chore(iroh): Update n0-watcher

iroh/Cargo.toml

@@ -41,9 +41,9 @@ iroh-base = { version = "0.90.0", default-features = false, features = ["key", "
 iroh-relay = { version = "0.90", path = "../iroh-relay", default-features = false }
 n0-future = "0.1.2"
 n0-snafu = "0.2.1"
-n0-watcher = "0.2"
+n0-watcher = "0.3"
 nested_enum_utils = "0.2.1"
-netwatch = { version = "0.7" }
+netwatch = { version = "0.8" }
 pin-project = "1"
 pkarr = { version = "3.7", default-features = false, features = [
     "relays",

iroh/bench/Cargo.toml

@@ -12,7 +12,6 @@ iroh = { path = ".." }
 iroh-metrics = "0.35"
 n0-future = "0.1.1"
 n0-snafu = "0.2.0"
-n0-watcher = "0.2"
 quinn = { package = "iroh-quinn", version = "0.14" }
 rand = "0.8"
 rcgen = "0.14"

iroh/bench/src/iroh.rs

@@ -5,11 +5,10 @@ use std::{
 
 use bytes::Bytes;
 use iroh::{
-    Endpoint, NodeAddr, RelayMode, RelayUrl,
+    Endpoint, NodeAddr, RelayMode, RelayUrl, Watcher as _,
     endpoint::{Connection, ConnectionError, RecvStream, SendStream, TransportConfig},
 };
 use n0_snafu::{Result, ResultExt};
-use n0_watcher::Watcher as _;
 use tracing::{trace, warn};
 
 use crate::{
@@ -50,7 +49,7 @@ pub fn server_endpoint(
             .unwrap();
 
         if relay_url.is_some() {
-            ep.home_relay().initialized().await.unwrap();
+            ep.home_relay().initialized().await;
         }
 
         let addr = ep.bound_sockets();
@@ -111,7 +110,7 @@ pub async fn connect_client(
         .unwrap();
 
     if relay_url.is_some() {
-        endpoint.home_relay().initialized().await?;
+        endpoint.home_relay().initialized().await;
     }
 
     // TODO: We don't support passing client transport config currently

iroh/examples/connect-unreliable.rs

@@ -52,7 +52,7 @@ async fn main() -> n0_snafu::Result<()> {
         .bind()
         .await?;
 
-    let node_addr = endpoint.node_addr().initialized().await?;
+    let node_addr = endpoint.node_addr().initialized().await;
     let me = node_addr.node_id;
     println!("node id: {me}");
     println!("node listening addresses:");

iroh/examples/connect.rs

@@ -55,19 +55,13 @@ async fn main() -> Result<()> {
     let me = endpoint.node_id();
     println!("node id: {me}");
     println!("node listening addresses:");
-    for local_endpoint in endpoint
-        .direct_addresses()
-        .initialized()
-        .await
-        .context("no direct addresses")?
-    {
+    for local_endpoint in endpoint.direct_addresses().initialized().await {
         println!("\t{}", local_endpoint.addr)
     }
 
     let relay_url = endpoint
         .home_relay()
         .get()
-        .unwrap()
         .first()
         .cloned()
         .expect("should be connected to a relay server, try calling `endpoint.local_endpoints()` or `endpoint.connect()` first, to ensure the endpoint has actually attempted a connection before checking for the connected relay server");

iroh/examples/echo-no-router.rs

@@ -20,7 +20,7 @@ const ALPN: &[u8] = b"iroh-example/echo/0";
 #[tokio::main]
 async fn main() -> Result<()> {
     let endpoint = start_accept_side().await?;
-    let node_addr = endpoint.node_addr().initialized().await?;
+    let node_addr = endpoint.node_addr().initialized().await;
 
     connect_side(node_addr).await?;
 

iroh/examples/echo.rs

@@ -23,7 +23,7 @@ const ALPN: &[u8] = b"iroh-example/echo/0";
 #[tokio::main]
 async fn main() -> Result<()> {
     let router = start_accept_side().await?;
-    let node_addr = router.endpoint().node_addr().initialized().await?;
+    let node_addr = router.endpoint().node_addr().initialized().await;
 
     connect_side(node_addr).await?;
 

iroh/examples/listen-unreliable.rs

@@ -37,7 +37,7 @@ async fn main() -> Result<()> {
     println!("node id: {me}");
     println!("node listening addresses:");
 
-    let node_addr = endpoint.node_addr().initialized().await?;
+    let node_addr = endpoint.node_addr().initialized().await;
     let local_addrs = node_addr
         .direct_addresses
         .into_iter()

iroh/examples/listen.rs

@@ -42,7 +42,7 @@ async fn main() -> n0_snafu::Result<()> {
     let local_addrs = endpoint
         .direct_addresses()
         .initialized()
-        .await?
+        .await
         .into_iter()
         .map(|addr| {
             let addr = addr.addr.to_string();
@@ -51,7 +51,7 @@ async fn main() -> n0_snafu::Result<()> {
         })
         .collect::<Vec<_>>()
         .join(" ");
-    let relay_url = endpoint.home_relay().initialized().await?;
+    let relay_url = endpoint.home_relay().initialized().await;
     println!("node relay server url: {relay_url}");
     println!("\nin a separate terminal run:");
 

iroh/examples/transfer.rs

@@ -244,22 +244,18 @@ impl EndpointArgs {
         let node_id = endpoint.node_id();
         println!("Our node id:\n\t{node_id}");
 
-        let eps = endpoint.direct_addresses().initialized().await?;
+        let eps = endpoint.direct_addresses().initialized().await;
         println!("Our direct addresses:");
         for local_endpoint in eps {
             println!("\t{} (type: {:?})", local_endpoint.addr, local_endpoint.typ)
         }
 
         if self.relay_only {
-            let relay_url = endpoint.home_relay().initialized().await?;
+            let relay_url = endpoint.home_relay().initialized().await;
             println!("Our home relay server:\t{relay_url}");
         } else if !self.no_relay {
             let relay_url = tokio::time::timeout(Duration::from_secs(2), async {
-                endpoint
-                    .home_relay()
-                    .initialized()
-                    .await
-                    .expect("disconnected")
+                endpoint.home_relay().initialized().await
             })
             .await
             .ok();
@@ -278,11 +274,11 @@ impl EndpointArgs {
 async fn provide(endpoint: Endpoint, size: u64) -> Result<()> {
     let node_id = endpoint.node_id();
 
-    let node_addr = endpoint.node_addr().initialized().await?;
+    let node_addr = endpoint.node_addr().initialized().await;
     let ticket = NodeTicket::new(node_addr);
     println!("Ticket with our home relay and direct addresses:\n{ticket}\n",);
 
-    let mut node_addr = endpoint.node_addr().initialized().await?;
+    let mut node_addr = endpoint.node_addr().initialized().await;
     node_addr.direct_addresses = Default::default();
     let ticket = NodeTicket::new(node_addr);
     println!("Ticket with our home relay but no direct addresses:\n{ticket}\n",);

iroh/src/discovery.rs

@@ -872,7 +872,7 @@ mod tests {
         };
         let ep1_addr = NodeAddr::new(ep1.node_id());
         // wait for our address to be updated and thus published at least once
-        ep1.node_addr().initialized().await?;
+        ep1.node_addr().initialized().await;
         let _conn = ep2.connect(ep1_addr, TEST_ALPN).await?;
         Ok(())
     }
@@ -898,10 +898,7 @@ mod tests {
         };
         let ep1_addr = NodeAddr::new(ep1.node_id());
         // wait for out address to be updated and thus published at least once
-        ep1.node_addr()
-            .initialized()
-            .await
-            .context("waiting for NodeAddr")?;
+        ep1.node_addr().initialized().await;
         let _conn = ep2
             .connect(ep1_addr, TEST_ALPN)
             .await
@@ -933,7 +930,7 @@ mod tests {
             new_endpoint(secret, disco).await
         };
         // wait for out address to be updated and thus published at least once
-        ep1.node_addr().initialized().await?;
+        ep1.node_addr().initialized().await;
         let _conn = ep2.connect(ep1.node_id(), TEST_ALPN).await?;
         Ok(())
     }
@@ -955,7 +952,7 @@ mod tests {
             new_endpoint(secret, disco).await
         };
         // wait for out address to be updated and thus published at least once
-        ep1.node_addr().initialized().await?;
+        ep1.node_addr().initialized().await;
 
         // 10x faster test via a 3s idle timeout instead of the 30s default
         let mut config = TransportConfig::default();
@@ -988,7 +985,7 @@ mod tests {
             new_endpoint(secret, disco).await
         };
         // wait for out address to be updated and thus published at least once
-        ep1.node_addr().initialized().await?;
+        ep1.node_addr().initialized().await;
         let ep1_wrong_addr = NodeAddr {
             node_id: ep1.node_id(),
             relay_url: None,
@@ -1016,7 +1013,7 @@ mod tests {
         let mut stream = ep1.discovery_stream();
 
         // wait for ep2 node addr to be updated and connect from ep1 -> discovery via resolve
-        ep2.node_addr().initialized().await?;
+        ep2.node_addr().initialized().await;
         let _ = ep1.connect(ep2.node_id(), TEST_ALPN).await?;
 
         let item = tokio::time::timeout(Duration::from_secs(1), stream.next())

iroh/src/discovery/pkarr.rs

@@ -349,10 +349,10 @@ impl PublisherService {
         let republish = time::sleep(Duration::MAX);
         tokio::pin!(republish);
         loop {
-            let Ok(info) = self.watcher.get() else {
-                break; // disconnected
-            };
-            if let Some(info) = info {
+            if !self.watcher.is_connected() {
+                break;
+            }
+            if let Some(info) = self.watcher.get() {
                 match self.publish_current(info).await {
                     Err(err) => {
                         failed_attempts += 1;