perf(stats): index rolling expiry to avoid full-window scans

crates/irtt-stats/src/rolling.rs

@@ -1,5 +1,5 @@
 use std::{
-    collections::VecDeque,
+    collections::{BTreeSet, VecDeque},
     time::{Duration, Instant},
 };
 
@@ -9,21 +9,41 @@ use crate::{
     core::CoreStats, normalization::StatsEvent, LateReplyMode, SampleMode, Snapshot, StatsConfig,
 };
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone)]
 pub(crate) struct RollingEvents {
     time_limit: Option<Duration>,
     count_events: Option<CountWindow<StatsEvent>>,
-    time_events: Option<VecDeque<StatsEvent>>,
+    time_events: Option<VecDeque<Option<StatsEvent>>>,
+    time_expiry: BTreeSet<(Instant, usize)>,
+    time_front: usize,
     time_anchor: Option<Instant>,
     late_replies: LateReplyMode,
 }
 
+impl PartialEq for RollingEvents {
+    fn eq(&self, other: &Self) -> bool {
+        // Expiry ordinals and tombstones are bookkeeping, not retained events.
+        let same_time_events = match (&self.time_events, &other.time_events) {
+            (Some(left), Some(right)) => left.iter().flatten().eq(right.iter().flatten()),
+            (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_expiry: BTreeSet::new(),
+            time_front: 0,
             time_anchor: None,
             late_replies: config.late_replies,
         }
@@ -38,12 +58,26 @@ impl RollingEvents {
             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 {
-                // Delayed timeout discovery can backdate events behind newer
+                // Expiry uses timestamp order; replay keeps arrival order.
-                // ones. Filter the whole window, preserving replay order.
+                while self.time_expiry.first().is_some_and(|(at, _)| *at < cutoff) {
-                window.retain(|event| event.at() >= cutoff);
+                    let (_, ordinal) = self.time_expiry.pop_first().unwrap();
+                    window[ordinal.wrapping_sub(self.time_front)] = None;
+                }
+                while window.front().is_some_and(Option::is_none) {
+                    window.pop_front();
+                    self.time_front = self.time_front.wrapping_add(1);
+                }
+                if event.at() < cutoff {
+                    return;
+                }
             }
+            // Logical deque positions allow constant-time expiry lookup. The
+            // queue cannot hold enough slots for live ordinals to alias, so
+            // wrapping arithmetic also handles a long-running ordinal rollover.
+            let ordinal = self.time_front.wrapping_add(window.len());
+            self.time_expiry.insert((event.at(), ordinal));
+            window.push_back(Some(event));
         }
     }
 
@@ -54,9 +88,9 @@ impl RollingEvents {
     }
 
     pub(crate) fn time_snapshot(&self) -> Option<Snapshot> {
-        self.time_events
+        self.time_events.as_ref().map(|events| {
-            .as_ref()
+            snapshot_window(events.iter().filter_map(Option::as_ref), self.late_replies)
-            .map(|events| snapshot_window(events.iter(), self.late_replies))
+        })
     }
 }
 
@@ -72,3 +106,41 @@ fn snapshot_window<'a>(
     }
     core.snapshot()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // Storage reclamation and ordinal rollover cannot be observed in snapshots.
+    #[test]
+    fn expired_storage_is_reclaimed_across_ordinal_rollover() {
+        let base = Instant::now();
+        let mut rolling = RollingEvents::new(StatsConfig {
+            rolling_time: Some(Duration::from_millis(10)),
+            ..StatsConfig::continuous()
+        });
+        rolling.time_front = usize::MAX - 1;
+        for cycle in 1..=100 {
+            let anchor_ms = cycle * 10;
+            // Each cycle hides an older event behind a newer one, then expires
+            // it while that newer event is still live.
+            for at_ms in [anchor_ms, anchor_ms - 9, anchor_ms + 2] {
+                rolling.push(StatsEvent::Warning {
+                    at: base + Duration::from_millis(at_ms),
+                });
+            }
+            // Both stores must stay tied to recent volume, not total arrivals.
+            assert!(rolling.time_events.as_ref().unwrap().len() <= 6);
+            assert!(rolling.time_expiry.len() <= 6);
+            assert_eq!(
+                rolling.time_snapshot().unwrap().events.warning_events,
+                2 + u64::from(cycle > 1)
+            );
+        }
+        rolling.push(StatsEvent::Warning {
+            at: base + Duration::from_millis(2_000),
+        });
+        assert_eq!(rolling.time_events.as_ref().unwrap().len(), 1);
+        assert_eq!(rolling.time_expiry.len(), 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.
-//! Includes delayed timeout discovery with backdated window timestamps.
+//! Includes delayed timeout discovery and replies with backdated timestamps.
 
 use std::time::{Duration, Instant, UNIX_EPOCH};
 
@@ -13,7 +13,7 @@ use proptest::prelude::*;
 
 /// One generated operation. `Advance` moves the shared monotonic clock
 /// forward without producing an event; other variants produce a normalized
-/// event, with delayed losses timestamped before the clock's current value.
+/// 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.
@@ -41,6 +43,7 @@ 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),
@@ -86,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; delayed losses use their earlier deadline
+/// shared clock's current value; delayed events use their earlier deadline or
-/// as the event's windowing timestamp.
+/// 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 {
@@ -100,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 {
@@ -205,6 +212,65 @@ 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, 10] {
+        left.process(&build_event(&Op::Warning, 0, at_ms, base));
+    }
+    for at_ms in [10, 1] {
+        right.process(&build_event(&Op::Warning, 0, at_ms, base));
+    }
+    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() })]
 
@@ -257,7 +323,7 @@ proptest! {
                 let event = build_event(op, seq, clock_ms, base);
                 collector.process(&event);
                 let at_ms = match op {
-                    Op::DelayedLoss { age_ms } => clock_ms.saturating_sub(u64::from(*age_ms)),
+                    Op::DelayedLoss { age_ms } | Op::DelayedReply { age_ms, .. } => clock_ms.saturating_sub(u64::from(*age_ms)),
                     _ => clock_ms,
                 };
                 history.push((event, at_ms));