fix(stats): expire backdated events from rolling time windows

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<Duration>,
     /// Whether matched late replies contribute measurements.
     pub late_replies: LateReplyMode,

crates/irtt-stats/src/normalization.rs

@@ -118,10 +118,9 @@ pub(crate) fn normalize_event(event: &ClientEvent) -> Option<StatsEvent> {
             // `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
+            // to the send-call duration. Clamp to `sent_at` so the loss is
-            // events are pushed in non-decreasing `at()` order, so clamp to
+            // not earlier than its own send. Delayed timeout discovery can
-            // `sent_at` to preserve that invariant without changing the
+            // still put it before other probes' events.
-            // public `timeout_at` value on the event itself.
             at: (*timeout_at).max(sent_at.mono),
         }),
         ClientEvent::Warning { at, .. } => Some(StatsEvent::Warning { at: at.mono }),

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<Duration>,
     count_events: Option<CountWindow<StatsEvent>>,
     time_events: Option<VecDeque<StatsEvent>>,
+    time_anchor: Option<Instant>,
     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) {
+                // Delayed timeout discovery can backdate events behind newer
-                    window.pop_front();
+                // ones. Filter the whole window, preserving replay order.
-                }
+                window.retain(|event| event.at() >= cutoff);
             }
         }
     }

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
+/// forward without producing an event; other variants produce a normalized
-/// one normalized event at the clock's current value.
+/// 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<Value = Op> {
         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
+/// shared clock's current value; delayed losses use their earlier deadline
-/// (`at()`, see `normalization.rs`) for every variant.
+/// 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*
+            // within `time_limit_ms` of the maximum observed event timestamp,
-            // timestamp, ignoring the count bound entirely. This recompute-
+            // ignoring the count bound entirely. Delayed losses cannot move
-            // from-scratch filter is equivalent to the production window's
+            // this anchor backwards.
-            // incremental sliding-eviction only because `clock_ms` is
+            if let Some(latest_at_ms) = history.iter().map(|(_, at_ms)| at_ms).max() {
-            // non-decreasing across pushes (enforced by construction above:
-            // `Advance` never subtracts).
-            if let Some((_, latest_at_ms)) = history.last() {
                 let cutoff = latest_at_ms.checked_sub(time_limit_ms);
                 let time_ref: Vec<ClientEvent> = history
                     .iter()

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();