diff --git a/CHANGELOG.md b/CHANGELOG.md index 45ac845..f88cc88 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 - Typed `RejectReason` classification for receive-path silent-rejection 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); diff --git a/testing/lib/wait-converge-test.sh b/testing/lib/wait-converge-test.sh new file mode 100755 index 0000000..3abd2de --- /dev/null +++ b/testing/lib/wait-converge-test.sh @@ -0,0 +1,191 @@ +#!/bin/bash +# Unit tests for wait_until_connected() in wait-converge.sh. +# +# These tests drive the convergence gate with synthetic connectivity +# checks (ping_fn stand-ins) that report scripted PASSED/FAILED counts +# keyed off the same SECONDS clock the gate uses. No containers or +# network are involved, so the whole suite runs in a few seconds and is +# safe to run in CI. +# +# Run: +# ./wait-converge-test.sh +# Exits 0 only if every case passes; non-zero if any case fails. + +set -u + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +source "$SCRIPT_DIR/wait-converge.sh" + +RESULTS=() +FAILURES=0 + +# Record a single assertion result. +check() { + local name="$1" + local ok="$2" # 0 = pass, anything else = fail + local detail="${3:-}" + if [ "$ok" -eq 0 ]; then + RESULTS+=("PASS $name") + echo "PASS $name${detail:+ ($detail)}" + else + RESULTS+=("FAIL $name") + echo "FAIL $name${detail:+ ($detail)}" + FAILURES=$((FAILURES + 1)) + fi +} + +# Each ping_fn records its own start on first call so its schedule is +# measured from when the gate began polling it, regardless of the wall +# clock at suite start. PT (ping-elapsed time) is computed inline in each +# ping_fn — NOT via a subshell — so the PING_START assignment persists. +PING_START=-1 +reset_ping() { + PING_START=-1 +} + +# --- Synthetic connectivity checks ------------------------------------ + +# Sets global PT to ping-elapsed seconds. Must be called (not subshelled) +# at the top of each ping_fn so the first-call timestamp persists. +PT=0 +set_pt() { + if (( PING_START < 0 )); then + PING_START=$SECONDS + fi + PT=$(( SECONDS - PING_START )) +} + +# Case 1 trace: improves (16 reachable) early, climbs to 18, then holds +# at FAILED=1 (within slack) until late, then fully converges. Mirrors a +# deep node whose last pair clears only after stacked discovery backoff. +ping_near_converged_hold() { + set_pt; local t=$PT + if (( t < 3 )); then + PASSED=16; FAILED=4 + elif (( t < 6 )); then + PASSED=18; FAILED=2 + elif (( t < 12 )); then + # Stuck within slack: one straggling pair pending. + PASSED=19; FAILED=1 + else + PASSED=20; FAILED=0 + fi +} + +# Case 2 trace: climbs a little, then wedges far from convergence with +# FAILED well above the slack. Should fast-bail on stall. +ping_far_stall() { + set_pt; local t=$PT + if (( t < 3 )); then + PASSED=8; FAILED=12 + else + # Stuck for good, many pairs still pending (> slack). + PASSED=10; FAILED=10 + fi +} + +# Case 3 trace: never converges; always one pair pending and never makes +# further progress after the first reading. Should hit the hard cap. +# FAILED stays at exactly 1 here so it is within the default slack=2 and +# the near-converged hold keeps polling all the way to max_secs. +ping_never_converges() { + PASSED=19; FAILED=1 +} + +# Case 4 trace: backward-compat. Same shape as case 1 (near-converged +# straggler) but driven with only 4 args so the default slack applies. +ping_backcompat_hold() { + set_pt; local t=$PT + if (( t < 3 )); then + PASSED=16; FAILED=4 + elif (( t < 6 )); then + PASSED=18; FAILED=2 + elif (( t < 12 )); then + PASSED=19; FAILED=1 + else + PASSED=20; FAILED=0 + fi +} + +HOLD_MSG="holding for full budget" +STUCK_MSG="STUCK" + +# --- Case 1: near-converged hold -------------------------------------- +echo +echo "== Case 1: near-converged hold (slack saves it) ==" +reset_ping +out=$(wait_until_connected ping_near_converged_hold 20 4 1 2); rc=$? +echo "$out" +c1_rc_ok=1; [ "$rc" -eq 0 ] && c1_rc_ok=0 +check "case1: returns 0 (eventually converges)" "$c1_rc_ok" "rc=$rc" +c1_hold_ok=1; echo "$out" | grep -q "$HOLD_MSG" && c1_hold_ok=0 +check "case1: near-converged hold branch taken" "$c1_hold_ok" "expected '$HOLD_MSG' in output" + +# Same trace with slack=0 must fail (old behavior) — proves slack matters. +echo "-- Case 1b: same trace, slack=0 (old behavior) must bail --" +reset_ping +out0=$(wait_until_connected ping_near_converged_hold 20 4 1 0); rc0=$? +echo "$out0" +c1b_rc_ok=1; [ "$rc0" -ne 0 ] && c1b_rc_ok=0 +check "case1b: returns 1 with slack=0" "$c1b_rc_ok" "rc=$rc0" +c1b_stuck_ok=1; echo "$out0" | grep -q "$STUCK_MSG" && c1b_stuck_ok=0 +check "case1b: bailed via STUCK (not hold)" "$c1b_stuck_ok" "expected '$STUCK_MSG' in output" + +# --- Case 2: far-from-converged stall (fast-bail) --------------------- +echo +echo "== Case 2: far-from-converged stall (fast-bail) ==" +reset_ping +start=$SECONDS +out=$(wait_until_connected ping_far_stall 30 4 1 2); rc=$? +elapsed=$((SECONDS - start)) +echo "$out" +c2_rc_ok=1; [ "$rc" -ne 0 ] && c2_rc_ok=0 +check "case2: returns 1 (stall bail)" "$c2_rc_ok" "rc=$rc" +c2_stuck_ok=1; echo "$out" | grep -q "$STUCK_MSG" && c2_stuck_ok=0 +check "case2: bailed via STUCK message" "$c2_stuck_ok" +# Fast-bail: must finish well before the 30s hard cap. +c2_fast_ok=1; [ "$elapsed" -lt 20 ] && c2_fast_ok=0 +check "case2: fast-bail (before hard cap)" "$c2_fast_ok" "elapsed=${elapsed}s < 20s" +c2_nocap_ok=1; echo "$out" | grep -q "TIMEOUT" || c2_nocap_ok=0 +check "case2: did NOT hit hard-cap TIMEOUT" "$c2_nocap_ok" + +# --- Case 3: never converges (hard cap) ------------------------------- +echo +echo "== Case 3: never converges (hard cap) ==" +reset_ping +start=$SECONDS +out=$(wait_until_connected ping_never_converges 6 3 1 2); rc=$? +elapsed=$((SECONDS - start)) +echo "$out" +c3_rc_ok=1; [ "$rc" -ne 0 ] && c3_rc_ok=0 +check "case3: returns 1 (never converges)" "$c3_rc_ok" "rc=$rc" +c3_cap_ok=1; echo "$out" | grep -q "TIMEOUT" && c3_cap_ok=0 +check "case3: hit hard-cap TIMEOUT message" "$c3_cap_ok" +# Hard cap: must run roughly to max_secs (6s), not bail early. +c3_dur_ok=1; [ "$elapsed" -ge 6 ] && c3_dur_ok=0 +check "case3: ran to hard cap (~max_secs)" "$c3_dur_ok" "elapsed=${elapsed}s >= 6s" + +# --- Case 4: backward-compat (4 args, default slack=2) ---------------- +echo +echo "== Case 4: backward-compat, no slack arg (default=2) ==" +reset_ping +out=$(wait_until_connected ping_backcompat_hold 20 4 1); rc=$? +echo "$out" +c4_rc_ok=1; [ "$rc" -eq 0 ] && c4_rc_ok=0 +check "case4: returns 0 with 4 args" "$c4_rc_ok" "rc=$rc" +c4_hold_ok=1; echo "$out" | grep -q "$HOLD_MSG" && c4_hold_ok=0 +check "case4: default slack triggered near-converged hold" "$c4_hold_ok" + +# --- Summary ---------------------------------------------------------- +echo +echo "==============================================" +for r in "${RESULTS[@]}"; do + echo " $r" +done +echo "==============================================" +if [ "$FAILURES" -ne 0 ]; then + echo "RESULT: $FAILURES assertion(s) FAILED" + exit 1 +fi +echo "RESULT: all assertions passed" +exit 0 diff --git a/testing/lib/wait-converge.sh b/testing/lib/wait-converge.sh index c203eca..8832b67 100644 --- a/testing/lib/wait-converge.sh +++ b/testing/lib/wait-converge.sh @@ -8,7 +8,8 @@ # source "$(dirname "$0")/../../lib/wait-converge.sh" # wait_for_links [timeout_secs] # wait_for_peers [timeout_secs] -# wait_until_connected [poll_secs] +# wait_until_connected [poll_secs] \ +# [near_converged_slack] # Wait until a container has at least min_links active links. # Returns 0 on success, 1 on timeout. @@ -55,7 +56,8 @@ wait_for_peers() { # Wait until a connectivity check reports every pair reachable, using a # progress-aware deadline instead of a fixed one. # -# wait_until_connected [poll_secs] +# wait_until_connected [poll_secs] \ +# [near_converged_slack] # # is the name of a function that runs the suite's own # connectivity check and sets two globals each call: @@ -69,7 +71,16 @@ wait_for_peers() { # clock and keep waiting (slow-but-improving is not a failure, so it # does not false-time-out under CI load). # - stuck: PASSED has not improved for stall_secs -> return 1 (fail -# fast rather than burn the whole budget on a genuinely wedged pair). +# fast rather than burn the whole budget on a genuinely wedged pair), +# BUT only when FAILED > near_converged_slack (default 2). A mesh +# that is genuinely far from convergence still bails fast on stall. +# - near-converged hold: when FAILED <= near_converged_slack and the +# stall window has elapsed, do NOT bail. A handful of straggling +# pairs (e.g. a deep node whose last pair clears only after stacked +# discovery backoff + late bloom propagation) is a rare timing event, +# not a routing defect, so the gate keeps polling toward max_secs +# rather than emitting a false RED with budget still unspent. A +# genuinely never-converging single pair still hits the hard cap. # - hard cap: max_secs elapsed -> return 1 (never runs unbounded). # # Returns 0 once fully connected, 1 on stall or timeout. @@ -78,10 +89,12 @@ wait_until_connected() { local max_secs="$2" local stall_secs="$3" local poll_secs="${4:-1}" + local near_converged_slack="${5:-2}" local start_secs=$SECONDS local best=-1 local last_progress=$SECONDS + local held_for_budget=0 while (( SECONDS - start_secs < max_secs )); do "$ping_fn" @@ -94,8 +107,14 @@ wait_until_connected() { last_progress=$SECONDS echo " converge: $PASSED reachable, $FAILED pending (progressing) after $((SECONDS - start_secs))s" elif (( SECONDS - last_progress >= stall_secs )); then - echo " converge: STUCK at $PASSED reachable / $FAILED pending — no progress for ${stall_secs}s (after $((SECONDS - start_secs))s)" - return 1 + if (( FAILED > near_converged_slack )); then + echo " converge: STUCK at $PASSED reachable / $FAILED pending — no progress for ${stall_secs}s (after $((SECONDS - start_secs))s)" + return 1 + fi + if (( held_for_budget == 0 )); then + held_for_budget=1 + echo " converge: near-converged ($PASSED reachable / $FAILED pending <= slack=$near_converged_slack) — holding for full budget, not bailing (after $((SECONDS - start_secs))s)" + fi fi sleep "$poll_secs" done