Merge pull request #92 from Lochnair/refactor/stats-late-reply-policy

refactor(stats): make late-reply sampling explicit

crates/irtt-cli/src/cmd/tui/ui.rs

@@ -264,7 +264,7 @@ impl TuiState {
             let Some(target) = self.targets.get_mut(target_idx) else {
                 return;
             };
-            process_tui_stats(event, &mut target.stats);
+            target.stats.process(event);
             match event {
                 ClientEvent::SessionStarted {
                     remote,
@@ -622,43 +622,21 @@ impl GroupPacingArg {
     }
 }
 
+/// Statistics policy for the TUI.
+///
+/// The TUI treats late replies as diagnostics, so it counts a matched late
+/// reply without letting it contribute timing samples. That is an explicit
+/// statistics policy rather than something the TUI expresses by degrading the
+/// event it forwards.
 fn stats_config(continuous: bool) -> irtt_stats::StatsConfig {
-    if continuous {
+    let base = if continuous {
         irtt_stats::StatsConfig::continuous()
     } else {
         irtt_stats::StatsConfig::finite()
-    }
-}
-
-fn process_tui_stats(event: &ClientEvent, stats: &mut StatsCollector) {
-    if let ClientEvent::LateReply {
-        seq,
-        highest_seen,
-        remote,
-        received_at,
-        bytes,
-        packet_meta,
-        ..
-    } = event
-    {
-        // The TUI treats late replies as diagnostics, even when retained send
-        // metadata lets the client attach RTT fields to the event.
-        let late_counter_event = ClientEvent::LateReply {
-            seq: *seq,
-            highest_seen: *highest_seen,
-            remote: *remote,
-            sent_at: None,
-            received_at: *received_at,
-            rtt: None,
-            server_timing: None,
-            one_way: None,
-            received_stats: None,
-            bytes: *bytes,
-            packet_meta: *packet_meta,
     };
-        stats.process(&late_counter_event);
-    } else {
-        stats.process(event);
+    irtt_stats::StatsConfig {
+        late_replies: irtt_stats::LateReplyMode::CountOnly,
+        ..base
     }
 }
 
@@ -2243,12 +2221,15 @@ mod tests {
         assert!(target.graph_history.is_empty());
         let snapshot = target.stats.snapshot();
         assert_eq!(snapshot.events.duplicate_replies, 1);
-        assert_eq!(snapshot.events.late_unique_replies, 0);
-        assert_eq!(snapshot.events.untracked_late_replies, 1);
-        assert_eq!(snapshot.packets.unique_replies, 0);
+        // A matched late reply is a late unique reply. The TUI suppresses its
+        // measurements through the late-reply policy, not by reclassifying it.
+        assert_eq!(snapshot.events.late_unique_replies, 1);
+        assert_eq!(snapshot.events.untracked_late_replies, 0);
+        assert_eq!(snapshot.packets.unique_replies, 1);
         assert_eq!(snapshot.packets.duplicates, 1);
         assert_eq!(snapshot.packets.late_packets, 1);
         assert_eq!(snapshot.rtt.primary.count, 0);
+        assert_eq!(snapshot.ipdv.round_trip.count, 0);
         assert!(state
             .recent_events
             .iter()
@@ -2259,6 +2240,32 @@ mod tests {
             .any(|event| event.contains("late seq=8")));
     }
 
+    #[test]
+    fn untracked_late_replies_stay_untracked_in_the_tui() {
+        let mut state = TuiState::default();
+        let target = target_instance("target");
+        let late = ClientEvent::LateReply {
+            seq: 8,
+            highest_seen: 9,
+            remote: remote(),
+            sent_at: None,
+            received_at: ts(Duration::from_secs(2)),
+            rtt: None,
+            server_timing: None,
+            one_way: None,
+            received_stats: None,
+            bytes: 64,
+            packet_meta: PacketMeta::default(),
+        };
+        state.process_target_event(&target, &late);
+
+        let snapshot = primary_target(&state).stats.snapshot();
+        assert_eq!(snapshot.events.untracked_late_replies, 1);
+        assert_eq!(snapshot.events.late_unique_replies, 0);
+        assert_eq!(snapshot.packets.unique_replies, 0);
+        assert_eq!(snapshot.packets.late_packets, 1);
+    }
+
     #[test]
     fn target_scoped_echo_reply_updates_only_that_target() {
         let mut state =

crates/irtt-stats/src/core.rs

@@ -5,8 +5,8 @@ use crate::loss::loss_stats;
 use crate::normalization::{ReplySample, StatsEvent};
 use crate::time_stats::TimeMetric;
 use crate::{
-    EventCounts, EventStatsUpdate, IpdvPairUpdate, IpdvStats, OneWayDelayStats, PacketCounts,
-    RttStats, SampleMode, ServerProcessingStats, Snapshot,
+    EventCounts, EventStatsUpdate, IpdvPairUpdate, IpdvStats, LateReplyMode, OneWayDelayStats,
+    PacketCounts, RttStats, SampleMode, ServerProcessingStats, Snapshot,
 };
 
 const CONTINUOUS_SEQUENCE_LIMIT: usize = 4096;
@@ -27,10 +27,11 @@ pub(crate) struct CoreStats {
     receive_delay: TimeMetric,
     server_processing: TimeMetric,
     ipdv_tracker: IpdvTracker,
+    late_replies: LateReplyMode,
 }
 
 impl CoreStats {
-    pub(crate) fn new(sample_mode: SampleMode) -> Self {
+    pub(crate) fn new(sample_mode: SampleMode, late_replies: LateReplyMode) -> Self {
         let sequence_limit = if sample_mode == SampleMode::Exact {
             None
         } else {
@@ -52,6 +53,7 @@ impl CoreStats {
             receive_delay: TimeMetric::new(sample_mode == SampleMode::Exact),
             server_processing: TimeMetric::new(false),
             ipdv_tracker: IpdvTracker::new(sequence_limit),
+            late_replies,
         }
     }
 
@@ -97,12 +99,19 @@ impl CoreStats {
     }
 
     fn apply_unique_reply(&mut self, is_late: bool, sample: ReplySample) -> EventStatsUpdate {
+        self.account_unique_reply(is_late, &sample);
+
+        // A matched late reply is always a late unique reply. Whether it is
+        // also allowed to measure is the configured policy, not a property of
+        // the event.
+        if is_late && self.late_replies == LateReplyMode::CountOnly {
+            return EventStatsUpdate::default();
+        }
+
         let mut update = EventStatsUpdate {
             contributed_sample: true,
             ..EventStatsUpdate::default()
         };
-
-        self.account_unique_reply(is_late, &sample);
         self.record_reply_metrics(&sample);
         update.ipdv_pairs = self.apply_ipdv_sample(sample.ipdv);
 

crates/irtt-stats/src/lib.rs

@@ -36,6 +36,8 @@ pub struct StatsConfig {
     pub rolling_count: Option<usize>,
     /// Time span of recent normalized events retained for time-based rolling snapshots.
     pub rolling_time: Option<Duration>,
+    /// Whether matched late replies contribute measurements.
+    pub late_replies: LateReplyMode,
 }
 
 impl StatsConfig {
@@ -49,6 +51,7 @@ impl StatsConfig {
             samples: SampleMode::Exact,
             rolling_count: None,
             rolling_time: None,
+            late_replies: LateReplyMode::Measure,
         }
     }
 
@@ -62,10 +65,34 @@ impl StatsConfig {
             samples: SampleMode::RunningOnly,
             rolling_count: None,
             rolling_time: None,
+            late_replies: LateReplyMode::Measure,
         }
     }
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+/// Whether a matched late reply contributes measurements.
+///
+/// This is a measurement policy, not a classification. A late reply that the
+/// client matched to retained send state is a late unique reply in both modes,
+/// and a late reply that could not be matched remains an untracked late reply
+/// in both modes. Only what a matched late reply is allowed to measure differs,
+/// so a caller that treats late replies as diagnostics does not have to degrade
+/// the event itself to suppress its timing contribution.
+pub enum LateReplyMode {
+    /// Matched late replies contribute timing, one-way, server-processing, and
+    /// IPDV measurements like any other unique reply.
+    #[default]
+    Measure,
+    /// Matched late replies are counted but measure nothing.
+    ///
+    /// Packet, byte, late, and server-reported receive counters still update,
+    /// so the reply stays visible as a late unique reply. Timing metrics,
+    /// one-way and server-processing measurements, and IPDV do not update, and
+    /// [`EventStatsUpdate::contributed_sample`] is `false`.
+    CountOnly,
+}
+
 impl Default for StatsConfig {
     fn default() -> Self {
         Self::finite()
@@ -95,7 +122,7 @@ impl StatsCollector {
     /// Creates a collector with the supplied configuration.
     pub fn new(config: StatsConfig) -> Self {
         Self {
-            cumulative: CoreStats::new(config.samples),
+            cumulative: CoreStats::new(config.samples, config.late_replies),
             rolling: RollingEvents::new(config),
         }
     }

crates/irtt-stats/src/rolling.rs

@@ -1,6 +1,8 @@
 use std::{collections::VecDeque, time::Duration};
 
-use crate::{core::CoreStats, normalization::StatsEvent, SampleMode, Snapshot, StatsConfig};
+use crate::{
+    core::CoreStats, normalization::StatsEvent, LateReplyMode, SampleMode, Snapshot, StatsConfig,
+};
 
 #[derive(Debug, Clone, PartialEq)]
 pub(crate) struct RollingEvents {
@@ -8,6 +10,7 @@ pub(crate) struct RollingEvents {
     time_limit: Option<Duration>,
     count_events: Option<VecDeque<StatsEvent>>,
     time_events: Option<VecDeque<StatsEvent>>,
+    late_replies: LateReplyMode,
 }
 
 impl RollingEvents {
@@ -17,6 +20,7 @@ impl RollingEvents {
             time_limit: config.rolling_time,
             count_events: config.rolling_count.map(|_| VecDeque::new()),
             time_events: config.rolling_time.map(|_| VecDeque::new()),
+            late_replies: config.late_replies,
         }
     }
 
@@ -40,16 +44,22 @@ impl RollingEvents {
     }
 
     pub(crate) fn count_snapshot(&self) -> Option<Snapshot> {
-        self.count_events.as_ref().map(snapshot_window)
+        self.count_events
+            .as_ref()
+            .map(|events| snapshot_window(events, self.late_replies))
     }
 
     pub(crate) fn time_snapshot(&self) -> Option<Snapshot> {
-        self.time_events.as_ref().map(snapshot_window)
+        self.time_events
+            .as_ref()
+            .map(|events| snapshot_window(events, self.late_replies))
     }
 }
 
-fn snapshot_window(events: &VecDeque<StatsEvent>) -> Snapshot {
-    let mut core = CoreStats::new(SampleMode::RunningOnly);
+/// Recompute a window snapshot under the same normalized semantics as the
+/// cumulative collector, including its late-reply measurement policy.
+fn snapshot_window(events: &VecDeque<StatsEvent>, late_replies: LateReplyMode) -> Snapshot {
+    let mut core = CoreStats::new(SampleMode::RunningOnly, late_replies);
     for event in events {
         core.apply(event.clone());
     }

crates/irtt-stats/tests/stats.rs

@@ -8,7 +8,9 @@ use common::{
 use irtt_client::{
     ClientEvent, ClientTimestamp, OneWayDelaySample, PacketMeta, RttSample, SignedDuration,
 };
-use irtt_stats::{EventStatsUpdate, IpdvPairUpdate, SampleMode, StatsCollector, StatsConfig};
+use irtt_stats::{
+    EventStatsUpdate, IpdvPairUpdate, LateReplyMode, SampleMode, StatsCollector, StatsConfig,
+};
 
 #[test]
 fn running_only_samples_avoid_finite_retention() {
@@ -342,6 +344,7 @@ fn rolling_count_eviction_recomputes_from_bounded_events() {
         samples: SampleMode::RunningOnly,
         rolling_count: Some(2),
         rolling_time: None,
+        late_replies: LateReplyMode::Measure,
     });
     collector.process(&sent(0, ts(0)));
     collector.process(&unadjusted_reply(0, 10));
@@ -359,6 +362,7 @@ fn rolling_time_eviction_uses_event_timestamps() {
         samples: SampleMode::RunningOnly,
         rolling_count: None,
         rolling_time: Some(Duration::from_millis(15)),
+        late_replies: LateReplyMode::Measure,
     });
     collector.process(&sent(0, ts(0)));
     collector.process(&sent(1, ts(10)));
@@ -420,3 +424,120 @@ fn directional_loss_uses_server_received_count_when_available() {
     assert_eq!(loss.packet_loss_percent, 50.0);
     assert_eq!(loss.upstream_loss_percent, 50.0);
 }
+
+fn count_only(base: StatsConfig) -> StatsConfig {
+    StatsConfig {
+        late_replies: LateReplyMode::CountOnly,
+        ..base
+    }
+}
+
+fn untracked_late_reply(seq: u32, received_ms: u64) -> ClientEvent {
+    ClientEvent::LateReply {
+        seq,
+        highest_seen: seq + 1,
+        remote: "127.0.0.1:2112".parse().unwrap(),
+        sent_at: None,
+        received_at: ts(received_ms),
+        rtt: None,
+        server_timing: None,
+        one_way: None,
+        received_stats: None,
+        bytes: 64,
+        packet_meta: PacketMeta::default(),
+    }
+}
+
+#[test]
+fn matched_late_reply_measures_under_the_default_policy() {
+    let mut collector = StatsCollector::new(StatsConfig::finite());
+    collector.process(&sent(0, ts(0)));
+    let update = collector.process(&unadjusted_late_reply(0, 10));
+
+    assert!(update.contributed_sample);
+
+    let snapshot = collector.snapshot();
+    assert_eq!(snapshot.events.late_unique_replies, 1);
+    assert_eq!(snapshot.events.untracked_late_replies, 0);
+    assert_eq!(snapshot.packets.unique_replies, 1);
+    assert_eq!(snapshot.packets.late_packets, 1);
+    assert_eq!(snapshot.packets.packets_received, 1);
+    assert_eq!(snapshot.packets.bytes_received, 64);
+    assert_eq!(snapshot.rtt.primary.count, 1);
+    assert_eq!(snapshot.server_processing.processing.count, 1);
+    assert_eq!(snapshot.one_way_delay.send_delay.count, 1);
+}
+
+#[test]
+fn matched_late_reply_counts_without_measuring_under_count_only() {
+    let mut collector = StatsCollector::new(count_only(StatsConfig::finite()));
+    collector.process(&sent(0, ts(0)));
+    let update = collector.process(&unadjusted_late_reply(0, 10));
+
+    assert!(!update.contributed_sample);
+    assert!(update.ipdv_pairs.is_empty());
+
+    let snapshot = collector.snapshot();
+    // Still a matched late unique reply, not a reclassified untracked one.
+    assert_eq!(snapshot.events.late_unique_replies, 1);
+    assert_eq!(snapshot.events.untracked_late_replies, 0);
+    assert_eq!(snapshot.packets.unique_replies, 1);
+    assert_eq!(snapshot.packets.late_packets, 1);
+    assert_eq!(snapshot.packets.packets_received, 1);
+    assert_eq!(snapshot.packets.bytes_received, 64);
+    // ...but it measures nothing.
+    assert_eq!(snapshot.rtt.primary.count, 0);
+    assert_eq!(snapshot.rtt.raw.count, 0);
+    assert_eq!(snapshot.server_processing.processing.count, 0);
+    assert_eq!(snapshot.one_way_delay.send_delay.count, 0);
+    assert_eq!(snapshot.one_way_delay.receive_delay.count, 0);
+    assert_eq!(snapshot.ipdv.round_trip.count, 0);
+}
+
+#[test]
+fn count_only_late_replies_do_not_complete_ipdv_pairs() {
+    let mut collector = StatsCollector::new(count_only(StatsConfig::finite()));
+    collector.process(&unadjusted_reply(0, 10));
+    let update = collector.process(&unadjusted_late_reply(1, 13));
+
+    assert!(update.ipdv_pairs.is_empty());
+    assert_eq!(collector.snapshot().ipdv.round_trip.count, 0);
+}
+
+#[test]
+fn untracked_late_replies_stay_untracked_in_both_policies() {
+    for config in [StatsConfig::finite(), count_only(StatsConfig::finite())] {
+        let mut collector = StatsCollector::new(config);
+        let update = collector.process(&untracked_late_reply(9, 100));
+
+        assert!(!update.contributed_sample);
+
+        let snapshot = collector.snapshot();
+        assert_eq!(snapshot.events.untracked_late_replies, 1);
+        assert_eq!(snapshot.events.late_unique_replies, 0);
+        assert_eq!(snapshot.packets.unique_replies, 0);
+        assert_eq!(snapshot.packets.late_packets, 1);
+        assert_eq!(snapshot.packets.packets_received, 1);
+        assert_eq!(snapshot.rtt.primary.count, 0);
+    }
+}
+
+#[test]
+fn rolling_snapshots_use_the_same_late_reply_policy() {
+    let mut collector = StatsCollector::new(StatsConfig {
+        rolling_count: Some(8),
+        rolling_time: None,
+        ..count_only(StatsConfig::continuous())
+    });
+    collector.process(&sent(0, ts(0)));
+    collector.process(&unadjusted_reply(0, 10));
+    collector.process(&unadjusted_late_reply(1, 10));
+
+    let rolling = collector.rolling_count().unwrap();
+    assert_eq!(rolling.events.late_unique_replies, 1);
+    assert_eq!(rolling.events.untracked_late_replies, 0);
+    assert_eq!(rolling.packets.unique_replies, 2);
+    assert_eq!(rolling.packets.late_packets, 1);
+    assert_eq!(rolling.rtt.primary.count, 1);
+    assert_eq!(rolling.ipdv.round_trip.count, 0);
+}