feat(iroh-net): Add a Watchable struct for use in the Endpoint API (#2806)

## Description

This implements a `Watchable` struct and a `Watcher` which provides
access to the watched value in several ways, including some streams.

## Breaking Changes

- `iroh::Endpoint::conn_type_stream` is renamed to
`iroh::Endpoint::conn_type` and returns
`Result<Watcher<ConnectionType>>` instead of
`Result<ConnectionTypeStream>`
To migrate, use `endpoint.conn_type()?.stream()` instead of
`endpoint.conn_type_stream()?`.
- `iroh::Endpoint::home_relay` now returns `Watcher<Option<RelayUrl>>`
instead of `Option<RelayUrl>`.
To migrate, use `endpoint.home_relay().get()?` instead of
`endpoint.home_relay()`.
- removed `iroh::Endpoint::watch_home_relay`
To migrate, use `endpoint.home_relay().initialized().await?` instead of
`endpoint.watch_home_relay().next().await` and
use `endpoint.home_relay().stream()` instead of
`endpoint.watch_home_relay().next().await`.
- removed `DirectAddrsStream` and `ConnTypeStream`.
  Use `iroh::watchable::WatcherStream` for as named types instead.

## Change checklist

- [x] Self-review.
- [x] 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.
- [x] Tests if relevant.
- [x] All breaking changes documented.

---------

Co-authored-by: Philipp Krüger <[email protected]>

Author: flub

Cargo.lock

@@ -37,7 +37,8 @@ missing_debug_implementations = "warn"
 # require a feature enabled when using `--cfg docsrs` which we can not
 # do.  To enable for a crate set `#![cfg_attr(iroh_docsrs,
 # feature(doc_auto_cfg))]` in the crate.
-unexpected_cfgs = { level = "warn", check-cfg = ["cfg(iroh_docsrs)"] }
+# We also have our own `iroh_loom` cfg to enable tokio-rs/loom testing.
+unexpected_cfgs = { level = "warn", check-cfg = ["cfg(iroh_docsrs)", "cfg(iroh_loom)"] }
 
 [workspace.lints.clippy]
 unused-async = "warn"

Cargo.toml

@@ -102,7 +102,6 @@ tokio-tungstenite-wasm = "0.3"
 tokio-util = { version = "0.7", features = ["io-util", "io", "codec", "rt"] }
 tracing = "0.1"
 url = { version = "2.5", features = ["serde"] }
-watchable = "1.1.2"
 webpki = { package = "rustls-webpki", version = "0.102" }
 webpki-roots = "0.26"
 x509-parser = "0.16"

iroh/Cargo.toml

@@ -5,7 +5,6 @@ use std::{
 
 use anyhow::{Context, Result};
 use bytes::Bytes;
-use futures_lite::StreamExt as _;
 use iroh::{
     endpoint::{Connection, ConnectionError, RecvStream, SendStream, TransportConfig},
     Endpoint, NodeAddr, RelayMap, RelayMode, RelayUrl,
@@ -45,7 +44,7 @@ pub fn server_endpoint(
             .unwrap();
 
         if relay_url.is_some() {
-            ep.watch_home_relay().next().await;
+            ep.home_relay().initialized().await.unwrap();
         }
 
         let addr = ep.bound_sockets();
@@ -101,7 +100,7 @@ pub async fn connect_client(
         .unwrap();
 
     if relay_url.is_some() {
-        endpoint.watch_home_relay().next().await;
+        endpoint.home_relay().initialized().await?;
     }
 
     // TODO: We don't support passing client transport config currently

iroh/bench/src/iroh.rs

@@ -7,9 +7,7 @@
 //! Run the `listen-unreliable` example first (`iroh/examples/listen-unreliable.rs`), which will give you instructions on how to run this example to watch two nodes connect and exchange bytes.
 use std::net::SocketAddr;
 
-use anyhow::Context;
 use clap::Parser;
-use futures_lite::StreamExt;
 use iroh::{Endpoint, NodeAddr, RelayMode, RelayUrl, SecretKey};
 use tracing::info;
 
@@ -52,21 +50,17 @@ async fn main() -> anyhow::Result<()> {
         .bind()
         .await?;
 
-    let me = endpoint.node_id();
+    let node_addr = endpoint.node_addr().await?;
+    let me = node_addr.node_id;
     println!("node id: {me}");
     println!("node listening addresses:");
-    for local_endpoint in endpoint
-        .direct_addresses()
-        .next()
-        .await
-        .context("no endpoints")?
-    {
-        println!("\t{}", local_endpoint.addr)
-    }
-
-    let relay_url = endpoint
-        .home_relay()
-        .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");
+    node_addr
+        .direct_addresses
+        .iter()
+        .for_each(|addr| println!("\t{addr}"));
+    let relay_url = node_addr
+        .relay_url
+        .expect("Should have a relay URL, assuming a default endpoint setup.");
     println!("node relay server url: {relay_url}\n");
     // Build a `NodeAddr` from the node_id, relay url, and UDP addresses.
     let addr = NodeAddr::from_parts(args.node_id, Some(args.relay_url), args.addrs);

iroh/examples/connect-unreliable.rs

@@ -9,7 +9,6 @@ use std::net::SocketAddr;
 
 use anyhow::Context;
 use clap::Parser;
-use futures_lite::StreamExt;
 use iroh::{Endpoint, NodeAddr, RelayMode, RelayUrl, SecretKey};
 use tracing::info;
 
@@ -57,15 +56,17 @@ async fn main() -> anyhow::Result<()> {
     println!("node listening addresses:");
     for local_endpoint in endpoint
         .direct_addresses()
-        .next()
+        .initialized()
         .await
-        .context("no endpoints")?
+        .context("no direct addresses")?
     {
         println!("\t{}", local_endpoint.addr)
     }
 
     let relay_url = endpoint
         .home_relay()
+        .get()
+        .unwrap()
         .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");
     println!("node relay server url: {relay_url}\n");
     // Build a `NodeAddr` from the node_id, relay url, and UDP addresses.

iroh/examples/connect.rs

@@ -3,8 +3,6 @@
 //! This example uses the default relay servers to attempt to holepunch, and will use that relay server to relay packets if the two devices cannot establish a direct UDP connection.
 //! run this example from the project root:
 //!     $ cargo run --example listen-unreliable
-use anyhow::Context;
-use futures_lite::StreamExt;
 use iroh::{Endpoint, RelayMode, SecretKey};
 use tracing::{info, warn};
 
@@ -37,23 +35,20 @@ async fn main() -> anyhow::Result<()> {
     println!("node id: {me}");
     println!("node listening addresses:");
 
-    let local_addrs = endpoint
-        .direct_addresses()
-        .next()
-        .await
-        .context("no endpoints")?
+    let node_addr = endpoint.node_addr().await?;
+    let local_addrs = node_addr
+        .direct_addresses
         .into_iter()
-        .map(|endpoint| {
-            let addr = endpoint.addr.to_string();
+        .map(|addr| {
+            let addr = addr.to_string();
             println!("\t{addr}");
             addr
         })
         .collect::<Vec<_>>()
         .join(" ");
-
-    let relay_url = endpoint
-        .home_relay()
-        .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");
+    let relay_url = node_addr
+        .relay_url
+        .expect("Should have a relay URL, assuming a default endpoint setup.");
     println!("node relay server url: {relay_url}");
     println!("\nin a separate terminal run:");
 

iroh/examples/listen-unreliable.rs

@@ -5,8 +5,6 @@
 //!     $ cargo run --example listen
 use std::time::Duration;
 
-use anyhow::Context;
-use futures_lite::StreamExt;
 use iroh::{endpoint::ConnectionError, Endpoint, RelayMode, SecretKey};
 use tracing::{debug, info, warn};
 
@@ -39,23 +37,20 @@ async fn main() -> anyhow::Result<()> {
     println!("node id: {me}");
     println!("node listening addresses:");
 
-    let local_addrs = endpoint
-        .direct_addresses()
-        .next()
-        .await
-        .context("no endpoints")?
+    let node_addr = endpoint.node_addr().await?;
+    let local_addrs = node_addr
+        .direct_addresses
         .into_iter()
-        .map(|endpoint| {
-            let addr = endpoint.addr.to_string();
+        .map(|addr| {
+            let addr = addr.to_string();
             println!("\t{addr}");
             addr
         })
         .collect::<Vec<_>>()
         .join(" ");
-
-    let relay_url = endpoint
-        .home_relay()
-        .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");
+    let relay_url = node_addr
+        .relay_url
+        .expect("Should have a relay URL, assuming a default endpoint setup.");
     println!("node relay server url: {relay_url}");
     println!("\nin a separate terminal run:");
 

iroh/examples/listen.rs

@@ -6,7 +6,6 @@ use std::{
 use anyhow::{Context, Result};
 use bytes::Bytes;
 use clap::{Parser, Subcommand};
-use futures_lite::StreamExt;
 use indicatif::HumanBytes;
 use iroh::{
     endpoint::ConnectionError, Endpoint, NodeAddr, RelayMap, RelayMode, RelayUrl, SecretKey,
@@ -71,23 +70,18 @@ async fn provide(size: u64, relay_url: Option<String>) -> anyhow::Result<()> {
 
     let node_id = endpoint.node_id();
 
-    for local_endpoint in endpoint
-        .direct_addresses()
-        .next()
-        .await
-        .context("no endpoints")?
-    {
+    for local_endpoint in endpoint.direct_addresses().initialized().await? {
         println!("\t{}", local_endpoint.addr)
     }
 
     let relay_url = endpoint
         .home_relay()
+        .get()?
         .expect("should be connected to a relay server");
     let local_addrs = endpoint
         .direct_addresses()
-        .next()
-        .await
-        .context("no endpoints")?
+        .initialized()
+        .await?
         .into_iter()
         .map(|endpoint| endpoint.addr)
         .collect::<Vec<_>>();
@@ -171,17 +165,13 @@ async fn fetch(ticket: &str, relay_url: Option<String>) -> anyhow::Result<()> {
     let me = endpoint.node_id();
     println!("node id: {me}");
     println!("node listening addresses:");
-    for local_endpoint in endpoint
-        .direct_addresses()
-        .next()
-        .await
-        .context("no endpoints")?
-    {
+    for local_endpoint in endpoint.direct_addresses().initialized().await? {
         println!("\t{}", local_endpoint.addr)
     }
 
     let relay_url = endpoint
         .home_relay()
+        .get()?
         .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");
     println!("node relay server url: {relay_url}\n");
 

iroh/examples/transfer.rs

@@ -480,6 +480,7 @@ mod tests {
             }
         }
     }
+
     #[derive(Debug)]
     struct TestDiscovery {
         node_id: NodeId,
@@ -578,7 +579,7 @@ mod tests {
             new_endpoint(secret, disco).await
         };
         let ep1_addr = NodeAddr::new(ep1.node_id());
-        // wait for out address to be updated and thus published at least once
+        // wait for our address to be updated and thus published at least once
         ep1.node_addr().await?;
         let _conn = ep2.connect(ep1_addr, TEST_ALPN).await?;
         Ok(())
@@ -833,27 +834,6 @@ mod test_dns_pkarr {
         Ok(())
     }
 
-    #[tokio::test]
-    async fn pkarr_publish_dns_discover_empty_node_addr() -> Result<()> {
-        let _logging_guard = iroh_test::logging::setup();
-
-        let dns_pkarr_server = DnsPkarrServer::run().await?;
-        let (relay_map, _relay_url, _relay_guard) = run_relay_server().await?;
-
-        let (ep1, _guard1) = ep_with_discovery(&relay_map, &dns_pkarr_server).await?;
-        let (ep2, _guard2) = ep_with_discovery(&relay_map, &dns_pkarr_server).await?;
-
-        // wait until our shared state received the update from pkarr publishing
-        dns_pkarr_server
-            .on_node(&ep1.node_id(), PUBLISH_TIMEOUT)
-            .await?;
-
-        // we connect only by node id!
-        let res = ep2.connect(ep1.node_id(), TEST_ALPN).await;
-        assert!(res.is_ok(), "connection established");
-        Ok(())
-    }
-
     async fn ep_with_discovery(
         relay_map: &RelayMap,
         dns_pkarr_server: &DnsPkarrServer,

iroh/src/discovery.rs

@@ -51,10 +51,10 @@ use tokio::{
 };
 use tokio_util::task::AbortOnDropHandle;
 use tracing::{debug, error, info_span, trace, warn, Instrument};
-use watchable::Watchable;
 
 use crate::{
     discovery::{Discovery, DiscoveryItem},
+    watchable::Watchable,
     Endpoint,
 };
 
@@ -152,8 +152,8 @@ impl LocalSwarmDiscovery {
         )?;
 
         let local_addrs: Watchable<Option<(Option<RelayUrl>, BTreeSet<SocketAddr>)>> =
-            Watchable::new(None);
-        let addrs_change = local_addrs.watch();
+            Watchable::default();
+        let mut addrs_change = local_addrs.watch();
         let discovery_fut = async move {
             let mut node_addrs: HashMap<PublicKey, Peer> = HashMap::default();
             let mut subscribers = Subscribers::new();
@@ -169,7 +169,7 @@ impl LocalSwarmDiscovery {
                     msg = recv.recv() => {
                         msg
                     }
-                    Ok(Some((_url, addrs)))= addrs_change.next_value_async() => {
+                    Ok(Some((_url, addrs))) = addrs_change.updated() => {
                         tracing::trace!(?addrs, "LocalSwarmDiscovery address changed");
                         discovery.remove_all();
                         let addrs =
@@ -379,7 +379,8 @@ impl Discovery for LocalSwarmDiscovery {
 
     fn publish(&self, url: Option<&RelayUrl>, addrs: &BTreeSet<SocketAddr>) {
         self.local_addrs
-            .replace(Some((url.cloned(), addrs.clone())));
+            .set(Some((url.cloned(), addrs.clone())))
+            .ok();
     }
 
     fn subscribe(&self) -> Option<BoxStream<DiscoveryItem>> {