fix(stats): reject stale and discontinuous rolling counter baselines

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
+receive count within one observation segment across the retained packet events
-is window sends minus that increase; downstream loss is the increase minus all
+in arrival order. Upstream loss is window sends minus that increase; downstream
-window replies, including duplicates and untracked late replies, as in cumulative
+loss is the increase minus all window replies, including duplicates and untracked
-accounting. The baseline is the highest count observed before the first retained
+late replies, as in cumulative accounting. The baseline must be supplied by the
-packet event, or zero if no earlier packet events existed. Each retained event
+packet event immediately before
-keeps this prior position, so eviction cannot erase a known baseline. Reordered
+the first retained packet event, or is zero if no earlier packet events existed.
-counts do not move the observation backwards; only matched unique replies supply
+It must also equal the highest count observed in that segment. Intervening sends
-counter observations, regardless of the late-reply timing policy.
+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
+unknown or stale, or if time filtering excludes a packet event between retained
-events. Their percentage fields use the existing unavailable value of `0.0`.
+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
+Loss/warning events do not affect this interval. Signed estimates remain possible
-its reply can still produce signed estimates, and outstanding sends retain the
+when server counts exceed local expectations or extra replies arrive, and
-existing provisional-loss interpretation. These are changes in observed
+outstanding sends retain the existing provisional-loss interpretation. These are
-accounting, not exact network transit loss for probes sent in a wall-clock period.
+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.

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
+/// These remain estimates: outstanding sends, excess server counts, or extra
-/// boundary can produce signed values. [`PacketCounts::server_received_window`]
+/// replies can produce signed values. [`PacketCounts::server_received_window`]
 /// is never used to derive loss.
 pub struct LossStats {
     /// Locally inferred total lost packets.

crates/irtt-stats/src/rolling.rs

@@ -18,6 +18,8 @@ type ArrivalKey = (Instant, usize);
 struct DirectionalPosition {
     packet_events: u128,
     server_received: Option<u64>,
+    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(
+                    // A jump across half the 32-bit range can be a wrap or an
-                        self.server_received
+                    // old reply from the other side of a wrap. Start a new
-                            .map_or(count, |current| current.max(count)),
+                    // 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
+        let baseline = if start.packet_events == 0 {
-            .server_received
+            0
-            .or_else(|| (start.packet_events == 0).then_some(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();

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.
+        // The evicted send makes the earlier observation stale at this boundary.
-        assert_directional(&rolling, Some(-1), Some(0));
+        assert_directional(&rolling, None, None);
     }
     assert_directional(&collector.snapshot(), Some(0), Some(0));
 

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]
+            let endpoint = observations.into_iter().max()?;
-                .iter()
-                .filter_map(|(_, (event, _))| count(event))
-                .max()?;
             Some(endpoint - baseline)
         });
     snapshot.loss.upstream_loss_packets =