remove GetResponseError and rename get to fetch

Author: rklaehn

examples/fetch-fsm.rs

@@ -7,7 +7,7 @@ use std::net::SocketAddr;
 
 use anyhow::{Context, Result};
 use iroh_blobs::{
-    get::fsm::{AtInitial, ConnectedNext, EndBlobNext},
+    fetch::fsm::{AtInitial, ConnectedNext, EndBlobNext},
     hashseq::HashSeq,
     protocol::GetRequest,
     Hash,
@@ -63,14 +63,14 @@ async fn main() -> Result<()> {
         // create a request for a collection
         let request = GetRequest::all(hash);
         // create the initial state of the finite state machine
-        let initial = iroh_blobs::get::fsm::start(connection, request);
+        let initial = iroh_blobs::fetch::fsm::start(connection, request);
 
         write_collection(initial).await
     } else {
         // create a request for a single blob
         let request = GetRequest::single(hash);
         // create the initial state of the finite state machine
-        let initial = iroh_blobs::get::fsm::start(connection, request);
+        let initial = iroh_blobs::fetch::fsm::start(connection, request);
 
         write_blob(initial).await
     }
@@ -119,7 +119,7 @@ async fn write_collection(initial: AtInitial) -> Result<()> {
     }
 
     // move to the header
-    let header: iroh_blobs::get::fsm::AtBlobHeader = start_root.next();
+    let header: iroh_blobs::fetch::fsm::AtBlobHeader = start_root.next();
     let (root_end, hashes_bytes) = header.concatenate_into_vec().await?;
     let next = root_end.next();
     let EndBlobNext::MoreChildren(at_meta) = next else {

examples/fetch-stream.rs

@@ -11,7 +11,7 @@ use bytes::Bytes;
 use futures_lite::{Stream, StreamExt};
 use genawaiter::sync::{Co, Gen};
 use iroh_blobs::{
-    get::fsm::{AtInitial, BlobContentNext, ConnectedNext, EndBlobNext},
+    fetch::fsm::{AtInitial, BlobContentNext, ConnectedNext, EndBlobNext},
     hashseq::HashSeq,
     protocol::GetRequest,
     Hash,
@@ -68,7 +68,7 @@ async fn main() -> Result<()> {
         let request = GetRequest::all(hash);
 
         // create the initial state of the finite state machine
-        let initial = iroh_blobs::get::fsm::start(connection, request);
+        let initial = iroh_blobs::fetch::fsm::start(connection, request);
 
         // create a stream that yields all the data of the blob
         stream_children(initial).boxed_local()
@@ -77,7 +77,7 @@ async fn main() -> Result<()> {
         let request = GetRequest::single(hash);
 
         // create the initial state of the finite state machine
-        let initial = iroh_blobs::get::fsm::start(connection, request);
+        let initial = iroh_blobs::fetch::fsm::start(connection, request);
 
         // create a stream that yields all the data of the blob
         stream_blob(initial).boxed_local()
@@ -166,7 +166,7 @@ fn stream_children(initial: AtInitial) -> impl Stream<Item = io::Result<Bytes>>
             ));
         }
         // move to the header
-        let header: iroh_blobs::get::fsm::AtBlobHeader = start_root.next();
+        let header: iroh_blobs::fetch::fsm::AtBlobHeader = start_root.next();
         let (root_end, hashes_bytes) = header.concatenate_into_vec().await?;
 
         // parse the hashes from the hash sequence bytes

examples/local-swarm-discovery.rs

@@ -140,7 +140,7 @@ mod progress {
         ProgressStyle,
     };
     use iroh_blobs::{
-        get::{db::DownloadProgress, progress::BlobProgress, Stats},
+        fetch::{db::DownloadProgress, progress::BlobProgress, Stats},
         Hash,
     };
 

src/downloader.rs

@@ -55,7 +55,7 @@ use tokio_util::{either::Either, sync::CancellationToken, time::delay_queue};
 use tracing::{debug, error, error_span, trace, warn, Instrument};
 
 use crate::{
-    get::{db::DownloadProgress, Stats},
+    fetch::{db::DownloadProgress, Stats},
     metrics::Metrics,
     store::Store,
     util::{local_pool::LocalPoolHandle, progress::ProgressSender},

src/downloader/get.rs

@@ -7,7 +7,7 @@ use iroh::endpoint;
 
 use super::{progress::BroadcastProgressSender, DownloadKind, FailureAction, GetStartFut, Getter};
 use crate::{
-    get::{db::get_to_db_in_steps, error::GetError},
+    fetch::{db::get_to_db_in_steps, GetError},
     store::Store,
 };
 
@@ -34,7 +34,7 @@ pub(crate) struct IoGetter<S: Store> {
 
 impl<S: Store> Getter for IoGetter<S> {
     type Connection = endpoint::Connection;
-    type NeedsConn = crate::get::db::GetStateNeedsConn;
+    type NeedsConn = crate::fetch::db::GetStateNeedsConn;
 
     fn get(
         &mut self,
@@ -45,10 +45,10 @@ impl<S: Store> Getter for IoGetter<S> {
         async move {
             match get_to_db_in_steps(store, kind.hash_and_format(), progress_sender).await {
                 Err(err) => Err(err.into()),
-                Ok(crate::get::db::GetState::Complete(stats)) => {
+                Ok(crate::fetch::db::GetState::Complete(stats)) => {
                     Ok(super::GetOutput::Complete(stats))
                 }
-                Ok(crate::get::db::GetState::NeedsConn(needs_conn)) => {
+                Ok(crate::fetch::db::GetState::NeedsConn(needs_conn)) => {
                     Ok(super::GetOutput::NeedsConn(needs_conn))
                 }
             }
@@ -57,7 +57,7 @@ impl<S: Store> Getter for IoGetter<S> {
     }
 }
 
-impl super::NeedsConn<endpoint::Connection> for crate::get::db::GetStateNeedsConn {
+impl super::NeedsConn<endpoint::Connection> for crate::fetch::db::GetStateNeedsConn {
     fn proceed(self, conn: endpoint::Connection) -> super::GetProceedFut {
         async move {
             let res = self.proceed(conn).await;
@@ -73,13 +73,13 @@ impl super::NeedsConn<endpoint::Connection> for crate::get::db::GetStateNeedsCon
 }
 
 #[cfg(feature = "metrics")]
-fn track_metrics(res: &Result<crate::get::Stats, GetError>) {
+fn track_metrics(res: &Result<crate::fetch::Stats, GetError>) {
     use iroh_metrics::{inc, inc_by};
 
     use crate::metrics::Metrics;
     match res {
         Ok(stats) => {
-            let crate::get::Stats {
+            let crate::fetch::Stats {
                 bytes_written,
                 bytes_read: _,
                 elapsed,

src/downloader/progress.rs

@@ -11,7 +11,7 @@ use parking_lot::Mutex;
 
 use super::DownloadKind;
 use crate::{
-    get::{db::DownloadProgress, progress::TransferState},
+    fetch::{db::DownloadProgress, progress::TransferState},
     util::progress::{AsyncChannelProgressSender, IdGenerator, ProgressSendError, ProgressSender},
 };
 

src/downloader/test.rs

@@ -10,7 +10,7 @@ use iroh::SecretKey;
 
 use super::*;
 use crate::{
-    get::{
+    fetch::{
         db::BlobId,
         progress::{BlobProgress, TransferState},
     },

src/get.rs renamed to src/fetch.rs

@@ -13,7 +13,6 @@
 //!
 //! [iroh-net]: https://docs.rs/iroh-net
 use std::{
-    error::Error,
     fmt::{self, Debug},
     time::{Duration, Instant},
 };
@@ -30,8 +29,9 @@ use crate::{
     Hash, IROH_BLOCK_SIZE,
 };
 
+mod error;
+pub use error::GetError;
 pub mod db;
-pub mod error;
 pub mod progress;
 pub mod request;
 
@@ -882,64 +882,3 @@ pub mod fsm {
         ranges_iter: RangesIter,
     }
 }
-
-/// Error when processing a response
-#[derive(thiserror::Error, Debug)]
-pub enum GetResponseError {
-    /// Error when opening a stream
-    #[error("connection: {0}")]
-    Connection(#[from] endpoint::ConnectionError),
-    /// Error when writing the handshake or request to the stream
-    #[error("write: {0}")]
-    Write(#[from] endpoint::WriteError),
-    /// Error when reading from the stream
-    #[error("read: {0}")]
-    Read(#[from] endpoint::ReadError),
-    /// Error when decoding, e.g. hash mismatch
-    #[error("decode: {0}")]
-    Decode(bao_tree::io::DecodeError),
-    /// A generic error
-    #[error("generic: {0}")]
-    Generic(anyhow::Error),
-}
-
-impl From<postcard::Error> for GetResponseError {
-    fn from(cause: postcard::Error) -> Self {
-        Self::Generic(cause.into())
-    }
-}
-
-impl From<bao_tree::io::DecodeError> for GetResponseError {
-    fn from(cause: bao_tree::io::DecodeError) -> Self {
-        match cause {
-            bao_tree::io::DecodeError::Io(cause) => {
-                // try to downcast to specific quinn errors
-                if let Some(source) = cause.source() {
-                    if let Some(error) = source.downcast_ref::<endpoint::ConnectionError>() {
-                        return Self::Connection(error.clone());
-                    }
-                    if let Some(error) = source.downcast_ref::<endpoint::ReadError>() {
-                        return Self::Read(error.clone());
-                    }
-                    if let Some(error) = source.downcast_ref::<endpoint::WriteError>() {
-                        return Self::Write(error.clone());
-                    }
-                }
-                Self::Generic(cause.into())
-            }
-            _ => Self::Decode(cause),
-        }
-    }
-}
-
-impl From<anyhow::Error> for GetResponseError {
-    fn from(cause: anyhow::Error) -> Self {
-        Self::Generic(cause)
-    }
-}
-
-impl From<GetResponseError> for std::io::Error {
-    fn from(cause: GetResponseError) -> Self {
-        Self::new(std::io::ErrorKind::Other, cause)
-    }
-}

src/get/db.rs renamed to src/fetch/db.rs

@@ -16,7 +16,7 @@ use tokio::sync::oneshot;
 use tracing::trace;
 
 use crate::{
-    get::{
+    fetch::{
         self,
         error::GetError,
         fsm::{AtBlobHeader, AtEndBlob, ConnectedNext, EndBlobNext},
@@ -196,7 +196,7 @@ async fn get_blob<D: BaoStore>(
             let request = GetRequest::new(*hash, RangeSpecSeq::from_ranges([required_ranges]));
             // full request
             let conn = co.get_conn().await;
-            let request = get::fsm::start(conn, request);
+            let request = fetch::fsm::start(conn, request);
             // create a new bidi stream
             let connected = request.next().await?;
             // next step. we have requested a single hash, so this must be StartRoot
@@ -212,7 +212,7 @@ async fn get_blob<D: BaoStore>(
         None => {
             // full request
             let conn = co.get_conn().await;
-            let request = get::fsm::start(conn, GetRequest::single(*hash));
+            let request = fetch::fsm::start(conn, GetRequest::single(*hash));
             // create a new bidi stream
             let connected = request.next().await?;
             // next step. we have requested a single hash, so this must be StartRoot
@@ -454,7 +454,7 @@ async fn get_hash_seq<D: BaoStore>(
             log!("requesting chunks {:?}", missing_iter);
             let request = GetRequest::new(*root_hash, RangeSpecSeq::from_ranges(missing_iter));
             let conn = co.get_conn().await;
-            let request = get::fsm::start(conn, request);
+            let request = fetch::fsm::start(conn, request);
             // create a new bidi stream
             let connected = request.next().await?;
             log!("connected");
@@ -500,7 +500,7 @@ async fn get_hash_seq<D: BaoStore>(
             tracing::debug!("don't have collection - doing full download");
             // don't have the collection, so probably got nothing
             let conn = co.get_conn().await;
-            let request = get::fsm::start(conn, GetRequest::all(*root_hash));
+            let request = fetch::fsm::start(conn, GetRequest::all(*root_hash));
             // create a new bidi stream
             let connected = request.next().await?;
             // next step. we have requested a single hash, so this must be StartRoot

src/get/error.rs renamed to src/fetch/error.rs

@@ -117,9 +117,9 @@ impl From<endpoint::WriteError> for GetError {
     }
 }
 
-impl From<crate::get::fsm::ConnectedNextError> for GetError {
-    fn from(value: crate::get::fsm::ConnectedNextError) -> Self {
-        use crate::get::fsm::ConnectedNextError::*;
+impl From<crate::fetch::fsm::ConnectedNextError> for GetError {
+    fn from(value: crate::fetch::fsm::ConnectedNextError) -> Self {
+        use crate::fetch::fsm::ConnectedNextError::*;
         match value {
             e @ PostcardSer(_) => {
                 // serialization errors indicate something wrong with the request itself
@@ -139,9 +139,9 @@ impl From<crate::get::fsm::ConnectedNextError> for GetError {
     }
 }
 
-impl From<crate::get::fsm::AtBlobHeaderNextError> for GetError {
-    fn from(value: crate::get::fsm::AtBlobHeaderNextError) -> Self {
-        use crate::get::fsm::AtBlobHeaderNextError::*;
+impl From<crate::fetch::fsm::AtBlobHeaderNextError> for GetError {
+    fn from(value: crate::fetch::fsm::AtBlobHeaderNextError) -> Self {
+        use crate::fetch::fsm::AtBlobHeaderNextError::*;
         match value {
             e @ NotFound => {
                 // > This indicates that the provider does not have the requested data.
@@ -157,9 +157,9 @@ impl From<crate::get::fsm::AtBlobHeaderNextError> for GetError {
     }
 }
 
-impl From<crate::get::fsm::DecodeError> for GetError {
-    fn from(value: crate::get::fsm::DecodeError) -> Self {
-        use crate::get::fsm::DecodeError::*;
+impl From<crate::fetch::fsm::DecodeError> for GetError {
+    fn from(value: crate::fetch::fsm::DecodeError) -> Self {
+        use crate::fetch::fsm::DecodeError::*;
 
         match value {
             e @ NotFound => GetError::NotFound(e.into()),