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