feat: congestion metrics (#3491)

## Description

Needs n0-computer/iroh-metrics#35 and updating
the Cargo.toml's again.

## Breaking Changes

<!-- Optional, if there are any breaking changes document them,
including how to migrate older code. -->

## Notes & open questions

<!-- Any notes, remarks or open questions you have to make about the PR.
-->

## Change checklist
<!-- Remove any that are not relevant. -->
- [ ] Self-review.
- [ ] Documentation updates following the [style
guide](https://rust-lang.github.io/rfcs/1574-more-api-documentation-conventions.html#appendix-a-full-conventions-text),
if relevant.
- [ ] Tests if relevant.
- [ ] All breaking changes documented.
- [ ] List all breaking changes in the above "Breaking Changes" section.
- [ ] Open an issue or PR on any number0 repos that are affected by this
breaking change. Give guidance on how the updates should be handled or
do the actual updates themselves. The major ones are:
    - [ ] [`quic-rpc`](https://github.com/n0-computer/quic-rpc)
    - [ ] [`iroh-gossip`](https://github.com/n0-computer/iroh-gossip)
    - [ ] [`iroh-blobs`](https://github.com/n0-computer/iroh-blobs)
    - [ ] [`dumbpipe`](https://github.com/n0-computer/dumbpipe)
    - [ ] [`sendme`](https://github.com/n0-computer/sendme)

Author: Arqu

Cargo.lock

@@ -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 = { version = "0.36", features = ["service"] }
 lru = "0.16"
 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 = { version = "0.36", default-features = false }
 n0-future = "0.1.2"
 num_enum = "0.7"
 pin-project = "1"

iroh-relay/Cargo.toml

@@ -82,7 +82,7 @@ futures-buffered = "0.2.11"
 pkcs8 = "0.11.0-rc.7"
 
 # metrics
-iroh-metrics = { version = "0.35", default-features = false }
+iroh-metrics = { version = "0.36", default-features = false }
 
 # local-swarm-discovery
 swarm-discovery = { version = "0.4", optional = true }
@@ -104,7 +104,7 @@ parse-size = { version = "=1.0.0", optional = true, features = ['std'] } # pinne
 hickory-resolver = "0.25.1"
 igd-next = { version = "0.16", features = ["aio_tokio"] }
 netdev = { version = "0.36.0" }
-portmapper = { version = "0.9", default-features = false }
+portmapper = { version = "0.10", default-features = false }
 quinn = { package = "iroh-quinn", version = "0.14.0", default-features = false, features = ["runtime-tokio", "rustls-ring"] }
 tokio = { version = "1", features = [
     "io-util",

iroh/Cargo.toml

@@ -9,7 +9,7 @@ publish = false
 bytes = "1.7"
 hdrhistogram = { version = "7.2", default-features = false }
 iroh = { path = ".." }
-iroh-metrics = "0.35"
+iroh-metrics = "0.36"
 n0-future = "0.1.1"
 n0-snafu = "0.2.0"
 quinn = { package = "iroh-quinn", version = "0.14" }

iroh/bench/Cargo.toml

@@ -792,7 +792,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();
@@ -2049,7 +2050,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();
@@ -2249,7 +2252,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,12 +1,12 @@
-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")]
+#[metrics(name = "magicsock", default)]
 pub struct Metrics {
     pub update_direct_addrs: Counter,
 
@@ -77,4 +77,23 @@ 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.
+    #[default(Histogram::new(vec![0.0, 0.01, 0.05, 0.1, 0.2, 0.5, 1.0]))]
+    pub path_packet_loss_rate: Histogram,
+    /// Histogram of RTT variance (in milliseconds) as a congestion indicator.
+    #[default(Histogram::new(vec![0.0, 1.0, 5.0, 10.0, 20.0, 50.0, 100.0, 200.0]))]
+    pub path_rtt_variance_ms: Histogram,
+    /// Histogram of path quality scores (0.0-1.0).
+    #[default(Histogram::new(vec![0.0, 0.3, 0.5, 0.7, 0.85, 0.95, 1.0]))]
+    pub path_quality_score: Histogram,
 }

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);
         }
     }
 
@@ -518,9 +529,15 @@ impl NodeMapInner {
             .and_then(|ep| ep.latency())
     }
 
-    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);
             }
@@ -553,7 +570,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)
             }
         }
     }