feat: remote_info and watcher for selected path

Author: Frando

iroh/examples/monitor-connections.rs

@@ -110,7 +110,7 @@ impl Monitor {
                 Some(conn) = rx.recv() => {
                     let alpn = String::from_utf8_lossy(conn.alpn()).to_string();
                     let remote = conn.remote_id().fmt_short();
-                    info!(%remote, %alpn, rtt=?conn.latency(), "new connection");
+                    info!(%remote, %alpn, rtt=?conn.rtt(), "new connection");
                     tasks.spawn(async move {
                         match conn.closed().await {
                             Some((close_reason, stats)) => {

iroh/examples/transfer.rs

@@ -522,13 +522,21 @@ fn parse_byte_size(s: &str) -> std::result::Result<u64, parse_size::Error> {
 }
 
 fn watch_conn_type(endpoint: &Endpoint, endpoint_id: EndpointId) -> AbortOnDropHandle<()> {
-    let mut stream = endpoint.conn_type(endpoint_id).unwrap().stream();
+    let info = endpoint.remote_info(endpoint_id).unwrap();
+    let mut stream = info.selected_path().stream();
     let task = tokio::task::spawn(async move {
-        while let Some(conn_type) = stream.next().await {
-            println!(
-                "[{}] Connection type changed to: {conn_type}",
-                endpoint_id.fmt_short()
-            );
+        while let Some(selected_path) = stream.next().await {
+            if let Some(selected_path) = selected_path {
+                let label = match selected_path {
+                    TransportAddr::Ip(addr) => format!("direct ({addr})"),
+                    TransportAddr::Relay(url) => format!("relay ({url})"),
+                    _ => format!("unknown transport"),
+                };
+                println!(
+                    "[{}] Connection type changed to: {label}",
+                    endpoint_id.fmt_short()
+                );
+            }
         }
     });
     AbortOnDropHandle::new(task)

iroh/src/endpoint.rs

@@ -26,7 +26,7 @@ use url::Url;
 
 pub use super::magicsock::{
     AddEndpointAddrError, ConnectionType, DirectAddr, DirectAddrType, PathInfo, PathsInfo,
-    endpoint_map::Source,
+    RemoteInfo, endpoint_map::Source,
 };
 #[cfg(wasm_browser)]
 use crate::discovery::pkarr::PkarrResolver;
@@ -55,8 +55,8 @@ 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,
+    SendDatagramError, SendStream, ServerConfig, Side, StoppedError, StreamId, TransportConfig,
+    VarInt, WeakConnectionHandle, WriteError,
 };
 pub use quinn_proto::{
     FrameStats, PathStats, TransportError, TransportErrorCode, UdpStats, Written,
@@ -1005,38 +1005,24 @@ impl Endpoint {
     //
     // Partially they return things passed into the builder.
 
-    /// Returns a [`Watcher`] that reports the current connection type and any changes for
-    /// given remote endpoint.
+    /// Information about a remote endpoint.
     ///
-    /// This watcher allows observing a stream of [`ConnectionType`] items by calling
-    /// [`Watcher::stream()`]. If the underlying connection to a remote endpoint changes, it will
-    /// yield a new item.  These connection changes are when the connection switches between
-    /// using the Relay server and a direct connection.
+    /// From the [`RemoteInfo`] you can watch which path is selected, get the current
+    /// round-trip time (latency), and get a list of [`ConnectionInfo`].
     ///
-    /// Note that this does not guarantee each connection change is yielded in the stream.
-    /// If the connection type changes several times before this stream is polled, only the
-    /// last recorded state is returned.  This can be observed e.g. right at the start of a
-    /// connection when the switch from a relayed to a direct connection can be so fast that
-    /// the relayed state is never exposed.
-    ///
-    /// If there is currently a connection with the remote endpoint, then using [`Watcher::get`]
-    /// will immediately return either [`ConnectionType::Relay`], [`ConnectionType::Direct`]
-    /// or [`ConnectionType::Mixed`].
-    ///
-    /// It is possible for the connection type to be [`ConnectionType::None`] if you've
-    /// recently connected to this endpoint id but previous methods of reaching the endpoint have
-    /// become inaccessible.
-    ///
-    /// Will return `None` if we do not have any address information for the given `endpoint_id`.
-    pub fn conn_type(&self, endpoint_id: EndpointId) -> Option<n0_watcher::Direct<ConnectionType>> {
-        self.msock.conn_type(endpoint_id)
+    /// Returns `None` if we don't have any state for this remote.
+    pub fn remote_info(&self, endpoint_id: EndpointId) -> Option<RemoteInfo> {
+        self.msock.endpoint_map.remote_info(endpoint_id)
     }
 
-    /// Returns the currently lowest latency for this endpoint.
+    /// Returns a list of all remote endpoints that this endpoint is dealing with.
+    ///
+    /// This includes all endpoints to which we have active connections. It also may include endpoints
+    /// to which we are in the process of connecting, or have recently been connected to.
     ///
-    /// Will return `None` if we do not have any address information for the given `endpoint_id`.
-    pub async fn latency(&self, endpoint_id: EndpointId) -> Option<Duration> {
-        self.msock.latency(endpoint_id).await
+    /// TODO: Expand docs.
+    pub fn remotes(&self) -> Vec<RemoteInfo> {
+        self.msock.endpoint_map.remotes()
     }
 
     /// Returns the DNS resolver used in this [`Endpoint`].

iroh/src/endpoint/connection.rs

@@ -35,7 +35,7 @@ use n0_watcher::{Watchable, Watcher};
 use pin_project::pin_project;
 use quinn::{
     AcceptBi, AcceptUni, ConnectionError, ConnectionStats, OpenBi, OpenUni, ReadDatagram,
-    RetryError, SendDatagramError, ServerConfig, VarInt, WeakConnectionHandle,
+    RetryError, SendDatagramError, ServerConfig, Side, VarInt, WeakConnectionHandle,
 };
 use quinn_proto::PathId;
 use tracing::warn;
@@ -249,8 +249,7 @@ fn conn_from_quinn_conn(
     };
     let info = conn.to_info();
     // Register this connection with the magicsock.
-    ep.msock
-        .register_connection(info, paths_info_watchable.clone());
+    ep.msock.register_connection(info, paths_info_watchable);
     Ok(conn)
 }
 
@@ -1533,6 +1532,11 @@ impl Connection {
         self.inner.set_max_concurrent_bi_streams(count)
     }
 
+    /// Returns the side of the connection (client or server).
+    pub fn side(&self) -> Side {
+        self.inner.side()
+    }
+
     /// Returns a [`ConnectionInfo`], which is a weak handle to the connection
     /// that does not keep the connection alive, but does allow to access some information
     /// about the connection, and allows to wait for the connection to be closed.
@@ -1542,6 +1546,7 @@ impl Connection {
             remote_id: self.remote_id,
             inner: self.inner.weak_handle(),
             paths_info: self.paths_info.clone(),
+            side: self.side(),
         }
     }
 }
@@ -1550,6 +1555,7 @@ impl Connection {
 /// but does not keep the connection alive.
 #[derive(Debug, Clone)]
 pub struct ConnectionInfo {
+    pub(crate) side: Side,
     pub(crate) alpn: Vec<u8>,
     pub(crate) remote_id: EndpointId,
     pub(crate) inner: WeakConnectionHandle,
@@ -1584,10 +1590,10 @@ impl ConnectionInfo {
         self.paths_info.get().keys().any(|addr| addr.is_ip())
     }
 
-    /// Returns the latency for this connection.
+    /// Current best estimate of this connection's latency (round-trip-time)
     ///
     /// Returns `None` if the connection has been dropped.
-    pub fn latency(&self) -> Option<Duration> {
+    pub fn rtt(&self) -> Option<Duration> {
         self.inner.upgrade().map(|conn| conn.rtt())
     }
 
@@ -1599,10 +1605,17 @@ impl ConnectionInfo {
     }
 
     /// Waits for the connection to be closed, and returns the close reason and final connection stats.
+    ///
+    /// Returns `None` if the connection has been dropped already before this call.
     pub async fn closed(&self) -> Option<(ConnectionError, ConnectionStats)> {
         let fut = self.inner.upgrade()?.on_closed();
         Some(fut.await)
     }
+
+    /// Returns the side of the connection (client or server).
+    pub fn side(&self) -> Side {
+        self.side
+    }
 }
 
 #[cfg(test)]

iroh/src/magicsock.rs

@@ -78,7 +78,7 @@ pub(crate) mod transports;
 use mapped_addrs::{EndpointIdMappedAddr, MappedAddr};
 
 pub use self::{
-    endpoint_map::{ConnectionType, PathInfo, PathsInfo},
+    endpoint_map::{ConnectionType, PathInfo, PathsInfo, RemoteInfo},
     metrics::Metrics,
 };
 
@@ -299,9 +299,8 @@ impl MagicSock {
         // have a ZrttConnection::into_connection() function which can be async and actually
         // send this.  Before the handshake has completed we don't have anything useful to
         // do with this connection inside of the EndpointStateActor anyway.
-        let weak_handle = conn.inner.clone();
         let endpoint_state = self.endpoint_map.endpoint_state_actor(conn.remote_id);
-        let msg = EndpointStateMessage::AddConnection(weak_handle, paths_info);
+        let msg = EndpointStateMessage::AddConnection(conn, paths_info);
 
         task::spawn(async move {
             endpoint_state.send(msg).await.ok();
@@ -407,32 +406,6 @@ impl MagicSock {
         })
     }
 
-    /// Returns a [`n0_watcher::Direct`] that reports the [`ConnectionType`] we have to the
-    /// given `endpoint_id`.
-    ///
-    /// This gets us a copy of the [`n0_watcher::Direct`] for the [`Watchable`] with a
-    /// [`ConnectionType`] that the `EndpointMap` stores for each `endpoint_id`'s endpoint.
-    ///
-    /// # Errors
-    ///
-    /// Will return `None` if there is no address information known about the
-    /// given `endpoint_id`.
-    pub(crate) fn conn_type(&self, eid: EndpointId) -> Option<n0_watcher::Direct<ConnectionType>> {
-        self.endpoint_map.conn_type(eid)
-    }
-
-    // TODO: Build better info to expose to the user about remote nodes.  We probably want
-    // to expose this as part of path information instead.
-    pub(crate) async fn latency(&self, eid: EndpointId) -> Option<Duration> {
-        let endpoint_state = self.endpoint_map.endpoint_state_actor(eid);
-        let (tx, rx) = oneshot::channel();
-        endpoint_state
-            .send(EndpointStateMessage::Latency(tx))
-            .await
-            .ok();
-        rx.await.unwrap_or_default()
-    }
-
     /// Returns the socket address which can be used by the QUIC layer to dial this endpoint.
     pub(crate) fn get_endpoint_mapped_addr(&self, eid: EndpointId) -> EndpointIdMappedAddr {
         self.endpoint_map.endpoint_mapped_addr(eid)

iroh/src/magicsock/endpoint_map.rs

@@ -17,15 +17,15 @@ use super::{
     mapped_addrs::{AddrMap, EndpointIdMappedAddr, MultipathMappedAddr, RelayMappedAddr},
     transports::{self, OwnedTransmit, TransportsSender},
 };
-use crate::disco::{self};
+use crate::disco;
 // #[cfg(any(test, feature = "test-utils"))]
 // use crate::endpoint::PathSelection;
 
 mod endpoint_state;
 mod path_state;
 
 pub(super) use endpoint_state::EndpointStateMessage;
-pub use endpoint_state::{ConnectionType, PathInfo, PathsInfo};
+pub use endpoint_state::{ConnectionType, PathInfo, PathsInfo, RemoteInfo};
 use endpoint_state::{EndpointStateActor, EndpointStateHandle};
 
 // TODO: use this
@@ -122,17 +122,23 @@ impl EndpointMap {
         }
     }
 
-    /// Returns a [`n0_watcher::Direct`] for given endpoint's [`ConnectionType`].
-    ///
-    /// # Errors
-    ///
-    /// Will return `None` if there is not an entry in the [`EndpointMap`] for
-    /// the `endpoint_id`
-    pub(super) fn conn_type(
-        &self,
-        _endpoint_id: EndpointId,
-    ) -> Option<n0_watcher::Direct<ConnectionType>> {
-        todo!();
+    pub(crate) fn remote_info(&self, eid: EndpointId) -> Option<RemoteInfo> {
+        self.actor_handles
+            .lock()
+            .expect("poisoned")
+            .get(&eid)
+            .map(|handle| RemoteInfo::new(eid, handle.sender.clone(), handle.selected_path.watch()))
+    }
+
+    pub(crate) fn remotes(&self) -> Vec<RemoteInfo> {
+        self.actor_handles
+            .lock()
+            .expect("poisoned")
+            .iter()
+            .map(|(eid, handle)| {
+                RemoteInfo::new(*eid, handle.sender.clone(), handle.selected_path.watch())
+            })
+            .collect()
     }
 
     /// Returns the sender for the [`EndpointStateActor`].