refactor(ecstore): preserve rename observations in commit module

crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs

@@ -1170,6 +1170,7 @@ pub async fn save_manual_transition_job_record_if_current(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(current_etag.to_string()),
                 ..Default::default()
@@ -1242,6 +1243,7 @@ pub(crate) async fn save_manual_transition_worker_result_if_absent(
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -1270,6 +1272,7 @@ pub(crate) async fn save_manual_transition_task_if_absent(
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -1621,6 +1624,7 @@ pub async fn save_manual_transition_scope_admission_if_absent(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -1672,6 +1676,7 @@ pub async fn save_manual_transition_scope_admission_if_current(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(current_etag.to_string()),
                 ..Default::default()

crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs

@@ -1733,6 +1733,7 @@ async fn save_config_if_none_fenced(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -1832,6 +1833,7 @@ async fn save_decommission_manifest_checkpoint_if_match(
 
         let mut opts = ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             no_lock: true,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(observed_etag),
@@ -1960,6 +1962,7 @@ async fn save_config_if_match_fenced(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(etag.to_string()),
                 ..Default::default()
@@ -3780,6 +3783,7 @@ where
                 data.clone(),
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(HTTPPreconditions {
                         if_none_match: Some("*".to_string()),
                         ..Default::default()
@@ -3869,6 +3873,7 @@ where
                 data.clone(),
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(HTTPPreconditions {
                         if_match: Some(etag),
                         ..Default::default()
@@ -3893,6 +3898,7 @@ where
                 data.clone(),
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(HTTPPreconditions {
                         if_none_match: Some("*".to_string()),
                         ..Default::default()

crates/ecstore/src/bucket/lifecycle/transition_transaction.rs

@@ -612,6 +612,7 @@ pub(crate) async fn save_transition_transaction_record(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -658,6 +659,7 @@ pub(crate) async fn save_transition_transaction_record_if_current(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(etag),
                 ..Default::default()

crates/ecstore/src/bucket/on_demand_migration/backfill.rs

@@ -684,6 +684,7 @@ async fn write_checkpoint(
     };
     let opts = ObjectOptions {
         max_parity: true,
+        write_completion: crate::object_api::WriteCompletion::TailDrained,
         http_preconditions: Some(preconditions),
         ..Default::default()
     };

crates/ecstore/src/core/pools.rs

@@ -5493,6 +5493,7 @@ where
     fence.ensure_held()?;
     let mut opts = ObjectOptions {
         max_parity: true,
+        write_completion: crate::object_api::WriteCompletion::TailDrained,
         no_lock: true,
         http_preconditions: Some(pool_meta_cas_preconditions(token, object)?),
         ..Default::default()
@@ -14412,6 +14413,7 @@ impl ECStore {
                 encoded.clone(),
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(HTTPPreconditions {
                         if_none_match: Some("*".to_string()),
                         ..Default::default()
@@ -14566,6 +14568,7 @@ impl ECStore {
                 encoded,
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(http_preconditions),
                     ..Default::default()
                 },
@@ -14957,6 +14960,7 @@ impl ECStore {
                 encoded,
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(HTTPPreconditions {
                         if_match: Some(etag),
                         ..Default::default()

crates/ecstore/src/disk/disk_store.rs

@@ -317,6 +317,22 @@ impl DiskStoreRenameDataExt for LocalDiskWrapper {
         dst_path: &str,
         external_guard: Option<Arc<dyn Send + Sync>>,
     ) -> Result<RenameDataResp> {
+        self.rename_data_observed(src_volume, src_path, fi, dst_volume, dst_path, external_guard)
+            .await
+            .result
+    }
+}
+
+impl LocalDiskWrapper {
+    pub(in crate::disk) async fn rename_data_observed(
+        &self,
+        src_volume: &str,
+        src_path: &str,
+        fi: &FileInfo,
+        dst_volume: &str,
+        dst_path: &str,
+        external_guard: Option<Arc<dyn Send + Sync>>,
+    ) -> super::RenameDataObservation {
         let operation = self.clone();
         let src_volume = src_volume.to_owned();
         let src_path = src_path.to_owned();
@@ -333,22 +349,35 @@ impl DiskStoreRenameDataExt for LocalDiskWrapper {
         } else {
             get_max_timeout_duration()
         };
-        run_owned_mutation(external_guard, move || async move {
-            operation
+        let observed = run_owned_mutation(external_guard, move || async move {
+            let mut preflight_rejection = None;
+            let result = 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
+                        // 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;
+                        preflight_rejection = observed.preflight_rejection;
+                        observed.result
                     },
                     timeout_duration,
                 )
-                .await
+                .await;
+            // Health tracking must observe the real disk error, not an Ok tuple.
+            Ok(super::RenameDataObservation {
+                result,
+                preflight_rejection,
+            })
         })
-        .await
+        .await;
+        observed.unwrap_or_else(|error| super::RenameDataObservation::unknown(Err(error)))
     }
 }
 
@@ -2588,6 +2617,46 @@ mod tests {
         assert_eq!(wrapper.metrics_snapshot().api_calls.get("unknown"), Some(&1));
     }
 
+    #[tokio::test]
+    async fn rename_preflight_evidence_preserves_health_errors_and_owned_reply() {
+        for source_exists in [false, true] {
+            for guarded in [false, true] {
+                let dir = tempfile::tempdir().expect("temp dir should be created");
+                let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be valid UTF-8"))
+                    .expect("endpoint should parse");
+                let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created"));
+                if source_exists {
+                    disk.make_volume("source").await.expect("source volume should exist");
+                }
+                let wrapper = LocalDiskWrapper::new(disk, false);
+                let drops = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+                let external_guard = guarded.then(|| Arc::new(DropProbe(Arc::clone(&drops))) as Arc<dyn Send + Sync>);
+                let mut file_info = FileInfo::new("object", 1, 0);
+                file_info.mod_time = Some(::time::OffsetDateTime::now_utc());
+                file_info.erasure.index = 1;
+                let observed = wrapper
+                    .rename_data_observed("source", "object", &file_info, "missing-destination", "object", external_guard)
+                    .await;
+                assert!(observed.rejected_before_publication(), "normal access rejection must carry proof");
+                assert!(matches!(observed.result, Err(DiskError::VolumeNotFound)));
+                let snapshot = wrapper.metrics_snapshot();
+                assert_eq!(snapshot.api_calls.get("rename_data"), Some(&1));
+                assert_eq!(snapshot.total_writes, 0, "health tracking must not observe the rejection as Ok");
+                assert_eq!(drops.load(Ordering::SeqCst), usize::from(guarded));
+
+                wrapper.health.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
+                let observed = wrapper
+                    .rename_data_observed("source", "object", &file_info, "missing-destination", "object", None)
+                    .await;
+                assert!(!observed.rejected_before_publication(), "wrapper errors carry no local preflight proof");
+                assert!(matches!(observed.result, Err(DiskError::FaultyDisk)));
+                let snapshot = wrapper.metrics_snapshot();
+                assert_eq!(snapshot.total_errors_availability, 1);
+                assert_eq!(snapshot.total_writes, 0);
+            }
+        }
+    }
+
     #[tokio::test]
     async fn local_disk_health_wrapper_counts_returned_availability_errors() {
         let dir = tempfile::tempdir().expect("temp dir should be created");

crates/ecstore/src/disk/local.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub(in crate::disk) use self::commit::LocalRenamePreflightRejection;
 #[cfg(test)]
 use self::commit::lock_rename_commit_directories;
 
@@ -8866,7 +8867,6 @@ impl DiskAPI for LocalDisk {
         Ok(())
     }
 
-    #[tracing::instrument(level = "trace", skip_all)]
     async fn rename_data(
         &self,
         src_volume: &str,
@@ -8875,8 +8875,8 @@ impl DiskAPI for LocalDisk {
         dst_volume: &str,
         dst_path: &str,
     ) -> Result<RenameDataResp> {
-        crate::hp_guard!("LocalDisk::rename_data");
-        self.rename_data_commit(src_volume, src_path, fi, dst_volume, dst_path).await
+        self.rename_data_inner(src_volume, src_path, fi, dst_volume, dst_path, &mut None)
+            .await
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
@@ -12971,6 +12971,196 @@ mod test {
         );
     }
 
+    #[tokio::test]
+    async fn observed_rename_timeout_has_no_preflight_proof_and_retains_namespace_lease() {
+        use crate::disk::disk_store::LocalDiskWrapper;
+        use futures::FutureExt;
+        use std::sync::mpsc;
+
+        temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("60"))], async {
+            let dir = tempfile::tempdir().expect("temp dir should be created");
+            let endpoint =
+                Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
+            let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created"));
+            let bucket = "observed-timeout-bucket";
+            let object = "prefix/object";
+            let tmp_object = "observed-timeout-stage";
+            let data_dir = Uuid::new_v4();
+            ensure_test_volume(&disk, bucket).await;
+            ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
+            let staged_part = disk
+                .get_object_path(RUSTFS_META_TMP_BUCKET, &format!("{tmp_object}/{data_dir}/part.1"))
+                .expect("staged part path should resolve");
+            fs::create_dir_all(staged_part.parent().expect("staged part should have a parent"))
+                .await
+                .expect("staged data directory should be created");
+            fs::write(&staged_part, b"new-payload")
+                .await
+                .expect("staged part should be written");
+            let staged_metadata = disk
+                .get_object_path_for_io(RUSTFS_META_TMP_BUCKET, &format!("{tmp_object}/{STORAGE_FORMAT_FILE}"))
+                .expect("staged metadata path should resolve");
+            let destination = disk.io_get_object_path(bucket, object).expect("destination should resolve");
+            let (entered_tx, entered_rx) = mpsc::channel();
+            let (release_tx, release_rx) = mpsc::channel();
+            set_owned_file_write_before_open(&staged_metadata, move || {
+                entered_tx.send(()).expect("signal staged writer entry");
+                // Dropping the sender also unblocks the syscall if the test fails.
+                let _ = release_rx.recv();
+            });
+            let wrapper = LocalDiskWrapper::new(Arc::clone(&disk), false);
+            let operation = wrapper.clone();
+            let fi = test_file_info(object, Uuid::new_v4(), Some(data_dir), None);
+            let rename = tokio::spawn(async move {
+                operation
+                    .rename_data_observed(RUSTFS_META_TMP_BUCKET, tmp_object, &fi, bucket, object, None)
+                    .await
+            });
+            tokio::task::spawn_blocking(move || entered_rx.recv_timeout(Duration::from_secs(10)))
+                .await
+                .expect("staged writer waiter should run")
+                .expect("rename must enter the real staged metadata write");
+
+            // Advance only after the blocking syscall owns its lease and the wrapper's timer exists.
+            tokio::time::pause();
+            tokio::time::advance(Duration::from_secs(61)).await;
+            tokio::time::resume();
+            let observed = tokio::time::timeout(Duration::from_secs(5), rename)
+                .await
+                .expect("wrapper timeout must not wait for the blocked syscall")
+                .expect("the wrapper waiter must not panic");
+            assert!(!observed.rejected_before_publication(), "a timeout must carry no local preflight proof");
+            assert!(matches!(observed.result, Err(DiskError::Timeout)));
+            let snapshot = wrapper.metrics_snapshot();
+            assert_eq!(snapshot.api_calls.get("rename_data"), Some(&1));
+            assert_eq!(snapshot.total_errors_timeout, 1);
+            assert_eq!(snapshot.total_writes, 0);
+            assert_eq!(snapshot.total_waiting, 0);
+            let volume_lock = os::disk_volume_mutation_lock(&disk.root, bucket);
+            assert!(
+                Arc::clone(&volume_lock).try_write_owned().is_err(),
+                "the blocked syscall must retain its volume guard"
+            );
+            assert!(
+                os::acquire_rename_data_mutation_lease(&disk.root, bucket, &destination)
+                    .now_or_never()
+                    .is_none(),
+                "a same-object mutation must still wait for the blocked syscall"
+            );
+            assert_eq!(fs::read(&staged_part).await.expect("staged data must remain"), b"new-payload");
+            assert!(!destination.join(STORAGE_FORMAT_FILE).exists());
+
+            release_tx.send(()).expect("release timed-out staged writer");
+            let lease = tokio::time::timeout(
+                Duration::from_secs(5),
+                os::acquire_rename_data_mutation_lease(&disk.root, bucket, &destination),
+            )
+            .await
+            .expect("the namespace lease must be released when the syscall drains");
+            drop(lease);
+            let _exclusive = tokio::time::timeout(Duration::from_secs(5), volume_lock.write_owned())
+                .await
+                .expect("the volume guard must be released when the syscall drains");
+            assert!(
+                !destination.join(STORAGE_FORMAT_FILE).exists(),
+                "timed-out waiter must not publish metadata later"
+            );
+        })
+        .await;
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn observed_rename_owned_task_panic_has_no_preflight_proof_and_releases_guard() {
+        use crate::disk::disk_store::LocalDiskWrapper;
+        use std::sync::mpsc;
+
+        let dir = tempfile::tempdir().expect("temp dir should be created");
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
+        let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created"));
+        let bucket = "observed-panic-bucket";
+        let object = "prefix/object";
+        let tmp_object = "observed-panic-stage";
+        let data_dir = Uuid::new_v4();
+        ensure_test_volume(&disk, bucket).await;
+        ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
+        let staged_part = disk
+            .get_object_path(RUSTFS_META_TMP_BUCKET, &format!("{tmp_object}/{data_dir}/part.1"))
+            .expect("staged part path should resolve");
+        fs::create_dir_all(staged_part.parent().expect("staged part should have a parent"))
+            .await
+            .expect("staged data directory should be created");
+        fs::write(&staged_part, b"new-payload")
+            .await
+            .expect("staged part should be written");
+        let destination = disk.io_get_object_path(bucket, object).expect("destination should resolve");
+        let published_part = destination.join(data_dir.to_string()).join("part.1");
+        let (entered_tx, entered_rx) = mpsc::channel();
+        let (release_tx, release_rx) = mpsc::channel();
+        set_rename_data_after_first_publication(&disk.root, bucket, object, move || {
+            entered_tx.send(()).expect("signal data publication");
+            let _ = release_rx.recv();
+            // This hook runs in the owned async mutation, outside spawn_blocking.
+            panic!("injected observed rename owner panic after publication");
+        });
+        let external_guard = Arc::new(());
+        let guard_probe = Arc::downgrade(&external_guard);
+        let wrapper = LocalDiskWrapper::new(Arc::clone(&disk), false);
+        let operation = wrapper.clone();
+        let fi = test_file_info(object, Uuid::new_v4(), Some(data_dir), None);
+        let rename = tokio::spawn(async move {
+            operation
+                .rename_data_observed(RUSTFS_META_TMP_BUCKET, tmp_object, &fi, bucket, object, Some(external_guard))
+                .await
+        });
+        tokio::task::spawn_blocking(move || entered_rx.recv_timeout(Duration::from_secs(10)))
+            .await
+            .expect("publication waiter should run")
+            .expect("rename must publish data before the injected owner panic");
+        assert!(guard_probe.upgrade().is_some(), "the owned task must retain the publication guard");
+        assert_eq!(fs::read(&published_part).await.expect("new data must be published"), b"new-payload");
+        assert!(!staged_part.exists(), "the real data rename must have consumed staging");
+        assert!(!destination.join(STORAGE_FORMAT_FILE).exists());
+        let volume_lock = os::disk_volume_mutation_lock(&disk.root, bucket);
+        assert!(
+            Arc::clone(&volume_lock).try_write_owned().is_err(),
+            "the mutation must retain its volume guard"
+        );
+
+        release_tx.send(()).expect("release mutation owner into the injected panic");
+        let observed = tokio::time::timeout(Duration::from_secs(5), rename)
+            .await
+            .expect("owned task panic must reach the wrapper")
+            .expect("the wrapper must convert the inner task panic into an error");
+        assert!(
+            !observed.rejected_before_publication(),
+            "a join failure must carry no local preflight proof"
+        );
+        assert!(matches!(observed.result, Err(DiskError::Io(error)) if error.to_string() == "owned mutation task failed"));
+        assert!(
+            guard_probe.upgrade().is_none(),
+            "the guard must be released after the mutation owner unwinds"
+        );
+        let snapshot = wrapper.metrics_snapshot();
+        assert_eq!(snapshot.api_calls.get("rename_data"), Some(&1));
+        assert_eq!(snapshot.total_writes, 0);
+        assert_eq!(snapshot.total_waiting, 0);
+        let lease = tokio::time::timeout(
+            Duration::from_secs(5),
+            os::acquire_rename_data_mutation_lease(&disk.root, bucket, &destination),
+        )
+        .await
+        .expect("panic must release the namespace lease");
+        drop(lease);
+        let _exclusive = tokio::time::timeout(Duration::from_secs(5), volume_lock.write_owned())
+            .await
+            .expect("panic must release the volume guard");
+        assert_eq!(
+            fs::read(&published_part).await.expect("published recovery data must remain"),
+            b"new-payload"
+        );
+        assert!(!destination.join(STORAGE_FORMAT_FILE).exists());
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn windows_and_unix_cancelled_staged_metadata_write_serializes_same_object_retry() {
         use std::sync::{Arc, mpsc};

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

@@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! Single-disk object rename publication and rollback. The caller retains the
-//! DiskAPI instrumentation; mutation leases and commit guards follow the syscall.
+//! Single-disk object rename publication and rollback. The shared execution core
+//! retains instrumentation, mutation leases, and commit guards through the syscall.
 
 #[cfg(all(test, windows))]
 use super::run_destination_commit_directory_preparation;
@@ -226,15 +226,23 @@ async fn restore_published_data_source(
     restore_renamed_data_source(src_volume_dir, src_data_path, dst_data_path, publication_root, mutation_lease).await
 }
 
+/// Proof produced only when the local rename returns at an existing access
+/// preflight, before metadata, backups, or object data can be published.
+#[derive(Debug)]
+pub(in crate::disk) struct LocalRenamePreflightRejection(());
+
 impl LocalDisk {
-    pub(super) async fn rename_data_commit(
+    #[tracing::instrument(name = "rename_data", target = "rustfs_ecstore::disk::local", level = "trace", skip_all)]
+    pub(super) async fn rename_data_inner(
         &self,
         src_volume: &str,
         src_path: &str,
         fi: FileInfo,
         dst_volume: &str,
         dst_path: &str,
+        preflight_rejection: &mut Option<LocalRenamePreflightRejection>,
     ) -> Result<RenameDataResp> {
+        crate::hp_guard!("LocalDisk::rename_data");
         let mut fi = fi;
         // A non-force DeleteBucket must not remove a directory while a local
         // object commit is publishing into it. The peer's empty scan remains
@@ -294,6 +302,7 @@ impl LocalDisk {
                 error = %e,
                 "Disk local access check failed"
             );
+            *preflight_rejection = Some(LocalRenamePreflightRejection(()));
             return Err(to_access_error(e, DiskError::VolumeAccessDenied).into());
         }
 
@@ -311,6 +320,7 @@ impl LocalDisk {
                 error = %e,
                 "Disk local access check failed"
             );
+            *preflight_rejection = Some(LocalRenamePreflightRejection(()));
             return Err(to_access_error(e, DiskError::VolumeAccessDenied).into());
         }
 
@@ -1202,4 +1212,22 @@ impl LocalDisk {
             })
         }
     }
+
+    pub(in crate::disk) async fn rename_data_observed(
+        &self,
+        src_volume: &str,
+        src_path: &str,
+        fi: &FileInfo,
+        dst_volume: &str,
+        dst_path: &str,
+    ) -> super::super::RenameDataObservation {
+        let mut preflight_rejection = None;
+        let result = self
+            .rename_data_inner(src_volume, src_path, fi.clone(), dst_volume, dst_path, &mut preflight_rejection)
+            .await;
+        super::super::RenameDataObservation {
+            result,
+            preflight_rejection,
+        }
+    }
 }

crates/ecstore/src/disk/mod.rs

@@ -75,6 +75,25 @@ use time::OffsetDateTime;
 use tokio::io::{AsyncRead, AsyncWrite};
 use uuid::Uuid;
 
+/// Local preflight evidence stays outside DiskAPI and the RPC response format.
+pub(crate) struct RenameDataObservation {
+    pub(crate) result: Result<RenameDataResp>,
+    preflight_rejection: Option<local::LocalRenamePreflightRejection>,
+}
+
+impl RenameDataObservation {
+    fn unknown(result: Result<RenameDataResp>) -> Self {
+        Self {
+            result,
+            preflight_rejection: None,
+        }
+    }
+
+    pub(crate) fn rejected_before_publication(&self) -> bool {
+        self.result.is_err() && self.preflight_rejection.is_some()
+    }
+}
+
 const QUOTA_MUTATION_FENCE_PREFIX: &str = "tmp/quota-mutation-fences/";
 pub(crate) const QUOTA_MUTATION_FENCE_METADATA_SUFFIX: &str = "quota-mutation-fence-token";
 
@@ -711,6 +730,36 @@ impl Disk {
             .await
     }
 
+    pub(crate) async fn rename_data_borrowed_with_fence_observed(
+        &self,
+        src_volume: &str,
+        src_path: &str,
+        fi: &FileInfo,
+        dst_volume: &str,
+        dst_path: &str,
+        scanner_publication_lease_token: Option<Uuid>,
+    ) -> RenameDataObservation {
+        match self {
+            Disk::Local(local_disk) => {
+                local_disk
+                    .rename_data_observed(src_volume, src_path, fi, dst_volume, dst_path, None)
+                    .await
+            }
+            Disk::Remote(remote_disk) => RenameDataObservation::unknown(
+                remote_disk
+                    .rename_data_borrowed_with_fence(
+                        src_volume,
+                        src_path,
+                        fi,
+                        dst_volume,
+                        dst_path,
+                        scanner_publication_lease_token,
+                    )
+                    .await,
+            ),
+        }
+    }
+
     pub(crate) async fn rename_data_borrowed_with_fence(
         &self,
         src_volume: &str,

crates/ecstore/src/object_api/types.rs

@@ -870,6 +870,18 @@ impl TierFreeVersionReceiptSink {
     }
 }
 
+/// Internal PUT completion boundary; this does not change fsync or write quorum.
+#[doc(hidden)]
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+pub enum WriteCompletion {
+    /// Return at write quorum when the commit owner can retain its guards.
+    #[default]
+    Quorum,
+    /// Drain the rename fan-out before returning. Minority failures still heal
+    /// after a successful quorum commit; this does not require every disk to succeed.
+    TailDrained,
+}
+
 #[derive(Default, Clone)]
 pub struct ObjectOptions {
     // Use the maximum parity (N/2), used when saving server configuration files
@@ -896,6 +908,10 @@ pub struct ObjectOptions {
     /// Persisted bucket incarnation observed before authorization.
     pub expected_bucket_incarnation_id: Option<Uuid>,
     pub no_lock: bool,
+    /// Control-plane writers that immediately read or CAS the same namespace
+    /// key use TailDrained without changing namespace lock ownership.
+    #[doc(hidden)]
+    pub write_completion: WriteCompletion,
     /// True when an upper layer already holds the object read lock before
     /// forwarding a no_lock read to the set layer.
     pub metadata_cache_safe: bool,

crates/ecstore/src/services/tier/tier_mutation_intent.rs

@@ -460,6 +460,7 @@ where
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -556,6 +557,7 @@ where
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(current_etag.to_string()),
                 ..Default::default()

crates/ecstore/src/services/tier/tier_probe_intent.rs

@@ -494,6 +494,7 @@ where
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -549,6 +550,7 @@ where
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(current.record_etag.clone()),
                 ..Default::default()

crates/ecstore/src/set_disk/ops/object.rs

@@ -299,11 +299,11 @@ use crate::error::is_err_invalid_upload_id;
 use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppressed};
 use crate::object_api::{
     NamespaceLockFence, ReplicationStatusWritebackCondition, ReplicationStatusWritebackMode,
-    SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY,
+    SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, WriteCompletion,
 };
 use crate::services::notification_sys::RemoteVersionStateFleetProofToken;
 use crate::services::tier::tier::{TierConfigMgr, TierDestinationId, TierOperationLease, tier_destination_id_from_metadata};
-use crate::set_disk::core::io_primitives::{RenameTailCleanup, finish_rename_tail_heal};
+use crate::set_disk::core::io_primitives::{RenameRollbackReceipt, RenameTailCleanup, finish_rename_tail_heal};
 #[cfg(test)]
 use crate::storage_api_contracts::namespace::NamespaceLocking;
 #[cfg(test)]
@@ -3548,6 +3548,7 @@ impl SetDisks {
                 (None, None, None)
             };
         let mut tmp_cleanup_owned = false;
+        let rollback_receipt = RenameRollbackReceipt::default();
         let operation = async {
             let erasure = Arc::new(erasure_from_file_info(&fi, false)?);
 
@@ -4256,6 +4257,7 @@ impl SetDisks {
             let commit_bucket = bucket.to_owned();
             let commit_object = object.to_owned();
             let commit_tmp_dir = tmp_dir.clone();
+            let commit_rollback_receipt = rollback_receipt.clone();
             let commit_object_lock_guard = object_lock_guard.take();
             let commit_decommission_object_lock_guard = decommission_object_lock_guard.take();
             let commit_publication_guard = publication_commit_guard.take();
@@ -4266,13 +4268,17 @@ impl SetDisks {
             // complete rename fan-out drains. Keep this path synchronous so
             // its terminal state is known before the coordinator releases
             // remote leases.
-            let commit_allows_early_ack = !(opts.data_movement && opts.has_decommission_capacity_reservation())
-                && (commit_object_lock_guard.is_some()
-                    || commit_decommission_object_lock_guard.is_some()
-                    || commit_publication_guard.is_some())
+            let commit_owns_namespace_guard = commit_object_lock_guard.is_some()
+                || commit_decommission_object_lock_guard.is_some()
+                || commit_publication_guard.is_some();
+            let commit_allows_early_ack = opts.write_completion == WriteCompletion::Quorum
+                && !(opts.data_movement && opts.has_decommission_capacity_reservation())
+                && commit_owns_namespace_guard
                 && commit_scanner_publication_scope.is_none();
+            // Full-tail callers also transfer owned guards to the coordinator:
+            // cancelling their ACK waiter must not cancel an in-flight rename.
             let detach_commit_owner = commit_scanner_publication_scope.is_some()
-                || commit_allows_early_ack
+                || commit_owns_namespace_guard
                 || commit_bucket_lifecycle_guard.is_some()
                 || quota_mutation_fence;
             let commit_write_path_label = write_path.metric_label();
@@ -4452,7 +4458,8 @@ impl SetDisks {
                         write_quorum,
                         commit_scanner_publication_lease_tokens.as_ref(),
                     )
-                    .with_publication_scope(commit_scanner_publication_scope.clone()),
+                    .with_publication_scope(commit_scanner_publication_scope.clone())
+                    .with_rollback_receipt(commit_rollback_receipt.clone()),
                 )
                 .await;
                 if let Some(scope) = commit_scanner_publication_scope.as_ref() {
@@ -4585,6 +4592,11 @@ impl SetDisks {
                 let rename_commit = match rename_result {
                     Ok(commit) => commit,
                     Err(err) => {
+                        if commit_rollback_receipt.is_incomplete() {
+                            // Incomplete undo retains the staging source and
+                            // rollback backup for recovery; cleanup is unsafe.
+                            return Err(err.into());
+                        }
                         if let Err(cleanup_err) = commit_set.delete_all(RUSTFS_META_TMP_BUCKET, &commit_tmp_dir).await {
                             warn!(tmp_dir = %commit_tmp_dir, error = ?cleanup_err, "failed to cleanup put_object temporary data");
                         } else if issue3031_diag_enabled() {
@@ -4617,9 +4629,8 @@ impl SetDisks {
                     request.object_version_id = committed_version_id
                         .or_else(|| commit_version_suspended.then(Uuid::nil))
                         .map(|version_id| version_id.to_string());
-                    tokio::spawn(async move {
-                        let _ = rustfs_heal_contracts::heal_channel::send_heal_request(request).await;
-                    });
+                    let heal_set = commit_set.clone();
+                    tokio::spawn(async move { heal_set.submit_rename_tail_heal(request).await });
                 }
 
                 let rename_stage_elapsed = rename_stage_start.elapsed();
@@ -4885,7 +4896,7 @@ impl SetDisks {
                     );
                 }
             });
-        } else {
+        } else if !rollback_receipt.is_incomplete() {
             // Failure path (quorum loss / rollback): keep the cleanup inline so
             // a failed PUT never returns while its tmp shards are still on disk
             // (state-residue hardening tracked by backlog#864 / backlog#898).
@@ -17494,27 +17505,69 @@ mod put_object_tmp_cleanup_tests {
     }
 
     #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
     async fn put_object_failure_cleans_tmp_workspace_inline() {
-        let (temp_dirs, _disk_stores, set_disks) = hermetic_set_disks(4).await;
+        temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
+            for write_completion in [WriteCompletion::Quorum, WriteCompletion::TailDrained] {
+                let (temp_dirs, _disk_stores, set_disks) = hermetic_set_disks(4).await;
+                let bucket = "tmp-clean-missing-bucket";
+                let object = "orphan-object";
+                let barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::BeforeNamespace);
+                let writer = Arc::clone(&set_disks);
+                let put = tokio::spawn(async move {
+                    let mut reader = PutObjReader::from_vec(vec![9u8; TEST_OBJECT_SIZE]);
+                    writer
+                        .put_object(
+                            bucket,
+                            object,
+                            &mut reader,
+                            &ObjectOptions {
+                                write_completion,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                });
+                tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+                    .await
+                    .expect("missing-bucket PUT must stage before rename");
+                let staged = non_trash_tmp_entries(&temp_dirs).await;
+                assert_eq!(staged.len(), 4, "every disk must have a staged workspace before rejection");
+                for workspace in staged {
+                    let mut entries = tokio::fs::read_dir(&workspace)
+                        .await
+                        .expect("staged workspace should be readable");
+                    let mut shards = 0;
+                    while let Some(entry) = entries.next_entry().await.expect("staged data directory should be readable") {
+                        if entry.file_type().await.expect("staged entry type").is_dir() {
+                            let part = tokio::fs::metadata(entry.path().join("part.1"))
+                                .await
+                                .expect("staging must contain an actual erasure shard");
+                            assert!(part.len() > 0, "the shard must be written before the missing-bucket failure");
+                            shards += 1;
+                        }
+                    }
+                    assert_eq!(shards, 1);
+                }
+                assert!(temp_dirs.iter().all(|dir| !dir.path().join(bucket).exists()));
+                barrier.release();
+                let err = tokio::time::timeout(Duration::from_secs(30), put)
+                    .await
+                    .expect("missing-bucket PUT must finish")
+                    .expect("PUT task should join")
+                    .expect_err("put_object into a missing bucket volume must fail");
+                assert!(matches!(err, StorageError::VolumeNotFound), "original disk error expected: {err}");
 
-        // The bucket volume is never created, so the shards are written into
-        // the tmp workspace and the commit fails at rename_data with a quorum
-        // error — exercising the failure-path cleanup.
-        let mut reader = PutObjReader::from_vec(vec![9u8; TEST_OBJECT_SIZE]);
-        let err = set_disks
-            .put_object("tmp-clean-missing-bucket", "orphan-object", &mut reader, &ObjectOptions::default())
-            .await
-            .expect_err("put_object into a missing bucket volume must fail");
-
-        // No polling: the failure path must clean the tmp workspace inline,
-        // before put_object returns (backlog#864 / backlog#898 hardening).
-        let leftovers = non_trash_tmp_entries(&temp_dirs).await;
-        assert!(
-            leftovers.is_empty(),
-            "failed PUT must not leave tmp shards behind, leftovers: {leftovers:?}, err: {err}"
-        );
-
-        drop(temp_dirs);
+                // No polling: known pre-publication rejection must clean staging
+                // inline, before PUT returns (backlog#864 / backlog#898).
+                let leftovers = non_trash_tmp_entries(&temp_dirs).await;
+                assert!(
+                    leftovers.is_empty(),
+                    "failed PUT must not leave tmp shards behind, leftovers: {leftovers:?}, err: {err}"
+                );
+            }
+        })
+        .await;
     }
 
     #[tokio::test]
@@ -18157,6 +18210,354 @@ mod put_object_tmp_cleanup_tests {
         .await;
     }
 
+    async fn make_completion_test_bucket(disks: &[DiskStore], bucket: &str) {
+        for disk in disks {
+            disk.make_volume(bucket)
+                .await
+                .expect("completion test bucket should be created");
+        }
+    }
+
+    /// Observe the actual metadata quorum while the remaining rename is parked.
+    /// A completed task count alone can race tasks that have not started yet.
+    async fn wait_for_paused_tail_metadata_quorum(disks: &[DiskStore], bucket: &str, object: &str) {
+        tokio::time::timeout(Duration::from_secs(30), async {
+            loop {
+                let mut committed = 0;
+                for disk in disks {
+                    match disk.read_version("", bucket, object, "", &ReadOptions::default()).await {
+                        Ok(_) => committed += 1,
+                        Err(DiskError::FileNotFound | DiskError::FileVersionNotFound) => {}
+                        Err(err) => panic!("unexpected metadata error while observing {bucket}/{object}: {err}"),
+                    }
+                }
+                if committed == 3 {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("three disks must publish metadata while the fourth rename remains paused");
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn tail_drained_put_waits_for_tail_and_allows_immediate_cas() {
+        temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
+            for size in [4096, 1024 * 1024] {
+                let (_dirs, disks, set) = hermetic_set_disks(4).await;
+                let bucket = "put-full-tail-cas";
+                let object = "full-tail-cas-object";
+                make_completion_test_bucket(&disks, bucket).await;
+                let tasks = rename_fanout_barrier::observe_tasks(object);
+                let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+                let writer = Arc::clone(&set);
+                let put = tokio::spawn(async move {
+                    let mut reader = PutObjReader::from_vec(vec![b'1'; size]);
+                    writer
+                        .put_object(
+                            bucket,
+                            object,
+                            &mut reader,
+                            &ObjectOptions {
+                                write_completion: WriteCompletion::TailDrained,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                });
+                tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+                    .await
+                    .expect("full-tail PUT must reach the rename barrier");
+                wait_for_paused_tail_metadata_quorum(&disks, bucket, object).await;
+                assert!(!put.is_finished(), "full-tail PUT must remain pending after metadata quorum");
+                let mut lock_probe = Box::pin(set.acquire_write_lock_diag("full_tail_probe", bucket, object));
+                assert!(
+                    futures::poll!(lock_probe.as_mut()).is_pending(),
+                    "the owned namespace guard must remain held"
+                );
+                barrier.release();
+                let written = tokio::time::timeout(Duration::from_secs(30), put)
+                    .await
+                    .expect("full-tail PUT should finish after release")
+                    .expect("full-tail PUT task should join")
+                    .expect("full-tail PUT must commit");
+                assert_eq!(tasks.running(), 0, "full-tail response must follow every rename task");
+                drop(
+                    tokio::time::timeout(Duration::from_secs(5), lock_probe)
+                        .await
+                        .expect("same-key lock should be available on return")
+                        .expect("same-key lock probe should succeed"),
+                );
+                for disk in &disks {
+                    disk.read_version("", bucket, object, "", &ReadOptions::default())
+                        .await
+                        .expect("successful full-tail PUT must publish on every healthy disk");
+                }
+                drop(barrier);
+                let mut replacement = PutObjReader::from_vec(b"cas successor".to_vec());
+                set.put_object(
+                    bucket,
+                    object,
+                    &mut replacement,
+                    &ObjectOptions {
+                        write_completion: WriteCompletion::TailDrained,
+                        http_preconditions: Some(HTTPPreconditions {
+                            if_match: written.etag,
+                            ..Default::default()
+                        }),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("immediate same-key CAS must acquire the namespace guard");
+                let mut read = set
+                    .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+                    .await
+                    .expect("CAS successor must be immediately readable");
+                let mut body = Vec::new();
+                read.stream.read_to_end(&mut body).await.expect("successor body must drain");
+                assert_eq!(body, b"cas successor");
+            }
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn tail_drained_put_preserves_quorum_success_and_heals_failed_tail() {
+        let (_dirs, disks, set) = hermetic_set_disks(4).await;
+        let bucket = "put-full-tail-heal";
+        let object = "full-tail-heal-object";
+        make_completion_test_bucket(&disks, bucket).await;
+        let mut heals = set.capture_test_rename_tail_heals();
+        let tasks = rename_fanout_barrier::observe_tasks(object);
+        let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+        let _fault = rename_fault_injection::fail_rename_on(object, &[0]);
+        let writer = Arc::clone(&set);
+        let put = tokio::spawn(async move {
+            let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
+            writer
+                .put_object(
+                    bucket,
+                    object,
+                    &mut reader,
+                    &ObjectOptions {
+                        write_completion: WriteCompletion::TailDrained,
+                        ..Default::default()
+                    },
+                )
+                .await
+        });
+        tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+            .await
+            .expect("failed tail must first reach the rename barrier");
+        wait_for_paused_tail_metadata_quorum(&disks, bucket, object).await;
+        assert!(!put.is_finished(), "committed quorum must still wait for the failing tail");
+        barrier.release();
+        tokio::time::timeout(Duration::from_secs(30), put)
+            .await
+            .expect("failed tail should drain")
+            .expect("PUT task should join")
+            .expect("a minority tail error must not negate committed quorum");
+        assert_eq!(tasks.running(), 0);
+        let heal = tokio::time::timeout(Duration::from_secs(30), heals.recv())
+            .await
+            .expect("failed tail must schedule heal")
+            .expect("heal capture must remain connected");
+        assert_eq!(heal.bucket, bucket);
+        assert_eq!(heal.object_prefix.as_deref(), Some(object));
+        let info = set
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("committed object must remain readable despite the failed tail");
+        assert_eq!(info.size, TEST_OBJECT_SIZE as i64);
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn tail_drained_put_rejects_quorum_minus_one() {
+        let (_dirs, disks, set) = hermetic_set_disks(4).await;
+        let bucket = "put-full-tail-no-quorum";
+        let object = "full-tail-no-quorum-object";
+        make_completion_test_bucket(&disks, bucket).await;
+        let _fault = rename_fault_injection::fail_rename_on(object, &[0, 1]);
+        let tasks = rename_fanout_barrier::observe_tasks(object);
+        let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
+        let err = set
+            .put_object(
+                bucket,
+                object,
+                &mut reader,
+                &ObjectOptions {
+                    write_completion: WriteCompletion::TailDrained,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("draining two successful disks cannot satisfy write quorum three");
+        assert!(
+            matches!(err, Error::ErasureWriteQuorum | Error::InsufficientWriteQuorum(_, _)),
+            "original quorum error expected: {err}"
+        );
+        assert_eq!(tasks.running(), 0, "failed fan-out and rollback must complete before return");
+        assert!(
+            set.get_object_info(bucket, object, &ObjectOptions::default()).await.is_err(),
+            "failed fresh write must not become visible"
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn put_incomplete_rollback_preserves_staging_and_old_version_backup() {
+        use crate::set_disk::core::io_primitives::rollback_fault_injection;
+
+        temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
+            for write_completion in [WriteCompletion::Quorum, WriteCompletion::TailDrained] {
+                for fault in [
+                    rollback_fault_injection::Fault::Io,
+                    rollback_fault_injection::Fault::VolumeNotFoundAfterRename,
+                ] {
+                    let (dirs, disks, set) = hermetic_set_disks(4).await;
+                    let bucket = "put-incomplete-undo";
+                    let object = "incomplete-undo-object";
+                    make_completion_test_bucket(&disks, bucket).await;
+                    let mut old_reader = PutObjReader::from_vec(vec![b'0'; TEST_OBJECT_SIZE]);
+                    set.put_object(
+                        bucket,
+                        object,
+                        &mut old_reader,
+                        &ObjectOptions {
+                            write_completion: WriteCompletion::TailDrained,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect("old generation should be completely committed");
+                    wait_for_tmp_workspace_to_drain(&dirs, "old PUT must leave no unrelated staging").await;
+                    let old = disks[0]
+                        .read_version("", bucket, object, "", &ReadOptions::default())
+                        .await
+                        .expect("old metadata must be readable");
+                    let old_data_dir = old.data_dir.expect("non-inline old version needs a data directory");
+                    let tasks = rename_fanout_barrier::observe_tasks(object);
+                    let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+                    let _rename_fault = rename_fault_injection::fail_rename_on(object, &[2, 3]);
+                    let _undo_fault = rollback_fault_injection::arm(object, 0, fault);
+                    let writer = Arc::clone(&set);
+                    let put = tokio::spawn(async move {
+                        let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
+                        writer
+                            .put_object(
+                                bucket,
+                                object,
+                                &mut reader,
+                                &ObjectOptions {
+                                    write_completion,
+                                    ..Default::default()
+                                },
+                            )
+                            .await
+                    });
+                    tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+                        .await
+                        .expect("overwrite must enter the actual rename fan-out before failure injection");
+                    barrier.release();
+                    let err = tokio::time::timeout(Duration::from_secs(30), put)
+                        .await
+                        .expect("incomplete undo must return without hanging")
+                        .expect("PUT task should join")
+                        .expect_err("two renamed disks cannot satisfy write quorum three");
+                    assert!(
+                        matches!(err, Error::ErasureWriteQuorum | Error::InsufficientWriteQuorum(_, _)),
+                        "original quorum error expected: {err}"
+                    );
+                    assert_eq!(tasks.running(), 0, "every rename and undo task must be reaped before return");
+                    let leftovers = non_trash_tmp_entries(&dirs).await;
+                    assert!(!leftovers.is_empty(), "incomplete undo must retain the new staging source for recovery");
+                    let backups = dirs
+                        .iter()
+                        .filter(|dir| {
+                            dir.path()
+                                .join(bucket)
+                                .join(object)
+                                .join(old_data_dir.to_string())
+                                .join(crate::disk::STORAGE_FORMAT_FILE_BACKUP)
+                                .exists()
+                        })
+                        .count();
+                    assert_eq!(backups, 1, "exactly the failed undo disk must retain its old-version backup");
+                    // The remaining three disks still serve the old generation;
+                    // the failed minority must never become an acknowledged write.
+                    let mut read = set
+                        .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+                        .await
+                        .expect("old generation must remain readable after incomplete rollback");
+                    let mut body = Vec::new();
+                    read.stream
+                        .read_to_end(&mut body)
+                        .await
+                        .expect("old generation should stream");
+                    assert_eq!(body, vec![b'0'; TEST_OBJECT_SIZE]);
+                }
+            }
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn tail_drained_put_owned_commit_survives_waiter_cancellation() {
+        let (dirs, disks, set) = hermetic_set_disks(4).await;
+        let bucket = RUSTFS_META_BUCKET;
+        let object = "full-tail-cancelled-receipt";
+        // Internal config writes do not own a bucket lifecycle guard. The object
+        // guard alone must keep the full-tail coordinator alive after cancellation.
+        let tasks = rename_fanout_barrier::observe_tasks(object);
+        let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+        let writer = Arc::clone(&set);
+        let put = tokio::spawn(async move {
+            let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
+            writer
+                .put_object(
+                    bucket,
+                    object,
+                    &mut reader,
+                    &ObjectOptions {
+                        write_completion: WriteCompletion::TailDrained,
+                        ..Default::default()
+                    },
+                )
+                .await
+        });
+        tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+            .await
+            .expect("cancelled receipt must first reach the rename barrier");
+        wait_for_paused_tail_metadata_quorum(&disks, bucket, object).await;
+        put.abort();
+        assert!(put.await.expect_err("ACK waiter should cancel").is_cancelled());
+        let mut lock_probe = Box::pin(set.acquire_write_lock_diag("cancelled_full_tail_probe", bucket, object));
+        assert!(
+            futures::poll!(lock_probe.as_mut()).is_pending(),
+            "owned coordinator must retain the namespace guard after waiter cancellation"
+        );
+        barrier.release();
+        drop(
+            tokio::time::timeout(Duration::from_secs(30), lock_probe)
+                .await
+                .expect("cancelled coordinator must eventually release its guard")
+                .expect("post-commit lock probe should succeed"),
+        );
+        assert_eq!(tasks.running(), 0, "cancelled coordinator must reap every rename task");
+        for disk in &disks {
+            disk.read_version("", bucket, object, "", &ReadOptions::default())
+                .await
+                .expect("caller cancellation must not interrupt committed receipt materialization");
+        }
+        wait_for_tmp_workspace_to_drain(&dirs, "cancelled full-tail commit should release staging ownership").await;
+    }
+
     #[tokio::test]
     #[serial_test::serial(capacity_dirty_scope)]
     async fn no_lock_put_waits_for_rename_tail_under_outer_guard() {
@@ -18184,6 +18585,7 @@ mod put_object_tmp_cleanup_tests {
                         &mut reader,
                         &ObjectOptions {
                             no_lock: true,
+                            write_completion: WriteCompletion::TailDrained,
                             ..Default::default()
                         },
                     )
@@ -18209,7 +18611,18 @@ mod put_object_tmp_cleanup_tests {
             put.await
                 .expect("no-lock PUT task should join")
                 .expect("no-lock PUT should commit after the rename tail releases");
+            let mut lock_probe = Box::pin(set_disks.acquire_write_lock_diag("borrowed_full_tail_probe", bucket, object));
+            assert!(
+                futures::poll!(lock_probe.as_mut()).is_pending(),
+                "full-tail PUT must not release the caller's outer guard"
+            );
             drop(outer_guard);
+            drop(
+                tokio::time::timeout(Duration::from_secs(5), lock_probe)
+                    .await
+                    .expect("outer owner releasing its guard should unblock the probe")
+                    .expect("post-outer-guard probe should succeed"),
+            );
         })
         .await;
     }

crates/ecstore/src/set_disk/transition_matrix_tests.rs

@@ -18,6 +18,7 @@ use super::{
 };
 use crate::bucket::lifecycle::lifecycle::{TRANSITION_COMPLETE, TRANSITION_PENDING, TransitionOptions, expected_expiry_time};
 use crate::ecstore_validation_blackbox::make_local_set_disks;
+use crate::object_api::WriteCompletion;
 use crate::services::tier::test_util::register_mock_tier;
 use crate::storage_api_contracts::bucket::BucketOperations;
 use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
@@ -72,7 +73,7 @@ async fn transition_and_restore_reclaim_prior_metadata_generations() {
             object,
             &mut reader,
             &ObjectOptions {
-                no_lock: true,
+                write_completion: WriteCompletion::TailDrained,
                 ..Default::default()
             },
         )
@@ -185,7 +186,7 @@ async fn prepared_snapshot_transition_duplicate_and_late_get_use_committed_remot
             object,
             &mut reader,
             &ObjectOptions {
-                no_lock: true,
+                write_completion: WriteCompletion::TailDrained,
                 ..Default::default()
             },
         )

crates/ecstore/src/store/init.rs

@@ -8045,10 +8045,15 @@ mod tests {
         );
         assert!(com::read_config(store.pools[0].clone(), &second_page_path).await.is_ok());
 
-        com::save_config(store.pools[target_pool_idx].clone(), &second_page_path, receipt_bytes.clone())
+        let full_tail = ObjectOptions {
+            max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
+            ..Default::default()
+        };
+        com::save_config_with_opts(store.pools[target_pool_idx].clone(), &second_page_path, receipt_bytes.clone(), &full_tail)
             .await
             .expect("second page receipt should restore");
-        com::save_config(store.pools[target_pool_idx].clone(), &second_page_path, b"{corrupt".to_vec())
+        com::save_config_with_opts(store.pools[target_pool_idx].clone(), &second_page_path, b"{corrupt".to_vec(), &full_tail)
             .await
             .expect("second page receipt should corrupt deterministically");
         let corrupt = store

docs/testing/ecstore-validation-suite-design.md

@@ -54,6 +54,22 @@ Fail-closed invariants every row enforces:
 
 Fault injection is explicit and deterministic: local disk mocks for unit tests, process-level disk manipulation (`crates/e2e_test/src/chaos.rs`) for e2e tests. Property tests replay a fixed seed for payload, range, and missing-shard selection.
 
+### PUT completion fixtures
+
+`ObjectOptions::default()` uses `WriteCompletion::Quorum`: a namespace-lock-owning PUT may acknowledge write quorum while its rename tail retains the lock. A fixture that immediately inspects every disk or primes a metadata generation must set `write_completion: WriteCompletion::TailDrained` and keep normal locking. TailDrained waits for the existing rename fan-out; it does not require every disk to succeed or change fsync policy. Codec-only `no_lock` fixtures do not cover namespace locking.
+
+The object tests reuse `rename_fanout_barrier::arm(object, disk_slot, phase)` and `observe_tasks(object)`. Wait for the barrier with a deadline, observe actual metadata quorum with `wait_for_paused_tail_metadata_quorum`, then release or cancel. The metadata check distinguishes a real quorum from disk tasks that have not started. Assert zero remaining rename tasks after the owned coordinator releases its lock; cancellation tests also wait for staging cleanup.
+
+| Fixture | Completion boundary |
+|---|---|
+| `early_ack_tail_drain_retains_namespace_lock_until_background_rename_finishes` | Default PUT returns before the parked tail; a second writer remains blocked. |
+| `tail_drained_put_*` | Explicit full-tail PUT retains its guard, preserves quorum success with a failed minority, rejects quorum-minus-one, and survives ACK waiter cancellation. |
+| `transition_and_restore_reclaim_prior_metadata_generations` | Both source fixtures use TailDrained before cache priming, with normal namespace locks. |
+| `object_transaction_fencing_persists_epoch_on_multipart_commit` | Multipart completion already always drains rename before inspecting all per-disk transaction UUIDs. |
+| `decommission_durable_ilm_receipt_pagination_fails_closed_on_second_page`, `dispatch_completion_cas_is_bounded_and_reaches_the_tail` | Durable receipt, journal, and manifest writers choose TailDrained; the pagination fixture also drains deliberate receipt replacement writes. |
+
+Select these checks with `cargo nextest list -p rustfs-ecstore --features test-util -E 'test(tail_drained_put) | test(early_ack_tail_drain) | test(no_lock_put_waits_for_rename_tail) | test(object_transaction_fencing_persists_epoch_on_multipart_commit) | test(transition_and_restore_reclaim) | test(decommission_durable_ilm_receipt_pagination) | test(dispatch_completion_cas)'`, then run the same expression under the default and CI profiles without retries. Remaining crash, reopen, rollback, and lock-loss schedules use the existing domain tests; this completion fixture is not a replacement for those checks.
+
 ### Coverage gate
 
 `full` and `destructive` run `cargo llvm-cov -p rustfs-ecstore --lib` and fail when line coverage of the gate scope is below `--unit-coverage-min`. The default minimum and the 100% target for EC read, write, decode, heal, metadata-quorum, and rollback paths are the `UNIT_COVERAGE_*` constants at the top of the runner. `cargo-llvm-cov` must be installed unless `--skip-coverage` is passed explicitly. The default scope `ec-critical` is: