prettify the info output

Author: rklaehn

examples/expiring-tags.rs

@@ -1,19 +1,24 @@
+//! 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.
+//!
+//! Then we can just use
 use std::time::{Duration, SystemTime};
 
 use chrono::Utc;
 use futures_lite::StreamExt;
 use iroh::endpoint;
-use iroh_blobs::store::GcConfig;
-use iroh_blobs::{hashseq::HashSeq, BlobFormat, HashAndFormat};
-use iroh_blobs::Hash;
-
-use iroh_blobs::rpc::client::blobs::MemClient as BlobsClient;
+use iroh_blobs::{
+    hashseq::HashSeq, rpc::client::blobs::MemClient as BlobsClient, store::GcConfig, BlobFormat,
+    Hash, HashAndFormat, Tag,
+};
 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,
     hashes: &[Hash],
@@ -40,65 +45,78 @@ async fn create_expiring_tag(
     Ok(())
 }
 
-async fn delete_expired_tags(iroh: &BlobsClient, prefix: &str) -> anyhow::Result<()> {
-    let mut tags = iroh.tags().list().await?;
+async fn delete_expired_tags(blobs: &BlobsClient, prefix: &str, bulk: bool) -> anyhow::Result<()> {
+    let mut tags = blobs.tags().list().await?;
     let prefix = format!("{}-", prefix);
     let now = chrono::Utc::now();
-    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);
+    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 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);
-        iroh.tags().delete(tag).await?;
+        for tag in to_delete {
+            println!("Deleting expired tag {}", tag);
+            blobs.tags().delete(tag).await?;
+        }
     }
     Ok(())
 }
 
-async fn print_tags_task(blobs: BlobsClient) -> anyhow::Result<()> {
+async fn info_task(blobs: BlobsClient) -> anyhow::Result<()> {
+    tokio::time::sleep(Duration::from_secs(1)).await;
     loop {
         let now = chrono::Utc::now();
         let mut tags = blobs.tags().list().await?;
-        println!("Tags at {}:\n", now);
+        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);
         }
-        println!();
-        tokio::time::sleep(Duration::from_secs(5)).await;
-    }
-}
-
-async fn print_blobs_task(blobs: BlobsClient) -> anyhow::Result<()> {
-    loop {
-        let now = chrono::Utc::now();
         let mut blobs = blobs.list().await?;
-        println!("Blobs at {}:\n", now);
+        println!("Blobs:");
         while let Some(info) = blobs.next().await {
-            println!("  {:?}", info?);
+            let info = info?;
+            println!("  {} {} bytes", info.hash, info.size);
         }
         println!();
         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(blobs: BlobsClient, prefix: &str) -> anyhow::Result<()> {
     loop {
-        delete_expired_tags(&blobs, prefix).await?;
+        delete_expired_tags(&blobs, prefix, false).await?;
         tokio::time::sleep(Duration::from_secs(5)).await;
     }
 }
@@ -108,8 +126,7 @@ 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);
+    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),
@@ -120,36 +137,44 @@ async fn main() -> anyhow::Result<()> {
     // 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?;
+        .spawn()
+        .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 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?;
+        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?;
 
         // 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();
+        let expires_at = SystemTime::now()
+            .checked_add(Duration::from_secs(60))
+            .unwrap();
         create_expiring_tag(blobs.client(), &[*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 check_task = tokio::spawn(delete_expired_tags_task(client.clone(), "expiring"));
-    // print tags every 5 seconds
-    let print_tags_task = tokio::spawn(print_tags_task(client.clone()));
-    // print blobs every 5 seconds
-    let print_blobs_task = tokio::spawn(print_blobs_task(client));
+    let delete_task = tokio::spawn(delete_expired_tags_task(client.clone(), "expiring"));
+    // print all tags and blobs every 5 seconds
+    let info_task = tokio::spawn(info_task(client.clone()));
 
     ctrl_c().await?;
+    delete_task.abort();
+    info_task.abort();
     router.shutdown().await?;
-    check_task.abort();
-    print_tags_task.abort();
-    print_blobs_task.abort();
     Ok(())
-}
+}