From 1834663e37b3ef7720e613e42671476b214e60c1 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Tue, 28 Jul 2026 07:45:52 -0700 Subject: [PATCH] [adapters] Fix the tick-accounting race in test_clock The test sampled Run 1's post-checkpoint tick count right after `controller.pause()`. Pause is asynchronous, so a step that raced it was still journaled, and hence still replayed, while its tick went uncounted: the baseline came out one short of what Run 2 replays. The assertion allows one extra replayed tick, but that allowance is already spent. Once replay ends, the first step commits the last replayed transaction, and `input_step` queues every connector on that step regardless of the paused state, so the clock connector answers with a live tick. Run 2 therefore replays the journaled ticks plus one every time, which leaves no room for a miscounted tick in Run 1. Take the baseline after `stop`, which joins the circuit thread and so counts every journaled step, and wait for the replayed ticks instead of sleeping a fixed five seconds, so a slow replay cannot be sampled halfway through. Failure: https://github.com/feldera/feldera/actions/runs/30334386760/job/90198147938 Previously reported as #4516. --- crates/adapters/src/transport/clock.rs | 70 ++++++++++++++++++++------ 1 file changed, 56 insertions(+), 14 deletions(-) diff --git a/crates/adapters/src/transport/clock.rs b/crates/adapters/src/transport/clock.rs index f76c0b54d1d..336f4396957 100644 --- a/crates/adapters/src/transport/clock.rs +++ b/crates/adapters/src/transport/clock.rs @@ -440,6 +440,26 @@ mod test { } } + /// Waits for the circuit to process `expected` ticks, then returns the + /// tick count once it has settled. + /// + /// Waiting beats a fixed sleep for the lower bound: a loaded machine can + /// take longer to replay than any sleep the test would pick. The settle + /// period then gives a tick beyond `expected` time to arrive, so the + /// caller checks its upper bound against the final count rather than a + /// snapshot taken mid-replay. + fn settled_ticks(stats: &ClockStats, expected: usize) -> usize { + const TIMEOUT: Duration = Duration::from_secs(30); + const SETTLE: Duration = Duration::from_secs(2); + + let deadline = std::time::Instant::now() + TIMEOUT; + while stats.ticks() < expected && std::time::Instant::now() < deadline { + sleep(Duration::from_millis(10)); + } + sleep(SETTLE); + stats.ticks() + } + /// Create a simple test circuit that passes each of a number of streams right /// through to the corresponding output. The number of streams is the length of /// `persistent_output_ids`, which also specifies optional persistent output ids. @@ -573,17 +593,35 @@ mod test { sleep(Duration::from_secs(5)); - // Pause the controller first to prevent more ticks from being generated - // before we count them and stop the pipeline + // Stop the tick stream, so that Run 1 ends with a bounded number of + // post-checkpoint ticks. controller.pause(); - let old_ticks = ticks; - let ticks_after_checkpoint = test_stats.ticks() - old_ticks; + println!("Stopping the pipeline"); + controller.stop().unwrap(); + + // Count Run 1's post-checkpoint ticks only once the pipeline has + // stopped, which `stop` guarantees by joining the circuit thread. + // `pause` is asynchronous, and a step that races it is still journaled + // and hence still replayed, so a count taken before the stop can be + // one short of what Run 2 replays. + let ticks_after_checkpoint = test_stats.ticks() - ticks; println!("{ticks_after_checkpoint} additional ticks after the checkpoint"); - // Capture the last `NOW()` emitted in Run 1. Used below to verify - // the no-catchup contract across the restart. + // The replay assertion below tolerates one live tick, so a baseline of + // one is indistinguishable from a replay that emits nothing, and a + // baseline of zero asserts nothing at all. Two journaled ticks is the + // smallest baseline that makes the assertion detect a broken replay. + assert!( + ticks_after_checkpoint >= 2, + "only {ticks_after_checkpoint} ticks were journaled between the \ + checkpoint and the stop; the replay assertion needs at least two \ + to tell a real replay from the live tick that follows it" + ); + + // The last `NOW()` emitted in Run 1. Used below to verify the + // no-catchup contract across the restart. let last_v_run1 = test_stats.last_value(); println!("Run 1 last emitted NOW(): {last_v_run1} ms (anchor {anchor_ms})"); assert!( @@ -591,9 +629,6 @@ mod test { "Run 1 should have emitted at least one tick past the anchor" ); - println!("Stopping the pipeline"); - controller.stop().unwrap(); - test_stats.clear(); // The connector should replay the exact number of clock ticks. @@ -608,12 +643,19 @@ mod test { ) .unwrap(); - sleep(Duration::from_secs(5)); - - let ticks = test_stats.ticks(); + // Replay runs without `start`, and the pipeline stays paused once it + // completes, so the tick count settles as soon as replay is done. + let ticks = settled_ticks(&test_stats, ticks_after_checkpoint); println!("{ticks} ticks replayed after restart"); - // Allow at most one extra tick after pause. - assert!(ticks >= ticks_after_checkpoint && ticks <= ticks_after_checkpoint + 1); + // One extra tick is allowed: the first step after replay commits the + // last replayed transaction, and the controller queries every + // connector for input on that step, so the clock connector answers it + // with a live tick even though the pipeline is paused. + assert!( + ticks >= ticks_after_checkpoint && ticks <= ticks_after_checkpoint + 1, + "replayed {ticks} ticks, expected the {ticks_after_checkpoint} \ + journaled after the checkpoint, plus at most one live tick" + ); // No-catchup contract: post-replay emissions continue from // `last_v_run1` rather than jumping back to the anchor.