Add some simple smoke tests for tags

Author: rklaehn

src/hash.rs

@@ -256,6 +256,12 @@ pub struct HashAndFormat {
     pub format: BlobFormat,
 }
 
+impl From<Hash> for HashAndFormat {
+    fn from(hash: Hash) -> Self {
+        Self::raw(hash)
+    }
+}
+
 #[cfg(feature = "redb")]
 mod redb_support {
     use postcard::experimental::max_size::MaxSize;

src/rpc.rs

@@ -313,20 +313,10 @@ impl<D: crate::store::Store> Handler<D> {
         tracing::info!("blob_list_tags");
         let blobs = self;
         Gen::new(|co| async move {
-            let tags = blobs.store().tags().await.unwrap();
+            let tags = blobs.store().tags(msg.from, msg.to).await.unwrap();
             #[allow(clippy::manual_flatten)]
             for item in tags {
                 if let Ok((name, HashAndFormat { hash, format })) = item {
-                    if let Some(from) = msg.from.as_ref() {
-                        if &name < from {
-                            continue;
-                        }
-                    }
-                    if let Some(to) = msg.to.as_ref() {
-                        if &name >= to {
-                            break;
-                        }
-                    }
                     if (format.is_raw() && msg.raw) || (format.is_hash_seq() && msg.hash_seq) {
                         co.yield_(TagInfo { name, hash, format }).await;
                     }

src/rpc/client/tags.rs

@@ -144,12 +144,33 @@ pub struct DeleteOptions {
 
 impl DeleteOptions {
     /// Delete a single tag
-    pub fn single(name: Tag) -> Self {
+    pub fn single(name: &[u8]) -> Self {
+        let name = Tag::from(name);
         Self {
             to: Some(name.successor()),
             from: Some(name),
         }
     }
+
+    /// Delete a range of tags
+    pub fn range<R, E>(range: R) -> Self
+    where
+        R: RangeBounds<E>,
+        E: AsRef<[u8]>,
+    {
+        let (from, to) = tags_from_range(range);
+        Self { from, to }
+    }
+
+    /// Delete tags with a prefix
+    pub fn prefix(prefix: &[u8]) -> Self {
+        let from = Tag::from(prefix);
+        let to = from.next_prefix();
+        Self {
+            from: Some(from),
+            to,
+        }
+    }
 }
 
 /// A client that uses the memory connector.
@@ -209,7 +230,7 @@ where
         self.list_with_opts(ListOptions::range(range)).await
     }
 
-    /// Lists all tags.
+    /// Lists all tags with the given prefix.
     pub async fn list_prefix(
         &self,
         prefix: impl AsRef<[u8]>,
@@ -235,13 +256,29 @@ where
     }
 
     /// Deletes a tag.
-    pub async fn delete(&self, name: Tag) -> Result<()> {
-        self.delete_with_opts(DeleteOptions::single(name)).await
+    pub async fn delete(&self, name: impl AsRef<[u8]>) -> Result<()> {
+        self.delete_with_opts(DeleteOptions::single(name.as_ref()))
+            .await
+    }
+
+    /// Deletes a range of tags.
+    pub async fn delete_range<R, E>(&self, range: R) -> Result<()>
+    where
+        R: RangeBounds<E>,
+        E: AsRef<[u8]>,
+    {
+        self.delete_with_opts(DeleteOptions::range(range)).await
+    }
+
+    /// Lists all tags with the given prefix.
+    pub async fn delete_prefix(&self, prefix: impl AsRef<[u8]>) -> Result<()> {
+        self.delete_with_opts(DeleteOptions::prefix(prefix.as_ref()))
+            .await
     }
 }
 
 /// Information about a tag.
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
 pub struct TagInfo {
     /// Name of the tag
     pub name: Tag,

src/store/fs.rs

@@ -601,8 +601,8 @@ pub(crate) enum ActorMessage {
     },
     /// Bulk query method: get the entire tags table
     Tags {
-        #[debug(skip)]
-        filter: FilterPredicate<Tag, HashAndFormat>,
+        from: Option<Tag>,
+        to: Option<Tag>,
         #[allow(clippy::type_complexity)]
         tx: oneshot::Sender<
             ActorResult<Vec<std::result::Result<(Tag, HashAndFormat), StorageError>>>,
@@ -863,11 +863,13 @@ impl StoreInner {
         Ok(res)
     }
 
-    async fn tags(&self) -> OuterResult<Vec<io::Result<(Tag, HashAndFormat)>>> {
+    async fn tags(
+        &self,
+        from: Option<Tag>,
+        to: Option<Tag>,
+    ) -> OuterResult<Vec<io::Result<(Tag, HashAndFormat)>>> {
         let (tx, rx) = oneshot::channel();
-        let filter: FilterPredicate<Tag, HashAndFormat> =
-            Box::new(|_i, k, v| Some((k.value(), v.value())));
-        self.tx.send(ActorMessage::Tags { filter, tx }).await?;
+        self.tx.send(ActorMessage::Tags { from, to, tx }).await?;
         let tags = rx.await?;
         // transform the internal error type into io::Error
         let tags = tags?
@@ -1299,8 +1301,12 @@ impl super::ReadableStore for Store {
         Ok(Box::new(self.0.partial_blobs().await?.into_iter()))
     }
 
-    async fn tags(&self) -> io::Result<super::DbIter<(Tag, HashAndFormat)>> {
-        Ok(Box::new(self.0.tags().await?.into_iter()))
+    async fn tags(
+        &self,
+        from: Option<Tag>,
+        to: Option<Tag>,
+    ) -> io::Result<super::DbIter<(Tag, HashAndFormat)>> {
+        Ok(Box::new(self.0.tags(from, to).await?.into_iter()))
     }
 
     fn temp_tags(&self) -> Box<dyn Iterator<Item = HashAndFormat> + Send + Sync + 'static> {
@@ -1985,23 +1991,21 @@ impl ActorState {
     fn tags(
         &mut self,
         tables: &impl ReadableTables,
-        filter: FilterPredicate<Tag, HashAndFormat>,
+        from: Option<Tag>,
+        to: Option<Tag>,
     ) -> ActorResult<Vec<std::result::Result<(Tag, HashAndFormat), StorageError>>> {
         let mut res = Vec::new();
-        let mut index = 0u64;
-        #[allow(clippy::explicit_counter_loop)]
-        for item in tables.tags().iter()? {
+        let from = from.map(Bound::Included).unwrap_or(Bound::Unbounded);
+        let to = to.map(Bound::Excluded).unwrap_or(Bound::Unbounded);
+        for item in tables.tags().range((from, to))? {
             match item {
                 Ok((k, v)) => {
-                    if let Some(item) = filter(index, k, v) {
-                        res.push(Ok(item));
-                    }
+                    res.push(Ok((k.value(), v.value())));
                 }
                 Err(e) => {
                     res.push(Err(e));
                 }
             }
-            index += 1;
         }
         Ok(res)
     }
@@ -2342,8 +2346,8 @@ impl ActorState {
                 let res = self.blobs(tables, filter);
                 tx.send(res).ok();
             }
-            ActorMessage::Tags { filter, tx } => {
-                let res = self.tags(tables, filter);
+            ActorMessage::Tags { from, to, tx } => {
+                let res = self.tags(tables, from, to);
                 tx.send(res).ok();
             }
             ActorMessage::GcStart { tx } => {

src/store/mem.rs

@@ -446,10 +446,30 @@ impl ReadableStore for Store {
         ))
     }
 
-    async fn tags(&self) -> io::Result<crate::store::DbIter<(crate::Tag, crate::HashAndFormat)>> {
+    async fn tags(
+        &self,
+        from: Option<Tag>,
+        to: Option<Tag>,
+    ) -> io::Result<crate::store::DbIter<(crate::Tag, crate::HashAndFormat)>> {
         #[allow(clippy::mutable_key_type)]
         let tags = self.read_lock().tags.clone();
-        Ok(Box::new(tags.into_iter().map(Ok)))
+        let tags = tags
+            .into_iter()
+            .filter(move |(tag, _)| {
+                if let Some(from) = &from {
+                    if tag < from {
+                        return false;
+                    }
+                }
+                if let Some(to) = &to {
+                    if tag >= to {
+                        return false;
+                    }
+                }
+                true
+            })
+            .map(Ok);
+        Ok(Box::new(tags))
     }
 
     fn temp_tags(&self) -> Box<dyn Iterator<Item = crate::HashAndFormat> + Send + Sync + 'static> {

src/store/readonly_mem.rs

@@ -232,7 +232,11 @@ impl ReadableStore for Store {
         ))
     }
 
-    async fn tags(&self) -> io::Result<DbIter<(Tag, HashAndFormat)>> {
+    async fn tags(
+        &self,
+        _from: Option<Tag>,
+        _to: Option<Tag>,
+    ) -> io::Result<DbIter<(Tag, HashAndFormat)>> {
         Ok(Box::new(std::iter::empty()))
     }
 

src/store/traits.rs

@@ -262,7 +262,11 @@ pub trait ReadableStore: Map {
     /// been imported, and hash sequences that have been created internally.
     fn blobs(&self) -> impl Future<Output = io::Result<DbIter<Hash>>> + Send;
     /// list all tags (collections or other explicitly added things) in the database
-    fn tags(&self) -> impl Future<Output = io::Result<DbIter<(Tag, HashAndFormat)>>> + Send;
+    fn tags(
+        &self,
+        from: Option<Tag>,
+        to: Option<Tag>,
+    ) -> impl Future<Output = io::Result<DbIter<(Tag, HashAndFormat)>>> + Send;
 
     /// Temp tags
     fn temp_tags(&self) -> Box<dyn Iterator<Item = HashAndFormat> + Send + Sync + 'static>;
@@ -635,7 +639,7 @@ pub(super) async fn gc_mark_task<'a>(
     }
     let mut roots = BTreeSet::new();
     debug!("traversing tags");
-    for item in store.tags().await? {
+    for item in store.tags(None, None).await? {
         let (name, haf) = item?;
         debug!("adding root {:?} {:?}", name, haf);
         roots.insert(haf);

tests/tags.rs

@@ -0,0 +1,115 @@
+#![cfg(all(feature = "net_protocol", feature = "rpc"))]
+use futures_lite::StreamExt;
+use futures_util::Stream;
+use iroh::Endpoint;
+use iroh_blobs::{
+    net_protocol::Blobs,
+    rpc::{
+        client::tags::{self, TagInfo},
+        proto::RpcService,
+    },
+    BlobFormat, Hash,
+};
+use testresult::TestResult;
+
+async fn to_vec<T>(stream: impl Stream<Item = anyhow::Result<T>>) -> anyhow::Result<Vec<T>> {
+    let res = stream.collect::<Vec<_>>().await;
+    res.into_iter().collect::<anyhow::Result<Vec<_>>>()
+}
+
+fn expected(tags: impl IntoIterator<Item = &'static str>) -> Vec<TagInfo> {
+    tags.into_iter()
+        .map(|tag| TagInfo {
+            name: tag.into(),
+            hash: Hash::new(tag),
+            format: BlobFormat::Raw,
+        })
+        .collect()
+}
+
+async fn tags_smoke<C: quic_rpc::Connector<RpcService>>(tags: tags::Client<C>) -> TestResult<()> {
+    tags.set("a", Hash::new("a")).await?;
+    tags.set("b", Hash::new("b")).await?;
+    tags.set("c", Hash::new("c")).await?;
+    tags.set("d", Hash::new("d")).await?;
+    tags.set("e", Hash::new("e")).await?;
+    let stream = tags.list().await?;
+    let res = to_vec(stream).await?;
+    assert_eq!(res, expected(["a", "b", "c", "d", "e"]));
+
+    let stream = tags.list_range("b".."d").await?;
+    let res = to_vec(stream).await?;
+    assert_eq!(res, expected(["b", "c"]));
+
+    let stream = tags.list_range("b"..).await?;
+    let res = to_vec(stream).await?;
+    assert_eq!(res, expected(["b", "c", "d", "e"]));
+
+    let stream = tags.list_range(.."d").await?;
+    let res = to_vec(stream).await?;
+    assert_eq!(res, expected(["a", "b", "c"]));
+
+    let stream = tags.list_range(..="d").await?;
+    let res = to_vec(stream).await?;
+    assert_eq!(res, expected(["a", "b", "c", "d"]));
+
+    tags.delete_range("b"..).await?;
+    let stream = tags.list().await?;
+    let res = to_vec(stream).await?;
+    assert_eq!(res, expected(["a"]));
+
+    tags.delete_range(..="a").await?;
+    let stream = tags.list().await?;
+    let res = to_vec(stream).await?;
+    assert_eq!(res, expected([]));
+
+    tags.set("a", Hash::new("a")).await?;
+    tags.set("aa", Hash::new("aa")).await?;
+    tags.set("aaa", Hash::new("aaa")).await?;
+    tags.set("aab", Hash::new("aab")).await?;
+    tags.set("b", Hash::new("b")).await?;
+
+    let stream = tags.list_prefix("aa").await?;
+    let res = to_vec(stream).await?;
+    assert_eq!(res, expected(["aa", "aaa", "aab"]));
+
+    tags.delete_prefix("aa").await?;
+    let stream = tags.list().await?;
+    let res = to_vec(stream).await?;
+    assert_eq!(res, expected(["a", "b"]));
+
+    tags.delete_prefix("").await?;
+    let stream = tags.list().await?;
+    let res = to_vec(stream).await?;
+    assert_eq!(res, expected([]));
+
+    tags.set("a", Hash::new("a")).await?;
+    tags.set("b", Hash::new("b")).await?;
+    tags.set("c", Hash::new("c")).await?;
+
+    tags.delete("b").await?;
+    let stream = tags.list().await?;
+    let res = to_vec(stream).await?;
+    assert_eq!(res, expected(["a", "c"]));
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn tags_smoke_mem() -> TestResult<()> {
+    let endpoint = Endpoint::builder().bind().await?;
+    let blobs = Blobs::memory().build(&endpoint);
+    let client = blobs.client();
+    tags_smoke(client.tags()).await
+}
+
+#[tokio::test]
+async fn tags_smoke_fs() -> TestResult<()> {
+    let td = tempfile::tempdir()?;
+    let endpoint = Endpoint::builder().bind().await?;
+    let blobs = Blobs::persistent(td.path().join("blobs.db"))
+        .await?
+        .build(&endpoint);
+    let client = blobs.client();
+    tags_smoke(client.tags()).await
+}