refactor(multipath): Store Watcher, not Watchable on Connection

Cargo.toml

@@ -46,6 +46,7 @@ unused-async = "warn"
 iroh-quinn = { git = "https://github.com/n0-computer/quinn", branch = "main-iroh" }
 iroh-quinn-proto = { git = "https://github.com/n0-computer/quinn", branch = "main-iroh" }
 iroh-quinn-udp = { git = "https://github.com/n0-computer/quinn", branch = "main-iroh" }
+n0-watcher = { git = "https://github.com/n0-computer/n0-watcher", branch = "Frando/lazy-direct" }
 
 # iroh-quinn = { path = "../iroh-quinn/quinn" }
 # iroh-quinn-proto = { path = "../iroh-quinn/quinn-proto" }

iroh/src/endpoint.rs

@@ -26,7 +26,7 @@ use url::Url;
 
 pub use super::magicsock::{
     AddEndpointAddrError, ConnectionType, DirectAddr, DirectAddrType, PathInfo,
-    endpoint_map::{PathInfoList, Source},
+    endpoint_map::{PathInfoIter, Source},
 };
 #[cfg(wasm_browser)]
 use crate::discovery::pkarr::PkarrResolver;
@@ -1804,9 +1804,9 @@ mod tests {
             send.write_all(b"hello").await.anyerr()?;
             let mut paths = conn.paths().stream();
             info!("Waiting for direct connection");
-            while let Some(infos) = paths.next().await {
+            while let Some(mut infos) = paths.next().await {
                 info!(?infos, "new PathInfos");
-                if infos.iter().any(|info| info.is_ip()) {
+                if infos.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().get();
+            let mut path_info = conn.paths().get();
             assert_eq!(path_info.len(), 1);
-            assert!(path_info.iter().next().unwrap().is_ip());
+            assert!(path_info.next().unwrap().is_ip());
 
             let mut paths = conn.paths().stream();
             time::timeout(Duration::from_secs(5), async move {
-                while let Some(infos) = paths.next().await {
+                while let Some(mut infos) = paths.next().await {
                     info!(?infos, "new PathInfos");
-                    if infos.iter().any(|info| info.is_relay()) {
+                    if infos.any(|info| info.is_relay()) {
                         break;
                     }
                 }
@@ -1944,9 +1944,9 @@ mod tests {
             // being added on this side too.
             let mut paths = conn.paths().stream();
             time::timeout(Duration::from_secs(5), async move {
-                while let Some(infos) = paths.next().await {
+                while let Some(mut infos) = paths.next().await {
                     info!(?infos, "new PathInfos");
-                    if infos.iter().any(|path| path.is_relay()) {
+                    if infos.any(|path| path.is_relay()) {
                         break;
                     }
                 }

iroh/src/endpoint/connection.rs

@@ -43,7 +43,7 @@ use crate::{
     discovery::DiscoveryTask,
     magicsock::{
         EndpointStateActorStoppedError,
-        endpoint_map::{PathInfoList, PathsWatchable},
+        endpoint_map::{PathInfoIter, PathsWatcher},
     },
 };
 
@@ -1213,7 +1213,7 @@ pub struct Connection {
     inner: quinn::Connection,
     remote_id: EndpointId,
     alpn: Vec<u8>,
-    paths: PathsWatchable,
+    paths: PathsWatcher,
 }
 
 #[allow(missing_docs)]
@@ -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 = PathInfoIter> {
         self.paths.watch(self.inner.weak_handle())
     }
 

iroh/src/magicsock.rs

@@ -64,7 +64,7 @@ use crate::{
     disco::{self, SendAddr},
     discovery::{ConcurrentDiscovery, Discovery, EndpointData, UserData},
     key::{DecryptionError, SharedSecret, public_ed_box, secret_ed_box},
-    magicsock::endpoint_map::PathsWatchable,
+    magicsock::endpoint_map::PathsWatcher,
     metrics::EndpointMetrics,
     net_report::{self, IfStateDetails, Report},
 };
@@ -275,14 +275,14 @@ impl MagicSock {
     /// The actor is responsible for holepunching and opening additional paths to this
     /// connection.
     ///
-    /// Returns a future that resolves to [`PathsWatchable`].
+    /// Returns a future that resolves to [`PathsWatcher`].
     ///
     /// The returned future is `'static`, so it can be stored without being liftetime-bound to `&self`.
     pub(crate) fn register_connection(
         &self,
         remote: EndpointId,
         conn: WeakConnectionHandle,
-    ) -> impl Future<Output = Result<PathsWatchable, EndpointStateActorStoppedError>> + Send + 'static
+    ) -> impl Future<Output = Result<PathsWatcher, EndpointStateActorStoppedError>> + Send + 'static
     {
         let (tx, rx) = oneshot::channel();
         let sender = self.endpoint_map.endpoint_state_actor(remote);
@@ -1985,7 +1985,7 @@ mod tests {
         info!("stats: {:#?}", stats);
         // TODO: ensure panics in this function are reported ok
         if matches!(loss, ExpectedLoss::AlmostNone) {
-            for info in conn.paths().get().iter() {
+            for info in conn.paths().get() {
                 assert!(
                     info.stats().lost_packets < 10,
                     "[receiver] path {:?} should not loose many packets",

iroh/src/magicsock/endpoint_map.rs

@@ -25,8 +25,8 @@ mod endpoint_state;
 mod path_state;
 
 pub(super) use endpoint_state::EndpointStateMessage;
-pub(crate) use endpoint_state::PathsWatchable;
-pub use endpoint_state::{ConnectionType, PathInfo, PathInfoList};
+pub(crate) use endpoint_state::PathsWatcher;
+pub use endpoint_state::{ConnectionType, PathInfo, PathInfoIter};
 use endpoint_state::{EndpointStateActor, EndpointStateHandle};
 
 // TODO: use this