fix(storage): retain publication guard through fenced mutations

Co-Authored-By: heihutu <heihutu@gmail.com>

crates/ecstore/src/disk/disk_store.rs

@@ -250,6 +250,37 @@ pub(crate) trait DiskStoreRenameDataExt {
         dst_volume: &str,
         dst_path: &str,
     ) -> Result<RenameDataResp>;
+
+    async fn rename_data_borrowed_with_guard(
+        &self,
+        src_volume: &str,
+        src_path: &str,
+        fi: &FileInfo,
+        dst_volume: &str,
+        dst_path: &str,
+        external_guard: Option<Arc<dyn Send + Sync>>,
+    ) -> Result<RenameDataResp> {
+        let _ = external_guard;
+        self.rename_data_borrowed(src_volume, src_path, fi, dst_volume, dst_path)
+            .await
+    }
+}
+
+/// Run a mutation in an owned task when a caller supplied publication guard.
+/// RPC cancellation drops only the waiter; the mutation owner keeps the guard
+/// until its operation has returned, including any detached blocking syscall.
+async fn run_owned_mutation<T, F, Fut>(external_guard: Option<Arc<dyn Send + Sync>>, operation: F) -> Result<T>
+where
+    T: Send + 'static,
+    F: FnOnce() -> Fut + Send + 'static,
+    Fut: std::future::Future<Output = Result<T>> + Send + 'static,
+{
+    tokio::spawn(async move {
+        let _external_guard = external_guard;
+        operation().await
+    })
+    .await
+    .map_err(|err| Error::other(format!("owned mutation task failed: {err}")))?
 }
 
 impl DiskStoreRenameDataExt for LocalDiskWrapper {
@@ -273,6 +304,49 @@ impl DiskStoreRenameDataExt for LocalDiskWrapper {
         )
         .await
     }
+
+    async fn rename_data_borrowed_with_guard(
+        &self,
+        src_volume: &str,
+        src_path: &str,
+        fi: &FileInfo,
+        dst_volume: &str,
+        dst_path: &str,
+        external_guard: Option<Arc<dyn Send + Sync>>,
+    ) -> Result<RenameDataResp> {
+        let operation = self.clone();
+        let src_volume = src_volume.to_owned();
+        let src_path = src_path.to_owned();
+        let fi = fi.clone();
+        let dst_volume = dst_volume.to_owned();
+        let dst_path = dst_path.to_owned();
+        let timeout_duration = if external_guard.is_some() {
+            // A fenced mutation owns the publication guard until the storage
+            // operation returns. Timing out this waiter would cancel the
+            // LocalDisk future while a spawn_blocking namespace syscall could
+            // still be committing, reopening the movement window. The caller
+            // may drop its waiter; the owned task drains the mutation.
+            Duration::ZERO
+        } else {
+            get_max_timeout_duration()
+        };
+        run_owned_mutation(external_guard, move || async move {
+            operation
+                .track_disk_health_mutation(
+                    "rename_data",
+                    DiskMetricMutation::Write,
+                    || async {
+                        operation
+                            .disk
+                            .rename_data_borrowed(&src_volume, &src_path, &fi, &dst_volume, &dst_path)
+                            .await
+                    },
+                    timeout_duration,
+                )
+                .await
+        })
+        .await
+    }
 }
 
 pub fn get_drive_walkdir_timeout() -> Duration {
@@ -1097,6 +1171,37 @@ impl LocalDiskWrapper {
         )
     }
 
+    /// Run a delete under an owned coordinator task when a publication guard
+    /// is present. This keeps the guard alive if the RPC waiter is cancelled
+    /// while the local namespace mutation is still in progress.
+    pub(crate) async fn delete_with_publication_guard(
+        &self,
+        volume: &str,
+        path: &str,
+        options: DeleteOptions,
+        external_guard: Option<Arc<dyn Send + Sync>>,
+    ) -> Result<()> {
+        let operation = self.clone();
+        let volume = volume.to_owned();
+        let path = path.to_owned();
+        let timeout_duration = if external_guard.is_some() {
+            Duration::ZERO
+        } else {
+            get_max_timeout_duration()
+        };
+        run_owned_mutation(external_guard, move || async move {
+            operation
+                .track_disk_health_mutation(
+                    "delete",
+                    DiskMetricMutation::Delete,
+                    || async { operation.disk.delete(&volume, &path, options).await },
+                    timeout_duration,
+                )
+                .await
+        })
+        .await
+    }
+
     pub(crate) fn new_with_reconnect_state(
         disk: Arc<LocalDisk>,
         health_check: bool,
@@ -2247,6 +2352,44 @@ mod tests {
     };
     use tokio::io::AsyncWrite;
 
+    struct DropProbe(Arc<std::sync::atomic::AtomicUsize>);
+
+    impl Drop for DropProbe {
+        fn drop(&mut self) {
+            self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        }
+    }
+
+    #[tokio::test]
+    async fn owned_mutation_keeps_publication_guard_after_waiter_cancellation() {
+        let drops = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let guard: Arc<dyn Send + Sync> = Arc::new(DropProbe(Arc::clone(&drops)));
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        let (release_tx, release_rx) = tokio::sync::oneshot::channel();
+        let (finished_tx, finished_rx) = tokio::sync::oneshot::channel();
+
+        let waiter = tokio::spawn(run_owned_mutation(Some(guard), move || async move {
+            started_tx.send(()).expect("mutation should signal start");
+            release_rx.await.expect("mutation should be released");
+            finished_tx.send(()).expect("mutation should signal completion");
+            Ok::<_, Error>(())
+        }));
+
+        started_rx.await.expect("mutation owner should start");
+        waiter.abort();
+        assert_eq!(drops.load(std::sync::atomic::Ordering::SeqCst), 0);
+
+        release_tx.send(()).expect("mutation owner should still be alive");
+        finished_rx.await.expect("mutation owner should finish");
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while drops.load(std::sync::atomic::Ordering::SeqCst) == 0 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("publication guard should be released after mutation completion");
+    }
+
     struct PendingWriter;
 
     #[test]

crates/ecstore/src/disk/mod.rs

@@ -694,6 +694,28 @@ impl Disk {
         }
     }
 
+    pub async fn delete_with_scanner_publication_lease_and_guard(
+        &self,
+        volume: &str,
+        path: &str,
+        opts: DeleteOptions,
+        scanner_publication_lease_token: Option<Uuid>,
+        external_guard: Option<Arc<dyn Send + Sync>>,
+    ) -> Result<()> {
+        match self {
+            Disk::Local(local_disk) => {
+                local_disk
+                    .delete_with_publication_guard(volume, path, opts, external_guard)
+                    .await
+            }
+            Disk::Remote(remote_disk) => {
+                remote_disk
+                    .delete_with_scanner_publication_lease(volume, path, opts, scanner_publication_lease_token)
+                    .await
+            }
+        }
+    }
+
     pub(crate) async fn rename_data_borrowed(
         &self,
         src_volume: &str,
@@ -714,11 +736,33 @@ impl Disk {
         dst_volume: &str,
         dst_path: &str,
         scanner_publication_lease_token: Option<Uuid>,
+    ) -> Result<RenameDataResp> {
+        self.rename_data_borrowed_with_fence_and_guard(
+            src_volume,
+            src_path,
+            fi,
+            dst_volume,
+            dst_path,
+            scanner_publication_lease_token,
+            None,
+        )
+        .await
+    }
+
+    pub async fn rename_data_borrowed_with_fence_and_guard(
+        &self,
+        src_volume: &str,
+        src_path: &str,
+        fi: &FileInfo,
+        dst_volume: &str,
+        dst_path: &str,
+        scanner_publication_lease_token: Option<Uuid>,
+        external_guard: Option<Arc<dyn Send + Sync>>,
     ) -> Result<RenameDataResp> {
         match self {
             Disk::Local(local_disk) => {
                 local_disk
-                    .rename_data_borrowed(src_volume, src_path, fi, dst_volume, dst_path)
+                    .rename_data_borrowed_with_guard(src_volume, src_path, fi, dst_volume, dst_path, external_guard)
                     .await
             }
             Disk::Remote(remote_disk) => {

rustfs/src/storage/rpc/node_service/disk.rs

@@ -33,6 +33,7 @@ use rustfs_io_metrics::internode_metrics::{
 use rustfs_protos::proto_gen::node_service::*;
 use serde::de::DeserializeOwned;
 use std::io::Cursor;
+use std::sync::Arc;
 use std::time::Instant;
 use tonic::{Request, Response, Status};
 use tracing::debug;
@@ -1230,37 +1231,40 @@ impl NodeService {
             // The target owns this read guard.  It must span the complete
             // disk rename, not merely the preflight, so a movement transition
             // cannot restart after validation and before rename linearization.
-            let _scanner_publication_lease_guard = if let Some(token) = scanner_publication_lease_token {
-                let Some(store) = self.resolve_object_store() else {
-                    return Ok(Response::new(RenameDataResponse {
-                        success: false,
-                        rename_data_resp: String::new(),
-                        rename_data_resp_bin: Vec::new().into(),
-                        error: Some(DiskError::other("scanner publication lease owner is unavailable").into()),
-                    }));
-                };
-                match store.acquire_scanner_publication_lease_guard(token).await {
-                    Ok(guard) => Some(guard),
-                    Err(err) => {
+            let scanner_publication_lease_guard: Option<Arc<dyn Send + Sync>> =
+                if let Some(token) = scanner_publication_lease_token {
+                    let Some(store) = self.resolve_object_store() else {
                         return Ok(Response::new(RenameDataResponse {
                             success: false,
                             rename_data_resp: String::new(),
                             rename_data_resp_bin: Vec::new().into(),
-                            error: Some(DiskError::other(err.to_string()).into()),
+                            error: Some(DiskError::other("scanner publication lease owner is unavailable").into()),
                         }));
+                    };
+                    match store.acquire_scanner_publication_lease_guard(token).await {
+                        Ok(guard) => Some(Arc::new(guard)),
+                        Err(err) => {
+                            return Ok(Response::new(RenameDataResponse {
+                                success: false,
+                                rename_data_resp: String::new(),
+                                rename_data_resp_bin: Vec::new().into(),
+                                error: Some(DiskError::other(err.to_string()).into()),
+                            }));
+                        }
                     }
-                }
-            } else {
-                None
-            };
+                } else {
+                    None
+                };
             let request_decoded_from_msgpack = decoded_file_info.from_msgpack;
             match disk
-                .rename_data(
+                .rename_data_borrowed_with_fence_and_guard(
                     &request.src_volume,
                     &request.src_path,
                     &decoded_file_info.value,
                     &request.dst_volume,
                     &request.dst_path,
+                    scanner_publication_lease_token,
+                    scanner_publication_lease_guard,
                 )
                 .await
             {
@@ -1641,26 +1645,36 @@ impl NodeService {
             // The target-side guard spans the complete delete operation. A
             // lease expiry or movement transition cannot occur between this
             // validation and the disk delete linearization point.
-            let _scanner_publication_lease_guard = if let Some(token) = scanner_publication_lease_token {
-                let Some(store) = self.resolve_object_store() else {
-                    return Ok(Response::new(DeleteResponse {
-                        success: false,
-                        error: Some(DiskError::other("scanner publication lease owner is unavailable").into()),
-                    }));
-                };
-                match store.acquire_scanner_publication_lease_guard(token).await {
-                    Ok(guard) => Some(guard),
-                    Err(err) => {
+            let scanner_publication_lease_guard: Option<Arc<dyn Send + Sync>> =
+                if let Some(token) = scanner_publication_lease_token {
+                    let Some(store) = self.resolve_object_store() else {
                         return Ok(Response::new(DeleteResponse {
                             success: false,
-                            error: Some(DiskError::other(err.to_string()).into()),
+                            error: Some(DiskError::other("scanner publication lease owner is unavailable").into()),
                         }));
+                    };
+                    match store.acquire_scanner_publication_lease_guard(token).await {
+                        Ok(guard) => Some(Arc::new(guard)),
+                        Err(err) => {
+                            return Ok(Response::new(DeleteResponse {
+                                success: false,
+                                error: Some(DiskError::other(err.to_string()).into()),
+                            }));
+                        }
                     }
-                }
-            } else {
-                None
-            };
-            match disk.delete(&request.volume, &request.path, options).await {
+                } else {
+                    None
+                };
+            match disk
+                .delete_with_scanner_publication_lease_and_guard(
+                    &request.volume,
+                    &request.path,
+                    options,
+                    scanner_publication_lease_token,
+                    scanner_publication_lease_guard,
+                )
+                .await
+            {
                 Ok(_) => Ok(Response::new(DeleteResponse {
                     success: true,
                     error: None,