use latest irpc, iroh-quinn, and iroh

Author: “ramfox”

Cargo.lock

@@ -9,11 +9,10 @@ bao-tree = { version = "0.15.1", features = ["experimental-mixed", "tokio_fsm",
 bytes = { version = "1", features = ["serde"] }
 derive_more = { version = "2.0.1", features = ["from", "try_from", "into", "debug", "display", "deref", "deref_mut"] }
 futures-lite = "2.6.0"
-quinn = { package = "iroh-quinn", version = "0.13.0" }
+quinn = { package = "iroh-quinn", version = "0.14.0" }
 n0-future = "0.1.2"
 n0-snafu = "0.2.0"
 n0-watcher = "0.2.0"
-nested_enum_utils = "0.2.1"
 range-collections = { version = "0.4.6", features = ["serde"] }
 redb = "2.4.0"
 smallvec = { version = "1", features = ["serde", "const_new"] }
@@ -28,19 +27,20 @@ serde = "1.0.217"
 postcard = { version = "1.1.1", features = ["experimental-derive", "use-std"] }
 data-encoding = "2.8.0"
 chrono = "0.4.39"
-nested_enum_utils = "0.1.0"
+nested_enum_utils = "0.2.1"
 ref-cast = "1.0.24"
 arrayvec = "0.7.6"
 iroh = "0.35"
 self_cell = "1.1.0"
 genawaiter = { version = "0.99.1", features = ["futures03"] }
 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 }
+irpc = { version = "0.4.0", features = ["rpc", "quinn_endpoint_setup", "message_spans", "stream"], default-features = false }
+irpc-derive = { version = "0.4.0", default-features = false }
 iroh-metrics = { version = "0.32.0" }
 hashlink = "0.10.0"
 futures-buffered = "0.2.11"
+thiserror = "2.0.12"
 
 [dev-dependencies]
 clap = { version = "4.5.31", features = ["derive"] }
@@ -64,3 +64,5 @@ 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" }
+irpc = { git = "https://github.com/n0-computer/irpc.git", branch = "iroh-quinn-latest" }
+irpc-derive = { git = "https://github.com/n0-computer/irpc.git", branch = "iroh-quinn-latest" }

Cargo.toml

@@ -24,7 +24,7 @@ use bao_tree::{
 use bytes::Bytes;
 use genawaiter::sync::Gen;
 use iroh_io::{AsyncStreamReader, TokioStreamReader};
-use irpc::channel::{oneshot, spsc};
+use irpc::channel::{mpsc, oneshot};
 use n0_future::{Stream, StreamExt, future, stream};
 use quinn::SendStream;
 use range_collections::{RangeSet2, range_set::RangeSetRange};
@@ -225,7 +225,7 @@ impl Blobs {
         };
         let client = self.client.clone();
         let stream = Gen::new(|co| async move {
-            let (mut sender, mut receiver) = match client.bidi_streaming(inner, 32, 32).await {
+            let (sender, mut receiver) = match client.bidi_streaming(inner, 32, 32).await {
                 Ok(x) => x,
                 Err(cause) => {
                     co.yield_(AddProgressItem::Error(cause.into())).await;
@@ -338,7 +338,7 @@ impl Blobs {
         trace!("{:?}", options);
         if options.hash == Hash::EMPTY {
             return ObserveProgress::new(async move {
-                let (mut tx, rx) = spsc::channel(1);
+                let (tx, rx) = mpsc::channel(1);
                 tx.send(Bitfield::complete(0)).await.ok();
                 Ok(rx)
             });
@@ -405,7 +405,7 @@ impl Blobs {
         let tree = BaoTree::new(size.get(), IROH_BLOCK_SIZE);
         let mut decoder = ResponseDecoder::new(hash.into(), ranges, tree, reader);
         let options = ImportBaoOptions { hash, size };
-        let mut handle = self.import_bao_with_opts(options, 32).await?;
+        let handle = self.import_bao_with_opts(options, 32).await?;
         let driver = async move {
             let reader = loop {
                 match decoder.next().await {
@@ -509,7 +509,7 @@ impl<'a> BatchAddProgress<'a> {
 pub struct Batch<'a> {
     scope: Scope,
     blobs: &'a Blobs,
-    _tx: spsc::Sender<BatchResponse>,
+    _tx: mpsc::Sender<BatchResponse>,
 }
 
 impl<'a> Batch<'a> {
@@ -647,7 +647,7 @@ impl<'a> AddProgress<'a> {
 /// Calling [`ObserveProgress::aggregated`] will return a stream of states,
 /// where each state is the current state at the time of the update.
 pub struct ObserveProgress {
-    inner: future::Boxed<irpc::Result<spsc::Receiver<Bitfield>>>,
+    inner: future::Boxed<irpc::Result<mpsc::Receiver<Bitfield>>>,
 }
 
 impl IntoFuture for ObserveProgress {
@@ -668,7 +668,7 @@ impl IntoFuture for ObserveProgress {
 
 impl ObserveProgress {
     fn new(
-        fut: impl Future<Output = irpc::Result<spsc::Receiver<Bitfield>>> + Send + 'static,
+        fut: impl Future<Output = irpc::Result<mpsc::Receiver<Bitfield>>> + Send + 'static,
     ) -> Self {
         Self {
             inner: Box::pin(fut),
@@ -710,7 +710,7 @@ impl ObserveProgress {
 /// It also implements [`IntoFuture`], so you can await it to get the size of the
 /// exported blob.
 pub struct ExportProgress {
-    inner: future::Boxed<irpc::Result<spsc::Receiver<ExportProgressItem>>>,
+    inner: future::Boxed<irpc::Result<mpsc::Receiver<ExportProgressItem>>>,
 }
 
 impl IntoFuture for ExportProgress {
@@ -725,7 +725,7 @@ impl IntoFuture for ExportProgress {
 
 impl ExportProgress {
     fn new(
-        fut: impl Future<Output = irpc::Result<spsc::Receiver<ExportProgressItem>>> + Send + 'static,
+        fut: impl Future<Output = irpc::Result<mpsc::Receiver<ExportProgressItem>>> + Send + 'static,
     ) -> Self {
         Self {
             inner: Box::pin(fut),
@@ -768,15 +768,15 @@ impl ExportProgress {
 
 /// A handle for an ongoing bao import operation.
 pub struct ImportBaoHandle {
-    pub tx: spsc::Sender<BaoContentItem>,
+    pub tx: mpsc::Sender<BaoContentItem>,
     pub rx: oneshot::Receiver<super::Result<()>>,
 }
 
 impl ImportBaoHandle {
     pub(crate) async fn new(
         fut: impl Future<
             Output = irpc::Result<(
-                spsc::Sender<BaoContentItem>,
+                mpsc::Sender<BaoContentItem>,
                 oneshot::Receiver<super::Result<()>>,
             )>,
         > + Send
@@ -789,20 +789,20 @@ impl ImportBaoHandle {
 
 /// A progress handle for a blobs list operation.
 pub struct BlobsListProgress {
-    inner: future::Boxed<irpc::Result<spsc::Receiver<super::Result<Hash>>>>,
+    inner: future::Boxed<irpc::Result<mpsc::Receiver<super::Result<Hash>>>>,
 }
 
 impl BlobsListProgress {
     fn new(
-        fut: impl Future<Output = irpc::Result<spsc::Receiver<super::Result<Hash>>>> + Send + 'static,
+        fut: impl Future<Output = irpc::Result<mpsc::Receiver<super::Result<Hash>>>> + Send + 'static,
     ) -> Self {
         Self {
             inner: Box::pin(fut),
         }
     }
 
     pub async fn hashes(self) -> RequestResult<Vec<Hash>> {
-        let mut rx: spsc::Receiver<Result<Hash, super::Error>> = self.inner.await?;
+        let mut rx: mpsc::Receiver<Result<Hash, super::Error>> = self.inner.await?;
         let mut hashes = Vec::new();
         while let Some(item) = rx.recv().await? {
             hashes.push(item?);
@@ -829,13 +829,13 @@ impl BlobsListProgress {
 /// You can get access to the underlying stream using the [`ExportBaoResult::stream`] method.
 pub struct ExportRangesProgress {
     ranges: RangeSet2<u64>,
-    inner: future::Boxed<irpc::Result<spsc::Receiver<ExportRangesItem>>>,
+    inner: future::Boxed<irpc::Result<mpsc::Receiver<ExportRangesItem>>>,
 }
 
 impl ExportRangesProgress {
     fn new(
         ranges: RangeSet2<u64>,
-        fut: impl Future<Output = irpc::Result<spsc::Receiver<ExportRangesItem>>> + Send + 'static,
+        fut: impl Future<Output = irpc::Result<mpsc::Receiver<ExportRangesItem>>> + Send + 'static,
     ) -> Self {
         Self {
             ranges,
@@ -909,12 +909,12 @@ impl ExportRangesProgress {
 ///
 /// You can get access to the underlying stream using the [`ExportBaoResult::stream`] method.
 pub struct ExportBaoProgress {
-    inner: future::Boxed<irpc::Result<spsc::Receiver<EncodedItem>>>,
+    inner: future::Boxed<irpc::Result<mpsc::Receiver<EncodedItem>>>,
 }
 
 impl ExportBaoProgress {
     fn new(
-        fut: impl Future<Output = irpc::Result<spsc::Receiver<EncodedItem>>> + Send + 'static,
+        fut: impl Future<Output = irpc::Result<mpsc::Receiver<EncodedItem>>> + Send + 'static,
     ) -> Self {
         Self {
             inner: Box::pin(fut),