From 743a0504165299506050a5dd67f4db36719dfdc3 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 4 Oct 2026 12:20:47 +0700 Subject: [PATCH 1/2] fix(stats): align rolling directional loss with window counts --- crates/irtt-stats/README.md | 20 ++ crates/irtt-stats/src/core.rs | 2 +- crates/irtt-stats/src/lib.rs | 5 +- crates/irtt-stats/src/loss.rs | 16 +- crates/irtt-stats/src/retention.rs | 9 +- crates/irtt-stats/src/rolling.rs | 90 +++++- crates/irtt-stats/tests/rolling_loss.rs | 281 ++++++++++++++++++ crates/irtt-stats/tests/rolling_properties.rs | 100 ++++++- 8 files changed, 489 insertions(+), 34 deletions(-) create mode 100644 crates/irtt-stats/tests/rolling_loss.rs diff --git a/crates/irtt-stats/README.md b/crates/irtt-stats/README.md index ddde8ef..5459473 100644 --- a/crates/irtt-stats/README.md +++ b/crates/irtt-stats/README.md @@ -38,6 +38,26 @@ 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. +Rolling directional loss uses the increase in the highest observed server +receive count across the retained packet events in arrival order. Upstream loss +is window sends minus that increase; downstream loss is the increase minus all +window replies, including duplicates and untracked late replies, as in cumulative +accounting. The baseline is the highest count observed before the first retained +packet event, or zero if no earlier packet events existed. Each retained event +keeps this prior position, so eviction cannot erase a known baseline. Reordered +counts do not move the observation backwards; only matched unique replies supply +counter observations, regardless of the late-reply timing policy. + +Both directional packet estimates are `None` if the baseline or endpoint is +unknown, or if time filtering excludes a packet event between retained packet +events. Their percentage fields use the existing unavailable value of `0.0`. +Loss/warning events do not affect this interval. A boundary between a send and +its reply can still produce signed estimates, and outstanding sends retain the +existing provisional-loss interpretation. These are changes in observed +accounting, not exact network transit loss for probes sent in a wall-clock period. +Raw `server_packets_received`, ordinary packet counts, and cumulative loss retain +their existing meanings. + Exact snapshots sort a temporary copy of each metric's retained samples, adding O(n) scratch memory and O(n log n) computation for that metric. The retention estimate is not peak snapshot or allocator accounting. diff --git a/crates/irtt-stats/src/core.rs b/crates/irtt-stats/src/core.rs index e3dd5de..c812ec9 100644 --- a/crates/irtt-stats/src/core.rs +++ b/crates/irtt-stats/src/core.rs @@ -234,7 +234,7 @@ impl CoreStats { Snapshot { events: self.events, packets, - loss: loss_stats(packets), + loss: loss_stats(packets, packets.server_packets_received), send_call: self.send_call.snapshot(), timer_error: self.timer_error.snapshot(), rtt: self.rtt_stats(), diff --git a/crates/irtt-stats/src/lib.rs b/crates/irtt-stats/src/lib.rs index 824bcfc..6c566ae 100644 --- a/crates/irtt-stats/src/lib.rs +++ b/crates/irtt-stats/src/lib.rs @@ -599,8 +599,9 @@ pub struct PacketCounts { pub bytes_received: u64, /// Highest cumulative server-reported packets received, when available. /// - /// This is the only server-reported input to the directional loss estimates - /// in [`LossStats`]. + /// Rolling snapshots also preserve this raw cumulative value from retained + /// replies. Their directional loss uses an interval-relative increase + /// instead; see [`LossStats`] for baseline and availability semantics. pub server_packets_received: Option, /// Raw server-reported receive window from the most recently processed /// reply that carried one, when available. diff --git a/crates/irtt-stats/src/loss.rs b/crates/irtt-stats/src/loss.rs index 5740dd3..8f886e4 100644 --- a/crates/irtt-stats/src/loss.rs +++ b/crates/irtt-stats/src/loss.rs @@ -8,9 +8,15 @@ use crate::PacketCounts; /// server-reported counts exceed local expectations. /// /// The directional fields are derived from -/// [`PacketCounts::server_packets_received`] and are `None` when the server -/// reported no receive count. [`PacketCounts::server_received_window`] is never -/// used to derive loss. +/// [`PacketCounts::server_packets_received`] for cumulative snapshots. Rolling +/// snapshots use the increase in the highest observed count over the retained +/// packet-event arrival interval, with the preceding observation as baseline +/// (zero at the start of the collector's history). They are `None` without a +/// known baseline/endpoint or when time filtering leaves an interior packet gap. +/// The percentage fields are `0.0` when the corresponding estimate is unavailable. +/// These remain estimates: outstanding sends and replies crossing a window +/// boundary can produce signed values. [`PacketCounts::server_received_window`] +/// is never used to derive loss. pub struct LossStats { /// Locally inferred total lost packets. pub lost_packets: u64, @@ -30,7 +36,7 @@ pub struct LossStats { pub late_packets_percent: f64, } -pub(crate) fn loss_stats(packets: PacketCounts) -> LossStats { +pub(crate) fn loss_stats(packets: PacketCounts, server_received: Option) -> LossStats { let lost = packets.packets_sent.saturating_sub(packets.unique_replies); let packet_loss_percent = if packets.packets_sent == 0 { 0.0 @@ -45,7 +51,7 @@ pub(crate) fn loss_stats(packets: PacketCounts) -> LossStats { upstream_loss_percent, downstream_loss_packets, downstream_loss_percent, - ) = if let Some(server_received) = packets.server_packets_received { + ) = if let Some(server_received) = server_received { let upstream = i128::from(packets.packets_sent) - i128::from(server_received); let downstream = i128::from(server_received) - i128::from(packets.packets_received); let upstream_percent = if packets.packets_sent == 0 { diff --git a/crates/irtt-stats/src/retention.rs b/crates/irtt-stats/src/retention.rs index cfeb8df..3331f14 100644 --- a/crates/irtt-stats/src/retention.rs +++ b/crates/irtt-stats/src/retention.rs @@ -9,7 +9,8 @@ use std::mem::size_of; use crate::core::CONTINUOUS_SEQUENCE_LIMIT; use crate::ipdv::IpdvSample; -use crate::normalization::{ReplySample, StatsEvent}; +use crate::normalization::ReplySample; +use crate::rolling::RollingEvent; use crate::{SampleMode, StatsConfig}; /// Capacity headroom applied to the live element bytes. @@ -94,8 +95,8 @@ fn rolling_count_bytes(config: &StatsConfig, probe_count: u64) -> u64 { retained.saturating_mul(rolling_bytes_per_event()) } -/// A retained event is the enum itself plus, for a unique reply, the boxed -/// reply sample it owns. +/// A retained event includes its directional baseline plus, for a unique reply, +/// the boxed reply sample it owns. fn rolling_bytes_per_event() -> u64 { - (size_of::() + size_of::()) as u64 + (size_of::() + size_of::()) as u64 } diff --git a/crates/irtt-stats/src/rolling.rs b/crates/irtt-stats/src/rolling.rs index bc58055..adf8a8a 100644 --- a/crates/irtt-stats/src/rolling.rs +++ b/crates/irtt-stats/src/rolling.rs @@ -6,20 +6,58 @@ use std::{ use measurement_stats::CountWindow; use crate::{ - core::CoreStats, normalization::StatsEvent, LateReplyMode, SampleMode, Snapshot, StatsConfig, + core::CoreStats, loss::loss_stats, normalization::StatsEvent, LateReplyMode, SampleMode, + Snapshot, StatsConfig, }; type ArrivalKey = (Instant, usize); +// Capture the observed cumulative position before eviction can discard it. +// Packet ordinals also reveal holes left by time filtering backdated events. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +struct DirectionalPosition { + packet_events: u128, + server_received: Option, +} + +impl DirectionalPosition { + fn advance(&mut self, event: &StatsEvent) -> bool { + match event { + StatsEvent::UniqueReply { sample, .. } => { + if let Some(count) = sample.received_count { + let count = u64::from(count); + self.server_received = Some( + self.server_received + .map_or(count, |current| current.max(count)), + ); + } + } + StatsEvent::Sent { .. } + | StatsEvent::DuplicateReply { .. } + | StatsEvent::UntrackedLate { .. } => {} + StatsEvent::Loss { .. } | StatsEvent::Warning { .. } => return false, + } + self.packet_events += 1; + true + } +} + +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct RollingEvent { + event: StatsEvent, + before: DirectionalPosition, +} + #[derive(Debug, Clone)] pub(crate) struct RollingEvents { time_limit: Option, - count_events: Option>, + count_events: Option>, // Both indexes contain exactly the live events; expiry removes from both. - time_events: Option>, + time_events: Option>, time_expiry: BTreeSet<(Instant, ArrivalKey)>, time_anchor: Option, late_replies: LateReplyMode, + directional_position: DirectionalPosition, } impl PartialEq for RollingEvents { @@ -34,6 +72,7 @@ impl PartialEq for RollingEvents { && self.count_events == other.count_events && self.time_anchor == other.time_anchor && self.late_replies == other.late_replies + && self.directional_position == other.directional_position && same_time_events } } @@ -47,16 +86,24 @@ impl RollingEvents { time_expiry: BTreeSet::new(), time_anchor: None, late_replies: config.late_replies, + directional_position: DirectionalPosition::default(), } } pub(crate) fn push(&mut self, event: StatsEvent) { + if self.count_events.is_none() && self.time_events.is_none() { + return; + } + let at = event.at(); + let before = self.directional_position; + self.directional_position.advance(&event); + let event = RollingEvent { event, before }; if let Some(window) = self.count_events.as_mut() { window.push(event.clone()); } if let (Some(duration), Some(window)) = (self.time_limit, self.time_events.as_mut()) { - let anchor = self.time_anchor.map_or(event.at(), |at| at.max(event.at())); + let anchor = self.time_anchor.map_or(at, |previous| previous.max(at)); self.time_anchor = Some(anchor); let cutoff = anchor.checked_sub(duration); if let Some(cutoff) = cutoff { @@ -65,7 +112,7 @@ impl RollingEvents { let (_, arrival) = self.time_expiry.pop_first().unwrap(); window.remove(&arrival); } - if event.at() < cutoff { + if at < cutoff { return; } } @@ -73,7 +120,7 @@ impl RollingEvents { // 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)); + self.time_expiry.insert((at, arrival)); window.insert(arrival, event); } } @@ -92,16 +139,39 @@ impl RollingEvents { } /// Recompute a window snapshot under the same normalized semantics as the -/// cumulative collector, including its late-reply measurement policy. +/// cumulative collector, except directional loss uses an interval-relative +/// server count. Raw server counters and all other metrics still replay normally. fn snapshot_window<'a>( - events: impl Iterator, + events: impl Iterator, late_replies: LateReplyMode, ) -> Snapshot { let mut core = CoreStats::new(SampleMode::RunningOnly, late_replies); + let mut start = None; + let mut end: Option = None; + let mut contiguous = true; for event in events { - core.apply(event.clone()); + let mut after = event.before; + if after.advance(&event.event) { + start.get_or_insert(event.before); + if end.is_some_and(|previous| previous.packet_events != event.before.packet_events) { + contiguous = false; + } + end = Some(after); + } + core.apply(event.event.clone()); } - core.snapshot() + let server_delta = start.zip(end).and_then(|(start, end)| { + if !contiguous { + return None; + } + let baseline = start + .server_received + .or_else(|| (start.packet_events == 0).then_some(0))?; + end.server_received.map(|count| count - baseline) + }); + let mut snapshot = core.snapshot(); + snapshot.loss = loss_stats(snapshot.packets, server_delta); + snapshot } #[cfg(test)] diff --git a/crates/irtt-stats/tests/rolling_loss.rs b/crates/irtt-stats/tests/rolling_loss.rs new file mode 100644 index 0000000..cd04912 --- /dev/null +++ b/crates/irtt-stats/tests/rolling_loss.rs @@ -0,0 +1,281 @@ +use std::time::{Duration, Instant, UNIX_EPOCH}; + +use irtt_client::{ + ClientEvent, ClientTimestamp, PacketMeta, ReceivedStatsSample, RttSample, SignedDuration, +}; +use irtt_stats::{LateReplyMode, Snapshot, StatsCollector, StatsConfig}; + +fn ts(base: Instant, ms: u64) -> ClientTimestamp { + ClientTimestamp { + mono: base + Duration::from_millis(ms), + wall: UNIX_EPOCH + Duration::from_millis(ms), + } +} + +fn sent(base: Instant, seq: u32) -> ClientEvent { + ClientEvent::EchoSent { + seq, + remote: "127.0.0.1:2112".parse().unwrap(), + scheduled_at: None, + sent_at: ts(base, u64::from(seq) * 10), + bytes: 32, + send_call: Duration::ZERO, + timer_error: None, + } +} + +fn reply(base: Instant, seq: u32, count: u32) -> ClientEvent { + ClientEvent::EchoReply { + seq, + remote: "127.0.0.1:2112".parse().unwrap(), + sent_at: ts(base, u64::from(seq) * 10), + received_at: ts(base, u64::from(seq) * 10 + 1), + rtt: RttSample { + raw: Duration::from_millis(1), + adjusted: None, + effective: SignedDuration::from_duration(Duration::from_millis(1)), + }, + server_timing: None, + one_way: None, + received_stats: Some(ReceivedStatsSample { + count: Some(count), + window: None, + }), + bytes: 64, + packet_meta: PacketMeta::default(), + } +} + +fn assert_directional(snapshot: &Snapshot, upstream: Option, downstream: Option) { + assert_eq!(snapshot.loss.upstream_loss_packets, upstream); + assert_eq!(snapshot.loss.downstream_loss_packets, downstream); +} + +#[test] +fn lossless_count_and_time_windows_use_the_server_counter_increase() { + let base = Instant::now(); + let mut collector = StatsCollector::new(StatsConfig { + rolling_count: Some(2), + rolling_time: Some(Duration::from_millis(2)), + ..StatsConfig::continuous() + }); + for seq in 0..100 { + collector.process(&sent(base, seq)); + collector.process(&reply(base, seq, seq + 1)); + for rolling in [collector.rolling_count(), collector.rolling_time()] { + let rolling = rolling.unwrap(); + assert_eq!(rolling.packets.packets_sent, 1); + assert_eq!(rolling.packets.packets_received, 1); + // Preserve the raw cumulative field, independently of the loss delta. + assert_eq!( + rolling.packets.server_packets_received, + Some(u64::from(seq + 1)) + ); + assert_directional(&rolling, Some(0), Some(0)); + assert_eq!(rolling.loss.upstream_loss_percent, 0.0); + assert_eq!(rolling.loss.downstream_loss_percent, 0.0); + } + } + let cumulative = collector.snapshot(); + assert_eq!(cumulative.packets.packets_sent, 100); + assert_directional(&cumulative, Some(0), Some(0)); +} + +#[test] +fn rolling_windows_distinguish_upstream_and_downstream_loss() { + let base = Instant::now(); + // After the first pair expires, three sends and two replies remain. + // A server delta of two means upstream loss; three means downstream loss. + // Four also protects signed estimates when server counts exceed local sends. + for (final_count, upstream, downstream, upstream_percent, downstream_percent) in [ + (3, 1, 0, 100.0 / 3.0, 0.0), + (4, 0, 1, 0.0, 100.0 / 3.0), + (5, -1, 2, -100.0 / 3.0, 50.0), + ] { + let mut collector = StatsCollector::new(StatsConfig { + rolling_count: Some(5), + rolling_time: Some(Duration::from_millis(25)), + ..StatsConfig::continuous() + }); + for event in [ + sent(base, 0), + reply(base, 0, 1), + sent(base, 1), + reply(base, 1, 2), + sent(base, 2), + sent(base, 3), + reply(base, 3, final_count), + ] { + collector.process(&event); + } + for rolling in [collector.rolling_count(), collector.rolling_time()] { + let rolling = rolling.unwrap(); + assert_eq!(rolling.packets.packets_sent, 3); + assert_eq!(rolling.packets.packets_received, 2); + assert_directional(&rolling, Some(upstream), Some(downstream)); + assert_eq!(rolling.loss.upstream_loss_percent, upstream_percent); + assert_eq!(rolling.loss.downstream_loss_percent, downstream_percent); + } + assert_directional(&collector.snapshot(), Some(upstream), Some(downstream)); + } +} + +#[test] +fn split_probe_boundaries_keep_signed_estimates_and_require_a_baseline() { + let base = Instant::now(); + let mut collector = StatsCollector::new(StatsConfig { + rolling_count: Some(1), + rolling_time: Some(Duration::ZERO), + ..StatsConfig::continuous() + }); + collector.process(&sent(base, 0)); + for rolling in [collector.rolling_count(), collector.rolling_time()] { + assert_directional(&rolling.unwrap(), None, None); + } + collector.process(&reply(base, 0, 1)); + // The first send has gone and no earlier server observation exists. + for rolling in [collector.rolling_count(), collector.rolling_time()] { + let rolling = rolling.unwrap(); + assert_eq!(rolling.packets.server_packets_received, Some(1)); + assert_directional(&rolling, None, None); + assert_eq!(rolling.loss.upstream_loss_percent, 0.0); + assert_eq!(rolling.loss.downstream_loss_percent, 0.0); + } + collector.process(&sent(base, 1)); + for rolling in [collector.rolling_count(), collector.rolling_time()] { + // No new server observation yet: one outstanding send, no counter increase. + assert_directional(&rolling.unwrap(), Some(1), Some(0)); + } + collector.process(&reply(base, 1, 2)); + for rolling in [collector.rolling_count(), collector.rolling_time()] { + let rolling = rolling.unwrap(); + assert_eq!(rolling.packets.packets_sent, 0); + assert_eq!(rolling.packets.packets_received, 1); + // The retained reply advances the count, but its send is outside the window. + assert_directional(&rolling, Some(-1), Some(0)); + } + assert_directional(&collector.snapshot(), Some(0), Some(0)); + + // A first observation after many evicted sends does not reveal how much + // of the cumulative counter belongs to the final retained send/reply pair. + let mut collector = StatsCollector::new(StatsConfig { + rolling_count: Some(2), + rolling_time: Some(Duration::from_millis(2)), + ..StatsConfig::continuous() + }); + for seq in 0..100 { + collector.process(&sent(base, seq)); + } + collector.process(&reply(base, 99, 100)); + for rolling in [collector.rolling_count(), collector.rolling_time()] { + let rolling = rolling.unwrap(); + assert_eq!(rolling.packets.packets_sent, 1); + assert_eq!(rolling.packets.packets_received, 1); + assert_eq!(rolling.packets.server_packets_received, Some(100)); + assert_directional(&rolling, None, None); + } + assert_directional(&collector.snapshot(), Some(0), Some(99)); +} + +#[test] +fn time_filtering_with_an_interior_packet_gap_has_no_directional_estimate() { + let base = Instant::now(); + let mut collector = StatsCollector::new(StatsConfig { + rolling_count: Some(3), + rolling_time: Some(Duration::from_millis(2)), + ..StatsConfig::continuous() + }); + collector.process(&sent(base, 0)); + collector.process(&reply(base, 0, 1)); + collector.process(&sent(base, 1)); + // Arrives after send 1, but the time window excludes its backdated receipt. + collector.process(&reply(base, 0, 2)); + collector.process(&reply(base, 1, 3)); + let rolling = collector.rolling_time().unwrap(); + assert_eq!(rolling.packets.packets_sent, 1); + assert_eq!(rolling.packets.packets_received, 1); + assert_directional(&rolling, None, None); + // Count retention has no gap, and the anomalous server increase stays signed. + assert_directional(&collector.rolling_count().unwrap(), Some(-1), Some(0)); +} + +#[test] +fn reordered_late_and_nonprimary_replies_preserve_directional_accounting() { + let base = Instant::now(); + for policy in [LateReplyMode::Measure, LateReplyMode::CountOnly] { + let mut collector = StatsCollector::new(StatsConfig { + rolling_count: Some(6), + rolling_time: Some(Duration::from_millis(25)), + late_replies: policy, + ..StatsConfig::continuous() + }); + collector.process(&sent(base, 0)); + collector.process(&reply(base, 0, 1)); + collector.process(&sent(base, 1)); + collector.process(&sent(base, 2)); + collector.process(&reply(base, 2, 3)); + let ClientEvent::EchoReply { + sent_at, + rtt, + received_stats, + remote, + .. + } = reply(base, 1, 2) + else { + unreachable!(); + }; + collector.process(&ClientEvent::LateReply { + seq: 1, + highest_seen: 2, + remote, + sent_at: Some(sent_at), + received_at: ts(base, 22), + rtt: Some(rtt), + server_timing: None, + one_way: None, + received_stats, + bytes: 64, + packet_meta: PacketMeta::default(), + }); + collector.process(&ClientEvent::DuplicateReply { + seq: 2, + remote, + received_at: ts(base, 23), + bytes: 64, + }); + // Even a supplied large counter on an untracked late reply is not a + // primary server measurement under the existing normalization policy. + collector.process(&ClientEvent::LateReply { + seq: 0, + highest_seen: 2, + remote, + sent_at: None, + received_at: ts(base, 30), + rtt: None, + server_timing: None, + one_way: None, + received_stats: Some(ReceivedStatsSample { + count: Some(100), + window: None, + }), + bytes: 64, + packet_meta: PacketMeta::default(), + }); + for rolling in [collector.rolling_count(), collector.rolling_time()] { + let rolling = rolling.unwrap(); + assert_eq!(rolling.packets.unique_replies, 2); + assert_eq!(rolling.packets.packets_received, 4); + assert_eq!(rolling.packets.server_packets_received, Some(3)); + assert_directional(&rolling, Some(0), Some(-2)); + assert_eq!( + rolling.rtt.primary.count, + if policy == LateReplyMode::Measure { + 2 + } else { + 1 + } + ); + } + assert_directional(&collector.snapshot(), Some(0), Some(-2)); + } +} diff --git a/crates/irtt-stats/tests/rolling_properties.rs b/crates/irtt-stats/tests/rolling_properties.rs index 46f06cd..8f4e509 100644 --- a/crates/irtt-stats/tests/rolling_properties.rs +++ b/crates/irtt-stats/tests/rolling_properties.rs @@ -1,5 +1,6 @@ //! Independent count and time windows must match a filtered replay through -//! StatsCollector, while cumulative accounting retains the complete history. +//! StatsCollector for ordinary metrics, while directional loss has an independent +//! interval model and cumulative accounting retains the complete history. //! Includes delayed timeout discovery and replies with backdated timestamps. use std::time::{Duration, Instant, UNIX_EPOCH}; @@ -212,6 +213,90 @@ fn replay(events: &[ClientEvent], late_replies: LateReplyMode) -> Snapshot { collector.snapshot() } +/// Retain ordinary metrics through replay, but derive directional loss directly +/// from the full public event history, including the discarded prefix. +fn rolling_reference( + history: &[(ClientEvent, u64)], + retain: impl Fn(usize, u64) -> bool, + late_replies: LateReplyMode, +) -> Snapshot { + let retained: Vec = history + .iter() + .enumerate() + .filter(|(index, (_, at))| retain(*index, *at)) + .map(|(_, (event, _))| event.clone()) + .collect(); + let mut snapshot = replay(&retained, late_replies); + let packets: Vec<_> = history + .iter() + .enumerate() + .filter(|(_, (event, _))| { + matches!( + event, + ClientEvent::EchoSent { .. } + | ClientEvent::EchoReply { .. } + | ClientEvent::LateReply { .. } + | ClientEvent::DuplicateReply { .. } + ) + }) + .collect(); + let selected: Vec<_> = packets + .iter() + .enumerate() + .filter(|(_, (index, (_, at)))| retain(*index, *at)) + .collect(); + let count = |event: &ClientEvent| match event { + ClientEvent::EchoReply { received_stats, .. } + | ClientEvent::LateReply { + sent_at: Some(_), + rtt: Some(_), + received_stats, + .. + } => received_stats.and_then(|stats| stats.count).map(u64::from), + _ => None, + }; + let delta = selected + .first() + .zip(selected.last()) + .and_then(|((first, _), (last, _))| { + // Timestamp filtering must not omit a packet between the endpoints. + if last - first + 1 != selected.len() { + return None; + } + let baseline = if *first == 0 { + Some(0) + } else { + packets[..*first] + .iter() + .filter_map(|(_, (event, _))| count(event)) + .max() + }?; + let endpoint = packets[..=*last] + .iter() + .filter_map(|(_, (event, _))| count(event)) + .max()?; + Some(endpoint - baseline) + }); + snapshot.loss.upstream_loss_packets = + delta.map(|delta| i128::from(snapshot.packets.packets_sent) - i128::from(delta)); + snapshot.loss.downstream_loss_packets = + delta.map(|delta| i128::from(delta) - i128::from(snapshot.packets.packets_received)); + snapshot.loss.upstream_loss_percent = if snapshot.packets.packets_sent == 0 { + 0.0 + } else { + snapshot.loss.upstream_loss_packets.map_or(0.0, |loss| { + 100.0 * loss as f64 / snapshot.packets.packets_sent as f64 + }) + }; + snapshot.loss.downstream_loss_percent = match delta { + Some(delta) if delta != 0 => { + 100.0 * snapshot.loss.downstream_loss_packets.unwrap() as f64 / delta as f64 + } + _ => 0.0, + }; + snapshot +} + #[test] fn collector_equality_ignores_expired_time_window_bookkeeping() { let base = Instant::now(); @@ -342,13 +427,9 @@ proptest! { // Count window reference: the last `count_limit` retained // events, ignoring the time bound entirely. let count_start = history.len().saturating_sub(count_limit); - let count_ref: Vec = history[count_start..] - .iter() - .map(|(event, _)| event.clone()) - .collect(); prop_assert_eq!( collector.rolling_count(), - Some(replay(&count_ref, late_replies)), + Some(rolling_reference(&history, |index, _| index >= count_start, late_replies)), "count window diverged from the count-only reference after op {}", idx ); @@ -358,14 +439,9 @@ proptest! { // 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() - .filter(|(_, at_ms)| cutoff.is_none_or(|cutoff| *at_ms >= cutoff)) - .map(|(event, _)| event.clone()) - .collect(); prop_assert_eq!( collector.rolling_time(), - Some(replay(&time_ref, late_replies)), + Some(rolling_reference(&history, |_, at_ms| cutoff.is_none_or(|cutoff| at_ms >= cutoff), late_replies)), "time window diverged from the time-only reference after op {}", idx ); } else { From a45396c2360f235938afc23d671e7249453b591f Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 4 Oct 2026 13:12:46 +0700 Subject: [PATCH 2/2] fix(stats): reject stale and discontinuous rolling counter baselines --- crates/irtt-stats/README.md | 40 ++++++--- crates/irtt-stats/src/loss.rs | 13 ++- crates/irtt-stats/src/rolling.rs | 35 ++++++-- crates/irtt-stats/tests/rolling_loss.rs | 86 ++++++++++++++++++- crates/irtt-stats/tests/rolling_properties.rs | 37 ++++++-- 5 files changed, 177 insertions(+), 34 deletions(-) diff --git a/crates/irtt-stats/README.md b/crates/irtt-stats/README.md index 5459473..b6e3709 100644 --- a/crates/irtt-stats/README.md +++ b/crates/irtt-stats/README.md @@ -39,25 +39,37 @@ 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. Rolling directional loss uses the increase in the highest observed server -receive count across the retained packet events in arrival order. Upstream loss -is window sends minus that increase; downstream loss is the increase minus all -window replies, including duplicates and untracked late replies, as in cumulative -accounting. The baseline is the highest count observed before the first retained -packet event, or zero if no earlier packet events existed. Each retained event -keeps this prior position, so eviction cannot erase a known baseline. Reordered -counts do not move the observation backwards; only matched unique replies supply -counter observations, regardless of the late-reply timing policy. +receive count within one observation segment across the retained packet events +in arrival order. Upstream loss is window sends minus that increase; downstream +loss is the increase minus all window replies, including duplicates and untracked +late replies, as in cumulative accounting. The baseline must be supplied by the +packet event immediately before +the first retained packet event, or is zero if no earlier packet events existed. +It must also equal the highest count observed in that segment. Intervening sends +or replies without a current count make an older baseline stale: its next increase +may cover packets outside the window. Each retained event keeps the count's observation +ordinal, so eviction cannot erase this distinction. Reordered lower counts do not +refresh the baseline; only matched unique replies supply counter observations, +regardless of the late-reply timing policy. Both directional packet estimates are `None` if the baseline or endpoint is -unknown, or if time filtering excludes a packet event between retained packet -events. Their percentage fields use the existing unavailable value of `0.0`. -Loss/warning events do not affect this interval. A boundary between a send and -its reply can still produce signed estimates, and outstanding sends retain the -existing provisional-loss interpretation. These are changes in observed -accounting, not exact network transit loss for probes sent in a wall-clock period. +unknown or stale, or if time filtering excludes a packet event between retained +packet events. Their percentage fields use the existing unavailable value of `0.0`. +Loss/warning events do not affect this interval. Signed estimates remain possible +when server counts exceed local expectations or extra replies arrive, and +outstanding sends retain the existing provisional-loss interpretation. These are +changes in observed accounting, not exact network transit loss for probes sent +in a wall-clock period. Raw `server_packets_received`, ordinary packet counts, and cumulative loss retain their existing meanings. +A counter jump of at least half the 32-bit range is ambiguous between counter +wrap and a very old reordered reply from the other side of a wrap. It starts a +new observation segment; windows crossing the discontinuity have no directional +estimate. Later windows can recover using a fresh baseline within that segment. +The collector does not infer an epoch from incomplete arrival history. Smaller +regressions retain the existing maximum-count interpretation for reordered replies. + Exact snapshots sort a temporary copy of each metric's retained samples, adding O(n) scratch memory and O(n log n) computation for that metric. The retention estimate is not peak snapshot or allocator accounting. diff --git a/crates/irtt-stats/src/loss.rs b/crates/irtt-stats/src/loss.rs index 8f886e4..6c9f385 100644 --- a/crates/irtt-stats/src/loss.rs +++ b/crates/irtt-stats/src/loss.rs @@ -10,12 +10,17 @@ use crate::PacketCounts; /// The directional fields are derived from /// [`PacketCounts::server_packets_received`] for cumulative snapshots. Rolling /// snapshots use the increase in the highest observed count over the retained -/// packet-event arrival interval, with the preceding observation as baseline +/// packet-event arrival interval within one observation segment. The baseline +/// must be that segment's highest count, reported by the preceding packet event /// (zero at the start of the collector's history). They are `None` without a -/// known baseline/endpoint or when time filtering leaves an interior packet gap. +/// current baseline or known endpoint, or when time filtering leaves an interior +/// packet gap. A jump of at least half the +/// 32-bit counter range starts a new observation segment: wrap and old replies +/// are ambiguous. Windows crossing that discontinuity have no directional loss; +/// later windows can use a fresh baseline within the new segment. /// The percentage fields are `0.0` when the corresponding estimate is unavailable. -/// These remain estimates: outstanding sends and replies crossing a window -/// boundary can produce signed values. [`PacketCounts::server_received_window`] +/// These remain estimates: outstanding sends, excess server counts, or extra +/// replies can produce signed values. [`PacketCounts::server_received_window`] /// is never used to derive loss. pub struct LossStats { /// Locally inferred total lost packets. diff --git a/crates/irtt-stats/src/rolling.rs b/crates/irtt-stats/src/rolling.rs index adf8a8a..7e459d2 100644 --- a/crates/irtt-stats/src/rolling.rs +++ b/crates/irtt-stats/src/rolling.rs @@ -18,6 +18,8 @@ type ArrivalKey = (Instant, usize); struct DirectionalPosition { packet_events: u128, server_received: Option, + server_observed_at: u128, + counter_reset_at: u128, } impl DirectionalPosition { @@ -26,10 +28,21 @@ impl DirectionalPosition { StatsEvent::UniqueReply { sample, .. } => { if let Some(count) = sample.received_count { let count = u64::from(count); - self.server_received = Some( - self.server_received - .map_or(count, |current| current.max(count)), - ); + // A jump across half the 32-bit range can be a wrap or an + // old reply from the other side of a wrap. Start a new + // observation segment without guessing an epoch. Windows + // spanning the discontinuity have no directional estimate. + if self + .server_received + .is_some_and(|current| current.abs_diff(count) >= (1_u64 << 31)) + { + self.server_received = None; + self.counter_reset_at = self.packet_events + 1; + } + if self.server_received.is_none_or(|current| count >= current) { + self.server_received = Some(count); + self.server_observed_at = self.packet_events + 1; + } } } StatsEvent::Sent { .. } @@ -161,12 +174,18 @@ fn snapshot_window<'a>( core.apply(event.event.clone()); } let server_delta = start.zip(end).and_then(|(start, end)| { - if !contiguous { + if !contiguous || start.counter_reset_at != end.counter_reset_at { return None; } - let baseline = start - .server_received - .or_else(|| (start.packet_events == 0).then_some(0))?; + let baseline = if start.packet_events == 0 { + 0 + } else if start.server_observed_at == start.packet_events { + start.server_received? + } else { + // Packet events after the observation but outside the window can + // have advanced the server count by an unknown amount. + return None; + }; end.server_received.map(|count| count - baseline) }); let mut snapshot = core.snapshot(); diff --git a/crates/irtt-stats/tests/rolling_loss.rs b/crates/irtt-stats/tests/rolling_loss.rs index cd04912..3de8d6b 100644 --- a/crates/irtt-stats/tests/rolling_loss.rs +++ b/crates/irtt-stats/tests/rolling_loss.rs @@ -121,7 +121,87 @@ fn rolling_windows_distinguish_upstream_and_downstream_loss() { } #[test] -fn split_probe_boundaries_keep_signed_estimates_and_require_a_baseline() { +fn rolling_baselines_cannot_include_unobserved_evicted_packets() { + let base = Instant::now(); + let mut collector = StatsCollector::new(StatsConfig { + rolling_count: Some(2), + rolling_time: Some(Duration::from_millis(2)), + ..StatsConfig::continuous() + }); + for event in [ + sent(base, 0), + reply(base, 0, 1), + sent(base, 1), + sent(base, 2), + sent(base, 3), + reply(base, 3, 4), + ] { + collector.process(&event); + } + for rolling in [collector.rolling_count(), collector.rolling_time()] { + let rolling = rolling.unwrap(); + assert_eq!(rolling.packets.packets_sent, 1); + assert_eq!(rolling.packets.packets_received, 1); + assert_eq!(rolling.packets.server_packets_received, Some(4)); + assert_directional(&rolling, None, None); + } + assert_directional(&collector.snapshot(), Some(0), Some(2)); + + // The next pair has an observation at its boundary, so estimates recover. + collector.process(&sent(base, 4)); + collector.process(&reply(base, 4, 5)); + for rolling in [collector.rolling_count(), collector.rolling_time()] { + assert_directional(&rolling.unwrap(), Some(0), Some(0)); + } +} + +#[test] +fn rolling_counter_discontinuities_are_unavailable_until_a_fresh_interval() { + let base = Instant::now(); + let mut collector = StatsCollector::new(StatsConfig { + rolling_count: Some(2), + rolling_time: Some(Duration::from_millis(2)), + ..StatsConfig::continuous() + }); + // Also cover an ambiguous high observation after wrap, followed by a return + // to post-wrap counts. Neither jump may be treated as a precise delta. + let counts = [u32::MAX - 1, u32::MAX, 0, 1, u32::MAX - 2, 2, 3]; + for (seq, count) in counts.into_iter().enumerate() { + let seq = u32::try_from(seq).unwrap(); + collector.process(&sent(base, seq)); + collector.process(&reply(base, seq, count)); + if seq == 0 { + continue; + } + for rolling in [collector.rolling_count(), collector.rolling_time()] { + let rolling = rolling.unwrap(); + if matches!(seq, 1 | 3 | 6) { + assert_directional(&rolling, Some(0), Some(0)); + } else { + assert_directional(&rolling, None, None); + assert_eq!(rolling.loss.upstream_loss_percent, 0.0); + assert_eq!(rolling.loss.downstream_loss_percent, 0.0); + assert_eq!( + rolling.packets.server_packets_received, + Some(u64::from(count)) + ); + } + } + } + let cumulative = collector.snapshot(); + assert_eq!( + cumulative.packets.server_packets_received, + Some(u64::from(u32::MAX)) + ); + assert_directional( + &cumulative, + Some(7 - i128::from(u32::MAX)), + Some(i128::from(u32::MAX) - 7), + ); +} + +#[test] +fn split_probe_boundaries_require_a_current_server_baseline() { let base = Instant::now(); let mut collector = StatsCollector::new(StatsConfig { rolling_count: Some(1), @@ -151,8 +231,8 @@ fn split_probe_boundaries_keep_signed_estimates_and_require_a_baseline() { let rolling = rolling.unwrap(); assert_eq!(rolling.packets.packets_sent, 0); assert_eq!(rolling.packets.packets_received, 1); - // The retained reply advances the count, but its send is outside the window. - assert_directional(&rolling, Some(-1), Some(0)); + // The evicted send makes the earlier observation stale at this boundary. + assert_directional(&rolling, None, None); } assert_directional(&collector.snapshot(), Some(0), Some(0)); diff --git a/crates/irtt-stats/tests/rolling_properties.rs b/crates/irtt-stats/tests/rolling_properties.rs index 8f4e509..9fc531b 100644 --- a/crates/irtt-stats/tests/rolling_properties.rs +++ b/crates/irtt-stats/tests/rolling_properties.rs @@ -263,18 +263,45 @@ fn rolling_reference( if last - first + 1 != selected.len() { return None; } + // Partition the full observed history at ambiguous discontinuities. + // This independent oracle keeps observation lists for each segment + // rather than relying on the collector's retained metadata. + let mut observations = Vec::new(); + let mut segment_start = 0; + for (index, (_, (event, _))) in packets[..=*last].iter().enumerate() { + if let Some(count) = count(event) { + if observations + .iter() + .max() + .is_some_and(|highest: &u64| highest.abs_diff(count) >= (1_u64 << 31)) + { + segment_start = index + 1; + observations.clear(); + } + observations.push(count); + } + } + if segment_start > *first { + return None; + } let baseline = if *first == 0 { Some(0) } else { - packets[..*first] + // The immediately preceding packet must itself report the + // highest count. Earlier observations cannot bound this window. + let (_, (previous_event, _)) = packets[*first - 1]; + let previous = count(previous_event)?; + if packets[segment_start.saturating_sub(1)..*first - 1] .iter() .filter_map(|(_, (event, _))| count(event)) .max() + .is_some_and(|highest| highest > previous) + { + return None; + } + Some(previous) }?; - let endpoint = packets[..=*last] - .iter() - .filter_map(|(_, (event, _))| count(event)) - .max()?; + let endpoint = observations.into_iter().max()?; Some(endpoint - baseline) }); snapshot.loss.upstream_loss_packets =