test(irtt-stats): add memory growth benchmarks

Introduce tests with a custom global allocator to measure and report heap allocation trends across different statistics configurations. Note: temporarily commented out `#![forbid(unsafe_code)]` to support the counting allocator implementation.

crates/irtt-cli/src/lib.rs

@@ -1456,6 +1456,7 @@ mod tests {
                     interval_ns: 1_000_000_000,
                     ..Params::default()
                 },
+                restrictions: vec![],
             },
             at: test_timestamp(Duration::from_secs(1)),
         };

crates/irtt-cli/src/main.rs

@@ -237,7 +237,7 @@ impl<W: Write> EventOutput<'_, W> {
                 self.out,
                 "{}",
                 irtt_cli::summary::format_summary_with_options(
-                    &self.stats.summary(),
+                    &self.stats.snapshot(),
                     irtt_cli::summary::SummaryFormatOptions {
                         verbose: self.human_options.verbose,
                     },
@@ -390,7 +390,7 @@ mod tests {
         }
 
         let rendered = String::from_utf8(out).unwrap();
-        let summary = stats.summary();
+        let summary = stats.snapshot();
         assert_eq!(rendered, "1200\n");
         assert_eq!(summary.packets.packets_sent, 1);
         assert_eq!(summary.packets.unique_replies, 1);
@@ -462,7 +462,7 @@ mod tests {
             });
         }
 
-        assert_eq!(collector.summary().rtt.primary.median_ns, None);
+        assert_eq!(collector.snapshot().rtt.primary.median_ns, None);
     }
 
     #[test]

crates/irtt-cli/src/summary.rs

@@ -1,10 +1,8 @@
 use std::fmt::Write as _;
 
-use irtt_stats::{
+use irtt_stats::{Snapshot, TimeStats};
-    DurationStats, DurationStatsWithMedian, FiniteSummary, SignedDurationStatsWithMedian,
-};
 
-pub fn format_summary(summary: &FiniteSummary) -> String {
+pub fn format_summary(summary: &Snapshot) -> String {
     format_summary_with_options(summary, SummaryFormatOptions::default())
 }
 
@@ -13,10 +11,7 @@ pub struct SummaryFormatOptions {
     pub verbose: bool,
 }
 
-pub fn format_summary_with_options(
+pub fn format_summary_with_options(summary: &Snapshot, options: SummaryFormatOptions) -> String {
-    summary: &FiniteSummary,
-    options: SummaryFormatOptions,
-) -> String {
     let mut out = String::new();
     let packets = summary.packets;
     let loss = summary.loss;
@@ -32,27 +27,27 @@ pub fn format_summary_with_options(
     .unwrap();
     writeln!(out, "  {}", "-".repeat(82)).unwrap();
 
-    write_signed_duration_row(&mut out, "RTT", &summary.rtt.primary);
+    write_time_row(&mut out, "RTT", &summary.rtt.primary);
     if options.verbose {
-        write_duration_row(&mut out, "raw RTT", &summary.rtt.raw);
+        write_time_row(&mut out, "raw RTT", &summary.rtt.raw);
-        write_signed_duration_row(&mut out, "adjusted RTT", &summary.rtt.adjusted);
+        write_time_row(&mut out, "adjusted RTT", &summary.rtt.adjusted);
     }
-    write_duration_row(&mut out, "IPDV/jitter", &summary.ipdv.round_trip);
+    write_time_row(&mut out, "IPDV/jitter", &summary.ipdv.round_trip);
-    write_duration_row(&mut out, "send IPDV", &summary.ipdv.send);
+    write_time_row(&mut out, "send IPDV", &summary.ipdv.send);
-    write_duration_row(&mut out, "receive IPDV", &summary.ipdv.receive);
+    write_time_row(&mut out, "receive IPDV", &summary.ipdv.receive);
-    write_duration_row(&mut out, "send delay", &summary.one_way_delay.send_delay);
+    write_time_row(&mut out, "send delay", &summary.one_way_delay.send_delay);
-    write_duration_row(
+    write_time_row(
         &mut out,
         "receive delay",
         &summary.one_way_delay.receive_delay,
     );
-    write_duration_row_no_median(
+    write_time_row(
         &mut out,
         "server processing",
         &summary.server_processing.processing,
     );
-    write_duration_row_no_median(&mut out, "send call", &summary.send_call);
+    write_time_row(&mut out, "send call", &summary.send_call);
-    write_duration_row_no_median(&mut out, "timer error", &summary.timer_error);
+    write_time_row(&mut out, "timer error", &summary.timer_error);
 
     writeln!(out).unwrap();
     writeln!(
@@ -97,24 +92,7 @@ pub fn format_summary_with_options(
     out
 }
 
-fn write_duration_row(out: &mut String, label: &str, value: &DurationStatsWithMedian) {
+fn write_time_row(out: &mut String, label: &str, value: &TimeStats) {
-    if value.stats.count == 0 {
-        return;
-    }
-    writeln!(
-        out,
-        "  {label:<18} {:>7} {:>10} {:>10} {:>10} {:>10} {:>10}",
-        value.stats.count,
-        format_ns_u64(value.stats.min_ns),
-        format_ns_f64(value.stats.mean_ns),
-        format_ns_f64_opt(value.median_ns),
-        format_ns_u64(value.stats.max_ns),
-        format_ns_f64(value.stddev_ns())
-    )
-    .unwrap();
-}
-
-fn write_duration_row_no_median(out: &mut String, label: &str, value: &DurationStats) {
     if value.count == 0 {
         return;
     }
@@ -122,27 +100,10 @@ fn write_duration_row_no_median(out: &mut String, label: &str, value: &DurationS
         out,
         "  {label:<18} {:>7} {:>10} {:>10} {:>10} {:>10} {:>10}",
         value.count,
-        format_ns_u64(value.min_ns),
+        format_ns_i128(value.min_ns),
         format_ns_f64(value.mean_ns),
-        "-",
-        format_ns_u64(value.max_ns),
-        format_ns_f64(value.stddev_ns())
-    )
-    .unwrap();
-}
-
-fn write_signed_duration_row(out: &mut String, label: &str, value: &SignedDurationStatsWithMedian) {
-    if value.stats.count == 0 {
-        return;
-    }
-    writeln!(
-        out,
-        "  {label:<18} {:>7} {:>10} {:>10} {:>10} {:>10} {:>10}",
-        value.stats.count,
-        format_ns_i128(value.stats.min_ns),
-        format_ns_f64(value.stats.mean_ns),
         format_ns_f64_opt(value.median_ns),
-        format_ns_i128(value.stats.max_ns),
+        format_ns_i128(value.max_ns),
         format_ns_f64(value.stddev_ns())
     )
     .unwrap();
@@ -152,12 +113,6 @@ fn format_percent(value: f64) -> String {
     format!("{value:.2}%")
 }
 
-fn format_ns_u64(value: Option<u64>) -> String {
-    value
-        .map(|value| format_ns_f64(value as f64))
-        .unwrap_or_else(|| "-".to_owned())
-}
-
 fn format_ns_i128(value: Option<i128>) -> String {
     value
         .map(|value| format_ns_f64(value as f64))
@@ -188,7 +143,7 @@ mod tests {
     use irtt_client::{
         ClientEvent, ClientTimestamp, PacketMeta, RttSample, ServerTiming, SignedDuration,
     };
-    use irtt_stats::{SignedDurationStats, StatsCollector, StatsConfig};
+    use irtt_stats::{StatsCollector, StatsConfig};
     use std::{
         net::{IpAddr, Ipv4Addr, SocketAddr},
         time::{Duration, Instant, UNIX_EPOCH},
@@ -207,7 +162,7 @@ mod tests {
 
     #[test]
     fn empty_summary_omits_optional_metric_sections() {
-        let summary = StatsCollector::new(StatsConfig::finite()).summary();
+        let summary = StatsCollector::new(StatsConfig::finite()).snapshot();
         let output = format_summary(&summary);
 
         assert!(output.contains("Metric"));
@@ -263,7 +218,7 @@ mod tests {
             packet_meta: PacketMeta::default(),
         });
 
-        let output = format_summary(&collector.summary());
+        let output = format_summary(&collector.snapshot());
 
         assert!(output.contains("packets: sent=1 received=1 unique=1 lost=0 loss=0.00%"));
         assert!(output.contains("bytes: sent=64 received=64"));
@@ -303,7 +258,7 @@ mod tests {
         });
 
         let output = format_summary_with_options(
-            &collector.summary(),
+            &collector.snapshot(),
             SummaryFormatOptions { verbose: true },
         );
 
@@ -313,20 +268,18 @@ mod tests {
 
     #[test]
     fn signed_stats_can_format_negative_values() {
-        let stats = SignedDurationStatsWithMedian {
+        let stats = TimeStats {
-            stats: SignedDurationStats {
             count: 1,
             total_ns: -500,
             min_ns: Some(-500),
             max_ns: Some(-500),
             mean_ns: -500.0,
-                variance_ns2: 0.0,
-            },
             median_ns: Some(-500.0),
+            variance_ns2: 0.0,
         };
 
         let mut out = String::new();
-        write_signed_duration_row(&mut out, "RTT", &stats);
+        write_time_row(&mut out, "RTT", &stats);
         assert!(out.contains("-500ns"));
     }
 }

crates/irtt-client/src/client.rs

@@ -23,7 +23,7 @@ use crate::{
     metadata::ReceiveMeta,
     probe::{CompletedSet, PendingMap, PendingProbe, TimedOutMap},
     receive::recv_datagram,
-    session::{validate_negotiated_params, ActiveSession, ClientPhase, NegotiatedParams},
+    session::{negotiate_params, ActiveSession, ClientPhase, NegotiatedParams},
     socket::{connect_udp_socket, resolve_remote, validate_open_timeouts},
     socket_options::{apply_dscp_to_socket, clear_dscp_on_socket},
     timing::ClientTimestamp,
@@ -328,8 +328,9 @@ impl Client {
     }
 
     #[cfg(test)]
-    fn recv_buffer_size(&self) -> usize {
+    pub(crate) fn recv_buffer_size(&self) -> usize {
         recv_buffer_size(self.config.hmac_key.is_some(), self.negotiated.as_ref())
+            .expect("test params must have a non-negative packet length")
     }
 
     /// Polls for probes that have timed out as of the current monotonic time.
@@ -604,15 +605,13 @@ impl Client {
         reply: OpenReply,
         now: ClientTimestamp,
     ) -> Result<OpenOutcome, ClientError> {
-        validate_negotiated_params(
+        let token = reply.token;
+        let negotiated = negotiate_params(
             &self.requested,
-            &reply.params,
+            reply.params,
             self.config.negotiation_policy,
         )?;
-        let negotiated = NegotiatedParams {
+        let recv_buffer_size = recv_buffer_size(self.config.hmac_key.is_some(), Some(&negotiated))?;
-            params: reply.params.clone(),
-        };
-        let recv_buffer_size = recv_buffer_size(self.config.hmac_key.is_some(), Some(&negotiated));
         let negotiated_dscp =
             u8::try_from(negotiated.params.dscp).map_err(|_| ClientError::InvalidConfig {
                 reason: "negotiated dscp must be in range 0..=63".to_owned(),
@@ -620,7 +619,7 @@ impl Client {
         apply_dscp_to_socket(&self.socket, self.remote, negotiated_dscp)?;
         self.negotiated = Some(negotiated.clone());
         self.recv_buffer.resize(recv_buffer_size, 0);
-        self.phase = ClientPhase::Open { token: reply.token };
+        self.phase = ClientPhase::Open { token };
 
         let end_mono = if negotiated.params.duration_ns > 0 {
             Some(negotiated_end_mono(
@@ -646,14 +645,14 @@ impl Client {
 
         let event = ClientEvent::SessionStarted {
             remote: self.remote,
-            token: reply.token,
+            token,
             negotiated: negotiated.clone(),
             at: now,
         };
 
         Ok(OpenOutcome::Started {
             remote: self.remote,
-            token: reply.token,
+            token,
             negotiated,
             event,
         })
@@ -664,14 +663,11 @@ impl Client {
         reply: OpenReply,
         now: ClientTimestamp,
     ) -> Result<OpenOutcome, ClientError> {
-        validate_negotiated_params(
+        let negotiated = negotiate_params(
             &self.requested,
-            &reply.params,
+            reply.params,
             self.config.negotiation_policy,
         )?;
-        let negotiated = NegotiatedParams {
-            params: reply.params.clone(),
-        };
         self.negotiated = Some(negotiated.clone());
         self.phase = ClientPhase::NoTestCompleted;
         let event = ClientEvent::NoTestCompleted {
@@ -735,13 +731,16 @@ fn instant_abs_diff(left: Instant, right: Instant) -> Duration {
         .unwrap_or(Duration::ZERO)
 }
 
-fn recv_buffer_size(has_hmac: bool, negotiated: Option<&NegotiatedParams>) -> usize {
+pub(crate) fn recv_buffer_size(
-    match negotiated {
+    has_hmac: bool,
-        Some(negotiated) => echo_packet_len(has_hmac, &negotiated.params)
+    negotiated: Option<&NegotiatedParams>,
+) -> Result<usize, ClientError> {
+    Ok(match negotiated {
+        Some(negotiated) => echo_packet_len(has_hmac, &negotiated.params)?
             .saturating_add(1)
             .max(MIN_RECV_BUFFER_SIZE),
         None => MIN_RECV_BUFFER_SIZE,
-    }
+    })
 }
 
 fn compute_rtt(

crates/irtt-client/src/client/tests/dscp.rs

@@ -60,6 +60,13 @@ fn normal_open_uses_negotiated_dscp_not_requested_dscp() {
 
     let negotiated = assert_open_started(client.open().unwrap());
     assert_eq!(negotiated.params.dscp, 0);
+    assert_eq!(
+        negotiated.restrictions,
+        vec![crate::NegotiationRestriction::DscpChanged {
+            requested: 46,
+            negotiated: 0,
+        }]
+    );
     assert_eq!(
         socket_traffic_class(&client.socket, client.remote).unwrap(),
         0

crates/irtt-client/src/client/tests/mod.rs

@@ -1,8 +1,8 @@
 use super::*;
 use crate::config::{NegotiationPolicy, DEFAULT_OPEN_TIMEOUTS};
 use irtt_proto::{
-    echo_packet_len, flags::FLAG_HMAC, flags::FLAG_OPEN, flags::FLAG_REPLY, verify_hmac, Clock,
+    flags::FLAG_HMAC, flags::FLAG_OPEN, flags::FLAG_REPLY, verify_hmac, Clock, ReceivedStats,
-    ReceivedStats, StampAt, HMAC_SIZE, MAGIC,
+    StampAt, HMAC_SIZE, MAGIC,
 };
 use std::{thread, time::SystemTime};
 

crates/irtt-client/src/client/tests/negotiation.rs

@@ -1,32 +1,72 @@
 use super::*;
+use crate::NegotiationRestriction;
+
+fn assert_negotiates(
+    requested: &Params,
+    returned: &Params,
+    policy: NegotiationPolicy,
+) -> NegotiatedParams {
+    negotiate_params(requested, returned.clone(), policy)
+        .unwrap_or_else(|err| panic!("expected negotiation success, got {err:?}"))
+}
+
+fn rejection_reason(requested: &Params, returned: &Params, policy: NegotiationPolicy) -> String {
+    match negotiate_params(requested, returned.clone(), policy) {
+        Err(ClientError::NegotiationRejected { reason }) => reason,
+        other => panic!("expected negotiation rejection, got {other:?}"),
+    }
+}
 
 #[test]
 fn strict_negotiation_accepts_identical_params() {
     let config = ClientConfig::default();
     let params = params_from_config(&config).unwrap();
-    assert!(validate_negotiated_params(&params, &params, NegotiationPolicy::Strict).is_ok());
+    let negotiated = assert_negotiates(&params, &params, NegotiationPolicy::Strict);
+    assert_eq!(negotiated.params, params);
+    assert!(negotiated.restrictions.is_empty());
 }
 
 #[test]
 fn strict_negotiation_rejects_changed_params() {
-    let config = ClientConfig::default();
+    let config = ClientConfig {
+        dscp: 46,
+        ..ClientConfig::default()
+    };
     let requested = params_from_config(&config).unwrap();
     let mut returned = requested.clone();
-    returned.dscp = 1;
+    returned.dscp = 0;
-    assert!(matches!(
+    assert_eq!(
-        validate_negotiated_params(&requested, &returned, NegotiationPolicy::Strict),
+        rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
-        Err(ClientError::NegotiationRejected { .. })
+        NegotiationRestriction::DscpChanged {
-    ));
+            requested: requested.dscp,
+            negotiated: returned.dscp,
+        }
+        .message()
+    );
 }
 
 #[test]
 fn loose_negotiation_accepts_server_restricted_params() {
     let config = ClientConfig::default();
-    let requested = params_from_config(&config).unwrap();
+    let mut requested = params_from_config(&config).unwrap();
+    requested.length = 128;
     let mut returned = requested.clone();
     returned.duration_ns /= 2;
     returned.length = 0;
-    assert!(validate_negotiated_params(&requested, &returned, NegotiationPolicy::Loose).is_ok());
+    let negotiated = assert_negotiates(&requested, &returned, NegotiationPolicy::Loose);
+    assert_eq!(
+        negotiated.restrictions,
+        vec![
+            NegotiationRestriction::DurationReduced {
+                requested_ns: requested.duration_ns,
+                negotiated_ns: returned.duration_ns,
+            },
+            NegotiationRestriction::LengthReduced {
+                requested: requested.length,
+                negotiated: returned.length,
+            },
+        ]
+    );
 }
 
 #[test]
@@ -37,11 +77,10 @@ fn loose_negotiation_rejects_non_positive_returned_interval() {
     for interval_ns in [0, -1] {
         let mut returned = requested.clone();
         returned.interval_ns = interval_ns;
-        assert!(matches!(
+        assert_eq!(
-            validate_negotiated_params(&requested, &returned, NegotiationPolicy::Loose),
+            rejection_reason(&requested, &returned, NegotiationPolicy::Loose),
-            Err(ClientError::NegotiationRejected { reason })
+            "interval must be positive"
-                if reason == "interval must be positive"
+        );
-        ));
     }
 }
 
@@ -51,21 +90,28 @@ fn loose_negotiation_accepts_documented_duration_combinations() {
     let requested_finite = params_from_config(&finite_config).unwrap();
 
     let returned_same_finite = requested_finite.clone();
-    assert!(validate_negotiated_params(
+    assert!(assert_negotiates(
         &requested_finite,
         &returned_same_finite,
         NegotiationPolicy::Loose
     )
-    .is_ok());
+    .restrictions
+    .is_empty());
 
     let mut returned_shorter_finite = requested_finite.clone();
     returned_shorter_finite.duration_ns /= 2;
-    assert!(validate_negotiated_params(
+    assert_eq!(
+        assert_negotiates(
             &requested_finite,
             &returned_shorter_finite,
             NegotiationPolicy::Loose
         )
-    .is_ok());
+        .restrictions,
+        vec![NegotiationRestriction::DurationReduced {
+            requested_ns: requested_finite.duration_ns,
+            negotiated_ns: returned_shorter_finite.duration_ns,
+        }]
+    );
 
     let continuous_config = ClientConfig {
         duration: None,
@@ -74,21 +120,28 @@ fn loose_negotiation_accepts_documented_duration_combinations() {
     let requested_continuous = params_from_config(&continuous_config).unwrap();
 
     let returned_continuous = requested_continuous.clone();
-    assert!(validate_negotiated_params(
+    assert!(assert_negotiates(
         &requested_continuous,
         &returned_continuous,
         NegotiationPolicy::Loose
     )
-    .is_ok());
+    .restrictions
+    .is_empty());
 
     let mut returned_finite = requested_continuous.clone();
     returned_finite.duration_ns = 1_000_000_000;
-    assert!(validate_negotiated_params(
+    assert_eq!(
+        assert_negotiates(
             &requested_continuous,
             &returned_finite,
             NegotiationPolicy::Loose
         )
-    .is_ok());
+        .restrictions,
+        vec![NegotiationRestriction::DurationReduced {
+            requested_ns: requested_continuous.duration_ns,
+            negotiated_ns: returned_finite.duration_ns,
+        }]
+    );
 }
 
 #[test]
@@ -98,11 +151,10 @@ fn loose_negotiation_rejects_finite_request_returned_continuous() {
     let mut returned = requested.clone();
     returned.duration_ns = 0;
 
-    assert!(matches!(
+    assert_eq!(
-        validate_negotiated_params(&requested, &returned, NegotiationPolicy::Loose),
+        rejection_reason(&requested, &returned, NegotiationPolicy::Loose),
-        Err(ClientError::NegotiationRejected { reason })
+        "server returned continuous duration for finite request"
-            if reason == "server returned continuous duration for finite request"
+    );
-    ));
 }
 
 #[test]
@@ -111,10 +163,10 @@ fn loose_negotiation_rejects_negative_returned_length() {
     let requested = params_from_config(&config).unwrap();
     let mut returned = requested.clone();
     returned.length = -1;
-    assert!(matches!(
+    assert_eq!(
-        validate_negotiated_params(&requested, &returned, NegotiationPolicy::Loose),
+        rejection_reason(&requested, &returned, NegotiationPolicy::Loose),
-        Err(ClientError::NegotiationRejected { reason }) if reason == "length must be non-negative"
+        "length must be non-negative"
-    ));
+    );
 }
 
 #[test]
@@ -123,10 +175,10 @@ fn strict_negotiation_rejects_negative_returned_length() {
     let requested = params_from_config(&config).unwrap();
     let mut returned = requested.clone();
     returned.length = -1;
-    assert!(matches!(
+    assert_eq!(
-        validate_negotiated_params(&requested, &returned, NegotiationPolicy::Strict),
+        rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
-        Err(ClientError::NegotiationRejected { reason }) if reason == "length must be non-negative"
+        "length must be non-negative"
-    ));
+    );
 }
 
 #[test]
@@ -137,10 +189,80 @@ fn loose_negotiation_rejects_runtime_invalid_returned_dscp() {
     for dscp in [-1, 64] {
         let mut returned = requested.clone();
         returned.dscp = dscp;
-        assert!(matches!(
+        assert_eq!(
-            validate_negotiated_params(&requested, &returned, NegotiationPolicy::Loose),
+            rejection_reason(&requested, &returned, NegotiationPolicy::Loose),
-            Err(ClientError::NegotiationRejected { reason })
+            "dscp must be in range 0..=63"
-                if reason == "dscp must be in range 0..=63"
+        );
-        ));
     }
 }
+
+#[test]
+fn loose_negotiation_records_dscp_disabled_by_server() {
+    let config = ClientConfig {
+        dscp: 46,
+        ..ClientConfig::default()
+    };
+    let requested = params_from_config(&config).unwrap();
+    let mut returned = requested.clone();
+    returned.dscp = 0;
+
+    let negotiated = assert_negotiates(&requested, &returned, NegotiationPolicy::Loose);
+
+    assert_eq!(
+        negotiated.restrictions,
+        vec![NegotiationRestriction::DscpChanged {
+            requested: 46,
+            negotiated: 0,
+        }]
+    );
+}
+
+#[test]
+fn strict_negotiation_rejects_dscp_disabled_by_server_as_specific_restriction() {
+    let config = ClientConfig {
+        dscp: 46,
+        ..ClientConfig::default()
+    };
+    let requested = params_from_config(&config).unwrap();
+    let mut returned = requested.clone();
+    returned.dscp = 0;
+
+    assert_eq!(
+        rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
+        NegotiationRestriction::DscpChanged {
+            requested: 46,
+            negotiated: 0,
+        }
+        .message()
+    );
+}
+
+#[test]
+fn negotiation_rejects_unsupported_dscp_changes() {
+    let config = ClientConfig {
+        dscp: 46,
+        ..ClientConfig::default()
+    };
+    let requested = params_from_config(&config).unwrap();
+    let mut returned = requested.clone();
+    returned.dscp = 8;
+
+    assert_eq!(
+        rejection_reason(&requested, &returned, NegotiationPolicy::Loose),
+        "server returned unsupported DSCP change"
+    );
+    assert_eq!(
+        rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
+        "server returned unsupported DSCP change"
+    );
+
+    let zero_config = ClientConfig::default();
+    let zero_requested = params_from_config(&zero_config).unwrap();
+    let mut returned = zero_requested.clone();
+    returned.dscp = 46;
+
+    assert_eq!(
+        rejection_reason(&zero_requested, &returned, NegotiationPolicy::Loose),
+        "server returned unsupported DSCP change"
+    );
+}

crates/irtt-client/src/client/tests/no_test.rs

@@ -79,13 +79,21 @@ fn no_test_loose_negotiation_accepts_restricted_params() {
     let mut config = default_test_config(SocketAddr::from(([127, 0, 0, 1], 1)));
     config.run_mode = RunMode::NoTest;
     config.negotiation_policy = NegotiationPolicy::Loose;
-    let mut params = params_from_config(&config).unwrap();
+    let requested = params_from_config(&config).unwrap();
+    let mut params = requested.clone();
     params.duration_ns /= 2;
     let server = no_test_server(params.clone(), 0);
     config.server_addr = server.addr.to_string();
     let mut client = Client::connect(config).unwrap();
     let negotiated = assert_no_test_completed(client.open().unwrap());
     assert_eq!(negotiated.params, params);
+    assert_eq!(
+        negotiated.restrictions,
+        vec![crate::NegotiationRestriction::DurationReduced {
+            requested_ns: requested.duration_ns,
+            negotiated_ns: params.duration_ns,
+        }]
+    );
     server.join();
 }
 

crates/irtt-client/src/client/tests/probes_replies.rs

@@ -1,4 +1,5 @@
 use super::*;
+use support::echo_packet_len;
 
 #[cfg(not(all(target_os = "linux", feature = "ancillary")))]
 fn assert_packet_meta_unavailable(packet_meta: &crate::event::PacketMeta) {

crates/irtt-client/src/client/tests/support.rs

@@ -1,7 +1,8 @@
 use super::*;
 use irtt_proto::{
-    compute_hmac_in_place, echo_packet_len, flags::FLAG_HMAC, flags::FLAG_OPEN, flags::FLAG_REPLY,
+    compute_hmac_in_place, echo_packet_len as checked_echo_packet_len, flags::FLAG_HMAC,
-    layout::PacketLayout, Clock, ReceivedStats, StampAt, HMAC_SIZE, MAGIC,
+    flags::FLAG_OPEN, flags::FLAG_REPLY, layout::PacketLayout, Clock, ReceivedStats, StampAt,
+    HMAC_SIZE, MAGIC,
 };
 use std::{
     net::UdpSocket,
@@ -74,6 +75,11 @@ pub(super) fn open_reply(
     packet
 }
 
+pub(crate) fn echo_packet_len(hmac: bool, params: &Params) -> usize {
+    checked_echo_packet_len(hmac, params)
+        .expect("test params must have a non-negative packet length")
+}
+
 pub(super) fn echo_reply_packet(
     token: u64,
     seq: u32,

crates/irtt-client/src/lib.rs

@@ -31,5 +31,5 @@ pub use managed::{
     CancellationToken, EventHub, EventSubscription, ManagedClient, ManagedClientSession,
     SessionEndReason, SessionOutcome, SubscriberConfig, SubscriberOverflow,
 };
-pub use session::NegotiatedParams;
+pub use session::{NegotiatedParams, NegotiationRestriction};
 pub use timing::ClientTimestamp;

crates/irtt-client/src/managed/runner.rs

@@ -311,8 +311,10 @@ mod tests {
     use super::*;
     use crate::{config::NegotiationPolicy, SubscriberOverflow};
     use irtt_proto::{
-        echo_packet_len, flags, flags::FLAG_OPEN, flags::FLAG_REPLY, layout::PacketLayout, Clock,
+        echo_packet_len,
-        Params, ReceivedStats, StampAt, TimestampFields, MAGIC, PROTOCOL_VERSION,
+        flags::{self, FLAG_OPEN, FLAG_REPLY},
+        layout::PacketLayout,
+        Clock, Params, ReceivedStats, StampAt, TimestampFields, MAGIC, PROTOCOL_VERSION,
     };
     use std::{
         net::{SocketAddr, UdpSocket},
@@ -331,6 +333,11 @@ mod tests {
         }
     }
 
+    fn test_echo_packet_len(hmac: bool, params: &Params) -> usize {
+        echo_packet_len(hmac, params)
+            .expect("managed runner test params must have a non-negative packet length")
+    }
+
     fn test_params(duration: Option<Duration>, interval: Duration) -> Params {
         Params {
             protocol_version: PROTOCOL_VERSION,
@@ -461,7 +468,7 @@ mod tests {
         timestamps: &TimestampFields,
     ) -> Vec<u8> {
         let layout = PacketLayout::echo(false, params);
-        let packet_len = echo_packet_len(false, params);
+        let packet_len = test_echo_packet_len(false, params);
         let mut packet = Vec::with_capacity(packet_len);
 
         packet.extend_from_slice(&MAGIC);

crates/irtt-client/src/session.rs

@@ -1,6 +1,6 @@
-use std::time::Instant;
+use std::{fmt, time::Instant};
 
-use irtt_proto::{Params, PROTOCOL_VERSION};
+use irtt_proto::{Clock, Params, ReceivedStats, StampAt, PROTOCOL_VERSION};
 
 use crate::{
     config::{NegotiationPolicy, MAX_DSCP_CODEPOINT},
@@ -11,6 +11,141 @@ use crate::{
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct NegotiatedParams {
     pub params: Params,
+    pub restrictions: Vec<NegotiationRestriction>,
+}
+
+/// A server-side restriction applied during session parameter negotiation.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum NegotiationRestriction {
+    /// Run duration was reduced.
+    ///
+    /// A requested duration of `0` means the client requested continuous mode.
+    /// When `requested_ns == 0` and `negotiated_ns > 0`, the server limited
+    /// that continuous request to a finite duration.
+    DurationReduced {
+        requested_ns: i64,
+        negotiated_ns: i64,
+    },
+    /// Probe interval was increased.
+    IntervalIncreased {
+        requested_ns: i64,
+        negotiated_ns: i64,
+    },
+    /// Probe interval was reduced.
+    IntervalReduced {
+        requested_ns: i64,
+        negotiated_ns: i64,
+    },
+    /// Packet length was reduced.
+    LengthReduced { requested: i64, negotiated: i64 },
+    /// Returned received-statistics mode differs from the request.
+    ReceivedStatsChanged {
+        requested: ReceivedStats,
+        negotiated: ReceivedStats,
+    },
+    /// Returned timestamp placement differs from the request.
+    StampAtChanged {
+        requested: StampAt,
+        negotiated: StampAt,
+    },
+    /// Returned clock source differs from the request.
+    ClockChanged { requested: Clock, negotiated: Clock },
+    /// Returned DSCP codepoint differs from the request.
+    DscpChanged { requested: i64, negotiated: i64 },
+    /// Returned server payload fill behavior differs from the request.
+    ServerFillChanged,
+}
+
+impl NegotiationRestriction {
+    pub fn message(&self) -> String {
+        self.to_string()
+    }
+}
+
+impl fmt::Display for NegotiationRestriction {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::DurationReduced {
+                requested_ns: 0,
+                negotiated_ns,
+            } => {
+                write!(
+                    f,
+                    "server limited continuous duration to {negotiated_ns} ns"
+                )
+            }
+            Self::DurationReduced {
+                requested_ns,
+                negotiated_ns,
+            } => {
+                write!(
+                    f,
+                    "server reduced duration from {requested_ns} ns to {negotiated_ns} ns"
+                )
+            }
+            Self::IntervalIncreased {
+                requested_ns,
+                negotiated_ns,
+            } => {
+                write!(
+                    f,
+                    "server increased interval from {requested_ns} ns to {negotiated_ns} ns"
+                )
+            }
+            Self::IntervalReduced {
+                requested_ns,
+                negotiated_ns,
+            } => {
+                write!(
+                    f,
+                    "server reduced interval from {requested_ns} ns to {negotiated_ns} ns"
+                )
+            }
+            Self::LengthReduced {
+                requested,
+                negotiated,
+            } => {
+                write!(
+                    f,
+                    "server reduced packet length from {requested} bytes to {negotiated} bytes"
+                )
+            }
+            Self::ReceivedStatsChanged {
+                requested,
+                negotiated,
+            } => {
+                write!(
+                    f,
+                    "server changed received-stats from {requested:?} to {negotiated:?}"
+                )
+            }
+            Self::StampAtChanged {
+                requested,
+                negotiated,
+            } => {
+                write!(
+                    f,
+                    "server changed stamp-at from {requested:?} to {negotiated:?}"
+                )
+            }
+            Self::ClockChanged {
+                requested,
+                negotiated,
+            } => {
+                write!(
+                    f,
+                    "server changed clock from {requested:?} to {negotiated:?}"
+                )
+            }
+            Self::DscpChanged {
+                requested,
+                negotiated,
+            } => {
+                write!(f, "server changed DSCP from {requested} to {negotiated}")
+            }
+            Self::ServerFillChanged => write!(f, "server changed payload fill behavior"),
+        }
+    }
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -35,17 +170,19 @@ pub(crate) struct ActiveSession {
     pub sending_done: bool,
 }
 
-pub(crate) fn validate_negotiated_params(
+pub(crate) fn negotiate_params(
     requested: &Params,
-    returned: &Params,
+    returned: Params,
     policy: NegotiationPolicy,
-) -> Result<(), ClientError> {
+) -> Result<NegotiatedParams, ClientError> {
     if returned.protocol_version != PROTOCOL_VERSION {
         return Err(ClientError::ProtocolVersionMismatch {
             requested: PROTOCOL_VERSION,
             received: returned.protocol_version,
         });
     }
+    let mut restrictions = Vec::new();
+
     validate_duration_restriction(requested.duration_ns, returned.duration_ns)?;
     if returned.length < 0 {
         return Err(ClientError::NegotiationRejected {
@@ -64,12 +201,105 @@ pub(crate) fn validate_negotiated_params(
     }
     validate_dscp_restriction(returned.dscp)?;
 
-    if policy == NegotiationPolicy::Strict && returned != requested {
+    if returned.duration_ns < requested.duration_ns
+        || (requested.duration_ns == 0 && returned.duration_ns > 0)
+    {
+        record_restriction(
+            policy,
+            &mut restrictions,
+            NegotiationRestriction::DurationReduced {
+                requested_ns: requested.duration_ns,
+                negotiated_ns: returned.duration_ns,
+            },
+        )?;
+    }
+    if returned.interval_ns > requested.interval_ns {
+        record_restriction(
+            policy,
+            &mut restrictions,
+            NegotiationRestriction::IntervalIncreased {
+                requested_ns: requested.interval_ns,
+                negotiated_ns: returned.interval_ns,
+            },
+        )?;
+    }
+    if returned.interval_ns < requested.interval_ns {
+        record_restriction(
+            policy,
+            &mut restrictions,
+            NegotiationRestriction::IntervalReduced {
+                requested_ns: requested.interval_ns,
+                negotiated_ns: returned.interval_ns,
+            },
+        )?;
+    }
+    if returned.length < requested.length {
+        record_restriction(
+            policy,
+            &mut restrictions,
+            NegotiationRestriction::LengthReduced {
+                requested: requested.length,
+                negotiated: returned.length,
+            },
+        )?;
+    }
+    if returned.received_stats != requested.received_stats {
+        record_restriction(
+            policy,
+            &mut restrictions,
+            NegotiationRestriction::ReceivedStatsChanged {
+                requested: requested.received_stats,
+                negotiated: returned.received_stats,
+            },
+        )?;
+    }
+    if returned.stamp_at != requested.stamp_at {
+        record_restriction(
+            policy,
+            &mut restrictions,
+            NegotiationRestriction::StampAtChanged {
+                requested: requested.stamp_at,
+                negotiated: returned.stamp_at,
+            },
+        )?;
+    }
+    if returned.clock != requested.clock {
+        record_restriction(
+            policy,
+            &mut restrictions,
+            NegotiationRestriction::ClockChanged {
+                requested: requested.clock,
+                negotiated: returned.clock,
+            },
+        )?;
+    }
+    if returned.dscp != requested.dscp && returned.dscp != 0 {
         return Err(ClientError::NegotiationRejected {
-            reason: "returned params differ from requested params".to_owned(),
+            reason: "server returned unsupported DSCP change".to_owned(),
         });
     }
-    Ok(())
+    if returned.dscp == 0 && requested.dscp != 0 {
+        record_restriction(
+            policy,
+            &mut restrictions,
+            NegotiationRestriction::DscpChanged {
+                requested: requested.dscp,
+                negotiated: returned.dscp,
+            },
+        )?;
+    }
+    if returned.server_fill != requested.server_fill {
+        record_restriction(
+            policy,
+            &mut restrictions,
+            NegotiationRestriction::ServerFillChanged,
+        )?;
+    }
+
+    Ok(NegotiatedParams {
+        params: returned,
+        restrictions,
+    })
 }
 
 fn validate_duration_restriction(requested: i64, returned: i64) -> Result<(), ClientError> {
@@ -104,6 +334,21 @@ fn validate_dscp_restriction(returned: i64) -> Result<(), ClientError> {
     Ok(())
 }
 
+fn record_restriction(
+    policy: NegotiationPolicy,
+    restrictions: &mut Vec<NegotiationRestriction>,
+    restriction: NegotiationRestriction,
+) -> Result<(), ClientError> {
+    if policy == NegotiationPolicy::Strict {
+        return Err(ClientError::NegotiationRejected {
+            reason: restriction.message(),
+        });
+    }
+
+    restrictions.push(restriction);
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -127,17 +372,26 @@ mod tests {
 
     fn assert_rejected(requested: &Params, returned: &Params, policy: NegotiationPolicy) {
         assert!(matches!(
-            validate_negotiated_params(requested, returned, policy),
+            negotiate_params(requested, returned.clone(), policy),
             Err(ClientError::NegotiationRejected { .. })
         ));
     }
 
+    fn assert_negotiates(
+        requested: &Params,
+        returned: &Params,
+        policy: NegotiationPolicy,
+    ) -> NegotiatedParams {
+        negotiate_params(requested, returned.clone(), policy)
+            .unwrap_or_else(|err| panic!("expected negotiation success, got {err:?}"))
+    }
+
     fn rejection_reason(
         requested: &Params,
         returned: &Params,
         policy: NegotiationPolicy,
     ) -> String {
-        match validate_negotiated_params(requested, returned, policy) {
+        match negotiate_params(requested, returned.clone(), policy) {
             Err(ClientError::NegotiationRejected { reason }) => reason,
             other => panic!("expected negotiation rejection, got {other:?}"),
         }
@@ -149,27 +403,65 @@ mod tests {
 
         let mut returned = requested.clone();
         returned.length = 128;
-        assert_rejected(&requested, &returned, NegotiationPolicy::Strict);
+        assert_eq!(
+            rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
+            NegotiationRestriction::LengthReduced {
+                requested: requested.length,
+                negotiated: returned.length,
+            }
+            .message()
+        );
 
         let mut returned = requested.clone();
-        returned.dscp = 8;
+        returned.dscp = 0;
-        assert_rejected(&requested, &returned, NegotiationPolicy::Strict);
+        assert_eq!(
+            rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
+            NegotiationRestriction::DscpChanged {
+                requested: requested.dscp,
+                negotiated: returned.dscp,
+            }
+            .message()
+        );
 
         let mut returned = requested.clone();
         returned.received_stats = ReceivedStats::Count;
-        assert_rejected(&requested, &returned, NegotiationPolicy::Strict);
+        assert_eq!(
+            rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
+            NegotiationRestriction::ReceivedStatsChanged {
+                requested: requested.received_stats,
+                negotiated: returned.received_stats,
+            }
+            .message()
+        );
 
         let mut returned = requested.clone();
         returned.stamp_at = StampAt::Midpoint;
-        assert_rejected(&requested, &returned, NegotiationPolicy::Strict);
+        assert_eq!(
+            rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
+            NegotiationRestriction::StampAtChanged {
+                requested: requested.stamp_at,
+                negotiated: returned.stamp_at,
+            }
+            .message()
+        );
 
         let mut returned = requested.clone();
         returned.clock = Clock::Wall;
-        assert_rejected(&requested, &returned, NegotiationPolicy::Strict);
+        assert_eq!(
+            rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
+            NegotiationRestriction::ClockChanged {
+                requested: requested.clock,
+                negotiated: returned.clock,
+            }
+            .message()
+        );
 
         let mut returned = requested.clone();
         returned.server_fill = None;
-        assert_rejected(&requested, &returned, NegotiationPolicy::Strict);
+        assert_eq!(
+            rejection_reason(&requested, &returned, NegotiationPolicy::Strict),
+            NegotiationRestriction::ServerFillChanged.message()
+        );
     }
 
     #[test]
@@ -178,8 +470,13 @@ mod tests {
 
         let mut returned = requested.clone();
         returned.duration_ns = requested.duration_ns / 2;
-        assert!(
+        let negotiated = assert_negotiates(&requested, &returned, NegotiationPolicy::Loose);
-            validate_negotiated_params(&requested, &returned, NegotiationPolicy::Loose).is_ok()
+        assert_eq!(
+            negotiated.restrictions,
+            vec![NegotiationRestriction::DurationReduced {
+                requested_ns: requested.duration_ns,
+                negotiated_ns: returned.duration_ns,
+            }]
         );
 
         let mut returned = requested.clone();
@@ -200,12 +497,18 @@ mod tests {
         continuous_requested.duration_ns = 0;
         let mut finite_returned = continuous_requested.clone();
         finite_returned.duration_ns = 1_000_000_000;
-        assert!(validate_negotiated_params(
+        let negotiated = assert_negotiates(
             &continuous_requested,
             &finite_returned,
-            NegotiationPolicy::Loose
+            NegotiationPolicy::Loose,
-        )
+        );
-        .is_ok());
+        assert_eq!(
+            negotiated.restrictions,
+            vec![NegotiationRestriction::DurationReduced {
+                requested_ns: 0,
+                negotiated_ns: finite_returned.duration_ns,
+            }]
+        );
 
         assert_rejected(
             &continuous_requested,
@@ -213,9 +516,9 @@ mod tests {
             NegotiationPolicy::Strict,
         );
 
-        assert!(validate_negotiated_params(
+        assert!(negotiate_params(
-            &continuous_requested,
             &continuous_requested,
+            continuous_requested.clone(),
             NegotiationPolicy::Strict
         )
         .is_ok());

crates/irtt-client/tests/hmac.rs

@@ -5,7 +5,7 @@ use std::time::Duration;
 use irtt_client::{
     Client, ClientConfig, ClientError, ClientEvent, OpenOutcome, SocketConfig, WarningKind,
 };
-use irtt_proto::{echo_packet_len, ProtoError, TimestampFields};
+use irtt_proto::{ProtoError, TimestampFields};
 
 use support::{
     config_for_params, default_params, params_for_modes, run_one_probe_with_config,
@@ -13,6 +13,8 @@ use support::{
     start_hmac_required_open_drop_server, BackendPeer, ServerObservation, TOKEN,
 };
 
+use crate::support::test_echo_packet_len;
+
 #[test]
 fn hmac_open_success_negotiates_without_warnings() {
     let key = b"compat-secret".to_vec();
@@ -62,7 +64,7 @@ fn hmac_echo_success_verifies_request_and_accepts_reply() {
             token,
             sequence,
         } => {
-            assert_eq!(*len, echo_packet_len(true, &params));
+            assert_eq!(*len, test_echo_packet_len(true, &params));
             assert!(*hmac);
             assert_eq!(*token, TOKEN);
             assert_eq!(*sequence, 0);
@@ -78,7 +80,7 @@ fn hmac_echo_success_verifies_request_and_accepts_reply() {
             ..
         } => {
             assert_eq!(*seq, 0);
-            assert_eq!(*bytes, echo_packet_len(true, &params));
+            assert_eq!(*bytes, test_echo_packet_len(true, &params));
             assert!(server_timing.is_some());
         }
         other => panic!("expected EchoReply, got {other:?}"),

crates/irtt-client/tests/large_packets.rs

@@ -1,10 +1,12 @@
 mod support;
 
 use irtt_client::{Client, ClientConfig, ClientEvent, NegotiationPolicy};
-use irtt_proto::{echo_header_len, echo_packet_len, Params, TimestampFields};
+use irtt_proto::{echo_header_len, Params, TimestampFields};
 
 use support::{config_for_params, default_params, run_one_probe, BackendPeer, ServerObservation};
 
+use crate::support::test_echo_packet_len;
+
 fn params_with_length(length: i64) -> Params {
     Params {
         length,
@@ -42,7 +44,7 @@ fn assert_reply_bytes(event: &ClientEvent, expected: usize) {
 #[test]
 fn length_zero_baseline() {
     let params = params_with_length(0);
-    let expected = echo_packet_len(false, &params);
+    let expected = test_echo_packet_len(false, &params);
     let header = echo_header_len(false, &params);
     assert_eq!(expected, header);
 }
@@ -50,21 +52,21 @@ fn length_zero_baseline() {
 #[test]
 fn length_one_accepted() {
     let params = params_with_length(1);
-    let expected = echo_packet_len(false, &params);
+    let expected = test_echo_packet_len(false, &params);
     assert!(expected >= 1);
 }
 
 #[test]
 fn length_1472_accepted() {
     let params = params_with_length(1472);
-    let expected = echo_packet_len(false, &params);
+    let expected = test_echo_packet_len(false, &params);
     assert_eq!(expected, 1472);
 }
 
 #[test]
 fn length_4096_accepted() {
     let params = params_with_length(4096);
-    let expected = echo_packet_len(false, &params);
+    let expected = test_echo_packet_len(false, &params);
     assert_eq!(expected, 4096);
 }
 
@@ -74,7 +76,7 @@ fn length_4096_accepted() {
 fn echo_request_size_matches_echo_packet_len_for_zero_length() {
     let params = params_with_length(0);
     let run = run_one_probe(params.clone(), TimestampFields::default());
-    let expected = echo_packet_len(false, &params);
+    let expected = test_echo_packet_len(false, &params);
     assert_echo_size(&run, expected);
     assert_sent_bytes(&run.sent, expected);
 }
@@ -84,7 +86,7 @@ fn echo_request_size_matches_echo_packet_len_above_header() {
     for length in [16, 60, 128, 1472, 4096] {
         let params = params_with_length(length);
         let run = run_one_probe(params.clone(), TimestampFields::default());
-        let expected = echo_packet_len(false, &params);
+        let expected = test_echo_packet_len(false, &params);
         assert_eq!(
             expected,
             echo_header_len(false, &params).max(length as usize)
@@ -100,7 +102,7 @@ fn echo_request_size_matches_echo_packet_len_above_header() {
 fn echo_reply_decoded_for_large_length() {
     let params = params_with_length(4096);
     let run = run_one_probe(params.clone(), TimestampFields::default());
-    let expected = echo_packet_len(false, &params);
+    let expected = test_echo_packet_len(false, &params);
     assert_reply(&run.reply);
     assert_reply_bytes(&run.reply, expected);
 }
@@ -110,7 +112,7 @@ fn echo_reply_bytes_consistent_with_echo_packet_len_for_various_lengths() {
     for length in [0, 1472, 4096] {
         let params = params_with_length(length);
         let run = run_one_probe(params.clone(), TimestampFields::default());
-        let expected = echo_packet_len(false, &params);
+        let expected = test_echo_packet_len(false, &params);
         assert_reply_bytes(&run.reply, expected);
     }
 }

crates/irtt-client/tests/protocol_options.rs

@@ -6,7 +6,7 @@ use irtt_client::{
     Client, ClientConfig, ClientError, ClientEvent, OneWayDelaySample, ReceivedStatsSample,
     RttSample, ServerTiming,
 };
-use irtt_proto::{echo_packet_len, Clock, Params, ReceivedStats, StampAt, TimestampFields};
+use irtt_proto::{Clock, Params, ReceivedStats, StampAt, TimestampFields};
 
 use support::{
     config_for_params, default_params, params_for_modes, run_one_probe, run_one_probe_with_config,
@@ -14,6 +14,8 @@ use support::{
     ServerObservation, RECV_COUNT, RECV_WINDOW, TOKEN,
 };
 
+use crate::support::test_echo_packet_len;
+
 struct ReplyView<'a> {
     seq: u32,
     rtt: &'a RttSample,
@@ -37,7 +39,7 @@ fn received_stats_modes_drive_negotiated_echo_and_reply_events() {
 
         let reply = expect_echo_reply(&run.reply);
         assert_eq!(reply.seq, 0);
-        assert_eq!(reply.bytes, echo_packet_len(false, &params));
+        assert_eq!(reply.bytes, test_echo_packet_len(false, &params));
         assert!(reply.server_timing.is_none());
         assert!(reply.one_way.is_none());
         assert_received_stats(reply.received_stats, expected_count, expected_window);
@@ -146,7 +148,7 @@ fn hmac_rich_mode_uses_negotiated_echo_layout_and_decodes_reply() {
     assert_negotiated_echo_use(&run, &params, true);
 
     let reply = expect_echo_reply(&run.reply);
-    assert_eq!(reply.bytes, echo_packet_len(true, &params));
+    assert_eq!(reply.bytes, test_echo_packet_len(true, &params));
     assert_received_stats(reply.received_stats, Some(RECV_COUNT), Some(RECV_WINDOW));
     assert_timing_fields(reply.server_timing, true, true, false, false, true, true);
 }
@@ -183,7 +185,7 @@ fn loose_open_uses_returned_params_for_echo_layout_and_reply_parsing() {
     returned.received_stats = ReceivedStats::Count;
     returned.stamp_at = StampAt::Receive;
     returned.clock = Clock::Wall;
-    returned.dscp = 8;
+    returned.dscp = 0;
     returned.server_fill = None;
 
     let requested_for_config = requested.clone();
@@ -243,7 +245,7 @@ fn assert_negotiated_echo_use(run: &OneProbeRun, params: &Params, hmac: bool) {
 
 fn assert_echo_uses_params(run: &OneProbeRun, params: &Params, hmac: bool) {
     let (echo_len, echo_hmac, token, sequence) = echo_observation(run);
-    assert_eq!(echo_len, echo_packet_len(hmac, params));
+    assert_eq!(echo_len, test_echo_packet_len(hmac, params));
     assert_eq!(echo_hmac, hmac);
     assert_eq!(token, TOKEN);
     assert_eq!(sequence, 0);
@@ -251,7 +253,7 @@ fn assert_echo_uses_params(run: &OneProbeRun, params: &Params, hmac: bool) {
     match &run.sent {
         ClientEvent::EchoSent { seq, bytes, .. } => {
             assert_eq!(*seq, 0);
-            assert_eq!(*bytes, echo_packet_len(hmac, params));
+            assert_eq!(*bytes, test_echo_packet_len(hmac, params));
         }
         other => panic!("expected EchoSent, got {other:?}"),
     }

crates/irtt-client/tests/server_fill.rs

@@ -1,13 +1,15 @@
 mod support;
 
 use irtt_client::{Client, ClientConfig, ClientError, ClientEvent, NegotiationPolicy};
-use irtt_proto::{echo_packet_len, Params, TimestampFields};
+use irtt_proto::{Params, TimestampFields};
 
 use support::{
     config_for_params, default_params, run_one_probe, run_one_probe_with_config, server_fill,
     start_open_server, OneProbeRun, ServerObservation,
 };
 
+use crate::support::test_echo_packet_len;
+
 fn open_observation_params(run: &OneProbeRun) -> &Params {
     match &run.observations[0] {
         ServerObservation::Open { params, .. } => params,
@@ -202,11 +204,11 @@ fn server_fill_does_not_affect_client_side_packet_layout() {
     let (echo_len_without, _, _, _) = echo_observation(&run_without);
     assert_eq!(echo_len_with, echo_len_without);
     assert_eq!(
-        echo_packet_len(false, &run_with.negotiated.params),
+        test_echo_packet_len(false, &run_with.negotiated.params),
         echo_len_with
     );
     assert_eq!(
-        echo_packet_len(false, &run_without.negotiated.params),
+        test_echo_packet_len(false, &run_without.negotiated.params),
         echo_len_without
     );
 }

crates/irtt-client/tests/support/mod.rs

@@ -520,6 +520,11 @@ fn open_reply(flags: u8, token: u64, params: &Params, hmac_key: Option<&[u8]>) -
     packet
 }
 
+pub(super) fn test_echo_packet_len(hmac: bool, params: &Params) -> usize {
+    echo_packet_len(hmac, params)
+        .expect("managed runner test params must have a non-negative packet length")
+}
+
 fn echo_reply_packet(
     token: u64,
     seq: u32,
@@ -529,7 +534,7 @@ fn echo_reply_packet(
 ) -> Vec<u8> {
     let has_hmac = hmac_key.is_some();
     let layout = irtt_proto::PacketLayout::echo(has_hmac, params);
-    let packet_len = echo_packet_len(has_hmac, params);
+    let packet_len = test_echo_packet_len(has_hmac, params);
     let mut packet = Vec::with_capacity(packet_len);
 
     packet.extend_from_slice(&MAGIC);

crates/irtt-proto/src/echo.rs

@@ -1,7 +1,7 @@
 use crate::{
     flags::{has, FLAG_HMAC, FLAG_OPEN, FLAG_REPLY},
     hmac,
-    layout::{try_echo_packet_len, PacketLayout},
+    layout::{echo_packet_len, PacketLayout},
     open::{check_hmac_presence, reject},
     params::Params,
     validate_header, write_header, ProtoError, Result, HEADER_SIZE, HMAC_SIZE, RECV_COUNT_SIZE,
@@ -43,7 +43,7 @@ pub fn encode_echo_request(request: &EchoRequest, hmac_key: Option<&[u8]>) -> Re
         flags |= FLAG_HMAC;
     }
     let layout = PacketLayout::echo(hmac_key.is_some(), &request.params);
-    let len = try_echo_packet_len(hmac_key.is_some(), &request.params)?;
+    let len = echo_packet_len(hmac_key.is_some(), &request.params)?;
     let payload_offset = layout.header_len();
     let available_payload_len = len.saturating_sub(payload_offset);
     if request.payload.len() > available_payload_len {
@@ -87,7 +87,7 @@ pub fn decode_echo_reply(
             actual: packet.len(),
         });
     }
-    let expected_len = try_echo_packet_len(has(flags, FLAG_HMAC), params)?;
+    let expected_len = echo_packet_len(has(flags, FLAG_HMAC), params)?;
     if packet.len() != expected_len {
         return Err(ProtoError::PacketLengthMismatch {
             expected: expected_len,
@@ -371,7 +371,7 @@ mod tests {
         };
         params.received_stats = ReceivedStats::Both;
         let layout = PacketLayout::echo(true, &params);
-        let mut packet = Vec::with_capacity(try_echo_packet_len(true, &params).unwrap());
+        let mut packet = Vec::with_capacity(echo_packet_len(true, &params).unwrap());
         write_header(&mut packet, FLAG_REPLY | FLAG_HMAC);
         packet.extend_from_slice(&[0; HMAC_SIZE]);
         packet.extend_from_slice(&0x7896_b6ab_8771_5213u64.to_le_bytes());

crates/irtt-proto/src/layout.rs

@@ -111,13 +111,7 @@ pub fn echo_header_len(hmac: bool, params: &Params) -> usize {
     PacketLayout::echo(hmac, params).header_len()
 }
 
-pub fn echo_packet_len(hmac: bool, params: &Params) -> usize {
+pub fn echo_packet_len(hmac: bool, params: &Params) -> Result<usize> {
-    let header_len = echo_header_len(hmac, params);
-    let requested = usize::try_from(params.length).unwrap_or(0);
-    header_len.max(requested)
-}
-
-pub fn try_echo_packet_len(hmac: bool, params: &Params) -> Result<usize> {
     let header_len = echo_header_len(hmac, params);
     let requested =
         usize::try_from(params.length).map_err(|_| ProtoError::NegativePacketLength {
@@ -300,9 +294,9 @@ mod tests {
     fn negotiated_length_never_smaller_than_header() {
         let mut p = params(ReceivedStats::Both, StampAt::Both, Clock::Both);
         p.length = 20;
-        assert_eq!(echo_packet_len(false, &p), 60);
+        assert_eq!(echo_packet_len(false, &p).unwrap(), 60);
         p.length = 92;
-        assert_eq!(echo_packet_len(false, &p), 92);
+        assert_eq!(echo_packet_len(false, &p).unwrap(), 92);
     }
 
     #[test]
@@ -311,7 +305,7 @@ mod tests {
         p.length = -1;
 
         assert_eq!(
-            try_echo_packet_len(false, &p),
+            echo_packet_len(false, &p),
             Err(ProtoError::NegativePacketLength { length: -1 })
         );
     }
@@ -320,9 +314,9 @@ mod tests {
     fn checked_packet_len_preserves_non_negative_header_floor() {
         let mut p = params(ReceivedStats::Both, StampAt::Both, Clock::Both);
         p.length = 0;
-        assert_eq!(try_echo_packet_len(false, &p), Ok(60));
+        assert_eq!(echo_packet_len(false, &p), Ok(60));
 
         p.length = 92;
-        assert_eq!(try_echo_packet_len(false, &p), Ok(92));
+        assert_eq!(echo_packet_len(false, &p), Ok(92));
     }
 }

crates/irtt-proto/src/lib.rs

@@ -15,7 +15,7 @@ pub use echo::{decode_echo_reply, encode_echo_request, EchoReply, EchoRequest, T
 pub use error::{ProtoError, Result};
 pub use flags::*;
 pub use hmac::{compute_hmac, compute_hmac_in_place, verify_hmac};
-pub use layout::{echo_header_len, echo_packet_len, try_echo_packet_len, PacketLayout};
+pub use layout::{echo_header_len, echo_packet_len, PacketLayout};
 pub use open::{decode_open_reply, encode_open_request, OpenReply, OpenRequest};
 pub use params::{Clock, Params, ReceivedStats, ServerFill, StampAt};
 

crates/irtt-stats/src/core.rs

@@ -0,0 +1,249 @@
+use std::time::Duration;
+
+use crate::ipdv::{IpdvSample, IpdvTracker};
+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,
+};
+
+const CONTINUOUS_SEQUENCE_LIMIT: usize = 4096;
+
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) struct CoreStats {
+    events: EventCounts,
+    packets: PacketCounts,
+    send_call: TimeMetric,
+    timer_error: TimeMetric,
+    rtt_primary: TimeMetric,
+    rtt_raw: TimeMetric,
+    rtt_adjusted: TimeMetric,
+    ipdv_round_trip: TimeMetric,
+    ipdv_send: TimeMetric,
+    ipdv_receive: TimeMetric,
+    send_delay: TimeMetric,
+    receive_delay: TimeMetric,
+    server_processing: TimeMetric,
+    ipdv_tracker: IpdvTracker,
+}
+
+impl CoreStats {
+    pub(crate) fn new(sample_mode: SampleMode) -> Self {
+        let sequence_limit = if sample_mode == SampleMode::Exact {
+            None
+        } else {
+            Some(CONTINUOUS_SEQUENCE_LIMIT)
+        };
+
+        Self {
+            events: EventCounts::default(),
+            packets: PacketCounts::default(),
+            send_call: TimeMetric::new(false),
+            timer_error: TimeMetric::new(false),
+            rtt_primary: TimeMetric::new(sample_mode == SampleMode::Exact),
+            rtt_raw: TimeMetric::new(sample_mode == SampleMode::Exact),
+            rtt_adjusted: TimeMetric::new(sample_mode == SampleMode::Exact),
+            ipdv_round_trip: TimeMetric::new(sample_mode == SampleMode::Exact),
+            ipdv_send: TimeMetric::new(sample_mode == SampleMode::Exact),
+            ipdv_receive: TimeMetric::new(sample_mode == SampleMode::Exact),
+            send_delay: TimeMetric::new(sample_mode == SampleMode::Exact),
+            receive_delay: TimeMetric::new(sample_mode == SampleMode::Exact),
+            server_processing: TimeMetric::new(false),
+            ipdv_tracker: IpdvTracker::new(sequence_limit),
+        }
+    }
+
+    pub(crate) fn apply(&mut self, event: StatsEvent) -> EventStatsUpdate {
+        match event {
+            StatsEvent::Sent {
+                bytes,
+                send_call_ns,
+                timer_error_ns,
+                ..
+            } => {
+                self.apply_sent(bytes, send_call_ns, timer_error_ns);
+                EventStatsUpdate::default()
+            }
+            StatsEvent::UniqueReply {
+                is_late, sample, ..
+            } => self.apply_unique_reply(is_late, *sample),
+            StatsEvent::DuplicateReply { .. } => {
+                self.apply_duplicate_reply();
+                EventStatsUpdate::default()
+            }
+            StatsEvent::Loss { .. } => {
+                self.apply_loss();
+                EventStatsUpdate::default()
+            }
+            StatsEvent::Warning { .. } => {
+                self.apply_warning();
+                EventStatsUpdate::default()
+            }
+            StatsEvent::UntrackedLate { .. } => {
+                self.apply_untracked_late();
+                EventStatsUpdate::default()
+            }
+        }
+    }
+
+    fn apply_sent(&mut self, bytes: usize, send_call_ns: i128, timer_error_ns: i128) {
+        self.events.sent_events += 1;
+        self.packets.packets_sent += 1;
+        self.packets.bytes_sent = self.packets.bytes_sent.saturating_add(bytes as u64);
+        self.send_call.push_ns(send_call_ns);
+        self.timer_error.push_ns(timer_error_ns);
+    }
+
+    fn apply_unique_reply(&mut self, is_late: bool, sample: ReplySample) -> EventStatsUpdate {
+        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);
+
+        update
+    }
+
+    fn account_unique_reply(&mut self, is_late: bool, sample: &ReplySample) {
+        self.events.echo_replies += u64::from(!is_late);
+        self.events.late_unique_replies += u64::from(is_late);
+        self.packets.packets_received += 1;
+        self.packets.unique_replies += 1;
+        self.packets.late_packets += u64::from(is_late);
+        self.packets.bytes_received = self
+            .packets
+            .bytes_received
+            .saturating_add(sample.bytes as u64);
+
+        if let Some(count) = sample.received_count {
+            self.packets.server_packets_received = Some(u64::from(count));
+        }
+        if let Some(window) = sample.received_window {
+            self.packets.server_received_window = Some(window);
+        }
+    }
+
+    fn record_reply_metrics(&mut self, sample: &ReplySample) {
+        self.rtt_primary.push_ns(sample.ipdv.rtt_primary_ns);
+        self.rtt_raw.push_ns(sample.rtt_raw_ns);
+        if let Some(adjusted) = sample.rtt_adjusted_ns {
+            self.rtt_adjusted.push_ns(adjusted);
+        }
+        if let Some(processing) = sample.server_processing_ns {
+            self.server_processing.push_ns(processing);
+        }
+        if let Some(delay) = sample.send_delay_ns {
+            self.send_delay.push_ns(delay);
+        }
+        if let Some(delay) = sample.receive_delay_ns {
+            self.receive_delay.push_ns(delay);
+        }
+    }
+
+    fn apply_ipdv_sample(&mut self, sample: IpdvSample) -> Vec<IpdvPairUpdate> {
+        let mut updates = Vec::new();
+
+        for pair in self.ipdv_tracker.insert(sample) {
+            let Some(rtt_ipdv) = duration_from_non_negative_i128_ns(pair.rtt_ipdv_ns) else {
+                continue;
+            };
+            let send_ipdv = pair
+                .send_ipdv_ns
+                .and_then(duration_from_non_negative_i128_ns);
+            let receive_ipdv = pair
+                .receive_ipdv_ns
+                .and_then(duration_from_non_negative_i128_ns);
+
+            self.ipdv_round_trip.push_ns(pair.rtt_ipdv_ns);
+
+            if let Some(value) = pair.send_ipdv_ns {
+                self.ipdv_send.push_ns(value);
+            }
+
+            if let Some(value) = pair.receive_ipdv_ns {
+                self.ipdv_receive.push_ns(value);
+            }
+
+            updates.push(IpdvPairUpdate {
+                previous_seq: pair.previous_seq,
+                current_seq: pair.current_seq,
+                rtt_ipdv,
+                send_ipdv,
+                receive_ipdv,
+            });
+        }
+
+        updates
+    }
+
+    fn apply_duplicate_reply(&mut self) {
+        self.events.duplicate_replies += 1;
+        self.packets.packets_received += 1;
+        self.packets.duplicates += 1;
+    }
+
+    fn apply_loss(&mut self) {
+        self.events.loss_events += 1;
+    }
+
+    fn apply_warning(&mut self) {
+        self.events.warning_events += 1;
+    }
+
+    fn apply_untracked_late(&mut self) {
+        self.events.untracked_late_replies += 1;
+    }
+
+    pub(crate) fn snapshot(&self) -> Snapshot {
+        let packets = self.packets;
+        Snapshot {
+            events: self.events,
+            packets,
+            loss: loss_stats(packets),
+            send_call: self.send_call.stats(),
+            timer_error: self.timer_error.stats(),
+            rtt: self.rtt_stats(),
+            ipdv: self.ipdv_stats(),
+            one_way_delay: self.one_way_delay_stats(),
+            server_processing: self.server_processing_stats(),
+        }
+    }
+
+    fn rtt_stats(&self) -> RttStats {
+        RttStats {
+            primary: self.rtt_primary.stats(),
+            raw: self.rtt_raw.stats(),
+            adjusted: self.rtt_adjusted.stats(),
+        }
+    }
+
+    fn ipdv_stats(&self) -> IpdvStats {
+        IpdvStats {
+            round_trip: self.ipdv_round_trip.stats(),
+            send: self.ipdv_send.stats(),
+            receive: self.ipdv_receive.stats(),
+        }
+    }
+
+    fn one_way_delay_stats(&self) -> OneWayDelayStats {
+        OneWayDelayStats {
+            send_delay: self.send_delay.stats(),
+            receive_delay: self.receive_delay.stats(),
+        }
+    }
+
+    fn server_processing_stats(&self) -> ServerProcessingStats {
+        ServerProcessingStats {
+            processing: self.server_processing.stats(),
+        }
+    }
+}
+
+fn duration_from_non_negative_i128_ns(value: i128) -> Option<Duration> {
+    u64::try_from(value).ok().map(Duration::from_nanos)
+}

crates/irtt-stats/src/ipdv.rs

@@ -0,0 +1,263 @@
+use std::{
+    collections::{HashMap, HashSet, VecDeque},
+    time::{Duration, Instant},
+};
+
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) struct IpdvSample {
+    pub(crate) seq: u32,
+    pub(crate) rtt_primary_ns: i128,
+    pub(crate) client_send_mono: Instant,
+    pub(crate) client_receive_mono: Instant,
+    pub(crate) client_send_wall_ns: Option<i128>,
+    pub(crate) client_receive_wall_ns: Option<i128>,
+    pub(crate) server_receive_mono_ns: Option<i64>,
+    pub(crate) server_send_mono_ns: Option<i64>,
+    pub(crate) server_receive_wall_ns: Option<i64>,
+    pub(crate) server_send_wall_ns: Option<i64>,
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) struct IpdvTracker {
+    samples: HashMap<u32, IpdvSample>,
+    sample_order: VecDeque<u32>,
+    completed_pairs: HashSet<u32>,
+    sequence_limit: Option<usize>,
+}
+
+impl IpdvTracker {
+    pub(crate) fn new(sequence_limit: Option<usize>) -> Self {
+        Self {
+            samples: HashMap::new(),
+            sample_order: VecDeque::new(),
+            completed_pairs: HashSet::new(),
+            sequence_limit,
+        }
+    }
+
+    pub(crate) fn insert(&mut self, sample: IpdvSample) -> Vec<CompletedIpdvPair> {
+        let seq = sample.seq;
+        if self.samples.insert(seq, sample).is_some() {
+            return Vec::new();
+        }
+
+        self.sample_order.push_back(seq);
+        self.enforce_sequence_limit();
+
+        let mut pairs = Vec::with_capacity(2);
+        if let Some(pair) = self.try_pair(seq) {
+            pairs.push(pair);
+        }
+        if let Some(next) = seq.checked_add(1) {
+            if let Some(pair) = self.try_pair(next) {
+                pairs.push(pair);
+            }
+        }
+        pairs
+    }
+
+    fn enforce_sequence_limit(&mut self) {
+        let Some(limit) = self.sequence_limit else {
+            return;
+        };
+        while self.samples.len() > limit {
+            let Some(seq) = self.sample_order.pop_front() else {
+                break;
+            };
+            if self.samples.remove(&seq).is_some() {
+                self.completed_pairs.remove(&seq);
+                if let Some(next) = seq.checked_add(1) {
+                    self.completed_pairs.remove(&next);
+                }
+            }
+        }
+    }
+
+    fn try_pair(&mut self, current_seq: u32) -> Option<CompletedIpdvPair> {
+        let previous_seq = current_seq.checked_sub(1)?;
+
+        if !self.completed_pairs.insert(current_seq) {
+            return None;
+        }
+
+        let Some(previous) = self.samples.get(&previous_seq) else {
+            self.completed_pairs.remove(&current_seq);
+            return None;
+        };
+
+        let Some(current) = self.samples.get(&current_seq) else {
+            self.completed_pairs.remove(&current_seq);
+            return None;
+        };
+
+        Some(CompletedIpdvPair {
+            previous_seq,
+            current_seq,
+            rtt_ipdv_ns: abs_i128_ns(current.rtt_primary_ns - previous.rtt_primary_ns),
+            send_ipdv_ns: send_ipdv_ns(previous, current).map(abs_i128_ns),
+            receive_ipdv_ns: receive_ipdv_ns(previous, current).map(abs_i128_ns),
+        })
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct CompletedIpdvPair {
+    pub(crate) previous_seq: u32,
+    pub(crate) current_seq: u32,
+    pub(crate) rtt_ipdv_ns: i128,
+    pub(crate) send_ipdv_ns: Option<i128>,
+    pub(crate) receive_ipdv_ns: Option<i128>,
+}
+
+fn send_ipdv_ns(previous: &IpdvSample, current: &IpdvSample) -> Option<i128> {
+    if let (Some(prev_server), Some(cur_server)) = (
+        previous.server_receive_mono_ns,
+        current.server_receive_mono_ns,
+    ) {
+        return Some(
+            i128::from(cur_server)
+                - i128::from(prev_server)
+                - instant_diff_ns(current.client_send_mono, previous.client_send_mono),
+        );
+    }
+    if let (Some(prev_server), Some(cur_server), Some(prev_client), Some(cur_client)) = (
+        previous.server_receive_wall_ns,
+        current.server_receive_wall_ns,
+        previous.client_send_wall_ns,
+        current.client_send_wall_ns,
+    ) {
+        return Some(i128::from(cur_server) - i128::from(prev_server) - (cur_client - prev_client));
+    }
+    None
+}
+
+fn receive_ipdv_ns(previous: &IpdvSample, current: &IpdvSample) -> Option<i128> {
+    if let (Some(prev_server), Some(cur_server)) =
+        (previous.server_send_mono_ns, current.server_send_mono_ns)
+    {
+        return Some(
+            instant_diff_ns(current.client_receive_mono, previous.client_receive_mono)
+                - (i128::from(cur_server) - i128::from(prev_server)),
+        );
+    }
+    if let (Some(prev_server), Some(cur_server), Some(prev_client), Some(cur_client)) = (
+        previous.server_send_wall_ns,
+        current.server_send_wall_ns,
+        previous.client_receive_wall_ns,
+        current.client_receive_wall_ns,
+    ) {
+        return Some(
+            (cur_client - prev_client) - (i128::from(cur_server) - i128::from(prev_server)),
+        );
+    }
+    None
+}
+
+fn instant_diff_ns(current: Instant, previous: Instant) -> i128 {
+    if let Some(diff) = current.checked_duration_since(previous) {
+        duration_ns_i128(diff)
+    } else {
+        -duration_ns_i128(previous.duration_since(current))
+    }
+}
+
+fn duration_ns_i128(duration: Duration) -> i128 {
+    i128::try_from(duration.as_nanos()).unwrap_or(i128::MAX)
+}
+
+fn abs_i128_ns(value: i128) -> i128 {
+    value.saturating_abs()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn ipdv_sample(seq: u32, rtt_primary_ns: i128) -> IpdvSample {
+        let now = Instant::now();
+        IpdvSample {
+            seq,
+            rtt_primary_ns,
+            client_send_mono: now,
+            client_receive_mono: now,
+            client_send_wall_ns: None,
+            client_receive_wall_ns: None,
+            server_receive_mono_ns: None,
+            server_send_mono_ns: None,
+            server_receive_wall_ns: None,
+            server_send_wall_ns: None,
+        }
+    }
+
+    #[test]
+    fn ipdv_tracker_completes_adjacent_pair() {
+        let mut tracker = IpdvTracker::new(None);
+        assert!(tracker.insert(ipdv_sample(0, 10)).is_empty());
+
+        let pairs = tracker.insert(ipdv_sample(1, 14));
+
+        assert_eq!(
+            pairs,
+            vec![CompletedIpdvPair {
+                previous_seq: 0,
+                current_seq: 1,
+                rtt_ipdv_ns: 4,
+                send_ipdv_ns: None,
+                receive_ipdv_ns: None,
+            }]
+        );
+    }
+
+    #[test]
+    fn ipdv_tracker_gap_fill_completes_both_adjacent_pairs() {
+        let mut tracker = IpdvTracker::new(None);
+        assert!(tracker.insert(ipdv_sample(0, 10)).is_empty());
+        assert!(tracker.insert(ipdv_sample(2, 20)).is_empty());
+
+        let pairs = tracker.insert(ipdv_sample(1, 13));
+
+        assert_eq!(
+            pairs,
+            vec![
+                CompletedIpdvPair {
+                    previous_seq: 0,
+                    current_seq: 1,
+                    rtt_ipdv_ns: 3,
+                    send_ipdv_ns: None,
+                    receive_ipdv_ns: None,
+                },
+                CompletedIpdvPair {
+                    previous_seq: 1,
+                    current_seq: 2,
+                    rtt_ipdv_ns: 7,
+                    send_ipdv_ns: None,
+                    receive_ipdv_ns: None,
+                },
+            ]
+        );
+    }
+
+    #[test]
+    fn ipdv_tracker_duplicate_sequence_does_not_emit_pair_again() {
+        let mut tracker = IpdvTracker::new(None);
+        assert!(tracker.insert(ipdv_sample(0, 10)).is_empty());
+        assert_eq!(tracker.insert(ipdv_sample(1, 14)).len(), 1);
+
+        assert!(tracker.insert(ipdv_sample(1, 18)).is_empty());
+    }
+
+    #[test]
+    fn ipdv_tracker_bounded_mode_evicts_old_sequence_state() {
+        let mut tracker = IpdvTracker::new(Some(2));
+        assert!(tracker.insert(ipdv_sample(0, 10)).is_empty());
+        assert_eq!(tracker.insert(ipdv_sample(1, 14)).len(), 1);
+
+        let pairs = tracker.insert(ipdv_sample(2, 20));
+
+        assert_eq!(pairs.len(), 1);
+        assert!(!tracker.samples.contains_key(&0));
+        assert!(!tracker.completed_pairs.contains(&0));
+        assert!(!tracker.completed_pairs.contains(&1));
+        assert!(tracker.completed_pairs.contains(&2));
+    }
+}

crates/irtt-stats/src/loss.rs

@@ -0,0 +1,106 @@
+use crate::PacketCounts;
+
+#[derive(Debug, Clone, Copy, PartialEq)]
+/// Packet loss, duplicate, and late-packet statistics.
+///
+/// Percentage fields are percentages from `0.0` to `100.0` in normal cases.
+/// Directional upstream and downstream loss can be negative when
+/// server-reported counts exceed local expectations.
+pub struct LossStats {
+    /// Locally inferred total lost packets.
+    pub lost_packets: u64,
+    /// Server-assisted upstream loss estimate, when server counts are available.
+    pub upstream_loss_packets: Option<i128>,
+    /// Server-assisted downstream loss estimate, when server counts are available.
+    pub downstream_loss_packets: Option<i128>,
+    /// Total packet loss percentage.
+    pub packet_loss_percent: f64,
+    /// Server-assisted upstream loss percentage.
+    pub upstream_loss_percent: f64,
+    /// Server-assisted downstream loss percentage.
+    pub downstream_loss_percent: f64,
+    /// Duplicate reply percentage.
+    pub duplicate_percent: f64,
+    /// Late reply packet percentage.
+    pub late_packets_percent: f64,
+}
+
+pub(crate) fn loss_stats(packets: PacketCounts) -> LossStats {
+    let lost = packets.packets_sent.saturating_sub(packets.unique_replies);
+    let packet_loss_percent = if packets.packets_sent == 0 {
+        0.0
+    } else if packets.unique_replies == 0 {
+        100.0
+    } else {
+        percent(lost as f64, packets.packets_sent as f64)
+    };
+
+    let (
+        upstream_loss_packets,
+        upstream_loss_percent,
+        downstream_loss_packets,
+        downstream_loss_percent,
+    ) = if let Some(server_received) = packets.server_packets_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 {
+            0.0
+        } else {
+            percent(upstream as f64, packets.packets_sent as f64)
+        };
+        let downstream_percent = if server_received == 0 {
+            0.0
+        } else {
+            percent(downstream as f64, server_received as f64)
+        };
+        (
+            Some(upstream),
+            upstream_percent,
+            Some(downstream),
+            downstream_percent,
+        )
+    } else {
+        (None, 0.0, None, 0.0)
+    };
+
+    LossStats {
+        lost_packets: lost,
+        upstream_loss_packets,
+        downstream_loss_packets,
+        packet_loss_percent,
+        upstream_loss_percent,
+        downstream_loss_percent,
+        duplicate_percent: if packets.packets_received == 0 {
+            0.0
+        } else {
+            percent(packets.duplicates as f64, packets.packets_received as f64)
+        },
+        late_packets_percent: if packets.packets_received == 0 {
+            0.0
+        } else {
+            percent(packets.late_packets as f64, packets.packets_received as f64)
+        },
+    }
+}
+
+fn percent(numerator: f64, denominator: f64) -> f64 {
+    100.0 * numerator / denominator
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn duplicate_and_late_percentages_use_packets_received_denominator() {
+        let loss = loss_stats(PacketCounts {
+            packets_received: 4,
+            duplicates: 1,
+            late_packets: 2,
+            ..PacketCounts::default()
+        });
+
+        assert_eq!(loss.duplicate_percent, 25.0);
+        assert_eq!(loss.late_packets_percent, 50.0);
+    }
+}

crates/irtt-stats/src/normalization.rs

@@ -0,0 +1,214 @@
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use irtt_client::{
+    ClientEvent, ClientTimestamp, OneWayDelaySample, ReceivedStatsSample, RttSample, ServerTiming,
+    SignedDuration,
+};
+
+use crate::ipdv::IpdvSample;
+
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) enum StatsEvent {
+    Sent {
+        at: Instant,
+        bytes: usize,
+        send_call_ns: i128,
+        timer_error_ns: i128,
+    },
+    UniqueReply {
+        at: Instant,
+        is_late: bool,
+        sample: Box<ReplySample>,
+    },
+    DuplicateReply {
+        at: Instant,
+    },
+    Loss {
+        at: Instant,
+    },
+    Warning {
+        at: Instant,
+    },
+    UntrackedLate {
+        at: Instant,
+    },
+}
+
+pub(crate) fn normalize_event(event: &ClientEvent) -> Option<StatsEvent> {
+    match event {
+        ClientEvent::EchoSent {
+            sent_at,
+            bytes,
+            send_call,
+            timer_error,
+            ..
+        } => Some(StatsEvent::Sent {
+            at: sent_at.mono,
+            bytes: *bytes,
+            send_call_ns: duration_ns_i128(*send_call),
+            timer_error_ns: duration_ns_i128(*timer_error),
+        }),
+        ClientEvent::EchoReply {
+            seq,
+            sent_at,
+            received_at,
+            rtt,
+            server_timing,
+            one_way,
+            received_stats,
+            bytes,
+            ..
+        } => Some(StatsEvent::UniqueReply {
+            at: received_at.mono,
+            is_late: false,
+            sample: Box::new(ReplySample::from_reply_parts(
+                *seq,
+                *sent_at,
+                *received_at,
+                *rtt,
+                *server_timing,
+                *one_way,
+                *received_stats,
+                *bytes,
+            )),
+        }),
+        ClientEvent::LateReply {
+            seq,
+            sent_at: Some(sent_at),
+            received_at,
+            rtt: Some(rtt),
+            server_timing,
+            one_way,
+            received_stats,
+            bytes,
+            ..
+        } => Some(StatsEvent::UniqueReply {
+            at: received_at.mono,
+            is_late: true,
+            sample: Box::new(ReplySample::from_reply_parts(
+                *seq,
+                *sent_at,
+                *received_at,
+                *rtt,
+                *server_timing,
+                *one_way,
+                *received_stats,
+                *bytes,
+            )),
+        }),
+        ClientEvent::LateReply { received_at, .. } => Some(StatsEvent::UntrackedLate {
+            at: received_at.mono,
+        }),
+        ClientEvent::DuplicateReply { received_at, .. } => Some(StatsEvent::DuplicateReply {
+            at: received_at.mono,
+        }),
+        ClientEvent::EchoLoss { timeout_at, .. } => Some(StatsEvent::Loss { at: *timeout_at }),
+        ClientEvent::Warning { at, .. } => Some(StatsEvent::Warning { at: at.mono }),
+        ClientEvent::SessionStarted { .. }
+        | ClientEvent::NoTestCompleted { .. }
+        | ClientEvent::SessionClosed { .. } => None,
+    }
+}
+
+impl StatsEvent {
+    pub(crate) fn at(&self) -> Instant {
+        match self {
+            Self::Sent { at, .. }
+            | Self::UniqueReply { at, .. }
+            | Self::DuplicateReply { at }
+            | Self::Loss { at }
+            | Self::Warning { at }
+            | Self::UntrackedLate { at } => *at,
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) struct ReplySample {
+    pub(crate) bytes: usize,
+    pub(crate) rtt_raw_ns: i128,
+    pub(crate) rtt_adjusted_ns: Option<i128>,
+    pub(crate) send_delay_ns: Option<i128>,
+    pub(crate) receive_delay_ns: Option<i128>,
+    pub(crate) server_processing_ns: Option<i128>,
+    pub(crate) received_count: Option<u32>,
+    pub(crate) received_window: Option<u64>,
+    pub(crate) ipdv: IpdvSample,
+}
+
+impl ReplySample {
+    #[allow(clippy::too_many_arguments)]
+    fn from_reply_parts(
+        seq: u32,
+        sent_at: ClientTimestamp,
+        received_at: ClientTimestamp,
+        rtt: RttSample,
+        server_timing: Option<ServerTiming>,
+        one_way: Option<OneWayDelaySample>,
+        received_stats: Option<ReceivedStatsSample>,
+        bytes: usize,
+    ) -> Self {
+        let rtt_primary_ns = signed_duration_ns(rtt.effective_signed);
+        Self {
+            bytes,
+            rtt_raw_ns: duration_ns_i128(rtt.raw),
+            rtt_adjusted_ns: rtt.adjusted_signed.map(signed_duration_ns),
+            send_delay_ns: one_way
+                .and_then(|sample| sample.client_to_server)
+                .map(duration_ns_i128),
+            receive_delay_ns: one_way
+                .and_then(|sample| sample.server_to_client)
+                .map(duration_ns_i128),
+            server_processing_ns: server_timing
+                .and_then(|timing| timing.processing)
+                .map(duration_ns_i128),
+            received_count: received_stats.and_then(|stats| stats.count),
+            received_window: received_stats.and_then(|stats| stats.window),
+            ipdv: IpdvSample {
+                seq,
+                rtt_primary_ns,
+                client_send_mono: sent_at.mono,
+                client_receive_mono: received_at.mono,
+                client_send_wall_ns: system_time_ns(sent_at.wall),
+                client_receive_wall_ns: system_time_ns(received_at.wall),
+                server_receive_mono_ns: server_timing.and_then(|timing| timing.receive_mono_ns),
+                server_send_mono_ns: server_timing.and_then(|timing| timing.send_mono_ns),
+                server_receive_wall_ns: server_timing.and_then(|timing| timing.receive_wall_ns),
+                server_send_wall_ns: server_timing.and_then(|timing| timing.send_wall_ns),
+            },
+        }
+    }
+}
+
+fn system_time_ns(time: SystemTime) -> Option<i128> {
+    if let Ok(duration) = time.duration_since(UNIX_EPOCH) {
+        return Some(duration_ns_i128(duration));
+    }
+    UNIX_EPOCH
+        .duration_since(time)
+        .ok()
+        .map(|duration| -duration_ns_i128(duration))
+}
+
+fn duration_ns_i128(duration: Duration) -> i128 {
+    i128::try_from(duration.as_nanos()).unwrap_or(i128::MAX)
+}
+
+fn signed_duration_ns(duration: SignedDuration) -> i128 {
+    duration.ns
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn system_time_before_epoch_is_supported() {
+        let before = UNIX_EPOCH - Duration::from_nanos(7);
+        assert_eq!(system_time_ns(before), Some(-7));
+        let after = UNIX_EPOCH + Duration::from_nanos(7);
+        assert_eq!(system_time_ns(after), Some(7));
+        let now = SystemTime::now();
+        assert!(system_time_ns(now).is_some());
+    }
+}

crates/irtt-stats/src/rolling.rs

@@ -0,0 +1,57 @@
+use std::{collections::VecDeque, time::Duration};
+
+use crate::{core::CoreStats, normalization::StatsEvent, SampleMode, Snapshot, StatsConfig};
+
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) struct RollingEvents {
+    count_limit: Option<usize>,
+    time_limit: Option<Duration>,
+    count_events: Option<VecDeque<StatsEvent>>,
+    time_events: Option<VecDeque<StatsEvent>>,
+}
+
+impl RollingEvents {
+    pub(crate) fn new(config: StatsConfig) -> Self {
+        Self {
+            count_limit: config.rolling_count,
+            time_limit: config.rolling_time,
+            count_events: config.rolling_count.map(|_| VecDeque::new()),
+            time_events: config.rolling_time.map(|_| VecDeque::new()),
+        }
+    }
+
+    pub(crate) fn push(&mut self, event: StatsEvent) {
+        if let (Some(limit), Some(window)) = (self.count_limit, self.count_events.as_mut()) {
+            window.push_back(event.clone());
+            while window.len() > limit {
+                window.pop_front();
+            }
+        }
+
+        if let (Some(duration), Some(window)) = (self.time_limit, self.time_events.as_mut()) {
+            let cutoff = event.at().checked_sub(duration);
+            window.push_back(event);
+            if let Some(cutoff) = cutoff {
+                while window.front().is_some_and(|event| event.at() < cutoff) {
+                    window.pop_front();
+                }
+            }
+        }
+    }
+
+    pub(crate) fn count_snapshot(&self) -> Option<Snapshot> {
+        self.count_events.as_ref().map(snapshot_window)
+    }
+
+    pub(crate) fn time_snapshot(&self) -> Option<Snapshot> {
+        self.time_events.as_ref().map(snapshot_window)
+    }
+}
+
+fn snapshot_window(events: &VecDeque<StatsEvent>) -> Snapshot {
+    let mut core = CoreStats::new(SampleMode::RunningOnly);
+    for event in events {
+        core.apply(event.clone());
+    }
+    core.snapshot()
+}

crates/irtt-stats/src/time_stats.rs

@@ -0,0 +1,157 @@
+#[derive(Debug, Clone, Copy, PartialEq)]
+/// Summary statistics for timing values.
+///
+/// All fields use nanoseconds except `variance_ns2`, which is nanoseconds
+/// squared. Values may represent signed timing quantities such as scheduling
+/// error or one-way delay.
+pub struct TimeStats {
+    /// Number of samples included in this summary.
+    pub count: u64,
+    /// Sum of all samples, in nanoseconds.
+    pub total_ns: i128,
+    /// Smallest sample, in nanoseconds.
+    pub min_ns: Option<i128>,
+    /// Largest sample, in nanoseconds.
+    pub max_ns: Option<i128>,
+    /// Arithmetic mean, in nanoseconds.
+    pub mean_ns: f64,
+    /// Median, in nanoseconds, when exact samples were retained.
+    pub median_ns: Option<f64>,
+    /// Sample variance, in nanoseconds squared.
+    pub variance_ns2: f64,
+}
+
+impl TimeStats {
+    /// Returns the sample standard deviation, in nanoseconds.
+    pub fn stddev_ns(&self) -> f64 {
+        self.variance_ns2.sqrt()
+    }
+
+    /// Returns whether this summary contains no samples.
+    pub fn is_empty(&self) -> bool {
+        self.count == 0
+    }
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) struct TimeMetric {
+    running: RunningStats,
+    samples: Option<Vec<i128>>,
+}
+
+impl TimeMetric {
+    pub(crate) fn new(retain_samples: bool) -> Self {
+        Self {
+            running: RunningStats::default(),
+            samples: retain_samples.then(Vec::new),
+        }
+    }
+
+    pub(crate) fn push_ns(&mut self, value: i128) {
+        self.running.push(value);
+        if let Some(samples) = self.samples.as_mut() {
+            samples.push(value);
+        }
+    }
+
+    pub(crate) fn stats(&self) -> TimeStats {
+        self.running
+            .stats(self.samples.as_ref().and_then(|samples| median_ns(samples)))
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Default)]
+struct RunningStats {
+    count: u64,
+    total_ns: i128,
+    min_ns: Option<i128>,
+    max_ns: Option<i128>,
+    mean_ns: f64,
+    m2_ns2: f64,
+}
+
+impl RunningStats {
+    fn push(&mut self, value: i128) {
+        self.count += 1;
+        self.total_ns = self.total_ns.saturating_add(value);
+        self.min_ns = Some(self.min_ns.map_or(value, |min| min.min(value)));
+        self.max_ns = Some(self.max_ns.map_or(value, |max| max.max(value)));
+        let x = value as f64;
+        let delta = x - self.mean_ns;
+        self.mean_ns += delta / self.count as f64;
+        let delta2 = x - self.mean_ns;
+        self.m2_ns2 += delta * delta2;
+    }
+
+    fn stats(&self, median_ns: Option<f64>) -> TimeStats {
+        TimeStats {
+            count: self.count,
+            total_ns: self.total_ns,
+            min_ns: self.min_ns,
+            max_ns: self.max_ns,
+            mean_ns: if self.count == 0 { 0.0 } else { self.mean_ns },
+            median_ns,
+            variance_ns2: sample_variance(self.count, self.m2_ns2),
+        }
+    }
+}
+
+fn sample_variance(count: u64, m2: f64) -> f64 {
+    if count < 2 {
+        0.0
+    } else {
+        m2 / (count - 1) as f64
+    }
+}
+
+fn median_ns(samples: &[i128]) -> Option<f64> {
+    if samples.is_empty() {
+        return None;
+    }
+    let mut sorted = samples.to_vec();
+    sorted.sort_unstable();
+    let mid = sorted.len() / 2;
+    Some(if sorted.len() % 2 == 1 {
+        sorted[mid] as f64
+    } else {
+        (sorted[mid - 1] as f64 + sorted[mid] as f64) / 2.0
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn running_duration_stats_use_sample_variance() {
+        let mut metric = TimeMetric::new(false);
+        metric.push_ns(1);
+        metric.push_ns(2);
+        metric.push_ns(3);
+        let stats = metric.stats();
+        assert_eq!(stats.count, 3);
+        assert_eq!(stats.total_ns, 6);
+        assert_eq!(stats.min_ns, Some(1));
+        assert_eq!(stats.max_ns, Some(3));
+        assert_eq!(stats.mean_ns, 2.0);
+        assert_eq!(stats.variance_ns2, 1.0);
+        assert_eq!(stats.stddev_ns(), 1.0);
+    }
+
+    #[test]
+    fn exact_median_handles_odd_and_even_samples() {
+        assert_eq!(median_ns(&[3, 1, 2]), Some(2.0));
+        assert_eq!(median_ns(&[4, 1, 2, 3]), Some(2.5));
+        assert_eq!(median_ns(&[-5, 1, 3]), Some(1.0));
+        assert_eq!(median_ns(&[-5, 1, 3, 7]), Some(2.0));
+    }
+
+    #[test]
+    fn single_sample_stddev_is_zero() {
+        let mut metric = TimeMetric::new(false);
+        metric.push_ns(42);
+        let stats = metric.stats();
+        assert_eq!(stats.variance_ns2, 0.0);
+        assert_eq!(stats.stddev_ns(), 0.0);
+    }
+}

crates/irtt-stats/tests/common/mod.rs

@@ -0,0 +1,114 @@
+use std::time::{Duration, Instant, UNIX_EPOCH};
+
+use irtt_client::{
+    ClientEvent, ClientTimestamp, OneWayDelaySample, PacketMeta, ReceivedStatsSample, RttSample,
+    ServerTiming, SignedDuration,
+};
+
+pub fn ts(ms: u64) -> ClientTimestamp {
+    ClientTimestamp {
+        mono: Instant::now() + Duration::from_millis(ms),
+        wall: UNIX_EPOCH + Duration::from_millis(ms),
+    }
+}
+
+pub fn rtt(raw_ms: u64, effective_ms: i128) -> RttSample {
+    RttSample {
+        raw: Duration::from_millis(raw_ms),
+        adjusted: u64::try_from(effective_ms).ok().map(Duration::from_millis),
+        effective: u64::try_from(effective_ms)
+            .ok()
+            .map(Duration::from_millis)
+            .unwrap_or_else(|| Duration::from_millis(raw_ms)),
+        adjusted_signed: Some(SignedDuration {
+            ns: effective_ms * 1_000_000,
+        }),
+        effective_signed: SignedDuration {
+            ns: effective_ms * 1_000_000,
+        },
+    }
+}
+
+pub fn sent(seq: u32, sent_at: ClientTimestamp) -> ClientEvent {
+    ClientEvent::EchoSent {
+        seq,
+        remote: "127.0.0.1:2112".parse().unwrap(),
+        scheduled_at: sent_at.mono,
+        sent_at,
+        bytes: 32,
+        send_call: Duration::from_micros(10),
+        timer_error: Duration::from_micros(2),
+    }
+}
+
+pub fn reply(seq: u32, raw_ms: u64, effective_ms: i128) -> ClientEvent {
+    let sent_at = ts(seq as u64 * 10);
+    let received_at = ClientTimestamp {
+        mono: sent_at.mono + Duration::from_millis(raw_ms),
+        wall: sent_at.wall + Duration::from_millis(raw_ms),
+    };
+    ClientEvent::EchoReply {
+        seq,
+        remote: "127.0.0.1:2112".parse().unwrap(),
+        sent_at,
+        received_at,
+        rtt: rtt(raw_ms, effective_ms),
+        server_timing: Some(ServerTiming {
+            receive_wall_ns: Some(unix_time_ns_after_epoch(sent_at.wall) as i64 + 1_000_000),
+            receive_mono_ns: Some(seq as i64 * 10_000_000 + 1_000_000),
+            send_wall_ns: Some(unix_time_ns_after_epoch(sent_at.wall) as i64 + 2_000_000),
+            send_mono_ns: Some(seq as i64 * 10_000_000 + 2_000_000),
+            midpoint_wall_ns: None,
+            midpoint_mono_ns: None,
+            processing: Some(Duration::from_millis(1)),
+        }),
+        one_way: Some(OneWayDelaySample {
+            client_to_server: Some(Duration::from_millis(1)),
+            server_to_client: Some(Duration::from_millis(2)),
+        }),
+        received_stats: Some(ReceivedStatsSample {
+            count: Some(seq + 1),
+            window: Some(0xff),
+        }),
+        bytes: 64,
+        packet_meta: PacketMeta::default(),
+    }
+}
+
+pub fn late_reply(seq: u32, raw_ms: u64, effective_ms: i128) -> ClientEvent {
+    let ClientEvent::EchoReply {
+        seq,
+        remote,
+        sent_at,
+        received_at,
+        rtt,
+        server_timing,
+        one_way,
+        received_stats,
+        bytes,
+        packet_meta,
+        ..
+    } = reply(seq, raw_ms, effective_ms)
+    else {
+        unreachable!();
+    };
+    ClientEvent::LateReply {
+        seq,
+        highest_seen: seq + 1,
+        remote,
+        sent_at: Some(sent_at),
+        received_at,
+        rtt: Some(rtt),
+        server_timing,
+        one_way,
+        received_stats,
+        bytes,
+        packet_meta,
+    }
+}
+
+fn unix_time_ns_after_epoch(time: std::time::SystemTime) -> i128 {
+    time.duration_since(UNIX_EPOCH)
+        .map(|duration| duration.as_nanos() as i128)
+        .unwrap()
+}

crates/irtt-stats/tests/stats.rs

@@ -0,0 +1,272 @@
+mod common;
+
+use std::time::{Duration, Instant};
+
+use common::{late_reply, reply, sent, ts};
+use irtt_client::{ClientEvent, PacketMeta};
+use irtt_stats::{EventStatsUpdate, IpdvPairUpdate, SampleMode, StatsCollector, StatsConfig};
+
+#[test]
+fn running_only_samples_avoid_finite_retention() {
+    let mut collector = StatsCollector::new(StatsConfig::continuous());
+    collector.process(&reply(0, 10, 9));
+    collector.process(&reply(1, 20, 19));
+
+    let snapshot = collector.snapshot();
+    assert_eq!(snapshot.rtt.primary.median_ns, None);
+    assert_eq!(snapshot.rtt.raw.median_ns, None);
+    assert_eq!(snapshot.rtt.adjusted.median_ns, None);
+    assert_eq!(snapshot.ipdv.round_trip.median_ns, None);
+    assert_eq!(snapshot.one_way_delay.send_delay.median_ns, None);
+    assert_eq!(snapshot.one_way_delay.receive_delay.median_ns, None);
+}
+
+#[test]
+fn continuous_mode_tracks_running_samples_without_exact_medians() {
+    let mut collector = StatsCollector::new(StatsConfig::continuous());
+    for seq in 0..4104 {
+        collector.process(&reply(seq, 10, 10));
+    }
+
+    let snapshot = collector.snapshot();
+    assert_eq!(snapshot.rtt.primary.count, 4104);
+    assert_eq!(snapshot.ipdv.round_trip.count, 4103);
+    assert_eq!(snapshot.rtt.primary.median_ns, None);
+    assert_eq!(snapshot.ipdv.round_trip.median_ns, None);
+}
+
+#[test]
+fn cumulative_rtt_uses_signed_effective_and_tracks_raw() {
+    let mut collector = StatsCollector::new(StatsConfig::finite());
+    collector.process(&reply(0, 1, -2));
+    collector.process(&reply(1, 10, 8));
+
+    let snapshot = collector.snapshot();
+    assert_eq!(snapshot.rtt.primary.count, 2);
+    assert_eq!(snapshot.rtt.primary.min_ns, Some(-2_000_000));
+    assert_eq!(snapshot.rtt.primary.median_ns, Some(3_000_000.0));
+    assert_eq!(snapshot.rtt.raw.total_ns, 11_000_000);
+}
+
+#[test]
+fn late_unique_counts_and_duplicates_do_not_update_measurements() {
+    let mut collector = StatsCollector::new(StatsConfig::finite());
+    collector.process(&sent(0, ts(0)));
+    collector.process(&sent(1, ts(10)));
+    collector.process(&reply(1, 10, 9));
+    collector.process(&late_reply(0, 20, 19));
+    collector.process(&ClientEvent::DuplicateReply {
+        seq: 0,
+        remote: "127.0.0.1:2112".parse().unwrap(),
+        received_at: ts(50),
+        bytes: 64,
+    });
+
+    let snapshot = collector.snapshot();
+    assert_eq!(snapshot.packets.packets_sent, 2);
+    assert_eq!(snapshot.packets.packets_received, 3);
+    assert_eq!(snapshot.packets.unique_replies, 2);
+    assert_eq!(snapshot.packets.duplicates, 1);
+    assert_eq!(snapshot.packets.late_packets, 1);
+    assert_eq!(snapshot.rtt.primary.count, 2);
+    assert_eq!(snapshot.loss.lost_packets, 0);
+    assert_eq!(snapshot.loss.duplicate_percent, 100.0 / 3.0);
+}
+
+#[test]
+fn final_loss_uses_sent_minus_unique_replies_not_echo_loss_events() {
+    let mut collector = StatsCollector::new(StatsConfig::finite());
+    collector.process(&sent(0, ts(0)));
+    collector.process(&sent(1, ts(10)));
+    collector.process(&ClientEvent::EchoLoss {
+        seq: 0,
+        sent_at: ts(0),
+        timeout_at: Instant::now(),
+    });
+    collector.process(&late_reply(0, 10, 9));
+
+    let snapshot = collector.snapshot();
+    assert_eq!(snapshot.events.loss_events, 1);
+    assert_eq!(snapshot.packets.unique_replies, 1);
+    assert_eq!(snapshot.loss.lost_packets, 1);
+    assert_eq!(snapshot.loss.packet_loss_percent, 50.0);
+}
+
+fn assert_no_ipdv_pairs(update: &EventStatsUpdate) {
+    assert!(update.ipdv_pairs.is_empty(), "{update:?}");
+}
+
+fn assert_one_ipdv_pair(
+    update: &EventStatsUpdate,
+    previous_seq: u32,
+    current_seq: u32,
+    rtt_ipdv: Duration,
+) -> &IpdvPairUpdate {
+    assert_eq!(update.ipdv_pairs.len(), 1, "{update:?}");
+    let pair = &update.ipdv_pairs[0];
+    assert_eq!(pair.previous_seq, previous_seq);
+    assert_eq!(pair.current_seq, current_seq);
+    assert_eq!(pair.rtt_ipdv, rtt_ipdv);
+    pair
+}
+
+#[test]
+fn ipdv_is_sequence_adjacent_and_gap_preserving() {
+    let mut collector = StatsCollector::new(StatsConfig::finite());
+    let first = collector.process(&reply(0, 10, 10));
+    let gap = collector.process(&reply(2, 15, 15));
+    let adjacent = collector.process(&reply(3, 12, 12));
+
+    let snapshot = collector.snapshot();
+    assert!(first.contributed_sample);
+    assert_no_ipdv_pairs(&first);
+
+    assert!(gap.contributed_sample);
+    assert_no_ipdv_pairs(&gap);
+
+    assert!(adjacent.contributed_sample);
+    assert_one_ipdv_pair(&adjacent, 2, 3, Duration::from_millis(3));
+    assert_eq!(snapshot.ipdv.round_trip.count, 1);
+    assert_eq!(snapshot.ipdv.round_trip.total_ns, 3_000_000);
+}
+
+#[test]
+fn late_reply_can_complete_ipdv_pair() {
+    let mut collector = StatsCollector::new(StatsConfig::finite());
+    collector.process(&reply(1, 20, 20));
+    let update = collector.process(&late_reply(0, 10, 10));
+
+    let snapshot = collector.snapshot();
+
+    assert!(update.contributed_sample);
+    assert_one_ipdv_pair(&update, 0, 1, Duration::from_millis(10));
+
+    assert_eq!(snapshot.ipdv.round_trip.count, 1);
+    assert_eq!(snapshot.ipdv.round_trip.total_ns, 10_000_000);
+}
+
+#[test]
+fn update_exposes_directional_ipdv_when_available() {
+    let mut collector = StatsCollector::new(StatsConfig::finite());
+    collector.process(&reply(0, 10, 10));
+    let update = collector.process(&reply(1, 13, 13));
+
+    assert!(update.contributed_sample);
+
+    let pair = assert_one_ipdv_pair(&update, 0, 1, Duration::from_millis(3));
+    assert!(pair.send_ipdv.is_some());
+    assert!(pair.receive_ipdv.is_some());
+}
+
+#[test]
+fn gap_fill_update_exposes_both_completed_ipdv_pairs() {
+    let mut collector = StatsCollector::new(StatsConfig::finite());
+
+    let first = collector.process(&reply(0, 10, 10));
+    let gap = collector.process(&reply(2, 20, 20));
+    let fill = collector.process(&reply(1, 13, 13));
+
+    assert!(first.contributed_sample);
+    assert!(first.ipdv_pairs.is_empty());
+
+    assert!(gap.contributed_sample);
+    assert!(gap.ipdv_pairs.is_empty());
+
+    assert!(fill.contributed_sample);
+    assert_eq!(fill.ipdv_pairs.len(), 2);
+
+    assert_eq!(fill.ipdv_pairs[0].previous_seq, 0);
+    assert_eq!(fill.ipdv_pairs[0].current_seq, 1);
+    assert_eq!(fill.ipdv_pairs[0].rtt_ipdv, Duration::from_millis(3));
+
+    assert_eq!(fill.ipdv_pairs[1].previous_seq, 1);
+    assert_eq!(fill.ipdv_pairs[1].current_seq, 2);
+    assert_eq!(fill.ipdv_pairs[1].rtt_ipdv, Duration::from_millis(7));
+
+    let snapshot = collector.snapshot();
+    assert_eq!(snapshot.ipdv.round_trip.count, 2);
+    assert_eq!(snapshot.ipdv.round_trip.total_ns, 10_000_000);
+}
+
+#[test]
+fn server_processing_and_one_way_require_available_samples() {
+    let mut collector = StatsCollector::new(StatsConfig::finite());
+    collector.process(&reply(0, 10, 9));
+    collector.process(&ClientEvent::LateReply {
+        seq: 9,
+        highest_seen: 10,
+        remote: "127.0.0.1:2112".parse().unwrap(),
+        sent_at: None,
+        received_at: ts(100),
+        rtt: None,
+        server_timing: None,
+        one_way: None,
+        received_stats: None,
+        bytes: 64,
+        packet_meta: PacketMeta::default(),
+    });
+
+    let snapshot = collector.snapshot();
+    assert_eq!(snapshot.server_processing.processing.count, 1);
+    assert_eq!(snapshot.one_way_delay.send_delay.count, 1);
+    assert_eq!(snapshot.events.untracked_late_replies, 1);
+}
+
+#[test]
+fn rolling_count_eviction_recomputes_from_bounded_events() {
+    let mut collector = StatsCollector::new(StatsConfig {
+        samples: SampleMode::RunningOnly,
+        rolling_count: Some(2),
+        rolling_time: None,
+    });
+    collector.process(&sent(0, ts(0)));
+    collector.process(&reply(0, 10, 10));
+    collector.process(&reply(1, 20, 20));
+
+    let rolling = collector.rolling_count().unwrap();
+    assert_eq!(rolling.packets.packets_sent, 0);
+    assert_eq!(rolling.packets.unique_replies, 2);
+    assert_eq!(rolling.rtt.primary.count, 2);
+}
+
+#[test]
+fn rolling_time_eviction_uses_event_timestamps() {
+    let mut collector = StatsCollector::new(StatsConfig {
+        samples: SampleMode::RunningOnly,
+        rolling_count: None,
+        rolling_time: Some(Duration::from_millis(15)),
+    });
+    collector.process(&sent(0, ts(0)));
+    collector.process(&sent(1, ts(10)));
+    collector.process(&sent(2, ts(30)));
+
+    let rolling = collector.rolling_time().unwrap();
+    assert_eq!(rolling.packets.packets_sent, 1);
+}
+
+#[test]
+fn empty_and_all_lost_edges_are_defined() {
+    let empty = StatsCollector::new(StatsConfig::finite()).snapshot();
+    assert_eq!(empty.loss.packet_loss_percent, 0.0);
+
+    let mut collector = StatsCollector::new(StatsConfig::finite());
+    collector.process(&sent(0, ts(0)));
+    let all_lost = collector.snapshot();
+    assert_eq!(all_lost.loss.lost_packets, 1);
+    assert_eq!(all_lost.loss.packet_loss_percent, 100.0);
+}
+
+#[test]
+fn directional_loss_uses_server_received_count_when_available() {
+    let mut collector = StatsCollector::new(StatsConfig::finite());
+    collector.process(&sent(0, ts(0)));
+    collector.process(&sent(1, ts(10)));
+    collector.process(&reply(0, 10, 10));
+
+    let loss = collector.snapshot().loss;
+    assert_eq!(loss.lost_packets, 1);
+    assert_eq!(loss.upstream_loss_packets, Some(1));
+    assert_eq!(loss.downstream_loss_packets, Some(0));
+    assert_eq!(loss.packet_loss_percent, 50.0);
+    assert_eq!(loss.upstream_loss_percent, 50.0);
+}