Merge branch 'fix/rolling-time-ordering'

crates/irtt-stats/README.md

@@ -32,8 +32,10 @@ unbounded sessions. Rolling-window snapshots are always running-only and
 report no medians, whichever `StatsConfig` produced them.
 
 An enabled time-based rolling window has no event-count cap, including in
-continuous mode. It expires only on new events, anchored at their timestamps;
-reading a snapshot does not advance time. Use count-based rolling storage for a
+continuous mode. It expires only on new events, anchored at the maximum observed
+normalized timestamp, with events exactly at the cutoff retained. Retained events
+are replayed in arrival order; backdated events cannot move the window backwards.
+Reading a snapshot does not advance time. Use count-based rolling storage for a
 hard bound on retained events. The memory estimate excludes time-window storage.
 
 Exact snapshots sort a temporary copy of each metric's retained samples, adding

crates/irtt-stats/src/lib.rs

@@ -58,7 +58,10 @@ 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. Retained
+    /// events are replayed in arrival order.
     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
-            // 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 }),

crates/irtt-stats/src/rolling.rs

@@ -1,4 +1,7 @@
-use std::{collections::VecDeque, time::Duration};
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    time::{Duration, Instant},
+};
 
 use measurement_stats::CountWindow;
 
@@ -6,20 +9,43 @@ use crate::{
     core::CoreStats, normalization::StatsEvent, LateReplyMode, SampleMode, Snapshot, StatsConfig,
 };
 
-#[derive(Debug, Clone, PartialEq)]
+type ArrivalKey = (Instant, usize);
+
+#[derive(Debug, Clone)]
 pub(crate) struct RollingEvents {
     time_limit: Option<Duration>,
     count_events: Option<CountWindow<StatsEvent>>,
-    time_events: Option<VecDeque<StatsEvent>>,
+    // Both indexes contain exactly the live events; expiry removes from both.
+    time_events: Option<BTreeMap<ArrivalKey, StatsEvent>>,
+    time_expiry: BTreeSet<(Instant, ArrivalKey)>,
+    time_anchor: Option<Instant>,
     late_replies: LateReplyMode,
 }
 
+impl PartialEq for RollingEvents {
+    fn eq(&self, other: &Self) -> bool {
+        // Arrival keys are bookkeeping, not retained events.
+        let same_time_events = match (&self.time_events, &other.time_events) {
+            (Some(left), Some(right)) => left.values().eq(right.values()),
+            (None, None) => true,
+            _ => false,
+        };
+        self.time_limit == other.time_limit
+            && self.count_events == other.count_events
+            && self.time_anchor == other.time_anchor
+            && self.late_replies == other.late_replies
+            && same_time_events
+    }
+}
+
 impl RollingEvents {
     pub(crate) fn new(config: StatsConfig) -> Self {
         Self {
             time_limit: config.rolling_time,
             count_events: config.rolling_count.map(CountWindow::new),
-            time_events: config.rolling_time.map(|_| VecDeque::new()),
+            time_events: config.rolling_time.map(|_| BTreeMap::new()),
+            time_expiry: BTreeSet::new(),
+            time_anchor: None,
             late_replies: config.late_replies,
         }
     }
@@ -30,13 +56,25 @@ impl RollingEvents {
         }
 
         if let (Some(duration), Some(window)) = (self.time_limit, self.time_events.as_mut()) {
-            let cutoff = event.at().checked_sub(duration);
-            window.push_back(event);
+            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);
             if let Some(cutoff) = cutoff {
-                while window.front().is_some_and(|event| event.at() < cutoff) {
-                    window.pop_front();
+                // Expiry uses timestamp order; replay keeps arrival order.
+                while self.time_expiry.first().is_some_and(|(at, _)| *at < cutoff) {
+                    let (_, arrival) = self.time_expiry.pop_first().unwrap();
+                    window.remove(&arrival);
+                }
+                if event.at() < cutoff {
+                    return;
                 }
             }
+            // Within one anchor, no accepted event expires, so length orders
+            // arrivals. A newer anchor sorts after all earlier arrivals even
+            // when expiry reduces the length. No lifetime counter can wrap.
+            let arrival = (anchor, window.len());
+            self.time_expiry.insert((event.at(), arrival));
+            window.insert(arrival, event);
         }
     }
 
@@ -49,7 +87,7 @@ impl RollingEvents {
     pub(crate) fn time_snapshot(&self) -> Option<Snapshot> {
         self.time_events
             .as_ref()
-            .map(|events| snapshot_window(events.iter(), self.late_replies))
+            .map(|events| snapshot_window(events.values(), self.late_replies))
     }
 }
 
@@ -65,3 +103,68 @@ fn snapshot_window<'a>(
     }
     core.snapshot()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // Snapshots alone cannot reveal storage or traversal over expired events.
+    #[test]
+    fn expired_interior_batches_are_reclaimed_while_the_front_stays_live() {
+        let base = Instant::now();
+        let warning = |seconds| StatsEvent::Warning {
+            at: base + Duration::from_secs(seconds),
+        };
+        let mut rolling = RollingEvents::new(StatsConfig {
+            rolling_time: Some(Duration::from_secs(100)),
+            ..StatsConfig::continuous()
+        });
+        rolling.push(warning(100));
+        for second in 0..64 {
+            // A large batch sits exactly on the cutoff behind the live front.
+            for _ in 0..2_048 {
+                rolling.push(warning(second));
+            }
+            assert_eq!(
+                rolling.time_snapshot().unwrap().events.warning_events,
+                second + 1 + 2_048
+            );
+            rolling.push(warning(101 + second));
+            // This now-expired arrival must not allocate another retained slot.
+            rolling.push(warning(second));
+            let live = usize::try_from(second + 2).unwrap();
+            assert_eq!(rolling.time_events.as_ref().unwrap().len(), live);
+            assert_eq!(rolling.time_expiry.len(), live);
+            assert_eq!(
+                rolling.time_snapshot().unwrap().events.warning_events,
+                u64::try_from(live).unwrap()
+            );
+        }
+        rolling.push(warning(1_000));
+        assert_eq!(rolling.time_events.as_ref().unwrap().len(), 1);
+        assert_eq!(rolling.time_expiry.len(), 1);
+    }
+
+    #[test]
+    fn chronological_stream_retains_only_the_current_window() {
+        let base = Instant::now();
+        let window_ms = 16_384;
+        let mut rolling = RollingEvents::new(StatsConfig {
+            rolling_time: Some(Duration::from_millis(window_ms)),
+            ..StatsConfig::continuous()
+        });
+        // Fill a substantial uncapped window, then run through several expiries.
+        for at_ms in 0..4 * window_ms {
+            rolling.push(StatsEvent::Warning {
+                at: base + Duration::from_millis(at_ms),
+            });
+            let live = usize::try_from(at_ms.min(window_ms) + 1).unwrap();
+            assert_eq!(rolling.time_events.as_ref().unwrap().len(), live);
+            assert_eq!(rolling.time_expiry.len(), live);
+        }
+        assert_eq!(
+            rolling.time_snapshot().unwrap().events.warning_events,
+            window_ms + 1
+        );
+    }
+}

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 and replies with backdated 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 events timestamped before the clock's current value.
 #[derive(Debug, Clone)]
 enum Op {
     /// Advance the shared clock by this many milliseconds.
@@ -22,6 +22,8 @@ enum Op {
     Send,
     /// An on-time unique reply with the given client-observed RTT.
     Reply { raw_ms: u16 },
+    /// A reply processed after later events, but timestamped at receipt.
+    DelayedReply { raw_ms: u16, age_ms: u16 },
     /// A late reply matched to retained send state (measurable).
     LateReplyMatched { raw_ms: u16 },
     /// A late reply that could not be matched to retained state.
@@ -30,6 +32,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,
 }
@@ -39,10 +43,12 @@ fn op_strategy() -> impl Strategy<Value = Op> {
         3 => (0u16..40).prop_map(Op::Advance),
         4 => Just(Op::Send),
         4 => (0u16..60).prop_map(|raw_ms| Op::Reply { raw_ms }),
+        2 => (0u16..60, 0u16..100).prop_map(|(raw_ms, age_ms)| Op::DelayedReply { raw_ms, age_ms }),
         2 => (0u16..60).prop_map(|raw_ms| Op::LateReplyMatched { raw_ms }),
         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,9 +89,13 @@ 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 events use their earlier deadline or
+/// receipt time as the event's windowing timestamp.
 fn build_event(op: &Op, seq: u32, clock_ms: u64, base: Instant) -> ClientEvent {
+    let clock_ms = match *op {
+        Op::DelayedReply { age_ms, .. } => clock_ms.saturating_sub(u64::from(age_ms)),
+        _ => clock_ms,
+    };
     match *op {
         Op::Advance(_) => unreachable!("Advance does not produce an event"),
         Op::Send => ClientEvent::EchoSent {
@@ -97,7 +107,7 @@ fn build_event(op: &Op, seq: u32, clock_ms: u64, base: Instant) -> ClientEvent {
             send_call: Duration::from_micros(10),
             timer_error: Some(Duration::from_micros(2)),
         },
-        Op::Reply { raw_ms } => {
+        Op::Reply { raw_ms } | Op::DelayedReply { raw_ms, .. } => {
             let sent_at = ts_at(base, clock_ms.saturating_sub(u64::from(raw_ms)));
             let received_at = ts_at(base, clock_ms);
             ClientEvent::EchoReply {
@@ -172,6 +182,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(),
@@ -194,6 +212,66 @@ fn replay(events: &[ClientEvent], late_replies: LateReplyMode) -> Snapshot {
     collector.snapshot()
 }
 
+#[test]
+fn collector_equality_ignores_expired_time_window_bookkeeping() {
+    let base = Instant::now();
+    let config = StatsConfig {
+        rolling_time: Some(Duration::from_millis(3)),
+        ..StatsConfig::continuous()
+    };
+    let mut left = StatsCollector::new(config);
+    let mut right = StatsCollector::new(config);
+    for at_ms in [1, 2, 4, 5, 6] {
+        left.process(&build_event(&Op::Warning, 0, at_ms, base));
+    }
+    for at_ms in [1, 4, 2, 5, 6] {
+        right.process(&build_event(&Op::Warning, 0, at_ms, base));
+    }
+    // Different expired histories leave the same live events in arrival order.
+    assert_eq!(left.snapshot(), right.snapshot());
+    assert_eq!(left.rolling_time(), right.rolling_time());
+    assert_eq!(left, right);
+}
+
+#[test]
+fn time_expiry_preserves_arrival_order_for_server_receive_window() {
+    let base = Instant::now();
+    let mut collector = StatsCollector::new(StatsConfig {
+        rolling_time: Some(Duration::from_millis(3)),
+        ..StatsConfig::continuous()
+    });
+    for (seq, at_ms, window) in [(1, 5, 0x5), (0, 4, 0x4)] {
+        let mut event = build_event(&Op::Reply { raw_ms: 1 }, seq, at_ms, base);
+        let ClientEvent::EchoReply { received_stats, .. } = &mut event else {
+            unreachable!();
+        };
+        received_stats.as_mut().unwrap().window = Some(window);
+        collector.process(&event);
+    }
+    // The last arrival owns the observation even with an older receive time.
+    assert_eq!(
+        collector
+            .rolling_time()
+            .unwrap()
+            .packets
+            .server_received_window,
+        Some(0x4)
+    );
+    collector.process(&build_event(&Op::Warning, 2, 8, base));
+    let rolling = collector.rolling_time().unwrap();
+    assert_eq!(rolling.rtt.primary.count, 1);
+    assert_eq!(rolling.packets.server_received_window, Some(0x5));
+    // The hidden, expired observation remains in cumulative accounting.
+    assert_eq!(
+        collector.snapshot().packets.server_received_window,
+        Some(0x4)
+    );
+    collector.process(&build_event(&Op::Warning, 3, 9, base));
+    let rolling = collector.rolling_time().unwrap();
+    assert_eq!(rolling.rtt.primary.count, 0);
+    assert_eq!(rolling.packets.server_received_window, None);
+}
+
 proptest! {
     #![proptest_config(ProptestConfig { cases: 128, .. ProptestConfig::default() })]
 
@@ -211,7 +289,7 @@ proptest! {
     fn rolling_count_and_rolling_time_windows_are_independent(
         ops in prop::collection::vec(op_strategy(), 1..40),
         count_limit in 1usize..6,
-        time_limit_ms in 1u64..80,
+        time_limit_ms in 0u64..80,
         count_only in prop::bool::ANY,
     ) {
         let late_replies = if count_only {
@@ -245,7 +323,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 } | Op::DelayedReply { 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 +353,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<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();