diff --git a/CHANGELOG.md b/CHANGELOG.md index e861899..dd59372 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- MMP sender metrics now ignore duplicate or regressed receiver reports + before updating RTT, loss, goodput, or ETX. Receiver reports also + suppress timestamp echo when dwell time overflows, so stale reports + cannot inflate SRTT. + ### Added - `pool_inbound` and `pool_outbound` counters on the TCP and Tor diff --git a/docs/design/fips-mmp.md b/docs/design/fips-mmp.md index 562b0a0..006befa 100644 --- a/docs/design/fips-mmp.md +++ b/docs/design/fips-mmp.md @@ -100,6 +100,11 @@ inter-frame processing delays inflate spin bit RTT measurements unpredictably. Timestamp-echo from ReceiverReports (with dwell-time compensation) is the sole SRTT source. +Duplicate or regressed ReceiverReports are ignored before any RTT, loss, +goodput, or ETX update. If receiver-side dwell time exceeds the wire +field, the report keeps its counters but sends a zero timestamp echo so +the sender cannot form an invalid RTT sample. + The spin bit lives in the link-layer FMP inner header, so this mechanism applies to link-layer MMP only. Session-layer MMP carries its spin bit in the FSP encrypted inner header but uses it the same diff --git a/src/mmp/metrics.rs b/src/mmp/metrics.rs index c12fef7..a04d664 100644 --- a/src/mmp/metrics.rs +++ b/src/mmp/metrics.rs @@ -118,25 +118,64 @@ impl MmpMetrics { ) -> bool { let had_srtt = self.srtt.initialized(); + if self.has_prev_rr { + let counters_regressed = rr.highest_counter < self.prev_rr_highest_counter + || rr.cumulative_packets_recv < self.prev_rr_cum_packets + || rr.cumulative_bytes_recv < self.prev_rr_cum_bytes + || rr.ecn_ce_count < self.prev_rr_ecn_ce + || rr.cumulative_reorder_count < self.prev_rr_reorder; + let duplicate_counters = rr.highest_counter == self.prev_rr_highest_counter + && rr.cumulative_packets_recv == self.prev_rr_cum_packets + && rr.cumulative_bytes_recv == self.prev_rr_cum_bytes + && rr.ecn_ce_count == self.prev_rr_ecn_ce + && rr.cumulative_reorder_count == self.prev_rr_reorder; + // Safe to drop: reports are only built after interval data, so + // a fresh report always advances at least one cumulative counter. + if counters_regressed || duplicate_counters { + trace!( + highest_counter = rr.highest_counter, + prev_highest_counter = self.prev_rr_highest_counter, + cumulative_packets_recv = rr.cumulative_packets_recv, + prev_cumulative_packets_recv = self.prev_rr_cum_packets, + cumulative_bytes_recv = rr.cumulative_bytes_recv, + prev_cumulative_bytes_recv = self.prev_rr_cum_bytes, + "Ignoring stale MMP ReceiverReport" + ); + return false; + } + } + // --- RTT from timestamp echo --- // RTT = now - echoed_timestamp - dwell_time if rr.timestamp_echo > 0 { let echo_ms = rr.timestamp_echo; - let dwell_ms = rr.dwell_time as u32; - // Guard against timestamp wrap or bogus values - if our_timestamp_ms > echo_ms + dwell_ms { - let rtt_ms = our_timestamp_ms - echo_ms - dwell_ms; - let rtt_us = (rtt_ms as i64) * 1000; - trace!( - our_ts = our_timestamp_ms, - echo = echo_ms, - dwell = dwell_ms, - rtt_ms = rtt_ms, - srtt_ms = self.srtt.srtt_us() as f64 / 1000.0, - "RTT sample from timestamp echo" - ); - self.srtt.update(rtt_us); - self.rtt_trend.update(rtt_us as f64); + let dwell_ms = u32::from(rr.dwell_time); + let rtt_sample_ms = echo_ms + .checked_add(dwell_ms) + .and_then(|send_done_ms| our_timestamp_ms.checked_sub(send_done_ms)); + + match rtt_sample_ms { + Some(rtt_ms) if rtt_ms > 0 => { + let rtt_us = (rtt_ms as i64) * 1000; + trace!( + our_ts = our_timestamp_ms, + echo = echo_ms, + dwell = dwell_ms, + rtt_ms = rtt_ms, + srtt_ms = self.srtt.srtt_us() as f64 / 1000.0, + "RTT sample from timestamp echo" + ); + self.srtt.update(rtt_us); + self.rtt_trend.update(rtt_us as f64); + } + _ => { + trace!( + our_ts = our_timestamp_ms, + echo = echo_ms, + dwell = dwell_ms, + "Ignoring invalid MMP RTT sample" + ); + } } } @@ -318,6 +357,70 @@ mod tests { assert!((srtt_ms - 45.0).abs() < 1.0, "srtt={srtt_ms}, expected ~45"); } + #[test] + fn test_ignores_duplicate_receiver_report_after_valid_sample() { + let mut m = MmpMetrics::new(); + let t0 = Instant::now(); + + let rr1 = make_rr(10, 10, 5_000, 1_000, 5, 0); + m.process_receiver_report(&rr1, 1_050, t0); + + let rr2 = make_rr(20, 18, 14_000, 1_100, 5, 0); + m.process_receiver_report(&rr2, 1_150, t0 + Duration::from_secs(1)); + let baseline_srtt_ms = m.srtt_ms().unwrap(); + let baseline_loss = m.loss_rate(); + let baseline_goodput = m.goodput_bps(); + + assert!(baseline_loss > 0.0); + assert!(baseline_goodput > 0.0); + + // A duplicate of the same counters arriving later would be a 4.895s + // RTT sample if accepted. It is stale and must not move metrics. + m.process_receiver_report(&rr2, 6_000, t0 + Duration::from_secs(5)); + + assert_eq!(m.srtt_ms().unwrap(), baseline_srtt_ms); + assert_eq!(m.loss_rate(), baseline_loss); + assert_eq!(m.goodput_bps(), baseline_goodput); + } + + #[test] + fn test_ignores_out_of_order_receiver_report_after_valid_sample() { + let mut m = MmpMetrics::new(); + let now = Instant::now(); + + let valid_rr = make_rr(20, 20, 10000, 1000, 5, 0); + m.process_receiver_report(&valid_rr, 1050, now); + let baseline_srtt_ms = m.srtt_ms().unwrap(); + + let old_rr = make_rr(10, 10, 5000, 1000, 0, 0); + m.process_receiver_report(&old_rr, 6000, now + Duration::from_secs(5)); + + let srtt_ms = m.srtt_ms().unwrap(); + assert_eq!(srtt_ms, baseline_srtt_ms); + } + + #[test] + fn test_ignores_wrapped_rtt_sample() { + let mut m = MmpMetrics::new(); + let now = Instant::now(); + + let wrapped_rr = make_rr(10, 10, 5000, u32::MAX - 10, 20, 0); + m.process_receiver_report(&wrapped_rr, 15, now); + + assert!(m.srtt_ms().is_none()); + } + + #[test] + fn test_ignores_future_rtt_sample() { + let mut m = MmpMetrics::new(); + let now = Instant::now(); + + let future_rr = make_rr(10, 10, 5_000, 2_000, 5, 0); + m.process_receiver_report(&future_rr, 1_000, now); + + assert!(m.srtt_ms().is_none()); + } + #[test] fn test_loss_rate_computation() { let mut m = MmpMetrics::new(); diff --git a/src/mmp/receiver.rs b/src/mmp/receiver.rs index 0582431..3c3b5e0 100644 --- a/src/mmp/receiver.rs +++ b/src/mmp/receiver.rs @@ -323,11 +323,20 @@ impl ReceiverState { return None; } - // Dwell time: ms between last frame reception and report generation - let dwell_time = self + // Dwell time: ms between last frame reception and report generation. + // If it no longer fits on the wire, the timestamp echo cannot produce + // a valid RTT sample. Preserve the counters but suppress the echo. + let (timestamp_echo, dwell_time) = self .last_recv_time - .map(|t| now.duration_since(t).as_millis() as u16) - .unwrap_or(0); + .map(|t| { + let dwell_ms = now.duration_since(t).as_millis(); + if dwell_ms > u128::from(u16::MAX) { + (0, u16::MAX) + } else { + (self.last_sender_timestamp, dwell_ms as u16) + } + }) + .unwrap_or((0, 0)); let (burst_count, max_burst, mean_burst) = self.gap_tracker.take_interval_stats(); @@ -335,7 +344,7 @@ impl ReceiverState { highest_counter: self.highest_counter, cumulative_packets_recv: self.cumulative_packets_recv, cumulative_bytes_recv: self.cumulative_bytes_recv, - timestamp_echo: self.last_sender_timestamp, + timestamp_echo, dwell_time, max_burst_loss: max_burst, mean_burst_loss: mean_burst, @@ -505,6 +514,21 @@ mod tests { assert_eq!(report.interval_bytes_recv, 1100); } + #[test] + fn test_build_report_suppresses_rtt_echo_when_dwell_overflows() { + let mut r = ReceiverState::new(32); + let t0 = Instant::now(); + r.record_recv(1, 100, 500, false, t0); + + let report = r + .build_report(t0 + Duration::from_millis(u64::from(u16::MAX) + 1)) + .unwrap(); + + assert_eq!(report.timestamp_echo, 0); + assert_eq!(report.dwell_time, u16::MAX); + assert_eq!(report.cumulative_packets_recv, 1); + } + #[test] fn test_build_report_resets_interval() { let mut r = ReceiverState::new(32);