Merge branch 'entity-manager' into lightweight-watcher

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" }
 atomic_refcell = "0.1.13"
 
@@ -60,15 +60,9 @@ tracing-subscriber = { version = "0.3.19", features = ["fmt"] }
 tracing-test = "0.2.5"
 walkdir = "2.5.0"
 atomic_refcell = "0.1.13"
-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

@@ -0,0 +1,186 @@
+//! This example shows how to create tags that expire after a certain time.
+//!
+//! We use a prefix so we can distinguish between expiring and normal tags, and
+//! then encode the expiry date in the tag name after the prefix, in a format
+//! that sorts in the same order as the expiry date.
+//!
+//! The example creates a number of blobs and protects them directly or indirectly
+//! with expiring tags. Watch as the expired tags are deleted and the blobs
+//! are removed from the store.
+use std::{
+    ops::Deref,
+    time::{Duration, SystemTime},
+};
+
+use chrono::Utc;
+use futures_lite::StreamExt;
+use iroh_blobs::{
+    api::{blobs::AddBytesOptions, Store, Tag},
+    hashseq::HashSeq,
+    store::fs::options::{BatchOptions, GcConfig, InlineOptions, Options, PathOptions},
+    BlobFormat, Hash,
+};
+use tokio::signal::ctrl_c;
+
+/// Using an iroh rpc client, create a tag that is marked to expire at `expiry` for all the given hashes.
+///
+/// The tag name will be `prefix`- followed by the expiry date in iso8601 format (e.g. `expiry-2025-01-01T12:00:00Z`).
+async fn create_expiring_tag(
+    store: &Store,
+    hashes: &[Hash],
+    prefix: &str,
+    expiry: SystemTime,
+) -> anyhow::Result<()> {
+    let expiry = chrono::DateTime::<chrono::Utc>::from(expiry);
+    let expiry = expiry.to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
+    let tagname = format!("{prefix}-{expiry}");
+    if hashes.is_empty() {
+        return Ok(());
+    } else if hashes.len() == 1 {
+        let hash = hashes[0];
+        store.tags().set(&tagname, hash).await?;
+    } else {
+        let hs = hashes.iter().copied().collect::<HashSeq>();
+        store
+            .add_bytes_with_opts(AddBytesOptions {
+                data: hs.into(),
+                format: BlobFormat::HashSeq,
+            })
+            .with_named_tag(&tagname)
+            .await?;
+    };
+    println!("Created tag {tagname}");
+    Ok(())
+}
+
+async fn delete_expired_tags(blobs: &Store, prefix: &str, bulk: bool) -> anyhow::Result<()> {
+    let prefix = format!("{prefix}-");
+    let now = chrono::Utc::now();
+    let end = format!(
+        "{}-{}",
+        prefix,
+        now.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
+    );
+    if bulk {
+        // delete all tags with the prefix and an expiry date before now
+        //
+        // this should be very efficient, since it is just a single database operation
+        blobs
+            .tags()
+            .delete_range(Tag::from(prefix.clone())..Tag::from(end))
+            .await?;
+    } else {
+        // find tags to delete one by one and then delete them
+        //
+        // this allows us to print the tags before deleting them
+        let mut tags = blobs.tags().list().await?;
+        let mut to_delete = Vec::new();
+        while let Some(tag) = tags.next().await {
+            let tag = tag?.name;
+            if let Some(rest) = tag.0.strip_prefix(prefix.as_bytes()) {
+                let Ok(expiry) = std::str::from_utf8(rest) else {
+                    tracing::warn!("Tag {} does have non utf8 expiry", tag);
+                    continue;
+                };
+                let Ok(expiry) = chrono::DateTime::parse_from_rfc3339(expiry) else {
+                    tracing::warn!("Tag {} does have invalid expiry date", tag);
+                    continue;
+                };
+                let expiry = expiry.with_timezone(&Utc);
+                if expiry < now {
+                    to_delete.push(tag);
+                }
+            }
+        }
+        for tag in to_delete {
+            println!("Deleting expired tag {tag}\n");
+            blobs.tags().delete(tag).await?;
+        }
+    }
+    Ok(())
+}
+
+async fn print_store_info(store: &Store) -> anyhow::Result<()> {
+    let now = chrono::Utc::now();
+    let mut tags = store.tags().list().await?;
+    println!(
+        "Current time: {}",
+        now.to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
+    );
+    println!("Tags:");
+    while let Some(tag) = tags.next().await {
+        let tag = tag?;
+        println!("  {tag:?}");
+    }
+    let mut blobs = store.list().stream().await?;
+    println!("Blobs:");
+    while let Some(item) = blobs.next().await {
+        println!("  {}", item?);
+    }
+    println!();
+    Ok(())
+}
+
+async fn info_task(store: Store) -> anyhow::Result<()> {
+    tokio::time::sleep(Duration::from_secs(1)).await;
+    loop {
+        print_store_info(&store).await?;
+        tokio::time::sleep(Duration::from_secs(5)).await;
+    }
+}
+
+async fn delete_expired_tags_task(store: Store, prefix: &str) -> anyhow::Result<()> {
+    loop {
+        delete_expired_tags(&store, prefix, false).await?;
+        tokio::time::sleep(Duration::from_secs(5)).await;
+    }
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt::init();
+    let path = std::env::current_dir()?.join("blobs");
+    let options = Options {
+        path: PathOptions::new(&path),
+        gc: Some(GcConfig {
+            add_protected: None,
+            interval: Duration::from_secs(10),
+        }),
+        inline: InlineOptions::default(),
+        batch: BatchOptions::default(),
+    };
+    let store =
+        iroh_blobs::store::fs::FsStore::load_with_opts(path.join("blobs.db"), options).await?;
+
+    // setup: add some data and tag it
+    {
+        // add several blobs and tag them with an expiry date 10 seconds in the future
+        let batch = store.batch().await?;
+        let a = batch.add_bytes("blob 1".as_bytes()).await?;
+        let b = batch.add_bytes("blob 2".as_bytes()).await?;
+
+        let expires_at = SystemTime::now()
+            .checked_add(Duration::from_secs(10))
+            .unwrap();
+        create_expiring_tag(&store, &[*a.hash(), *b.hash()], "expiring", expires_at).await?;
+
+        // add a single blob and tag it with an expiry date 60 seconds in the future
+        let c = batch.add_bytes("blob 3".as_bytes()).await?;
+        let expires_at = SystemTime::now()
+            .checked_add(Duration::from_secs(60))
+            .unwrap();
+        create_expiring_tag(&store, &[*c.hash()], "expiring", expires_at).await?;
+        // batch goes out of scope, so data is only protected by the tags we created
+    }
+
+    // delete expired tags every 5 seconds
+    let delete_task = tokio::spawn(delete_expired_tags_task(store.deref().clone(), "expiring"));
+    // print all tags and blobs every 5 seconds
+    let info_task = tokio::spawn(info_task(store.deref().clone()));
+
+    ctrl_c().await?;
+    delete_task.abort();
+    info_task.abort();
+    store.shutdown().await?;
+    Ok(())
+}

examples/expiring-tags.rs

@@ -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

@@ -1266,7 +1266,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 {