diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 5c0f6454e..cc475ed33 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -250,6 +250,37 @@ pub(crate) trait DiskStoreRenameDataExt { dst_volume: &str, dst_path: &str, ) -> Result; + + 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>, + ) -> Result { + 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(external_guard: Option>, operation: F) -> Result +where + T: Send + 'static, + F: FnOnce() -> Fut + Send + 'static, + Fut: std::future::Future> + 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>, + ) -> Result { + 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>, + ) -> 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, health_check: bool, @@ -2247,6 +2352,44 @@ mod tests { }; use tokio::io::AsyncWrite; + struct DropProbe(Arc); + + 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 = 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] diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index fb69085c4..ff959e9e6 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/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, + external_guard: Option>, + ) -> 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, + ) -> Result { + 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, + external_guard: Option>, ) -> Result { 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) => { diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index e2457c996..19fb86c80 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/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> = + 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> = + 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,