chore: upgrade to latest iroh main, and add n0-watcher dependency

Author: “ramfox”

Cargo.lock

@@ -11,6 +11,7 @@ derive_more = { version = "2.0.1", features = ["from", "try_from", "into", "debu
 futures-lite = "2.6.0"
 quinn = { package = "iroh-quinn", version = "0.13.0" }
 n0-future = "0.1.2"
+n0-watcher = "0.2.0"
 range-collections = { version = "0.4.6", features = ["serde"] }
 redb = "2.4.0"
 smallvec = { version = "1", features = ["serde", "const_new"] }
@@ -28,10 +29,10 @@ chrono = "0.4.39"
 nested_enum_utils = "0.1.0"
 ref-cast = "1.0.24"
 arrayvec = "0.7.6"
-iroh = "0.35.0"
+iroh = "0.35"
 self_cell = "1.1.0"
 genawaiter = { version = "0.99.1", features = ["futures03"] }
-iroh-base = "0.35.0"
+iroh-base = "0.35"
 reflink-copy = "0.1.24"
 irpc = { version = "0.3.0", features = ["rpc", "quinn_endpoint_setup", "message_spans", "stream"], default-features = false }
 irpc-derive = { version = "0.3.0", default-features = false }
@@ -57,3 +58,7 @@ walkdir = "2.5.0"
 hide-proto-docs = []
 metrics = []
 default = ["hide-proto-docs"]
+
+[patch.crates-io]
+iroh = { git = "https://github.com/n0-computer/iroh.git", branch = "main" }
+iroh-base = { git = "https://github.com/n0-computer/iroh.git", branch = "main" }

Cargo.toml

@@ -430,7 +430,7 @@ impl Blobs {
         &self,
         hash: Hash,
         ranges: ChunkRanges,
-        stream: &mut quinn::RecvStream,
+        stream: &mut iroh::endpoint::RecvStream,
     ) -> RequestResult<()> {
         let reader = TokioStreamReader::new(stream);
         self.import_bao_reader(hash, ranges, reader).await?;

src/api/blobs.rs

@@ -630,7 +630,7 @@ impl DialNode {
             Ok(Err(e)) => {
                 warn!("Failed to connect to node {}: {}", self.id, e);
                 *guard = SlotState::AttemptFailed(SystemTime::now());
-                Err(e)
+                Err(e.into())
             }
             Err(e) => {
                 warn!("Failed to connect to node {}: {}", self.id, e);
@@ -690,6 +690,7 @@ mod tests {
 
     use bao_tree::ChunkRanges;
     use n0_future::StreamExt;
+    use n0_watcher::Watcher;
     use testresult::TestResult;
 
     use crate::{
@@ -710,9 +711,9 @@ mod tests {
         let (r3, store3, _) = node_test_setup_fs(testdir.path().join("c")).await?;
         let tt1 = store1.add_slice("hello world").await?;
         let tt2 = store2.add_slice("hello world 2").await?;
-        let node1_addr = r1.endpoint().node_addr().await?;
+        let node1_addr = r1.endpoint().node_addr().initialized().await?;
         let node1_id = node1_addr.node_id;
-        let node2_addr = r2.endpoint().node_addr().await?;
+        let node2_addr = r2.endpoint().node_addr().initialized().await?;
         let node2_id = node2_addr.node_id;
         let swarm = Downloader::new(&store3, r3.endpoint());
         r3.endpoint().add_node_addr(node1_addr.clone())?;
@@ -749,9 +750,9 @@ mod tests {
                 format: crate::BlobFormat::HashSeq,
             })
             .await?;
-        let node1_addr = r1.endpoint().node_addr().await?;
+        let node1_addr = r1.endpoint().node_addr().initialized().await?;
         let node1_id = node1_addr.node_id;
-        let node2_addr = r2.endpoint().node_addr().await?;
+        let node2_addr = r2.endpoint().node_addr().initialized().await?;
         let node2_id = node2_addr.node_id;
         let swarm = Downloader::new(&store3, r3.endpoint());
         r3.endpoint().add_node_addr(node1_addr.clone())?;
@@ -818,9 +819,9 @@ mod tests {
                 format: crate::BlobFormat::HashSeq,
             })
             .await?;
-        let node1_addr = r1.endpoint().node_addr().await?;
+        let node1_addr = r1.endpoint().node_addr().initialized().await?;
         let node1_id = node1_addr.node_id;
-        let node2_addr = r2.endpoint().node_addr().await?;
+        let node2_addr = r2.endpoint().node_addr().initialized().await?;
         let node2_id = node2_addr.node_id;
         let swarm = Downloader::new(&store3, r3.endpoint());
         r3.endpoint().add_node_addr(node1_addr.clone())?;

src/api/downloader.rs

@@ -2,9 +2,9 @@
 //!
 //! The entry point is the [`Download`] struct.
 use genawaiter::sync::{Co, Gen};
+use iroh::endpoint::SendStream;
 use irpc::util::{AsyncReadVarintExt, WriteVarintExt};
 use n0_future::{Stream, StreamExt, io};
-use quinn::SendStream;
 use ref_cast::RefCast;
 
 use super::blobs::Bitfield;

src/api/remote.rs

@@ -253,7 +253,7 @@ pub mod fsm {
         RequestTooBig,
         /// Error when writing the request to the [`SendStream`].
         #[error("write: {0}")]
-        Write(#[from] quinn::WriteError),
+        Write(#[from] iroh::endpoint::WriteError),
         /// Quic connection is closed.
         #[error("closed")]
         Closed(#[from] quinn::ClosedStream),

src/get.rs

@@ -38,9 +38,12 @@
 
 use std::{fmt::Debug, sync::Arc};
 
-use anyhow::Result;
-use iroh::{Endpoint, endpoint::Connection, protocol::ProtocolHandler};
-use n0_future::future::Boxed as BoxedFuture;
+use iroh::{
+    Endpoint,
+    endpoint::Connection,
+    protocol::{AcceptError, ProtocolHandler},
+};
+use n0_watcher::Watcher;
 use tokio::sync::mpsc;
 use tracing::error;
 
@@ -89,14 +92,17 @@ impl Blobs {
     /// just a convenience method to create a ticket from content and the address of this node.
     pub async fn ticket(&self, content: impl Into<HashAndFormat>) -> anyhow::Result<BlobTicket> {
         let content = content.into();
-        let addr = self.inner.endpoint.node_addr().await?;
+        let addr = self.inner.endpoint.node_addr().initialized().await?;
         let ticket = BlobTicket::new(addr, content.hash, content.format);
         Ok(ticket)
     }
 }
 
 impl ProtocolHandler for Blobs {
-    fn accept(&self, conn: Connection) -> BoxedFuture<Result<()>> {
+    fn accept(
+        &self,
+        conn: Connection,
+    ) -> impl Future<Output = std::result::Result<(), AcceptError>> + Send {
         let store = self.store().clone();
         let events = self.inner.events.clone();
 
@@ -106,7 +112,7 @@ impl ProtocolHandler for Blobs {
         })
     }
 
-    fn shutdown(&self) -> BoxedFuture<()> {
+    fn shutdown(&self) -> impl Future<Output = ()> + Send {
         let store = self.store().clone();
         Box::pin(async move {
             if let Err(cause) = store.shutdown().await {

src/net_protocol.rs

@@ -437,7 +437,7 @@ pub enum RequestType {
 
 impl Request {
     pub async fn read_async(
-        reader: &mut CountingReader<&mut quinn::RecvStream>,
+        reader: &mut CountingReader<&mut iroh::endpoint::RecvStream>,
     ) -> io::Result<Self> {
         let request_type = reader.read_u8().await?;
         let request_type: RequestType = postcard::from_bytes(std::slice::from_ref(&request_type))

src/protocol.rs

@@ -732,7 +732,7 @@ impl<R> CountingReader<R> {
     }
 }
 
-impl CountingReader<&mut quinn::RecvStream> {
+impl CountingReader<&mut iroh::endpoint::RecvStream> {
     pub async fn read_to_end_as<T: DeserializeOwned>(&mut self, max_size: usize) -> io::Result<T> {
         let data = self
             .inner

src/provider.rs

@@ -5,6 +5,7 @@ use bytes::Bytes;
 use iroh::{Endpoint, NodeId, protocol::Router};
 use irpc::RpcMessage;
 use n0_future::{StreamExt, task::AbortOnDropHandle};
+use n0_watcher::Watcher;
 use tempfile::TempDir;
 use testresult::TestResult;
 use tokio::sync::{mpsc, watch};
@@ -226,7 +227,7 @@ async fn two_nodes_get_blobs(
     for size in sizes {
         tts.push(store1.add_bytes(test_data(size)).await?);
     }
-    let addr1 = r1.endpoint().node_addr().await?;
+    let addr1 = r1.endpoint().node_addr().initialized().await?;
     let conn = r2.endpoint().connect(addr1, crate::ALPN).await?;
     for size in sizes {
         let hash = Hash::new(test_data(size));
@@ -259,7 +260,7 @@ async fn two_nodes_observe(
     let size = 1024 * 1024 * 8 + 1;
     let data = test_data(size);
     let (hash, bao) = create_n0_bao(&data, &ChunkRanges::all())?;
-    let addr1 = r1.endpoint().node_addr().await?;
+    let addr1 = r1.endpoint().node_addr().initialized().await?;
     let conn = r2.endpoint().connect(addr1, crate::ALPN).await?;
     let mut stream = store2
         .remote()
@@ -308,7 +309,7 @@ async fn two_nodes_get_many(
         tts.push(store1.add_bytes(test_data(size)).await?);
     }
     let hashes = tts.iter().map(|tt| tt.hash).collect::<Vec<_>>();
-    let addr1 = r1.endpoint().node_addr().await?;
+    let addr1 = r1.endpoint().node_addr().initialized().await?;
     let conn = r2.endpoint().connect(addr1, crate::ALPN).await?;
     store2
         .remote()
@@ -381,7 +382,7 @@ async fn two_nodes_push_blobs(
     for size in sizes {
         tts.push(store1.add_bytes(test_data(size)).await?);
     }
-    let addr2 = r2.endpoint().node_addr().await?;
+    let addr2 = r2.endpoint().node_addr().initialized().await?;
     let conn = r1.endpoint().connect(addr2, crate::ALPN).await?;
     for size in sizes {
         let hash = Hash::new(test_data(size));
@@ -542,7 +543,7 @@ async fn two_nodes_hash_seq(
     r2: Router,
     store2: &Store,
 ) -> TestResult<()> {
-    let addr1 = r1.endpoint().node_addr().await?;
+    let addr1 = r1.endpoint().node_addr().initialized().await?;
     let sizes = INTERESTING_SIZES;
     let root = add_test_hash_seq(store1, sizes).await?;
     let conn = r2.endpoint().connect(addr1, crate::ALPN).await?;
@@ -569,7 +570,7 @@ async fn two_nodes_hash_seq_mem() -> TestResult<()> {
 async fn two_nodes_hash_seq_progress() -> TestResult<()> {
     tracing_subscriber::fmt::try_init().ok();
     let (_testdir, (r1, store1, _), (r2, store2, _)) = two_node_test_setup_fs().await?;
-    let addr1 = r1.endpoint().node_addr().await?;
+    let addr1 = r1.endpoint().node_addr().initialized().await?;
     let sizes = INTERESTING_SIZES;
     let root = add_test_hash_seq(&store1, sizes).await?;
     let conn = r2.endpoint().connect(addr1, crate::ALPN).await?;
@@ -605,7 +606,7 @@ async fn node_serve_hash_seq() -> TestResult<()> {
     let r1 = Router::builder(endpoint)
         .accept(crate::protocol::ALPN, blobs)
         .spawn();
-    let addr1 = r1.endpoint().node_addr().await?;
+    let addr1 = r1.endpoint().node_addr().initialized().await?;
     info!("node addr: {addr1:?}");
     let endpoint2 = Endpoint::builder().discovery_n0().bind().await?;
     let conn = endpoint2.connect(addr1, crate::protocol::ALPN).await?;
@@ -636,7 +637,7 @@ async fn node_serve_blobs() -> TestResult<()> {
     let r1 = Router::builder(endpoint)
         .accept(crate::protocol::ALPN, blobs)
         .spawn();
-    let addr1 = r1.endpoint().node_addr().await?;
+    let addr1 = r1.endpoint().node_addr().initialized().await?;
     info!("node addr: {addr1:?}");
     let endpoint2 = Endpoint::builder().discovery_n0().bind().await?;
     let conn = endpoint2.connect(addr1, crate::protocol::ALPN).await?;
@@ -678,7 +679,7 @@ async fn node_smoke(store: &Store) -> TestResult<()> {
     let r1 = Router::builder(endpoint)
         .accept(crate::protocol::ALPN, blobs)
         .spawn();
-    let addr1 = r1.endpoint().node_addr().await?;
+    let addr1 = r1.endpoint().node_addr().initialized().await?;
     info!("node addr: {addr1:?}");
     let endpoint2 = Endpoint::builder().discovery_n0().bind().await?;
     let conn = endpoint2.connect(addr1, crate::protocol::ALPN).await?;

src/tests.rs

@@ -79,8 +79,9 @@ impl Ticket for BlobTicket {
         postcard::to_stdvec(&data).expect("postcard serialization failed")
     }
 
-    fn from_bytes(bytes: &[u8]) -> std::result::Result<Self, ticket::Error> {
-        let res: TicketWireFormat = postcard::from_bytes(bytes).map_err(ticket::Error::Postcard)?;
+    fn from_bytes(bytes: &[u8]) -> std::result::Result<Self, ticket::ParseError> {
+        let res: TicketWireFormat =
+            postcard::from_bytes(bytes).map_err(ticket::ParseError::Postcard)?;
         let TicketWireFormat::Variant0(Variant0BlobTicket { node, format, hash }) = res;
         Ok(Self {
             node: NodeAddr {
@@ -95,7 +96,7 @@ impl Ticket for BlobTicket {
 }
 
 impl FromStr for BlobTicket {
-    type Err = ticket::Error;
+    type Err = ticket::ParseError;
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         Ticket::deserialize(s)