Merge 5edec02 into 57112e6

Author: rklaehn

Cargo.toml

@@ -96,7 +96,7 @@ futures-util = "0.3.30"
 testdir = "0.9.1"
 
 [features]
-default = ["fs-store", "net_protocol"]
+default = ["fs-store", "net_protocol", "formats-collection"]
 downloader = ["dep:parking_lot", "tokio-util/time", "dep:hashlink"]
 net_protocol = ["downloader", "dep:futures-util"]
 fs-store = ["dep:reflink-copy", "redb", "dep:tempfile"]
@@ -114,6 +114,8 @@ rpc = [
   "dep:ssh-key",
   "downloader",
 ]
+formats = []
+formats-collection = ["formats"]
 
 example-iroh = [
   "dep:clap",
@@ -129,6 +131,7 @@ rustdoc-args = ["--cfg", "iroh_docsrs"]
 
 [[example]]
 name = "provide-bytes"
+required-features = ["formats-collection"]
 
 [[example]]
 name = "fetch-fsm"

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,10 @@ mod progress {
         ProgressStyle,
     };
     use iroh_blobs::{
-        get::{db::DownloadProgress, progress::BlobProgress, Stats},
+        fetch::{
+            progress::{BlobProgress, DownloadProgress},
+            Stats,
+        },
         Hash,
     };
 

src/cli.rs

@@ -19,8 +19,10 @@ use iroh::{NodeAddr, PublicKey, RelayUrl};
 use tokio::io::AsyncWriteExt;
 
 use crate::{
-    get::{db::DownloadProgress, progress::BlobProgress, Stats},
-    net_protocol::DownloadMode,
+    fetch::{
+        progress::{BlobProgress, DownloadProgress},
+        Stats,
+    },
     provider::AddProgress,
     rpc::client::blobs::{
         self, BlobInfo, BlobStatus, CollectionInfo, DownloadOptions, IncompleteBlobInfo, WrapOption,

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::{progress::DownloadProgress, Stats},
     metrics::Metrics,
     store::Store,
     util::{local_pool::LocalPoolHandle, progress::ProgressSender},

src/downloader/get.rs

@@ -7,20 +7,20 @@ use iroh::endpoint;
 
 use super::{progress::BroadcastProgressSender, DownloadKind, FailureAction, GetStartFut, Getter};
 use crate::{
-    get::{db::get_to_db_in_steps, error::GetError},
-    store::Store,
+    fetch::Error,
+    store::{fetch_to_db_in_steps, FetchState, FetchStateNeedsConn, Store},
 };
 
-impl From<GetError> for FailureAction {
-    fn from(e: GetError) -> Self {
+impl From<Error> for FailureAction {
+    fn from(e: Error) -> Self {
         match e {
-            e @ GetError::NotFound(_) => FailureAction::AbortRequest(e.into()),
-            e @ GetError::RemoteReset(_) => FailureAction::RetryLater(e.into()),
-            e @ GetError::NoncompliantNode(_) => FailureAction::DropPeer(e.into()),
-            e @ GetError::Io(_) => FailureAction::RetryLater(e.into()),
-            e @ GetError::BadRequest(_) => FailureAction::AbortRequest(e.into()),
+            e @ Error::NotFound(_) => FailureAction::AbortRequest(e.into()),
+            e @ Error::RemoteReset(_) => FailureAction::RetryLater(e.into()),
+            e @ Error::NoncompliantNode(_) => FailureAction::DropPeer(e.into()),
+            e @ Error::Io(_) => FailureAction::RetryLater(e.into()),
+            e @ Error::BadRequest(_) => FailureAction::AbortRequest(e.into()),
             // TODO: what do we want to do on local failures?
-            e @ GetError::LocalFailure(_) => FailureAction::AbortRequest(e.into()),
+            e @ Error::LocalFailure(_) => FailureAction::AbortRequest(e.into()),
         }
     }
 }
@@ -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 = FetchStateNeedsConn;
 
     fn get(
         &mut self,
@@ -43,12 +43,10 @@ impl<S: Store> Getter for IoGetter<S> {
     ) -> GetStartFut<Self::NeedsConn> {
         let store = self.store.clone();
         async move {
-            match get_to_db_in_steps(store, kind.hash_and_format(), progress_sender).await {
+            match fetch_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(super::GetOutput::Complete(stats))
-                }
-                Ok(crate::get::db::GetState::NeedsConn(needs_conn)) => {
+                Ok(FetchState::Complete(stats)) => Ok(super::GetOutput::Complete(stats)),
+                Ok(FetchState::NeedsConn(needs_conn)) => {
                     Ok(super::GetOutput::NeedsConn(needs_conn))
                 }
             }
@@ -57,7 +55,7 @@ impl<S: Store> Getter for IoGetter<S> {
     }
 }
 
-impl super::NeedsConn<endpoint::Connection> for crate::get::db::GetStateNeedsConn {
+impl super::NeedsConn<endpoint::Connection> for FetchStateNeedsConn {
     fn proceed(self, conn: endpoint::Connection) -> super::GetProceedFut {
         async move {
             let res = self.proceed(conn).await;
@@ -73,13 +71,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, Error>) {
     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,
@@ -90,7 +88,7 @@ fn track_metrics(res: &Result<crate::get::Stats, GetError>) {
             inc_by!(Metrics, download_time_total, elapsed.as_millis() as u64);
         }
         Err(e) => match &e {
-            GetError::NotFound(_) => inc!(Metrics, downloads_notfound),
+            Error::NotFound(_) => inc!(Metrics, downloads_notfound),
             _ => inc!(Metrics, downloads_error),
         },
     }

src/downloader/progress.rs

@@ -11,7 +11,7 @@ use parking_lot::Mutex;
 
 use super::DownloadKind;
 use crate::{
-    get::{db::DownloadProgress, progress::TransferState},
+    fetch::progress::{DownloadProgress, TransferState},
     util::progress::{AsyncChannelProgressSender, IdGenerator, ProgressSendError, ProgressSender},
 };
 
@@ -21,7 +21,7 @@ pub type ProgressSubscriber = AsyncChannelProgressSender<DownloadProgress>;
 /// Track the progress of downloads.
 ///
 /// This struct allows to create [`ProgressSender`] structs to be passed to
-/// [`crate::get::db::get_to_db`]. Each progress sender can be subscribed to by any number of
+/// [`crate::store::fetch_to_db`]. Each progress sender can be subscribed to by any number of
 /// [`ProgressSubscriber`] channel senders, which will receive each progress update (if they have
 /// capacity). Additionally, the [`ProgressTracker`] maintains a [`TransferState`] for each
 /// transfer, applying each progress update to update this state. When subscribing to an already

src/downloader/test.rs

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

src/downloader/test/getter.rs

@@ -36,14 +36,14 @@ impl Getter for TestingGetter {
     // since for testing we don't need a real connection, just keep track of what peer is the
     // request being sent to
     type Connection = NodeId;
-    type NeedsConn = GetStateNeedsConn;
+    type NeedsConn = FetchStateNeedsConn;
 
     fn get(
         &mut self,
         kind: DownloadKind,
         progress_sender: BroadcastProgressSender,
     ) -> GetStartFut<Self::NeedsConn> {
-        std::future::ready(Ok(downloader::GetOutput::NeedsConn(GetStateNeedsConn(
+        std::future::ready(Ok(downloader::GetOutput::NeedsConn(FetchStateNeedsConn(
             self.clone(),
             kind,
             progress_sender,
@@ -53,11 +53,11 @@ impl Getter for TestingGetter {
 }
 
 #[derive(Debug)]
-pub(super) struct GetStateNeedsConn(TestingGetter, DownloadKind, BroadcastProgressSender);
+pub(super) struct FetchStateNeedsConn(TestingGetter, DownloadKind, BroadcastProgressSender);
 
-impl downloader::NeedsConn<NodeId> for GetStateNeedsConn {
+impl downloader::NeedsConn<NodeId> for FetchStateNeedsConn {
     fn proceed(self, peer: NodeId) -> super::GetProceedFut {
-        let GetStateNeedsConn(getter, kind, progress_sender) = self;
+        let FetchStateNeedsConn(getter, kind, progress_sender) = self;
         let mut inner = getter.0.write();
         inner.request_history.push((kind, peer));
         let request_duration = inner.request_duration;