Merge branch 'fix/rolling-directional-loss'

crates/irtt-stats/README.md

@@ -38,6 +38,38 @@ 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 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 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.

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(),

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<u64>,
     /// Raw server-reported receive window from the most recently processed
     /// reply that carried one, when available.

crates/irtt-stats/src/loss.rs

@@ -8,9 +8,20 @@ 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 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
+/// 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, 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.
     pub lost_packets: u64,
@@ -30,7 +41,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<u64>) -> LossStats {
     let lost = packets.packets_sent.saturating_sub(packets.unique_replies);
     let packet_loss_percent = if packets.packets_sent == 0 {
         0.0
@@ -45,7 +56,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 {

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::<StatsEvent>() + size_of::<ReplySample>()) as u64
+    (size_of::<RollingEvent>() + size_of::<ReplySample>()) as u64
 }

crates/irtt-stats/src/rolling.rs

@@ -6,20 +6,71 @@ 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<u64>,
+    server_observed_at: u128,
+    counter_reset_at: u128,
+}
+
+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);
+                    // 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 { .. }
+            | 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<Duration>,
-    count_events: Option<CountWindow<StatsEvent>>,
+    count_events: Option<CountWindow<RollingEvent>>,
     // Both indexes contain exactly the live events; expiry removes from both.
-    time_events: Option<BTreeMap<ArrivalKey, StatsEvent>>,
+    time_events: Option<BTreeMap<ArrivalKey, RollingEvent>>,
     time_expiry: BTreeSet<(Instant, ArrivalKey)>,
     time_anchor: Option<Instant>,
     late_replies: LateReplyMode,
+    directional_position: DirectionalPosition,
 }
 
 impl PartialEq for RollingEvents {
@@ -34,6 +85,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 +99,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 +125,7 @@ impl RollingEvents {
                     let (_, arrival) = self.time_expiry.pop_first().unwrap();
                     window.remove(&arrival);
                 }
-                if event.at() < cutoff {
+                if at < cutoff {
                     return;
                 }
             }
@@ -73,7 +133,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 +152,45 @@ 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<Item = &'a StatsEvent>,
+    events: impl Iterator<Item = &'a RollingEvent>,
     late_replies: LateReplyMode,
 ) -> Snapshot {
     let mut core = CoreStats::new(SampleMode::RunningOnly, late_replies);
+    let mut start = None;
+    let mut end: Option<DirectionalPosition> = 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 || start.counter_reset_at != end.counter_reset_at {
+            return None;
+        }
+        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();
+    snapshot.loss = loss_stats(snapshot.packets, server_delta);
+    snapshot
 }
 
 #[cfg(test)]

crates/irtt-stats/tests/rolling_loss.rs

@@ -0,0 +1,361 @@
+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<i128>, downstream: Option<i128>) {
+    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 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),
+        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 evicted send makes the earlier observation stale at this boundary.
+        assert_directional(&rolling, None, None);
+    }
+    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));
+    }
+}

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,117 @@ 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<ClientEvent> = 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;
+            }
+            // 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 {
+                // 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 = observations.into_iter().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 +454,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<ClientEvent> = 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 +466,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<ClientEvent> = 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 {