Make tag more rich and add a richer API for tag deletion as well

todo: make it compile and add tests

Author: rklaehn

src/rpc/client/tags.rs

@@ -10,6 +10,8 @@
 //! [`Client::list_hash_seq`] can be used to list all tags with a hash_seq format.
 //!
 //! [`Client::delete`] can be used to delete a tag.
+use std::ops::{Bound, RangeBounds};
+
 use anyhow::Result;
 use futures_lite::{Stream, StreamExt};
 use quic_rpc::{client::BoxedConnector, Connector, RpcClient};
@@ -30,6 +32,125 @@ pub struct Client<C = BoxedConnector<RpcService>> {
     pub(super) rpc: RpcClient<RpcService, C>,
 }
 
+/// Options for a list operation.
+#[derive(Debug, Clone)]
+pub struct ListOptions {
+    /// List tags to hash seqs
+    pub hash_seq: bool,
+    /// List tags to raw blobs
+    pub raw: bool,
+    /// Optional from tag (inclusive)
+    pub from: Option<Tag>,
+    /// Optional to tag (exclusive)
+    pub to: Option<Tag>,
+}
+
+fn tags_from_range<R, E>(range: R) -> (Option<Tag>, Option<Tag>)
+where
+    R: RangeBounds<E>,
+    E: AsRef<[u8]>,
+{
+    let from = match range.start_bound() {
+        Bound::Included(start) => Some(Tag::from(start.as_ref())),
+        Bound::Excluded(start) => Some(Tag::from(start.as_ref()).successor()),
+        Bound::Unbounded => None,
+    };
+    let to = match range.end_bound() {
+        Bound::Included(end) => Some(Tag::from(end.as_ref()).successor()),
+        Bound::Excluded(end) => Some(Tag::from(end.as_ref())),
+        Bound::Unbounded => None,
+    };
+    (from, to)
+}
+
+impl ListOptions {
+    /// List 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,
+            raw: true,
+            hash_seq: true,
+        }
+    }
+
+    /// List tags with a prefix
+    pub fn prefix(prefix: &[u8]) -> Self {
+        let from = Tag::from(prefix);
+        let to = from.next_prefix();
+        Self {
+            raw: true,
+            hash_seq: true,
+            from: Some(from),
+            to,
+        }
+    }
+
+    /// List a single tag
+    pub fn single(name: &[u8]) -> Self {
+        let from = Tag::from(name);
+        Self {
+            to: Some(from.successor()),
+            from: Some(from),
+            raw: true,
+            hash_seq: true,
+        }
+    }
+
+    /// List all tags
+    pub fn all() -> Self {
+        Self {
+            raw: true,
+            hash_seq: true,
+            from: None,
+            to: None,
+        }
+    }
+
+    /// List raw tags
+    pub fn raw() -> Self {
+        Self {
+            raw: true,
+            hash_seq: false,
+            from: None,
+            to: None,
+        }
+    }
+
+    /// List hash seq tags
+    pub fn hash_seq() -> Self {
+        Self {
+            raw: false,
+            hash_seq: true,
+            from: None,
+            to: None,
+        }
+    }
+}
+
+/// Options for a delete operation.
+#[derive(Debug, Clone)]
+pub struct DeleteOptions {
+    /// Optional from tag (inclusive)
+    pub from: Option<Tag>,
+    /// Optional to tag (exclusive)
+    pub to: Option<Tag>,
+}
+
+impl DeleteOptions {
+    pub fn single(name: Tag) -> Self {
+        Self {
+            to: Some(name.successor()),
+            from: Some(name),
+        }
+    }
+}
+
 /// A client that uses the memory connector.
 pub type MemClient = Client<crate::rpc::MemConnector>;
 
@@ -42,44 +163,67 @@ where
         Self { rpc }
     }
 
+    /// List all tags with options.
+    ///
+    /// This is the most flexible way to list tags. All the other list methods are just convenience
+    /// methods that call this one with the appropriate options.
+    pub async fn list_with_opts(
+        &self,
+        options: ListOptions,
+    ) -> Result<impl Stream<Item = Result<TagInfo>>> {
+        let stream = self
+            .rpc
+            .server_streaming(ListRequest::from(options))
+            .await?;
+        Ok(stream.map(|res| res.map_err(anyhow::Error::from)))
+    }
+
     /// Get the value of a single tag
     pub async fn get(&self, name: impl AsRef<[u8]>) -> Result<Option<TagInfo>> {
         let mut stream = self
-            .rpc
-            .server_streaming(ListRequest::single(name.as_ref()))
+            .list_with_opts(ListOptions::single(name.as_ref()))
             .await?;
         Ok(stream.next().await.transpose()?)
     }
 
+    /// List a range of tags
+    pub async fn list_range<R, E>(&self, range: R) -> Result<impl Stream<Item = Result<TagInfo>>>
+    where
+        R: RangeBounds<E>,
+        E: AsRef<[u8]>,
+    {
+        self.list_with_opts(ListOptions::range(range)).await
+    }
+
     /// Lists all tags.
     pub async fn list_prefix(
         &self,
         prefix: impl AsRef<[u8]>,
     ) -> Result<impl Stream<Item = Result<TagInfo>>> {
-        let stream = self
-            .rpc
-            .server_streaming(ListRequest::prefix(prefix.as_ref()))
-            .await?;
-        Ok(stream.map(|res| res.map_err(anyhow::Error::from)))
+        self.list_with_opts(ListOptions::prefix(prefix.as_ref()))
+            .await
     }
 
     /// Lists all tags.
     pub async fn list(&self) -> Result<impl Stream<Item = Result<TagInfo>>> {
-        let stream = self.rpc.server_streaming(ListRequest::all()).await?;
-        Ok(stream.map(|res| res.map_err(anyhow::Error::from)))
+        self.list_with_opts(ListOptions::all()).await
     }
 
     /// Lists all tags with a hash_seq format.
     pub async fn list_hash_seq(&self) -> Result<impl Stream<Item = Result<TagInfo>>> {
-        let stream = self.rpc.server_streaming(ListRequest::hash_seq()).await?;
-        Ok(stream.map(|res| res.map_err(anyhow::Error::from)))
+        self.list_with_opts(ListOptions::hash_seq()).await
     }
 
     /// Deletes a tag.
-    pub async fn delete(&self, name: Tag) -> Result<()> {
-        self.rpc.rpc(DeleteRequest { name }).await??;
+    pub async fn delete_with_opts(&self, options: DeleteOptions) -> Result<()> {
+        self.rpc.rpc(DeleteRequest::from(options)).await??;
         Ok(())
     }
+
+    /// Deletes a tag.
+    pub async fn delete(&self, name: Tag) -> Result<()> {
+        self.delete_with_opts(DeleteOptions::single(name)).await
+    }
 }
 
 /// Information about a tag.

src/rpc/proto/tags.rs

@@ -1,14 +1,12 @@
 //! Tags RPC protocol
-use bytes::Bytes;
 use nested_enum_utils::enum_conversions;
 use quic_rpc_derive::rpc_requests;
 use serde::{Deserialize, Serialize};
 
 use super::{RpcResult, RpcService};
 use crate::{
     net_protocol::BatchId,
-    rpc::client::tags::TagInfo,
-    util::{increment_vec, next_prefix},
+    rpc::client::tags::{DeleteOptions, ListOptions, TagInfo},
     HashAndFormat, Tag,
 };
 
@@ -79,80 +77,37 @@ pub struct ListRequest {
     pub raw: bool,
     /// List hash seq tags
     pub hash_seq: bool,
-    /// From tag
+    /// From tag (inclusive)
     pub from: Option<Tag>,
     /// To tag (exclusive)
     pub to: Option<Tag>,
 }
 
-impl ListRequest {
-    /// List tags with a prefix
-    pub fn prefix(prefix: &[u8]) -> Self {
-        let from = prefix.to_vec();
-        let mut to = from.clone();
-        let from = Bytes::from(from).into();
-        let to = if next_prefix(&mut to) {
-            Some(Bytes::from(to).into())
-        } else {
-            None
-        };
+impl From<ListOptions> for ListRequest {
+    fn from(options: ListOptions) -> Self {
         Self {
-            raw: true,
-            hash_seq: true,
-            from: Some(from),
-            to,
-        }
-    }
-
-    /// List a single tag
-    pub fn single(name: &[u8]) -> Self {
-        let from = name.to_vec();
-        let mut next = from.clone();
-        increment_vec(&mut next);
-        let from = Bytes::from(from).into();
-        let to = Bytes::from(next).into();
-        Self {
-            raw: true,
-            hash_seq: true,
-            from: Some(from),
-            to: Some(to),
-        }
-    }
-
-    /// List all tags
-    pub fn all() -> Self {
-        Self {
-            raw: true,
-            hash_seq: true,
-            from: None,
-            to: None,
-        }
-    }
-
-    /// List raw tags
-    pub fn raw() -> Self {
-        Self {
-            raw: true,
-            hash_seq: false,
-            from: None,
-            to: None,
-        }
-    }
-
-    /// List hash seq tags
-    pub fn hash_seq() -> Self {
-        Self {
-            raw: false,
-            hash_seq: true,
-            from: None,
-            to: None,
+            raw: options.raw,
+            hash_seq: options.hash_seq,
+            from: options.from,
+            to: options.to,
         }
     }
 }
 
 /// Delete a tag
 #[derive(Debug, Serialize, Deserialize)]
 pub struct DeleteRequest {
-    /// Name of the tag
-    pub name: Tag,
+    /// From tag (inclusive)
+    pub from: Option<Tag>,
+    /// To tag (exclusive)
+    pub to: Option<Tag>,
+}
+
+impl From<DeleteOptions> for DeleteRequest {
+    fn from(options: DeleteOptions) -> Self {
+        Self {
+            from: options.from,
+            to: options.to,
+        }
+    }
 }

src/util.rs

@@ -74,6 +74,18 @@ mod redb_support {
     }
 }
 
+impl From<&[u8]> for Tag {
+    fn from(value: &[u8]) -> Self {
+        Self(Bytes::copy_from_slice(value))
+    }
+}
+
+impl AsRef<[u8]> for Tag {
+    fn as_ref(&self) -> &[u8] {
+        self.0.as_ref()
+    }
+}
+
 impl Borrow<[u8]> for Tag {
     fn borrow(&self) -> &[u8] {
         self.0.as_ref()
@@ -132,6 +144,25 @@ impl Tag {
             i += 1;
         }
     }
+
+    /// The successor of this tag in lexicographic order.
+    pub fn successor(&self) -> Self {
+        let mut bytes = self.0.to_vec();
+        increment_vec(&mut bytes);
+        Self(bytes.into())
+    }
+
+    /// If this is a prefix, get the next prefix.
+    ///
+    /// This is like successor, except that it will return None if the prefix is all 0xFF instead of appending a 0 byte.
+    pub fn next_prefix(&self) -> Option<Self> {
+        let mut bytes = self.0.to_vec();
+        if next_prefix(&mut bytes) {
+            Some(Self(bytes.into()))
+        } else {
+            None
+        }
+    }
 }
 
 /// Option for commands that allow setting a tag