From a45396c2360f235938afc23d671e7249453b591f Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 4 Oct 2026 13:12:46 +0700 Subject: [PATCH] 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 =