diff --git a/crates/irtt-cli/src/cmd/tui/ui.rs b/crates/irtt-cli/src/cmd/tui/ui.rs index d468bac..72c8bcd 100644 --- a/crates/irtt-cli/src/cmd/tui/ui.rs +++ b/crates/irtt-cli/src/cmd/tui/ui.rs @@ -258,7 +258,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, @@ -616,43 +616,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 } } @@ -2267,12 +2245,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() @@ -2283,6 +2264,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 = diff --git a/crates/irtt-stats/src/core.rs b/crates/irtt-stats/src/core.rs index e5f076e..6ca8b9d 100644 --- a/crates/irtt-stats/src/core.rs +++ b/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); diff --git a/crates/irtt-stats/src/lib.rs b/crates/irtt-stats/src/lib.rs index b89809a..d76152b 100644 --- a/crates/irtt-stats/src/lib.rs +++ b/crates/irtt-stats/src/lib.rs @@ -36,6 +36,8 @@ pub struct StatsConfig { pub rolling_count: Option, /// Time span of recent normalized events retained for time-based rolling snapshots. pub rolling_time: Option, + /// 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), } } diff --git a/crates/irtt-stats/src/rolling.rs b/crates/irtt-stats/src/rolling.rs index 3c2fd82..0f03361 100644 --- a/crates/irtt-stats/src/rolling.rs +++ b/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, count_events: Option>, time_events: Option>, + 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 { - 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 { - 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) -> 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, late_replies: LateReplyMode) -> Snapshot { + let mut core = CoreStats::new(SampleMode::RunningOnly, late_replies); for event in events { core.apply(event.clone()); } diff --git a/crates/irtt-stats/tests/stats.rs b/crates/irtt-stats/tests/stats.rs index a4a92ed..1af22cc 100644 --- a/crates/irtt-stats/tests/stats.rs +++ b/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); +}