Merge branch 'main' into expiring-tags-example

Author: rklaehn

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "iroh-blobs"
-version = "0.91.0"
+version = "0.93.0"
 edition = "2021"
 description = "content-addressed blobs for iroh"
 license = "MIT OR Apache-2.0"
@@ -37,12 +37,12 @@ chrono = "0.4.39"
 nested_enum_utils = "0.2.1"
 ref-cast = "1.0.24"
 arrayvec = "0.7.6"
-iroh = "0.90"
+iroh = "0.91"
 self_cell = "1.1.0"
 genawaiter = { version = "0.99.1", features = ["futures03"] }
-iroh-base = "0.90"
+iroh-base = "0.91"
 reflink-copy = "0.1.24"
-irpc = { version = "0.5.0", features = ["rpc", "quinn_endpoint_setup", "message_spans", "stream", "derive"], default-features = false }
+irpc = { version = "0.7.0", features = ["rpc", "quinn_endpoint_setup", "spans", "stream", "derive"], default-features = false }
 iroh-metrics = { version = "0.35" }
 
 [dev-dependencies]
@@ -58,15 +58,9 @@ testresult = "0.4.1"
 tracing-subscriber = { version = "0.3.19", features = ["fmt"] }
 tracing-test = "0.2.5"
 walkdir = "2.5.0"
-iroh = { version = "0.90", features = ["discovery-local-network"]}
+iroh = { version = "0.91", features = ["discovery-local-network"]}
 
 [features]
 hide-proto-docs = []
 metrics = []
 default = ["hide-proto-docs"]
-
-[patch.crates-io]
-iroh = { git = "https://github.com/n0-computer/iroh.git", branch = "main" }
-iroh-base = { git = "https://github.com/n0-computer/iroh.git", branch = "main" }
-irpc = { git = "https://github.com/n0-computer/irpc.git", branch = "main" }
-irpc-derive = { git = "https://github.com/n0-computer/irpc.git", branch = "main" }

Cargo.toml

@@ -39,9 +39,3 @@ name = "ring"
 [[licenses.clarify.license-files]]
 hash = 3171872035
 path = "LICENSE"
-
-[sources]
-allow-git = [
-    "https://github.com/n0-computer/irpc.git",
-    "https://github.com/n0-computer/iroh.git",
-]

deny.toml

@@ -241,7 +241,7 @@ async fn provide(args: ProvideArgs) -> anyhow::Result<()> {
     let router = iroh::protocol::Router::builder(endpoint.clone())
         .accept(iroh_blobs::ALPN, blobs)
         .spawn();
-    let addr = router.endpoint().node_addr().initialized().await?;
+    let addr = router.endpoint().node_addr().initialized().await;
     let ticket = NodeTicket::from(addr.clone());
     println!("Node address: {addr:?}");
     println!("ticket:\n{ticket}");

examples/random_store.rs

@@ -12,11 +12,11 @@
 //!
 //! You can also [`connect`](Store::connect) to a remote store that is listening
 //! to rpc requests.
-use std::{io, net::SocketAddr, ops::Deref, sync::Arc};
+use std::{io, net::SocketAddr, ops::Deref};
 
 use bao_tree::io::EncodeError;
 use iroh::Endpoint;
-use irpc::rpc::{listen, Handler};
+use irpc::rpc::{listen, RemoteService};
 use n0_snafu::SpanTrace;
 use nested_enum_utils::common_fields;
 use proto::{Request, ShutdownRequest, SyncDbRequest};
@@ -32,7 +32,7 @@ pub mod remote;
 pub mod tags;
 pub use crate::{store::util::Tag, util::temp_tag::TempTag};
 
-pub(crate) type ApiClient = irpc::Client<proto::Command, proto::Request, proto::StoreService>;
+pub(crate) type ApiClient = irpc::Client<proto::Request>;
 
 #[common_fields({
     backtrace: Option<Backtrace>,
@@ -281,42 +281,8 @@ impl Store {
 
     /// Listen on a quinn endpoint for incoming rpc connections.
     pub async fn listen(self, endpoint: quinn::Endpoint) {
-        let local = self.client.local().unwrap().clone();
-        let handler: Handler<Request> = Arc::new(move |req, rx, tx| {
-            let local = local.clone();
-            Box::pin({
-                match req {
-                    Request::SetTag(msg) => local.send((msg, tx)),
-                    Request::CreateTag(msg) => local.send((msg, tx)),
-                    Request::DeleteTags(msg) => local.send((msg, tx)),
-                    Request::RenameTag(msg) => local.send((msg, tx)),
-                    Request::ListTags(msg) => local.send((msg, tx)),
-
-                    Request::ListTempTags(msg) => local.send((msg, tx)),
-                    Request::CreateTempTag(msg) => local.send((msg, tx)),
-
-                    Request::BlobStatus(msg) => local.send((msg, tx)),
-
-                    Request::ImportBytes(msg) => local.send((msg, tx)),
-                    Request::ImportByteStream(msg) => local.send((msg, tx, rx)),
-                    Request::ImportBao(msg) => local.send((msg, tx, rx)),
-                    Request::ImportPath(msg) => local.send((msg, tx)),
-                    Request::ListBlobs(msg) => local.send((msg, tx)),
-                    Request::DeleteBlobs(msg) => local.send((msg, tx)),
-                    Request::Batch(msg) => local.send((msg, tx, rx)),
-
-                    Request::ExportBao(msg) => local.send((msg, tx)),
-                    Request::ExportRanges(msg) => local.send((msg, tx)),
-                    Request::ExportPath(msg) => local.send((msg, tx)),
-
-                    Request::Observe(msg) => local.send((msg, tx)),
-
-                    Request::ClearProtected(msg) => local.send((msg, tx)),
-                    Request::SyncDb(msg) => local.send((msg, tx)),
-                    Request::Shutdown(msg) => local.send((msg, tx)),
-                }
-            })
-        });
+        let local = self.client.as_local().unwrap().clone();
+        let handler = Request::remote_handler(local);
         listen::<Request>(endpoint, handler).await
     }
 

src/api.rs

@@ -29,15 +29,10 @@ use crate::{
 
 #[derive(Debug, Clone)]
 pub struct Downloader {
-    client: irpc::Client<SwarmMsg, SwarmProtocol, DownloaderService>,
+    client: irpc::Client<SwarmProtocol>,
 }
 
-#[derive(Debug, Clone)]
-pub struct DownloaderService;
-
-impl irpc::Service for DownloaderService {}
-
-#[rpc_requests(DownloaderService, message = SwarmMsg, alias = "Msg")]
+#[rpc_requests(message = SwarmMsg, alias = "Msg")]
 #[derive(Debug, Serialize, Deserialize)]
 enum SwarmProtocol {
     #[rpc(tx = mpsc::Sender<DownloadProgessItem>)]
@@ -711,9 +706,9 @@ mod tests {
         let (r3, store3, _) = node_test_setup_fs(testdir.path().join("c")).await?;
         let tt1 = store1.add_slice("hello world").await?;
         let tt2 = store2.add_slice("hello world 2").await?;
-        let node1_addr = r1.endpoint().node_addr().initialized().await?;
+        let node1_addr = r1.endpoint().node_addr().initialized().await;
         let node1_id = node1_addr.node_id;
-        let node2_addr = r2.endpoint().node_addr().initialized().await?;
+        let node2_addr = r2.endpoint().node_addr().initialized().await;
         let node2_id = node2_addr.node_id;
         let swarm = Downloader::new(&store3, r3.endpoint());
         r3.endpoint().add_node_addr(node1_addr.clone())?;
@@ -750,9 +745,9 @@ mod tests {
                 format: crate::BlobFormat::HashSeq,
             })
             .await?;
-        let node1_addr = r1.endpoint().node_addr().initialized().await?;
+        let node1_addr = r1.endpoint().node_addr().initialized().await;
         let node1_id = node1_addr.node_id;
-        let node2_addr = r2.endpoint().node_addr().initialized().await?;
+        let node2_addr = r2.endpoint().node_addr().initialized().await;
         let node2_id = node2_addr.node_id;
         let swarm = Downloader::new(&store3, r3.endpoint());
         r3.endpoint().add_node_addr(node1_addr.clone())?;
@@ -819,9 +814,9 @@ mod tests {
                 format: crate::BlobFormat::HashSeq,
             })
             .await?;
-        let node1_addr = r1.endpoint().node_addr().initialized().await?;
+        let node1_addr = r1.endpoint().node_addr().initialized().await;
         let node1_id = node1_addr.node_id;
-        let node2_addr = r2.endpoint().node_addr().initialized().await?;
+        let node2_addr = r2.endpoint().node_addr().initialized().await;
         let node2_id = node2_addr.node_id;
         let swarm = Downloader::new(&store3, r3.endpoint());
         r3.endpoint().add_node_addr(node1_addr.clone())?;

src/api/downloader.rs

@@ -86,11 +86,7 @@ impl HashSpecific for CreateTagMsg {
     }
 }
 
-#[derive(Debug, Clone)]
-pub struct StoreService;
-impl irpc::Service for StoreService {}
-
-#[rpc_requests(StoreService, message = Command, alias = "Msg")]
+#[rpc_requests(message = Command, alias = "Msg")]
 #[derive(Debug, Serialize, Deserialize)]
 pub enum Request {
     #[rpc(tx = mpsc::Sender<super::Result<Hash>>)]

src/api/proto.rs

@@ -99,7 +99,7 @@ impl BlobsProtocol {
     /// just a convenience method to create a ticket from content and the address of this node.
     pub async fn ticket(&self, content: impl Into<HashAndFormat>) -> anyhow::Result<BlobTicket> {
         let content = content.into();
-        let addr = self.inner.endpoint.node_addr().initialized().await?;
+        let addr = self.inner.endpoint.node_addr().initialized().await;
         let ticket = BlobTicket::new(addr, content.hash, content.format);
         Ok(ticket)
     }

src/net_protocol.rs

@@ -1259,7 +1259,7 @@ impl AsRef<Store> for FsStore {
 
 impl FsStore {
     fn new(
-        sender: irpc::LocalSender<proto::Command, proto::StoreService>,
+        sender: irpc::LocalSender<proto::Request>,
         db: tokio::sync::mpsc::Sender<InternalCommand>,
     ) -> Self {
         Self {

src/store/fs.rs

@@ -1,9 +1,9 @@
-use std::collections::HashSet;
+use std::{collections::HashSet, pin::Pin, sync::Arc};
 
 use bao_tree::ChunkRanges;
 use genawaiter::sync::{Co, Gen};
 use n0_future::{Stream, StreamExt};
-use tracing::{debug, error, warn};
+use tracing::{debug, error, info, warn};
 
 use crate::{api::Store, Hash, HashAndFormat};
 
@@ -130,14 +130,52 @@ fn gc_sweep<'a>(
     })
 }
 
-#[derive(Debug, Clone)]
+/// Configuration for garbage collection.
+#[derive(derive_more::Debug, Clone)]
 pub struct GcConfig {
+    /// Interval in which to run garbage collection.
     pub interval: std::time::Duration,
+    /// Optional callback to manually add protected blobs.
+    ///
+    /// The callback is called before each garbage collection run. It gets a `&mut HashSet<Hash>`
+    /// and returns a future that returns [`ProtectOutcome`]. All hashes that are added to the
+    /// [`HashSet`] will be protected from garbage collection during this run.
+    ///
+    /// In normal operation, return [`ProtectOutcome::Continue`] from the callback. If you return
+    /// [`ProtectOutcome::Abort`], the garbage collection run will be aborted.Use this if your
+    /// source of hashes to protect returned an error, and thus garbage collection should be skipped
+    /// completely to not unintentionally delete blobs that should be protected.
+    #[debug("ProtectCallback")]
+    pub add_protected: Option<ProtectCb>,
 }
 
+/// Returned from [`ProtectCb`].
+///
+/// See [`GcConfig::add_protected] for details.
+#[derive(Debug)]
+pub enum ProtectOutcome {
+    /// Continue with the garbage collection run.
+    Continue,
+    /// Abort the garbage collection run.
+    Abort,
+}
+
+/// The type of the garbage collection callback.
+///
+/// See [`GcConfig::add_protected] for details.
+pub type ProtectCb = Arc<
+    dyn for<'a> Fn(
+            &'a mut HashSet<Hash>,
+        )
+            -> Pin<Box<dyn std::future::Future<Output = ProtectOutcome> + Send + Sync + 'a>>
+        + Send
+        + Sync
+        + 'static,
+>;
+
 pub async fn gc_run_once(store: &Store, live: &mut HashSet<Hash>) -> crate::api::Result<()> {
+    debug!(externally_protected = live.len(), "gc: start");
     {
-        live.clear();
         store.clear_protected().await?;
         let mut stream = gc_mark(store, live);
         while let Some(ev) = stream.next().await {
@@ -155,6 +193,7 @@ pub async fn gc_run_once(store: &Store, live: &mut HashSet<Hash>) -> crate::api:
             }
         }
     }
+    debug!(total_protected = live.len(), "gc: sweep");
     {
         let mut stream = gc_sweep(store, live);
         while let Some(ev) = stream.next().await {
@@ -172,14 +211,26 @@ pub async fn gc_run_once(store: &Store, live: &mut HashSet<Hash>) -> crate::api:
             }
         }
     }
+    debug!("gc: done");
 
     Ok(())
 }
 
 pub async fn run_gc(store: Store, config: GcConfig) {
+    debug!("gc enabled with interval {:?}", config.interval);
     let mut live = HashSet::new();
     loop {
+        live.clear();
         tokio::time::sleep(config.interval).await;
+        if let Some(ref cb) = config.add_protected {
+            match (cb)(&mut live).await {
+                ProtectOutcome::Continue => {}
+                ProtectOutcome::Abort => {
+                    info!("abort gc run: protect callback indicated abort");
+                    continue;
+                }
+            }
+        }
         if let Err(e) = gc_run_once(&store, &mut live).await {
             error!("error during gc run: {e}");
             break;
@@ -288,6 +339,7 @@ mod tests {
             assert!(!data_path.exists());
             assert!(!outboard_path.exists());
         }
+        live.clear();
         // create a large partial file and check that the data and outboard file as well as
         // the sizes and bitfield files are deleted by gc
         {