fix(ecstore): retain namespace owners through local disk completion (#7245)

* fix(ecstore): retain namespace owners through local physical tails

(cherry picked from commit a2f242463316e87604feadbdac5e4148140e72c0)

* test(ecstore): expose stale fsync group cleanup

* fix(ecstore): capture complete fsync worker guard

(cherry picked from commit 1dc90bb836e20ea9ee45d0a629a9201e20d231c4)

* fix(ecstore): preserve successor fsync group registration

(cherry picked from commit c7dfaad90526052e56c57dafffa4813bdcde46ca)

* test(ecstore): mark physical owner fixtures as inline

* test(ecstore): wait for namespace owner release before asserting

The namespace owner tests decided that ownership had ended when the Weak probe stopped upgrading or when the mutation lease could be reacquired. Both signals fire before the owner guard's Drop decrements the pending counter: Arc releases its strong count before running Drop, and the lease drops its locks before its owner field. The rio-v2 lane hit that window in undo_fresh_version_keeps_physical_namespace_owner_after_timeout.

Extend every drain wait to also require namespace_commits_pending() to be false, so the assertions observe the completed release instead of racing it.

crates/ecstore/src/disk/disk_store.rs

@@ -324,6 +324,30 @@ impl DiskStoreRenameDataExt for LocalDiskWrapper {
 }
 
 impl LocalDiskWrapper {
+    pub(in crate::disk) async fn undo_write_with_namespace_owner(
+        &self,
+        volume: &str,
+        path: &str,
+        fi: FileInfo,
+        opts: DeleteOptions,
+        namespace_owner: Option<Arc<dyn Send + Sync>>,
+    ) -> Result<()> {
+        self.track_disk_health_mutation(
+            "delete_version",
+            DiskMetricMutation::Delete,
+            || async {
+                // Preserve the old DiskAPI future's boxing boundary.
+                Box::pin(
+                    self.disk
+                        .undo_write_with_namespace_owner(volume, path, fi, opts, namespace_owner),
+                )
+                .await
+            },
+            get_max_timeout_duration(),
+        )
+        .await
+    }
+
     pub(in crate::disk) async fn rename_data_observed(
         &self,
         src_volume: &str,
@@ -333,6 +357,34 @@ impl LocalDiskWrapper {
         dst_path: &str,
         external_guard: Option<Arc<dyn Send + Sync>>,
     ) -> super::RenameDataObservation {
+        self.rename_data_observed_with_guards(
+            src_volume,
+            src_path,
+            fi,
+            dst_volume,
+            dst_path,
+            super::RenameDataGuards {
+                external_guard,
+                ..Default::default()
+            },
+        )
+        .await
+    }
+
+    pub(in crate::disk) async fn rename_data_observed_with_guards(
+        &self,
+        src_volume: &str,
+        src_path: &str,
+        fi: &FileInfo,
+        dst_volume: &str,
+        dst_path: &str,
+        guards: super::RenameDataGuards,
+    ) -> super::RenameDataObservation {
+        let super::RenameDataGuards {
+            external_guard,
+            namespace_owner,
+            ..
+        } = guards;
         let operation = self.clone();
         let src_volume = src_volume.to_owned();
         let src_path = src_path.to_owned();
@@ -357,13 +409,15 @@ impl LocalDiskWrapper {
                     DiskMetricMutation::Write,
                     || async {
                         // Preserve the former DiskAPI future's single boxing boundary.
-                        let observed =
-                            Box::pin(
-                                operation
-                                    .disk
-                                    .rename_data_observed(&src_volume, &src_path, &fi, &dst_volume, &dst_path),
-                            )
-                            .await;
+                        let observed = Box::pin(operation.disk.rename_data_observed(
+                            &src_volume,
+                            &src_path,
+                            &fi,
+                            &dst_volume,
+                            &dst_path,
+                            namespace_owner,
+                        ))
+                        .await;
                         preflight_rejection = observed.preflight_rejection;
                         observed.result
                     },

crates/ecstore/src/disk/local/commit.rs

@@ -34,7 +34,7 @@ use crate::disk::{
     error::{DiskError, Result},
     error_conv::{to_access_error, to_file_error},
     os,
-    os::{check_path_length, rename_all},
+    os::check_path_length,
 };
 use bytes::Bytes;
 use rustfs_filemeta::{FileInfo, FileMeta};
@@ -74,6 +74,8 @@ fn rollback_inline_metadata_commit_std(
     rollback_data_dir: Option<Uuid>,
     local_rollback_path: Option<&Path>,
 ) -> std::io::Result<()> {
+    #[cfg(all(test, not(windows)))]
+    os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Rollback, dst_file_path);
     if let Some(backup_path) = local_rollback_path {
         // The commit immediately before this rollback renamed the staged
         // xl.meta from the same directory as `backup_path` onto
@@ -233,6 +235,12 @@ async fn restore_published_data_source(
 #[derive(Debug)]
 pub(in crate::disk) struct LocalRenamePreflightRejection(());
 
+#[derive(Default)]
+pub(super) struct RenameDataState {
+    namespace_owner: Option<Arc<dyn Send + Sync>>,
+    preflight_rejection: Option<LocalRenamePreflightRejection>,
+}
+
 impl LocalDisk {
     #[tracing::instrument(name = "rename_data", target = "rustfs_ecstore::disk::local", level = "trace", skip_all)]
     pub(super) async fn rename_data_inner(
@@ -242,7 +250,7 @@ impl LocalDisk {
         fi: FileInfo,
         dst_volume: &str,
         dst_path: &str,
-        preflight_rejection: &mut Option<LocalRenamePreflightRejection>,
+        state: &mut RenameDataState,
     ) -> Result<RenameDataResp> {
         crate::hp_guard!("LocalDisk::rename_data");
         let mut fi = fi;
@@ -271,7 +279,13 @@ impl LocalDisk {
             Some(token) => Some(self.claim_quota_mutation_fence(dst_volume, dst_path, token).await?),
             None => None,
         };
-        let mutation_lease = os::acquire_rename_data_mutation_lease(&self.root, dst_volume, &destination_object_path).await;
+        let mutation_lease = os::acquire_rename_data_mutation_lease_with_owner(
+            &self.root,
+            dst_volume,
+            &destination_object_path,
+            state.namespace_owner.take(),
+        )
+        .await;
         if let Some(claim) = quota_fence_claim {
             mutation_lease.attach_external_guard(claim);
         }
@@ -304,7 +318,7 @@ impl LocalDisk {
                 error = %e,
                 "Disk local access check failed"
             );
-            *preflight_rejection = Some(LocalRenamePreflightRejection(()));
+            state.preflight_rejection = Some(LocalRenamePreflightRejection(()));
             return Err(to_access_error(e, DiskError::VolumeAccessDenied).into());
         }
 
@@ -322,7 +336,7 @@ impl LocalDisk {
                 error = %e,
                 "Disk local access check failed"
             );
-            *preflight_rejection = Some(LocalRenamePreflightRejection(()));
+            state.preflight_rejection = Some(LocalRenamePreflightRejection(()));
             return Err(to_access_error(e, DiskError::VolumeAccessDenied).into());
         }
 
@@ -530,7 +544,9 @@ impl LocalDisk {
             // rename below.
             if fi_healing
                 && let Some((_, dst_data_path)) = has_data_dir_path.as_ref()
-                && let Err(err) = self.move_to_trash(dst_data_path, true, false).await
+                && let Err(err) = self
+                    .move_to_trash_with_namespace_owner(dst_data_path, true, false, Some(mutation_lease.clone()))
+                    .await
             {
                 warn!(
                     target: "rustfs_ecstore::disk::local",
@@ -757,7 +773,7 @@ impl LocalDisk {
                 && let Some(parent) = dst_file_path.parent()
             {
                 let fsync_started = rustfs_io_metrics::put_stage_timer();
-                if let Err(err) = os::fsync_dst_dir_group_commit(parent).await {
+                if let Err(err) = os::fsync_dst_dir_group_commit(parent, Some(mutation_lease.clone())).await {
                     rustfs_io_metrics::record_put_object_stage_duration_from(
                         rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC,
                         fsync_started,
@@ -795,7 +811,7 @@ impl LocalDisk {
                         break;
                     }
                     let fsync_started = rustfs_io_metrics::put_stage_timer();
-                    if let Err(err) = os::fsync_dir(dir).await {
+                    if let Err(err) = os::fsync_dir_with_owner(dir, Some(mutation_lease.clone())).await {
                         rustfs_io_metrics::record_put_object_stage_duration_from(
                             rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC,
                             fsync_started,
@@ -1026,7 +1042,15 @@ impl LocalDisk {
                 // rename_all acquires the backup path's namespace lease. Do not
                 // hold a disk admission while acquiring another namespace lock.
                 drop(file_sync_admission.take());
-                if let Err(err) = rename_all(staged_backup, &backup_path, &dst_volume_dir, &self.publication_root).await {
+                if let Err(err) = os::rename_all_with_owner(
+                    staged_backup,
+                    &backup_path,
+                    &dst_volume_dir,
+                    &self.publication_root,
+                    Some(mutation_lease.clone()),
+                )
+                .await
+                {
                     let _ = remove_file_if_exists(staged_backup);
                     return Err(err);
                 }
@@ -1222,14 +1246,18 @@ impl LocalDisk {
         fi: &FileInfo,
         dst_volume: &str,
         dst_path: &str,
+        namespace_owner: Option<Arc<dyn Send + Sync>>,
     ) -> super::super::RenameDataObservation {
-        let mut preflight_rejection = None;
+        let mut state = RenameDataState {
+            namespace_owner,
+            ..Default::default()
+        };
         let result = self
-            .rename_data_inner(src_volume, src_path, fi.clone(), dst_volume, dst_path, &mut preflight_rejection)
+            .rename_data_inner(src_volume, src_path, fi.clone(), dst_volume, dst_path, &mut state)
             .await;
         super::super::RenameDataObservation {
             result,
-            preflight_rejection,
+            preflight_rejection: state.preflight_rejection,
         }
     }
 }

crates/ecstore/src/disk/mod.rs

@@ -75,6 +75,14 @@ use time::OffsetDateTime;
 use tokio::io::{AsyncRead, AsyncWrite};
 use uuid::Uuid;
 
+/// Independent admission and physical ownership for one disk rename.
+#[derive(Default)]
+pub(crate) struct RenameDataGuards {
+    pub(crate) scanner_publication_lease_token: Option<Uuid>,
+    pub(crate) external_guard: Option<Arc<dyn Send + Sync>>,
+    pub(crate) namespace_owner: Option<Arc<dyn Send + Sync>>,
+}
+
 /// Local preflight evidence stays outside DiskAPI and the RPC response format.
 pub(crate) struct RenameDataObservation {
     pub(crate) result: Result<RenameDataResp>,
@@ -724,6 +732,25 @@ impl Disk {
         }
     }
 
+    /// Keep local undo publication owned independently of the wrapper deadline.
+    /// Remote undo retains its existing RPC contract; this is not a remote drain proof.
+    pub(crate) async fn undo_write_with_namespace_owner(
+        &self,
+        volume: &str,
+        path: &str,
+        fi: FileInfo,
+        opts: DeleteOptions,
+        namespace_owner: Option<Arc<dyn Send + Sync>>,
+    ) -> Result<()> {
+        match self {
+            Self::Local(disk) => {
+                disk.undo_write_with_namespace_owner(volume, path, fi, opts, namespace_owner)
+                    .await
+            }
+            Self::Remote(disk) => disk.delete_version(volume, path, fi, false, opts).await,
+        }
+    }
+
     pub(crate) async fn rename_data_borrowed(
         &self,
         src_volume: &str,
@@ -743,12 +770,12 @@ impl Disk {
         fi: &FileInfo,
         dst_volume: &str,
         dst_path: &str,
-        scanner_publication_lease_token: Option<Uuid>,
+        guards: RenameDataGuards,
     ) -> RenameDataObservation {
         match self {
             Disk::Local(local_disk) => {
                 local_disk
-                    .rename_data_observed(src_volume, src_path, fi, dst_volume, dst_path, None)
+                    .rename_data_observed_with_guards(src_volume, src_path, fi, dst_volume, dst_path, guards)
                     .await
             }
             Disk::Remote(remote_disk) => RenameDataObservation::unknown(
@@ -759,7 +786,7 @@ impl Disk {
                         fi,
                         dst_volume,
                         dst_path,
-                        scanner_publication_lease_token,
+                        guards.scanner_publication_lease_token,
                     )
                     .await,
             ),

crates/ecstore/src/disk/os.rs

@@ -246,6 +246,50 @@ pub(crate) mod fsync_dir_recorder {
     }
 }
 
+/// Pause a real namespace mutation inside its physical executor.
+#[cfg(all(test, not(windows)))]
+pub(crate) mod prepared_publication_test_hooks {
+    use super::*;
+
+    #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+    pub(crate) enum Stage {
+        PreparedRename,
+        Rename,
+        Remove,
+        Rollback,
+        DirFsync,
+    }
+
+    type Hook = Box<dyn FnOnce() + Send>;
+    type Key = (Stage, PathBuf);
+    static BEFORE_PUBLICATION: LazyLock<Mutex<HashMap<Key, Hook>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
+
+    pub(crate) struct Guard(Key);
+
+    impl Drop for Guard {
+        fn drop(&mut self) {
+            BEFORE_PUBLICATION.lock().remove(&self.0);
+        }
+    }
+
+    pub(crate) fn install(path: &Path, hook: impl FnOnce() + Send + 'static) -> Guard {
+        install_at(Stage::PreparedRename, path, hook)
+    }
+
+    pub(crate) fn install_at(stage: Stage, path: &Path, hook: impl FnOnce() + Send + 'static) -> Guard {
+        let key = (stage, path.to_path_buf());
+        assert!(BEFORE_PUBLICATION.lock().insert(key.clone(), Box::new(hook)).is_none());
+        Guard(key)
+    }
+
+    pub(crate) fn run(stage: Stage, path: &Path) {
+        let hook = BEFORE_PUBLICATION.lock().remove(&(stage, path.to_path_buf()));
+        if let Some(hook) = hook {
+            hook();
+        }
+    }
+}
+
 #[cfg(all(test, windows))]
 pub(crate) mod windows_rename_test_hooks {
     use super::*;
@@ -580,6 +624,7 @@ impl OpenedDstDirFsyncGroup {
 }
 
 struct DstDirFsyncWaiter {
+    namespace_owner: Option<Arc<dyn Send + Sync>>,
     result_tx: oneshot::Sender<SharedDstDirFsyncResult>,
 }
 
@@ -638,6 +683,7 @@ impl DstDirFsyncGroupCommit {
     fn enqueue_opened(
         &self,
         opened: OpenedDstDirFsyncGroup,
+        namespace_owner: Option<Arc<dyn Send + Sync>>,
     ) -> io::Result<(oneshot::Receiver<SharedDstDirFsyncResult>, Option<Arc<DstDirFsyncGroup>>)> {
         let (result_tx, result_rx) = oneshot::channel();
         let mut registry = self.inner.lock();
@@ -668,7 +714,10 @@ impl DstDirFsyncGroupCommit {
             group
         };
         let mut group_state = group.inner.lock();
-        group_state.pending.push_back(DstDirFsyncWaiter { result_tx });
+        group_state.pending.push_back(DstDirFsyncWaiter {
+            result_tx,
+            namespace_owner,
+        });
         let start_worker = !group_state.worker_running;
         if start_worker {
             group_state.worker_running = true;
@@ -690,7 +739,13 @@ impl DstDirFsyncGroupCommit {
     fn remove_idle_group(&self, group: &Arc<DstDirFsyncGroup>) {
         let mut registry = self.inner.lock();
         let group_state = group.inner.lock();
-        if !group_state.worker_running && group_state.pending.is_empty() {
+        if !group_state.worker_running
+            && group_state.pending.is_empty()
+            && registry
+                .groups
+                .get(&group.key)
+                .is_some_and(|registered| Arc::ptr_eq(registered, group))
+        {
             registry.groups.remove(&group.key);
         }
     }
@@ -713,16 +768,20 @@ impl DstDirFsyncGroupCommit {
         &self,
         dir: &Path,
     ) -> io::Result<(oneshot::Receiver<SharedDstDirFsyncResult>, Option<Arc<DstDirFsyncGroup>>)> {
-        self.enqueue_opened(OpenedDstDirFsyncGroup::open(dir)?)
+        self.enqueue_opened(OpenedDstDirFsyncGroup::open(dir)?, None)
     }
 }
 
 #[cfg(unix)]
-async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup) -> io::Result<()> {
+async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup, namespace_owners: Vec<Arc<dyn Send + Sync>>) -> io::Result<()> {
     #[cfg(test)]
     let dir = group.dir.clone();
     let dir_file = group.dir_file.clone();
     fsync_spawn_blocking(move || {
+        // The batch worker may be cancelled while this syscall is still running.
+        let _namespace_owners = namespace_owners;
+        #[cfg(all(test, not(windows)))]
+        prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::DirFsync, &dir);
         #[cfg(test)]
         {
             if let Some(kind) = fsync_dir_recorder::take_grouped_failure(&dir) {
@@ -737,66 +796,118 @@ async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup) -> io::Result<()> {
 }
 
 #[cfg(not(unix))]
-async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup) -> io::Result<()> {
+async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup, namespace_owners: Vec<Arc<dyn Send + Sync>>) -> io::Result<()> {
+    let _namespace_owners = namespace_owners;
     fsync_dir(&group.dir).await
 }
 
-async fn run_dst_dir_fsync_group_worker(group: Arc<DstDirFsyncGroup>) {
-    loop {
-        #[cfg(test)]
-        fsync_dir_recorder::run_before_group_batch(&group.dir);
-        tokio::task::yield_now().await;
-        let batch: Vec<DstDirFsyncWaiter> = {
-            let mut group_state = group.inner.lock();
-            group_state.pending.drain(..).collect()
-        };
-        if batch.is_empty() {
-            let mut group_state = group.inner.lock();
+struct DstDirFsyncWorkerGuard {
+    group: Arc<DstDirFsyncGroup>,
+    in_flight: usize,
+    armed: bool,
+}
+
+impl Drop for DstDirFsyncWorkerGuard {
+    fn drop(&mut self) {
+        if !self.armed {
+            return;
+        }
+        // Cancellation must release queued owners, but the physical batch keeps
+        // its own owners until its blocking syscall returns.
+        let pending = {
+            let mut registry = DST_DIR_FSYNC_GROUP_COMMIT.inner.lock();
+            let mut group_state = self.group.inner.lock();
+            let pending = std::mem::take(&mut group_state.pending);
             group_state.worker_running = false;
-            drop(group_state);
-            DST_DIR_FSYNC_GROUP_COMMIT.remove_idle_group(&group);
-            return;
-        }
-
-        #[cfg(test)]
-        fsync_dir_recorder::record_grouped(&group.dir, batch.len());
-        let result = fsync_open_dst_dir_group(&group)
-            .await
-            .map_err(SharedDstDirFsyncError::from_error);
-        let batch_len = batch.len();
-        DST_DIR_FSYNC_GROUP_COMMIT.complete_batch(batch_len);
-
-        let should_stop = {
-            let mut group_state = group.inner.lock();
-            if group_state.pending.is_empty() {
-                group_state.worker_running = false;
-                true
-            } else {
-                false
+            if registry
+                .groups
+                .get(&self.group.key)
+                .is_some_and(|group| Arc::ptr_eq(group, &self.group))
+            {
+                registry.total_waiters = registry.total_waiters.saturating_sub(pending.len() + self.in_flight);
+                registry.groups.remove(&self.group.key);
             }
+            pending
         };
-        if should_stop {
-            DST_DIR_FSYNC_GROUP_COMMIT.remove_idle_group(&group);
-        }
-        for waiter in batch {
-            let _ = waiter.result_tx.send(result.clone());
-        }
-        if should_stop {
-            return;
+        // Lease and channel destructors must run outside the registry locks.
+        drop(pending);
+    }
+}
+
+fn run_dst_dir_fsync_group_worker(group: Arc<DstDirFsyncGroup>) -> impl std::future::Future<Output = ()> {
+    // Capture before spawning: shutdown may drop the future without polling it.
+    let worker_guard = DstDirFsyncWorkerGuard {
+        group: group.clone(),
+        in_flight: 0,
+        armed: true,
+    };
+    async move {
+        let mut worker_guard = worker_guard;
+        loop {
+            #[cfg(test)]
+            fsync_dir_recorder::run_before_group_batch(&group.dir);
+            tokio::task::yield_now().await;
+            let mut batch: Vec<DstDirFsyncWaiter> = {
+                let mut group_state = group.inner.lock();
+                group_state.pending.drain(..).collect()
+            };
+            if batch.is_empty() {
+                let mut group_state = group.inner.lock();
+                worker_guard.armed = false;
+                group_state.worker_running = false;
+                drop(group_state);
+                DST_DIR_FSYNC_GROUP_COMMIT.remove_idle_group(&group);
+                return;
+            }
+            worker_guard.in_flight = batch.len();
+
+            #[cfg(test)]
+            fsync_dir_recorder::record_grouped(&group.dir, batch.len());
+            let namespace_owners = batch.iter_mut().filter_map(|waiter| waiter.namespace_owner.take()).collect();
+            let result = fsync_open_dst_dir_group(&group, namespace_owners)
+                .await
+                .map_err(SharedDstDirFsyncError::from_error);
+            let batch_len = batch.len();
+            DST_DIR_FSYNC_GROUP_COMMIT.complete_batch(batch_len);
+            worker_guard.in_flight = 0;
+
+            let should_stop = {
+                let mut group_state = group.inner.lock();
+                if group_state.pending.is_empty() {
+                    worker_guard.armed = false;
+                    group_state.worker_running = false;
+                    true
+                } else {
+                    false
+                }
+            };
+            if should_stop {
+                DST_DIR_FSYNC_GROUP_COMMIT.remove_idle_group(&group);
+            }
+            for waiter in batch {
+                let _ = waiter.result_tx.send(result.clone());
+            }
+            if should_stop {
+                return;
+            }
         }
     }
 }
 
-async fn fsync_dst_dir_group_commit_with_enabled(dir: impl AsRef<Path>, enabled: bool) -> io::Result<()> {
+async fn fsync_dst_dir_group_commit_with_enabled(
+    dir: impl AsRef<Path>,
+    enabled: bool,
+    namespace_owner: Option<Arc<dyn Send + Sync>>,
+) -> io::Result<()> {
     if !enabled {
-        return fsync_dir(dir).await;
+        return fsync_dir_with_owner(dir.as_ref(), namespace_owner).await;
     }
 
     let dir = dir.as_ref().to_path_buf();
     let opened = tokio::task::spawn_blocking(move || OpenedDstDirFsyncGroup::open(&dir))
         .await
         .map_err(|err| io::Error::other(format!("blocking dst dir group open failed: {err}")))??;
-    let (result_rx, worker) = DST_DIR_FSYNC_GROUP_COMMIT.enqueue_opened(opened)?;
+    let (result_rx, worker) = DST_DIR_FSYNC_GROUP_COMMIT.enqueue_opened(opened, namespace_owner)?;
     if let Some(group) = worker {
         tokio::spawn(run_dst_dir_fsync_group_worker(group));
     }
@@ -808,8 +919,11 @@ async fn fsync_dst_dir_group_commit_with_enabled(dir: impl AsRef<Path>, enabled:
     }
 }
 
-pub(crate) async fn fsync_dst_dir_group_commit(dir: impl AsRef<Path>) -> io::Result<()> {
-    fsync_dst_dir_group_commit_with_enabled(dir, dst_dir_fsync_group_commit_enabled()).await
+pub(crate) async fn fsync_dst_dir_group_commit(
+    dir: impl AsRef<Path>,
+    namespace_owner: Option<Arc<dyn Send + Sync>>,
+) -> io::Result<()> {
+    fsync_dst_dir_group_commit_with_enabled(dir, dst_dir_fsync_group_commit_enabled(), namespace_owner).await
 }
 
 pub(crate) async fn fsync_dst_dir_group_commit_or_namespace_file_sync_limit(
@@ -818,7 +932,7 @@ pub(crate) async fn fsync_dst_dir_group_commit_or_namespace_file_sync_limit(
     admission: &FileSyncAdmission,
 ) -> io::Result<()> {
     if dst_dir_fsync_group_commit_enabled() {
-        fsync_dst_dir_group_commit_with_enabled(dir, true).await
+        fsync_dst_dir_group_commit_with_enabled(dir, true, Some(lease)).await
     } else {
         fsync_dir_with_namespace_file_sync_limit(dir, lease, admission).await
     }
@@ -826,7 +940,7 @@ pub(crate) async fn fsync_dst_dir_group_commit_or_namespace_file_sync_limit(
 
 #[cfg(test)]
 pub(crate) async fn fsync_dst_dir_group_commit_for_test(dir: impl AsRef<Path>, enabled: bool) -> io::Result<()> {
-    fsync_dst_dir_group_commit_with_enabled(dir, enabled).await
+    fsync_dst_dir_group_commit_with_enabled(dir, enabled, None).await
 }
 
 #[cfg(test)]
@@ -1233,6 +1347,8 @@ pub(crate) struct NamespaceMutationLease {
     _namespace_guard: OwnedMutexGuard<()>,
     _volume_guard: Option<OwnedRwLockReadGuard<()>>,
     external_guard: Mutex<Option<Arc<dyn Send + Sync>>>,
+    // Independent of the quota claim; both survive cancellation of the waiter.
+    _namespace_owner: Option<Arc<dyn Send + Sync>>,
 }
 
 impl NamespaceMutationLease {
@@ -1242,10 +1358,18 @@ impl NamespaceMutationLease {
 }
 
 async fn acquire_namespace_mutation_lease(path: &Path) -> Arc<NamespaceMutationLease> {
+    acquire_namespace_mutation_lease_with_owner(path, None).await
+}
+
+async fn acquire_namespace_mutation_lease_with_owner(
+    path: &Path,
+    namespace_owner: Option<Arc<dyn Send + Sync>>,
+) -> Arc<NamespaceMutationLease> {
     Arc::new(NamespaceMutationLease {
         _namespace_guard: disk_namespace_mutation_lock(path).lock_owned().await,
         _volume_guard: None,
         external_guard: Mutex::new(None),
+        _namespace_owner: namespace_owner,
     })
 }
 
@@ -1255,6 +1379,15 @@ pub(crate) async fn acquire_rename_data_mutation_lease(
     root: &Path,
     volume: &str,
     destination_object: &Path,
+) -> Arc<NamespaceMutationLease> {
+    acquire_rename_data_mutation_lease_with_owner(root, volume, destination_object, None).await
+}
+
+pub(crate) async fn acquire_rename_data_mutation_lease_with_owner(
+    root: &Path,
+    volume: &str,
+    destination_object: &Path,
+    namespace_owner: Option<Arc<dyn Send + Sync>>,
 ) -> Arc<NamespaceMutationLease> {
     let namespace_guard = disk_namespace_mutation_lock(destination_object).lock_owned().await;
     let volume_guard = disk_volume_mutation_lock(root, volume).read_owned().await;
@@ -1262,6 +1395,7 @@ pub(crate) async fn acquire_rename_data_mutation_lease(
         _namespace_guard: namespace_guard,
         _volume_guard: Some(volume_guard),
         external_guard: Mutex::new(None),
+        _namespace_owner: namespace_owner,
     })
 }
 
@@ -1751,6 +1885,69 @@ pub async fn rename_all(
     Ok(())
 }
 
+pub(crate) async fn fsync_dir_with_owner(path: &Path, namespace_owner: Option<Arc<dyn Send + Sync>>) -> io::Result<()> {
+    #[cfg(unix)]
+    {
+        if namespace_owner.is_none() {
+            return fsync_dir(path).await;
+        }
+        let path = path.to_path_buf();
+        fsync_spawn_blocking(move || {
+            let _namespace_owner = namespace_owner;
+            fsync_dir_std(path)
+        })
+        .await?
+    }
+    #[cfg(not(unix))]
+    {
+        let _ = namespace_owner;
+        fsync_dir(path).await
+    }
+}
+
+/// Retain namespace ownership in the actual filesystem executor after timeout.
+pub(crate) async fn remove_file_with_owner(
+    path: impl AsRef<Path>,
+    namespace_owner: Option<Arc<dyn Send + Sync>>,
+) -> io::Result<()> {
+    if namespace_owner.is_none() {
+        return tokio::fs::remove_file(path).await;
+    }
+    let path = path.as_ref().to_path_buf();
+    let lease = acquire_namespace_mutation_lease_with_owner(&path, namespace_owner).await;
+    run_blocking_namespace_operation(lease, move || {
+        #[cfg(all(test, not(windows)))]
+        prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Remove, &path);
+        std::fs::remove_file(path)
+    })
+    .await
+}
+
+/// Retain namespace ownership in the actual filesystem executor after timeout.
+pub(crate) async fn remove_dir_with_owner(
+    path: impl AsRef<Path>,
+    namespace_owner: Option<Arc<dyn Send + Sync>>,
+) -> io::Result<()> {
+    if namespace_owner.is_none() {
+        return tokio::fs::remove_dir(path).await;
+    }
+    let path = path.as_ref().to_path_buf();
+    let lease = acquire_namespace_mutation_lease_with_owner(&path, namespace_owner).await;
+    run_blocking_namespace_operation(lease, move || std::fs::remove_dir(path)).await
+}
+
+#[tracing::instrument(name = "rename_all", level = "debug", skip_all)]
+pub(crate) async fn rename_all_with_owner(
+    src_file_path: impl AsRef<Path>,
+    dst_file_path: impl AsRef<Path>,
+    base_dir: impl AsRef<Path>,
+    publication_root: &PublicationRoot,
+    namespace_owner: Option<Arc<dyn Send + Sync>>,
+) -> Result<()> {
+    let lease = acquire_namespace_mutation_lease_with_owner(dst_file_path.as_ref(), namespace_owner).await;
+    rename_all_with_lease(src_file_path, dst_file_path, base_dir, publication_root, lease).await
+}
+
 pub(crate) async fn rename_all_with_lease(
     src_file_path: impl AsRef<Path>,
     dst_file_path: impl AsRef<Path>,
@@ -1943,6 +2140,8 @@ pub(crate) async fn rename_all_with_prepared_source(
         move || {
             validate_prepared_rename_source(&prepared_source, &src_file_path)?;
             let preparation = prepare_rename_with_retry(&src_file_path, &dst_file_path, &base_dir, &publication_root)?;
+            #[cfg(test)]
+            prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::PreparedRename, &dst_file_path);
             rename_prepared(&src_file_path, &dst_file_path, &preparation)
         }
     };
@@ -1981,6 +2180,32 @@ pub async fn rename_all_ignore_missing_source(
     }
 }
 
+#[tracing::instrument(name = "rename_all_ignore_missing_source", level = "debug", skip_all)]
+pub(crate) async fn rename_all_ignore_missing_source_with_owner(
+    src_file_path: impl AsRef<Path>,
+    dst_file_path: impl AsRef<Path>,
+    base_dir: impl AsRef<Path>,
+    publication_root: &PublicationRoot,
+    namespace_owner: Option<Arc<dyn Send + Sync>>,
+) -> Result<()> {
+    let src_file_path = src_file_path.as_ref();
+    let lease = acquire_namespace_mutation_lease_with_owner(dst_file_path.as_ref(), namespace_owner).await;
+    match reliable_rename_inner_with_lease(
+        src_file_path.to_path_buf(),
+        dst_file_path.as_ref().to_path_buf(),
+        base_dir.as_ref().to_path_buf(),
+        publication_root.clone(),
+        false,
+        lease,
+    )
+    .await
+    {
+        Ok(()) => Ok(()),
+        Err(err) if err.kind() == io::ErrorKind::NotFound && rename_source_is_missing(src_file_path, publication_root) => Ok(()),
+        Err(err) => Err(to_file_error(err).into()),
+    }
+}
+
 #[cfg(windows)]
 pub(crate) fn rename_source_is_missing(src_file_path: &Path, publication_root: &PublicationRoot) -> bool {
     let Some(source_parent) = src_file_path.parent() else {
@@ -2046,6 +2271,11 @@ async fn reliable_rename_inner_with_lease(
         let base_dir = base_dir.clone();
         move || {
             let preparation = prepare_rename_with_retry(&src_file_path, &dst_file_path, &base_dir, &publication_root)?;
+            #[cfg(all(test, not(windows)))]
+            {
+                prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &src_file_path);
+                prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, &dst_file_path);
+            }
             rename_prepared(&src_file_path, &dst_file_path, &preparation)
         }
     };
@@ -6140,6 +6370,245 @@ mod tests {
         wait_for_dst_dir_fsync_group_commit_idle().await;
     }
 
+    #[cfg(unix)]
+    #[tokio::test]
+    #[serial_test::serial(dst_dir_fsync_group_commit)]
+    async fn grouped_fsync_physical_batch_keeps_all_owners_after_worker_cancellation() {
+        let temp_dir = tempdir().expect("fixture directory");
+        let dir = temp_dir.path().canonicalize().expect("canonical fsync path");
+        let first_ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
+        let second_ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
+        let first_owner = first_ctx.begin_namespace_commit();
+        let second_owner = second_ctx.begin_namespace_commit();
+        let first_probe = Arc::downgrade(&first_owner);
+        let second_probe = Arc::downgrade(&second_owner);
+        let (first_rx, group) = DST_DIR_FSYNC_GROUP_COMMIT
+            .enqueue_opened(
+                OpenedDstDirFsyncGroup::open(&dir).expect("open first waiter directory"),
+                Some(first_owner),
+            )
+            .expect("queue first real waiter");
+        let group = group.expect("first waiter starts the group");
+        let (second_rx, second_worker) = DST_DIR_FSYNC_GROUP_COMMIT
+            .enqueue_opened(
+                OpenedDstDirFsyncGroup::open(&dir).expect("open second waiter directory"),
+                Some(second_owner),
+            )
+            .expect("queue second real waiter");
+        assert!(second_worker.is_none(), "same directory must join the same batch");
+        assert_eq!(group.inner.lock().pending.len(), 2);
+        let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
+        let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
+        let _hook =
+            prepared_publication_test_hooks::install_at(prepared_publication_test_hooks::Stage::DirFsync, &dir, move || {
+                let _ = entered_tx.send(());
+                let _ = release_rx.recv();
+            });
+        let worker = tokio::spawn(run_dst_dir_fsync_group_worker(group.clone()));
+        tokio::time::timeout(Duration::from_secs(5), entered_rx)
+            .await
+            .expect("batch must reach its physical fsync")
+            .expect("physical fsync entry");
+        assert_eq!(fsync_dir_recorder::grouped_batch_sizes(&dir), vec![2]);
+        assert!(
+            group.inner.lock().pending.is_empty(),
+            "both waiters were transferred into the physical batch"
+        );
+        let queued_ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
+        let queued_owner = queued_ctx.begin_namespace_commit();
+        let queued_probe = Arc::downgrade(&queued_owner);
+        let queued_generation = queued_ctx.namespace_commit_generation();
+        let (queued_rx, queued_worker) = DST_DIR_FSYNC_GROUP_COMMIT
+            .enqueue_opened(
+                OpenedDstDirFsyncGroup::open(&dir).expect("open queued waiter directory"),
+                Some(queued_owner),
+            )
+            .expect("queue a waiter after the physical batch was frozen");
+        assert!(queued_worker.is_none());
+        assert_eq!(group.inner.lock().pending.len(), 1);
+        drop((first_rx, second_rx));
+        worker.abort();
+        assert!(worker.await.expect_err("cancel the async batch owner").is_cancelled());
+        assert!(queued_rx.await.is_err(), "an undispatched waiter must observe worker cancellation");
+        assert!(queued_probe.upgrade().is_none());
+        assert!(!queued_ctx.namespace_commits_pending());
+        assert!(queued_ctx.namespace_commit_generation() > queued_generation);
+        assert!(group.inner.lock().pending.is_empty());
+        assert!(!group.inner.lock().worker_running);
+        assert_eq!(DST_DIR_FSYNC_GROUP_COMMIT.counts_for_test(), (0, 0));
+        let first_pending = first_ctx.namespace_commits_pending() && first_probe.upgrade().is_some();
+        let second_pending = second_ctx.namespace_commits_pending() && second_probe.upgrade().is_some();
+        let generations = (first_ctx.namespace_commit_generation(), second_ctx.namespace_commit_generation());
+        drop(release_tx);
+        tokio::time::timeout(Duration::from_secs(5), async {
+            while Arc::strong_count(&group.dir_file) != 1
+                || first_probe.upgrade().is_some()
+                || second_probe.upgrade().is_some()
+                || first_ctx.namespace_commits_pending()
+                || second_ctx.namespace_commits_pending()
+            {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("physical fsync must release every batch owner");
+        assert!(fsync_dir_recorder::was_fsynced(&dir), "the detached syscall must really execute");
+        assert!(
+            first_pending && second_pending,
+            "one physical batch must preserve both independent namespace owners"
+        );
+        assert!(!first_ctx.namespace_commits_pending());
+        assert!(!second_ctx.namespace_commits_pending());
+        assert!(first_ctx.namespace_commit_generation() > generations.0);
+        assert!(second_ctx.namespace_commit_generation() > generations.1);
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    #[serial_test::serial(dst_dir_fsync_group_commit)]
+    async fn grouped_fsync_unpolled_worker_releases_queued_owner() {
+        let temp_dir = tempdir().expect("fixture directory");
+        let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
+        let owner = ctx.begin_namespace_commit();
+        let probe = Arc::downgrade(&owner);
+        let generation = ctx.namespace_commit_generation();
+        let (rx, group) = DST_DIR_FSYNC_GROUP_COMMIT
+            .enqueue_opened(
+                OpenedDstDirFsyncGroup::open(temp_dir.path()).expect("open queued waiter directory"),
+                Some(owner),
+            )
+            .expect("queue a real waiter");
+        let group = group.expect("first waiter starts the group");
+        let worker = run_dst_dir_fsync_group_worker(group.clone());
+        assert!(ctx.namespace_commits_pending());
+        drop(worker);
+        assert!(rx.await.is_err(), "shutdown before first poll must release the waiter");
+        assert!(probe.upgrade().is_none());
+        assert!(!ctx.namespace_commits_pending());
+        assert!(ctx.namespace_commit_generation() > generation);
+        assert!(group.inner.lock().pending.is_empty());
+        assert!(!group.inner.lock().worker_running);
+        assert_eq!(DST_DIR_FSYNC_GROUP_COMMIT.counts_for_test(), (0, 0));
+        assert!(
+            fsync_dir_recorder::grouped_batch_sizes(temp_dir.path()).is_empty(),
+            "the dropped future must not dispatch a physical batch"
+        );
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn stale_idle_group_cleanup_preserves_successor_registration() {
+        let temp_dir = tempdir().expect("fixture directory");
+        let registry = DstDirFsyncGroupCommit::default();
+        let (mut first_rx, first_worker) = registry.enqueue_for_test(temp_dir.path()).expect("enqueue first worker");
+        let old_group = first_worker.expect("first waiter starts a worker");
+        // W1 has completed its batch and marked G idle, but has not cleaned G up.
+        let first_waiter = old_group.inner.lock().pending.pop_front().expect("first batch waiter");
+        registry.complete_batch(1);
+        old_group.inner.lock().worker_running = false;
+
+        let (mut second_rx, second_worker) = registry.enqueue_for_test(temp_dir.path()).expect("enqueue second worker");
+        let reused_group = second_worker.expect("idle G starts another worker");
+        assert!(Arc::ptr_eq(&old_group, &reused_group));
+        let second_waiter = reused_group.inner.lock().pending.pop_front().expect("second batch waiter");
+        registry.complete_batch(1);
+        reused_group.inner.lock().worker_running = false;
+        registry.remove_idle_group(&reused_group);
+        assert_eq!(registry.counts_for_test(), (0, 0), "normal idle cleanup must remove G");
+        assert!(second_waiter.result_tx.send(Ok(())).is_ok());
+        assert!(second_rx.try_recv().expect("second worker reports completion").is_ok());
+
+        let (mut successor_rx, successor_worker) = registry.enqueue_for_test(temp_dir.path()).expect("enqueue successor");
+        let successor = successor_worker.expect("successor starts a new group");
+        assert!(!Arc::ptr_eq(&old_group, &successor));
+        assert_eq!(registry.counts_for_test(), (1, 1));
+        // W1 resumes with its old Arc after W2 removed G and W3 installed G2.
+        registry.remove_idle_group(&old_group);
+        assert!(first_waiter.result_tx.send(Ok(())).is_ok());
+        assert!(first_rx.try_recv().expect("first worker reports completion").is_ok());
+        assert!(
+            registry
+                .inner
+                .lock()
+                .groups
+                .get(&successor.key)
+                .is_some_and(|registered| Arc::ptr_eq(registered, &successor)),
+            "stale cleanup must retain the exact successor Arc"
+        );
+        assert_eq!(registry.counts_for_test(), (1, 1));
+        assert!(successor.inner.lock().worker_running);
+        assert_eq!(successor.inner.lock().pending.len(), 1);
+        assert!(matches!(successor_rx.try_recv(), Err(oneshot::error::TryRecvError::Empty)));
+
+        let (_joined_rx, new_worker) = registry.enqueue_for_test(temp_dir.path()).expect("join successor");
+        assert!(new_worker.is_none(), "a later waiter must join G2 instead of creating G3");
+        assert_eq!(successor.inner.lock().pending.len(), 2);
+        assert_eq!(registry.counts_for_test(), (1, 2));
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    #[serial_test::serial(dst_dir_fsync_group_commit)]
+    async fn stale_idle_cleanup_then_unpolled_worker_drop_releases_waiter_budget() {
+        wait_for_dst_dir_fsync_group_commit_idle().await;
+        let temp_dir = tempdir().expect("fixture directory");
+        let (old_rx, old_worker) = DST_DIR_FSYNC_GROUP_COMMIT
+            .enqueue_for_test(temp_dir.path())
+            .expect("enqueue old group");
+        let old_group = old_worker.expect("old group starts a worker");
+        tokio::time::timeout(Duration::from_secs(5), run_dst_dir_fsync_group_worker(old_group.clone()))
+            .await
+            .expect("old worker must finish its actual fsync");
+        assert!(old_rx.await.expect("old worker reports completion").is_ok());
+        assert!(fsync_dir_recorder::was_fsynced(temp_dir.path()));
+        assert_eq!(fsync_dir_recorder::grouped_batch_sizes(temp_dir.path()), vec![1]);
+        assert_eq!(DST_DIR_FSYNC_GROUP_COMMIT.counts_for_test(), (0, 0));
+
+        let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
+        let owner = ctx.begin_namespace_commit();
+        let probe = Arc::downgrade(&owner);
+        let generation = ctx.namespace_commit_generation();
+        let (rx, successor_worker) = DST_DIR_FSYNC_GROUP_COMMIT
+            .enqueue_opened(
+                OpenedDstDirFsyncGroup::open(temp_dir.path()).expect("open successor directory"),
+                Some(owner),
+            )
+            .expect("enqueue successor owner");
+        let successor = successor_worker.expect("successor starts a new group");
+        assert!(!Arc::ptr_eq(&old_group, &successor));
+        let worker = run_dst_dir_fsync_group_worker(successor.clone());
+        // The stale Arc represents W1 resuming after another worker removed G.
+        DST_DIR_FSYNC_GROUP_COMMIT.remove_idle_group(&old_group);
+        assert!(ctx.namespace_commits_pending());
+        assert!(probe.upgrade().is_some());
+        drop(worker);
+        let channel_closed = tokio::time::timeout(Duration::from_secs(5), rx)
+            .await
+            .expect("dropping the unpolled worker must release its channel")
+            .is_err();
+        let counts_after_drop = DST_DIR_FSYNC_GROUP_COMMIT.counts_for_test();
+        let owner_released = probe.upgrade().is_none();
+        let namespace_pending = ctx.namespace_commits_pending();
+        let generation_after_drop = ctx.namespace_commit_generation();
+        let successor_pending = successor.inner.lock().pending.len();
+        let worker_running = successor.inner.lock().worker_running;
+        // Preserve the observed result before cleanup, so a RED run cannot leak
+        // its phantom count into unrelated tests in the same process.
+        clear_dst_dir_fsync_group_commit_for_test();
+        assert!(channel_closed);
+        assert!(owner_released);
+        assert!(!namespace_pending);
+        assert!(generation_after_drop > generation);
+        assert_eq!(successor_pending, 0);
+        assert!(!worker_running);
+        assert_eq!(
+            fsync_dir_recorder::grouped_batch_sizes(temp_dir.path()),
+            vec![1],
+            "dropping the successor before its first poll must not dispatch another fsync"
+        );
+        assert_eq!(counts_after_drop, (0, 0), "stale cleanup must not strand a phantom waiter");
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     #[serial_test::serial(dst_dir_fsync_group_commit)]
     async fn dst_dir_fsync_group_commit_cancellation_releases_waiter_state() {

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -3663,22 +3663,22 @@ async fn rollback_failed_rename(
             let object = object.to_string();
             let disk_namespace_commit_guard = namespace_commit_guard.clone();
             let task = tokio::spawn(async move {
-                let _namespace_commit_guard = disk_namespace_commit_guard;
+                let _namespace_commit_guard = disk_namespace_commit_guard.clone();
                 #[allow(clippy::let_unit_value)]
                 let _task_guard = SetDisks::rename_fanout_task_guard(&object);
                 SetDisks::rename_fanout_barrier(&object, disk_index, rename_fanout_barrier_phase::ROLLBACK).await;
                 #[cfg(test)]
                 rollback_fault_injection::before_undo(&object, disk_index)?;
-                disk.delete_version(
+                disk.undo_write_with_namespace_owner(
                     &bucket,
                     &object,
                     fi,
-                    false,
                     DeleteOptions {
                         undo_write: true,
                         old_data_dir: rollback_dir,
                         ..Default::default()
                     },
+                    disk_namespace_commit_guard.map(|owner| owner as Arc<dyn Send + Sync>),
                 )
                 .await
             });
@@ -4238,7 +4238,7 @@ impl SetDisks {
                     let successful_rename_completion_rank = successful_rename_completion_rank.clone();
                     let namespace_commit_guard = namespace_commit_guard.clone();
                     tasks.spawn(async move {
-                        let _namespace_commit_guard = namespace_commit_guard;
+                        let _namespace_commit_guard = namespace_commit_guard.clone();
                         let mut dispatch_state = RenameDispatchState::NotDispatched;
                         let result = std::panic::AssertUnwindSafe(async {
                             #[allow(clippy::let_unit_value)]
@@ -4273,7 +4273,13 @@ impl SetDisks {
                                     &file_info,
                                     &dst_bucket,
                                     &dst_object,
-                                    scanner_publication_lease_token,
+                                    crate::disk::RenameDataGuards {
+                                        scanner_publication_lease_token,
+                                        namespace_owner: namespace_commit_guard
+                                            .clone()
+                                            .map(|owner| owner as Arc<dyn Send + Sync>),
+                                        ..Default::default()
+                                    },
                                 )
                                 .await;
                             let rejected_before_publication = observed.rejected_before_publication();
@@ -4602,7 +4608,7 @@ impl SetDisks {
             // Keep the storage-owned movement permit attached to the actual
             // fan-out owner, even if the caller future is cancelled.
             let _fanout_publication_scope = fanout_publication_scope;
-            let _namespace_commit_guard = fanout_namespace_commit_guard;
+            let _namespace_commit_guard = fanout_namespace_commit_guard.clone();
             let successful_rename_completion_rank =
                 rustfs_io_metrics::put_stage_metrics_enabled().then(|| Arc::new(AtomicUsize::new(0)));
             let futures = fanout_disks
@@ -4617,6 +4623,7 @@ impl SetDisks {
                     let dst_bucket = fanout_dst_bucket.clone();
                     let successful_rename_completion_rank = successful_rename_completion_rank.clone();
                     let publication_scope = scanner_publication_commit_scope.clone();
+                    let namespace_commit_guard = fanout_namespace_commit_guard.clone();
 
                     async move {
                         let mut dispatch_state = RenameDispatchState::NotDispatched;
@@ -4669,7 +4676,13 @@ impl SetDisks {
                                     file_info,
                                     &dst_bucket,
                                     &dst_object,
-                                    scanner_publication_lease_token,
+                                    crate::disk::RenameDataGuards {
+                                        scanner_publication_lease_token,
+                                        namespace_owner: namespace_commit_guard
+                                            .clone()
+                                            .map(|owner| owner as Arc<dyn Send + Sync>),
+                                        ..Default::default()
+                                    },
                                 )
                                 .await;
                             let rejected_before_publication = observed.rejected_before_publication();
@@ -10860,6 +10873,358 @@ mod tests {
         .await;
     }
 
+    #[cfg(not(windows))]
+    async fn assert_namespace_owner_survives_physical_publication_timeout(allow_early_ack: bool) {
+        use crate::disk::os;
+        use futures::FutureExt;
+
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("60")),
+                (ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true")),
+            ],
+            async {
+                const DISKS: usize = 4;
+                let bucket = "namespace-physical-tail";
+                let object = "inline-overwrite";
+                let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
+                let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
+                prepare_rename_source_dirs(&dirs, &disks, "source").await;
+                let mut old = metadata_test_fileinfo(object);
+                old.mod_time = Some(OffsetDateTime::now_utc());
+                old.size = 15;
+                old.parts.clear();
+                old.add_object_part(1, "old-etag".to_string(), 15, None, 15, None, None);
+                old.data = Some(Bytes::from_static(b"old-inline-body"));
+                old.set_inline_data();
+                old.metadata.insert("etag".to_string(), "old-etag".to_string());
+                let mut infos = rename_commit_fileinfos(object, DISKS, "new-etag");
+                let mut hooks = Vec::new();
+                let mut entered = Vec::new();
+                let mut releases = Vec::new();
+                let mut publication_paths = Vec::new();
+                for (disk, info) in disks.iter().flatten().zip(&mut infos) {
+                    disk.write_metadata(bucket, bucket, object, old.clone())
+                        .await
+                        .expect("the old inline version must be readable before overwrite");
+                    info.size = 11;
+                    info.parts.clear();
+                    info.add_object_part(1, "new-etag".to_string(), 11, None, 11, None, None);
+                    let crate::disk::Disk::Local(local) = disk.as_ref() else {
+                        panic!("physical publication fixture requires local disks");
+                    };
+                    // Linux IO paths use a mount FD, which is also the namespace lock key.
+                    let destination = local
+                        .get_disk()
+                        .get_object_path_for_io(bucket, object)
+                        .expect("the publication path must resolve through the disk's mount lease");
+                    let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
+                    let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
+                    hooks.push(os::prepared_publication_test_hooks::install(
+                        &destination.join(STORAGE_FORMAT_FILE),
+                        move || {
+                            let _ = entered_tx.send(());
+                            // Sender drop also releases the syscall when an earlier assertion fails.
+                            let _ = release_rx.recv();
+                        },
+                    ));
+                    entered.push(entered_rx);
+                    releases.push(release_tx);
+                    publication_paths.push(destination);
+                }
+                let namespace_owner = ctx.begin_namespace_commit();
+                let namespace_probe = Arc::downgrade(&namespace_owner);
+                let receipt = RenameRollbackReceipt::default();
+                let mut rename = Box::pin(SetDisks::rename_data_owned_with_fence(
+                    &disks,
+                    (RUSTFS_META_TMP_BUCKET, "source"),
+                    infos,
+                    (bucket, object),
+                    allow_early_ack,
+                    RenameDataFenceOptions::new(3, None)
+                        .with_rollback_receipt(receipt.clone())
+                        .with_namespace_commit_guard(Some(namespace_owner)),
+                ));
+                tokio::time::timeout(Duration::from_secs(10), async {
+                    tokio::select! {
+                        signals = join_all(entered) => {
+                            assert!(signals.into_iter().all(|signal| signal.is_ok()), "all physical publishers must enter");
+                        }
+                        _ = rename.as_mut() => panic!("rename must not finish before physical publication is paused"),
+                    }
+                })
+                .await
+                .expect("all four prepared metadata renames must reach their blocking syscall");
+                assert!(ctx.namespace_commits_pending());
+                assert_eq!(ctx.namespace_commit_generation(), 1);
+
+                // Every wrapper timer exists before advancing; the physical closures stay blocked.
+                tokio::time::pause();
+                tokio::time::advance(Duration::from_secs(61)).await;
+                tokio::time::resume();
+                let result = tokio::time::timeout(Duration::from_secs(5), rename)
+                    .await
+                    .expect("ordinary disk timeout must not wait for the physical rename");
+                assert!(result.is_err(), "four timed-out disks cannot satisfy write quorum");
+                let report = receipt.0.get().expect("failed fanout must finish rollback accounting");
+                assert_eq!(report.disks.len(), DISKS);
+                assert!(
+                    report
+                        .disks
+                        .iter()
+                        .all(|disk| matches!(disk.outcome, RenameRollbackOutcome::Indeterminate(DiskError::Timeout)))
+                );
+                let pending_before_release = ctx.namespace_commits_pending();
+                let owner_alive_before_release = namespace_probe.upgrade().is_some();
+                let old_snapshot_generation = ctx.namespace_commit_generation();
+                for (disk, destination) in disks.iter().flatten().zip(&publication_paths) {
+                    let root = disk.path();
+                    assert!(
+                        os::acquire_rename_data_mutation_lease(&root, bucket, destination)
+                            .now_or_never()
+                            .is_none(),
+                        "the physical publication must still own object serialization after the async timeout"
+                    );
+                    assert!(
+                        root.join(RUSTFS_META_TMP_BUCKET)
+                            .join("source")
+                            .join(STORAGE_FORMAT_FILE)
+                            .exists()
+                    );
+                    let stored = disk
+                        .read_version(
+                            "",
+                            bucket,
+                            object,
+                            "",
+                            &ReadOptions {
+                                read_data: true,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                        .expect("a scanner can still read the complete old metadata while publication is paused");
+                    assert_eq!(stored.size, 15);
+                    assert_eq!(stored.data.as_deref(), Some(b"old-inline-body".as_slice()));
+                }
+                assert_eq!(ctx.namespace_commit_generation(), old_snapshot_generation);
+
+                // Drain real syscalls before checking the regression, including on the RED run.
+                drop(releases);
+                for (disk, destination) in disks.iter().flatten().zip(&publication_paths) {
+                    let root = disk.path();
+                    let lease = tokio::time::timeout(
+                        Duration::from_secs(5),
+                        os::acquire_rename_data_mutation_lease(&root, bucket, destination),
+                    )
+                    .await
+                    .expect("released physical publishers must drain");
+                    drop(lease);
+                }
+                for dir in &dirs {
+                    let reopened = reopen_local_disk(dir).await;
+                    let stored = reopened
+                        .read_version(
+                            "",
+                            bucket,
+                            object,
+                            "",
+                            &ReadOptions {
+                                read_data: true,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                        .expect("the detached prepared rename must actually publish after timeout");
+                    assert_eq!(stored.size, 11);
+                    assert_eq!(stored.data.as_deref(), Some(b"inline-body".as_slice()));
+                }
+                // The lease releases its locks before dropping the namespace owner, and the
+                // owner's `Drop` runs after its `Weak` probe stops upgrading, so wait for the
+                // pending counter itself instead of asserting it right after the drain.
+                tokio::time::timeout(Duration::from_secs(5), async {
+                    while ctx.namespace_commits_pending() || namespace_probe.upgrade().is_some() {
+                        tokio::task::yield_now().await;
+                    }
+                })
+                .await
+                .expect("released physical publishers must release namespace ownership");
+                let generation_after_publication = ctx.namespace_commit_generation();
+                assert!(!ctx.namespace_commits_pending());
+                assert!(namespace_probe.upgrade().is_none());
+                assert!(receipt.is_incomplete(), "late publication must not erase failed-write recovery evidence");
+                assert!(
+                    pending_before_release && owner_alive_before_release,
+                    "physical publication outlived namespace accounting: early_ack={allow_early_ack}, \
+                     pending={pending_before_release}, owner_alive={owner_alive_before_release}, \
+                     old_snapshot_generation={old_snapshot_generation}, after_late_publication={generation_after_publication}"
+                );
+                assert!(
+                    generation_after_publication > old_snapshot_generation,
+                    "physical completion must invalidate the scanner's old metadata snapshot"
+                );
+            },
+        )
+        .await;
+    }
+
+    #[cfg(not(windows))]
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn rename_full_wait_timeout_keeps_namespace_owner_until_physical_publication() {
+        assert_namespace_owner_survives_physical_publication_timeout(false).await;
+    }
+
+    #[cfg(not(windows))]
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn rename_early_ack_timeout_keeps_namespace_owner_until_physical_publication() {
+        assert_namespace_owner_survives_physical_publication_timeout(true).await;
+    }
+
+    #[cfg(not(windows))]
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn successful_rename_ack_keeps_physical_tail_owner_after_caller_cancellation() {
+        use crate::disk::os;
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("60")),
+                (ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true")),
+            ],
+            async {
+                let bucket = "physical-ack-tail";
+                let object = "ack-object";
+                let (dirs, disks) = call_counter_local_disks(bucket, 4).await;
+                prepare_rename_source_dirs(&dirs, &disks, "source").await;
+                let mut infos = rename_commit_fileinfos(object, 4, "new-etag");
+                for info in &mut infos {
+                    info.size = 11;
+                    info.parts.clear();
+                    info.add_object_part(1, "new-etag".to_string(), 11, None, 11, None, None);
+                }
+                let disk = disks[3].as_ref().expect("tail disk");
+                let crate::disk::Disk::Local(local) = disk.as_ref() else {
+                    panic!("local fixture");
+                };
+                let destination = local.get_disk().get_object_path_for_io(bucket, object).expect("tail IO path");
+                let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
+                let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
+                let _hook = os::prepared_publication_test_hooks::install(&destination.join(STORAGE_FORMAT_FILE), move || {
+                    let _ = entered_tx.send(());
+                    let _ = release_rx.recv();
+                });
+                let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
+                let owner = ctx.begin_namespace_commit();
+                let owner_probe = Arc::downgrade(&owner);
+                let receipt = RenameRollbackReceipt::default();
+                let caller_receipt = receipt.clone();
+                let caller_disks = disks.clone();
+                let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();
+                let caller = tokio::spawn(async move {
+                    let commit = SetDisks::rename_data_owned_with_fence(
+                        &caller_disks,
+                        (RUSTFS_META_TMP_BUCKET, "source"),
+                        infos,
+                        (bucket, object),
+                        true,
+                        RenameDataFenceOptions::new(3, None)
+                            .with_namespace_commit_guard(Some(owner))
+                            .with_rollback_receipt(caller_receipt),
+                    )
+                    .await
+                    .expect("three real disk publications must produce a successful ACK");
+                    assert!(ack_tx.send(commit).is_ok(), "deliver successful ACK");
+                    std::future::pending::<()>().await;
+                });
+                let mut commit = tokio::time::timeout(Duration::from_secs(10), async {
+                    entered_rx.await.expect("physical tail entry");
+                    ack_rx
+                        .await
+                        .expect("ACK must arrive while the fourth disk is physically paused")
+                })
+                .await
+                .expect("successful quorum ACK must not wait for its physical tail");
+                assert_eq!(commit.online_disks.iter().flatten().count(), 3);
+                assert!(!destination.join(STORAGE_FORMAT_FILE).exists(), "tail has not published at ACK");
+                let tail_drain = commit.tail_drain.take().expect("early ACK transfers a real tail handle");
+                drop(commit);
+                caller.abort();
+                assert!(caller.await.expect_err("cancel caller after it delivered ACK").is_cancelled());
+                tokio::time::pause();
+                tokio::time::advance(Duration::from_secs(61)).await;
+                tokio::time::resume();
+                let tail = tokio::time::timeout(Duration::from_secs(5), tail_drain)
+                    .await
+                    .expect("ordinary tail timeout stays bounded after ACK")
+                    .expect("tail owner must not panic")
+                    .expect("successful ACK keeps its convergence result");
+                assert_eq!(tail.convergence, RenameConvergence::PartialCommit);
+                assert!(receipt.0.get().is_none(), "an acknowledged write must never enter rollback");
+                let pending = ctx.namespace_commits_pending();
+                let alive = owner_probe.upgrade().is_some();
+                let generation = ctx.namespace_commit_generation();
+                for disk in disks.iter().flatten().take(3) {
+                    let stored = disk
+                        .read_version(
+                            "",
+                            bucket,
+                            object,
+                            "",
+                            &ReadOptions {
+                                read_data: true,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                        .expect("all ACK voters keep the new object after caller cancellation");
+                    assert_eq!(stored.data.as_deref(), Some(b"inline-body".as_slice()));
+                }
+                drop(release_tx);
+                let lease = tokio::time::timeout(
+                    Duration::from_secs(5),
+                    os::acquire_rename_data_mutation_lease(&disk.path(), bucket, &destination),
+                )
+                .await
+                .expect("late physical tail drains");
+                drop(lease);
+                tokio::time::timeout(Duration::from_secs(5), async {
+                    while ctx.namespace_commits_pending() || owner_probe.upgrade().is_some() {
+                        tokio::task::yield_now().await;
+                    }
+                })
+                .await
+                .expect("late physical tail must release namespace ownership");
+                for dir in &dirs {
+                    let stored = reopen_local_disk(dir)
+                        .await
+                        .read_version(
+                            "",
+                            bucket,
+                            object,
+                            "",
+                            &ReadOptions {
+                                read_data: true,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                        .expect("successful ACK remains committed on every disk after late publication");
+                    assert_eq!(stored.data.as_deref(), Some(b"inline-body".as_slice()));
+                }
+                assert!(
+                    pending && alive,
+                    "physical ACK tail must retain namespace ownership after the coordinator exits"
+                );
+                assert!(!ctx.namespace_commits_pending());
+                assert!(owner_probe.upgrade().is_none());
+                assert!(ctx.namespace_commit_generation() > generation);
+                assert!(receipt.0.get().is_none(), "late publication cannot change success into rollback");
+            },
+        )
+        .await;
+    }
+
     #[tokio::test]
     #[serial_test::serial(capacity_dirty_scope)]
     async fn rename_rollback_incomplete_receipt_waits_for_undo_barrier() {