refactor(stats): make late-reply sampling explicit

The TUI suppressed late-reply timing by rebuilding the event with sent_at,
rtt, server_timing, one_way and received_stats erased. That leaked stats
policy into event construction and, as a side effect, reclassified a matched
late reply as an untracked late reply.

StatsConfig now carries an explicit LateReplyMode. Measure keeps today's
behavior for every existing caller. CountOnly keeps the reply a late unique
reply and keeps packet, byte and late counters updating, while suppressing
timing, one-way, server-processing and IPDV contributions and reporting
contributed_sample == false. Cumulative and rolling snapshots share the
policy. Genuinely untracked late replies stay untracked in both modes.

The TUI selects CountOnly and passes the original event straight to the
collector.
This commit is contained in:
Nils Andreas Svee committed 2026-08-15 01:17:46 +07:00
1 parent 90c04b6569
commit e0dec9ac5f
5 files changed
+223 -49

No files matched your search

+44 -37
View File
@@ -258,7 +258,7 @@ impl TuiState {
let Some(target) = self.targets.get_mut(target_idx) else { let Some(target) = self.targets.get_mut(target_idx) else {
return; return;
}; };
process_tui_stats(event, &mut target.stats); target.stats.process(event);
match event { match event {
ClientEvent::SessionStarted { ClientEvent::SessionStarted {
remote, 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 { fn stats_config(continuous: bool) -> irtt_stats::StatsConfig {
if continuous { let base = if continuous {
irtt_stats::StatsConfig::continuous() irtt_stats::StatsConfig::continuous()
} else { } else {
irtt_stats::StatsConfig::finite() irtt_stats::StatsConfig::finite()
} };
} irtt_stats::StatsConfig {
late_replies: irtt_stats::LateReplyMode::CountOnly,
fn process_tui_stats(event: &ClientEvent, stats: &mut StatsCollector) { ..base
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);
} }
} }
@@ -2267,12 +2245,15 @@ mod tests {
assert!(target.graph_history.is_empty()); assert!(target.graph_history.is_empty());
let snapshot = target.stats.snapshot(); let snapshot = target.stats.snapshot();
assert_eq!(snapshot.events.duplicate_replies, 1); assert_eq!(snapshot.events.duplicate_replies, 1);
assert_eq!(snapshot.events.late_unique_replies, 0); // A matched late reply is a late unique reply. The TUI suppresses its
assert_eq!(snapshot.events.untracked_late_replies, 1); // measurements through the late-reply policy, not by reclassifying it.
assert_eq!(snapshot.packets.unique_replies, 0); 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.duplicates, 1);
assert_eq!(snapshot.packets.late_packets, 1); assert_eq!(snapshot.packets.late_packets, 1);
assert_eq!(snapshot.rtt.primary.count, 0); assert_eq!(snapshot.rtt.primary.count, 0);
assert_eq!(snapshot.ipdv.round_trip.count, 0);
assert!(state assert!(state
.recent_events .recent_events
.iter() .iter()
@@ -2283,6 +2264,32 @@ mod tests {
.any(|event| event.contains("late seq=8"))); .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] #[test]
fn target_scoped_echo_reply_updates_only_that_target() { fn target_scoped_echo_reply_updates_only_that_target() {
let mut state = let mut state =
+14 -5
View File
@@ -5,8 +5,8 @@ use crate::loss::loss_stats;
use crate::normalization::{ReplySample, StatsEvent}; use crate::normalization::{ReplySample, StatsEvent};
use crate::time_stats::TimeMetric; use crate::time_stats::TimeMetric;
use crate::{ use crate::{
EventCounts, EventStatsUpdate, IpdvPairUpdate, IpdvStats, OneWayDelayStats, PacketCounts, EventCounts, EventStatsUpdate, IpdvPairUpdate, IpdvStats, LateReplyMode, OneWayDelayStats,
RttStats, SampleMode, ServerProcessingStats, Snapshot, PacketCounts, RttStats, SampleMode, ServerProcessingStats, Snapshot,
}; };
const CONTINUOUS_SEQUENCE_LIMIT: usize = 4096; const CONTINUOUS_SEQUENCE_LIMIT: usize = 4096;
@@ -27,10 +27,11 @@ pub(crate) struct CoreStats {
receive_delay: TimeMetric, receive_delay: TimeMetric,
server_processing: TimeMetric, server_processing: TimeMetric,
ipdv_tracker: IpdvTracker, ipdv_tracker: IpdvTracker,
late_replies: LateReplyMode,
} }
impl CoreStats { 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 { let sequence_limit = if sample_mode == SampleMode::Exact {
None None
} else { } else {
@@ -52,6 +53,7 @@ impl CoreStats {
receive_delay: TimeMetric::new(sample_mode == SampleMode::Exact), receive_delay: TimeMetric::new(sample_mode == SampleMode::Exact),
server_processing: TimeMetric::new(false), server_processing: TimeMetric::new(false),
ipdv_tracker: IpdvTracker::new(sequence_limit), 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 { 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 { let mut update = EventStatsUpdate {
contributed_sample: true, contributed_sample: true,
..EventStatsUpdate::default() ..EventStatsUpdate::default()
}; };
self.account_unique_reply(is_late, &sample);
self.record_reply_metrics(&sample); self.record_reply_metrics(&sample);
update.ipdv_pairs = self.apply_ipdv_sample(sample.ipdv); update.ipdv_pairs = self.apply_ipdv_sample(sample.ipdv);
+28 -1
View File
@@ -36,6 +36,8 @@ pub struct StatsConfig {
pub rolling_count: Option<usize>, pub rolling_count: Option<usize>,
/// Time span of recent normalized events retained for time-based rolling snapshots. /// Time span of recent normalized events retained for time-based rolling snapshots.
pub rolling_time: Option<Duration>, pub rolling_time: Option<Duration>,
/// Whether matched late replies contribute measurements.
pub late_replies: LateReplyMode,
} }
impl StatsConfig { impl StatsConfig {
@@ -49,6 +51,7 @@ impl StatsConfig {
samples: SampleMode::Exact, samples: SampleMode::Exact,
rolling_count: None, rolling_count: None,
rolling_time: None, rolling_time: None,
late_replies: LateReplyMode::Measure,
} }
} }
@@ -62,10 +65,34 @@ impl StatsConfig {
samples: SampleMode::RunningOnly, samples: SampleMode::RunningOnly,
rolling_count: None, rolling_count: None,
rolling_time: 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 { impl Default for StatsConfig {
fn default() -> Self { fn default() -> Self {
Self::finite() Self::finite()
@@ -95,7 +122,7 @@ impl StatsCollector {
/// Creates a collector with the supplied configuration. /// Creates a collector with the supplied configuration.
pub fn new(config: StatsConfig) -> Self { pub fn new(config: StatsConfig) -> Self {
Self { Self {
cumulative: CoreStats::new(config.samples), cumulative: CoreStats::new(config.samples, config.late_replies),
rolling: RollingEvents::new(config), rolling: RollingEvents::new(config),
} }
} }
+15 -5
View File
@@ -1,6 +1,8 @@
use std::{collections::VecDeque, time::Duration}; 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)] #[derive(Debug, Clone, PartialEq)]
pub(crate) struct RollingEvents { pub(crate) struct RollingEvents {
@@ -8,6 +10,7 @@ pub(crate) struct RollingEvents {
time_limit: Option<Duration>, time_limit: Option<Duration>,
count_events: Option<VecDeque<StatsEvent>>, count_events: Option<VecDeque<StatsEvent>>,
time_events: Option<VecDeque<StatsEvent>>, time_events: Option<VecDeque<StatsEvent>>,
late_replies: LateReplyMode,
} }
impl RollingEvents { impl RollingEvents {
@@ -17,6 +20,7 @@ impl RollingEvents {
time_limit: config.rolling_time, time_limit: config.rolling_time,
count_events: config.rolling_count.map(|_| VecDeque::new()), count_events: config.rolling_count.map(|_| VecDeque::new()),
time_events: config.rolling_time.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> { 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> { 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 { /// Recompute a window snapshot under the same normalized semantics as the
let mut core = CoreStats::new(SampleMode::RunningOnly); /// 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 { for event in events {
core.apply(event.clone()); core.apply(event.clone());
} }
+122 -1
View File
@@ -8,7 +8,9 @@ use common::{
use irtt_client::{ use irtt_client::{
ClientEvent, ClientTimestamp, OneWayDelaySample, PacketMeta, RttSample, SignedDuration, ClientEvent, ClientTimestamp, OneWayDelaySample, PacketMeta, RttSample, SignedDuration,
}; };
use irtt_stats::{EventStatsUpdate, IpdvPairUpdate, SampleMode, StatsCollector, StatsConfig}; use irtt_stats::{
EventStatsUpdate, IpdvPairUpdate, LateReplyMode, SampleMode, StatsCollector, StatsConfig,
};
#[test] #[test]
fn running_only_samples_avoid_finite_retention() { fn running_only_samples_avoid_finite_retention() {
@@ -342,6 +344,7 @@ fn rolling_count_eviction_recomputes_from_bounded_events() {
samples: SampleMode::RunningOnly, samples: SampleMode::RunningOnly,
rolling_count: Some(2), rolling_count: Some(2),
rolling_time: None, rolling_time: None,
late_replies: LateReplyMode::Measure,
}); });
collector.process(&sent(0, ts(0))); collector.process(&sent(0, ts(0)));
collector.process(&unadjusted_reply(0, 10)); collector.process(&unadjusted_reply(0, 10));
@@ -359,6 +362,7 @@ fn rolling_time_eviction_uses_event_timestamps() {
samples: SampleMode::RunningOnly, samples: SampleMode::RunningOnly,
rolling_count: None, rolling_count: None,
rolling_time: Some(Duration::from_millis(15)), rolling_time: Some(Duration::from_millis(15)),
late_replies: LateReplyMode::Measure,
}); });
collector.process(&sent(0, ts(0))); collector.process(&sent(0, ts(0)));
collector.process(&sent(1, ts(10))); 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.packet_loss_percent, 50.0);
assert_eq!(loss.upstream_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);
}