refactor: remove Endpoint::conn_type (#3647)

## Description

* Remove `Endpoint::conn_type`
* Update `transfer.rs` example to use `Connection::paths` instead
* Change return type of `Connection::paths` to have more guarantees on
the return type (Send, Unpin, 'static)

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

iroh/examples/transfer.rs

@@ -10,16 +10,16 @@ use data_encoding::HEXLOWER;
 use indicatif::HumanBytes;
 use iroh::{
     Endpoint, EndpointAddr, EndpointId, RelayMap, RelayMode, RelayUrl, SecretKey, TransportAddr,
+    Watcher,
     discovery::{
         dns::DnsDiscovery,
         pkarr::{N0_DNS_PKARR_RELAY_PROD, N0_DNS_PKARR_RELAY_STAGING, PkarrPublisher},
     },
     dns::{DnsResolver, N0_DNS_ENDPOINT_ORIGIN_PROD, N0_DNS_ENDPOINT_ORIGIN_STAGING},
-    endpoint::ConnectionError,
+    endpoint::{ConnectionError, PathInfoList},
 };
 use n0_error::{Result, StackResultExt, StdResultExt};
 use n0_future::task::AbortOnDropHandle;
-use n0_watcher::Watcher as _;
 use tokio_stream::StreamExt;
 use tracing::{info, warn};
 use url::Url;
@@ -337,7 +337,6 @@ async fn provide(endpoint: Endpoint, size: u64) -> Result<()> {
             }
         };
         // spawn a task to handle reading and writing off of the connection
-        let endpoint_clone = endpoint.clone();
         tokio::spawn(async move {
             let conn = accepting.await.anyerr()?;
             let endpoint_id = conn.remote_id();
@@ -350,7 +349,7 @@ async fn provide(endpoint: Endpoint, size: u64) -> Result<()> {
             println!("[{remote}] Connected");
 
             // Spawn a background task that prints connection type changes. Will be aborted on drop.
-            let _guard = watch_conn_type(&endpoint_clone, endpoint_id);
+            let _guard = watch_conn_type(conn.remote_id(), conn.paths());
 
             // accept a bi-directional QUIC connection
             // use the `quinn` APIs to send and recv content
@@ -404,7 +403,7 @@ async fn fetch(endpoint: Endpoint, remote_addr: EndpointAddr) -> Result<()> {
     let conn = endpoint.connect(remote_addr, TRANSFER_ALPN).await?;
     println!("Connected to {}", remote_id);
     // Spawn a background task that prints connection type changes. Will be aborted on drop.
-    let _guard = watch_conn_type(&endpoint, remote_id);
+    let _guard = watch_conn_type(conn.remote_id(), conn.paths());
 
     // Use the Quinn API to send and recv content.
     let (mut send, mut recv) = conn.open_bi().await.anyerr()?;
@@ -521,14 +520,36 @@ fn parse_byte_size(s: &str) -> std::result::Result<u64, parse_size::Error> {
     cfg.parse_size(s)
 }
 
-fn watch_conn_type(endpoint: &Endpoint, endpoint_id: EndpointId) -> AbortOnDropHandle<()> {
-    let mut stream = endpoint.conn_type(endpoint_id).unwrap().stream();
+fn watch_conn_type(
+    endpoint_id: EndpointId,
+    paths_watcher: impl Watcher<Value = PathInfoList> + Send + Unpin + 'static,
+) -> AbortOnDropHandle<()> {
+    let id = endpoint_id.fmt_short();
     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()
-            );
+        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()) {
+                // We can get path updates without the selected path changing. We don't want to log again in that case.
+                if Some(path) == previous.as_ref() {
+                    continue;
+                }
+                println!(
+                    "[{id}] Connection type changed to: {:?} (RTT: {:?})",
+                    path.remote_addr(),
+                    path.rtt()
+                );
+                previous = Some(path.clone());
+            } else if !paths.is_empty() {
+                println!(
+                    "[{id}] Connection type changed to: mixed ({} paths)",
+                    paths.len()
+                );
+                previous = None;
+            } else {
+                println!("[{id}] Connection type changed to none (no active transmission paths)",);
+                previous = None;
+            }
         }
     });
     AbortOnDropHandle::new(task)

iroh/src/endpoint.rs

@@ -25,7 +25,7 @@ use tracing::{debug, instrument, trace, warn};
 use url::Url;
 
 pub use super::magicsock::{
-    AddEndpointAddrError, ConnectionType, DirectAddr, DirectAddrType, PathInfo,
+    AddEndpointAddrError, DirectAddr, DirectAddrType, PathInfo,
     endpoint_map::{PathInfoList, Source},
 };
 #[cfg(wasm_browser)]
@@ -963,33 +963,6 @@ 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.
-    ///
-    /// 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.
-    ///
-    /// 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 the currently lowest latency for this endpoint.
     ///
     /// Will return `None` if we do not have any address information for the given `endpoint_id`.

iroh/src/endpoint/connection.rs

@@ -1462,7 +1462,7 @@ impl Connection {
     ///
     /// [`PathInfo::is_selected`]: crate::magicsock::PathInfo::is_selected
     /// [`PathInfo`]: crate::magicsock::PathInfo
-    pub fn paths(&self) -> impl Watcher<Value = PathInfoList> {
+    pub fn paths(&self) -> impl Watcher<Value = PathInfoList> + Unpin + Send + Sync + 'static {
         self.paths.watch(self.inner.weak_handle())
     }
 

iroh/src/magicsock.rs

@@ -77,10 +77,7 @@ pub(crate) mod transports;
 
 use mapped_addrs::{EndpointIdMappedAddr, MappedAddr};
 
-pub use self::{
-    endpoint_map::{ConnectionType, PathInfo},
-    metrics::Metrics,
-};
+pub use self::{endpoint_map::PathInfo, metrics::Metrics};
 
 // TODO: Use this
 // /// How long we consider a QAD-derived endpoint valid for. UDP NAT mappings typically
@@ -394,20 +391,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> {

iroh/src/magicsock/endpoint_map.rs

@@ -25,8 +25,8 @@ mod path_state;
 
 pub(super) use endpoint_state::EndpointStateMessage;
 pub(crate) use endpoint_state::PathsWatchable;
-pub use endpoint_state::{ConnectionType, PathInfo, PathInfoList};
 use endpoint_state::{EndpointStateActor, EndpointStateHandle};
+pub use endpoint_state::{PathInfo, PathInfoList};
 
 // TODO: use this
 // /// Number of endpoints that are inactive for which we keep info about. This limit is enforced
@@ -107,19 +107,6 @@ impl EndpointMap {
         self.endpoint_mapped_addrs.get(&eid)
     }
 
-    /// 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!();
-    }
-
     /// Returns the sender for the [`EndpointStateActor`].
     ///
     /// If needed a new actor is started on demand.

iroh/src/magicsock/endpoint_map/endpoint_state.rs

@@ -16,7 +16,6 @@ use n0_watcher::{Watchable, Watcher};
 use quinn::{PathStats, WeakConnectionHandle};
 use quinn_proto::{PathError, PathEvent, PathId, PathStatus};
 use rustc_hash::FxHashMap;
-use serde::{Deserialize, Serialize};
 use smallvec::SmallVec;
 use tokio::sync::{mpsc, oneshot};
 use tokio_stream::wrappers::{BroadcastStream, errors::BroadcastStreamRecvError};
@@ -1055,27 +1054,6 @@ struct HolepunchAttempt {
     remote_addrs: BTreeSet<SocketAddr>,
 }
 
-/// The type of connection we have to the endpoint.
-#[derive(derive_more::Display, Default, Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
-pub enum ConnectionType {
-    /// Direct UDP connection
-    #[display("direct({_0})")]
-    Direct(SocketAddr),
-    /// Relay connection over relay
-    #[display("relay({_0})")]
-    Relay(RelayUrl),
-    /// Both a UDP and a relay connection are used.
-    ///
-    /// This is the case if we do have a UDP address, but are missing a recent confirmation that
-    /// the address works.
-    #[display("mixed(udp: {_0}, relay: {_1})")]
-    Mixed(SocketAddr, RelayUrl),
-    /// We have no verified connection to this PublicKey
-    #[default]
-    #[display("none")]
-    None,
-}
-
 /// Newtype to track Connections.
 ///
 /// The wrapped value is the [`quinn::Connection::stable_id`] value, and is thus only valid
@@ -1167,7 +1145,7 @@ impl PathsWatchable {
     pub(crate) fn watch(
         &self,
         conn_handle: WeakConnectionHandle,
-    ) -> impl Watcher<Value = PathInfoList> {
+    ) -> impl Watcher<Value = PathInfoList> + Unpin + Send + Sync + 'static {
         let joined_watcher = (self.open_paths.watch(), self.selected_path.watch());
         joined_watcher.map(move |(open_paths, selected_path)| {
             let selected_path: Option<TransportAddr> = selected_path.map(Into::into);