Merge maint into master (MMP stale-report rejection; convergence-gate near-budget hold)

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();

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);