Merge pull request #98 from Lochnair/refactor/stats-exact-medians

refactor(stats): retain exact finite timing samples

crates/irtt-cli/src/cmd/client/summary.rs

@@ -218,6 +218,97 @@ mod tests {
         assert!(!output.contains("medians unavailable"));
     }
 
+    /// Returns the Median cell of the row with this label.
+    fn median_cell(output: &str, label: &str) -> String {
+        let row = output
+            .lines()
+            .find(|line| line.trim_start().starts_with(label))
+            .unwrap_or_else(|| panic!("summary should contain a {label:?} row:\n{output}"));
+        // Metric, Count, Min, Mean, Median, ...
+        // Counting from the right keeps multi-word labels out of the way.
+        row.split_whitespace()
+            .rev()
+            .nth(2)
+            .unwrap_or_else(|| panic!("row {row:?} should have a Median cell"))
+            .to_owned()
+    }
+
+    /// Collects probes whose send-call, timer-error, and server-processing
+    /// values each have a median distinct from their min, mean, and max, so
+    /// the rendered cell can only come from a real median calculation.
+    fn spread_finite_summary(config: StatsConfig) -> Snapshot {
+        let mut collector = StatsCollector::new(config);
+        for (seq, (send_call_us, timer_error_us, processing_us)) in
+            [(10_u64, 2_u64, 100_u64), (40, 12, 700), (20, 4, 300)]
+                .into_iter()
+                .enumerate()
+        {
+            let seq = seq as u32;
+            let offset = Duration::from_secs(1) + Duration::from_millis(u64::from(seq));
+            let sent_at = test_timestamp(offset);
+            let received_at = test_timestamp(offset + Duration::from_micros(1500));
+
+            collector.process(&ClientEvent::EchoSent {
+                seq,
+                remote: test_remote(),
+                scheduled_at: sent_at.mono,
+                sent_at,
+                bytes: 64,
+                send_call: Duration::from_micros(send_call_us),
+                timer_error: Duration::from_micros(timer_error_us),
+            });
+            collector.process(&ClientEvent::EchoReply {
+                seq,
+                remote: test_remote(),
+                sent_at,
+                received_at,
+                rtt: RttSample {
+                    raw: Duration::from_micros(1500),
+                    adjusted: None,
+                    effective: SignedDuration::from_nanos(1_500_000),
+                },
+                server_timing: Some(ServerTiming {
+                    receive_wall_ns: None,
+                    receive_mono_ns: None,
+                    send_wall_ns: None,
+                    send_mono_ns: None,
+                    midpoint_wall_ns: None,
+                    midpoint_mono_ns: None,
+                    processing: Some(Duration::from_micros(processing_us)),
+                }),
+                one_way: None,
+                received_stats: None,
+                bytes: 64,
+                packet_meta: PacketMeta::default(),
+            });
+        }
+        collector.snapshot()
+    }
+
+    #[test]
+    fn finite_summary_shows_medians_for_send_call_timer_error_and_processing() {
+        let output = format_summary(&spread_finite_summary(StatsConfig::finite()));
+
+        // Medians of 10/20/40, 2/4/12, and 100/300/700 µs. None of these
+        // equals the row's min, mean, or max.
+        assert_eq!(median_cell(&output, "send call"), "20.0µs");
+        assert_eq!(median_cell(&output, "timer error"), "4.0µs");
+        assert_eq!(median_cell(&output, "server processing"), "300.0µs");
+    }
+
+    #[test]
+    fn continuous_summary_still_shows_no_medians_for_those_rows() {
+        let output = format_summary(&spread_finite_summary(StatsConfig::continuous()));
+
+        for label in ["send call", "timer error", "server processing"] {
+            assert_eq!(
+                median_cell(&output, label),
+                "-",
+                "continuous mode retains no exact samples for {label}"
+            );
+        }
+    }
+
     #[test]
     fn verbose_summary_includes_raw_and_adjusted_rtt_rows() {
         let mut collector = StatsCollector::new(StatsConfig::finite());

crates/irtt-stats/src/core.rs

@@ -32,7 +32,11 @@ pub(crate) struct CoreStats {
 
 impl CoreStats {
     pub(crate) fn new(sample_mode: SampleMode, late_replies: LateReplyMode) -> Self {
-        let sequence_limit = if sample_mode == SampleMode::Exact {
+        // One decision for every public timing metric: in exact mode they all
+        // retain their samples so they can all report an exact median. The
+        // retention estimate in `retention.rs` counts the same set.
+        let retain_exact = sample_mode == SampleMode::Exact;
+        let sequence_limit = if retain_exact {
             None
         } else {
             Some(CONTINUOUS_SEQUENCE_LIMIT)
@@ -41,17 +45,17 @@ impl CoreStats {
         Self {
             events: EventCounts::default(),
             packets: PacketCounts::default(),
-            send_call: TimeMetric::new(false),
+            send_call: TimeMetric::new(retain_exact),
-            timer_error: TimeMetric::new(false),
+            timer_error: TimeMetric::new(retain_exact),
-            rtt_primary: TimeMetric::new(sample_mode == SampleMode::Exact),
+            rtt_primary: TimeMetric::new(retain_exact),
-            rtt_raw: TimeMetric::new(sample_mode == SampleMode::Exact),
+            rtt_raw: TimeMetric::new(retain_exact),
-            rtt_adjusted: TimeMetric::new(sample_mode == SampleMode::Exact),
+            rtt_adjusted: TimeMetric::new(retain_exact),
-            ipdv_round_trip: TimeMetric::new(sample_mode == SampleMode::Exact),
+            ipdv_round_trip: TimeMetric::new(retain_exact),
-            ipdv_send: TimeMetric::new(sample_mode == SampleMode::Exact),
+            ipdv_send: TimeMetric::new(retain_exact),
-            ipdv_receive: TimeMetric::new(sample_mode == SampleMode::Exact),
+            ipdv_receive: TimeMetric::new(retain_exact),
-            send_delay: TimeMetric::new(sample_mode == SampleMode::Exact),
+            send_delay: TimeMetric::new(retain_exact),
-            receive_delay: TimeMetric::new(sample_mode == SampleMode::Exact),
+            receive_delay: TimeMetric::new(retain_exact),
-            server_processing: TimeMetric::new(false),
+            server_processing: TimeMetric::new(retain_exact),
             ipdv_tracker: IpdvTracker::new(sequence_limit),
             late_replies,
         }

crates/irtt-stats/src/lib.rs

@@ -28,7 +28,8 @@ pub use time_stats::TimeStats;
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 /// Configuration for statistics collection.
 pub struct StatsConfig {
-    /// How timing samples are retained for median-capable metrics.
+    /// How timing samples are retained, and so whether the cumulative snapshot
+    /// reports medians.
     pub samples: SampleMode,
     /// Number of recent normalized events retained for count-based rolling snapshots.
     ///
@@ -44,9 +45,10 @@ pub struct StatsConfig {
 impl StatsConfig {
     /// Returns the default configuration for finite runs.
     ///
-    /// Finite mode retains exact samples where needed for medians and keeps
+    /// Finite mode retains exact samples for every timing metric, so each of
-    /// unbounded adjacent-sequence IPDV tracking so late adjacent replies can
+    /// them reports a median in the cumulative snapshot once it has samples,
-    /// still complete IPDV pairs.
+    /// and keeps unbounded adjacent-sequence IPDV tracking so late adjacent
+    /// replies can still complete IPDV pairs.
     pub fn finite() -> Self {
         Self {
             samples: SampleMode::Exact,
@@ -58,9 +60,9 @@ impl StatsConfig {
 
     /// Returns a configuration for long-running use.
     ///
-    /// Continuous mode uses running statistics, does not retain exact samples
+    /// Continuous mode uses running statistics, retains no exact samples so no
-    /// for medians, and bounds adjacent-sequence IPDV tracking for long-running
+    /// timing metric reports a median, and bounds adjacent-sequence IPDV
-    /// sessions.
+    /// tracking for long-running sessions.
     pub fn continuous() -> Self {
         Self {
             samples: SampleMode::RunningOnly,
@@ -81,7 +83,7 @@ impl StatsConfig {
     ///
     /// # What is included
     ///
-    /// - Exact timing samples, for the metrics that retain them.
+    /// - Exact timing samples, for every metric that retains them.
     /// - Adjacent-sequence IPDV tracking state.
     /// - Count-based rolling storage, when [`StatsConfig::rolling_count`] is
     ///   set, assuming a probe contributes one send event and one unique
@@ -98,17 +100,18 @@ impl StatsConfig {
     ///
     /// # Sample mode
     ///
-    /// [`SampleMode::Exact`] retains one value per metric per measurable
+    /// [`SampleMode::Exact`] retains one value per timing metric per ordinary
-    /// reply and keeps unbounded IPDV state, so the estimate grows with
+    /// successful probe and keeps unbounded IPDV state, so the estimate grows
-    /// `probe_count`. [`SampleMode::RunningOnly`] retains no exact samples
+    /// with `probe_count`. [`SampleMode::RunningOnly`] retains no exact
-    /// and bounds its IPDV state, so the estimate reaches a ceiling and stops
+    /// samples and bounds its IPDV state, so the estimate reaches a ceiling
-    /// growing.
+    /// and stops growing.
     ///
     /// # Optional measurements
     ///
-    /// Adjusted RTT, one-way delay, and send/receive IPDV are only measured
+    /// Adjusted RTT, one-way delay, send/receive IPDV, and server processing
-    /// when the negotiated session supplies the necessary timestamps. The
+    /// are only measured when the negotiated session supplies the necessary
-    /// estimate assumes all of them are available, which is the upper bound.
+    /// timestamps. The estimate assumes all of them are available, which is
+    /// the upper bound.
     ///
     /// # Saturation
     ///
@@ -151,10 +154,19 @@ impl Default for StatsConfig {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 /// Controls whether exact timing samples are retained.
+///
+/// This governs the cumulative snapshot from [`StatsCollector::snapshot`],
+/// where the choice applies uniformly to every timing metric, so that snapshot
+/// never mixes median-capable and median-incapable timing metrics. Rolling
+/// snapshots are recomputed with running statistics whatever this is set to;
+/// see [`StatsCollector::rolling_count`].
 pub enum SampleMode {
-    /// Keep only running statistics; exact medians are not available.
+    /// Keep only running statistics; medians are not available for any timing
+    /// metric, and retention stays bounded for long-running sessions.
     RunningOnly,
-    /// Retain exact samples for metrics that report exact medians.
+    /// Retain exact samples for every timing metric, so each one reports a
+    /// median once it has samples in the cumulative snapshot. Retention grows
+    /// with the probe count.
     Exact,
 }
 
@@ -193,20 +205,28 @@ impl StatsCollector {
     }
 
     /// Returns a snapshot of all events processed by this collector.
+    ///
+    /// This is the snapshot [`StatsConfig::samples`] applies to, so under
+    /// [`SampleMode::Exact`] every timing metric here reports a median once it
+    /// has samples.
     pub fn snapshot(&self) -> Snapshot {
         self.cumulative.snapshot()
     }
 
     /// Returns the configured count-based rolling snapshot, if enabled.
     ///
-    /// The snapshot is recomputed from the retained rolling events.
+    /// The snapshot is recomputed from the retained rolling events using
+    /// running statistics only, whatever [`StatsConfig::samples`] is set to,
+    /// so its timing metrics report no median. A rolling window is a bounded
+    /// recent view rather than the run's retained history.
     pub fn rolling_count(&self) -> Option<Snapshot> {
         self.rolling.count_snapshot()
     }
 
     /// Returns the configured time-based rolling snapshot, if enabled.
     ///
-    /// The snapshot is recomputed from the retained rolling events.
+    /// Like [`StatsCollector::rolling_count`], this is recomputed with running
+    /// statistics only and reports no medians.
     pub fn rolling_time(&self) -> Option<Snapshot> {
         self.rolling.time_snapshot()
     }

crates/irtt-stats/src/retention.rs

@@ -23,17 +23,23 @@ const GROWTH_FACTOR: u64 = 2;
 /// Approximate per-entry control overhead for the hashed containers.
 const HASH_ENTRY_OVERHEAD: u64 = 1;
 
-/// Exact-sample vectors that retain one value per measurable unique reply.
+/// Exact timing samples an ordinary successful probe can retain.
 ///
-/// In [`SampleMode::Exact`] these are the primary, raw, and adjusted RTT
+/// In [`SampleMode::Exact`] every public timing metric retains its samples, so
-/// metrics, the three IPDV metrics, and the two one-way delay metrics.
+/// one probe that is sent and answered can add one `i128` to each of them:
-/// `send_call`, `timer_error`, and `server_processing` do not retain exact
+/// send call and timer error from the send, and primary, raw, and adjusted
-/// samples in any mode.
+/// RTT, the three IPDV metrics, the two one-way delays, and server processing
+/// from the reply. [`SampleMode::RunningOnly`] retains none of them.
 ///
-/// The adjusted-RTT, one-way, and send/receive IPDV metrics only receive a
+/// Adjusted RTT, one-way delay, send/receive IPDV, and server processing only
-/// sample when the negotiated session supplies the corresponding optional
+/// receive a sample when the negotiated session supplies the corresponding
-/// measurement, so counting all eight is the upper bound.
+/// optional measurement, so counting all eleven is the upper bound.
-const EXACT_SAMPLE_VECS_PER_REPLY: u64 = 8;
+///
+/// This count must track the metrics `CoreStats::new` builds with exact
+/// retention. `exact_mode_retains_one_sample_per_counted_metric` below counts
+/// the metrics a fully measured probe actually retains, so the two cannot
+/// drift apart unnoticed.
+pub(crate) const EXACT_SAMPLES_PER_PROBE: u64 = 11;
 
 /// Normalized events a probe usually contributes to a rolling window: one send
 /// event and one unique reply event.
@@ -51,13 +57,14 @@ pub(crate) fn estimated_retained_bytes(config: &StatsConfig, probe_count: u64) -
         .saturating_mul(GROWTH_FACTOR)
 }
 
-/// Exact mode retains one sample per metric and one IPDV tracker entry for
+/// Exact mode retains one sample per timing metric and one IPDV tracker entry
-/// every measurable reply, so its storage grows with the probe count.
+/// for every ordinary successful probe, so its storage grows with the probe
+/// count.
 fn exact_bytes(probe_count: u64) -> u64 {
-    let per_reply = EXACT_SAMPLE_VECS_PER_REPLY
+    let per_probe = EXACT_SAMPLES_PER_PROBE
         .saturating_mul(size_of::<i128>() as u64)
         .saturating_add(ipdv_tracker_bytes_per_sample());
-    probe_count.saturating_mul(per_reply)
+    probe_count.saturating_mul(per_probe)
 }
 
 /// Running-only mode retains no exact samples and bounds the IPDV tracker at a
@@ -100,7 +107,136 @@ fn rolling_bytes_per_event() -> u64 {
 
 #[cfg(test)]
 mod tests {
+    use std::time::{Duration, Instant, UNIX_EPOCH};
+
+    use irtt_client::{
+        ClientEvent, ClientTimestamp, OneWayDelaySample, PacketMeta, ReceivedStatsSample,
+        RttSample, ServerTiming, SignedDuration,
+    };
+
     use super::*;
+    use crate::{Snapshot, StatsCollector, TimeStats};
+
+    /// Every public timing metric a snapshot exposes.
+    fn timing_metrics(snapshot: &Snapshot) -> Vec<&TimeStats> {
+        vec![
+            &snapshot.send_call,
+            &snapshot.timer_error,
+            &snapshot.rtt.primary,
+            &snapshot.rtt.raw,
+            &snapshot.rtt.adjusted,
+            &snapshot.ipdv.round_trip,
+            &snapshot.ipdv.send,
+            &snapshot.ipdv.receive,
+            &snapshot.one_way_delay.send_delay,
+            &snapshot.one_way_delay.receive_delay,
+            &snapshot.server_processing.processing,
+        ]
+    }
+
+    fn timestamp(offset_ms: u64) -> ClientTimestamp {
+        ClientTimestamp {
+            mono: Instant::now() + Duration::from_millis(offset_ms),
+            wall: UNIX_EPOCH + Duration::from_millis(offset_ms),
+        }
+    }
+
+    fn sent(seq: u32) -> ClientEvent {
+        let sent_at = timestamp(u64::from(seq) * 10);
+        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 + u64::from(seq)),
+            timer_error: Duration::from_micros(2 + u64::from(seq)),
+        }
+    }
+
+    /// A reply carrying every optional measurement, so one probe exercises the
+    /// upper bound the estimate assumes.
+    fn fully_measured_reply(seq: u32) -> ClientEvent {
+        let sent_at = timestamp(u64::from(seq) * 10);
+        let raw = Duration::from_millis(10 + u64::from(seq));
+        let received_at = ClientTimestamp {
+            mono: sent_at.mono + raw,
+            wall: sent_at.wall + raw,
+        };
+        let base_ns = i64::from(seq) * 10_000_000;
+        ClientEvent::EchoReply {
+            seq,
+            remote: "127.0.0.1:2112".parse().unwrap(),
+            sent_at,
+            received_at,
+            rtt: RttSample {
+                raw,
+                adjusted: Some(SignedDuration::from_duration(raw)),
+                effective: SignedDuration::from_duration(raw),
+            },
+            server_timing: Some(ServerTiming {
+                receive_mono_ns: Some(base_ns + 1_000_000),
+                send_mono_ns: Some(base_ns + 2_000_000),
+                receive_wall_ns: Some(base_ns + 1_000_000),
+                send_wall_ns: Some(base_ns + 2_000_000),
+                midpoint_mono_ns: None,
+                midpoint_wall_ns: None,
+                processing: Some(Duration::from_millis(1 + u64::from(seq))),
+            }),
+            one_way: Some(OneWayDelaySample {
+                client_to_server: Some(SignedDuration::from_nanos(1_000_000 + i128::from(seq))),
+                server_to_client: Some(SignedDuration::from_nanos(2_000_000 + i128::from(seq))),
+            }),
+            received_stats: Some(ReceivedStatsSample {
+                count: Some(seq + 1),
+                window: Some(0xff),
+            }),
+            bytes: 64,
+            packet_meta: PacketMeta::default(),
+        }
+    }
+
+    /// Two probes fill every metric, including the IPDV pair metrics that need
+    /// adjacent replies.
+    fn fully_measured_snapshot(config: StatsConfig) -> Snapshot {
+        let mut collector = StatsCollector::new(config);
+        for seq in 0..2 {
+            collector.process(&sent(seq));
+            collector.process(&fully_measured_reply(seq));
+        }
+        collector.snapshot()
+    }
+
+    /// The estimate is only honest while it counts the metrics that actually
+    /// retain samples. Adding or removing an exact-retaining public timing
+    /// metric without revisiting [`EXACT_SAMPLES_PER_PROBE`] fails here.
+    #[test]
+    fn exact_mode_retains_one_sample_per_counted_metric() {
+        let snapshot = fully_measured_snapshot(StatsConfig::finite());
+        let with_medians = timing_metrics(&snapshot)
+            .into_iter()
+            .filter(|stats| stats.median_ns.is_some())
+            .count() as u64;
+
+        assert_eq!(
+            with_medians, EXACT_SAMPLES_PER_PROBE,
+            "the estimate counts {EXACT_SAMPLES_PER_PROBE} exact sample streams per probe, \
+             but {with_medians} public timing metrics retained samples"
+        );
+    }
+
+    #[test]
+    fn running_only_retains_no_exact_samples() {
+        let snapshot = fully_measured_snapshot(StatsConfig::continuous());
+
+        for stats in timing_metrics(&snapshot) {
+            assert!(stats.count > 0, "the fixture should populate every metric");
+            assert_eq!(
+                stats.median_ns, None,
+                "running-only mode retains no exact samples"
+            );
+        }
+    }
 
     #[test]
     fn per_sample_components_are_non_zero() {

crates/irtt-stats/src/time_stats.rs

@@ -16,6 +16,9 @@ pub struct TimeStats {
     /// Arithmetic mean, in nanoseconds.
     pub mean_ns: f64,
     /// Median, in nanoseconds, when exact samples were retained.
+    ///
+    /// This is `None` when the collector was not configured to retain exact
+    /// samples, and also when it was but this metric received none.
     pub median_ns: Option<f64>,
     /// Sample variance, in nanoseconds squared.
     pub variance_ns2: f64,

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

@@ -100,6 +100,50 @@ pub fn reply_with_received_stats(
     event
 }
 
+/// Builds a send event with the supplied send-call duration and timer error.
+pub fn sent_with_timings(
+    seq: u32,
+    sent_at: ClientTimestamp,
+    send_call_us: u64,
+    timer_error_us: u64,
+) -> ClientEvent {
+    let mut event = sent(seq, sent_at);
+    let ClientEvent::EchoSent {
+        send_call,
+        timer_error,
+        ..
+    } = &mut event
+    else {
+        unreachable!();
+    };
+    *send_call = Duration::from_micros(send_call_us);
+    *timer_error = Duration::from_micros(timer_error_us);
+    event
+}
+
+/// Builds a normal reply reporting exactly the supplied server processing time.
+pub fn reply_with_processing(seq: u32, raw_ms: u64, processing_ms: u64) -> ClientEvent {
+    let mut event = unadjusted_reply(seq, raw_ms);
+    let ClientEvent::EchoReply { server_timing, .. } = &mut event else {
+        unreachable!();
+    };
+    if let Some(timing) = server_timing.as_mut() {
+        timing.processing = Some(Duration::from_millis(processing_ms));
+    }
+    event
+}
+
+/// Builds a normal reply from a session that supplied no server timestamps, so
+/// no server processing measurement exists.
+pub fn reply_without_server_timing(seq: u32, raw_ms: u64) -> ClientEvent {
+    let mut event = unadjusted_reply(seq, raw_ms);
+    let ClientEvent::EchoReply { server_timing, .. } = &mut event else {
+        unreachable!();
+    };
+    *server_timing = None;
+    event
+}
+
 pub fn unadjusted_late_reply(seq: u32, raw_ms: u64) -> ClientEvent {
     late_reply_from(unadjusted_reply(seq, raw_ms))
 }

crates/irtt-stats/tests/retention.rs

@@ -35,6 +35,25 @@ fn exact_mode_scales_linearly_with_probe_count() {
     assert_eq!(config.estimated_retained_bytes(1_000), single * 1_000);
 }
 
+/// Exact mode retains eleven `i128` timing streams per probe alongside its
+/// IPDV tracker state, so the timing samples alone are at least that much.
+///
+/// This checks the estimate against the storage it claims to model. It is an
+/// implementation test, not a stable public constant: the figure is free to
+/// move whenever the retention model does.
+#[test]
+fn exact_estimate_covers_every_retained_timing_sample() {
+    const EXACT_TIMING_STREAMS: u64 = 11;
+    let per_probe = StatsConfig::finite().estimated_retained_bytes(1);
+    let timing_sample_bytes = EXACT_TIMING_STREAMS * std::mem::size_of::<i128>() as u64;
+
+    assert!(
+        per_probe > timing_sample_bytes,
+        "the estimate should cover {EXACT_TIMING_STREAMS} retained timing samples plus \
+         IPDV tracker state, but a probe estimated only {per_probe} bytes"
+    );
+}
+
 #[test]
 fn enormous_probe_counts_saturate_instead_of_wrapping() {
     for config in [StatsConfig::finite(), StatsConfig::continuous()] {

crates/irtt-stats/tests/stats.rs

@@ -3,7 +3,8 @@ mod common;
 use std::time::{Duration, Instant, UNIX_EPOCH};
 
 use common::{
-    adjusted_reply, reply_with_received_stats, sent, ts, unadjusted_late_reply, unadjusted_reply,
+    adjusted_reply, reply_with_processing, reply_with_received_stats, reply_without_server_timing,
+    sent, sent_with_timings, ts, unadjusted_late_reply, unadjusted_reply,
 };
 use irtt_client::{
     ClientEvent, ClientTimestamp, OneWayDelaySample, PacketMeta, RttSample, SignedDuration,
@@ -12,6 +13,144 @@ use irtt_stats::{
     EventStatsUpdate, IpdvPairUpdate, LateReplyMode, SampleMode, StatsCollector, StatsConfig,
 };
 
+#[test]
+fn exact_mode_reports_a_send_call_median() {
+    let mut collector = StatsCollector::new(StatsConfig::finite());
+    for (seq, send_call_us) in [10_u64, 30, 20].into_iter().enumerate() {
+        collector.process(&sent_with_timings(
+            seq as u32,
+            ts(seq as u64 * 10),
+            send_call_us,
+            2,
+        ));
+    }
+
+    let snapshot = collector.snapshot();
+    assert_eq!(snapshot.send_call.count, 3);
+    assert_eq!(snapshot.send_call.min_ns, Some(10_000));
+    assert_eq!(snapshot.send_call.max_ns, Some(30_000));
+    // Sorted 10, 20, 30 us: the median is the middle sample, not the mean.
+    assert_eq!(snapshot.send_call.median_ns, Some(20_000.0));
+}
+
+#[test]
+fn exact_mode_reports_a_timer_error_median() {
+    let mut collector = StatsCollector::new(StatsConfig::finite());
+    for (seq, timer_error_us) in [2_u64, 8, 4, 6].into_iter().enumerate() {
+        collector.process(&sent_with_timings(
+            seq as u32,
+            ts(seq as u64 * 10),
+            10,
+            timer_error_us,
+        ));
+    }
+
+    let snapshot = collector.snapshot();
+    assert_eq!(snapshot.timer_error.count, 4);
+    // Sorted 2, 4, 6, 8 us: an even count averages the two middle samples.
+    assert_eq!(snapshot.timer_error.median_ns, Some(5_000.0));
+}
+
+#[test]
+fn exact_mode_reports_a_server_processing_median() {
+    let mut collector = StatsCollector::new(StatsConfig::finite());
+    for (seq, processing_ms) in [1_u64, 5, 3].into_iter().enumerate() {
+        collector.process(&reply_with_processing(seq as u32, 10, processing_ms));
+    }
+
+    let snapshot = collector.snapshot();
+    assert_eq!(snapshot.server_processing.processing.count, 3);
+    assert_eq!(
+        snapshot.server_processing.processing.median_ns,
+        Some(3_000_000.0)
+    );
+}
+
+#[test]
+fn absent_server_processing_measurements_retain_no_samples() {
+    let mut collector = StatsCollector::new(StatsConfig::finite());
+    for seq in 0..3 {
+        collector.process(&reply_without_server_timing(seq, 10));
+    }
+
+    let snapshot = collector.snapshot();
+    // The replies measured RTT, so the run is not simply empty.
+    assert_eq!(snapshot.rtt.primary.count, 3);
+    assert!(snapshot.rtt.primary.median_ns.is_some());
+    // ...but a metric the server never supplied stays empty rather than
+    // gaining zero-valued samples.
+    assert_eq!(snapshot.server_processing.processing.count, 0);
+    assert_eq!(snapshot.server_processing.processing.median_ns, None);
+    assert_eq!(snapshot.server_processing.processing.total_ns, 0);
+}
+
+#[test]
+fn exact_mode_reports_medians_for_every_timing_metric_with_samples() {
+    let mut collector = StatsCollector::new(StatsConfig::finite());
+    for seq in 0..3 {
+        collector.process(&sent(seq, ts(u64::from(seq) * 10)));
+        collector.process(&adjusted_reply(
+            seq,
+            10 + u64::from(seq),
+            9 + i128::from(seq),
+        ));
+    }
+
+    let snapshot = collector.snapshot();
+    for (label, stats) in [
+        ("send call", &snapshot.send_call),
+        ("timer error", &snapshot.timer_error),
+        ("primary RTT", &snapshot.rtt.primary),
+        ("raw RTT", &snapshot.rtt.raw),
+        ("adjusted RTT", &snapshot.rtt.adjusted),
+        ("round-trip IPDV", &snapshot.ipdv.round_trip),
+        ("send IPDV", &snapshot.ipdv.send),
+        ("receive IPDV", &snapshot.ipdv.receive),
+        ("send delay", &snapshot.one_way_delay.send_delay),
+        ("receive delay", &snapshot.one_way_delay.receive_delay),
+        ("server processing", &snapshot.server_processing.processing),
+    ] {
+        assert!(stats.count > 0, "{label} should have samples");
+        assert!(
+            stats.median_ns.is_some(),
+            "{label} has samples, so exact mode should report its median"
+        );
+    }
+}
+
+#[test]
+fn running_only_tracks_send_side_and_processing_without_medians() {
+    let mut collector = StatsCollector::new(StatsConfig::continuous());
+    for (seq, value) in [10_u64, 30, 20].into_iter().enumerate() {
+        let seq = seq as u32;
+        collector.process(&sent_with_timings(
+            seq,
+            ts(u64::from(seq) * 10),
+            value,
+            value,
+        ));
+        collector.process(&reply_with_processing(seq, 10, 1 + u64::from(seq)));
+    }
+
+    let snapshot = collector.snapshot();
+    for (label, stats) in [
+        ("send call", &snapshot.send_call),
+        ("timer error", &snapshot.timer_error),
+        ("server processing", &snapshot.server_processing.processing),
+    ] {
+        assert_eq!(stats.count, 3, "{label} should still count its samples");
+        assert!(
+            stats.min_ns.is_some(),
+            "{label} should still track a minimum"
+        );
+        assert!(stats.mean_ns > 0.0, "{label} should still track a mean");
+        assert_eq!(
+            stats.median_ns, None,
+            "{label} retains no exact samples in running-only mode"
+        );
+    }
+}
+
 #[test]
 fn running_only_samples_avoid_finite_retention() {
     let mut collector = StatsCollector::new(StatsConfig::continuous());
@@ -492,6 +631,9 @@ fn matched_late_reply_counts_without_measuring_under_count_only() {
     assert_eq!(snapshot.one_way_delay.send_delay.count, 0);
     assert_eq!(snapshot.one_way_delay.receive_delay.count, 0);
     assert_eq!(snapshot.ipdv.round_trip.count, 0);
+    // The reply carried a processing time, but the policy withheld it, so
+    // exact retention has nothing to report a median from either.
+    assert_eq!(snapshot.server_processing.processing.median_ns, None);
 }
 
 #[test]
@@ -522,6 +664,45 @@ fn untracked_late_replies_stay_untracked_in_both_policies() {
     }
 }
 
+#[test]
+fn rolling_snapshots_report_no_medians_even_in_exact_mode() {
+    let mut collector = StatsCollector::new(StatsConfig {
+        rolling_count: Some(8),
+        rolling_time: Some(Duration::from_secs(60)),
+        ..StatsConfig::finite()
+    });
+    for seq in 0..3 {
+        collector.process(&sent(seq, ts(u64::from(seq) * 10)));
+        collector.process(&adjusted_reply(
+            seq,
+            10 + u64::from(seq),
+            9 + i128::from(seq),
+        ));
+    }
+
+    // The cumulative snapshot keeps the exact-mode contract...
+    let cumulative = collector.snapshot();
+    assert!(cumulative.rtt.primary.median_ns.is_some());
+    assert!(cumulative.send_call.median_ns.is_some());
+    assert!(cumulative.server_processing.processing.median_ns.is_some());
+
+    // ...while a rolling window is a bounded recent view recomputed from
+    // running statistics, so it reports no median whatever the sample mode.
+    for rolling in [
+        collector.rolling_count().unwrap(),
+        collector.rolling_time().unwrap(),
+    ] {
+        assert!(
+            rolling.rtt.primary.count > 0,
+            "the window should have samples"
+        );
+        assert_eq!(rolling.rtt.primary.median_ns, None);
+        assert_eq!(rolling.send_call.median_ns, None);
+        assert_eq!(rolling.timer_error.median_ns, None);
+        assert_eq!(rolling.server_processing.processing.median_ns, None);
+    }
+}
+
 #[test]
 fn rolling_snapshots_use_the_same_late_reply_policy() {
     let mut collector = StatsCollector::new(StatsConfig {