From db05a3164becedce98a7fc43d8560161820ebb9f Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Thu, 9 Jul 2026 23:56:30 -0700 Subject: [PATCH] [adapters] Increase wait_for_records timeout in tests. suspend_barrier5 times out waiting for 4000 input records while suspending in CI. The log seems to indicate that the connector is making progress, just slowly under load. We increase the timeout from 10 to 100s and improve progress logging. If the issue occurs again, this means there's a real bug somewhere. Signed-off-by: Leonid Ryzhyk --- crates/adapters/src/controller/test.rs | 81 +++++++++++++++++++------- 1 file changed, 59 insertions(+), 22 deletions(-) diff --git a/crates/adapters/src/controller/test.rs b/crates/adapters/src/controller/test.rs index 6f2e34368a9..f19f98f3cde 100644 --- a/crates/adapters/src/controller/test.rs +++ b/crates/adapters/src/controller/test.rs @@ -431,9 +431,40 @@ fn collect_endpoint_records(controller: &Controller, n: usize) -> Vec { .collect() } +/// Default budget for [`wait_for_records`] to observe the expected output. +const OUTPUT_TIMEOUT_MS: u128 = 10_000; + +/// Longer output budget for waits taken while a suspend is in progress. +/// +/// While a suspend is pending, the controller advances only the barrier inputs +/// while re-attempting the checkpoint each step, so under CI CPU contention +/// output can lag far behind input even though stepping keeps making forward +/// progress. Using the default [`OUTPUT_TIMEOUT_MS`] in that window makes the +/// suspend tests flake. Matches the 100s budget already used for the +/// suspend-completion `recv_timeout` in these tests. +const SUSPEND_OUTPUT_TIMEOUT_MS: u128 = 100_000; + +/// Like `println!`, but prefixes a UTC timestamp so test output interleaves +/// legibly with the controller's tracing logs when diagnosing timing issues. +macro_rules! tprintln { + ($($arg:tt)*) => { + println!("[{}] {}", chrono::Utc::now().format("%H:%M:%S%.6f"), format_args!($($arg)*)) + }; +} + +/// Wait until every output endpoint has received at least the expected number +/// of records, then verify the counts match exactly, using the default +/// [`OUTPUT_TIMEOUT_MS`] budget. #[track_caller] fn wait_for_records(controller: &Controller, expect_n: &[usize]) { - println!("waiting for {expect_n:?} records..."); + wait_for_records_within(controller, expect_n, OUTPUT_TIMEOUT_MS); +} + +/// Like [`wait_for_records`], but waits up to `timeout_ms` for the records to +/// arrive. +#[track_caller] +fn wait_for_records_within(controller: &Controller, expect_n: &[usize], timeout_ms: u128) { + tprintln!("waiting for {expect_n:?} records..."); let n = expect_n.len(); let mut last_n = repeat_n(0, n).collect::>(); wait( @@ -441,7 +472,7 @@ fn wait_for_records(controller: &Controller, expect_n: &[usize]) { let new_n = collect_endpoint_records(controller, n); for i in 0..n { if new_n[i] > last_n[i] { - println!("received {} records on test_output{}", new_n[i], i + 1); + tprintln!("received {} records on test_output{}", new_n[i], i + 1); } } last_n = new_n; @@ -450,7 +481,7 @@ fn wait_for_records(controller: &Controller, expect_n: &[usize]) { .zip(expect_n.iter()) .all(|(&last, &expect)| last >= expect) }, - 10_000, + timeout_ms, ) .unwrap(); @@ -2028,25 +2059,25 @@ fn suspend_barrier() { .from_writer(&input_file); // Write records to the input file. - println!("Writing records 0..4000"); + tprintln!("Writing records 0..4000"); for id in 0..4000 { writer.serialize(TestStruct::for_id(id as u32)).unwrap(); } writer.flush().unwrap(); // Start pipeline. - println!("start pipeline"); + tprintln!("start pipeline"); let controller = start_controller(&storage_dir, &[5000]); // Wait for the records that are not in the checkpoint to be // processed or replayed. - println!("wait for 4000 records 0..4000"); + tprintln!("wait for 4000 records 0..4000"); wait_for_records(&controller, &[4000]); // Suspend. let (sender, receiver) = mpsc::channel(); - println!("start suspend"); + tprintln!("start suspend"); let suspend_request_step = controller.status().global_metrics.total_initiated_steps(); controller.start_suspend(Box::new(move |result| sender.send(result).unwrap())); @@ -2068,7 +2099,7 @@ fn suspend_barrier() { assert_bounded_suspend_steps(&controller, suspend_request_step); // Stop controller. - println!("stop controller"); + tprintln!("stop controller"); controller.stop().unwrap(); // Read output and compare. Our output adapter, which is not @@ -2082,7 +2113,7 @@ fn suspend_barrier() { let controller = start_controller(&storage_dir, &[5000]); wait_for_records(&controller, &[5000]); - println!("start suspend"); + tprintln!("start suspend"); let (sender, receiver) = mpsc::channel(); let mut sender = Some(sender); @@ -2092,13 +2123,16 @@ fn suspend_barrier() { let start = (i + 5) * 1000; let end = start + 1000; - println!("writing records {start}..{end}"); + tprintln!("writing records {start}..{end}"); for id in start..end { writer.serialize(TestStruct::for_id(id as u32)).unwrap(); } writer.flush().unwrap(); - println!("waiting for {end} records"); - wait_for_records(&controller, &[end]); + tprintln!("waiting for {end} records"); + // After the first iteration a suspend is pending: the controller only + // advances barrier inputs while re-attempting the checkpoint each step, + // so output can lag under CI CPU load. Use the longer budget. + wait_for_records_within(&controller, &[end], SUSPEND_OUTPUT_TIMEOUT_MS); check_file_contents(&(output_path(&storage_dir, 0)), 5000..end); if let Some(sender) = sender.take() { @@ -2157,7 +2191,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { .collect::>(); // Write records to the input files. - println!("Writing 1000 records to each of {n_inputs} files"); + tprintln!("Writing 1000 records to each of {n_inputs} files"); for writer in writers.iter_mut().take(n_inputs) { for id in 0..1000 { writer.serialize(TestStruct::for_id(id as u32)).unwrap(); @@ -2166,7 +2200,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { } // Start pipeline. - println!("start pipeline"); + tprintln!("start pipeline"); // The barrier for input 0 is record 0, // for input 1 is record 1000, @@ -2177,7 +2211,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { // Wait for the first 1000 records in each file to be read and copied to the // output. - println!("wait for 1000 records in each file"); + tprintln!("wait for 1000 records in each file"); let mut written = repeat_n(1000, n_inputs).collect::>(); wait_for_records(&controller, &written); @@ -2187,7 +2221,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { // we're past all the barriers; otherwise, it will not complete due to // barriers, since each input only has 1000 records so far. let (sender, receiver) = mpsc::channel(); - println!("start suspend"); + tprintln!("start suspend"); let suspend_request_step = controller.status().global_metrics.total_initiated_steps(); controller.start_suspend(Box::new(move |result| sender.send(result).unwrap())); @@ -2212,7 +2246,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { receiver.try_recv().unwrap_err(); // Write 1000 more records to the `next` input. - println!("writing 1000 more records to test_input{}", next + 1); + tprintln!("writing 1000 more records to test_input{}", next + 1); for id in written[next]..written[next] + 1000 { writers[next] .serialize(TestStruct::for_id(id as u32)) @@ -2228,9 +2262,12 @@ fn suspend_multiple_barriers(n_inputs: usize) { // We won't get any more records on output from inputs that have reached // their barrier, so the writes to inputs 0 and 1 won't have any effect // here. - println!("total written: {written:?}"); + tprintln!("total written: {written:?}"); let expect = expectations(&written, &barriers); - wait_for_records(&controller, &expect); + // A suspend is pending here: the controller only advances barrier inputs + // while re-attempting the checkpoint each step, so output can lag far + // behind under CI CPU load. Use the longer budget. + wait_for_records_within(&controller, &expect, SUSPEND_OUTPUT_TIMEOUT_MS); for (i, expectation) in expect.iter().enumerate().take(n_inputs) { check_file_contents(&output_path(&storage_dir, i), 0..*expectation); } @@ -2244,11 +2281,11 @@ fn suspend_multiple_barriers(n_inputs: usize) { assert_bounded_suspend_steps(&controller, suspend_request_step); // Stop controller. - println!("stop controller"); + tprintln!("stop controller"); controller.stop().unwrap(); // Check output one more time. - println!("check output one more time now that controller is stopped"); + tprintln!("check output one more time now that controller is stopped"); let expect = expectations(&written, &barriers); for (i, e) in expect.iter().enumerate().take(n_inputs) { check_file_contents(&output_path(&storage_dir, i), 0..*e); @@ -2256,7 +2293,7 @@ fn suspend_multiple_barriers(n_inputs: usize) { // Now restart the controller and wait for all the records that we wrote // beyond the barriers get copied to output (and nothing else). - println!("restart controller and wait for records beyond the barriers"); + tprintln!("restart controller and wait for records beyond the barriers"); let controller = start_controller(&storage_dir, &barriers); wait_for_records(&controller, &written); for i in 0..n_inputs {