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 {