feat(iroh)!: Retain stats for closed and abandoned paths in the path watcher (#3899)

## Description

Improves the path watcher for connections to reliably allow accessing
the stats of abandoned paths.

* Abandoned paths are not removed from the value of the watchable for a
connection's paths, as long as there are active watchers (subscribers)
* Only once all watchers are dropped, we drop the abandoned paths from
the watchable's value on the next update
* This means that if you keep a `PathWatcher` alive for the duration of
a connection, it will contain *all* paths the connection ever used.
* We use the changes from n0-computer/noq#386
and keep a `WeakPathHandle` for all paths in the watchable's value. This
prevents the path stats to be dropped within a quinn connection even if
the path is abandoned.
* We do no longer store path stats anywhere in iroh directly, but only
ever access them from quinn. Due to keeping the `WeakPathHandle`, we can
ensure that the stats are always available as long as the connection
hasn't been dropped.

The combination of all this gives us a reliable way to access final path
stats for all paths used in a connection, as long as you keep a
reference to the connection around, which is quite straightforward to do
and document.

## Breaking Changes

* `Connection::paths` and `ConnectionInfo::paths` now return a
`PathWatcher` (which still implements `n0_watcher::Watcher` but now also
is a named struct)
* `PathInfo::stats` and `PathInfo::rtt` now return `Option`. They return
`None` if the underlying connection has been droped.

## Notes & open questions

I spend quite some time going back-and-forth over different approaches.
Very open to other ideas, but I'm a bit out of further ideas currently
and this is the best I could come up with so far.

## 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: Frando

iroh/examples/monitor-connections.rs

@@ -111,7 +111,7 @@ impl Monitor {
                 Some(conn) = rx.recv() => {
                     let alpn = String::from_utf8_lossy(conn.alpn()).to_string();
                     let remote = conn.remote_id().fmt_short();
-                    let rtt = conn.paths().peek().iter().map(|p| p.stats().rtt).min();
+                    let rtt = conn.paths().peek().iter().map(|p| p.stats().expect("conn is not dropped").rtt).min();
                     info!(%remote, %alpn, ?rtt, "new connection");
                     tasks.spawn(async move {
                         match conn.closed().await {

iroh/examples/remote-info.rs

@@ -217,10 +217,11 @@ mod remote_map {
             if path.is_relay() {
                 self.relay_path = true;
             }
-            let stats = path.stats();
-            debug!("path update addr {:?} {stats:?}", path.remote_addr());
-            self.rtt_min = self.rtt_min.min(stats.rtt);
-            self.rtt_max = self.rtt_max.max(stats.rtt);
+            if let Some(stats) = path.stats() {
+                debug!("path update addr {:?} {stats:?}", path.remote_addr());
+                self.rtt_min = self.rtt_min.min(stats.rtt);
+                self.rtt_max = self.rtt_max.max(stats.rtt);
+            }
         }
     }
 
@@ -237,8 +238,9 @@ mod remote_map {
         /// Returns `None` if there are no active connections.
         pub fn current_min_rtt(&self) -> Option<Duration> {
             self.connections()
-                .flat_map(|c| c.paths().get())
-                .map(|path| path.stats().rtt)
+                .flat_map(|c| c.paths().get().into_iter())
+                .flat_map(|p| p.stats())
+                .map(|s| s.rtt)
                 .min()
         }
 

iroh/examples/transfer.rs

@@ -22,7 +22,6 @@
 //! ```
 
 use std::{
-    collections::BTreeMap,
     fmt,
     fs::File,
     net::{SocketAddr, SocketAddrV4, SocketAddrV6},
@@ -49,8 +48,8 @@ use iroh::{
     },
     dns::{DnsResolver, N0_DNS_ENDPOINT_ORIGIN_PROD, N0_DNS_ENDPOINT_ORIGIN_STAGING},
     endpoint::{
-        BindOpts, Connection, ConnectionError, PathId, PathInfoList, RecvStream, SendStream,
-        VarInt, WriteError,
+        BindOpts, Connection, ConnectionError, PathId, PathWatcher, RecvStream, SendStream, VarInt,
+        WriteError,
     },
 };
 use n0_error::{Result, StackResultExt, StdResultExt, anyerr, ensure_any};
@@ -582,8 +581,8 @@ async fn provide(endpoint: Endpoint, output: Output) -> Result<()> {
 async fn handle_connection(conn: Connection, output: Output) {
     let start = Instant::now();
     let remote_id = conn.remote_id();
-    let _guard = watch_conn_type(conn.paths(), Some(remote_id), output);
-    let stats_task = watch_path_stats(conn.clone());
+    let watcher = conn.paths();
+    let _guard = watch_conn_type(&conn, Some(remote_id), output);
 
     // Accept incoming streams in a loop until the connection is closed by the remote.
     let close_reason = loop {
@@ -615,8 +614,7 @@ async fn handle_connection(conn: Connection, output: Output) {
             duration: start.elapsed(),
         },
     );
-    let path_stats = stats_task.await.expect("path stats task panicked");
-    output.emit_with_remote(remote_id, path_stats);
+    output.emit_with_remote(remote_id, PathStats::from_watcher(watcher));
 }
 
 #[instrument("handle", skip_all, fields(id=send.id().index()))]
@@ -658,9 +656,9 @@ async fn fetch(
         remote_id,
         duration: start.elapsed(),
     });
+    let watcher = conn.paths();
     // Spawn a background task that prints connection type changes. Will be aborted on drop.
-    let _guard = watch_conn_type(conn.paths(), None, output);
-    let stats_task = watch_path_stats(conn.clone());
+    let _guard = watch_conn_type(&conn, None, output);
 
     output.emit(StartRequest { mode, length });
     // Perform requests depending on the request mode.
@@ -706,8 +704,7 @@ async fn fetch(
     });
 
     close_endpoint_with_timeout(&endpoint, output).await;
-    let path_stats = stats_task.await.expect("path stats task panicked");
-    output.emit(path_stats);
+    output.emit(PathStats::from_watcher(watcher));
 
     res
 }
@@ -866,7 +863,7 @@ fn parse_byte_size(s: &str) -> std::result::Result<u64, parse_size::Error> {
 }
 
 fn watch_conn_type(
-    paths_watcher: impl Watcher<Value = PathInfoList> + Send + Unpin + 'static,
+    conn: &Connection,
     remote_id: Option<EndpointId>,
     output: Output,
 ) -> AbortOnDropHandle<()> {
@@ -878,8 +875,8 @@ fn watch_conn_type(
             output.emit(event)
         }
     };
+    let mut stream = conn.paths().stream();
     let task = tokio::task::spawn(async move {
-        let mut stream = paths_watcher.stream();
         let mut previous = None;
         while let Some(paths) = stream.next().await {
             if let Some(path) = paths.iter().find(|p| p.is_selected()) {
@@ -890,7 +887,7 @@ fn watch_conn_type(
                 print(SelectedPath::Selected {
                     id: path.id(),
                     addr: path.remote_addr().clone(),
-                    rtt: path.rtt(),
+                    rtt: path.rtt().expect("conn is not dropped"),
                 });
                 previous = Some(path.clone());
             } else {
@@ -902,45 +899,6 @@ fn watch_conn_type(
     AbortOnDropHandle::new(task)
 }
 
-fn watch_path_stats(conn: iroh::endpoint::Connection) -> AbortOnDropHandle<PathStats> {
-    let task = tokio::spawn(async move {
-        let mut watcher = conn.paths();
-        let mut latest_stats_by_path = BTreeMap::new();
-        while conn.close_reason().is_none() {
-            n0_future::future::race(
-                async {
-                    conn.closed().await;
-                },
-                async {
-                    let _ = watcher.updated().await;
-                },
-            )
-            .await;
-            // Insert what could possibly be new path stats.
-            for path in watcher.get() {
-                let stats = path.stats();
-                latest_stats_by_path.insert(path.remote_addr().clone(), (path, stats));
-            }
-            // Update all stat values, even for paths that are removed by now.
-            for (path, stats) in latest_stats_by_path.values_mut() {
-                *stats = path.stats();
-            }
-        }
-        let list = latest_stats_by_path
-            .into_iter()
-            .map(|(addr, (info, stats))| PathData {
-                id: info.id(),
-                remote_addr: addr,
-                rtt: stats.rtt,
-                bytes_sent: stats.udp_tx.bytes,
-                bytes_recv: stats.udp_tx.bytes,
-            })
-            .collect();
-        PathStats { paths: list }
-    });
-    AbortOnDropHandle::new(task)
-}
-
 fn parse_ipv4_net(s: &str) -> Result<(SocketAddrV4, u8)> {
     let (net, port) = s.split_once(":").std_context("missing colon")?;
     let net: Ipv4Net = net.parse().std_context("invalid net")?;
@@ -1132,6 +1090,26 @@ struct PathStats {
     paths: Vec<PathData>,
 }
 
+impl PathStats {
+    fn from_watcher(mut watcher: PathWatcher) -> Self {
+        let list = watcher
+            .get()
+            .iter()
+            .filter_map(|info| {
+                let stats = info.stats()?;
+                Some(PathData {
+                    id: info.id(),
+                    remote_addr: info.remote_addr().clone(),
+                    rtt: stats.rtt,
+                    bytes_sent: stats.udp_tx.bytes,
+                    bytes_recv: stats.udp_rx.bytes,
+                })
+            })
+            .collect();
+        PathStats { paths: list }
+    }
+}
+
 impl fmt::Display for PathStats {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "Path stats:")?;

iroh/src/endpoint.rs

@@ -30,8 +30,11 @@ use url::Url;
 
 use self::hooks::EndpointHooksList;
 pub use super::socket::{
-    BindError, DirectAddr, DirectAddrType, PathInfo,
-    remote_map::{PathInfoList, RemoteInfo, Source, TransportAddrInfo, TransportAddrUsage},
+    BindError, DirectAddr, DirectAddrType,
+    remote_map::{
+        PathInfo, PathInfoList, PathInfoListIter, PathWatcher, RemoteInfo, Source,
+        TransportAddrInfo, TransportAddrUsage,
+    },
 };
 #[cfg(wasm_browser)]
 use crate::address_lookup::PkarrResolver;
@@ -1677,11 +1680,10 @@ mod tests {
     use iroh_base::{EndpointAddr, EndpointId, RelayUrl, SecretKey, TransportAddr};
     use iroh_relay::endpoint_info::UserData;
     use n0_error::{AnyError as Error, Result, StdResultExt};
-    use n0_future::{
-        BufferedStreamExt, StreamExt, future::now_or_never, stream, task::AbortOnDropHandle, time,
-    };
+    use n0_future::{BufferedStreamExt, StreamExt, future::now_or_never, stream, time};
     use n0_tracing_test::traced_test;
     use n0_watcher::Watcher;
+    use quinn::PathStats;
     use rand::SeedableRng;
     use rand_chacha::ChaCha8Rng;
     use tokio::sync::oneshot;
@@ -1693,7 +1695,7 @@ mod tests {
         address_lookup::memory::MemoryLookup,
         endpoint::{
             ApplicationClose, BindError, BindOpts, ConnectError, ConnectOptions,
-            ConnectWithOptsError, Connection, ConnectionError, PathInfo,
+            ConnectWithOptsError, Connection, ConnectionError, PathWatcher,
         },
         protocol::{AcceptError, ProtocolHandler, Router},
         test_utils::{QlogFileGroup, run_relay_server, run_relay_server_with},
@@ -2193,7 +2195,7 @@ mod tests {
             info!("Waiting for connection");
             'outer: while let Some(infos) = paths.next().await {
                 info!(?infos, "new PathInfos");
-                for info in infos {
+                for info in infos.iter() {
                     if info.is_ip() {
                         panic!("should not happen: {:?}", info);
                     }
@@ -3152,15 +3154,17 @@ mod tests {
         ));
         let transfer_size = 1_000_000;
 
-        async fn collect_path_infos(conn: Connection) -> BTreeMap<TransportAddr, PathInfo> {
-            let mut path_infos = BTreeMap::new();
-            let mut paths = conn.paths().stream();
-            while let Some(path_list) = paths.next().await {
-                for path in path_list {
-                    path_infos.insert(path.remote_addr().clone(), path);
-                }
-            }
-            path_infos
+        fn collect_stats(mut watcher: PathWatcher) -> BTreeMap<TransportAddr, PathStats> {
+            watcher
+                .get()
+                .iter()
+                .map(|info| {
+                    (
+                        info.remote_addr().clone(),
+                        info.stats().expect("conn is not yet dropped"),
+                    )
+                })
+                .collect()
         }
 
         let client = Endpoint::empty_builder(RelayMode::Custom(relay_map.clone()))
@@ -3178,49 +3182,49 @@ mod tests {
         let server_task = tokio::spawn(async move {
             let incoming = server.accept().await.anyerr()?;
             let conn = incoming.await.anyerr()?;
-            let stats_task = AbortOnDropHandle::new(tokio::spawn(collect_path_infos(conn.clone())));
+            let watcher = conn.paths();
             let (mut send, mut recv) = conn.accept_bi().await.anyerr()?;
             let msg = recv.read_to_end(transfer_size).await.anyerr()?;
             send.write_all(&msg).await.anyerr()?;
             send.finish().anyerr()?;
             conn.closed().await;
-            let stats = stats_task.await.std_context("server stats task failed")?;
+            let stats = collect_stats(watcher);
             Ok::<_, Error>(stats)
         });
 
         let conn = client.connect(server_addr, TEST_ALPN).await?;
-        let stats_task = AbortOnDropHandle::new(tokio::spawn(collect_path_infos(conn.clone())));
+        let watcher = conn.paths();
         let (mut send, mut recv) = conn.open_bi().await.anyerr()?;
         send.write_all(&vec![42u8; transfer_size]).await.anyerr()?;
         send.finish().anyerr()?;
         recv.read_to_end(transfer_size).await.anyerr()?;
         conn.close(0u32.into(), b"thanks, bye!");
         client.close().await;
-        let client_stats = stats_task.await.std_context("client stats task failed")?;
+        let client_stats = collect_stats(watcher);
         let server_stats = server_task.await.anyerr()??;
 
         info!("client stats: {client_stats:#?}");
         info!("server stats: {server_stats:#?}");
 
         let client_total_relay_tx = client_stats
-            .values()
-            .filter(|p| p.is_relay())
-            .map(|p| p.stats().udp_tx.bytes)
+            .iter()
+            .filter(|(remote, _stats)| remote.is_relay())
+            .map(|(_, stats)| stats.udp_tx.bytes)
             .sum::<u64>();
         let client_total_relay_rx = client_stats
-            .values()
-            .filter(|p| p.is_relay())
-            .map(|p| p.stats().udp_rx.bytes)
+            .iter()
+            .filter(|(remote, _stats)| remote.is_relay())
+            .map(|(_, stats)| stats.udp_rx.bytes)
             .sum::<u64>();
         let server_total_relay_tx = server_stats
-            .values()
-            .filter(|p| p.is_relay())
-            .map(|p| p.stats().udp_tx.bytes)
+            .iter()
+            .filter(|(remote, _stats)| remote.is_relay())
+            .map(|(_, stats)| stats.udp_tx.bytes)
             .sum::<u64>();
         let server_total_relay_rx = server_stats
-            .values()
-            .filter(|p| p.is_relay())
-            .map(|p| p.stats().udp_rx.bytes)
+            .iter()
+            .filter(|(remote, _stats)| remote.is_relay())
+            .map(|(_, stats)| stats.udp_rx.bytes)
             .sum::<u64>();
 
         info!(?client_total_relay_tx, "total");