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 82235f4..824bcfc 100644 --- a/crates/irtt-stats/src/lib.rs +++ b/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, /// 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..bc58055 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::{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, count_events: Option>, - time_events: Option>, + // 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 { + // 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 { 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 + ); + } +} diff --git a/crates/irtt-stats/tests/rolling_properties.rs b/crates/irtt-stats/tests/rolling_properties.rs index 91222e5..46f06cd 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 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 { 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 = 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();