refactor: improve path watching, add path stats (#3622)

## Description

This improves how we expose paths and path stats for connections, and
also updates feat-multipath to use
n0-computer/quinn#168.

* The watcher for open paths internally uses a SmallVec to not allocate
in the common case of not-too-many paths
* The path info for a path now includes a boolean whether this is the
currently selected primary transmission path
* We no longer expose PathIds to users
* We expose stats for paths from `Connection::path_stats`

## 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)

---------

Co-authored-by: Floris Bruynooghe <[email protected]>

Author: Frando

Cargo.lock

@@ -26,14 +26,7 @@ backon = { version = "1.4" }
 bytes = "1.7"
 crypto_box = { version = "0.10.0-pre.0", features = ["serde", "chacha20"] }
 data-encoding = "2.2"
-derive_more = { version = "2.0.1", features = [
-    "debug",
-    "display",
-    "from",
-    "try_into",
-    "deref",
-    "from_str"
-] }
+derive_more = { version = "2.0.1", features = ["debug", "display", "from", "try_into", "deref", "from_str", "into_iterator"] }
 ed25519-dalek = { version = "3.0.0-pre.1", features = ["serde", "rand_core", "zeroize", "pkcs8", "pem"] }
 http = "1"
 iroh-base = { version = "0.95.1", default-features = false, features = ["key", "relay"], path = "../iroh-base" }

iroh/Cargo.toml

@@ -25,8 +25,8 @@ use tracing::{debug, instrument, trace, warn};
 use url::Url;
 
 pub use super::magicsock::{
-    AddEndpointAddrError, ConnectionType, DirectAddr, DirectAddrType, PathInfo, PathsInfo,
-    endpoint_map::Source,
+    AddEndpointAddrError, ConnectionType, DirectAddr, DirectAddrType, PathInfo,
+    endpoint_map::{PathInfoList, Source},
 };
 #[cfg(wasm_browser)]
 use crate::discovery::pkarr::PkarrResolver;
@@ -54,12 +54,12 @@ pub mod presets;
 pub use quinn::{
     AcceptBi, AcceptUni, AckFrequencyConfig, ApplicationClose, Chunk, ClosedStream,
     ConnectionClose, ConnectionError, ConnectionStats, MtuDiscoveryConfig, OpenBi, OpenUni,
-    ReadDatagram, ReadError, ReadExactError, ReadToEndError, RecvStream, ResetError, RetryError,
-    SendDatagramError, SendStream, ServerConfig, StoppedError, StreamId, TransportConfig, VarInt,
-    WeakConnectionHandle, WriteError,
+    PathStats, ReadDatagram, ReadError, ReadExactError, ReadToEndError, RecvStream, ResetError,
+    RetryError, SendDatagramError, SendStream, ServerConfig, StoppedError, StreamId,
+    TransportConfig, VarInt, WeakConnectionHandle, WriteError,
 };
 pub use quinn_proto::{
-    FrameStats, PathStats, TransportError, TransportErrorCode, UdpStats, Written,
+    FrameStats, TransportError, TransportErrorCode, UdpStats, Written,
     congestion::{Controller, ControllerFactory},
     crypto::{
         AeadKey, CryptoError, ExportKeyingMaterialError, HandshakeTokenKey,
@@ -1802,11 +1802,11 @@ mod tests {
             let conn = ep.connect(dst, TEST_ALPN).await?;
             let mut send = conn.open_uni().await.anyerr()?;
             send.write_all(b"hello").await.anyerr()?;
-            let mut paths = conn.paths_info().stream();
+            let mut paths = conn.paths().stream();
             info!("Waiting for direct connection");
             while let Some(infos) = paths.next().await {
                 info!(?infos, "new PathInfos");
-                if infos.keys().any(|addr| addr.is_ip()) {
+                if infos.iter().any(|info| info.is_ip()) {
                     break;
                 }
             }
@@ -1893,15 +1893,15 @@ mod tests {
 
             // We should be connected via IP, because it is faster than the relay server.
             // TODO: Maybe not panic if this is not true?
-            let path_info = conn.paths_info().get();
+            let path_info = conn.paths().get();
             assert_eq!(path_info.len(), 1);
-            assert!(path_info.keys().next().unwrap().is_ip());
+            assert!(path_info.iter().next().unwrap().is_ip());
 
-            let mut paths = conn.paths_info().stream();
+            let mut paths = conn.paths().stream();
             time::timeout(Duration::from_secs(5), async move {
                 while let Some(infos) = paths.next().await {
                     info!(?infos, "new PathInfos");
-                    if infos.keys().any(|a| a.is_relay()) {
+                    if infos.iter().any(|info| info.is_relay()) {
                         break;
                     }
                 }
@@ -1942,11 +1942,11 @@ mod tests {
             // Wait for a relay connection to be added.  Client does all the asserting here,
             // we just want to wait so we get to see all the mechanics of the connection
             // being added on this side too.
-            let mut paths = conn.paths_info().stream();
+            let mut paths = conn.paths().stream();
             time::timeout(Duration::from_secs(5), async move {
                 while let Some(infos) = paths.next().await {
                     info!(?infos, "new PathInfos");
-                    if infos.keys().any(|a| a.is_relay()) {
+                    if infos.iter().any(|path| path.is_relay()) {
                         break;
                     }
                 }

iroh/src/endpoint.rs

@@ -18,7 +18,6 @@
 //! [module docs]: crate
 use std::{
     any::Any,
-    collections::HashMap,
     future::{Future, IntoFuture},
     net::{IpAddr, SocketAddr},
     pin::Pin,
@@ -31,19 +30,18 @@ use futures_util::{FutureExt, future::Shared};
 use iroh_base::EndpointId;
 use n0_error::{e, stack_error};
 use n0_future::time::Duration;
-use n0_watcher::{Watchable, Watcher};
+use n0_watcher::Watcher;
 use pin_project::pin_project;
 use quinn::{
     AcceptBi, AcceptUni, ConnectionError, ConnectionStats, OpenBi, OpenUni, ReadDatagram,
     RetryError, SendDatagramError, ServerConfig, VarInt,
 };
-use quinn_proto::PathId;
 use tracing::warn;
 
 use crate::{
     Endpoint,
     discovery::DiscoveryTask,
-    magicsock::{PathInfo, PathsInfo},
+    magicsock::endpoint_map::{PathInfoList, PathsWatchable},
 };
 
 /// Future produced by [`Endpoint::accept`].
@@ -239,38 +237,16 @@ fn conn_from_quinn_conn(
     }
     let remote_id = remote_id_from_quinn_conn(&conn)?;
     let alpn = alpn_from_quinn_conn(&conn).ok_or_else(|| e!(AuthenticationError::NoAlpn))?;
-    let paths_info_watchable = init_paths_info_watcher(&conn, ep);
-    let paths_info = paths_info_watchable.watch();
     // Register this connection with the magicsock.
-    ep.msock
-        .register_connection(remote_id, &conn, paths_info_watchable.clone());
+    let paths = ep.msock.register_connection(remote_id, &conn);
     Ok(Connection {
         remote_id,
         alpn,
         inner: conn,
-        paths_info,
+        paths,
     })
 }
 
-fn init_paths_info_watcher(conn: &quinn::Connection, ep: &Endpoint) -> Watchable<PathsInfo> {
-    let mut paths_info = HashMap::with_capacity(5);
-    if let Some(path0) = conn.path(PathId::ZERO) {
-        // This all is supposed to be infallible, but anyway.
-        if let Ok(remote) = path0.remote_address() {
-            if let Some(remote) = ep.msock.endpoint_map.transport_addr_from_mapped(remote) {
-                paths_info.insert(
-                    remote.clone(),
-                    PathInfo {
-                        remote,
-                        path_id: PathId::ZERO,
-                    },
-                );
-            }
-        }
-    }
-    n0_watcher::Watchable::new(paths_info)
-}
-
 /// Returns the [`EndpointId`] from the peer's TLS certificate.
 ///
 /// The [`PublicKey`] of an endpoint is also known as an [`EndpointId`].  This [`PublicKey`] is
@@ -1245,7 +1221,7 @@ pub struct Connection {
     inner: quinn::Connection,
     remote_id: EndpointId,
     alpn: Vec<u8>,
-    paths_info: n0_watcher::Direct<PathsInfo>,
+    paths: PathsWatchable,
 }
 
 #[allow(missing_docs)]
@@ -1480,13 +1456,21 @@ impl Connection {
         self.inner.stable_id()
     }
 
-    /// Returns information about the network paths in use by this connection.
+    /// Returns a [`Watcher`] for the network paths of this connection.
     ///
     /// A connection can have several network paths to the remote endpoint, commonly there
-    /// will be a path via the relay server and a holepunched path.  This returns all the
-    /// paths in use by this connection.
-    pub fn paths_info(&self) -> impl Watcher<Value = PathsInfo> {
-        self.paths_info.clone()
+    /// will be a path via the relay server and a holepunched path.
+    ///
+    /// The watcher is updated whenever a path is opened or closed, or when the path selected
+    /// for transmission changes (see [`PathInfo::is_selected`]).
+    ///
+    /// The [`PathInfoList`] returned from the watcher contains a [`PathInfo`] for each
+    /// transmission path.
+    ///
+    /// [`PathInfo::is_selected`]: crate::magicsock::PathInfo::is_selected
+    /// [`PathInfo`]: crate::magicsock::PathInfo
+    pub fn paths(&self) -> impl Watcher<Value = PathInfoList> {
+        self.paths.watch(self.inner.weak_handle())
     }
 
     /// Derives keying material from this connection's TLS session secrets.