Refactor error and make get and provide side generic

rklaehn · rklaehn · commit 4e8387a07180 · 2025-09-04T16:04:44.000+02:00
diff --git a/src/api.rs b/src/api.rs
@@ -98,7 +98,7 @@ pub enum ExportBaoError {
     #[snafu(display("encode error: {source}"))]
     ExportBaoInner { source: bao_tree::io::EncodeError },
     #[snafu(display("client error: {source}"))]
-    ClientError { source: ProgressError },
+    Progress { source: ProgressError },
 }
 
 impl From<ExportBaoError> for Error {
@@ -109,7 +109,7 @@ impl From<ExportBaoError> for Error {
             ExportBaoError::Request { source, .. } => Self::Io(source.into()),
             ExportBaoError::ExportBaoIo { source, .. } => Self::Io(source),
             ExportBaoError::ExportBaoInner { source, .. } => Self::Io(source.into()),
-            ExportBaoError::ClientError { source, .. } => Self::Io(source.into()),
+            ExportBaoError::Progress { source, .. } => Self::Io(source.into()),
         }
     }
 }
@@ -157,7 +157,7 @@ impl From<bao_tree::io::EncodeError> for ExportBaoError {
 
 impl From<ProgressError> for ExportBaoError {
     fn from(value: ProgressError) -> Self {
-        ClientSnafu.into_error(value)
+        ProgressSnafu.into_error(value)
     }
 }
 
diff --git a/src/api/downloader.rs b/src/api/downloader.rs
@@ -563,7 +563,7 @@ mod tests {
             .download(request, Shuffled::new(vec![node1_id, node2_id]))
             .stream()
             .await?;
-        while let Some(_) = progress.next().await {}
+        while progress.next().await.is_some() {}
         assert_eq!(store3.get_bytes(tt1.hash).await?.deref(), b"hello world");
         assert_eq!(store3.get_bytes(tt2.hash).await?.deref(), b"hello world 2");
         Ok(())
@@ -606,7 +606,7 @@ mod tests {
                 ))
                 .stream()
                 .await?;
-            while let Some(_) = progress.next().await {}
+            while progress.next().await.is_some() {}
         }
         if false {
             let conn = r3.endpoint().connect(node1_addr, crate::ALPN).await?;
@@ -668,7 +668,7 @@ mod tests {
             ))
             .stream()
             .await?;
-        while let Some(_) = progress.next().await {}
+        while progress.next().await.is_some() {}
         Ok(())
     }
 }
diff --git a/src/api/remote.rs b/src/api/remote.rs
@@ -99,8 +99,7 @@ impl GetProgress {
 
     pub async fn complete(self) -> GetResult<Stats> {
         just_result(self.stream()).await.unwrap_or_else(|| {
-            Err(LocalFailureSnafu
-                .into_error(anyhow::anyhow!("stream closed without result").into()))
+            Err(LocalFailureSnafu.into_error(anyhow::anyhow!("stream closed without result")))
         })
     }
 }
@@ -512,15 +511,15 @@ impl Remote {
         let local = self
             .local(content)
             .await
-            .map_err(|e: anyhow::Error| LocalFailureSnafu.into_error(e.into()))?;
+            .map_err(|e: anyhow::Error| LocalFailureSnafu.into_error(e))?;
         if local.is_complete() {
             return Ok(Default::default());
         }
         let request = local.missing();
         let conn = conn
             .connection()
             .await
-            .map_err(|e| LocalFailureSnafu.into_error(e.into()))?;
+            .map_err(|e| LocalFailureSnafu.into_error(e))?;
         let stats = self.execute_get_sink(&conn, request, progress).await?;
         Ok(stats)
     }
@@ -914,8 +913,7 @@ async fn get_blob_ranges_impl(
     };
     let complete = async move {
         handle.rx.await.map_err(|e| {
-            LocalFailureSnafu
-                .into_error(anyhow::anyhow!("error reading from import stream: {e}").into())
+            LocalFailureSnafu.into_error(anyhow::anyhow!("error reading from import stream: {e}"))
         })
     };
     let (_, end) = tokio::try_join!(complete, write)?;
diff --git a/src/get.rs b/src/get.rs
@@ -53,7 +53,7 @@ impl AsyncStreamWriter for IrohStreamWriter {
     }
 
     async fn sync(&mut self) -> io::Result<()> {
-        Ok(self.0.flush().await?)
+        self.0.flush().await
     }
 }
 
@@ -716,7 +716,7 @@ pub mod fsm {
                 DecodeError::LeafNotFound { .. } => {
                     io::Error::new(io::ErrorKind::UnexpectedEof, cause)
                 }
-                DecodeError::Read { source, .. } => source.into(),
+                DecodeError::Read { source, .. } => source,
                 DecodeError::Write { source, .. } => source,
                 _ => io::Error::other(cause),
             }
diff --git a/src/get/error.rs b/src/get/error.rs
@@ -82,7 +82,7 @@ impl GetError {
             Self::ConnectedNext {
                 source: ConnectedNextError::Write { source, .. },
                 ..
-            } => Some(&source),
+            } => Some(source),
             _ => None,
         }
     }
@@ -92,7 +92,7 @@ impl GetError {
             Self::InitialNext {
                 source: InitialNextError::Open { source, .. },
                 ..
-            } => Some(&source),
+            } => Some(source),
             _ => None,
         }
     }
@@ -102,15 +102,15 @@ impl GetError {
             Self::AtBlobHeaderNext {
                 source: AtBlobHeaderNextError::Read { source, .. },
                 ..
-            } => Some(&source),
+            } => Some(source),
             Self::Decode {
                 source: DecodeError::Read { source, .. },
                 ..
-            } => Some(&source),
+            } => Some(source),
             Self::AtClosingNext {
                 source: AtClosingNextError::Read { source, .. },
                 ..
-            } => Some(&source),
+            } => Some(source),
             _ => None,
         }
     }
@@ -120,7 +120,7 @@ impl GetError {
             Self::Decode {
                 source: DecodeError::Write { source, .. },
                 ..
-            } => Some(&source),
+            } => Some(source),
             _ => None,
         }
     }
diff --git a/src/get/request.rs b/src/get/request.rs
@@ -58,7 +58,7 @@ impl GetBlobResult {
         let mut parts = Vec::new();
         let stats = loop {
             let Some(item) = self.next().await else {
-                return Err(LocalFailureSnafu.into_error(anyhow::anyhow!("unexpected end").into()));
+                return Err(LocalFailureSnafu.into_error(anyhow::anyhow!("unexpected end")));
             };
             match item {
                 GetBlobItem::Item(item) => {
@@ -238,11 +238,11 @@ pub async fn get_hash_seq_and_sizes(
     let (at_blob_content, size) = at_start_root.next().await?;
     // check the size to avoid parsing a maliciously large hash seq
     if size > max_size {
-        return Err(BadRequestSnafu.into_error(anyhow::anyhow!("size too large").into()));
+        return Err(BadRequestSnafu.into_error(anyhow::anyhow!("size too large")));
     }
     let (mut curr, hash_seq) = at_blob_content.concatenate_into_vec().await?;
-    let hash_seq = HashSeq::try_from(Bytes::from(hash_seq))
-        .map_err(|e| BadRequestSnafu.into_error(e.into()))?;
+    let hash_seq =
+        HashSeq::try_from(Bytes::from(hash_seq)).map_err(|e| BadRequestSnafu.into_error(e))?;
     let mut sizes = Vec::with_capacity(hash_seq.len());
     let closing = loop {
         match curr.next() {
diff --git a/src/protocol.rs b/src/protocol.rs
@@ -382,15 +382,14 @@ use bao_tree::{io::round_up_to_chunks, ChunkNum};
 use builder::GetRequestBuilder;
 use derive_more::From;
 use iroh::endpoint::VarInt;
-use irpc::util::AsyncReadVarintExt;
+use iroh_io::AsyncStreamReader;
 use postcard::experimental::max_size::MaxSize;
 use range_collections::{range_set::RangeSetEntry, RangeSet2};
 use serde::{Deserialize, Serialize};
 mod range_spec;
 pub use bao_tree::ChunkRanges;
 pub use range_spec::{ChunkRangesSeq, NonEmptyRequestRangeSpecIter, RangeSpec};
 use snafu::{GenerateImplicitData, Snafu};
-use tokio::io::AsyncReadExt;
 
 use crate::{api::blobs::Bitfield, provider::RecvStreamExt, BlobFormat, Hash, HashAndFormat};
 
@@ -448,7 +447,7 @@ pub enum RequestType {
 }
 
 impl Request {
-    pub async fn read_async(reader: &mut iroh::endpoint::RecvStream) -> io::Result<(Self, usize)> {
+    pub async fn read_async<R: AsyncStreamReader>(reader: &mut R) -> io::Result<(Self, usize)> {
         let request_type = reader.read_u8().await?;
         let request_type: RequestType = postcard::from_bytes(std::slice::from_ref(&request_type))
             .map_err(|_| {
diff --git a/src/provider.rs b/src/provider.rs
diff --git a/src/provider/events.rs b/src/provider/events.rs
diff --git a/src/tests.rs b/src/tests.rs

Original file line number	Diff line number	Diff line change
`@@ -98,7 +98,7 @@ pub enum ExportBaoError {`
`98`	`98`	`#[snafu(display("encode error: {source}"))]`
`99`	`99`	`ExportBaoInner { source: bao_tree::io::EncodeError },`
`100`	`100`	`#[snafu(display("client error: {source}"))]`
`101`		`- ClientError { source: ProgressError },`
	`101`	`+ Progress { source: ProgressError },`
`102`	`102`	`}`
`103`	`103`
`104`	`104`	`impl From<ExportBaoError> for Error {`
`@@ -109,7 +109,7 @@ impl From<ExportBaoError> for Error {`
`109`	`109`	`ExportBaoError::Request { source, .. } => Self::Io(source.into()),`
`110`	`110`	`ExportBaoError::ExportBaoIo { source, .. } => Self::Io(source),`
`111`	`111`	`ExportBaoError::ExportBaoInner { source, .. } => Self::Io(source.into()),`
`112`		`- ExportBaoError::ClientError { source, .. } => Self::Io(source.into()),`
	`112`	`+ ExportBaoError::Progress { source, .. } => Self::Io(source.into()),`
`113`	`113`	`}`
`114`	`114`	`}`
`115`	`115`	`}`
`@@ -157,7 +157,7 @@ impl From<bao_tree::io::EncodeError> for ExportBaoError {`
`157`	`157`
`158`	`158`	`impl From<ProgressError> for ExportBaoError {`
`159`	`159`	`fn from(value: ProgressError) -> Self {`
`160`		`- ClientSnafu.into_error(value)`
	`160`	`+ ProgressSnafu.into_error(value)`
`161`	`161`	`}`
`162`	`162`	`}`
`163`	`163`
Original file line number	Diff line number	Diff line change
`@@ -53,7 +53,7 @@ impl AsyncStreamWriter for IrohStreamWriter {`
`53`	`53`	`}`
`54`	`54`
`55`	`55`	`async fn sync(&mut self) -> io::Result<()> {`
`56`		`- Ok(self.0.flush().await?)`
	`56`	`+ self.0.flush().await`
`57`	`57`	`}`
`58`	`58`	`}`
`59`	`59`
`@@ -716,7 +716,7 @@ pub mod fsm {`
`716`	`716`	`DecodeError::LeafNotFound { .. } => {`
`717`	`717`	`io::Error::new(io::ErrorKind::UnexpectedEof, cause)`
`718`	`718`	`}`
`719`		`- DecodeError::Read { source, .. } => source.into(),`
	`719`	`+ DecodeError::Read { source, .. } => source,`
`720`	`720`	`DecodeError::Write { source, .. } => source,`
`721`	`721`	`_ => io::Error::other(cause),`
`722`	`722`	`}`
Original file line number	Diff line number	Diff line change
`@@ -82,7 +82,7 @@ impl GetError {`
`82`	`82`	`Self::ConnectedNext {`
`83`	`83`	`source: ConnectedNextError::Write { source, .. },`
`84`	`84`	`..`
`85`		`- } => Some(&source),`
	`85`	`+ } => Some(source),`
`86`	`86`	`_ => None,`
`87`	`87`	`}`
`88`	`88`	`}`
`@@ -92,7 +92,7 @@ impl GetError {`
`92`	`92`	`Self::InitialNext {`
`93`	`93`	`source: InitialNextError::Open { source, .. },`
`94`	`94`	`..`
`95`		`- } => Some(&source),`
	`95`	`+ } => Some(source),`
`96`	`96`	`_ => None,`
`97`	`97`	`}`
`98`	`98`	`}`
`@@ -102,15 +102,15 @@ impl GetError {`
`102`	`102`	`Self::AtBlobHeaderNext {`
`103`	`103`	`source: AtBlobHeaderNextError::Read { source, .. },`
`104`	`104`	`..`
`105`		`- } => Some(&source),`
	`105`	`+ } => Some(source),`
`106`	`106`	`Self::Decode {`
`107`	`107`	`source: DecodeError::Read { source, .. },`
`108`	`108`	`..`
`109`		`- } => Some(&source),`
	`109`	`+ } => Some(source),`
`110`	`110`	`Self::AtClosingNext {`
`111`	`111`	`source: AtClosingNextError::Read { source, .. },`
`112`	`112`	`..`
`113`		`- } => Some(&source),`
	`113`	`+ } => Some(source),`
`114`	`114`	`_ => None,`
`115`	`115`	`}`
`116`	`116`	`}`
`@@ -120,7 +120,7 @@ impl GetError {`
`120`	`120`	`Self::Decode {`
`121`	`121`	`source: DecodeError::Write { source, .. },`
`122`	`122`	`..`
`123`		`- } => Some(&source),`
	`123`	`+ } => Some(source),`
`124`	`124`	`_ => None,`
`125`	`125`	`}`
`126`	`126`	`}`