From edff75e55aed598c5290aa42100c125642bca4a2 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sat, 3 Oct 2026 22:29:42 +0700 Subject: [PATCH 1/3] fix(stats): expire backdated events from rolling time windows --- crates/irtt-stats/src/lib.rs | 4 +- crates/irtt-stats/src/normalization.rs | 7 ++- crates/irtt-stats/src/rolling.rs | 17 +++++-- crates/irtt-stats/tests/rolling_properties.rs | 38 ++++++++++----- crates/irtt-stats/tests/stats.rs | 47 +++++++++++++++++++ 5 files changed, 90 insertions(+), 23 deletions(-) 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(); From 8986caecb06feed2a9cb21690865f95388c7b4ee Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 4 Oct 2026 00:06:15 +0700 Subject: [PATCH 2/3] perf(stats): index rolling expiry to avoid full-window scans --- crates/irtt-stats/src/rolling.rs | 92 +++++++++++++++++-- crates/irtt-stats/tests/rolling_properties.rs | 78 ++++++++++++++-- 2 files changed, 154 insertions(+), 16 deletions(-) diff --git a/crates/irtt-stats/src/rolling.rs b/crates/irtt-stats/src/rolling.rs index 6e54c79..ba8be2a 100644 --- a/crates/irtt-stats/src/rolling.rs +++ b/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, count_events: Option>, - time_events: Option>, + time_events: Option>>, + time_expiry: BTreeSet<(Instant, usize)>, + time_front: usize, time_anchor: Option, 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 - // ones. Filter the whole window, preserving replay order. - window.retain(|event| event.at() >= cutoff); + // Expiry uses timestamp order; replay keeps arrival order. + while self.time_expiry.first().is_some_and(|(at, _)| *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 { - self.time_events - .as_ref() - .map(|events| snapshot_window(events.iter(), self.late_replies)) + self.time_events.as_ref().map(|events| { + snapshot_window(events.iter().filter_map(Option::as_ref), 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); + } +} diff --git a/crates/irtt-stats/tests/rolling_properties.rs b/crates/irtt-stats/tests/rolling_properties.rs index d8cfcd6..16e15ec 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. -//! 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 { 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 -/// as the event's windowing timestamp. +/// 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 { @@ -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)); From bdc5d152d4e1bb24e3441c90e2d4d63f41bce0b8 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 4 Oct 2026 00:52:58 +0700 Subject: [PATCH 3/3] fix(stats): reclaim expired rolling events without tombstones --- crates/irtt-stats/README.md | 6 +- crates/irtt-stats/src/lib.rs | 3 +- crates/irtt-stats/src/rolling.rs | 108 +++++++++++------- crates/irtt-stats/tests/rolling_properties.rs | 7 +- 4 files changed, 76 insertions(+), 48 deletions(-) diff --git a/crates/irtt-stats/README.md b/crates/irtt-stats/README.md index ed46b97..ddde8ef 100644 --- a/crates/irtt-stats/README.md +++ b/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 diff --git a/crates/irtt-stats/src/lib.rs b/crates/irtt-stats/src/lib.rs index 0047b15..824bcfc 100644 --- a/crates/irtt-stats/src/lib.rs +++ b/crates/irtt-stats/src/lib.rs @@ -60,7 +60,8 @@ pub struct StatsConfig { /// on event volume inside the interval. Expiry occurs on insertion, anchored /// 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. + /// cannot move the anchor backwards or restore expired history. Retained + /// events are replayed in arrival order. pub rolling_time: Option, /// Whether matched late replies contribute measurements. pub late_replies: LateReplyMode, diff --git a/crates/irtt-stats/src/rolling.rs b/crates/irtt-stats/src/rolling.rs index ba8be2a..bc58055 100644 --- a/crates/irtt-stats/src/rolling.rs +++ b/crates/irtt-stats/src/rolling.rs @@ -1,5 +1,5 @@ use std::{ - collections::{BTreeSet, VecDeque}, + collections::{BTreeMap, BTreeSet}, time::{Duration, Instant}, }; @@ -9,22 +9,24 @@ use crate::{ core::CoreStats, normalization::StatsEvent, LateReplyMode, SampleMode, Snapshot, StatsConfig, }; +type ArrivalKey = (Instant, usize); + #[derive(Debug, Clone)] pub(crate) struct RollingEvents { time_limit: Option, count_events: Option>, - time_events: Option>>, - time_expiry: BTreeSet<(Instant, usize)>, - time_front: usize, + // Both indexes contain exactly the live events; expiry removes from both. + time_events: Option>, + time_expiry: BTreeSet<(Instant, ArrivalKey)>, time_anchor: Option, late_replies: LateReplyMode, } impl PartialEq for RollingEvents { fn eq(&self, other: &Self) -> bool { - // Expiry ordinals and tombstones are bookkeeping, not retained events. + // Arrival keys 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()), + (Some(left), Some(right)) => left.values().eq(right.values()), (None, None) => true, _ => false, }; @@ -41,9 +43,8 @@ impl RollingEvents { 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_front: 0, time_anchor: None, late_replies: config.late_replies, } @@ -61,23 +62,19 @@ impl RollingEvents { if let Some(cutoff) = cutoff { // Expiry uses timestamp order; replay keeps arrival order. while self.time_expiry.first().is_some_and(|(at, _)| *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); + let (_, arrival) = self.time_expiry.pop_first().unwrap(); + window.remove(&arrival); } 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)); + // 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); } } @@ -88,9 +85,9 @@ impl RollingEvents { } pub(crate) fn time_snapshot(&self) -> Option { - self.time_events.as_ref().map(|events| { - snapshot_window(events.iter().filter_map(Option::as_ref), self.late_replies) - }) + self.time_events + .as_ref() + .map(|events| snapshot_window(events.values(), self.late_replies)) } } @@ -111,36 +108,63 @@ fn snapshot_window<'a>( mod tests { use super::*; - // Storage reclamation and ordinal rollover cannot be observed in snapshots. + // Snapshots alone cannot reveal storage or traversal over expired events. #[test] - fn expired_storage_is_reclaimed_across_ordinal_rollover() { + 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_millis(10)), + rolling_time: Some(Duration::from_secs(100)), ..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), - }); + 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)); } - // 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) + 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(StatsEvent::Warning { - at: base + Duration::from_millis(2_000), - }); + 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 + ); + } } diff --git a/crates/irtt-stats/tests/rolling_properties.rs b/crates/irtt-stats/tests/rolling_properties.rs index 16e15ec..46f06cd 100644 --- a/crates/irtt-stats/tests/rolling_properties.rs +++ b/crates/irtt-stats/tests/rolling_properties.rs @@ -221,12 +221,13 @@ fn collector_equality_ignores_expired_time_window_bookkeeping() { }; let mut left = StatsCollector::new(config); let mut right = StatsCollector::new(config); - for at_ms in [1, 10] { + for at_ms in [1, 2, 4, 5, 6] { left.process(&build_event(&Op::Warning, 0, at_ms, base)); } - for at_ms in [10, 1] { + 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); @@ -288,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 {