Keep anyhow internally for now

Author: rklaehn

src/api/remote.rs

@@ -8,12 +8,12 @@ use n0_future::{Stream, StreamExt, io};
 use n0_snafu::SpanTrace;
 use nested_enum_utils::common_fields;
 use ref_cast::RefCast;
-use snafu::{Backtrace, Snafu};
+use snafu::{Backtrace, IntoError, Snafu};
 
 use super::blobs::Bitfield;
 use crate::{
     api::{ApiClient, blobs::WriteProgress},
-    get::{GetError, GetResult, Stats, fsm::DecodeError},
+    get::{BadRequestSnafu, GetError, GetResult, LocalFailureSnafu, Stats, fsm::DecodeError},
     protocol::{
         GetManyRequest, MAX_MESSAGE_SIZE, ObserveItem, ObserveRequest, PushRequest, Request,
         RequestType,
@@ -94,9 +94,7 @@ impl GetProgress {
 
     pub async fn complete(self) -> GetResult<Stats> {
         just_result(self.stream()).await.unwrap_or_else(|| {
-            Err(GetError::LocalFailure {
-                source: anyhow::anyhow!("stream closed without result"),
-            })
+            Err(LocalFailureSnafu.into_error(anyhow::anyhow!("stream closed without result")))
         })
     }
 }
@@ -505,15 +503,15 @@ impl Remote {
         let local = self
             .local(content)
             .await
-            .map_err(|source| GetError::LocalFailure { source })?;
+            .map_err(|e| LocalFailureSnafu.into_error(e))?;
         if local.is_complete() {
             return Ok(Default::default());
         }
         let request = local.missing();
         let conn = conn
             .connection()
             .await
-            .map_err(|source| GetError::LocalFailure { source })?;
+            .map_err(|e| LocalFailureSnafu.into_error(e))?;
         let stats = self.execute_get_sink(conn, request, progress).await?;
         Ok(stats)
     }
@@ -682,9 +680,9 @@ impl Remote {
                     store
                         .get_bytes(root)
                         .await
-                        .map_err(|e| GetError::LocalFailure { source: e.into() })?,
+                        .map_err(|e| LocalFailureSnafu.into_error(e.into()))?,
                 )
-                .map_err(|source| GetError::BadRequest { source })?;
+                .map_err(|source| BadRequestSnafu.into_error(source))?;
                 // let mut hash_seq = LazyHashSeq::new(store.blobs().clone(), root);
                 loop {
                     let at_start_child = match next_child {
@@ -889,7 +887,7 @@ async fn get_blob_ranges_impl(
     let handle = store
         .import_bao(hash, size, buffer_size)
         .await
-        .map_err(|e| GetError::LocalFailure { source: e.into() })?;
+        .map_err(|e| LocalFailureSnafu.into_error(e.into()))?;
     let write = async move {
         GetResult::Ok(loop {
             match content.next().await {
@@ -898,7 +896,7 @@ async fn get_blob_ranges_impl(
                     progress
                         .send(next.stats().payload_bytes_read)
                         .await
-                        .map_err(|e| GetError::LocalFailure { source: e.into() })?;
+                        .map_err(|e| LocalFailureSnafu.into_error(e.into()))?;
                     handle.tx.send(item).await?;
                     content = next;
                 }
@@ -910,8 +908,8 @@ async fn get_blob_ranges_impl(
         })
     };
     let complete = async move {
-        handle.rx.await.map_err(|e| GetError::LocalFailure {
-            source: anyhow::anyhow!("error reading from import stream: {e}"),
+        handle.rx.await.map_err(|e| {
+            LocalFailureSnafu.into_error(anyhow::anyhow!("error reading from import stream: {e}"))
         })
     };
     let (_, end) = tokio::try_join!(complete, write)?;

src/get.rs

@@ -37,6 +37,7 @@ use crate::{Hash, protocol::ChunkRangesSeq, store::IROH_BLOCK_SIZE};
 
 mod error;
 pub mod request;
+pub(crate) use error::{BadRequestSnafu, LocalFailureSnafu};
 pub use error::{GetError, GetResult};
 
 type WrappedRecvStream = TokioStreamReader<RecvStream>;
@@ -98,8 +99,11 @@ pub mod fsm {
     use iroh_io::{AsyncSliceWriter, AsyncStreamReader, TokioStreamReader};
 
     use super::*;
-    use crate::protocol::{
-        GetManyRequest, GetRequest, MAX_MESSAGE_SIZE, NonEmptyRequestRangeSpecIter, Request,
+    use crate::{
+        get::error::BadRequestSnafu,
+        protocol::{
+            GetManyRequest, GetRequest, MAX_MESSAGE_SIZE, NonEmptyRequestRangeSpecIter, Request,
+        },
     };
 
     self_cell::self_cell! {
@@ -128,10 +132,8 @@ pub mod fsm {
         let start = Instant::now();
         let (mut writer, reader) = connection.open_bi().await?;
         let request = Request::GetMany(request);
-        let request_bytes =
-            postcard::to_stdvec(&request).map_err(|source| GetError::BadRequest {
-                source: source.into(),
-            })?;
+        let request_bytes = postcard::to_stdvec(&request)
+            .map_err(|source| BadRequestSnafu.into_error(source.into()))?;
         writer.write_all(&request_bytes).await?;
         writer.finish()?;
         let Request::GetMany(request) = request else {

src/get/error.rs

@@ -3,7 +3,7 @@
 use iroh::endpoint::{self, ClosedStream};
 use n0_snafu::SpanTrace;
 use nested_enum_utils::common_fields;
-use snafu::{Backtrace, Snafu};
+use snafu::{Backtrace, IntoError, Snafu};
 
 #[common_fields({
     backtrace: Option<Backtrace>,
@@ -17,6 +17,7 @@ pub enum GetNotFoundError {
 
 /// Failures for a get operation
 #[derive(Debug, Snafu)]
+#[snafu(visibility(pub(crate)))]
 pub enum GetError {
     /// Hash not found, or a requested chunk for the hash not found.
     #[snafu(display("Data for hash not found"))]
@@ -44,17 +45,13 @@ pub type GetResult<T> = std::result::Result<T, GetError>;
 
 impl From<irpc::channel::SendError> for GetError {
     fn from(value: irpc::channel::SendError) -> Self {
-        Self::LocalFailure {
-            source: value.into(),
-        }
+        LocalFailureSnafu.into_error(value.into())
     }
 }
 
 impl<T: Send + Sync + 'static> From<tokio::sync::mpsc::error::SendError<T>> for GetError {
     fn from(value: tokio::sync::mpsc::error::SendError<T>) -> Self {
-        Self::LocalFailure {
-            source: value.into(),
-        }
+        LocalFailureSnafu.into_error(value.into())
     }
 }
 
@@ -66,40 +63,40 @@ impl From<endpoint::ConnectionError> for GetError {
             e @ ConnectionError::VersionMismatch => {
                 // > The peer doesn't implement any supported version
                 // unsupported version is likely a long time error, so this peer is not usable
-                GetError::NoncompliantNode { source: e.into() }
+                NoncompliantNodeSnafu.into_error(e.into())
             }
             e @ ConnectionError::TransportError(_) => {
                 // > The peer violated the QUIC specification as understood by this implementation
                 // bad peer we don't want to keep around
-                GetError::NoncompliantNode { source: e.into() }
+                NoncompliantNodeSnafu.into_error(e.into())
             }
             e @ ConnectionError::ConnectionClosed(_) => {
                 // > The peer's QUIC stack aborted the connection automatically
                 // peer might be disconnecting or otherwise unavailable, drop it
-                GetError::Io { source: e.into() }
+                IoSnafu.into_error(e.into())
             }
             e @ ConnectionError::ApplicationClosed(_) => {
                 // > The peer closed the connection
                 // peer might be disconnecting or otherwise unavailable, drop it
-                GetError::Io { source: e.into() }
+                IoSnafu.into_error(e.into())
             }
             e @ ConnectionError::Reset => {
                 // > The peer is unable to continue processing this connection, usually due to having restarted
-                GetError::RemoteReset { source: e.into() }
+                RemoteResetSnafu.into_error(e.into())
             }
             e @ ConnectionError::TimedOut => {
                 // > Communication with the peer has lapsed for longer than the negotiated idle timeout
-                GetError::Io { source: e.into() }
+                IoSnafu.into_error(e.into())
             }
             e @ ConnectionError::LocallyClosed => {
                 // > The local application closed the connection
                 // TODO(@divma): don't see how this is reachable but let's just not use the peer
-                GetError::Io { source: e.into() }
+                IoSnafu.into_error(e.into())
             }
             e @ ConnectionError::CidsExhausted => {
                 // > The connection could not be created because not enough of the CID space
                 // > is available
-                GetError::Io { source: e.into() }
+                IoSnafu.into_error(e.into())
             }
         }
     }
@@ -109,38 +106,32 @@ impl From<endpoint::ReadError> for GetError {
     fn from(value: endpoint::ReadError) -> Self {
         use endpoint::ReadError;
         match value {
-            e @ ReadError::Reset(_) => GetError::RemoteReset { source: e.into() },
+            e @ ReadError::Reset(_) => RemoteResetSnafu.into_error(e.into()),
             ReadError::ConnectionLost(conn_error) => conn_error.into(),
             ReadError::ClosedStream
             | ReadError::IllegalOrderedRead
             | ReadError::ZeroRttRejected => {
                 // all these errors indicate the peer is not usable at this moment
-                GetError::Io {
-                    source: value.into(),
-                }
+                IoSnafu.into_error(value.into())
             }
         }
     }
 }
 impl From<ClosedStream> for GetError {
     fn from(value: ClosedStream) -> Self {
-        GetError::Io {
-            source: value.into(),
-        }
+        IoSnafu.into_error(value.into())
     }
 }
 
 impl From<quinn::WriteError> for GetError {
     fn from(value: quinn::WriteError) -> Self {
         use quinn::WriteError;
         match value {
-            e @ WriteError::Stopped(_) => GetError::RemoteReset { source: e.into() },
+            e @ WriteError::Stopped(_) => RemoteResetSnafu.into_error(e.into()),
             WriteError::ConnectionLost(conn_error) => conn_error.into(),
             WriteError::ClosedStream | WriteError::ZeroRttRejected => {
                 // all these errors indicate the peer is not usable at this moment
-                GetError::Io {
-                    source: value.into(),
-                }
+                IoSnafu.into_error(value.into())
             }
         }
     }
@@ -152,17 +143,17 @@ impl From<crate::get::fsm::ConnectedNextError> for GetError {
         match value {
             e @ PostcardSer { .. } => {
                 // serialization errors indicate something wrong with the request itself
-                GetError::BadRequest { source: e.into() }
+                BadRequestSnafu.into_error(e.into())
             }
             e @ RequestTooBig { .. } => {
                 // request will never be sent, drop it
-                GetError::BadRequest { source: e.into() }
+                BadRequestSnafu.into_error(e.into())
             }
             Write { source, .. } => source.into(),
             Closed { source, .. } => source.into(),
             e @ Io { .. } => {
                 // io errors are likely recoverable
-                GetError::Io { source: e.into() }
+                IoSnafu.into_error(e.into())
             }
         }
     }
@@ -178,17 +169,15 @@ impl From<crate::get::fsm::AtBlobHeaderNextError> for GetError {
             } => {
                 // > This indicates that the provider does not have the requested data.
                 // peer might have the data later, simply retry it
-                GetError::NotFound {
-                    source: GetNotFoundError::AtBlobHeader {
-                        backtrace,
-                        span_trace,
-                    },
-                }
+                NotFoundSnafu.into_error(GetNotFoundError::AtBlobHeader {
+                    backtrace,
+                    span_trace,
+                })
             }
             EndpointRead { source, .. } => source.into(),
             e @ Io { .. } => {
                 // io errors are likely recoverable
-                GetError::Io { source: e.into() }
+                IoSnafu.into_error(e.into())
             }
         }
     }
@@ -202,41 +191,35 @@ impl From<crate::get::fsm::DecodeError> for GetError {
             ChunkNotFound {
                 backtrace,
                 span_trace,
-            } => GetError::NotFound {
-                source: GetNotFoundError::AtBlobHeader {
-                    backtrace,
-                    span_trace,
-                },
-            },
+            } => NotFoundSnafu.into_error(GetNotFoundError::AtBlobHeader {
+                backtrace,
+                span_trace,
+            }),
             ParentNotFound {
                 backtrace,
                 span_trace,
                 ..
-            } => GetError::NotFound {
-                source: GetNotFoundError::AtBlobHeader {
-                    backtrace,
-                    span_trace,
-                },
-            },
+            } => NotFoundSnafu.into_error(GetNotFoundError::AtBlobHeader {
+                backtrace,
+                span_trace,
+            }),
             LeafNotFound {
                 backtrace,
                 span_trace,
                 ..
-            } => GetError::NotFound {
-                source: GetNotFoundError::AtBlobHeader {
-                    backtrace,
-                    span_trace,
-                },
-            },
+            } => NotFoundSnafu.into_error(GetNotFoundError::AtBlobHeader {
+                backtrace,
+                span_trace,
+            }),
             e @ ParentHashMismatch { .. } => {
                 // TODO(@divma): did the peer sent wrong data? is it corrupted? did we sent a wrong
                 // request?
-                GetError::NoncompliantNode { source: e.into() }
+                NoncompliantNodeSnafu.into_error(e.into())
             }
             e @ LeafHashMismatch { .. } => {
                 // TODO(@divma): did the peer sent wrong data? is it corrupted? did we sent a wrong
                 // request?
-                GetError::NoncompliantNode { source: e.into() }
+                NoncompliantNodeSnafu.into_error(e.into())
             }
             Read { source, .. } => source.into(),
             DecodeIo { source, .. } => source.into(),
@@ -248,8 +231,6 @@ impl From<std::io::Error> for GetError {
     fn from(value: std::io::Error) -> Self {
         // generally consider io errors recoverable
         // we might want to revisit this at some point
-        GetError::Io {
-            source: value.into(),
-        }
+        IoSnafu.into_error(value.into())
     }
 }