cleanup

Author: Frando

iroh/src/endpoint/connection.rs

@@ -29,7 +29,7 @@ use ed25519_dalek::{VerifyingKey, pkcs8::DecodePublicKey};
 use futures_util::{FutureExt, future::Shared};
 use iroh_base::{EndpointId, RelayUrl};
 use n0_error::{e, stack_error};
-use n0_future::{TryFutureExt, future::Boxed as BoxFuture, time::Duration};
+use n0_future::{Stream, StreamExt, TryFutureExt, future::Boxed as BoxFuture, time::Duration};
 use noq::WeakConnectionHandle as NoqWeakConnectionHandle;
 use pin_project::pin_project;
 use tracing::{event, warn};
@@ -1134,7 +1134,20 @@ impl Connection<HandshakeCompleted> {
     /// [`PathEvent`]: crate::endpoint::PathEvent
     /// [`PathEvent::Lagged`]: crate::endpoint::PathEvent::Lagged
     pub fn path_events(&self) -> PathEventStream {
-        self.data.paths.updates()
+        self.data.paths.events()
+    }
+
+    /// Returns a new [`PathList`] whenever it changes, starting with the current value.
+    ///
+    /// TODO(Frando): Do we want this?
+    pub fn path_updates(&self) -> impl Stream<Item = PathList<'_>> {
+        let events = self.path_events();
+        n0_future::stream::unfold((true, events), async |(is_first, mut events)| {
+            if !is_first {
+                events.next().await?;
+            }
+            Some((self.paths(), (false, events)))
+        })
     }
 
     /// Returns the side of the connection (client or server).
@@ -1318,7 +1331,7 @@ mod tests {
     use super::Endpoint;
     use crate::{
         RelayMode,
-        endpoint::{ConnectOptions, Incoming, PathEvent, PathEventStream, ZeroRttStatus, presets},
+        endpoint::{ConnectOptions, Connection, Incoming, ZeroRttStatus, presets},
         test_utils::run_relay_server,
     };
 
@@ -1562,68 +1575,42 @@ mod tests {
             async { server.accept().await.unwrap().await.unwrap() }
         );
         info!("connected");
-        // Subscribe before the initial check to avoid missing events emitted
-        // during the early exchange.
+
         let mut events_client = conn_client.path_events();
         let mut events_server = conn_server.path_events();
 
-        /// Advance the event stream until we have seen at least one IP and
-        /// one relay Opened event (current snapshot may also already satisfy
-        /// the condition, which is checked first).
-        async fn wait_for_paths(
-            events: &mut PathEventStream,
-            snapshot: impl Fn() -> (bool, bool) + Send + Sync + 'static,
-        ) {
-            let (mut has_ip, mut has_relay) = snapshot();
-            while !(has_ip && has_relay) {
-                let Some(ev) = events.next().await else {
+        /// Advances the path stream until at least one IP and one relay paths are available.
+        ///
+        /// Panics if the path stream finishes before that happens.
+        async fn wait_for_paths(conn: &Connection) {
+            let mut events = conn.path_events();
+            loop {
+                let paths = conn.paths();
+                if paths.len() >= 2
+                    && paths.iter().any(|p| p.is_ip())
+                    && paths.iter().any(|p| p.is_relay())
+                {
+                    break;
+                }
+                if events.next().await.is_none() {
                     panic!("path event stream ended before reaching target state");
                 };
-                if let PathEvent::Opened { remote_addr, .. } = ev {
-                    if remote_addr.is_ip() {
-                        has_ip = true;
-                    }
-                    if remote_addr.is_relay() {
-                        has_relay = true;
-                    }
-                }
-                // Also re-check snapshot in case the opens happened before
-                // we subscribed (unlikely here since we subscribed right
-                // after connect, but cheap).
-                let (ip2, relay2) = snapshot();
-                has_ip |= ip2;
-                has_relay |= relay2;
             }
         }
 
-        let conn_server_for_snap = conn_server.clone();
-        let conn_client_for_snap = conn_client.clone();
-        // Verify that both connections are notified of path changes and get
-        // an IP and a relay path.
+        // Verify that both connections are notified of path changes and get an IP and a relay path.
         tokio::join!(
             async {
-                tokio::time::timeout(
-                    Duration::from_secs(1),
-                    wait_for_paths(&mut events_server, move || {
-                        let p = conn_server_for_snap.paths();
-                        (p.iter().any(|p| p.is_ip()), p.iter().any(|p| p.is_relay()))
-                    }),
-                )
-                .instrument(error_span!("paths-server"))
-                .await
-                .unwrap()
+                tokio::time::timeout(Duration::from_secs(1), wait_for_paths(&conn_server))
+                    .instrument(error_span!("paths-server"))
+                    .await
+                    .unwrap()
             },
             async {
-                tokio::time::timeout(
-                    Duration::from_secs(1),
-                    wait_for_paths(&mut events_client, move || {
-                        let p = conn_client_for_snap.paths();
-                        (p.iter().any(|p| p.is_ip()), p.iter().any(|p| p.is_relay()))
-                    }),
-                )
-                .instrument(error_span!("paths-client"))
-                .await
-                .unwrap()
+                tokio::time::timeout(Duration::from_secs(1), wait_for_paths(&conn_client))
+                    .instrument(error_span!("paths-client"))
+                    .await
+                    .unwrap()
             }
         );
 

iroh/src/socket/remote_map/remote_state/path_watcher.rs

@@ -331,7 +331,7 @@ impl PathStateReceiver {
     /// The stream ends as soon as the [`PathStateSender`] is dropped. If
     /// the sender has already been dropped at the time this is called,
     /// the returned stream yields no events and terminates immediately.
-    pub(crate) fn updates(&self) -> PathEventStream {
+    pub(crate) fn events(&self) -> PathEventStream {
         let receiver = if let Some(sender) = self.events.upgrade() {
             sender.subscribe()
         } else {
@@ -583,7 +583,7 @@ mod tests {
     #[tokio::test]
     async fn event_stream_ends_when_sender_drops() {
         let (sender, receiver) = PathStateSender::new();
-        let mut stream = receiver.updates();
+        let mut stream = receiver.events();
 
         sender.record_selected(Some((pid(1), ip_addr(1))));
         drop(sender);
@@ -600,14 +600,14 @@ mod tests {
     async fn event_stream_after_sender_drop_is_immediately_empty() {
         let (sender, receiver) = PathStateSender::new();
         drop(sender);
-        let mut stream = receiver.updates();
+        let mut stream = receiver.events();
         assert!(stream.next().await.is_none());
     }
 
     #[tokio::test]
     async fn lagged_surfaces_after_overflow() {
         let (sender, receiver) = PathStateSender::new();
-        let mut stream = receiver.updates();
+        let mut stream = receiver.events();
 
         // Fill the channel beyond its capacity. Every record_selected
         // call here is a real transition because each id differs from