diff --git a/crates/irtt-stats/src/lib.rs b/crates/irtt-stats/src/lib.rs index 82235f4..0047b15 100644 --- a/crates/irtt-stats/src/lib.rs +++ b/crates/irtt-stats/src/lib.rs @@ -58,7 +58,9 @@ pub struct StatsConfig { /// /// This window has no item cap, even in continuous mode. Its memory depends /// on event volume inside the interval. Expiry occurs on insertion, anchored - /// at the latest event timestamp, not when a snapshot is read. + /// at the maximum observed normalized event timestamp, not when a snapshot + /// is read. Events exactly at the cutoff are included. Backdated events + /// cannot move the anchor backwards or restore expired history. pub rolling_time: Option, /// Whether matched late replies contribute measurements. pub late_replies: LateReplyMode, diff --git a/crates/irtt-stats/src/normalization.rs b/crates/irtt-stats/src/normalization.rs index c3f8726..ea6cf65 100644 --- a/crates/irtt-stats/src/normalization.rs +++ b/crates/irtt-stats/src/normalization.rs @@ -118,10 +118,9 @@ pub(crate) fn normalize_event(event: &ClientEvent) -> Option { // `timeout_at` is anchored to the pre-send probe timestamp (see // irtt-client's AGENTS.md), so it can precede this same probe's // own post-send `sent_at` when a probe_timeout is small relative - // to the send-call duration. Rolling-window eviction assumes - // events are pushed in non-decreasing `at()` order, so clamp to - // `sent_at` to preserve that invariant without changing the - // public `timeout_at` value on the event itself. + // to the send-call duration. Clamp to `sent_at` so the loss is + // not earlier than its own send. Delayed timeout discovery can + // still put it before other probes' events. at: (*timeout_at).max(sent_at.mono), }), ClientEvent::Warning { at, .. } => Some(StatsEvent::Warning { at: at.mono }), diff --git a/crates/irtt-stats/src/rolling.rs b/crates/irtt-stats/src/rolling.rs index c521de1..6e54c79 100644 --- a/crates/irtt-stats/src/rolling.rs +++ b/crates/irtt-stats/src/rolling.rs @@ -1,4 +1,7 @@ -use std::{collections::VecDeque, time::Duration}; +use std::{ + collections::VecDeque, + time::{Duration, Instant}, +}; use measurement_stats::CountWindow; @@ -11,6 +14,7 @@ pub(crate) struct RollingEvents { time_limit: Option, count_events: Option>, time_events: Option>, + time_anchor: Option, late_replies: LateReplyMode, } @@ -20,6 +24,7 @@ impl RollingEvents { time_limit: config.rolling_time, count_events: config.rolling_count.map(CountWindow::new), time_events: config.rolling_time.map(|_| VecDeque::new()), + time_anchor: None, late_replies: config.late_replies, } } @@ -30,12 +35,14 @@ impl RollingEvents { } if let (Some(duration), Some(window)) = (self.time_limit, self.time_events.as_mut()) { - let cutoff = event.at().checked_sub(duration); + let anchor = self.time_anchor.map_or(event.at(), |at| at.max(event.at())); + self.time_anchor = Some(anchor); + let cutoff = anchor.checked_sub(duration); window.push_back(event); if let Some(cutoff) = cutoff { - while window.front().is_some_and(|event| event.at() < cutoff) { - window.pop_front(); - } + // Delayed timeout discovery can backdate events behind newer + // ones. Filter the whole window, preserving replay order. + window.retain(|event| event.at() >= cutoff); } } } diff --git a/crates/irtt-stats/tests/rolling_properties.rs b/crates/irtt-stats/tests/rolling_properties.rs index 91222e5..d8cfcd6 100644 --- a/crates/irtt-stats/tests/rolling_properties.rs +++ b/crates/irtt-stats/tests/rolling_properties.rs @@ -1,6 +1,6 @@ //! Independent count and time windows must match a filtered replay through //! StatsCollector, while cumulative accounting retains the complete history. -//! Generated events have non-decreasing normalized window timestamps. +//! Includes delayed timeout discovery with backdated window timestamps. use std::time::{Duration, Instant, UNIX_EPOCH}; @@ -12,8 +12,8 @@ use irtt_stats::{LateReplyMode, SampleMode, Snapshot, StatsCollector, StatsConfi use proptest::prelude::*; /// One generated operation. `Advance` moves the shared monotonic clock -/// forward without producing an event; every other variant produces exactly -/// one normalized event at the clock's current value. +/// forward without producing an event; other variants produce a normalized +/// event, with delayed losses timestamped before the clock's current value. #[derive(Debug, Clone)] enum Op { /// Advance the shared clock by this many milliseconds. @@ -30,6 +30,8 @@ enum Op { Duplicate, /// A probe timeout / loss. Loss, + /// A timeout discovered this many milliseconds after its deadline. + DelayedLoss { age_ms: u16 }, /// A diagnostic warning event. Warning, } @@ -43,6 +45,7 @@ fn op_strategy() -> impl Strategy { 1 => Just(Op::LateReplyUnmatched), 1 => Just(Op::Duplicate), 2 => Just(Op::Loss), + 2 => (0u16..100).prop_map(|age_ms| Op::DelayedLoss { age_ms }), 1 => Just(Op::Warning), ] } @@ -83,8 +86,8 @@ fn one_way(seq: u32) -> OneWayDelaySample { } /// Builds the `ClientEvent` for one non-`Advance` op. `clock_ms` is the -/// shared clock's current value, used as the event's windowing timestamp -/// (`at()`, see `normalization.rs`) for every variant. +/// shared clock's current value; delayed losses use their earlier deadline +/// as the event's windowing timestamp. fn build_event(op: &Op, seq: u32, clock_ms: u64, base: Instant) -> ClientEvent { match *op { Op::Advance(_) => unreachable!("Advance does not produce an event"), @@ -172,6 +175,14 @@ fn build_event(op: &Op, seq: u32, clock_ms: u64, base: Instant) -> ClientEvent { sent_at: ts_at(base, clock_ms), timeout_at: ts_at(base, clock_ms).mono - Duration::from_millis(10), }, + Op::DelayedLoss { age_ms } => { + let deadline_ms = clock_ms.saturating_sub(u64::from(age_ms)); + ClientEvent::EchoLoss { + seq, + sent_at: ts_at(base, deadline_ms.saturating_sub(3)), + timeout_at: ts_at(base, deadline_ms).mono, + } + } Op::Warning => ClientEvent::Warning { kind: irtt_client::WarningKind::UntrackedReply, message: "generated".to_owned(), @@ -245,7 +256,11 @@ proptest! { let seq = u32::try_from(idx).unwrap(); let event = build_event(op, seq, clock_ms, base); collector.process(&event); - history.push((event, clock_ms)); + let at_ms = match op { + Op::DelayedLoss { age_ms } => clock_ms.saturating_sub(u64::from(*age_ms)), + _ => clock_ms, + }; + history.push((event, at_ms)); } // Cumulative snapshot: unaffected by rolling eviction, so it must @@ -271,13 +286,10 @@ proptest! { ); // Time window reference: every retained event whose timestamp is - // within `time_limit_ms` of the *latest processed event's* - // timestamp, ignoring the count bound entirely. This recompute- - // from-scratch filter is equivalent to the production window's - // incremental sliding-eviction only because `clock_ms` is - // non-decreasing across pushes (enforced by construction above: - // `Advance` never subtracts). - if let Some((_, latest_at_ms)) = history.last() { + // within `time_limit_ms` of the maximum observed event timestamp, + // ignoring the count bound entirely. Delayed losses cannot move + // this anchor backwards. + if let Some(latest_at_ms) = history.iter().map(|(_, at_ms)| at_ms).max() { let cutoff = latest_at_ms.checked_sub(time_limit_ms); let time_ref: Vec = history .iter() diff --git a/crates/irtt-stats/tests/stats.rs b/crates/irtt-stats/tests/stats.rs index 6a2d1e5..b74e4b4 100644 --- a/crates/irtt-stats/tests/stats.rs +++ b/crates/irtt-stats/tests/stats.rs @@ -512,6 +512,53 @@ fn rolling_time_eviction_uses_event_timestamps() { assert_eq!(rolling.packets.packets_sent, 1); } +#[test] +fn rolling_time_expires_delayed_losses_without_moving_backwards() { + let base = Instant::now(); + let at = |ms| ClientTimestamp { + mono: base + Duration::from_millis(ms), + wall: UNIX_EPOCH + Duration::from_millis(ms), + }; + let mut collector = StatsCollector::new(StatsConfig { + rolling_count: Some(2), + rolling_time: Some(Duration::from_millis(3)), + ..StatsConfig::continuous() + }); + collector.process(&sent(0, at(1))); + collector.process(&sent(1, at(2))); + collector.process(&sent(2, at(5))); + // Timeout polling discovers probe 0's deadline after the later send. + collector.process(&ClientEvent::EchoLoss { + seq: 0, + sent_at: at(1), + timeout_at: at(4).mono, + }); + assert_eq!(collector.rolling_time().unwrap().events.loss_events, 1); + + collector.process(&sent(3, at(8))); + let rolling = collector.rolling_time().unwrap(); + // The cutoff is inclusive: the send at 5 remains, the loss at 4 expires. + assert_eq!(rolling.events.loss_events, 0); + assert_eq!(rolling.packets.packets_sent, 2); + assert_eq!(rolling.send_call.count, 2); + + collector.process(&sent(4, at(9))); + let rolling = collector.rolling_time().unwrap(); + assert_eq!(rolling.packets.packets_sent, 2); + // Another delayed timeout is already outside the window when discovered. + collector.process(&ClientEvent::EchoLoss { + seq: 1, + sent_at: at(2), + timeout_at: at(5).mono, + }); + assert_eq!(collector.rolling_time().unwrap(), rolling); + // Time expiry does not change cumulative or arrival-count accounting. + assert_eq!(collector.snapshot().packets.packets_sent, 5); + assert_eq!(collector.snapshot().events.loss_events, 2); + assert_eq!(collector.rolling_count().unwrap().events.loss_events, 1); + assert_eq!(collector.rolling_count().unwrap().packets.packets_sent, 1); +} + #[test] fn empty_and_all_lost_edges_are_defined() { let empty = StatsCollector::new(StatsConfig::finite()).snapshot();