From 5d72e7cee137e57004840215a04a3a1edd1d8b3d Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Mon, 23 Feb 2026 19:56:14 +0000 Subject: [PATCH] Add exponential backoff to session MMP probing on send failures Track consecutive send failures in SenderState. Apply 2^n backoff multiplier (capped at 32x) to the report interval. Suppress debug logs after 3 consecutive failures, emit recovery summary on success. --- src/mmp/sender.rs | 114 ++++++++++++++++++++++++++++++++++++++- src/node/handlers/mmp.rs | 69 +++++++++++++++++++++--- 2 files changed, 174 insertions(+), 9 deletions(-) diff --git a/src/mmp/sender.rs b/src/mmp/sender.rs index c3a4fa2..ba3b360 100644 --- a/src/mmp/sender.rs +++ b/src/mmp/sender.rs @@ -31,6 +31,10 @@ pub struct SenderState { // --- Report timing --- last_report_time: Option, report_interval: Duration, + + // --- Send failure backoff --- + /// Consecutive send failure count for backoff calculation. + consecutive_send_failures: u32, } impl SenderState { @@ -54,6 +58,7 @@ impl SenderState { interval_has_data: false, last_report_time: None, report_interval: Duration::from_millis(cold_start_ms), + consecutive_send_failures: 0, } } @@ -102,13 +107,44 @@ impl SenderState { } /// Check if it's time to send a report. + /// + /// When consecutive send failures have occurred, the effective interval + /// is multiplied by an exponential backoff factor (2^failures, capped at 32×). pub fn should_send_report(&self, now: Instant) -> bool { if !self.interval_has_data { return false; } match self.last_report_time { None => true, // Never sent a report — send immediately - Some(last) => now.duration_since(last) >= self.report_interval, + Some(last) => { + let effective = self.report_interval.mul_f64(self.send_failure_backoff_multiplier()); + now.duration_since(last) >= effective + } + } + } + + /// Record a send failure. Returns the new consecutive failure count. + pub fn record_send_failure(&mut self) -> u32 { + self.consecutive_send_failures += 1; + self.consecutive_send_failures + } + + /// Record a successful send. Returns the previous failure count (for summary logging). + pub fn record_send_success(&mut self) -> u32 { + let prev = self.consecutive_send_failures; + self.consecutive_send_failures = 0; + prev + } + + /// Get the backoff multiplier based on consecutive failures. + /// + /// Returns 1.0 for no failures, 2.0 for 1 failure, 4.0 for 2, ... + /// capped at 32.0 (5 failures). + pub fn send_failure_backoff_multiplier(&self) -> f64 { + if self.consecutive_send_failures == 0 { + 1.0 + } else { + 2.0_f64.powi(self.consecutive_send_failures.min(5) as i32) } } @@ -145,6 +181,10 @@ impl SenderState { pub fn report_interval(&self) -> Duration { self.report_interval } + + pub fn consecutive_send_failures(&self) -> u32 { + self.consecutive_send_failures + } } impl Default for SenderState { @@ -263,4 +303,76 @@ mod tests { s.update_report_interval_from_srtt(2_000_000); assert_eq!(s.report_interval(), Duration::from_millis(MAX_REPORT_INTERVAL_MS)); } + + #[test] + fn test_backoff_multiplier_progression() { + let mut s = SenderState::new(); + + // No failures → multiplier 1.0 + assert_eq!(s.send_failure_backoff_multiplier(), 1.0); + assert_eq!(s.consecutive_send_failures(), 0); + + // Progressive failures: 2^1, 2^2, 2^3, 2^4, 2^5 + let expected = [2.0, 4.0, 8.0, 16.0, 32.0]; + for (i, &exp) in expected.iter().enumerate() { + let count = s.record_send_failure(); + assert_eq!(count, (i + 1) as u32); + assert_eq!(s.send_failure_backoff_multiplier(), exp); + } + + // Beyond 5 failures: stays capped at 32.0 + s.record_send_failure(); // 6th + assert_eq!(s.send_failure_backoff_multiplier(), 32.0); + s.record_send_failure(); // 7th + assert_eq!(s.send_failure_backoff_multiplier(), 32.0); + } + + #[test] + fn test_backoff_reset_on_success() { + let mut s = SenderState::new(); + + // Accumulate failures + s.record_send_failure(); + s.record_send_failure(); + s.record_send_failure(); + assert_eq!(s.consecutive_send_failures(), 3); + assert_eq!(s.send_failure_backoff_multiplier(), 8.0); + + // Success resets and returns previous count + let prev = s.record_send_success(); + assert_eq!(prev, 3); + assert_eq!(s.consecutive_send_failures(), 0); + assert_eq!(s.send_failure_backoff_multiplier(), 1.0); + } + + #[test] + fn test_backoff_success_with_no_prior_failures() { + let mut s = SenderState::new(); + + // Success with no failures returns 0 + let prev = s.record_send_success(); + assert_eq!(prev, 0); + assert_eq!(s.consecutive_send_failures(), 0); + } + + #[test] + fn test_should_send_report_respects_backoff() { + let mut s = SenderState::new(); + let t0 = Instant::now(); + s.record_sent(1, 100, 500); + let _ = s.build_report(t0); + + // Record a failure: multiplier becomes 2.0 + s.record_send_failure(); + + s.record_sent(2, 200, 500); + + // At 1× interval: should NOT send (backoff requires 2×) + let t1 = t0 + s.report_interval() + Duration::from_millis(1); + assert!(!s.should_send_report(t1)); + + // At 2× interval: should send + let t2 = t0 + s.report_interval() * 2 + Duration::from_millis(1); + assert!(s.should_send_report(t2)); + } } diff --git a/src/node/handlers/mmp.rs b/src/node/handlers/mmp.rs index 9b4b420..7c2e079 100644 --- a/src/node/handlers/mmp.rs +++ b/src/node/handlers/mmp.rs @@ -313,15 +313,68 @@ impl Node { } } - // Send collected reports via session-layer encryption + // Send collected reports via session-layer encryption. + // Track per-destination success/failure for backoff and log suppression. + let mut send_results: Vec<(NodeAddr, bool)> = Vec::new(); for (dest_addr, msg_type, body) in reports { - if let Err(e) = self.send_session_msg(&dest_addr, msg_type, &body).await { - debug!( - dest = %self.peer_display_name(&dest_addr), - msg_type, - error = %e, - "Failed to send session MMP report" - ); + match self.send_session_msg(&dest_addr, msg_type, &body).await { + Ok(()) => { + send_results.push((dest_addr, true)); + } + Err(e) => { + // Peek at current failure count for log suppression + let failures = self.sessions.get(&dest_addr) + .and_then(|entry| entry.mmp()) + .map(|mmp| mmp.sender.consecutive_send_failures()) + .unwrap_or(0); + + if failures < 3 { + debug!( + dest = %self.peer_display_name(&dest_addr), + msg_type, + error = %e, + "Failed to send session MMP report" + ); + } else if failures == 3 { + debug!( + dest = %self.peer_display_name(&dest_addr), + "Suppressing further session MMP send failure logs" + ); + } + // failures > 3: silently suppressed + + send_results.push((dest_addr, false)); + } + } + } + + // Update backoff state from send results. + // Deduplicate: a destination counts as success if ANY report succeeded, + // failure only if ALL reports for that destination failed. + let mut dest_success: std::collections::HashMap = + std::collections::HashMap::new(); + for (dest, ok) in &send_results { + let entry = dest_success.entry(*dest).or_insert(false); + if *ok { + *entry = true; + } + } + for (dest_addr, success) in dest_success { + if let Some(entry) = self.sessions.get_mut(&dest_addr) + && let Some(mmp) = entry.mmp_mut() + { + if success { + let prev = mmp.sender.record_send_success(); + if prev > 3 { + debug!( + dest = %self.peer_display_name(&dest_addr), + consecutive_failures = prev, + "Resumed session MMP reporting" + ); + } + } else { + mmp.sender.record_send_failure(); + } } } }