WIP: testing some congestion metrics and fixes

Author: Arqu

Cargo.lock

@@ -40,3 +40,6 @@ unexpected_cfgs = { level = "warn", check-cfg = ["cfg(iroh_docsrs)", "cfg(iroh_l
 
 [workspace.lints.clippy]
 unused-async = "warn"
+
+[patch.crates-io]
+iroh-metrics = { git = "https://github.com/n0-computer/iroh-metrics", branch = "arqu/histograms" }

Cargo.toml

@@ -28,7 +28,7 @@ hickory-server = { version = "0.25.1", features = ["https-ring"] }
 http = "1.0.0"
 humantime = "2.2.0"
 humantime-serde = "1.1.1"
-iroh-metrics = { version = "0.35", features = ["service"] }
+iroh-metrics = { git = "https://github.com/n0-computer/iroh-metrics", branch = "arqu/histograms", features = ["service"] }
 lru = "0.13"
 n0-future = "0.1.2"
 n0-snafu = "0.2.2"

iroh-dns-server/Cargo.toml

@@ -32,7 +32,7 @@ http-body-util = "0.1.0"
 hyper = { version = "1", features = ["server", "client", "http1"] }
 hyper-util = "0.1.1"
 iroh-base = { version = "0.92.0", path = "../iroh-base", default-features = false, features = ["key", "relay"] }
-iroh-metrics = { version = "0.35", default-features = false }
+iroh-metrics = { git = "https://github.com/n0-computer/iroh-metrics", branch = "arqu/histograms", default-features = false }
 n0-future = "0.1.2"
 num_enum = "0.7"
 pin-project = "1"

iroh-relay/Cargo.toml

@@ -85,7 +85,7 @@ futures-buffered = "0.2.11"
 spki = { version = "0.7.3", features = ["std"] }
 
 # metrics
-iroh-metrics = { version = "0.35", default-features = false }
+iroh-metrics = { git = "https://github.com/n0-computer/iroh-metrics", branch = "arqu/histograms", default-features = false }
 
 # local-swarm-discovery
 swarm-discovery = { version = "0.4", optional = true }

iroh/Cargo.toml

@@ -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 {
@@ -795,7 +808,8 @@ impl MagicSock {
             }
             disco::Message::Pong(pong) => {
                 self.metrics.magicsock.recv_disco_pong.inc();
-                self.node_map.handle_pong(sender, src, pong);
+                self.node_map
+                    .handle_pong(sender, src, pong, &self.metrics.magicsock);
             }
             disco::Message::CallMeMaybe(cm) => {
                 self.metrics.magicsock.recv_disco_call_me_maybe.inc();
@@ -1849,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
@@ -2063,7 +2077,9 @@ impl Actor {
     async fn handle_actor_message(&mut self, msg: ActorMessage) {
         match msg {
             ActorMessage::EndpointPingExpired(id, txid) => {
-                self.msock.node_map.notify_ping_timeout(id, txid);
+                self.msock
+                    .node_map
+                    .notify_ping_timeout(id, txid, &self.msock.metrics.magicsock);
             }
             ActorMessage::NetworkChange => {
                 self.network_monitor.network_change().await.ok();
@@ -2255,7 +2271,9 @@ impl Actor {
     /// This is called when connectivity changes enough that we no longer trust the old routes.
     #[instrument(skip_all)]
     fn reset_endpoint_states(&mut self) {
-        self.msock.node_map.reset_node_states()
+        self.msock
+            .node_map
+            .reset_node_states(&self.msock.metrics.magicsock)
     }
 }
 

iroh/src/magicsock.rs

@@ -1,10 +1,10 @@
-use iroh_metrics::{Counter, MetricsGroup};
+use iroh_metrics::{Counter, Histogram, MetricsGroup};
 use serde::{Deserialize, Serialize};
 
 /// Enum of metrics for the module
 // TODO(frando): Add description doc strings for each metric.
 #[allow(missing_docs)]
-#[derive(Debug, Default, Serialize, Deserialize, MetricsGroup)]
+#[derive(Debug, Serialize, Deserialize, MetricsGroup)]
 #[non_exhaustive]
 #[metrics(name = "magicsock")]
 pub struct Metrics {
@@ -77,4 +77,88 @@ pub struct Metrics {
     pub connection_handshake_success: Counter,
     /// Number of connections with a successful handshake that became direct.
     pub connection_became_direct: Counter,
+
+    /*
+     * Path Congestion Metrics
+     */
+    /// Number of times a path was marked as outdated due to consecutive ping failures.
+    pub path_marked_outdated: Counter,
+    /// Number of ping failures recorded across all paths.
+    pub path_ping_failures: Counter,
+    /// Number of consecutive failure resets (path recovered).
+    pub path_failure_resets: Counter,
+    /// Histogram of packet loss rates (0.0-1.0) observed on UDP paths.
+    pub path_packet_loss_rate: Histogram,
+    /// Histogram of RTT variance (in milliseconds) as a congestion indicator.
+    pub path_rtt_variance_ms: Histogram,
+    /// Histogram of path quality scores (0.0-1.0).
+    pub path_quality_score: Histogram,
+}
+
+impl Default for Metrics {
+    fn default() -> Self {
+        Self {
+            update_direct_addrs: Counter::default(),
+            send_ipv4: Counter::default(),
+            send_ipv6: Counter::default(),
+            send_relay: Counter::default(),
+            send_relay_error: Counter::default(),
+            send_data: Counter::default(),
+            send_data_network_down: Counter::default(),
+            recv_data_relay: Counter::default(),
+            recv_data_ipv4: Counter::default(),
+            recv_data_ipv6: Counter::default(),
+            recv_datagrams: Counter::default(),
+            recv_gro_datagrams: Counter::default(),
+            send_disco_udp: Counter::default(),
+            send_disco_relay: Counter::default(),
+            sent_disco_udp: Counter::default(),
+            sent_disco_relay: Counter::default(),
+            sent_disco_ping: Counter::default(),
+            sent_disco_pong: Counter::default(),
+            sent_disco_call_me_maybe: Counter::default(),
+            recv_disco_bad_key: Counter::default(),
+            recv_disco_bad_parse: Counter::default(),
+            recv_disco_udp: Counter::default(),
+            recv_disco_relay: Counter::default(),
+            recv_disco_ping: Counter::default(),
+            recv_disco_pong: Counter::default(),
+            recv_disco_call_me_maybe: Counter::default(),
+            recv_disco_call_me_maybe_bad_disco: Counter::default(),
+            relay_home_change: Counter::default(),
+            num_direct_conns_added: Counter::default(),
+            num_direct_conns_removed: Counter::default(),
+            num_relay_conns_added: Counter::default(),
+            num_relay_conns_removed: Counter::default(),
+            actor_tick_main: Counter::default(),
+            actor_tick_msg: Counter::default(),
+            actor_tick_re_stun: Counter::default(),
+            actor_tick_portmap_changed: Counter::default(),
+            actor_tick_direct_addr_heartbeat: Counter::default(),
+            actor_link_change: Counter::default(),
+            actor_tick_other: Counter::default(),
+            nodes_contacted: Counter::default(),
+            nodes_contacted_directly: Counter::default(),
+            connection_handshake_success: Counter::default(),
+            connection_became_direct: Counter::default(),
+            path_marked_outdated: Counter::default(),
+            path_ping_failures: Counter::default(),
+            path_failure_resets: Counter::default(),
+            path_packet_loss_rate: packet_loss_buckets(),
+            path_rtt_variance_ms: rtt_variance_buckets(),
+            path_quality_score: quality_score_buckets(),
+        }
+    }
+}
+
+fn packet_loss_buckets() -> Histogram {
+    Histogram::new(vec![0.0, 0.01, 0.05, 0.1, 0.2, 0.5, 1.0])
+}
+
+fn rtt_variance_buckets() -> Histogram {
+    Histogram::new(vec![0.0, 1.0, 5.0, 10.0, 20.0, 50.0, 100.0, 200.0])
+}
+
+fn quality_score_buckets() -> Histogram {
+    Histogram::new(vec![0.0, 0.3, 0.5, 0.7, 0.85, 0.95, 1.0])
 }

iroh/src/magicsock/metrics.rs

@@ -192,14 +192,19 @@ impl NodeMap {
         }
     }
 
-    pub(super) fn notify_ping_timeout(&self, id: usize, tx_id: stun_rs::TransactionId) {
+    pub(super) fn notify_ping_timeout(
+        &self,
+        id: usize,
+        tx_id: stun_rs::TransactionId,
+        metrics: &Metrics,
+    ) {
         if let Some(ep) = self
             .inner
             .lock()
             .expect("poisoned")
             .get_mut(NodeStateKey::Idx(id))
         {
-            ep.ping_timeout(tx_id, Instant::now());
+            ep.ping_timeout(tx_id, Instant::now(), metrics);
         }
     }
 
@@ -228,11 +233,17 @@ impl NodeMap {
             .handle_ping(sender, src, tx_id)
     }
 
-    pub(super) fn handle_pong(&self, sender: PublicKey, src: &transports::Addr, pong: Pong) {
+    pub(super) fn handle_pong(
+        &self,
+        sender: PublicKey,
+        src: &transports::Addr,
+        pong: Pong,
+        metrics: &Metrics,
+    ) {
         self.inner
             .lock()
             .expect("poisoned")
-            .handle_pong(sender, src, pong)
+            .handle_pong(sender, src, pong, metrics)
     }
 
     #[must_use = "actions must be handled"]
@@ -268,11 +279,11 @@ impl NodeMap {
         Some((public_key, udp_addr, relay_url, ping_actions))
     }
 
-    pub(super) fn reset_node_states(&self) {
+    pub(super) fn reset_node_states(&self, metrics: &Metrics) {
         let now = Instant::now();
         let mut inner = self.inner.lock().expect("poisoned");
         for (_, ep) in inner.node_states_mut() {
-            ep.note_connectivity_change(now);
+            ep.note_connectivity_change(now, metrics);
         }
     }
 
@@ -506,9 +517,15 @@ impl NodeMapInner {
             .map(|ep| ep.conn_type())
     }
 
-    fn handle_pong(&mut self, sender: NodeId, src: &transports::Addr, pong: Pong) {
+    fn handle_pong(
+        &mut self,
+        sender: NodeId,
+        src: &transports::Addr,
+        pong: Pong,
+        metrics: &Metrics,
+    ) {
         if let Some(ns) = self.get_mut(NodeStateKey::NodeId(sender)).as_mut() {
-            let insert = ns.handle_pong(&pong, src.clone().into());
+            let insert = ns.handle_pong(&pong, src.clone().into(), metrics);
             if let Some((src, key)) = insert {
                 self.set_node_key_for_ip_port(src, &key);
             }
@@ -541,7 +558,7 @@ impl NodeMapInner {
             Some(ns) => {
                 debug!(endpoints = ?cm.my_numbers, "received call-me-maybe");
 
-                ns.handle_call_me_maybe(cm)
+                ns.handle_call_me_maybe(cm, metrics)
             }
         }
     }