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()),

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

@@ -8,7 +8,7 @@ use iroh_io::AsyncSliceReaderExt;
 use serde::{Deserialize, Serialize};
 
 use crate::{
-    get::{fsm, Stats},
+    fetch::{fsm, Stats},
     hashseq::HashSeq,
     store::MapEntry,
     util::TempTag,
@@ -142,7 +142,7 @@ impl Collection {
     ///
     /// Returns the collection, a map from blob offsets to bytes, and the stats.
     pub async fn read_fsm_all(
-        fsm_at_start_root: crate::get::fsm::AtStartRoot,
+        fsm_at_start_root: crate::fetch::fsm::AtStartRoot,
     ) -> anyhow::Result<(Collection, BTreeMap<u64, Bytes>, Stats)> {
         let (next, links, collection) = Self::read_fsm(fsm_at_start_root).await?;
         let mut res = BTreeMap::new();

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

@@ -34,7 +34,7 @@ pub mod cli;
 pub mod downloader;
 pub mod export;
 pub mod format;
-pub mod get;
+pub mod fetch;
 pub mod hashseq;
 pub mod metrics;
 #[cfg(feature = "net_protocol")]

src/format/collection.rs

@@ -45,7 +45,7 @@ use crate::{
     downloader::{DownloadRequest, Downloader},
     export::ExportProgress,
     format::collection::Collection,
-    get::{
+    fetch::{
         db::{DownloadProgress, GetState},
         Stats,
     },
@@ -1009,7 +1009,7 @@ impl<D: crate::store::Store> Handler<D> {
         let mut remaining_nodes = nodes.len();
         let mut nodes_iter = nodes.into_iter();
         'outer: loop {
-            match crate::get::db::get_to_db_in_steps(
+            match crate::fetch::db::get_to_db_in_steps(
                 self.store().clone(),
                 hash_and_format,
                 progress.clone(),

src/lib.rs

@@ -85,7 +85,7 @@ pub use crate::net_protocol::DownloadMode;
 use crate::{
     export::ExportProgress as BytesExportProgress,
     format::collection::{Collection, SimpleStore},
-    get::db::DownloadProgress as BytesDownloadProgress,
+    fetch::db::DownloadProgress as BytesDownloadProgress,
     net_protocol::BlobDownloadRequest,
     rpc::proto::RpcService,
     store::{BaoBlobSize, ConsistencyCheckProgress, ExportFormat, ExportMode, ValidateProgress},
@@ -675,7 +675,7 @@ pub struct DownloadOutcome {
     /// The size of the data we downloaded from the network
     pub downloaded_size: u64,
     /// Statistics about the download
-    pub stats: crate::get::Stats,
+    pub stats: crate::fetch::Stats,
 }
 
 /// Progress stream for blob download operations.

src/rpc.rs

@@ -10,7 +10,7 @@ use super::{RpcError, RpcResult, RpcService};
 use crate::{
     export::ExportProgress,
     format::collection::Collection,
-    get::db::DownloadProgress,
+    fetch::db::DownloadProgress,
     net_protocol::{BatchId, BlobDownloadRequest},
     provider::{AddProgress, BatchAddPathProgress},
     rpc::client::blobs::{BlobInfo, BlobStatus, IncompleteBlobInfo, ReadAtLen, WrapOption},