Add lots of plumbing to make rename atomic

Author: rklaehn

src/rpc.rs

@@ -30,7 +30,7 @@ use proto::{
     },
     tags::{
         CreateRequest as TagsCreateRequest, DeleteRequest as TagDeleteRequest,
-        ListRequest as TagListRequest, SetRequest as TagsSetRequest, SyncMode,
+        ListRequest as TagListRequest, RenameRequest, SetRequest as TagsSetRequest, SyncMode,
     },
     Request, RpcError, RpcResult, RpcService,
 };
@@ -158,6 +158,7 @@ impl<D: crate::store::Store> Handler<D> {
             Set(msg) => chan.rpc(msg, self, Self::tags_set).await,
             DeleteTag(msg) => chan.rpc(msg, self, Self::blob_delete_tag).await,
             ListTags(msg) => chan.server_streaming(msg, self, Self::blob_list_tags).await,
+            Rename(msg) => chan.rpc(msg, self, Self::tags_rename).await,
         }
     }
 
@@ -382,6 +383,16 @@ impl<D: crate::store::Store> Handler<D> {
         rx.map(AddPathResponse)
     }
 
+    async fn tags_rename(self, msg: RenameRequest) -> RpcResult<()> {
+        let blobs = self;
+        blobs
+            .store()
+            .rename_tag(msg.from, msg.to)
+            .await
+            .map_err(|e| RpcError::new(&e))?;
+        Ok(())
+    }
+
     async fn tags_set(self, msg: TagsSetRequest) -> RpcResult<()> {
         let blobs = self;
         blobs

src/rpc/proto/tags.rs

@@ -20,6 +20,8 @@ pub enum Request {
     #[rpc(response = RpcResult<()>)]
     Set(SetRequest),
     #[rpc(response = RpcResult<()>)]
+    Rename(RenameRequest),
+    #[rpc(response = RpcResult<()>)]
     DeleteTag(DeleteRequest),
     #[server_streaming(response = TagInfo)]
     ListTags(ListRequest),
@@ -111,3 +113,12 @@ impl From<DeleteOptions> for DeleteRequest {
         }
     }
 }
+
+/// Rename a tag atomically
+#[derive(Debug, Serialize, Deserialize)]
+pub struct RenameRequest {
+    /// Old tag name
+    pub from: Tag,
+    /// New tag name
+    pub to: Tag,
+}

src/store/fs.rs

@@ -625,6 +625,12 @@ pub(crate) enum ActorMessage {
         hash: HashAndFormat,
         tx: oneshot::Sender<ActorResult<Tag>>,
     },
+    /// Modification method: rename a tag atomically.
+    RenameTag {
+        from: Tag,
+        to: Tag,
+        tx: oneshot::Sender<ActorResult<()>>,
+    },
     /// Modification method: unconditional delete the data for a number of hashes
     Delete {
         hashes: Vec<Hash>,
@@ -680,6 +686,7 @@ impl ActorMessage {
             | Self::SetFullEntryState { .. }
             | Self::Delete { .. }
             | Self::DeleteTags { .. }
+            | Self::RenameTag { .. }
             | Self::GcDelete { .. } => MessageCategory::ReadWrite,
             Self::UpdateInlineOptions { .. }
             | Self::Sync { .. }
@@ -901,6 +908,14 @@ impl StoreInner {
         Ok(rx.await??)
     }
 
+    async fn rename_tag(&self, from: Tag, to: Tag) -> OuterResult<()> {
+        let (tx, rx) = oneshot::channel();
+        self.tx
+            .send(ActorMessage::RenameTag { from, to, tx })
+            .await?;
+        Ok(rx.await??)
+    }
+
     async fn delete(&self, hashes: Vec<Hash>) -> OuterResult<()> {
         let (tx, rx) = oneshot::channel();
         self.tx.send(ActorMessage::Delete { hashes, tx }).await?;
@@ -1404,6 +1419,10 @@ impl super::Store for Store {
         Ok(self.0.create_tag(hash).await?)
     }
 
+    async fn rename_tag(&self, from: Tag, to: Tag) -> io::Result<()> {
+        Ok(self.0.rename_tag(from, to).await?)
+    }
+
     async fn delete(&self, hashes: Vec<Hash>) -> io::Result<()> {
         Ok(self.0.delete(hashes).await?)
     }
@@ -2021,6 +2040,18 @@ impl ActorState {
         Ok(tag)
     }
 
+    fn rename_tag(&mut self, tables: &mut Tables, from: Tag, to: Tag) -> ActorResult<()> {
+        let value = tables
+            .tags
+            .get(from)?
+            .ok_or_else(|| {
+                ActorError::Io(io::Error::new(io::ErrorKind::NotFound, "tag not found"))
+            })?
+            .value();
+        tables.tags.insert(to, value)?;
+        Ok(())
+    }
+
     fn set_tag(&self, tables: &mut Tables, tag: Tag, value: HashAndFormat) -> ActorResult<()> {
         tables.tags.insert(tag, value)?;
         Ok(())
@@ -2393,6 +2424,10 @@ impl ActorState {
                 let res = self.create_tag(tables, hash);
                 tx.send(res).ok();
             }
+            ActorMessage::RenameTag { from, to, tx } => {
+                let res = self.rename_tag(tables, from, to);
+                tx.send(res).ok();
+            }
             ActorMessage::Delete { hashes, tx } => {
                 let res = self.delete(tables, hashes, true);
                 tx.send(res).ok();

src/store/mem.rs

@@ -208,6 +208,18 @@ impl super::Store for Store {
         .await?
     }
 
+    async fn rename_tag(&self, from: Tag, to: Tag) -> io::Result<()> {
+        let mut state = self.write_lock();
+        let value = state.tags.remove(&from).ok_or_else(|| {
+            io::Error::new(
+                io::ErrorKind::NotFound,
+                format!("tag not found: {:?}", from),
+            )
+        })?;
+        state.tags.insert(to, value);
+        Ok(())
+    }
+
     async fn set_tag(&self, name: Tag, value: HashAndFormat) -> io::Result<()> {
         let mut state = self.write_lock();
         state.tags.insert(name, value);

src/store/readonly_mem.rs

@@ -307,6 +307,10 @@ impl super::Store for Store {
         Err(io::Error::new(io::ErrorKind::Other, "not implemented"))
     }
 
+    async fn rename_tag(&self, _from: Tag, _to: Tag) -> io::Result<()> {
+        Err(io::Error::new(io::ErrorKind::Other, "not implemented"))
+    }
+
     async fn import_stream(
         &self,
         data: impl Stream<Item = io::Result<Bytes>> + Unpin + Send,

src/store/traits.rs

@@ -352,6 +352,9 @@ pub trait Store: ReadableStore + MapMut + std::fmt::Debug {
         hash: HashAndFormat,
     ) -> impl Future<Output = io::Result<()>> + Send;
 
+    /// Rename a tag
+    fn rename_tag(&self, from: Tag, to: Tag) -> impl Future<Output = io::Result<()>> + Send;
+
     /// Delete a single tag
     fn delete_tag(&self, name: Tag) -> impl Future<Output = io::Result<()>> + Send {
         self.delete_tags(Some(name.clone()), Some(name.successor()))