wip: some congestion trickery

Author: Arqu

iroh/src/magicsock.rs

@@ -95,6 +95,19 @@ const ENDPOINTS_FRESH_ENOUGH_DURATION: Duration = Duration::from_secs(27);
 
 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
 
+/// Jitter range for heartbeat intervals (±25% of base interval).
+/// This prevents synchronized heartbeat storms across many nodes.
+const HEARTBEAT_JITTER_PCT: f64 = 0.25;
+
+/// Create a jittered interval to prevent synchronized heartbeat storms.
+fn jittered_interval(base: Duration) -> time::Interval {
+    let jitter_range = base.as_secs_f64() * HEARTBEAT_JITTER_PCT;
+    let jitter = rand::thread_rng().gen_range(-jitter_range..=jitter_range);
+    let jittered = base.as_secs_f64() + jitter;
+    let duration = Duration::from_secs_f64(jittered.max(0.1));
+    time::interval(duration)
+}
+
 /// Contains options for `MagicSock::listen`.
 #[derive(derive_more::Debug)]
 pub(crate) struct Options {
@@ -1850,7 +1863,7 @@ impl Actor {
         let mut current_netmon_state = self.netmon_watcher.get();
 
         #[cfg(not(wasm_browser))]
-        let mut direct_addr_heartbeat_timer = time::interval(HEARTBEAT_INTERVAL);
+        let mut direct_addr_heartbeat_timer = jittered_interval(HEARTBEAT_INTERVAL);
 
         #[cfg(not(wasm_browser))]
         let mut portmap_watcher = self

iroh/src/magicsock/node_map/node_state.rs

@@ -460,11 +460,21 @@ impl NodeState {
                             // which we should have received the pong, clear best addr and
                             // pong.  Both are used to select this path again, but we know
                             // it's not a usable path now.
-                            path_state.validity = PathValidity::empty();
+                            // Record ping failure and only clear validity after threshold
+                            path_state.validity.record_ping_failure();
                             metrics.path_ping_failures.inc();
 
                             path_state.validity.record_metrics(metrics);
-                            metrics.path_marked_outdated.inc();
+
+                            if path_state.validity.should_mark_outdated() {
+                                debug!(
+                                    "path {} marked outdated after {} consecutive failures",
+                                    addr,
+                                    path_state.validity.consecutive_failures()
+                                );
+                                metrics.path_marked_outdated.inc();
+                                path_state.validity = PathValidity::empty();
+                            }
                         }
                     }
                 }

iroh/src/magicsock/node_map/path_state.rs

@@ -125,7 +125,14 @@ impl PathState {
             }
         }
 
+        let had_failures = self.validity.consecutive_failures() > 0;
+
         self.validity.update_pong(r.pong_at, r.latency);
+        self.validity.reset_failures();
+
+        if had_failures {
+            metrics.path_failure_resets.inc();
+        }
 
         self.validity.record_metrics(metrics);
     }
@@ -134,17 +141,20 @@ impl PathState {
         self.last_payload_msg = Some(now);
         self.validity
             .receive_payload(now, path_validity::Source::QuicPayload);
+        self.validity.reset_failures();
     }
 
     #[cfg(test)]
     pub(super) fn with_pong_reply(node_id: NodeId, r: PongReply) -> Self {
+        let mut validity = PathValidity::new(r.pong_at, r.latency);
+        validity.reset_failures();
         PathState {
             node_id,
             path: r.from.clone(),
             last_ping: None,
             last_got_ping: None,
             call_me_maybe_time: None,
-            validity: PathValidity::new(r.pong_at, r.latency),
+            validity,
             last_payload_msg: None,
             sources: HashMap::new(),
         }

iroh/src/magicsock/node_map/path_validity.rs

@@ -9,7 +9,14 @@ use crate::magicsock::Metrics as MagicsockMetrics;
 /// currently trusted.
 ///
 /// If trust goes away, it can be brought back with another valid DISCO UDP pong.
-const TRUST_UDP_ADDR_DURATION: Duration = Duration::from_millis(6500);
+///
+/// Increased from 6.5s to 12s to be more resilient under congestion.
+const TRUST_UDP_ADDR_DURATION: Duration = Duration::from_secs(12);
+
+/// Number of consecutive ping failures required before marking a path as outdated.
+///
+/// This implements a tolerance to prevent temporary packet loss from causing path degradation.
+const PING_FAILURE_THRESHOLD: u8 = 3;
 
 /// Tracks a path's validity.
 ///
@@ -27,6 +34,7 @@ struct Inner {
     latest_pong: Instant,
     latency: Duration,
     trust_until: Instant,
+    consecutive_failures: u8,
     congestion_metrics: CongestionMetrics,
 }
 
@@ -150,6 +158,7 @@ impl PathValidity {
             trust_until: pong_at + Source::ReceivedPong.trust_duration(),
             latest_pong: pong_at,
             latency,
+            consecutive_failures: 0,
             congestion_metrics: metrics,
         }))
     }
@@ -161,6 +170,7 @@ impl PathValidity {
                 inner.trust_until = pong_at + Source::ReceivedPong.trust_duration();
                 inner.latest_pong = pong_at;
                 inner.latency = latency;
+                inner.consecutive_failures = 0;
                 inner.congestion_metrics.add_latency_sample(latency);
             }
             None => {
@@ -226,6 +236,31 @@ impl PathValidity {
         Some(self.0.as_ref()?.latest_pong)
     }
 
+    /// Record a ping failure (timeout or no response).
+    ///
+    /// Only marks the path as outdated after PING_FAILURE_THRESHOLD consecutive failures.
+    pub(super) fn record_ping_failure(&mut self) {
+        let Some(state) = self.0.as_mut() else {
+            return;
+        };
+        state.consecutive_failures = state.consecutive_failures.saturating_add(1);
+    }
+
+    /// Check if path should be considered outdated based on consecutive failures.
+    pub(super) fn should_mark_outdated(&self) -> bool {
+        self.0
+            .as_ref()
+            .map(|state| state.consecutive_failures >= PING_FAILURE_THRESHOLD)
+            .unwrap_or(false)
+    }
+
+    /// Reset consecutive failure counter (called when we receive activity).
+    pub(super) fn reset_failures(&mut self) {
+        if let Some(state) = self.0.as_mut() {
+            state.consecutive_failures = 0;
+        }
+    }
+
     /// Record that a ping was sent on this path.
     pub(super) fn record_ping_sent(&mut self) {
         if let Some(state) = self.0.as_mut() {
@@ -267,6 +302,14 @@ impl PathValidity {
             .and_then(|state| state.congestion_metrics.avg_latency())
     }
 
+    /// Get the number of consecutive failures.
+    pub(super) fn consecutive_failures(&self) -> u8 {
+        self.0
+            .as_ref()
+            .map(|state| state.consecutive_failures)
+            .unwrap_or(0)
+    }
+
     /// Record congestion metrics to the metrics system.
     /// Should be called periodically or on significant events.
     pub(super) fn record_metrics(&self, metrics: &MagicsockMetrics) {
@@ -302,7 +345,7 @@ impl Inner {
 mod tests {
     use n0_future::time::{Duration, Instant};
 
-    use super::{PathValidity, Source, TRUST_UDP_ADDR_DURATION};
+    use super::{PING_FAILURE_THRESHOLD, PathValidity, Source, TRUST_UDP_ADDR_DURATION};
 
     #[tokio::test(start_paused = true)]
     async fn test_basic_path_validity_lifetime() {
@@ -330,6 +373,32 @@ mod tests {
         assert!(!validity.is_valid(Instant::now()));
         assert!(validity.is_outdated(Instant::now()));
     }
+
+    #[tokio::test]
+    async fn test_multiple_ping_failures() {
+        let mut validity = PathValidity::new(Instant::now(), Duration::from_millis(20));
+
+        // First failure should not mark as outdated
+        validity.record_ping_failure();
+        assert!(!validity.should_mark_outdated());
+        assert_eq!(validity.consecutive_failures(), 1);
+
+        // Second failure should not mark as outdated
+        validity.record_ping_failure();
+        assert!(!validity.should_mark_outdated());
+        assert_eq!(validity.consecutive_failures(), 2);
+
+        // Third failure should mark as outdated (threshold = 3)
+        validity.record_ping_failure();
+        assert!(validity.should_mark_outdated());
+        assert_eq!(validity.consecutive_failures(), PING_FAILURE_THRESHOLD);
+
+        // Receiving pong should reset failures
+        validity.update_pong(Instant::now(), Duration::from_millis(20));
+        assert_eq!(validity.consecutive_failures(), 0);
+        assert!(!validity.should_mark_outdated());
+    }
+
     #[tokio::test]
     async fn test_congestion_metrics() {
         let mut validity = PathValidity::new(Instant::now(), Duration::from_millis(10));

iroh/src/magicsock/transports/relay.rs

@@ -39,11 +39,14 @@ pub(crate) struct RelayTransport {
 
 impl RelayTransport {
     pub(crate) fn new(config: RelayActorConfig) -> Self {
-        let (relay_datagram_send_tx, relay_datagram_send_rx) = mpsc::channel(256);
+        // Increased from 256 to 1024 to better handle congestion and burst traffic
+        let (relay_datagram_send_tx, relay_datagram_send_rx) = mpsc::channel(1024);
 
-        let (relay_datagram_recv_tx, relay_datagram_recv_rx) = mpsc::channel(512);
+        // Increased from 512 to 2048 to reduce drops under load
+        let (relay_datagram_recv_tx, relay_datagram_recv_rx) = mpsc::channel(2048);
 
-        let (actor_sender, actor_receiver) = mpsc::channel(256);
+        // Increased from 256 to 512 for actor control messages
+        let (actor_sender, actor_receiver) = mpsc::channel(512);
 
         let my_node_id = config.secret_key.public();
         let my_relay = config.my_relay.clone();