clippy

Author: rklaehn

examples/expiring-tags.rs

@@ -4,50 +4,57 @@
 //! 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.
 //!
-//! Then we can just use
-use std::time::{Duration, SystemTime};
+//! 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::endpoint;
 use iroh_blobs::{
-    hashseq::HashSeq, rpc::client::blobs::MemClient as BlobsClient, store::GcConfig, BlobFormat,
-    Hash, HashAndFormat, Tag,
+    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(
-    iroh: &BlobsClient,
+    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);
-    let batch = iroh.batch().await?;
-    let tt = if hashes.is_empty() {
+    let tagname = format!("{prefix}-{expiry}");
+    if hashes.is_empty() {
         return Ok(());
     } else if hashes.len() == 1 {
         let hash = hashes[0];
-        batch.temp_tag(HashAndFormat::raw(hash)).await?
+        store.tags().set(&tagname, hash).await?;
     } else {
         let hs = hashes.iter().copied().collect::<HashSeq>();
-        batch
-            .add_bytes_with_opts(hs.into_inner(), BlobFormat::HashSeq)
-            .await?
+        store
+            .add_bytes_with_opts(AddBytesOptions {
+                data: hs.into(),
+                format: BlobFormat::HashSeq,
+            })
+            .with_named_tag(&tagname)
+            .await?;
     };
-    batch.persist_to(tt, tagname.as_str().into()).await?;
-    println!("Created tag {}", tagname);
+    println!("Created tag {tagname}");
     Ok(())
 }
 
-async fn delete_expired_tags(blobs: &BlobsClient, prefix: &str, bulk: bool) -> anyhow::Result<()> {
-    let mut tags = blobs.tags().list().await?;
-    let prefix = format!("{}-", prefix);
+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!(
         "{}-{}",
@@ -66,6 +73,7 @@ async fn delete_expired_tags(blobs: &BlobsClient, prefix: &str, bulk: bool) -> a
         // 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;
@@ -85,99 +93,93 @@ async fn delete_expired_tags(blobs: &BlobsClient, prefix: &str, bulk: bool) -> a
             }
         }
         for tag in to_delete {
-            println!("Deleting expired tag {}", tag);
+            println!("Deleting expired tag {tag}\n");
             blobs.tags().delete(tag).await?;
         }
     }
     Ok(())
 }
 
-async fn info_task(blobs: BlobsClient) -> anyhow::Result<()> {
+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 {
-        let now = chrono::Utc::now();
-        let mut tags = blobs.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 = blobs.list().await?;
-        println!("Blobs:");
-        while let Some(info) = blobs.next().await {
-            let info = info?;
-            println!("  {} {} bytes", info.hash, info.size);
-        }
-        println!();
+        print_store_info(&store).await?;
         tokio::time::sleep(Duration::from_secs(5)).await;
     }
 }
 
-async fn delete_expired_tags_task(blobs: BlobsClient, prefix: &str) -> anyhow::Result<()> {
+async fn delete_expired_tags_task(store: Store, prefix: &str) -> anyhow::Result<()> {
     loop {
-        delete_expired_tags(&blobs, prefix, false).await?;
+        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 endpoint = endpoint::Endpoint::builder().bind().await?;
-    let store = iroh_blobs::store::fs::Store::load("blobs").await?;
-    let blobs = iroh_blobs::net_protocol::Blobs::builder(store).build(&endpoint);
-    // enable gc with a short period
-    blobs.start_gc(GcConfig {
-        period: Duration::from_secs(1),
-        done_callback: None,
-    })?;
-    // create a router and add blobs as a service
-    //
-    // You can skip this if you don't want to serve the data over the network.
-    let router = iroh::protocol::Router::builder(endpoint)
-        .accept(iroh_blobs::ALPN, blobs.clone())
-        .spawn()
-        .await?;
+    let path = std::env::current_dir()?.join("blobs");
+    let options = Options {
+        path: PathOptions::new(&path),
+        gc: Some(GcConfig {
+            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 = blobs.client().batch().await?;
+        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(
-            blobs.client(),
-            &[*a.hash(), *b.hash()],
-            "expiring",
-            expires_at,
-        )
-        .await?;
+        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(blobs.client(), &[*c.hash()], "expiring", expires_at).await?;
+        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
     }
-    let client = blobs.client().clone();
 
     // delete expired tags every 5 seconds
-    let delete_task = tokio::spawn(delete_expired_tags_task(client.clone(), "expiring"));
+    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(client.clone()));
+    let info_task = tokio::spawn(info_task(store.deref().clone()));
 
     ctrl_c().await?;
     delete_task.abort();
     info_task.abort();
-    router.shutdown().await?;
+    store.shutdown().await?;
     Ok(())
 }

src/api/tags.rs

@@ -58,17 +58,22 @@ impl Tags {
         Ok(stream.next().await.transpose()?)
     }
 
-    pub async fn set_with_opts(&self, options: SetOptions) -> super::RequestResult<()> {
+    pub async fn set_with_opts(&self, options: SetOptions) -> super::RequestResult<TagInfo> {
         trace!("{:?}", options);
+        let info = TagInfo {
+            name: options.name.clone(),
+            hash: options.value.hash,
+            format: options.value.format,
+        };
         self.client.rpc(options).await??;
-        Ok(())
+        Ok(info)
     }
 
     pub async fn set(
         &self,
         name: impl AsRef<[u8]>,
         value: impl Into<HashAndFormat>,
-    ) -> super::RequestResult<()> {
+    ) -> super::RequestResult<TagInfo> {
         self.set_with_opts(SetOptions {
             name: Tag::from(name.as_ref()),
             value: value.into(),

src/store/fs.rs

@@ -693,7 +693,14 @@ impl Actor {
         fs::create_dir_all(db_path.parent().unwrap())?;
         let (db_send, db_recv) = tokio::sync::mpsc::channel(100);
         let (protect, ds) = delete_set::pair(Arc::new(options.path.clone()));
-        let db_actor = meta::Actor::new(db_path, db_recv, ds, options.batch.clone())?;
+        let db_actor = meta::Actor::new(db_path, db_recv, ds, options.batch.clone());
+        let db_actor = match db_actor {
+            Ok(actor) => actor,
+            Err(err) => {
+                println!("failed to create meta actor: {err}");
+                return Err(err);
+            }
+        };
         let slot_context = Arc::new(TaskContext {
             options: options.clone(),
             db: meta::Db::new(db_send),
@@ -1209,6 +1216,7 @@ impl FsStore {
         let (commands_tx, commands_rx) = tokio::sync::mpsc::channel(100);
         let (fs_commands_tx, fs_commands_rx) = tokio::sync::mpsc::channel(100);
         let gc_config = options.gc.clone();
+        println!("Creating actor");
         let actor = handle
             .spawn(Actor::new(
                 db_path,

src/store/fs/meta.rs

@@ -409,7 +409,8 @@ impl Actor {
         options: BatchOptions,
     ) -> anyhow::Result<Self> {
         debug!("creating or opening meta database at {}", db_path.display());
-        let db = match redb::Database::create(db_path) {
+        let res = redb::Database::create(&db_path);
+        let db = match res {
             Ok(db) => db,
             Err(DatabaseError::UpgradeRequired(1)) => {
                 return Err(anyhow::anyhow!("migration from v1 no longer supported"));

src/store/fs/options.rs

@@ -4,7 +4,8 @@ use std::{
     time::Duration,
 };
 
-use super::{gc::GcConfig, meta::raw_outboard_size, temp_name};
+pub use super::gc::GcConfig;
+use super::{meta::raw_outboard_size, temp_name};
 use crate::Hash;
 
 /// Options for directories used by the file store.