From c426884fd19822b1d19d5790a0d506e49c5a4e82 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Mon, 14 Sep 2026 15:33:37 +0700 Subject: [PATCH] fix(client): prevent stagger gate from accumulating wakeup latency --- crates/irtt-client/src/managed/task.rs | 26 ++++++++- crates/irtt-client/src/managed/tests.rs | 72 +++++++++++++++++++++++++ 2 files changed, 97 insertions(+), 1 deletion(-) diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index ccf81dc..15bb2e9 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -1613,6 +1613,30 @@ impl ManagedClientTask { self.send_gate = self.send_gate.filter(|gate| *gate > now); } + /// The next stagger gate after a send was accepted at `accepted_at`. + /// + /// The gate advances on its **own** cadence rather than being re-anchored + /// on every observed accept time. Anchoring on `accepted_at` folds each + /// send's wakeup latency into the next slot and then into every slot after + /// it, so the gate — and with it the whole group's send cadence — drifts + /// later without bound while the per-target schedules stay on their + /// absolute grid. The drift is not hypothetical: it shows up as a + /// monotonically growing `EchoSent::timer_error` and, once it exceeds one + /// interval, as probe slots the schedule then skips. + /// + /// `accepted_at` is still the anchor in the two cases where continuing the + /// old cadence would be wrong: there is no previous gate, or the pacer has + /// fallen a full slot behind it. Re-anchoring there is what keeps a gate + /// stranded in the past from releasing a burst of catch-up sends once the + /// group starts moving again. + fn next_stagger_gate(&self, spacing: Duration, accepted_at: Instant) -> Option { + let base = match self.send_gate { + Some(gate) if gate <= accepted_at && accepted_at.duration_since(gate) < spacing => gate, + _ => accepted_at, + }; + base.checked_add(spacing) + } + fn record_stagger_acceptance( &mut self, result: SendResult, @@ -1626,7 +1650,7 @@ impl ManagedClientTask { return; }; self.last_stagger_send = Some(accepted_at); - self.send_gate = accepted_at.checked_add(spacing); + self.send_gate = self.next_stagger_gate(spacing, accepted_at); #[cfg(test)] if let Some(observations) = &self.stagger_observations { observations.lock().unwrap().push((_active, spacing)); diff --git a/crates/irtt-client/src/managed/tests.rs b/crates/irtt-client/src/managed/tests.rs index 99b323e..d5a09d1 100644 --- a/crates/irtt-client/src/managed/tests.rs +++ b/crates/irtt-client/src/managed/tests.rs @@ -1895,6 +1895,78 @@ fn stagger_gate_tracks_active_membership() { second.finish(); } +/// The stagger gate must not fold each send's observed wakeup latency into +/// every later slot. +/// +/// The gate is the binding constraint on when the *group* may send next (see +/// `next_deadline`), while each target's own schedule stays on an absolute +/// grid anchored at its open. Re-anchoring the gate on every observed accept +/// time therefore made the whole group's cadence drift later without bound: +/// `EchoSent::timer_error` grew monotonically, and once the drift exceeded one +/// interval `skip_missed_probe_slots_at` started dropping probe slots +/// outright. Measured before this was fixed, a single-target 20 s run at a +/// 50 ms interval sent 381 probes instead of 400. +/// +/// This asserts the invariant directly rather than through a real-time run, +/// because reproducing accumulation behaviorally needs seconds of wall clock +/// and a jitter assumption. Two properties, both durable: +/// +/// - while the pacer keeps up, N accepted sends advance the gate by exactly +/// N spacings from the first anchor, whatever lateness each send is +/// observed with; +/// - a gate the pacer has fallen a full spacing behind re-anchors on the +/// accept time instead, so a stranded gate cannot release a burst of +/// catch-up sends. +#[test] +fn the_stagger_gate_advances_by_one_spacing_per_send_without_accumulating_lateness() { + // No socket is involved: the gate is driven directly and the task is never + // polled, so an unreachable address is all a valid target set needs here. + let (mut task, _) = ManagedClient::task( + config(ManagedPacing::Staggered), + vec![target("only", "127.0.0.1:9".parse().unwrap())], + ) + .unwrap(); + + let spacing = Duration::from_millis(50); + let start = Instant::now(); + let lateness = Duration::from_millis(3); + + task.record_stagger_acceptance( + SendResult::Ready { accepted: true }, + Some((1, spacing)), + start, + ); + assert_eq!(task.send_gate, Some(start + spacing), "first send anchors"); + + // Every later send is observed `lateness` after the gate that released it. + // The gate must still advance one spacing per send from `start`, not + // `spacing + lateness`. + for slot in 1..=8_u32 { + let released_at = start + spacing * slot; + task.record_stagger_acceptance( + SendResult::Ready { accepted: true }, + Some((1, spacing)), + released_at + lateness, + ); + assert_eq!( + task.send_gate, + Some(start + spacing * (slot + 1)), + "slot {slot} accumulated observed lateness into the gate" + ); + } + + // A full spacing behind is no longer the same cadence, so the gate + // re-anchors rather than staying stranded in the past. + let stalled_gate = task.send_gate.expect("the gate is set"); + let resumed_at = stalled_gate + spacing; + task.record_stagger_acceptance( + SendResult::Ready { accepted: true }, + Some((1, spacing)), + resumed_at, + ); + assert_eq!(task.send_gate, Some(resumed_at + spacing)); +} + #[test] fn timeout_discovery_inspects_at_most_one_budget_of_targets() { let target_count = TIMEOUT_WORK_BUDGET + 1;