WIP

Author: rklaehn

Cargo.toml

@@ -95,7 +95,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"]
@@ -112,6 +112,8 @@ rpc = [
     "dep:walkdir",
     "downloader",
 ]
+formats = []
+formats-collection = ["formats"]
 
 example-iroh = [
   "dep:clap",

examples/local-swarm-discovery.rs

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

src/cli.rs

@@ -19,7 +19,10 @@ use iroh::{NodeAddr, PublicKey, RelayUrl};
 use tokio::io::AsyncWriteExt;
 
 use crate::{
-    fetch::{db::DownloadProgress, progress::BlobProgress, Stats},
+    fetch::{
+        progress::{BlobProgress, DownloadProgress},
+        Stats,
+    },
     net_protocol::DownloadMode,
     provider::AddProgress,
     rpc::client::blobs::{

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

src/downloader/get.rs

@@ -7,8 +7,8 @@ use iroh::endpoint;
 
 use super::{progress::BroadcastProgressSender, DownloadKind, FailureAction, GetStartFut, Getter};
 use crate::{
-    fetch::{db::fetch_to_db_in_steps, Error},
-    store::Store,
+    fetch::Error,
+    store::{fetch_to_db_in_steps, FetchState, FetchStateNeedsConn, Store},
 };
 
 impl From<Error> for FailureAction {
@@ -34,7 +34,7 @@ pub(crate) struct IoGetter<S: Store> {
 
 impl<S: Store> Getter for IoGetter<S> {
     type Connection = endpoint::Connection;
-    type NeedsConn = crate::fetch::db::FetchStateNeedsConn;
+    type NeedsConn = FetchStateNeedsConn;
 
     fn get(
         &mut self,
@@ -45,10 +45,8 @@ impl<S: Store> Getter for IoGetter<S> {
         async move {
             match fetch_to_db_in_steps(store, kind.hash_and_format(), progress_sender).await {
                 Err(err) => Err(err.into()),
-                Ok(crate::fetch::db::FetchState::Complete(stats)) => {
-                    Ok(super::GetOutput::Complete(stats))
-                }
-                Ok(crate::fetch::db::FetchState::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::fetch::db::FetchStateNeedsConn {
+impl super::NeedsConn<endpoint::Connection> for FetchStateNeedsConn {
     fn proceed(self, conn: endpoint::Connection) -> super::GetProceedFut {
         async move {
             let res = self.proceed(conn).await;

src/downloader/progress.rs

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

src/downloader/test.rs

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

src/fetch.rs

@@ -31,7 +31,6 @@ use crate::{
 
 mod error;
 pub use error::Error;
-pub mod db;
 pub mod progress;
 pub mod request;
 

src/fetch/progress.rs

@@ -5,7 +5,7 @@ use std::{collections::HashMap, num::NonZeroU64};
 use serde::{Deserialize, Serialize};
 use tracing::warn;
 
-use super::db::{BlobId, DownloadProgress};
+use super::Stats;
 use crate::{protocol::RangeSpec, store::BaoBlobSize, Hash};
 
 /// The identifier for progress events.
@@ -183,3 +183,90 @@ impl TransferState {
         }
     }
 }
+
+/// Progress updates for the get operation.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum DownloadProgress {
+    /// Initial state if subscribing to a running or queued transfer.
+    InitialState(TransferState),
+    /// Data was found locally.
+    FoundLocal {
+        /// child offset
+        child: BlobId,
+        /// The hash of the entry.
+        hash: Hash,
+        /// The size of the entry in bytes.
+        size: BaoBlobSize,
+        /// The ranges that are available locally.
+        valid_ranges: RangeSpec,
+    },
+    /// A new connection was established.
+    Connected,
+    /// An item was found with hash `hash`, from now on referred to via `id`.
+    Found {
+        /// A new unique progress id for this entry.
+        id: u64,
+        /// Identifier for this blob within this download.
+        ///
+        /// Will always be [`BlobId::Root`] unless a hashseq is downloaded, in which case this
+        /// allows to identify the children by their offset in the hashseq.
+        child: BlobId,
+        /// The hash of the entry.
+        hash: Hash,
+        /// The size of the entry in bytes.
+        size: u64,
+    },
+    /// An item was found with hash `hash`, from now on referred to via `id`.
+    FoundHashSeq {
+        /// The name of the entry.
+        hash: Hash,
+        /// Number of children in the collection, if known.
+        children: u64,
+    },
+    /// We got progress ingesting item `id`.
+    Progress {
+        /// The unique id of the entry.
+        id: u64,
+        /// The offset of the progress, in bytes.
+        offset: u64,
+    },
+    /// We are done with `id`.
+    Done {
+        /// The unique id of the entry.
+        id: u64,
+    },
+    /// All operations finished.
+    ///
+    /// This will be the last message in the stream.
+    AllDone(Stats),
+    /// We got an error and need to abort.
+    ///
+    /// This will be the last message in the stream.
+    Abort(serde_error::Error),
+}
+
+/// The id of a blob in a transfer
+#[derive(
+    Debug, Copy, Clone, Ord, PartialOrd, Eq, PartialEq, std::hash::Hash, Serialize, Deserialize,
+)]
+pub enum BlobId {
+    /// The root blob (child id 0)
+    Root,
+    /// A child blob (child id > 0)
+    Child(NonZeroU64),
+}
+
+impl BlobId {
+    pub(crate) fn from_offset(id: u64) -> Self {
+        NonZeroU64::new(id).map(Self::Child).unwrap_or(Self::Root)
+    }
+}
+
+impl From<BlobId> for u64 {
+    fn from(value: BlobId) -> Self {
+        match value {
+            BlobId::Root => 0,
+            BlobId::Child(id) => id.into(),
+        }
+    }
+}

src/format.rs

@@ -13,4 +13,5 @@
 //! n-1 items, where n is the number of blobs in the HashSeq.
 //!
 //! [postcard]: https://docs.rs/postcard/latest/postcard/
+#[cfg(feature = "formats-collection")]
 pub mod collection;

src/lib.rs

@@ -32,8 +32,8 @@
 pub mod cli;
 #[cfg(feature = "downloader")]
 pub mod downloader;
-pub mod export;
 pub mod fetch;
+#[cfg(feature = "formats")]
 pub mod format;
 pub mod hashseq;
 pub mod metrics;

src/rpc.rs

@@ -43,15 +43,14 @@ use tracing::{debug, warn};
 
 use crate::{
     downloader::{DownloadRequest, Downloader},
-    export::ExportProgress,
-    fetch::{
-        db::{DownloadProgress, FetchState},
-        Stats,
-    },
+    fetch::{progress::DownloadProgress, Stats},
     format::collection::Collection,
     net_protocol::{BlobDownloadRequest, Blobs, BlobsInner},
     provider::{AddProgress, BatchAddPathProgress},
-    store::{ConsistencyCheckProgress, ImportProgress, MapEntry, ValidateProgress},
+    store::{
+        ConsistencyCheckProgress, ExportProgress, FetchState, ImportProgress, MapEntry,
+        ValidateProgress,
+    },
     util::{
         local_pool::LocalPoolHandle,
         progress::{AsyncChannelProgressSender, ProgressSender},
@@ -451,7 +450,7 @@ impl<D: crate::store::Store> Handler<D> {
         let progress = AsyncChannelProgressSender::new(tx);
         let rt = self.rt().clone();
         rt.spawn_detached(move || async move {
-            let res = crate::export::export(
+            let res = crate::store::export(
                 self.store(),
                 msg.hash,
                 msg.path,
@@ -1009,7 +1008,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::fetch::db::fetch_to_db_in_steps(
+            match crate::store::fetch_to_db_in_steps(
                 self.store().clone(),
                 hash_and_format,
                 progress.clone(),

src/rpc/client/blobs.rs

@@ -83,11 +83,11 @@ use tracing::warn;
 
 pub use crate::net_protocol::DownloadMode;
 use crate::{
-    export::ExportProgress as BytesExportProgress,
-    fetch::db::DownloadProgress as BytesDownloadProgress,
+    fetch::progress::DownloadProgress as BytesDownloadProgress,
     format::collection::{Collection, SimpleStore},
     net_protocol::BlobDownloadRequest,
     rpc::proto::RpcService,
+    store::ExportProgress as BytesExportProgress,
     store::{BaoBlobSize, ConsistencyCheckProgress, ExportFormat, ExportMode, ValidateProgress},
     util::SetTagOption,
     BlobFormat, Hash, Tag,

src/rpc/proto/blobs.rs

@@ -8,12 +8,11 @@ use serde::{Deserialize, Serialize};
 
 use super::{RpcError, RpcResult, RpcService};
 use crate::{
-    export::ExportProgress,
-    fetch::db::DownloadProgress,
     format::collection::Collection,
     net_protocol::{BatchId, BlobDownloadRequest},
     provider::{AddProgress, BatchAddPathProgress},
     rpc::client::blobs::{BlobInfo, BlobStatus, IncompleteBlobInfo, ReadAtLen, WrapOption},
+    store::ExportProgress,
     store::{
         BaoBlobSize, ConsistencyCheckProgress, ExportFormat, ExportMode, ImportMode,
         ValidateProgress,
@@ -111,7 +110,7 @@ pub struct AddPathResponse(pub AddProgress);
 
 /// Progress response for [`BlobDownloadRequest`]
 #[derive(Debug, Clone, Serialize, Deserialize, derive_more::From, derive_more::Into)]
-pub struct DownloadResponse(pub DownloadProgress);
+pub struct DownloadResponse(pub crate::fetch::progress::DownloadProgress);
 
 /// A request to the node to download and share the data specified by the hash.
 #[derive(Debug, Clone, Serialize, Deserialize)]

src/store.rs

@@ -10,6 +10,11 @@ pub mod readonly_mem;
 #[cfg(feature = "fs-store")]
 pub mod fs;
 
+mod fetch_to_db;
+pub use fetch_to_db::{fetch_to_db, fetch_to_db_in_steps, FetchState, FetchStateNeedsConn};
+mod export;
+pub use export::{export, export_blob, export_collection, ExportProgress};
+
 mod traits;
 use tracing::warn;
 pub use traits::*;

src/export.rs renamed to src/store/export.rs

@@ -1,6 +1,6 @@
 //! Functions that use the iroh-blobs protocol in conjunction with a bao store.
 
-use std::{future::Future, io, num::NonZeroU64, pin::Pin};
+use std::{future::Future, io, pin::Pin};
 
 use anyhow::anyhow;
 use bao_tree::{ChunkNum, ChunkRanges};
@@ -11,17 +11,15 @@ use genawaiter::{
 };
 use iroh::endpoint::Connection;
 use iroh_io::AsyncSliceReader;
-use serde::{Deserialize, Serialize};
 use tokio::sync::oneshot;
 use tracing::trace;
 
 use crate::{
     fetch::{
         self,
-        error::Error,
         fsm::{AtBlobHeader, AtEndBlob, ConnectedNext, EndBlobNext},
-        progress::TransferState,
-        Stats,
+        progress::{BlobId, DownloadProgress},
+        Error, Stats,
     },
     hashseq::parse_hash_seq,
     protocol::{GetRequest, RangeSpec, RangeSpecSeq},
@@ -608,91 +606,3 @@ impl<D: BaoStore> BlobInfo<D> {
         }
     }
 }
-
-/// Progress updates for the get operation.
-// TODO: Move to super::progress
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub enum DownloadProgress {
-    /// Initial state if subscribing to a running or queued transfer.
-    InitialState(TransferState),
-    /// Data was found locally.
-    FoundLocal {
-        /// child offset
-        child: BlobId,
-        /// The hash of the entry.
-        hash: Hash,
-        /// The size of the entry in bytes.
-        size: BaoBlobSize,
-        /// The ranges that are available locally.
-        valid_ranges: RangeSpec,
-    },
-    /// A new connection was established.
-    Connected,
-    /// An item was found with hash `hash`, from now on referred to via `id`.
-    Found {
-        /// A new unique progress id for this entry.
-        id: u64,
-        /// Identifier for this blob within this download.
-        ///
-        /// Will always be [`BlobId::Root`] unless a hashseq is downloaded, in which case this
-        /// allows to identify the children by their offset in the hashseq.
-        child: BlobId,
-        /// The hash of the entry.
-        hash: Hash,
-        /// The size of the entry in bytes.
-        size: u64,
-    },
-    /// An item was found with hash `hash`, from now on referred to via `id`.
-    FoundHashSeq {
-        /// The name of the entry.
-        hash: Hash,
-        /// Number of children in the collection, if known.
-        children: u64,
-    },
-    /// We got progress ingesting item `id`.
-    Progress {
-        /// The unique id of the entry.
-        id: u64,
-        /// The offset of the progress, in bytes.
-        offset: u64,
-    },
-    /// We are done with `id`.
-    Done {
-        /// The unique id of the entry.
-        id: u64,
-    },
-    /// All operations finished.
-    ///
-    /// This will be the last message in the stream.
-    AllDone(Stats),
-    /// We got an error and need to abort.
-    ///
-    /// This will be the last message in the stream.
-    Abort(serde_error::Error),
-}
-
-/// The id of a blob in a transfer
-#[derive(
-    Debug, Copy, Clone, Ord, PartialOrd, Eq, PartialEq, std::hash::Hash, Serialize, Deserialize,
-)]
-pub enum BlobId {
-    /// The root blob (child id 0)
-    Root,
-    /// A child blob (child id > 0)
-    Child(NonZeroU64),
-}
-
-impl BlobId {
-    fn from_offset(id: u64) -> Self {
-        NonZeroU64::new(id).map(Self::Child).unwrap_or(Self::Root)
-    }
-}
-
-impl From<BlobId> for u64 {
-    fn from(value: BlobId) -> Self {
-        match value {
-            BlobId::Root => 0,
-            BlobId::Child(id) => id.into(),
-        }
-    }
-}

src/fetch/db.rs renamed to src/store/fetch_to_db.rs

@@ -822,28 +822,6 @@ pub enum ExportFormat {
     Collection,
 }
 
-#[allow(missing_docs)]
-#[derive(Debug)]
-pub enum ExportProgress {
-    /// Starting to export to a file
-    ///
-    /// This will be the first message for an id
-    Start {
-        id: u64,
-        hash: Hash,
-        path: PathBuf,
-        stable: bool,
-    },
-    /// Progress when copying the file to the target
-    ///
-    /// This will be omitted if the store can move the file or use copy on write
-    ///
-    /// There will be multiple of these messages for an id
-    Progress { id: u64, offset: u64 },
-    /// Done exporting
-    Done { id: u64 },
-}
-
 /// Level for generic validation messages
 #[derive(
     Debug, Clone, Copy, derive_more::Display, Serialize, Deserialize, PartialOrd, Ord, PartialEq, Eq,