test(storage): require named core regressions and pinned fixtures (#8330)

.config/ecstore-required-tests.json

@@ -55,6 +55,211 @@
       "invariant": "odm-source-contract-gcs",
       "suite": "rustfs",
       "name": "on_demand_migration::gcs::tests::gcs_native_backend_satisfies_the_shared_backend_contract"
+    },
+    {
+      "invariant": "rollback-inspection",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::core::io_primitives::tests::rename_rollback_incomplete_preserves_overwrite_data_dirs_and_staging"
+    },
+    {
+      "invariant": "rollback-cancellation",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::core::io_primitives::tests::rename_rollback_incomplete_cancelled_task_is_not_success"
+    },
+    {
+      "invariant": "rollback-completion-equivalence",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::core::io_primitives::tests::rename_rollback_incomplete_matches_early_ack_and_full_wait_after_reopen"
+    },
+    {
+      "invariant": "rollback-physical-owner",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::core::io_primitives::tests::rename_rollback_children_keep_namespace_ownership_after_coordinator_panic"
+    },
+    {
+      "invariant": "late-ack-retains-commit",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::core::io_primitives::tests::rename_data_early_ack_post_mutation_tail_error_never_rolls_back_commit"
+    },
+    {
+      "invariant": "tail-drained-cas",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::object::put_object_tmp_cleanup_tests::tail_drained_put_waits_for_tail_and_allows_immediate_cas"
+    },
+    {
+      "invariant": "borrowed-lock-tail",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::object::put_object_tmp_cleanup_tests::no_lock_put_waits_for_rename_tail_under_outer_guard"
+    },
+    {
+      "invariant": "cancelled-writer-owner",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::object::put_object_tmp_cleanup_tests::tail_drained_put_owned_commit_survives_waiter_cancellation"
+    },
+    {
+      "invariant": "get-body-reconstruction",
+      "suite": "rustfs-ecstore",
+      "name": "ecstore_validation_blackbox::blackbox_get_restores_body_after_one_shard_file_is_removed"
+    },
+    {
+      "invariant": "range-reconstruction",
+      "suite": "rustfs-ecstore",
+      "name": "ecstore_validation_blackbox::blackbox_range_read_restores_exact_slice_with_one_offline_disk"
+    },
+    {
+      "invariant": "multipart-abort-linearization",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::multipart::tests::abort_and_complete_linearize_for_plain_sse_and_legacy_layouts"
+    },
+    {
+      "invariant": "multipart-write-quorum",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::multipart::tests::abort_enforces_delete_write_quorum_boundary"
+    },
+    {
+      "invariant": "multipart-physical-owner",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::multipart::tests::complete_multipart_timeout_keeps_namespace_owner_until_physical_publication"
+    },
+    {
+      "invariant": "restore-abort-cleanup",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::object::transition_commit_failure_tests::multipart_restore_aborts_every_post_create_failure"
+    },
+    {
+      "invariant": "legacy-metadata",
+      "suite": "rustfs-filemeta",
+      "name": "filemeta::test::test_issue_2288_legacy_xlmeta_compatibility"
+    },
+    {
+      "invariant": "legacy-nil-version",
+      "suite": "rustfs-filemeta",
+      "name": "filemeta::test::test_into_fileinfo_reads_legacy_nil_uuid_inline_key"
+    },
+    {
+      "invariant": "commit-phase-schedule",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::core::io_primitives::tests::commit_fault_schedule_keeps_multiple_disk_phases_independent"
+    },
+    {
+      "invariant": "commit-phase-ownership",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::core::io_primitives::tests::commit_fault_schedule_rejects_duplicate_checkpoint_without_disarming_owner"
+    },
+    {
+      "invariant": "metadata-observation-all-four-slot-states-and-arrival-orders-preserve-reduction",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::core::metadata_quorum::tests::metadata_observation_all_four_slot_states_and_arrival_orders_preserve_reduction"
+    },
+    {
+      "invariant": "metadata-observation-pending-newer-version-and-same-time-different-directory-force-full-wait",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::core::metadata_quorum::tests::metadata_observation_pending_newer_version_and_same_time_different_directory_force_full_wait"
+    },
+    {
+      "invariant": "metadata-observation-duplicate-shard-cannot-supply-a-read-reserve",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::core::metadata_quorum::tests::metadata_observation_duplicate_shard_cannot_supply_a_read_reserve"
+    },
+    {
+      "invariant": "metadata-observation-null-versions-markers-and-invalid-success-keep-their-meaning",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::core::metadata_quorum::tests::metadata_observation_null_versions_markers_and_invalid_success_keep_their_meaning"
+    },
+    {
+      "invariant": "list-objects-shared-corpus-static-namespace",
+      "suite": "rustfs-ecstore",
+      "name": "store::list_objects::differential_tests::list_objects_shared_corpus_static_namespace"
+    },
+    {
+      "invariant": "list-objects-shared-corpus-cross-subdirectory-limit",
+      "suite": "rustfs-ecstore",
+      "name": "store::list_objects::differential_tests::list_objects_shared_corpus_cross_subdirectory_limit"
+    },
+    {
+      "invariant": "list-objects-shared-corpus-max-and-max-plus-one",
+      "suite": "rustfs-ecstore",
+      "name": "store::list_objects::differential_tests::list_objects_shared_corpus_max_and_max_plus_one"
+    },
+    {
+      "invariant": "list-objects-shared-corpus-fixed-seed-marker-and-prefix",
+      "suite": "rustfs-ecstore",
+      "name": "store::list_objects::differential_tests::list_objects_shared_corpus_fixed_seed_marker_and_prefix"
+    },
+    {
+      "invariant": "list-objects-shared-corpus-offline-corrupt-and-minority-stale",
+      "suite": "rustfs-ecstore",
+      "name": "store::list_objects::differential_tests::list_objects_shared_corpus_offline_corrupt_and_minority_stale"
+    },
+    {
+      "invariant": "list-objects-shared-corpus-versions-delete-markers-and-free-versions",
+      "suite": "rustfs-ecstore",
+      "name": "store::list_objects::differential_tests::list_objects_shared_corpus_versions_delete_markers_and_free_versions"
+    },
+    {
+      "invariant": "list-objects-shared-corpus-orphan-cleanup-requires-authoritative-evidence",
+      "suite": "rustfs-ecstore",
+      "name": "store::list_objects::differential_tests::list_objects_shared_corpus_orphan_cleanup_requires_authoritative_evidence"
+    },
+    {
+      "invariant": "list-objects-shared-corpus-consumer-drop-joins-real-disk-producer",
+      "suite": "rustfs-ecstore",
+      "name": "store::list_objects::differential_tests::list_objects_shared_corpus_consumer_drop_joins_real_disk_producer"
+    },
+    {
+      "invariant": "list-objects-shared-corpus-cancelled-quorum-walk-releases-senders-and-disk-lock",
+      "suite": "rustfs-ecstore",
+      "name": "store::list_objects::differential_tests::list_objects_shared_corpus_cancelled_quorum_walk_releases_senders_and_disk_lock"
+    },
+    {
+      "invariant": "list-objects-shared-corpus-enabled-index-contracts",
+      "suite": "rustfs-ecstore",
+      "name": "store::list_objects::differential_tests::list_objects_shared_corpus_enabled_index_contracts"
+    },
+    {
+      "invariant": "borrowed-context-boundary",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::object::put_object_tmp_cleanup_tests::borrowed_write_context_rejects_wrong_namespace_and_cache_flag"
+    },
+    {
+      "invariant": "borrowed-put-cancellation",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::object::put_object_tmp_cleanup_tests::borrowed_tail_drained_put_keeps_owner_after_caller_cancel"
+    },
+    {
+      "invariant": "borrowed-mpu-cancellation",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::multipart::tests::borrowed_complete_keeps_namespace_owner_after_caller_cancel"
+    },
+    {
+      "invariant": "pool-cas-owner",
+      "suite": "rustfs-ecstore",
+      "name": "core::pools::pools_tests::pool_meta_cas_context_retains_actual_owner_for_exact_target"
+    },
+    {
+      "invariant": "pool-cas-released-owner",
+      "suite": "rustfs-ecstore",
+      "name": "core::pools::pools_tests::pool_meta_cas_rejects_released_owner"
+    },
+    {
+      "invariant": "copy-tail-drained",
+      "suite": "rustfs-ecstore",
+      "name": "store::object::tests::borrowed_copy_tail_drained_waits_for_rename_tail"
+    },
+    {
+      "invariant": "catalog-borrowed-publication",
+      "suite": "rustfs",
+      "name": "admin::handlers::table_catalog::tests::commit_backend_borrowed_put_preserves_owner_authorization_and_precondition"
+    },
+    {
+      "invariant": "directory-marker-quorum",
+      "suite": "rustfs-filemeta",
+      "name": "metacache::tests::resolve_directory_marker_and_prefix_preserve_quorum_in_both_orders"
+    },
+    {
+      "invariant": "logical-prefix-limit",
+      "suite": "rustfs-ecstore",
+      "name": "store::list_objects::test::list_path_gather_results_counts_common_prefixes_before_page_limit"
     }
   ],
   "fixtures": [
@@ -82,6 +287,21 @@
       "path": "crates/ecstore/tests/fixtures/minio/bucket_metadata_full.xlmeta.hex",
       "sha256": "3b6de589519c08a1614c8bd409bb8199c17d42043861b07bce513075e6fbfc12",
       "source": "MinIO RELEASE.2025-07-23T15-54-02Z; crates/ecstore/tests/fixtures/minio/README.md"
+    },
+    {
+      "path": "crates/filemeta/tests/fixtures/issue_2288_legacy_xlmeta.hex",
+      "sha256": "b3e8d32f9950a38b5dc71f679519cc830a2b2a807e8b0b39f78840b238d8eacf",
+      "source": "Legacy xl.meta captured for rustfs/backlog#2288; filemeta::test::test_issue_2288_legacy_xlmeta_compatibility"
+    },
+    {
+      "path": "crates/filemeta/tests/fixtures/issue_2265_legacy_meta_v2_object.hex",
+      "sha256": "89dd7ec92d16e54b5d31e0c49c9edffff8a7aa58890aaeff59bc2ff84da0a62f",
+      "source": "Legacy metadata v2 object captured for rustfs/backlog#2265; crates/filemeta/src/test_data.rs"
+    },
+    {
+      "path": "crates/ecstore/tests/fixtures/list_namespace_keys.json",
+      "sha256": "5afddb60565deba0938e617470255563f0e15a15fab699645c59bb7ccc84898e",
+      "source": "Shared static LIST namespace corpus; ECStore differential tests and ODM list_through_static_namespace_boundary_matrix"
     }
   ]
 }

crates/ecstore/src/api/mod.rs

@@ -519,7 +519,7 @@ pub mod object {
         ObjectInfo, ObjectLockConfigSnapshot, ObjectMutationHook, ObjectOptions, PutObjReader, QuotaAdmission,
         RangedDecompressReader, ReadEncryptionMaterial, ReadEncryptionMode, ReadEncryptionRequest,
         SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, ScannerPublicationCommitScope, ScannerPublicationCommitStartError,
-        ScannerPublicationCommitState, ShardIntegrityWriteMode, StreamConsumer, WriteCompletion,
+        ScannerPublicationCommitState, ShardIntegrityWriteMode, StreamConsumer, WriteCommitGuard, WriteCompletion,
         get_object_body_cache_plaintext_len, lookup_get_object_body_cache_hook, register_get_object_body_cache_hook,
         register_object_mutation_hook, unregister_get_object_body_cache_hook, unregister_object_mutation_hook,
     };

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

@@ -1122,9 +1122,11 @@ pub(crate) async fn resume_recovery_disposition(
     // Lock order is control record, then legacy source. Disposition revisions
     // use ETag CAS and never acquire either namespace lock internally.
     let control_lock = api.new_ns_lock(RUSTFS_META_BUCKET, &control_object).await?;
-    let control_guard = control_lock
-        .get_read_lock(crate::set_disk::get_lock_acquire_timeout())
-        .await?;
+    let control_guard = Arc::new(
+        control_lock
+            .get_read_lock(crate::set_disk::get_lock_acquire_timeout())
+            .await?,
+    );
     let source_lock = api
         .new_ns_lock(RUSTFS_META_BUCKET, &initial.disposition.identity.canonical_source_path)
         .await?;
@@ -1504,7 +1506,7 @@ fn control_is_abandoned_successor(disposition: &IlmRecoveryDisposition, control:
 async fn abandon_recovery_control_fenced(
     api: Arc<ECStore>,
     disposition: &IlmRecoveryDisposition,
-    control_guard: &rustfs_lock::NamespaceLockGuard,
+    control_guard: &Arc<rustfs_lock::NamespaceLockGuard>,
     source_guard: &rustfs_lock::NamespaceLockGuard,
     proof: &IlmRecoveryExportFleetProofToken,
 ) -> EcstoreResult<()> {
@@ -1557,7 +1559,7 @@ async fn abandon_recovery_control_fenced(
         }),
         ..Default::default()
     };
-    options.add_namespace_lock_guard(control_guard);
+    options.add_owned_write_lock(Arc::clone(control_guard), RUSTFS_META_BUCKET, &object);
     options.add_namespace_lock_guard(source_guard);
     let write_result = config_boundary::save_config_with_opts(api.clone(), &object, encoded, &options).await;
     let observed = match load_recovery_control_no_lock(api.clone(), disposition).await {
@@ -1630,7 +1632,7 @@ async fn delete_exact_local_recovery_copy(
     api: Arc<ECStore>,
     copy: &IlmRecoverySourceCopy,
     source_schema: &str,
-    control_guard: &rustfs_lock::NamespaceLockGuard,
+    control_guard: &Arc<rustfs_lock::NamespaceLockGuard>,
     source_guard: &rustfs_lock::NamespaceLockGuard,
 ) -> EcstoreResult<()> {
     let set = recovery_copy_set(&api, &copy.authority)?;
@@ -1704,7 +1706,7 @@ async fn save_recovery_disposition_step_fenced(
     api: Arc<ECStore>,
     current: &ObservedIlmRecoveryDisposition,
     next: &IlmRecoveryDisposition,
-    control_guard: &rustfs_lock::NamespaceLockGuard,
+    control_guard: &Arc<rustfs_lock::NamespaceLockGuard>,
     source_guard: &rustfs_lock::NamespaceLockGuard,
     proof: &IlmRecoveryExportFleetProofToken,
 ) -> EcstoreResult<ObservedIlmRecoveryDisposition> {

crates/ecstore/src/bucket/migration.rs

@@ -280,11 +280,11 @@ where
 
     debug!("Found {} migrating bucket metadata, migrating...", buckets.len());
 
-    let opts = ObjectOptions {
+    let mut opts = ObjectOptions {
         max_parity: true,
-        no_lock: true,
         ..Default::default()
     };
+    opts.use_offline_recovery_write();
     let h = HeaderMap::new();
 
     for bucket in buckets {
@@ -397,11 +397,11 @@ where
         return Ok(());
     }
 
-    let opts = ObjectOptions {
+    let mut opts = ObjectOptions {
         max_parity: true,
-        no_lock: true,
         ..Default::default()
     };
+    opts.use_offline_recovery_write();
     let h = HeaderMap::new();
     let prefix = format!("{IAM_CONFIG_PREFIX}/");
     let mut continuation: Option<String> = None;

crates/ecstore/src/bucket/quota/reservation.rs

@@ -273,7 +273,7 @@ impl QuotaContext {
             reap_stale_reservations(Arc::clone(&store), &ledger_data.bucket, &ledger_data.ledger_object).await?;
 
             let ledger_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &ledger_data.ledger_object).await?;
-            let ledger_guard = ledger_lock.get_write_lock(get_lock_acquire_timeout()).await?;
+            let ledger_guard = Arc::new(ledger_lock.get_write_lock(get_lock_acquire_timeout()).await?);
             fence_namespace_mutations(&store, RUSTFS_META_BUCKET, &ledger_data.ledger_object, None).await?;
             let mut ledger = load_current_ledger_locked(
                 Arc::clone(&store),
@@ -623,7 +623,7 @@ async fn mark_commit_started(data: &LedgerReservationData) -> Result<()> {
     let data = data.clone();
     tokio::spawn(async move {
         let ledger_lock = data.store.new_ns_lock(RUSTFS_META_BUCKET, &data.ledger_object).await?;
-        let ledger_guard = ledger_lock.get_write_lock(get_lock_acquire_timeout()).await?;
+        let ledger_guard = Arc::new(ledger_lock.get_write_lock(get_lock_acquire_timeout()).await?);
         fence_namespace_mutations(&data.store, RUSTFS_META_BUCKET, &data.ledger_object, None).await?;
         let mut ledger = load_ledger_locked(Arc::clone(&data.store), &data.ledger_object).await?;
         ledger.mark_commit_started(data.operation_id, &data.reservation)?;
@@ -644,7 +644,7 @@ async fn settle(data: &LedgerReservationData, committed: bool) -> Result<()> {
         // object lock is released and would otherwise revoke a later write's
         // newly acquired fence for the same object.
         let ledger_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &ledger_object).await?;
-        let ledger_guard = ledger_lock.get_write_lock(get_lock_acquire_timeout()).await?;
+        let ledger_guard = Arc::new(ledger_lock.get_write_lock(get_lock_acquire_timeout()).await?);
         fence_namespace_mutations(&store, RUSTFS_META_BUCKET, &ledger_object, None).await?;
         let mut ledger = load_ledger_locked(Arc::clone(&store), &ledger_object).await?;
         if committed {
@@ -727,7 +727,7 @@ async fn reap_stale_reservations(store: Arc<ECStore>, bucket: &str, ledger_objec
     }
 
     let ledger_lock = store.new_ns_lock(RUSTFS_META_BUCKET, ledger_object).await?;
-    let ledger_guard = ledger_lock.get_write_lock(get_lock_acquire_timeout()).await?;
+    let ledger_guard = Arc::new(ledger_lock.get_write_lock(get_lock_acquire_timeout()).await?);
     fence_namespace_mutations(&store, RUSTFS_META_BUCKET, ledger_object, None).await?;
     let mut ledger = load_ledger_locked(Arc::clone(&store), ledger_object).await?;
     let cursor_changed = next_cursor.is_some() && ledger.reap_cursor != next_cursor;
@@ -887,7 +887,7 @@ async fn save_ledger_locked(
     store: Arc<ECStore>,
     ledger_object: &str,
     ledger: &QuotaLedger,
-    ledger_guard: &NamespaceLockGuard,
+    ledger_guard: &Arc<NamespaceLockGuard>,
 ) -> Result<()> {
     if ledger_guard.is_lock_lost() {
         return Err(StorageError::NamespaceLockQuorumUnavailable {
@@ -908,7 +908,8 @@ async fn save_ledger_locked(
         ..Default::default()
     };
     let _ = opts.set_quota_admission(0, u64::MAX);
-    opts.add_namespace_lock_guard(ledger_guard);
+    opts.add_owned_write_lock(Arc::clone(ledger_guard), RUSTFS_META_BUCKET, ledger_object);
+    opts.write_completion = crate::object_api::WriteCompletion::TailDrained;
     save_config_with_opts(store, ledger_object, serde_json::to_vec(ledger)?, &opts).await
 }
 

crates/ecstore/src/bucket/replication/replication_config_store.rs

@@ -13,7 +13,9 @@
 // limitations under the License.
 
 use super::replication_error_boundary::Result;
-use super::replication_storage_boundary::{HTTPPreconditions, ObjectInfo, ObjectOptions, ReplicationObjectIO};
+use super::replication_storage_boundary::{
+    HTTPPreconditions, ObjectInfo, ObjectOptions, ReplicationObjectIO, WriteCommitGuard, WriteCompletion,
+};
 use crate::config::{com, storageclass};
 use std::sync::Arc;
 
@@ -77,13 +79,6 @@ impl ReplicationConfigStore {
         com::save_config(api, file, data).await
     }
 
-    pub(crate) async fn save_no_lock<S>(api: Arc<S>, file: &str, data: Vec<u8>) -> Result<()>
-    where
-        S: ReplicationObjectIO,
-    {
-        com::save_config_no_lock(api, file, data).await
-    }
-
     pub(crate) async fn save_conditional<S>(
         api: Arc<S>,
         file: &str,
@@ -111,21 +106,18 @@ impl ReplicationConfigStore {
         file: &str,
         data: Vec<u8>,
         http_preconditions: HTTPPreconditions,
+        guard: &WriteCommitGuard,
     ) -> Result<()>
     where
         S: ReplicationObjectIO,
     {
-        com::save_config_with_opts_quiet(
-            api,
-            file,
-            data,
-            &ObjectOptions {
-                max_parity: true,
-                no_lock: true,
-                http_preconditions: Some(http_preconditions),
-                ..Default::default()
-            },
-        )
-        .await
+        let mut opts = ObjectOptions {
+            max_parity: true,
+            http_preconditions: Some(http_preconditions),
+            write_completion: WriteCompletion::TailDrained,
+            ..Default::default()
+        };
+        opts.add_write_commit_guard(guard);
+        com::save_config_with_opts_quiet(api, file, data, &opts).await
     }
 }

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -44,7 +44,7 @@ use super::replication_resyncer::{
 use super::replication_state::ReplicationStats;
 use super::replication_storage_boundary::{
     HTTPPreconditions, ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationDeletedObject, ReplicationObjectIO,
-    ReplicationStorage,
+    ReplicationStorage, WriteCommitGuard,
 };
 use super::replication_target_boundary::{BucketTargetError, ReplicationTargetStore, replication_object_is_ssec_encrypted};
 use super::replication_versioning_boundary::ReplicationVersioningStore;
@@ -513,7 +513,7 @@ where
             // Lock order is transaction namespace lock -> force-delete journal object lock.
             // Keep the transaction guard alive through the conditional write so legacy
             // writers cannot interleave a read-modify-write transition within this process.
-            let guard = lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await?;
+            let guard = WriteCommitGuard::acquire(&lock, ReplicationLockTiming::acquire_timeout()).await?;
             let (mut entries, preconditions, exists) = read_force_delete_intents(storage.clone(), file).await?;
             if !update(&mut entries, exists)? {
                 return Ok(());
@@ -565,7 +565,7 @@ async fn read_force_delete_intents<S: ReplicationObjectIO>(
 async fn save_force_delete_intents<S: ReplicationStorage>(
     storage: Arc<S>,
     file: &str,
-    guard: &rustfs_lock::NamespaceLockGuard,
+    guard: &WriteCommitGuard,
     entries: Vec<MrfReplicateEntry>,
     preconditions: HTTPPreconditions,
 ) -> Result<(), EcstoreError> {
@@ -628,7 +628,7 @@ async fn acknowledge_mrf_recovery<S: ReplicationStorage>(
         let lock = storage
             .new_ns_lock(ReplicationMetadataStore::rustfs_meta_bucket(), file)
             .await?;
-        let guard = lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await?;
+        let guard = WriteCommitGuard::acquire(&lock, ReplicationLockTiming::acquire_timeout()).await?;
         let current = ReplicationConfigStore::read_no_lock_with_metadata_preserve_empty(storage.clone(), file).await;
         let (current_data, current_etag, current_exists) = match current {
             Ok((data, object_info)) => (data, object_info.etag, true),
@@ -658,7 +658,7 @@ async fn acknowledge_mrf_recovery<S: ReplicationStorage>(
         } else {
             encode_mrf_file(&retained)?
         };
-        match ReplicationConfigStore::save_conditional_no_lock(storage.clone(), file, data, preconditions).await {
+        match ReplicationConfigStore::save_conditional_no_lock(storage.clone(), file, data, preconditions, &guard).await {
             Ok(()) => return Ok(retained),
             Err(EcstoreError::PreconditionFailed) => continue,
             Err(error) => return Err(error),
@@ -2522,7 +2522,7 @@ async fn quarantine_mrf_file<S: ReplicationStorage>(storage: &Arc<S>, data: &[u8
                 continue;
             }
         };
-        let guard = match lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await {
+        let guard = match WriteCommitGuard::acquire(&lock, ReplicationLockTiming::acquire_timeout()).await {
             Ok(guard) => guard,
             Err(error) => {
                 warn!(
@@ -2573,6 +2573,7 @@ async fn quarantine_mrf_file<S: ReplicationStorage>(storage: &Arc<S>, data: &[u8
                     ReplicationMetadataStore::MRF_REPLICATION_FILE,
                     Vec::new(),
                     preconditions,
+                    &guard,
                 )
                 .await
                 {
@@ -2627,13 +2628,12 @@ async fn recover_corrupt_mrf_generation<S: ReplicationStorage>(
     entries: &[MrfReplicateEntry],
     preconditions: Option<HTTPPreconditions>,
     storage: &Arc<S>,
-    guard: &rustfs_lock::NamespaceLockGuard,
+    guard: &WriteCommitGuard,
     pending_payload: &mut Option<PendingMrfAppend>,
     started: Instant,
 ) -> Option<MrfAppendResult> {
     let quarantine_file = format!("{MRF_CORRUPT_FILE_PREFIX}.{}.bin", OffsetDateTime::now_utc().unix_timestamp_nanos());
-    if let Err(error) = ReplicationConfigStore::save_no_lock(storage.clone(), &quarantine_file, corrupt_generation.to_vec()).await
-    {
+    if let Err(error) = ReplicationConfigStore::save(storage.clone(), &quarantine_file, corrupt_generation.to_vec()).await {
         warn!(
             component = LOG_COMPONENT_ECSTORE,
             subsystem = LOG_SUBSYSTEM_REPLICATION,
@@ -2688,6 +2688,7 @@ async fn recover_corrupt_mrf_generation<S: ReplicationStorage>(
         ReplicationMetadataStore::MRF_REPLICATION_FILE,
         data,
         preconditions,
+        guard,
     )
     .await
     {
@@ -2735,7 +2736,7 @@ async fn append_mrf_entries_to_disk<S: ReplicationStorage>(
             return None;
         }
     };
-    let guard = match lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await {
+    let guard = match WriteCommitGuard::acquire(&lock, ReplicationLockTiming::acquire_timeout()).await {
         Ok(guard) => guard,
         Err(error) => {
             warn!(
@@ -2862,6 +2863,7 @@ async fn append_mrf_entries_to_disk<S: ReplicationStorage>(
         ReplicationMetadataStore::MRF_REPLICATION_FILE,
         data,
         preconditions,
+        &guard,
     )
     .await
     {

crates/ecstore/src/bucket/replication/replication_storage_boundary.rs

@@ -19,7 +19,8 @@ use super::replication_error_boundary::Error;
 use super::replication_filemeta_boundary::{replication_state_from_filemeta, version_purge_status_from_filemeta};
 pub(crate) type ReplicationObjectStore = crate::store::ECStore;
 pub(crate) use crate::object_api::{
-    GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader, ReplicationStatusWritebackCondition, ReplicationStatusWritebackMode,
+    GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader, ReplicationStatusWritebackCondition,
+    ReplicationStatusWritebackMode, WriteCommitGuard, WriteCompletion,
 };
 #[cfg(test)]
 pub(crate) use crate::object_api::{NamespaceLockFence, NamespaceLockSignalTestFence};

crates/ecstore/src/config/com.rs

@@ -74,7 +74,7 @@ fn config_task_join_error(operation: &'static str, error: tokio::task::JoinError
 /// The operation must use the corresponding no-lock read/save functions.
 pub async fn with_server_config_write_lock<F, Fut, T>(store: Arc<ECStore>, operation: F) -> Result<T>
 where
-    F: FnOnce() -> Fut + Send + 'static,
+    F: FnOnce(crate::object_api::WriteCommitGuard) -> Fut + Send + 'static,
     Fut: std::future::Future<Output = T> + Send + 'static,
     T: Send + 'static,
 {
@@ -85,8 +85,8 @@ where
         let transaction_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &transaction_lock).await?;
         let _transaction_guard = transaction_lock.get_write_lock(get_lock_acquire_timeout()).await?;
         let namespace_lock = store.new_ns_lock(RUSTFS_META_BUCKET, SERVER_CONFIG_OBJECT).await?;
-        let _write_guard = namespace_lock.get_write_lock(get_lock_acquire_timeout()).await?;
-        Ok(operation().await)
+        let write_guard = crate::object_api::WriteCommitGuard::acquire(&namespace_lock, get_lock_acquire_timeout()).await?;
+        Ok(operation(write_guard.clone()).await)
     })
     .await
     .map_err(|error| config_task_join_error("server config write", error))?
@@ -119,14 +119,14 @@ where
 /// one metadata config object. The operation must use no-lock object I/O.
 pub async fn with_config_object_write_lock<F, Fut, T>(store: Arc<ECStore>, object: String, operation: F) -> Result<T>
 where
-    F: FnOnce() -> Fut + Send + 'static,
+    F: FnOnce(crate::object_api::WriteCommitGuard) -> Fut + Send + 'static,
     Fut: std::future::Future<Output = T> + Send + 'static,
     T: Send + 'static,
 {
     tokio::spawn(async move {
         let namespace_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &object).await?;
-        let _write_guard = namespace_lock.get_write_lock(get_lock_acquire_timeout()).await?;
-        Ok(operation().await)
+        let write_guard = crate::object_api::WriteCommitGuard::acquire(&namespace_lock, get_lock_acquire_timeout()).await?;
+        Ok(operation(write_guard.clone()).await)
     })
     .await
     .map_err(|error| config_task_join_error("config object write", error))?
@@ -638,7 +638,12 @@ where
     .await
 }
 
-pub async fn save_config_no_lock<S>(api: Arc<S>, file: &str, data: Vec<u8>) -> Result<()>
+pub async fn save_config_no_lock<S>(
+    api: Arc<S>,
+    file: &str,
+    data: Vec<u8>,
+    guard: &crate::object_api::WriteCommitGuard,
+) -> Result<()>
 where
     S: ObjectIO<
             Error = Error,
@@ -650,17 +655,13 @@ where
             PutObjectReader = PutObjReader,
         >,
 {
-    save_config_with_opts(
-        api,
-        file,
-        data,
-        &ObjectOptions {
-            max_parity: true,
-            no_lock: true,
-            ..Default::default()
-        },
-    )
-    .await
+    let mut opts = ObjectOptions {
+        max_parity: true,
+        write_completion: crate::object_api::WriteCompletion::TailDrained,
+        ..Default::default()
+    };
+    opts.add_write_commit_guard(guard);
+    save_config_with_opts(api, file, data, &opts).await
 }
 
 /// `delete_config` with `no_lock` set — for callers already holding the
@@ -846,12 +847,12 @@ where
     Ok(cfg)
 }
 
-async fn new_and_save_server_config_no_lock<S>(api: Arc<S>) -> Result<Config>
+async fn new_and_save_server_config_no_lock<S>(api: Arc<S>, guard: &crate::object_api::WriteCommitGuard) -> Result<Config>
 where
     S: EcstoreObjectIO + StorageAdminApi,
 {
     let cfg = new_server_config();
-    save_server_config_no_lock(api, &cfg).await?;
+    save_server_config_no_lock(api, &cfg, guard).await?;
     Ok(cfg)
 }
 
@@ -2138,14 +2139,18 @@ where
 }
 
 /// Handle the situation where the configuration file does not exist, create and save a new configuration
-async fn handle_missing_config<S>(api: Arc<S>, context: &str, namespace_lock_held: bool) -> Result<Config>
+async fn handle_missing_config<S>(
+    api: Arc<S>,
+    context: &str,
+    write_guard: Option<&crate::object_api::WriteCommitGuard>,
+) -> Result<Config>
 where
     S: EcstoreObjectIO + StorageAdminApi + NamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
 {
     warn!("Configuration not found ({}): Start initializing new configuration", context);
     let cfg = if runtime_sources::first_cluster_node_is_local().await {
-        if namespace_lock_held {
-            new_and_save_server_config_no_lock(api.clone()).await?
+        if let Some(guard) = write_guard {
+            new_and_save_server_config_no_lock(api.clone(), guard).await?
         } else {
             new_and_save_server_config(api.clone()).await?
         }
@@ -2175,13 +2180,13 @@ where
         > + StorageAdminApi,
     S: NamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
 {
-    read_config_without_migrate_inner(api, false).await
+    read_config_without_migrate_inner(api, None).await
 }
 
 /// Reads the server config while an upper layer holds the namespace write lock
 /// for [`SERVER_CONFIG_OBJECT`]. Missing-config initialization uses the matching
 /// no-lock save path and therefore cannot recursively acquire the same lock.
-pub async fn read_config_without_migrate_no_lock<S>(api: Arc<S>) -> Result<Config>
+pub async fn read_config_without_migrate_no_lock<S>(api: Arc<S>, guard: &crate::object_api::WriteCommitGuard) -> Result<Config>
 where
     S: ObjectIO<
             Error = Error,
@@ -2194,7 +2199,7 @@ where
         > + StorageAdminApi
         + NamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
 {
-    read_config_without_migrate_inner(api, true).await
+    read_config_without_migrate_inner(api, Some(guard)).await
 }
 
 /// Reads an already-initialized server config while a caller owns a namespace
@@ -2204,7 +2209,10 @@ pub async fn read_existing_server_config_no_lock(api: Arc<ECStore>) -> Result<Co
     Ok(decode_persisted_server_config(&data)?.merge())
 }
 
-async fn read_config_without_migrate_inner<S>(api: Arc<S>, namespace_lock_held: bool) -> Result<Config>
+async fn read_config_without_migrate_inner<S>(
+    api: Arc<S>,
+    write_guard: Option<&crate::object_api::WriteCommitGuard>,
+) -> Result<Config>
 where
     S: ObjectIO<
             Error = Error,
@@ -2220,19 +2228,23 @@ where
     let config_file = server_config_path();
 
     // Try to read the configuration file.
-    let data = if namespace_lock_held {
+    let data = if write_guard.is_some() {
         read_config_no_lock(api.clone(), &config_file).await
     } else {
         read_config(api.clone(), &config_file).await
     };
     match data {
-        Ok(data) => read_server_config(api, &data, namespace_lock_held).await,
-        Err(Error::ConfigNotFound) => handle_missing_config(api, "Read the main configuration", namespace_lock_held).await,
+        Ok(data) => read_server_config(api, &data, write_guard).await,
+        Err(Error::ConfigNotFound) => handle_missing_config(api, "Read the main configuration", write_guard).await,
         Err(err) => handle_config_read_error(err, &config_file),
     }
 }
 
-async fn read_server_config<S>(api: Arc<S>, data: &[u8], namespace_lock_held: bool) -> Result<Config>
+async fn read_server_config<S>(
+    api: Arc<S>,
+    data: &[u8],
+    write_guard: Option<&crate::object_api::WriteCommitGuard>,
+) -> Result<Config>
 where
     S: EcstoreObjectIO + StorageAdminApi + NamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
 {
@@ -2242,7 +2254,7 @@ where
         warn!("Received empty configuration data, try to reread from '{}'", config_file);
 
         // Try to read the configuration again
-        let data = if namespace_lock_held {
+        let data = if write_guard.is_some() {
             read_config_no_lock(api.clone(), &config_file).await
         } else {
             read_config(api.clone(), &config_file).await
@@ -2253,7 +2265,7 @@ where
                 return Ok(cfg.merge());
             }
             Err(Error::ConfigNotFound) => {
-                return handle_missing_config(api, "Read alternate configuration", namespace_lock_held).await;
+                return handle_missing_config(api, "Read alternate configuration", write_guard).await;
             }
             Err(err) => return handle_config_read_error(err, &config_file),
         }
@@ -2687,7 +2699,7 @@ where
 
 /// Saves the server config while an upper layer holds the namespace write
 /// lock for [`SERVER_CONFIG_OBJECT`].
-pub async fn save_server_config_no_lock<S>(api: Arc<S>, cfg: &Config) -> Result<()>
+pub async fn save_server_config_no_lock<S>(api: Arc<S>, cfg: &Config, guard: &crate::object_api::WriteCommitGuard) -> Result<()>
 where
     S: ObjectIO<
             Error = Error,
@@ -2699,10 +2711,10 @@ where
             PutObjectReader = PutObjReader,
         >,
 {
-    save_server_config_inner(api, cfg, true).await
+    save_server_config_inner(api, cfg, guard).await
 }
 
-async fn save_server_config_inner<S>(api: Arc<S>, cfg: &Config, no_lock: bool) -> Result<()>
+async fn save_server_config_inner<S>(api: Arc<S>, cfg: &Config, guard: &crate::object_api::WriteCommitGuard) -> Result<()>
 where
     S: ObjectIO<
             Error = Error,
@@ -2715,11 +2727,7 @@ where
         >,
 {
     let config_file = get_config_file();
-    let existing = match if no_lock {
-        read_config_no_lock(api.clone(), &config_file).await
-    } else {
-        read_config(api.clone(), &config_file).await
-    } {
+    let existing = match read_config_no_lock(api.clone(), &config_file).await {
         Ok(v) => Some(v),
         Err(Error::ConfigNotFound) => None,
         Err(err) => {
@@ -2743,21 +2751,7 @@ where
         return Ok(());
     }
 
-    if no_lock {
-        save_config_with_opts(
-            api,
-            &config_file,
-            data,
-            &ObjectOptions {
-                max_parity: true,
-                no_lock: true,
-                ..Default::default()
-            },
-        )
-        .await
-    } else {
-        save_config(api, &config_file, data).await
-    }
+    save_config_no_lock(api, &config_file, data, guard).await
 }
 
 pub async fn lookup_configs<S>(cfg: &mut Config, api: Arc<S>) -> Result<()>

crates/ecstore/src/core/pools.rs

@@ -3692,8 +3692,8 @@ fn activation_pool_meta_lock_error(err: rustfs_lock::LockError) -> Error {
 }
 
 pub(crate) struct PoolRebalanceActivationFence {
-    pool_meta_guard: rustfs_lock::NamespaceLockGuard,
-    rebalance_meta_guard: rustfs_lock::NamespaceLockGuard,
+    pool_meta_guard: Arc<rustfs_lock::NamespaceLockGuard>,
+    rebalance_meta_guard: Arc<rustfs_lock::NamespaceLockGuard>,
     fleet_proof: Option<crate::services::notification_sys::CrossPoolFenceFleetProofToken>,
     decommission_target_fence_proof: Option<crate::services::notification_sys::DecommissionTargetFenceFleetProofToken>,
     #[cfg(test)]
@@ -3720,7 +3720,12 @@ impl PoolRebalanceActivationFence {
         let forced_lost = self.forced_lost.load(Ordering::Acquire);
         #[cfg(not(test))]
         let forced_lost = false;
-        if forced_lost || self.pool_meta_guard.is_lock_lost() || self.rebalance_meta_guard.is_lock_lost() {
+        if forced_lost
+            || self.pool_meta_guard.is_lock_lost()
+            || self.pool_meta_guard.is_released()
+            || self.rebalance_meta_guard.is_lock_lost()
+            || self.rebalance_meta_guard.is_released()
+        {
             return Err(Error::other("activation lock lost before metadata commit or worker admission"));
         }
         if self
@@ -3742,8 +3747,8 @@ impl PoolRebalanceActivationFence {
     }
 
     pub(crate) fn add_namespace_lock_fence(&self, opts: &mut ObjectOptions) {
-        opts.add_namespace_lock_guard(&self.pool_meta_guard);
-        opts.add_namespace_lock_guard(&self.rebalance_meta_guard);
+        opts.add_owned_write_lock(Arc::clone(&self.pool_meta_guard), RUSTFS_META_BUCKET, POOL_META_NAME);
+        opts.add_owned_write_lock(Arc::clone(&self.rebalance_meta_guard), RUSTFS_META_BUCKET, REBAL_META_NAME);
     }
 
     #[cfg(test)]
@@ -3784,8 +3789,8 @@ where
         .map_err(activation_pool_meta_lock_error)?;
 
     Ok(PoolRebalanceActivationFence {
-        pool_meta_guard,
-        rebalance_meta_guard,
+        pool_meta_guard: Arc::new(pool_meta_guard),
+        rebalance_meta_guard: Arc::new(rebalance_meta_guard),
         fleet_proof,
         decommission_target_fence_proof: None,
         #[cfg(test)]
@@ -4067,7 +4072,7 @@ fn rollback_start_decommission_pool_meta(pool_meta: &mut PoolMeta, previous_pool
 }
 
 fn ensure_pool_meta_write_fence(guard: &rustfs_lock::NamespaceLockGuard, operation: &str) -> Result<()> {
-    if guard.is_lock_lost() {
+    if guard.is_lock_lost() || guard.is_released() {
         return Err(Error::other(format!("{operation}: pool metadata distributed fence was lost")));
     }
     Ok(())
@@ -6339,6 +6344,9 @@ fn encode_pool_meta_v3_envelope(
 }
 
 enum PoolMetaPersistenceFence<'a> {
+    Owned(Arc<rustfs_lock::NamespaceLockGuard>),
+    OfflineRecovery,
+    #[cfg(test)]
     Distributed(Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>),
     Activation(&'a PoolRebalanceActivationFence),
 }
@@ -6346,6 +6354,10 @@ enum PoolMetaPersistenceFence<'a> {
 impl PoolMetaPersistenceFence<'_> {
     fn ensure_held(&self) -> Result<()> {
         match self {
+            Self::Owned(guard) if guard.is_lock_lost() || guard.is_released() => {
+                Err(Error::other("pool metadata distributed fence was lost before a replica write"))
+            }
+            #[cfg(test)]
             Self::Distributed(Some(signal)) if signal.is_lost() => {
                 Err(Error::other("pool metadata distributed fence was lost before a replica write"))
             }
@@ -6354,14 +6366,30 @@ impl PoolMetaPersistenceFence<'_> {
         }
     }
 
-    fn add_to_options(&self, opts: &mut ObjectOptions) {
+    fn add_to_options(&self, opts: &mut ObjectOptions, object: &str) {
         match self {
+            Self::Owned(guard) => opts.add_owned_write_lock(Arc::clone(guard), RUSTFS_META_BUCKET, object),
+            Self::OfflineRecovery => opts.use_offline_recovery_write(),
+            #[cfg(test)]
             Self::Distributed(Some(signal)) => opts.add_namespace_lock_lost_signal(Arc::clone(signal)),
-            Self::Activation(fence) => fence.add_namespace_lock_fence(opts),
+            Self::Activation(fence) => {
+                fence.add_namespace_lock_fence(opts);
+                opts.add_owned_write_lock(Arc::clone(&fence.pool_meta_guard), RUSTFS_META_BUCKET, object);
+            }
+            #[cfg(test)]
             Self::Distributed(None) => {}
         }
     }
 
+    fn allows_concurrent_phases(&self) -> bool {
+        match self {
+            Self::Owned(guard) => guard.lock_lost_signal().is_some(),
+            #[cfg(test)]
+            Self::Distributed(signal) => signal.is_some(),
+            _ => false,
+        }
+    }
+
     fn is_activation(&self) -> bool {
         matches!(self, Self::Activation(_))
     }
@@ -6662,7 +6690,7 @@ fn pool_meta_cas_options(object: &str, token: &PoolMetaCasToken, fence: &PoolMet
         http_preconditions: Some(pool_meta_cas_preconditions(token, object)?),
         ..Default::default()
     };
-    fence.add_to_options(&mut opts);
+    fence.add_to_options(&mut opts, object);
     Ok(opts)
 }
 
@@ -7049,7 +7077,7 @@ where
         write_state,
         initialized,
         PoolMetaIdentityWriteScope::All,
-        &PoolMetaPersistenceFence::Distributed(None),
+        &PoolMetaPersistenceFence::OfflineRecovery,
         &mut transaction_arm,
     )
     .await?;
@@ -7076,7 +7104,7 @@ where
         write_state,
         false,
         PoolMetaIdentityWriteScope::Pools(pool_indices),
-        &PoolMetaPersistenceFence::Distributed(None),
+        &PoolMetaPersistenceFence::OfflineRecovery,
         &mut transaction_arm,
     )
     .await?;
@@ -8154,11 +8182,16 @@ impl PoolMeta {
             .cloned()
             .ok_or_else(|| Error::other("pool metadata save failed: no storage pools available"))?;
         let pool_meta_lock = pool.new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME).await?;
-        let pool_meta_guard = pool_meta_lock.get_write_lock(get_lock_acquire_timeout()).await?;
+        let pool_meta_guard = Arc::new(pool_meta_lock.get_write_lock(get_lock_acquire_timeout()).await?);
         let mut write_state = PoolMetaWriteState::default();
         let indices = (0..self.pools.len()).collect::<Vec<_>>();
         let outcome = self
-            .save_no_lock_armed_scoped(pools, &mut write_state, pool_meta_guard.lock_lost_signal(), Some(&indices))
+            .save_no_lock_armed_scoped(
+                pools,
+                &mut write_state,
+                PoolMetaPersistenceFence::Owned(Arc::clone(&pool_meta_guard)),
+                Some(&indices),
+            )
             .await?;
         outcome.disarm();
         Ok(())
@@ -8182,14 +8215,16 @@ impl PoolMeta {
     where
         S: EcstoreObjectIO,
     {
-        let outcome = self.save_no_lock_armed_scoped(pools, write_state, None, None).await?;
+        let outcome = self
+            .save_no_lock_armed_scoped(pools, write_state, PoolMetaPersistenceFence::OfflineRecovery, None)
+            .await?;
         Ok(outcome.into_committed())
     }
 
     async fn save_no_lock_with_fence<S>(
         &self,
         pools: Vec<Arc<S>>,
-        lock_lost: Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
+        fence: PoolMetaPersistenceFence<'_>,
         indices: &[usize],
     ) -> Result<()>
     where
@@ -8197,7 +8232,7 @@ impl PoolMeta {
     {
         let mut write_state = PoolMetaWriteState::default();
         let outcome = self
-            .save_no_lock_armed_scoped(pools, &mut write_state, lock_lost, Some(indices))
+            .save_no_lock_armed_scoped(pools, &mut write_state, fence, Some(indices))
             .await?;
         outcome.disarm();
         Ok(())
@@ -8233,21 +8268,20 @@ impl PoolMeta {
         &self,
         pools: Vec<Arc<S>>,
         write_state: &mut PoolMetaWriteState,
-        lock_lost: Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
+        fence: PoolMetaPersistenceFence<'_>,
         indices: &[usize],
     ) -> Result<PoolMetaSaveOutcome>
     where
         S: EcstoreObjectIO,
     {
-        self.save_no_lock_armed_scoped(pools, write_state, lock_lost, Some(indices))
-            .await
+        self.save_no_lock_armed_scoped(pools, write_state, fence, Some(indices)).await
     }
 
     async fn save_no_lock_armed_scoped<S>(
         &self,
         pools: Vec<Arc<S>>,
         write_state: &mut PoolMetaWriteState,
-        lock_lost: Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
+        fence: PoolMetaPersistenceFence<'_>,
         indices: Option<&[usize]>,
     ) -> Result<PoolMetaSaveOutcome>
     where
@@ -8258,7 +8292,6 @@ impl PoolMeta {
         // returned outcome is dropped before publication, Drop latches the
         // sticky recovery gate.
         let mut transaction_arm = write_state.arm_transaction();
-        let fence = PoolMetaPersistenceFence::Distributed(lock_lost);
         let committed = self
             .save_no_lock_transaction(pools, write_state, &fence, indices, &mut transaction_arm)
             .await
@@ -8489,7 +8522,7 @@ impl PoolMeta {
             && selection.revision.is_generation_protocol()
             && !selection.replica_state.needs_repair
             && !bootstrap_generation_required
-            && matches!(fence, PoolMetaPersistenceFence::Distributed(Some(_)));
+            && fence.allows_concurrent_phases();
         #[cfg(feature = "e2e-test-hooks")]
         let phase_barrier = if concurrent_phases && pools.len() == 4 {
             PoolMetaPhaseBarrier::bind(&previous, &committed, revision, &durable).await?
@@ -8635,7 +8668,9 @@ impl PoolMeta {
         S: EcstoreObjectIO,
     {
         let indices = (0..self.pools.len()).collect::<Vec<_>>();
-        let outcome = self.save_no_lock_armed(pools, write_state, None, &indices).await?;
+        let outcome = self
+            .save_no_lock_armed(pools, write_state, PoolMetaPersistenceFence::OfflineRecovery, &indices)
+            .await?;
         outcome.disarm();
         Ok(())
     }
@@ -10831,7 +10866,7 @@ impl ECStore {
         &self,
         write_state: &mut PoolMetaWriteState,
         operation: &str,
-    ) -> Result<(rustfs_lock::NamespaceLockGuard, PoolMeta)> {
+    ) -> Result<(Arc<rustfs_lock::NamespaceLockGuard>, PoolMeta)> {
         self.acquire_pool_meta_write_guard_with_lock_error(write_state, operation, activation_pool_meta_lock_error)
             .await
     }
@@ -10841,7 +10876,7 @@ impl ECStore {
         write_state: &mut PoolMetaWriteState,
         operation: &str,
         map_lock_error: F,
-    ) -> Result<(rustfs_lock::NamespaceLockGuard, PoolMeta)>
+    ) -> Result<(Arc<rustfs_lock::NamespaceLockGuard>, PoolMeta)>
     where
         F: FnOnce(rustfs_lock::LockError) -> Error,
     {
@@ -10862,7 +10897,7 @@ impl ECStore {
         let selection = load_pool_meta_replicas_observing(self.pools.clone(), true, write_state).await?;
         write_state.observe_replicas(selection.replica_state);
         write_state.ensure_write_safe(operation)?;
-        Ok((pool_meta_guard, selection.meta))
+        Ok((Arc::new(pool_meta_guard), selection.meta))
     }
 
     async fn acquire_pool_meta_read_guard(
@@ -11132,7 +11167,7 @@ impl ECStore {
     pub(crate) async fn acquire_pool_meta_object_heal_fence(
         &self,
         target_pool_indices: &[usize],
-    ) -> Result<(rustfs_lock::NamespaceLockGuard, Vec<Result<()>>)> {
+    ) -> Result<(Arc<rustfs_lock::NamespaceLockGuard>, Vec<Result<()>>)> {
         let mut save_guard = self.pool_meta_save_gate.lock().await;
         let (pool_meta_guard, snapshot) = self
             .acquire_pool_meta_write_guard_with_lock_error(&mut save_guard, "pool metadata heal admission failed", Error::from)
@@ -11361,7 +11396,12 @@ impl ECStore {
         ensure_exact_delete_capacity_namespace_fences(opts, bucket, object)?;
 
         let outcome = snapshot
-            .save_no_lock_armed(self.pools.clone(), &mut save_guard, write_guard.lock_lost_signal(), &source_pool_indices)
+            .save_no_lock_armed(
+                self.pools.clone(),
+                &mut save_guard,
+                PoolMetaPersistenceFence::Owned(Arc::clone(&write_guard)),
+                &source_pool_indices,
+            )
             .await?;
         ensure_pool_meta_write_fence(&write_guard, "exact delete capacity reconciliation save failed")?;
         for (target_pool_index, target_guard) in &target_guards {
@@ -11470,7 +11510,7 @@ impl ECStore {
             .save_no_lock_armed(
                 self.pools.clone(),
                 &mut save_guard,
-                pool_meta_guard.lock_lost_signal(),
+                PoolMetaPersistenceFence::Owned(Arc::clone(&pool_meta_guard)),
                 &[source_pool_index],
             )
             .await?;
@@ -11659,7 +11699,7 @@ impl ECStore {
             .save_no_lock_armed(
                 self.pools.clone(),
                 &mut save_guard,
-                pool_meta_guard.lock_lost_signal(),
+                PoolMetaPersistenceFence::Owned(Arc::clone(&pool_meta_guard)),
                 &[source_pool_index],
             )
             .await?;
@@ -11965,7 +12005,12 @@ impl ECStore {
         };
         if reservation_grew || pending_added > 0 {
             let outcome = snapshot
-                .save_no_lock_armed(self.pools.clone(), &mut save_guard, write_guard.lock_lost_signal(), &[source_pool_index])
+                .save_no_lock_armed(
+                    self.pools.clone(),
+                    &mut save_guard,
+                    PoolMetaPersistenceFence::Owned(Arc::clone(&write_guard)),
+                    &[source_pool_index],
+                )
                 .await?;
             ensure_pool_meta_write_fence(&write_guard, "decommission target capacity intent save failed")?;
             snapshot = outcome.committed.clone();
@@ -12137,7 +12182,12 @@ impl ECStore {
             return result;
         }
         let outcome = snapshot
-            .save_no_lock_armed(self.pools.clone(), save_guard, write_guard.lock_lost_signal(), &[source_pool_index])
+            .save_no_lock_armed(
+                self.pools.clone(),
+                save_guard,
+                PoolMetaPersistenceFence::Owned(Arc::clone(write_guard)),
+                &[source_pool_index],
+            )
             .await?;
         ensure_pool_meta_write_fence(write_guard, "decommission target capacity progress save failed")?;
         if let Some(target_guard) = target_guard.as_ref() {
@@ -12272,8 +12322,8 @@ impl ECStore {
             .first()
             .ok_or_else(|| Error::other("pool metadata recovery has no storage pools"))?;
         let lock = pool.new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME).await?;
-        let guard = lock.get_write_lock(get_lock_acquire_timeout()).await?;
-        let fence = PoolMetaPersistenceFence::Distributed(guard.lock_lost_signal());
+        let guard = Arc::new(lock.get_write_lock(get_lock_acquire_timeout()).await?);
+        let fence = PoolMetaPersistenceFence::Owned(Arc::clone(&guard));
         let mut candidate = PoolMetaWriteState {
             aborted_transaction: Arc::new(AtomicBool::new(false)),
             transaction_failure: Arc::default(),
@@ -12449,7 +12499,12 @@ impl ECStore {
             merge_pool_meta_updates_for_save(&mut snapshot, &pool_meta, indices, "pool metadata save failed")?;
         }
         let outcome = snapshot
-            .save_no_lock_armed(self.pools.clone(), &mut save_guard, pool_meta_guard.lock_lost_signal(), indices)
+            .save_no_lock_armed(
+                self.pools.clone(),
+                &mut save_guard,
+                PoolMetaPersistenceFence::Owned(Arc::clone(&pool_meta_guard)),
+                indices,
+            )
             .await?;
         let mut pool_meta = self.pool_meta.write().await;
         ensure_pool_meta_write_fence(&pool_meta_guard, "pool metadata save failed")?;
@@ -12544,7 +12599,12 @@ impl ECStore {
         };
 
         let outcome = match snapshot
-            .save_no_lock_armed(self.pools.clone(), &mut save_guard, pool_meta_guard.lock_lost_signal(), &[idx])
+            .save_no_lock_armed(
+                self.pools.clone(),
+                &mut save_guard,
+                PoolMetaPersistenceFence::Owned(Arc::clone(&pool_meta_guard)),
+                &[idx],
+            )
             .await
         {
             Ok(outcome) => outcome,
@@ -12610,7 +12670,12 @@ impl ECStore {
         #[cfg(test)]
         decommission_test_wrap_result("bucket_completion_before_save", &bucket.name, &bucket.prefix, 0, Ok(()))?;
         let outcome = snapshot
-            .save_no_lock_armed(self.pools.clone(), &mut save_guard, pool_meta_guard.lock_lost_signal(), &[idx])
+            .save_no_lock_armed(
+                self.pools.clone(),
+                &mut save_guard,
+                PoolMetaPersistenceFence::Owned(Arc::clone(&pool_meta_guard)),
+                &[idx],
+            )
             .await?;
         #[cfg(test)]
         decommission_test_wrap_result("bucket_completion_after_save", &bucket.name, &bucket.prefix, 0, Ok(()))?;
@@ -12744,7 +12809,12 @@ impl ECStore {
             .await?;
         rollback_start_decommission_pool_meta(&mut snapshot, previous_pool_meta, indices);
         let outcome = snapshot
-            .save_no_lock_armed(self.pools.clone(), &mut save_guard, pool_meta_guard.lock_lost_signal(), indices)
+            .save_no_lock_armed(
+                self.pools.clone(),
+                &mut save_guard,
+                PoolMetaPersistenceFence::Owned(Arc::clone(&pool_meta_guard)),
+                indices,
+            )
             .await?;
         let mut pool_meta = self.pool_meta.write().await;
         ensure_pool_meta_write_fence(&pool_meta_guard, "decommission start rollback failed")?;
@@ -13044,7 +13114,7 @@ impl ECStore {
         save_pool_meta: Save,
     ) -> Result<()>
     where
-        Save: FnOnce(PoolMeta, Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>) -> SaveFuture + Send + 'static,
+        Save: FnOnce(PoolMeta, PoolMetaPersistenceFence<'static>) -> SaveFuture + Send + 'static,
         SaveFuture: Future<Output = Result<()>> + Send + 'static,
     {
         let store = self.clone();
@@ -13064,7 +13134,7 @@ impl ECStore {
         save_pool_meta: Save,
     ) -> Result<()>
     where
-        Save: FnOnce(PoolMeta, Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>) -> SaveFuture,
+        Save: FnOnce(PoolMeta, PoolMetaPersistenceFence<'static>) -> SaveFuture,
         SaveFuture: Future<Output = Result<()>>,
     {
         let owner = owner.as_ref();
@@ -13113,9 +13183,17 @@ impl ECStore {
                 ));
             }
         }
-        let pool_meta_fence = _pool_meta_guard
-            .as_ref()
-            .and_then(rustfs_lock::NamespaceLockGuard::lock_lost_signal);
+        let pool_meta_fence = match _pool_meta_guard.as_ref() {
+            Some(guard) => PoolMetaPersistenceFence::Owned(Arc::clone(guard)),
+            None => {
+                #[cfg(test)]
+                {
+                    PoolMetaPersistenceFence::OfflineRecovery
+                }
+                #[cfg(not(test))]
+                return Err(Error::other("runtime pool metadata cancellation requires its write-lock owner"));
+            }
+        };
 
         // Lock order: start gate, target gates, save gate, distributed pool
         // metadata fence, rebalance_meta, decommission_cancelers, then
@@ -13554,7 +13632,12 @@ impl ECStore {
             let (save_outcome, save_error) = if changed {
                 ensure_decommission_target_fence_fleet_proof(target_fence_proof.as_ref(), target_fence_proof_required)?;
                 match snapshot
-                    .save_no_lock_armed(self.pools.clone(), &mut save_guard, pool_meta_guard.lock_lost_signal(), &changed_indices)
+                    .save_no_lock_armed(
+                        self.pools.clone(),
+                        &mut save_guard,
+                        PoolMetaPersistenceFence::Owned(Arc::clone(&pool_meta_guard)),
+                        &changed_indices,
+                    )
                     .await
                 {
                     Ok(outcome) => (Some(outcome), None),
@@ -13646,7 +13729,12 @@ impl ECStore {
             .await?;
         snapshot.mark_decommission_capacity_blocked(idx, err.to_string(), OffsetDateTime::now_utc())?;
         let outcome = snapshot
-            .save_no_lock_armed(self.pools.clone(), &mut save_guard, pool_meta_guard.lock_lost_signal(), &[idx])
+            .save_no_lock_armed(
+                self.pools.clone(),
+                &mut save_guard,
+                PoolMetaPersistenceFence::Owned(Arc::clone(&pool_meta_guard)),
+                &[idx],
+            )
             .await?;
         ensure_pool_meta_write_fence(&pool_meta_guard, "decommission capacity pause failed")?;
         {
@@ -22023,7 +22111,7 @@ mod tests {
             let mut state = store.pool_meta_save_gate.lock().await;
             let (fence, _) = store.acquire_pool_meta_write_guard(&mut state, "test save").await.unwrap();
             let outcome = requested
-                .save_no_lock_armed(store.pools.clone(), &mut state, fence.lock_lost_signal(), &[0])
+                .save_no_lock_armed(store.pools.clone(), &mut state, PoolMetaPersistenceFence::Owned(Arc::clone(&fence)), &[0])
                 .await
                 .unwrap();
             assert_eq!(outcome.committed.pools[0].last_update, requested.pools[0].last_update);
@@ -24825,11 +24913,12 @@ mod pools_tests {
             config[3][0].gate = Some(gate.clone());
             let fixture = Fixture::new(config);
             let mut state = fixture.state();
-            let mut save = Box::pin(
-                fixture
-                    .desired
-                    .save_no_lock_armed(fixture.pools.clone(), &mut state, signal(), &[0]),
-            );
+            let mut save = Box::pin(fixture.desired.save_no_lock_armed(
+                fixture.pools.clone(),
+                &mut state,
+                PoolMetaPersistenceFence::Distributed(signal()),
+                &[0],
+            ));
             tokio::select! { _ = fixture.trace.wait_for(4, 3) => {}, _ = &mut save => panic!("commit crossed the last prepare") }
             assert_eq!(fixture.trace.commit_count(), 0);
             gate.add_permits(1);
@@ -24849,7 +24938,7 @@ mod pools_tests {
             let mut state = fixture.state();
             let err = fixture
                 .desired
-                .save_no_lock_armed(fixture.pools.clone(), &mut state, signal(), &[0])
+                .save_no_lock_armed(fixture.pools.clone(), &mut state, PoolMetaPersistenceFence::Distributed(signal()), &[0])
                 .await
                 .expect_err("a missing conditional-write revision must fail");
             assert!(err.to_string().contains("without a conditional-write revision"));
@@ -24866,11 +24955,12 @@ mod pools_tests {
             config[3][0].gate = Some(gate.clone());
             let fixture = Fixture::new(config);
             let mut state = fixture.state();
-            let mut save = Box::pin(
-                fixture
-                    .desired
-                    .save_no_lock_armed(fixture.pools.clone(), &mut state, signal(), &[0]),
-            );
+            let mut save = Box::pin(fixture.desired.save_no_lock_armed(
+                fixture.pools.clone(),
+                &mut state,
+                PoolMetaPersistenceFence::Distributed(signal()),
+                &[0],
+            ));
             tokio::select! { _ = fixture.trace.wait_for(4, 3) => {}, _ = &mut save => panic!("phase returned with a live replica") }
             assert_eq!(fixture.trace.commit_count(), 0);
             gate.add_permits(1);
@@ -25045,11 +25135,12 @@ mod pools_tests {
                 }
                 let fixture = Fixture::new(config);
                 let mut state = fixture.state();
-                let mut save = Box::pin(
-                    fixture
-                        .desired
-                        .save_no_lock_armed(fixture.pools.clone(), &mut state, signal(), &[0]),
-                );
+                let mut save = Box::pin(fixture.desired.save_no_lock_armed(
+                    fixture.pools.clone(),
+                    &mut state,
+                    PoolMetaPersistenceFence::Distributed(signal()),
+                    &[0],
+                ));
                 tokio::select! { _ = fixture.trace.wait_for(4 * (phase + 1), 4 * phase) => {}, _ = &mut save => panic!("blocked phase finished") }
                 drop(save);
                 assert!(state.ensure_write_safe("cancelled phase").is_err());
@@ -25060,7 +25151,7 @@ mod pools_tests {
             let mut state = fixture.state();
             let outcome = fixture
                 .desired
-                .save_no_lock_armed(fixture.pools.clone(), &mut state, signal(), &[0])
+                .save_no_lock_armed(fixture.pools.clone(), &mut state, PoolMetaPersistenceFence::Distributed(signal()), &[0])
                 .await
                 .expect("durable save before publication");
             drop(outcome);
@@ -25088,6 +25179,7 @@ mod pools_tests {
                 .await
                 .expect("lease request")
                 .expect("lease acquired");
+            let guard = Arc::new(guard);
             let gate = Arc::new(tokio::sync::Semaphore::new(0));
             let mut config = steps(6);
             for step in &mut config {
@@ -25096,7 +25188,7 @@ mod pools_tests {
             let fixture = Fixture::new(config);
             let state = fixture.state();
             let mut arm = state.arm_transaction();
-            let fence = PoolMetaPersistenceFence::Distributed(guard.lock_lost_signal());
+            let fence = PoolMetaPersistenceFence::Owned(Arc::clone(&guard));
             let mut pending = Box::pin(prepare(&fixture, &fence, &mut arm));
             tokio::select! { _ = fixture.trace.wait_for(4, 0) => {}, _ = &mut pending => panic!("gate ignored") }
             tokio::time::timeout(StdDuration::from_secs(3), guard.lock_lost_notified())
@@ -25122,7 +25214,7 @@ mod pools_tests {
                 let mut state = fixture.state();
                 let result = fixture
                     .desired
-                    .save_no_lock_armed(fixture.pools.clone(), &mut state, signal(), &[0])
+                    .save_no_lock_armed(fixture.pools.clone(), &mut state, PoolMetaPersistenceFence::Distributed(signal()), &[0])
                     .await;
                 assert_eq!(fixture.trace.commit_count(), 4);
                 if matches!(fault, WriteFault::TimeoutAfterWrite) {
@@ -25172,7 +25264,7 @@ mod pools_tests {
                 let mut save = Box::pin(fixture.desired.save_no_lock_armed(
                     fixture.pools.clone(),
                     &mut state,
-                    if legacy || needs_repair { signal() } else { None },
+                    PoolMetaPersistenceFence::Distributed(if legacy || needs_repair { signal() } else { None }),
                     &[0],
                 ));
                 tokio::select! { _ = fixture.trace.wait_for(1, 0) => {}, _ = &mut save => panic!("serial write gate ignored") }
@@ -25732,7 +25824,7 @@ mod pools_tests {
         let save_task = tokio::spawn(async move {
             let mut save_guard = save_store.pool_meta_save_gate.lock().await;
             let outcome = snapshot
-                .save_no_lock_armed(vec![save_committed], &mut save_guard, None, &[0])
+                .save_no_lock_armed(vec![save_committed], &mut save_guard, PoolMetaPersistenceFence::OfflineRecovery, &[0])
                 .await?;
             task_publish_started.notify_one();
             task_publish_release.notified().await;
@@ -25759,6 +25851,73 @@ mod pools_tests {
             .expect_err("cancellation after replica save but before publication must keep writes blocked");
     }
 
+    #[tokio::test]
+    async fn pool_meta_cas_context_retains_actual_owner_for_exact_target() {
+        use crate::object_api::write_commit_context::WriteCommitContext;
+
+        let lock = NamespaceLock::with_local_manager("pool-meta-cas-owner".to_string(), Arc::new(GlobalLockManager::new()));
+        let request = LockRequest::new(
+            ObjectKey::new(super::RUSTFS_META_BUCKET, POOL_META_NAME),
+            LockType::Exclusive,
+            "cas-writer",
+        )
+        .with_acquire_timeout(StdDuration::from_secs(1));
+        let owner = Arc::new(
+            lock.acquire_guard(&request)
+                .await
+                .expect("pool metadata lock acquisition should succeed")
+                .expect("pool metadata lock should be available"),
+        );
+        let weak_owner = Arc::downgrade(&owner);
+        let fence = PoolMetaPersistenceFence::Owned(Arc::clone(&owner));
+        let mut contexts = Vec::new();
+        for object in [POOL_META_NAME, POOL_META_IDENTITY_NAME] {
+            let options = super::pool_meta_cas_options(object, &PoolMetaCasToken::Missing, &fence)
+                .expect("a held pool metadata lock should authorize its CAS target");
+            let cloned_options = options.clone();
+            let context = WriteCommitContext::from_options(&cloned_options, super::RUSTFS_META_BUCKET, object, false)
+                .expect("the actual CAS target should retain its borrowed namespace owner");
+            assert!(context.has_borrowed_owner(), "CAS must borrow the real protocol lock owner");
+            assert!(
+                WriteCommitContext::from_options(&cloned_options, super::RUSTFS_META_BUCKET, "unrelated-object", false).is_err(),
+                "a pool metadata CAS capability must not authorize an unrelated object"
+            );
+            drop(options);
+            drop(cloned_options);
+            contexts.push(context);
+        }
+        drop(fence);
+        drop(owner);
+        assert!(
+            weak_owner.upgrade().is_some(),
+            "queued commit contexts must keep the actual lock guard alive"
+        );
+        drop(contexts);
+        assert!(weak_owner.upgrade().is_none(), "the last commit context must release its lock owner");
+    }
+
+    #[tokio::test]
+    async fn pool_meta_cas_rejects_released_owner() {
+        let lock = NamespaceLock::with_local_manager("pool-meta-released-owner".to_string(), Arc::new(GlobalLockManager::new()));
+        let request = LockRequest::new(
+            ObjectKey::new(super::RUSTFS_META_BUCKET, POOL_META_NAME),
+            LockType::Exclusive,
+            "released-cas-writer",
+        )
+        .with_acquire_timeout(StdDuration::from_secs(1));
+        let mut owner = lock
+            .acquire_guard(&request)
+            .await
+            .expect("pool metadata lock acquisition should succeed")
+            .expect("pool metadata lock should be available");
+        assert!(owner.release(), "the test must release a real namespace owner");
+        let fence = PoolMetaPersistenceFence::Owned(Arc::new(owner));
+        assert!(
+            super::pool_meta_cas_options(POOL_META_NAME, &PoolMetaCasToken::Missing, &fence).is_err(),
+            "a released namespace owner must not dispatch another CAS"
+        );
+    }
+
     #[tokio::test]
     async fn test_lost_pool_meta_fence_rejects_replica_write() {
         let client = Arc::new(LocalClient::with_manager(Arc::new(GlobalLockManager::new())));
@@ -25776,6 +25935,7 @@ mod pools_tests {
             .await
             .expect("pool metadata fence acquisition should not error")
             .expect("the stale writer should acquire the pool metadata fence");
+        let guard = Arc::new(guard);
         tokio::time::timeout(StdDuration::from_secs(2), guard.lock_lost_notified())
             .await
             .expect("the stale writer lease should expire");
@@ -25797,7 +25957,12 @@ mod pools_tests {
         };
         let mut write_state = super::PoolMetaWriteState::default();
         let err = snapshot
-            .save_no_lock_armed(vec![storage.clone()], &mut write_state, guard.lock_lost_signal(), &[0])
+            .save_no_lock_armed(
+                vec![storage.clone()],
+                &mut write_state,
+                PoolMetaPersistenceFence::Owned(Arc::clone(&guard)),
+                &[0],
+            )
             .await
             .expect_err("a writer must not persist after losing the distributed pool metadata fence");
 
@@ -25825,6 +25990,7 @@ mod pools_tests {
             .await
             .expect("pool metadata fence acquisition should not error")
             .expect("the publishing writer should acquire the pool metadata fence");
+        let guard = Arc::new(guard);
         let storage = Arc::new(PartialPoolMetaWriteStorage {
             fail_write: false,
             fail_after_first_write: false,
@@ -25845,7 +26011,7 @@ mod pools_tests {
         current.pools[0].last_update = OffsetDateTime::UNIX_EPOCH;
         let mut write_state = super::PoolMetaWriteState::for_test_bootstrap();
         let outcome = saved
-            .save_no_lock_armed(vec![storage], &mut write_state, guard.lock_lost_signal(), &[0])
+            .save_no_lock_armed(vec![storage], &mut write_state, PoolMetaPersistenceFence::Owned(Arc::clone(&guard)), &[0])
             .await
             .expect("the replica save should finish while the fence is valid");
 

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

@@ -84,46 +84,107 @@ async fn wait_for_heal_commit_test_barrier(root: &Path, bucket: &str, object: &s
     }
 }
 
-fn rollback_committed_rename_std(
-    dst_file_path: &Path,
-    new_data_path: Option<&Path>,
-    rollback_data_dir: Option<Uuid>,
-) -> std::io::Result<()> {
-    if let Some(old_data_dir) = rollback_data_dir {
-        let Some(dst_parent) = dst_file_path.parent() else {
-            return Err(std::io::Error::new(ErrorKind::InvalidInput, "missing object metadata parent"));
-        };
-        let backup_path = dst_parent.join(old_data_dir.to_string()).join(STORAGE_FORMAT_FILE_BACKUP);
-        std::fs::rename(backup_path, dst_file_path)?;
-    } else {
-        remove_file_if_exists(dst_file_path)?;
-    }
-
-    if let Some(new_data_path) = new_data_path {
-        remove_dir_all_if_exists(new_data_path)?;
-    }
-
-    Ok(())
+struct RenameCommitContext<'a> {
+    // Drop the directory identity guard before releasing mutation ownership.
+    directory_guard: os::RenameCommitGuard,
+    publication_root: &'a os::PublicationRoot,
+    mutation_lease: Arc<os::NamespaceMutationLease>,
+    source_metadata: &'a Path,
+    destination_metadata: &'a Path,
+    source_volume_dir: &'a Path,
+    data_paths: Option<&'a (PathBuf, PathBuf)>,
 }
 
-fn rollback_inline_metadata_commit_std(
-    dst_file_path: &Path,
-    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
-        // `dst_file_path`, proving both paths are on the same filesystem.
-        // Unix rename atomically replaces the committed destination; never
-        // unlink it first or an interrupted rollback could lose xl.meta.
-        std::fs::rename(backup_path, dst_file_path)?;
-    } else {
-        rollback_committed_rename_std(dst_file_path, None, rollback_data_dir)?;
+struct StagedRollbackBackup(PathBuf);
+
+impl StagedRollbackBackup {
+    fn path(&self) -> &Path {
+        &self.0
+    }
+}
+
+impl RenameCommitContext<'_> {
+    fn rollback_metadata_std(&self, rollback_data_dir: Option<Uuid>) -> std::io::Result<()> {
+        if let Some(old_data_dir) = rollback_data_dir {
+            let Some(dst_parent) = self.destination_metadata.parent() else {
+                return Err(std::io::Error::new(ErrorKind::InvalidInput, "missing object metadata parent"));
+            };
+            let backup_path = dst_parent.join(old_data_dir.to_string()).join(STORAGE_FORMAT_FILE_BACKUP);
+            std::fs::rename(backup_path, self.destination_metadata)?;
+        } else {
+            remove_file_if_exists(self.destination_metadata)?;
+        }
+        Ok(())
+    }
+
+    fn rollback_non_inline_std(&self, rollback_data_dir: Option<Uuid>) -> std::io::Result<()> {
+        self.rollback_metadata_std(rollback_data_dir)?;
+        if let Some((_, new_data_path)) = self.data_paths {
+            remove_dir_all_if_exists(new_data_path)?;
+        }
+        Ok(())
+    }
+
+    fn rollback_inline_std(
+        &self,
+        rollback_data_dir: Option<Uuid>,
+        local_backup: Option<&StagedRollbackBackup>,
+    ) -> std::io::Result<()> {
+        #[cfg(all(test, not(windows)))]
+        os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Rollback, self.destination_metadata);
+        if let Some(backup) = local_backup {
+            // The preceding commit used this staged directory on the same
+            // filesystem. Never unlink xl.meta before its atomic replacement.
+            std::fs::rename(backup.path(), self.destination_metadata)?;
+        } else {
+            self.rollback_metadata_std(rollback_data_dir)?;
+        }
+        Ok(())
+    }
+
+    async fn restore_published_data_source(&self) -> Result<()> {
+        let Some((src_data_path, dst_data_path)) = self.data_paths else {
+            return Ok(());
+        };
+        let src_volume_dir = self.source_volume_dir;
+        let publication_root = self.publication_root;
+        let mutation_lease = self.mutation_lease.clone();
+        if fs::symlink_metadata(src_data_path).await.is_ok() {
+            return Ok(());
+        }
+        let result =
+            match os::rename_all_with_lease(dst_data_path, src_data_path, src_volume_dir, publication_root, mutation_lease).await
+            {
+                Ok(()) => Ok(()),
+                Err(DiskError::FileNotFound) => {
+                    let source_exists = fs::symlink_metadata(src_data_path).await.is_ok();
+                    let destination_missing = matches!(
+                        fs::symlink_metadata(dst_data_path).await,
+                        Err(err) if err.kind() == ErrorKind::NotFound
+                    );
+                    if source_exists && destination_missing {
+                        Ok(())
+                    } else {
+                        Err(DiskError::FileNotFound)
+                    }
+                }
+                Err(err) => Err(err),
+            };
+        if let Err(err) = &result {
+            warn!(
+                target: "rustfs_ecstore::disk::local",
+                event = EVENT_DISK_LOCAL_RENAME_REJECTED,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
+                reason = "restore_staged_data_source_failed",
+                src_path = ?src_data_path,
+                dst_path = ?dst_data_path,
+                error = ?err,
+                "Failed to restore staged data after a metadata commit was rejected"
+            );
+        }
+        result
     }
-    Ok(())
 }
 
 #[cfg(any(not(windows), test))]
@@ -212,61 +273,6 @@ async fn read_rename_destination_metadata(
         .map_err(DiskError::from)
 }
 
-async fn restore_renamed_data_source(
-    src_volume_dir: &Path,
-    src_data_path: &Path,
-    dst_data_path: &Path,
-    publication_root: &os::PublicationRoot,
-    mutation_lease: Arc<os::NamespaceMutationLease>,
-) -> Result<()> {
-    if fs::symlink_metadata(src_data_path).await.is_ok() {
-        return Ok(());
-    }
-    let result =
-        match os::rename_all_with_lease(dst_data_path, src_data_path, src_volume_dir, publication_root, mutation_lease).await {
-            Ok(()) => Ok(()),
-            Err(DiskError::FileNotFound) => {
-                let source_exists = fs::symlink_metadata(src_data_path).await.is_ok();
-                let destination_missing = matches!(
-                    fs::symlink_metadata(dst_data_path).await,
-                    Err(err) if err.kind() == ErrorKind::NotFound
-                );
-                if source_exists && destination_missing {
-                    Ok(())
-                } else {
-                    Err(DiskError::FileNotFound)
-                }
-            }
-            Err(err) => Err(err),
-        };
-    if let Err(err) = &result {
-        warn!(
-            target: "rustfs_ecstore::disk::local",
-            event = EVENT_DISK_LOCAL_RENAME_REJECTED,
-            component = LOG_COMPONENT_ECSTORE,
-            subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
-            reason = "restore_staged_data_source_failed",
-            src_path = ?src_data_path,
-            dst_path = ?dst_data_path,
-            error = ?err,
-            "Failed to restore staged data after a metadata commit was rejected"
-        );
-    }
-    result
-}
-
-async fn restore_published_data_source(
-    data_paths: Option<&(PathBuf, PathBuf)>,
-    src_volume_dir: &Path,
-    publication_root: &os::PublicationRoot,
-    mutation_lease: Arc<os::NamespaceMutationLease>,
-) -> Result<()> {
-    let Some((src_data_path, dst_data_path)) = data_paths else {
-        return Ok(());
-    };
-    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)]
@@ -457,7 +463,18 @@ impl LocalDisk {
             mutation_lease.clone(),
         )
         .await?;
-        let has_dst_buf = read_rename_destination_metadata(&dst_file_path, &rename_commit_guard, mutation_lease.clone()).await?;
+        let mut commit = RenameCommitContext {
+            directory_guard: rename_commit_guard,
+            publication_root: &self.publication_root,
+            mutation_lease,
+            source_metadata: &src_file_path,
+            destination_metadata: &dst_file_path,
+            source_volume_dir: &src_volume_dir,
+            data_paths: has_data_dir_path.as_ref(),
+        };
+        let has_dst_buf =
+            read_rename_destination_metadata(commit.destination_metadata, &commit.directory_guard, commit.mutation_lease.clone())
+                .await?;
 
         if no_inline {
             // Non-inline: read xl.meta, parse, write, rename data dir, rename xl.meta
@@ -533,18 +550,18 @@ impl LocalDisk {
             // rename below to report through the existing rollback path. Payload
             // durability is kept by both strict and relaxed.
             let tmp_meta_write = {
-                let src_file_path = src_file_path.clone();
+                let src_file_path = commit.source_metadata.to_path_buf();
                 let dst_file_path = dst_file_path.clone();
-                let rename_commit_guard = rename_commit_guard.clone();
-                let mutation_lease = mutation_lease.clone();
+                let owned_directory_guard = commit.directory_guard.clone();
+                let owned_mutation_lease = commit.mutation_lease.clone();
                 async move {
-                    os::run_blocking_namespace_operation(mutation_lease, move || {
+                    os::run_blocking_namespace_operation(owned_mutation_lease, move || {
                         #[cfg(test)]
                         run_owned_file_write_before_open(&src_file_path);
                         let mut prepared_metadata_source = os::create_prepared_rename_source_with_commit_guard(
                             &src_file_path,
                             &dst_file_path,
-                            &rename_commit_guard,
+                            &owned_directory_guard,
                         )?;
                         prepared_metadata_source.write_all(&new_dst_buf, tmp_meta_sync != SyncMode::None)?;
                         Ok(prepared_metadata_source)
@@ -570,14 +587,14 @@ impl LocalDisk {
             // sequential version did.
             let prepared_metadata_source = tmp_meta_res?;
             shard_sync_res?;
-            let rename_commit_guard = remove_dst_base_before_commit(
+            commit.directory_guard = remove_dst_base_before_commit(
                 dst_path,
-                rename_commit_guard,
+                commit.directory_guard,
                 src_file_parent,
                 dst_file_parent,
                 &dst_volume_dir,
-                &self.publication_root,
-                mutation_lease.clone(),
+                commit.publication_root,
+                commit.mutation_lease.clone(),
             )
             .await?;
             if should_remove_staged_meta_before_commit(dst_path) {
@@ -595,7 +612,7 @@ impl LocalDisk {
             if fi_healing
                 && let Some((_, dst_data_path)) = has_data_dir_path.as_ref()
                 && let Err(err) = self
-                    .move_to_trash_with_namespace_owner(dst_data_path, true, false, Some(mutation_lease.clone()))
+                    .move_to_trash_with_namespace_owner(dst_data_path, true, false, Some(commit.mutation_lease.clone()))
                     .await
             {
                 warn!(
@@ -613,9 +630,9 @@ impl LocalDisk {
                     src_data_path,
                     dst_data_path,
                     &skip_parent,
-                    &self.publication_root,
-                    &rename_commit_guard,
-                    mutation_lease.clone(),
+                    commit.publication_root,
+                    &commit.directory_guard,
+                    commit.mutation_lease.clone(),
                 )
                 .await
             {
@@ -630,13 +647,7 @@ impl LocalDisk {
                     error = ?err,
                     "Disk local rename flow failed"
                 );
-                restore_published_data_source(
-                    has_data_dir_path.as_ref(),
-                    &src_volume_dir,
-                    &self.publication_root,
-                    mutation_lease.clone(),
-                )
-                .await?;
+                commit.restore_published_data_source().await?;
                 return Err(err);
             }
             #[cfg(test)]
@@ -661,13 +672,7 @@ impl LocalDisk {
                     reason = "test_fail_before_old_metadata_backup",
                     "Disk local rename flow failed before metadata commit"
                 );
-                restore_published_data_source(
-                    has_data_dir_path.as_ref(),
-                    &src_volume_dir,
-                    &self.publication_root,
-                    mutation_lease.clone(),
-                )
-                .await?;
+                commit.restore_published_data_source().await?;
                 return Err(DiskError::Unexpected);
             }
 
@@ -685,25 +690,16 @@ impl LocalDisk {
                 let backup_parent = dst_file_parent.join(old_data_dir.to_string());
                 #[cfg(not(windows))]
                 if let Err(err) = os::make_dir_all(&backup_parent, &skip_parent).await {
-                    restore_published_data_source(
-                        has_data_dir_path.as_ref(),
-                        &src_volume_dir,
-                        &self.publication_root,
-                        mutation_lease.clone(),
-                    )
-                    .await?;
+                    commit.restore_published_data_source().await?;
                     return Err(err);
                 }
-                let backup_path_guard = match rename_commit_guard.create_destination_directory_for_path_access(&backup_parent) {
+                let backup_path_guard = match commit
+                    .directory_guard
+                    .create_destination_directory_for_path_access(&backup_parent)
+                {
                     Ok(guard) => guard,
                     Err(err) => {
-                        restore_published_data_source(
-                            has_data_dir_path.as_ref(),
-                            &src_volume_dir,
-                            &self.publication_root,
-                            mutation_lease.clone(),
-                        )
-                        .await?;
+                        commit.restore_published_data_source().await?;
                         return Err(DiskError::from(to_file_error(err)));
                     }
                 };
@@ -711,13 +707,7 @@ impl LocalDisk {
                 if let Err(err) = check_path_length(backup_path.to_string_lossy().as_ref()) {
                     #[cfg(windows)]
                     drop(backup_path_guard);
-                    restore_published_data_source(
-                        has_data_dir_path.as_ref(),
-                        &src_volume_dir,
-                        &self.publication_root,
-                        mutation_lease.clone(),
-                    )
-                    .await?;
+                    commit.restore_published_data_source().await?;
                     return Err(err);
                 }
                 let backup_bytes = dst_buf.clone();
@@ -726,7 +716,7 @@ impl LocalDisk {
                 // sync finish. A detached spawn_blocking writer could survive
                 // cancellation and later truncate a newer transaction's
                 // deterministic rollback backup.
-                let write_result = os::run_blocking_namespace_operation(mutation_lease.clone(), move || {
+                let write_result = os::run_blocking_namespace_operation(commit.mutation_lease.clone(), move || {
                     #[cfg(test)]
                     run_owned_file_write_before_open(&backup_path);
                     backup_path_guard.write_file_for_path_access(
@@ -749,13 +739,7 @@ impl LocalDisk {
                         error = ?err,
                         "Disk local rename flow failed"
                     );
-                    restore_published_data_source(
-                        has_data_dir_path.as_ref(),
-                        &src_volume_dir,
-                        &self.publication_root,
-                        mutation_lease.clone(),
-                    )
-                    .await?;
+                    commit.restore_published_data_source().await?;
                     return Err(err);
                 }
             }
@@ -773,9 +757,9 @@ impl LocalDisk {
                 &src_file_path,
                 &dst_file_path,
                 &skip_parent,
-                &self.publication_root,
-                &rename_commit_guard,
-                mutation_lease.clone(),
+                commit.publication_root,
+                &commit.directory_guard,
+                commit.mutation_lease.clone(),
             )
             .await
             {
@@ -790,20 +774,11 @@ impl LocalDisk {
                     error = ?err,
                     "Disk local rename flow failed"
                 );
-                restore_published_data_source(
-                    has_data_dir_path.as_ref(),
-                    &src_volume_dir,
-                    &self.publication_root,
-                    mutation_lease.clone(),
-                )
-                .await?;
+                commit.restore_published_data_source().await?;
                 return Err(err);
             }
-
-            let committed_new_data_path = has_data_dir_path.as_ref().map(|(_, dst_data_path)| dst_data_path.as_path());
             if should_fail_after_metadata_commit(dst_path) {
-                rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir)
-                    .map_err(to_file_error)?;
+                commit.rollback_non_inline_std(rollback_data_dir).map_err(to_file_error)?;
                 return Err(DiskError::Unexpected);
             }
 
@@ -823,13 +798,12 @@ 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, Some(mutation_lease.clone())).await {
+                if let Err(err) = os::fsync_dst_dir_group_commit(parent, Some(commit.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,
                     );
-                    rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir)
-                        .map_err(to_file_error)?;
+                    commit.rollback_non_inline_std(rollback_data_dir).map_err(to_file_error)?;
                     // The commit rename changed the dst part inodes before this fsync
                     // failed and rolled them back; drop any fd cached during that
                     // window so readers re-open the restored inode (rustfs/backlog#1177).
@@ -861,13 +835,12 @@ impl LocalDisk {
                         break;
                     }
                     let fsync_started = rustfs_io_metrics::put_stage_timer();
-                    if let Err(err) = os::fsync_dir_with_owner(dir, Some(mutation_lease.clone())).await {
+                    if let Err(err) = os::fsync_dir_with_owner(dir, Some(commit.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,
                         );
-                        rollback_committed_rename_std(&dst_file_path, committed_new_data_path, rollback_data_dir)
-                            .map_err(to_file_error)?;
+                        commit.rollback_non_inline_std(rollback_data_dir).map_err(to_file_error)?;
                         // Same post-commit rollback window as above — drop cached
                         // dst part fds so readers re-open the restored inode
                         // (rustfs/backlog#1177).
@@ -891,7 +864,7 @@ impl LocalDisk {
             // complete. Do not retain the Windows object identity guard while
             // cleaning staging paths or invalidating cached descriptors.
             #[cfg(windows)]
-            drop(rename_commit_guard);
+            drop(commit.directory_guard);
 
             if let Some(src_file_path_parent) = src_file_path.parent() {
                 if src_volume != super::super::RUSTFS_META_MULTIPART_BUCKET {
@@ -939,7 +912,7 @@ impl LocalDisk {
             let dst_path_for_failpoint = dst_path.to_string();
             #[cfg(windows)]
             let source_parent = src_file_parent.to_path_buf();
-            let rename_commit_guard_for_preparation = rename_commit_guard.clone();
+            let rename_commit_guard_for_preparation = commit.directory_guard.clone();
             let sync = durability.syncs_commit_metadata();
             #[cfg(test)]
             run_inline_before_file_sync_admission(dst_path);
@@ -1022,7 +995,7 @@ impl LocalDisk {
                     if sync {
                         std::fs::File::open(&backup_path)?.sync_data()?;
                     }
-                    staged_rollback_path = Some(backup_path);
+                    staged_rollback_path = Some(StagedRollbackBackup(backup_path));
                 }
 
                 Ok::<_, std::io::Error>((
@@ -1036,9 +1009,10 @@ impl LocalDisk {
                 ))
             };
             let inline_preparation = if let Some(admission) = file_sync_admission.as_ref() {
-                os::run_blocking_namespace_file_sync_operation(mutation_lease.clone(), admission, prepare_inline_metadata).await
+                os::run_blocking_namespace_file_sync_operation(commit.mutation_lease.clone(), admission, prepare_inline_metadata)
+                    .await
             } else {
-                os::run_blocking_namespace_operation(mutation_lease.clone(), prepare_inline_metadata).await
+                os::run_blocking_namespace_operation(commit.mutation_lease.clone(), prepare_inline_metadata).await
             }
             .map_err(to_file_error)
             .map_err(DiskError::from);
@@ -1061,28 +1035,30 @@ impl LocalDisk {
                 }
             };
 
-            let rename_commit_guard = remove_dst_base_before_commit(
+            commit.directory_guard = remove_dst_base_before_commit(
                 dst_path,
-                rename_commit_guard,
+                commit.directory_guard,
                 src_file_parent,
                 dst_file_parent,
                 &dst_volume_dir,
-                &self.publication_root,
-                mutation_lease.clone(),
+                commit.publication_root,
+                commit.mutation_lease.clone(),
             )
             .await?;
 
             if should_remove_staged_meta_before_commit(dst_path) {
                 drop(prepared_metadata_source);
                 let remove_result = std::fs::remove_file(&src_file_path);
-                if let Some(backup_path) = local_rollback_path.as_deref() {
+                if let Some(backup_path) = local_rollback_path.as_ref().map(StagedRollbackBackup::path) {
                     let _ = remove_file_if_exists(backup_path);
                 }
                 remove_result.map_err(to_file_error)?;
                 return Err(DiskError::FileNotFound);
             }
 
-            if let (Some(rollback_data_dir), Some(staged_backup)) = (rollback_data_dir, local_rollback_path.as_deref()) {
+            if let (Some(rollback_data_dir), Some(staged_backup)) =
+                (rollback_data_dir, local_rollback_path.as_ref().map(StagedRollbackBackup::path))
+            {
                 let Some(dst_parent) = dst_file_path.parent() else {
                     return Err(DiskError::other("missing object metadata parent"));
                 };
@@ -1096,8 +1072,8 @@ impl LocalDisk {
                     staged_backup,
                     &backup_path,
                     &dst_volume_dir,
-                    &self.publication_root,
-                    Some(mutation_lease.clone()),
+                    commit.publication_root,
+                    Some(commit.mutation_lease.clone()),
                 )
                 .await
                 {
@@ -1119,7 +1095,8 @@ impl LocalDisk {
                 {
                     let fsync_started = rustfs_io_metrics::put_stage_timer();
                     if let Err(err) =
-                        os::fsync_dir_with_namespace_file_sync_limit(backup_parent, mutation_lease.clone(), admission).await
+                        os::fsync_dir_with_namespace_file_sync_limit(backup_parent, commit.mutation_lease.clone(), admission)
+                            .await
                     {
                         rustfs_io_metrics::record_put_object_stage_duration_from(
                             rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_BACKUP_DIR_FSYNC,
@@ -1143,14 +1120,14 @@ impl LocalDisk {
                     &src_file_path,
                     &dst_file_path,
                     &dst_volume_dir,
-                    &self.publication_root,
-                    &rename_commit_guard,
-                    mutation_lease.clone(),
+                    commit.publication_root,
+                    &commit.directory_guard,
+                    commit.mutation_lease.clone(),
                 )
                 .await
             };
             if let Err(err) = commit_result {
-                if let Some(backup_path) = local_rollback_path.as_deref() {
+                if let Some(backup_path) = local_rollback_path.as_ref().map(StagedRollbackBackup::path) {
                     let _ = remove_file_if_exists(backup_path);
                 }
                 for part_path in &invalidate_part_paths {
@@ -1161,7 +1138,7 @@ impl LocalDisk {
 
             let post_commit = async {
                 if should_fail_after_metadata_commit(dst_path) {
-                    rollback_inline_metadata_commit_std(&dst_file_path, rollback_data_dir, local_rollback_path.as_deref())?;
+                    commit.rollback_inline_std(rollback_data_dir, local_rollback_path.as_ref())?;
                     return Err(std::io::Error::other("test fail after metadata commit"));
                 }
 
@@ -1170,15 +1147,18 @@ impl LocalDisk {
                     && let Some(dst_parent) = dst_file_path.parent()
                 {
                     let fsync_started = rustfs_io_metrics::put_stage_timer();
-                    if let Err(err) =
-                        os::fsync_dst_dir_group_commit_or_namespace_file_sync_limit(dst_parent, mutation_lease.clone(), admission)
-                            .await
+                    if let Err(err) = os::fsync_dst_dir_group_commit_or_namespace_file_sync_limit(
+                        dst_parent,
+                        commit.mutation_lease.clone(),
+                        admission,
+                    )
+                    .await
                     {
                         rustfs_io_metrics::record_put_object_stage_duration_from(
                             rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_DST_DIR_FSYNC,
                             fsync_started,
                         );
-                        rollback_inline_metadata_commit_std(&dst_file_path, rollback_data_dir, local_rollback_path.as_deref())?;
+                        commit.rollback_inline_std(rollback_data_dir, local_rollback_path.as_ref())?;
                         return Err(err);
                     }
                     rustfs_io_metrics::record_put_object_stage_duration_from(
@@ -1204,17 +1184,14 @@ impl LocalDisk {
                         }
                         let fsync_started = rustfs_io_metrics::put_stage_timer();
                         if let Err(err) =
-                            os::fsync_dir_with_namespace_file_sync_limit(ancestor_dir, mutation_lease.clone(), admission).await
+                            os::fsync_dir_with_namespace_file_sync_limit(ancestor_dir, commit.mutation_lease.clone(), admission)
+                                .await
                         {
                             rustfs_io_metrics::record_put_object_stage_duration_from(
                                 rustfs_io_metrics::PUT_STAGE_SET_DISK_RENAME_ANCESTOR_DIR_FSYNC,
                                 fsync_started,
                             );
-                            rollback_inline_metadata_commit_std(
-                                &dst_file_path,
-                                rollback_data_dir,
-                                local_rollback_path.as_deref(),
-                            )?;
+                            commit.rollback_inline_std(rollback_data_dir, local_rollback_path.as_ref())?;
                             return Err(err);
                         }
                         rustfs_io_metrics::record_put_object_stage_duration_from(
@@ -1251,9 +1228,9 @@ impl LocalDisk {
             // The commit no longer has a rollback path. Release the Windows
             // object identity guard before best-effort staging cleanup.
             #[cfg(windows)]
-            drop(rename_commit_guard);
+            drop(commit.directory_guard);
 
-            if let Some(backup_path) = local_rollback_path.as_deref() {
+            if let Some(backup_path) = local_rollback_path.as_ref().map(StagedRollbackBackup::path) {
                 let _ = remove_file_if_exists(backup_path);
             }
 

crates/ecstore/src/ecstore_validation_blackbox.rs

@@ -259,7 +259,7 @@ async fn blackbox_put_unknown_actual_size_restores_body_and_records_written_size
         .map(|idx| ((idx * 17) % 251) as u8)
         .collect::<Vec<_>>();
     let opts = ObjectOptions {
-        no_lock: true,
+        write_completion: crate::object_api::WriteCompletion::TailDrained,
         ..Default::default()
     };
 
@@ -300,7 +300,7 @@ async fn blackbox_get_restores_body_after_one_shard_file_is_removed() {
         .map(|idx| ((idx * 19) % 251) as u8)
         .collect::<Vec<_>>();
     let opts = ObjectOptions {
-        no_lock: true,
+        write_completion: crate::object_api::WriteCompletion::TailDrained,
         ..Default::default()
     };
 
@@ -406,7 +406,7 @@ async fn blackbox_heal_requests_preserve_repair_scope() {
                         put_object,
                         &mut put_reader,
                         &ObjectOptions {
-                            no_lock: true,
+                            write_completion: crate::object_api::WriteCompletion::TailDrained,
                             versioned: true,
                             expected_bucket_incarnation_id: Some(put_incarnation),
                             ..Default::default()
@@ -469,7 +469,7 @@ async fn blackbox_heal_requests_preserve_repair_scope() {
             put_object,
             &mut healthy_reader,
             &ObjectOptions {
-                no_lock: true,
+                write_completion: crate::object_api::WriteCompletion::TailDrained,
                 ..Default::default()
             },
         )
@@ -509,7 +509,7 @@ async fn blackbox_heal_requests_preserve_repair_scope() {
             .map(|idx| ((idx * 29) % 251) as u8)
             .collect::<Vec<_>>();
         let opts = ObjectOptions {
-            no_lock: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             ..Default::default()
         };
 
@@ -567,7 +567,7 @@ async fn blackbox_heal_requests_preserve_repair_scope() {
         let suspended_object = "suspended.bin";
         let payload = vec![0x5a; 1 << 20];
         let mpu_opts = ObjectOptions {
-            no_lock: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             versioned: true,
             ..Default::default()
         };
@@ -675,7 +675,7 @@ async fn blackbox_heal_requests_preserve_repair_scope() {
                 .expect("fourth disk should be online before suspended completion")
         };
         let suspended_opts = ObjectOptions {
-            no_lock: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             version_suspended: true,
             ..Default::default()
         };
@@ -712,7 +712,7 @@ async fn blackbox_range_read_restores_exact_slice_with_one_offline_disk() {
     let range_start = 513usize;
     let range_len = 8192usize;
     let opts = ObjectOptions {
-        no_lock: true,
+        write_completion: crate::object_api::WriteCompletion::TailDrained,
         ..Default::default()
     };
 
@@ -756,7 +756,7 @@ async fn blackbox_delete_marker_hides_object_body_without_erasing_prior_version_
     let bucket = "bb-delete-marker-read-negative";
     let object = "object.bin";
     let opts = ObjectOptions {
-        no_lock: true,
+        write_completion: crate::object_api::WriteCompletion::TailDrained,
         version_suspended: true,
         object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(ObjectLockConfigState::ConfirmedAbsent))),
         ..Default::default()
@@ -853,7 +853,7 @@ async fn blackbox_issue3031_diag_covers_put_success_cleanup_and_error_summary()
         let first_payload = b"first diagnostic body".to_vec();
         let second_payload = b"second diagnostic body that replaces the first".to_vec();
         let opts = ObjectOptions {
-            no_lock: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             ..Default::default()
         };
 
@@ -959,7 +959,7 @@ mod old_current_size_backfill {
             .await
             .expect("bucket should be created");
         let opts = ObjectOptions {
-            no_lock: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             ..Default::default()
         };
 
@@ -993,7 +993,7 @@ mod old_current_size_backfill {
         // latest's size, not the incoming version's.
         let versioned = "versioned.bin";
         let first_version_opts = ObjectOptions {
-            no_lock: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             versioned: true,
             version_id: Some(uuid::Uuid::new_v4().to_string()),
             ..Default::default()
@@ -1005,7 +1005,7 @@ mod old_current_size_backfill {
         let expected = prelookup_expectation(&set_disks, bucket, versioned).await;
         assert_eq!(expected, OldCurrentSize::Present(111));
         let second_version_opts = ObjectOptions {
-            no_lock: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             versioned: true,
             version_id: Some(uuid::Uuid::new_v4().to_string()),
             ..Default::default()
@@ -1037,7 +1037,7 @@ mod old_current_size_backfill {
         let expected = prelookup_expectation(&set_disks, bucket, versioned).await;
         assert_eq!(expected, OldCurrentSize::Present(0));
         let third_version_opts = ObjectOptions {
-            no_lock: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             versioned: true,
             version_id: Some(uuid::Uuid::new_v4().to_string()),
             ..Default::default()
@@ -1061,7 +1061,7 @@ mod old_current_size_backfill {
             .await
             .expect("bucket should be created");
         let opts = ObjectOptions {
-            no_lock: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             ..Default::default()
         };
 

crates/ecstore/src/object_api/mod.rs

@@ -104,6 +104,7 @@ mod hook_slot;
 mod object_mutation_hook;
 mod readers;
 mod types;
+pub(crate) mod write_commit_context;
 
 #[cfg(test)]
 mod persisted_metadata_keys_tests;
@@ -125,3 +126,4 @@ pub(crate) use object_mutation_hook::notify_object_mutation;
 pub use object_mutation_hook::{ObjectMutationHook, register_object_mutation_hook, unregister_object_mutation_hook};
 pub use readers::*;
 pub use types::*;
+pub use write_commit_context::{WriteCommitGuard, WriteLockContext};

crates/ecstore/src/object_api/types.rs

@@ -935,6 +935,9 @@ pub struct ObjectOptions {
     /// Persisted bucket incarnation observed before authorization.
     pub expected_bucket_incarnation_id: Option<Uuid>,
     pub no_lock: bool,
+    /// A genuine target write guard; loss fences alone do not authorize a borrowed write.
+    #[doc(hidden)]
+    pub write_lock_context: Option<WriteLockContext>,
     /// Internal read-only inspection must not enqueue metadata or payload repairs.
     pub suppress_read_repair: bool,
     /// Control-plane writers that immediately read or CAS the same namespace

crates/ecstore/src/object_api/write_commit_context.rs

@@ -0,0 +1,186 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::{ObjectOptions, WriteCompletion};
+use crate::error::{Error, Result};
+use std::sync::Arc;
+use std::time::Duration;
+
+/// A real write guard shared with an inner commit, never a lock-loss signal alone.
+///
+/// Callers cannot fabricate a guard without acquiring its namespace lock.
+///
+/// ```compile_fail
+/// use rustfs_ecstore::api::object::WriteCommitGuard;
+/// use std::sync::Arc;
+///
+/// let guard = WriteCommitGuard {
+///     bucket: "bucket".into(),
+///     object: "object".into(),
+///     guards: Arc::new(Vec::new()),
+/// };
+/// ```
+#[doc(hidden)]
+#[derive(Clone, Debug)]
+pub struct WriteCommitGuard {
+    bucket: Arc<str>,
+    object: Arc<str>,
+    guards: Arc<Vec<Arc<rustfs_lock::NamespaceLockGuard>>>,
+}
+
+impl WriteCommitGuard {
+    /// Acquire the key recorded by the namespace wrapper and retain its owner.
+    pub async fn acquire(
+        lock: &rustfs_lock::NamespaceLockWrapper,
+        timeout: Duration,
+    ) -> std::result::Result<Self, rustfs_lock::LockError> {
+        let guard = Arc::new(lock.get_write_lock(timeout).await?);
+        let key = lock.resource();
+        Ok(Self::from_guard(guard, &key.bucket, &key.object))
+    }
+
+    pub(crate) fn from_guard(guard: Arc<rustfs_lock::NamespaceLockGuard>, bucket: &str, object: &str) -> Self {
+        Self {
+            bucket: bucket.into(),
+            object: object.into(),
+            guards: Arc::new(vec![guard]),
+        }
+    }
+
+    pub fn is_lock_lost(&self) -> bool {
+        self.guards.iter().any(|guard| guard.is_lock_lost() || guard.is_released())
+    }
+
+    fn matches(&self, bucket: &str, object: &str) -> bool {
+        self.bucket.as_ref() == bucket && self.object.as_ref() == object
+    }
+}
+
+/// Opaque boundary capability; only real acquisitions or offline startup construct it.
+///
+/// ```compile_fail
+/// use rustfs_ecstore::object_api::WriteLockContext;
+///
+/// let context = WriteLockContext {};
+/// ```
+#[doc(hidden)]
+#[derive(Clone, Debug)]
+pub struct WriteLockContext {
+    mode: WriteLockMode,
+}
+
+#[derive(Clone, Debug)]
+enum WriteLockMode {
+    Borrowed(Vec<WriteCommitGuard>),
+    OfflineRecovery,
+}
+
+pub(crate) enum WriteCommitContext {
+    Owned,
+    Borrowed(Vec<WriteCommitGuard>),
+    Publication,
+    OfflineRecovery,
+    #[cfg(test)]
+    UnlockedFixture,
+}
+
+impl WriteCommitContext {
+    pub(crate) fn from_options(opts: &ObjectOptions, bucket: &str, object: &str, publication: bool) -> Result<Self> {
+        if publication {
+            if let Some(WriteLockContext {
+                mode: WriteLockMode::Borrowed(guards),
+            }) = opts.write_lock_context.as_ref()
+                && !guards.iter().any(|guard| guard.matches(bucket, object))
+            {
+                return Err(Error::InvalidArgument(
+                    bucket.to_owned(),
+                    object.to_owned(),
+                    "publication carries a different borrowed namespace".to_owned(),
+                ));
+            }
+            return Ok(Self::Publication);
+        }
+        match (opts.no_lock, opts.write_lock_context.as_ref().map(|context| &context.mode)) {
+            (false, None | Some(WriteLockMode::Borrowed(_))) => Ok(Self::Owned),
+            (true, Some(WriteLockMode::Borrowed(guards))) if guards.iter().any(|guard| guard.matches(bucket, object)) => {
+                Ok(Self::Borrowed(guards.clone()))
+            }
+            (true, Some(WriteLockMode::OfflineRecovery)) => Ok(Self::OfflineRecovery),
+            #[cfg(test)]
+            (true, None) if !opts.metadata_cache_safe => Ok(Self::UnlockedFixture),
+            _ => Err(Error::InvalidArgument(
+                bucket.to_owned(),
+                object.to_owned(),
+                "write lock context does not prove the target namespace".to_owned(),
+            )),
+        }
+    }
+
+    pub(crate) fn acquires_namespace(&self) -> bool {
+        matches!(self, Self::Owned)
+    }
+
+    pub(crate) fn has_borrowed_owner(&self) -> bool {
+        matches!(self, Self::Borrowed(_))
+    }
+
+    pub(crate) fn is_lock_lost(&self) -> bool {
+        matches!(self, Self::Borrowed(guards) if guards.iter().any(WriteCommitGuard::is_lock_lost))
+    }
+
+    pub(crate) fn allows_early_ack(&self, completion: WriteCompletion) -> bool {
+        completion == WriteCompletion::Quorum && matches!(self, Self::Owned | Self::Borrowed(_) | Self::Publication)
+    }
+}
+
+impl ObjectOptions {
+    pub fn add_write_commit_guard(&mut self, guard: &WriteCommitGuard) {
+        for owner in guard.guards.iter() {
+            self.add_namespace_lock_guard(owner);
+        }
+        let mut guards = match self.write_lock_context.as_ref() {
+            Some(WriteLockContext {
+                mode: WriteLockMode::Borrowed(existing),
+            }) => existing.clone(),
+            _ => Vec::new(),
+        };
+        if let Some(existing) = guards.iter_mut().find(|held| held.matches(&guard.bucket, &guard.object)) {
+            let owners = Arc::make_mut(&mut existing.guards);
+            for owner in guard.guards.iter() {
+                if !owners.iter().any(|held| Arc::ptr_eq(held, owner)) {
+                    owners.push(Arc::clone(owner));
+                }
+            }
+        } else {
+            guards.push(guard.clone());
+        }
+        self.no_lock = true;
+        self.write_lock_context = Some(WriteLockContext {
+            mode: WriteLockMode::Borrowed(guards),
+        });
+    }
+
+    pub(crate) fn add_owned_write_lock(&mut self, guard: Arc<rustfs_lock::NamespaceLockGuard>, bucket: &str, object: &str) {
+        self.add_write_commit_guard(&WriteCommitGuard::from_guard(guard, bucket, object));
+    }
+
+    /// Only startup migration may publish without a running namespace owner.
+    pub(crate) fn use_offline_recovery_write(&mut self) {
+        self.no_lock = true;
+        self.write_lock_context = Some(WriteLockContext {
+            mode: WriteLockMode::OfflineRecovery,
+        });
+        self.write_completion = WriteCompletion::TailDrained;
+    }
+}

crates/ecstore/src/services/rebalance/control.rs

@@ -521,11 +521,12 @@ impl ECStore {
                 .map_err(|err| rebalance_meta_lock_error(err, "write"))
         })
         .await?;
+        let guard = Arc::new(guard);
         let mut opts = ObjectOptions {
-            no_lock: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             ..Default::default()
         };
-        opts.add_namespace_lock_guard(&guard);
+        opts.add_owned_write_lock(Arc::clone(&guard), crate::disk::RUSTFS_META_BUCKET, REBAL_META_NAME);
 
         merge_and_save_rebalance_meta_no_lock(pool, local_snapshot, stage, opts, None, expected_id).await?;
         if guard.is_lock_lost() {

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

@@ -29,6 +29,27 @@ use crate::disk::error_reduce::OBJECT_OP_IGNORED_ERRS;
 use crate::set_disk::file_info_is_valid_for_metadata;
 use rustfs_filemeta::FileInfo;
 
+/// One disk's observation. Pending is neither a successful vote nor an offline
+/// disk: the scheduler may not have issued this slot or may still be waiting.
+// Option<Result<...>> keeps FileInfo inline without a per-response Box.
+// None is pending; Some(Ok(_)) is success; Some(Err(_)) retains a disk error.
+pub(in crate::set_disk) type MetadataDiskResult = Option<crate::disk::error::Result<FileInfo>>;
+
+#[derive(Debug)]
+pub(in crate::set_disk) struct MetadataDiskObservation {
+    pub(in crate::set_disk) disk_index: usize,
+    pub(in crate::set_disk) result: MetadataDiskResult,
+}
+
+impl MetadataDiskObservation {
+    pub(in crate::set_disk) fn file_info(&self) -> Option<&FileInfo> {
+        match &self.result {
+            Some(Ok(metadata)) => Some(metadata),
+            None | Some(Err(_)) => None,
+        }
+    }
+}
+
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub(in crate::set_disk) struct MetadataEarlyStopDecision {
     pub(in crate::set_disk) reason: &'static str,
@@ -59,6 +80,14 @@ pub(in crate::set_disk) struct MetadataQuorumAccumulator {
 }
 
 impl MetadataQuorumAccumulator {
+    pub(in crate::set_disk) fn observe(&mut self, observation: &MetadataDiskObservation) {
+        match &observation.result {
+            None => {}
+            Some(Ok(metadata)) => self.observe_file_info_at(observation.disk_index, metadata),
+            Some(Err(error)) => self.observe_error(error),
+        }
+    }
+
     pub(in crate::set_disk) fn new(total_disks: usize, default_parity_count: usize, allow_early_stop: bool) -> Self {
         Self {
             total_disks,
@@ -86,6 +115,7 @@ impl MetadataQuorumAccumulator {
         self
     }
 
+    #[cfg(test)]
     pub(in crate::set_disk) fn observe_file_info(&mut self, file_info: &FileInfo) {
         self.observe_file_info_with_index(None, file_info);
     }
@@ -383,3 +413,186 @@ pub(in crate::set_disk) fn metadata_early_stop_candidate_matches(left: &FileInfo
 pub(in crate::set_disk) fn is_metadata_fanout_ignored_error(err: &DiskError) -> bool {
     OBJECT_OP_IGNORED_ERRS.iter().any(|ignored| ignored == err)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::set_disk::SetDisks;
+    use time::OffsetDateTime;
+    use uuid::Uuid;
+
+    fn payload(disk_index: usize) -> FileInfo {
+        let mut metadata = FileInfo::new("object", 2, 2);
+        metadata.volume = "bucket".to_string();
+        metadata.name = "object".to_string();
+        metadata.size = 1;
+        metadata.mod_time = Some(OffsetDateTime::UNIX_EPOCH);
+        metadata.data_dir = Some(Uuid::from_u128(1));
+        metadata.erasure.index = metadata.erasure.distribution[disk_index];
+        metadata.metadata.insert("etag".to_string(), "etag".to_string());
+        metadata.add_object_part(1, "etag".to_string(), 1, None, 1, None, None);
+        assert!(file_info_is_valid_for_metadata(&metadata), "the observation corpus needs a valid payload");
+        metadata
+    }
+
+    fn observation(disk_index: usize, state: usize) -> MetadataDiskObservation {
+        MetadataDiskObservation {
+            disk_index,
+            result: match state {
+                0 => Some(Ok(payload(disk_index))),
+                1 => Some(Err(DiskError::FileNotFound)),
+                2 => Some(Err(DiskError::FileCorrupt)),
+                3 => Some(Err(DiskError::DiskNotFound)),
+                4 => None,
+                _ => panic!("unexpected test state"),
+            },
+        }
+    }
+
+    fn assert_same_reduction(typed: &MetadataQuorumAccumulator, legacy: &MetadataQuorumAccumulator) {
+        assert_eq!(typed.early_stop_decision(), legacy.early_stop_decision());
+        assert_eq!(typed.version_early_stop_decision(), legacy.version_early_stop_decision());
+        assert_eq!(typed.final_miss_reason(), legacy.final_miss_reason());
+        assert_eq!(typed.valid_responses, legacy.valid_responses);
+        assert_eq!(typed.not_found_responses, legacy.not_found_responses);
+        assert_eq!(typed.version_not_found_responses, legacy.version_not_found_responses);
+        assert_eq!(typed.ignored_errors, legacy.ignored_errors);
+        assert_eq!(typed.hard_errors, legacy.hard_errors);
+        assert_eq!(typed.candidate_votes, legacy.candidate_votes);
+        assert_eq!(typed.candidate_shard_mask, legacy.candidate_shard_mask);
+        assert_eq!(typed.matching_version_votes, legacy.matching_version_votes);
+        assert_eq!(typed.delete_marker_votes, legacy.delete_marker_votes);
+        assert_eq!(typed.conflicting_metadata, legacy.conflicting_metadata);
+        assert_eq!(
+            typed.candidate.as_ref().map(SetDisks::file_info_quorum_hash),
+            legacy.candidate.as_ref().map(SetDisks::file_info_quorum_hash)
+        );
+    }
+
+    fn observe_legacy(accumulator: &mut MetadataQuorumAccumulator, observation: &MetadataDiskObservation) {
+        match &observation.result {
+            None => {}
+            Some(Ok(metadata)) => accumulator.observe_file_info_at(observation.disk_index, metadata),
+            Some(Err(error)) => accumulator.observe_error(error),
+        }
+    }
+
+    #[test]
+    fn metadata_observation_all_four_slot_states_and_arrival_orders_preserve_reduction() {
+        let permutations = (0..4)
+            .flat_map(|a| (0..4).filter(move |b| *b != a).map(move |b| (a, b)))
+            .flat_map(|(a, b)| (0..4).filter(move |c| *c != a && *c != b).map(move |c| (a, b, c)))
+            .map(|(a, b, c)| [a, b, c, 6 - a - b - c])
+            .collect::<Vec<_>>();
+        assert_eq!(permutations.len(), 24);
+        // 5^4 states and all 4! arrivals, both with early-stop enabled and disabled.
+        for encoded in 0..625usize {
+            let states = [encoded % 5, encoded / 5 % 5, encoded / 25 % 5, encoded / 125 % 5];
+            let inputs = std::array::from_fn::<_, 4, _>(|index| observation(index, states[index]));
+            for enabled in [false, true] {
+                for order in &permutations {
+                    let mut typed = MetadataQuorumAccumulator::new(4, 2, enabled);
+                    let mut legacy = MetadataQuorumAccumulator::new(4, 2, enabled);
+                    for &index in order {
+                        typed.observe(&inputs[index]);
+                        observe_legacy(&mut legacy, &inputs[index]);
+                        assert_same_reduction(&typed, &legacy);
+                    }
+                    assert_eq!(typed.valid_responses, states.iter().filter(|&&state| state == 0).count());
+                    assert_eq!(typed.not_found_responses, states.iter().filter(|&&state| state == 1).count());
+                    assert_eq!(typed.hard_errors, states.iter().filter(|&&state| state == 2).count());
+                    assert_eq!(typed.ignored_errors, states.iter().filter(|&&state| state == 3).count());
+                    let expected_early_stop =
+                        enabled && typed.valid_responses >= 3 && !states.iter().any(|&state| state == 1 || state == 2);
+                    assert_eq!(
+                        typed.early_stop_decision().is_some(),
+                        expected_early_stop,
+                        "states={states:?}, order={order:?}"
+                    );
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn metadata_observation_pending_newer_version_and_same_time_different_directory_force_full_wait() {
+        let mut accumulator = MetadataQuorumAccumulator::new(4, 2, true);
+        for index in 0..2 {
+            accumulator.observe(&observation(index, 0));
+        }
+        accumulator.observe(&observation(2, 4));
+        assert_eq!(accumulator.candidate_votes, 2, "an outstanding response cannot supply the deciding vote");
+        assert_eq!(accumulator.early_stop_decision(), None);
+
+        let mut newer = payload(2);
+        newer.mod_time = Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(1));
+        accumulator.observe(&MetadataDiskObservation {
+            disk_index: 2,
+            result: Some(Ok(newer)),
+        });
+        assert!(accumulator.conflicting_metadata);
+        assert_eq!(accumulator.early_stop_decision(), None);
+
+        let mut same_time_different_directory = payload(3);
+        same_time_different_directory.data_dir = Some(Uuid::from_u128(2));
+        accumulator.observe(&MetadataDiskObservation {
+            disk_index: 3,
+            result: Some(Ok(same_time_different_directory)),
+        });
+        assert_eq!(accumulator.early_stop_decision(), None);
+    }
+
+    #[test]
+    fn metadata_observation_duplicate_shard_cannot_supply_a_read_reserve() {
+        let mut accumulator = MetadataQuorumAccumulator::new(4, 2, true);
+        let first = payload(0);
+        for disk_index in 0..3 {
+            accumulator.observe(&MetadataDiskObservation {
+                disk_index,
+                result: Some(Ok(first.clone())),
+            });
+        }
+        assert_eq!(accumulator.candidate_votes, 3);
+        assert!(
+            !accumulator.candidate_has_read_reserve(),
+            "copied erasure indexes cannot supply independent data shards"
+        );
+    }
+
+    #[test]
+    fn metadata_observation_null_versions_markers_and_invalid_success_keep_their_meaning() {
+        let mut null_version = MetadataQuorumAccumulator::new(4, 2, true);
+        for index in 0..3 {
+            null_version.observe(&observation(index, 0));
+        }
+        assert!(null_version.early_stop_decision().is_some());
+        assert_eq!(null_version.matching_version_votes, 0);
+
+        let mut marker = MetadataQuorumAccumulator::new(4, 2, true);
+        for disk_index in 0..3 {
+            let metadata = FileInfo {
+                volume: "bucket".to_string(),
+                name: "object".to_string(),
+                deleted: true,
+                mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(1)),
+                ..Default::default()
+            };
+            assert!(metadata.is_canonical_delete_marker(), "the null marker fixture must reach marker voting");
+            marker.observe(&MetadataDiskObservation {
+                disk_index,
+                result: Some(Ok(metadata)),
+            });
+        }
+        assert_eq!(marker.delete_marker_votes, 3);
+        assert!(marker.early_stop_decision().is_some());
+
+        let mut invalid = MetadataQuorumAccumulator::new(4, 2, true);
+        invalid.observe(&MetadataDiskObservation {
+            disk_index: 0,
+            result: Some(Ok(FileInfo::default())),
+        });
+        assert_eq!(invalid.valid_responses, 0);
+        assert_eq!(invalid.hard_errors, 1);
+        assert_eq!(invalid.early_stop_decision(), None);
+    }
+}

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

@@ -18,4 +18,5 @@
 //! duplicating read/write/erasure logic.
 
 pub(crate) mod io_primitives;
-mod metadata_quorum;
+pub(in crate::set_disk) mod metadata_quorum;
+pub(in crate::set_disk) mod metadata_read;

crates/ecstore/src/set_disk/mod.rs

@@ -868,7 +868,7 @@ static OBJECT_LOCK_DIAG_ENABLED: OnceLock<bool> = OnceLock::new();
 mod core;
 #[cfg(test)]
 pub(crate) use core::io_primitives::disk_call_counters;
-#[cfg(all(test, feature = "test-util"))]
+#[cfg(test)]
 pub(crate) use core::io_primitives::{ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, rename_fanout_barrier};
 mod ctx;
 mod metadata;
@@ -5527,12 +5527,29 @@ fn collect_inline_data_shard_fileinfos_by_index_or_reason<'a>(
     parts_metadata: &'a [FileInfo],
     fi: &FileInfo,
     data_shards: usize,
+    disk_is_online: impl FnMut(usize) -> bool,
+) -> std::result::Result<Vec<&'a FileInfo>, &'static str> {
+    collect_inline_data_shard_fileinfos_from_observations(
+        parts_metadata
+            .iter()
+            .enumerate()
+            .map(|(index, metadata)| (index, Some(metadata))),
+        fi,
+        data_shards,
+        disk_is_online,
+    )
+}
+
+fn collect_inline_data_shard_fileinfos_from_observations<'a>(
+    observations: impl IntoIterator<Item = (usize, Option<&'a FileInfo>)>,
+    fi: &FileInfo,
+    data_shards: usize,
     mut disk_is_online: impl FnMut(usize) -> bool,
 ) -> std::result::Result<Vec<&'a FileInfo>, &'static str> {
     let distribution = &fi.erasure.distribution;
     let mut data_files = vec![None; data_shards];
 
-    for (disk_index, file_info) in parts_metadata.iter().enumerate() {
+    for (disk_index, file_info) in observations {
         if !disk_is_online(disk_index) {
             continue;
         }
@@ -5542,9 +5559,9 @@ fn collect_inline_data_shard_fileinfos_by_index_or_reason<'a>(
         if block_index == 0 || block_index > data_shards {
             continue;
         }
-        if file_info.name.is_empty() {
+        let Some(file_info) = file_info.filter(|metadata| !metadata.name.is_empty()) else {
             return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_MISSING_SHARD);
-        }
+        };
         if file_info.erasure.index != block_index {
             return Err(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_IDENTITY_MISMATCH);
         }

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

@@ -598,17 +598,18 @@ impl SetDisks {
         }
 
         scope.check()?;
-        let (_, errors) =
-            Self::read_all_fileinfo(disks.as_slice(), "", bucket, object, &version.to_string(), false, false, false).await?;
+        let metadata_read =
+            Self::read_metadata(disks.as_slice(), "", bucket, object, &version.to_string(), false, false, false).await?;
+        let all_absent = metadata_read
+            .slots
+            .iter()
+            .all(|slot| matches!(&slot.result, Some(Err(DiskError::FileNotFound | DiskError::FileVersionNotFound))));
+
         let selected = self.get_disks_internal().await;
         let same_targets = selected.len() == disks.len() && selected.iter().zip(&disks).all(|(current, original)| {
             matches!((current, original), (Some(current), Some(original)) if std::sync::Arc::ptr_eq(current, original))
         });
-        if !same_targets
-            || !errors
-                .iter()
-                .all(|error| matches!(error, Some(DiskError::FileNotFound | DiskError::FileVersionNotFound)))
-        {
+        if !same_targets || !all_absent {
             return Err(StorageError::SlowDown);
         }
         scope.check()?;
@@ -734,7 +735,7 @@ impl SetDisks {
         object: &str,
         version_id: &str,
     ) -> disk::error::Result<ReadRepairCommitFingerprint> {
-        let (parts_metadata, errs, _) = Self::read_all_fileinfo_observed(
+        let metadata_read = Self::read_metadata_observed(
             disks,
             "",
             bucket,
@@ -747,6 +748,7 @@ impl SetDisks {
             self.default_parity_count,
         )
         .await?;
+        let (parts_metadata, errs, _) = metadata_read.into_legacy();
         let (read_quorum, _) = Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count)?;
         let read_quorum = usize::try_from(read_quorum).map_err(|_| DiskError::ErasureReadQuorum)?;
         let (_, quorum_mod_time, quorum_etag) = Self::list_online_disks(disks, &parts_metadata, &errs, read_quorum);
@@ -931,8 +933,9 @@ impl SetDisks {
             }
         };
 
-        let (mut parts_metadata, errs) =
-            Self::read_all_fileinfo(&disks, "", bucket, object, version_id, true, true, false).await?;
+        let (mut parts_metadata, errs, _) = Self::read_metadata(&disks, "", bucket, object, version_id, true, true, false)
+            .await?
+            .into_legacy();
 
         trace!(
             event = EVENT_SET_DISK_HEAL,
@@ -1284,8 +1287,10 @@ impl SetDisks {
                             // in `parts_metadata` to defaults. Re-read only before
                             // destructive cleanup so the guard sees every original
                             // identity.
-                            let (delete_guard_metadata, delete_guard_errs) =
-                                Self::read_all_fileinfo(&disks, "", bucket, object, version_id, true, true, false).await?;
+                            let (delete_guard_metadata, delete_guard_errs, _) =
+                                Self::read_metadata(&disks, "", bucket, object, version_id, true, true, false)
+                                    .await?
+                                    .into_legacy();
                             if self
                                 .dangling_delete_safety(bucket, object, &delete_guard_metadata, &delete_guard_errs, &disks)
                                 .await?
@@ -2316,16 +2321,16 @@ impl SetDisks {
                 return Err(DiskError::retired_marker_deferred(format!("conditional marker deletion failed: {error}")));
             }
         }
-        let (_, errors) = Self::read_all_fileinfo(disks, "", bucket, object, &version.to_string(), false, false, false).await?;
+        let metadata_read = Self::read_metadata(disks, "", bucket, object, &version.to_string(), false, false, false).await?;
+        let all_absent = metadata_read
+            .slots
+            .iter()
+            .all(|slot| matches!(&slot.result, Some(Err(DiskError::FileNotFound | DiskError::FileVersionNotFound))));
         let selected = self.get_disks_internal().await;
         let same_targets = selected.len() == disks.len() && selected.iter().zip(disks).all(|(current, original)| {
             matches!((current, original), (Some(current), Some(original)) if std::sync::Arc::ptr_eq(current, original))
         });
-        if !same_targets
-            || !errors
-                .iter()
-                .all(|error| matches!(error, Some(DiskError::FileNotFound | DiskError::FileVersionNotFound)))
-        {
+        if !same_targets || !all_absent {
             return Err(DiskError::retired_marker_deferred(
                 "complete marker absence could not be verified under the original target set",
             ));
@@ -3383,9 +3388,10 @@ impl SetDisks {
         // The inner heal and missing-object report read the registry again;
         // release this snapshot guard before a topology writer can queue between reads.
         let disks = self.get_disks_internal().await;
-        let (_, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, version_id, false, false, false)
+        let (_, errs, _) = Self::read_metadata(&disks, "", bucket, object, version_id, false, false, false)
             .await
-            .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
+            .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?
+            .into_legacy();
         if DiskError::is_all_not_found(&errs) {
             debug!(
                 event = EVENT_SET_DISK_HEAL,

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

@@ -21,6 +21,7 @@
 //! unchanged; method bodies are moved verbatim and runtime behavior is the same.
 
 use crate::core::pools::DecommissionCapacityAdmission;
+use crate::object_api::write_commit_context::WriteCommitContext;
 
 #[cfg(test)]
 use super::super::GetObjectMetadataCacheKey;
@@ -2446,17 +2447,19 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         opts: &ObjectOptions,
     ) -> Result<ObjectInfo> {
         crate::hp_guard!("SetDisks::complete_multipart_upload");
+        let write_context = WriteCommitContext::from_options(opts, bucket, object, false)?;
         self.invalidate_get_object_metadata_cache(bucket, object).await;
 
         let upload_id_path = Self::get_multipart_upload_dir(bucket, object, upload_id, opts.data_movement);
-        let range_seek_rollout_enabled = crate::object_api::legacy_encrypted_range_seek_enabled() && !opts.no_lock;
+        let range_seek_rollout_enabled =
+            crate::object_api::legacy_encrypted_range_seek_enabled() && write_context.acquires_namespace();
         let mut object_lock_guard = None;
         let mut decommission_object_lock_guard = None;
         let mut decommission_target_lock_covered = false;
         let mut decommission_capacity_guard = None;
 
         if opts.http_preconditions.is_some() {
-            if !opts.no_lock {
+            if write_context.acquires_namespace() {
                 object_lock_guard = Some(
                     self.acquire_write_lock_diag("complete_multipart_upload_precondition", bucket, object)
                         .await?,
@@ -2479,7 +2482,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                     self.pool_index,
                     bucket,
                     object,
-                    opts.no_lock || object_lock_guard.is_some(),
+                    !write_context.acquires_namespace() || object_lock_guard.is_some(),
                     DecommissionCapacityAdmission::ExistingMultipart,
                 )
                 .await?;
@@ -2487,7 +2490,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             decommission_target_lock_covered = target_lock_covered;
             decommission_capacity_guard = capacity_guard;
         }
-        if !opts.no_lock && object_lock_guard.is_none() && !decommission_target_lock_covered {
+        if write_context.acquires_namespace() && object_lock_guard.is_none() && !decommission_target_lock_covered {
             object_lock_guard = Some(
                 self.acquire_write_lock_diag("complete_multipart_upload_commit", bucket, object)
                     .await?,
@@ -3354,8 +3357,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         // usable as a CopyObject source. Do not return on rename quorum while
         // a tail owner may still hold the object guard and finish shard moves.
         let commit_allows_early_ack = false;
-        let detach_commit_owner = commit_allows_early_ack || upload_guard.is_some() || quota_mutation_fence;
+        let detach_commit_owner =
+            write_context.has_borrowed_owner() || commit_allows_early_ack || upload_guard.is_some() || quota_mutation_fence;
         let commit = async move {
+            let write_context = write_context;
             let mut _object_lock_guard = commit_object_lock_guard;
             let mut _decommission_object_lock_guard = commit_decommission_object_lock_guard;
             let mut _upload_guard = upload_guard;
@@ -3377,6 +3382,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::BeforeQuotaRename).await;
                 if quota_reservation.is_lock_lost()
                     || !quota_reservation.capability_proof_matches()
+                    || write_context.is_lock_lost()
                     || _object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
                     || _decommission_object_lock_guard
                         .as_ref()
@@ -3431,6 +3437,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 }
                 if quota_reservation.is_lock_lost()
                     || !quota_reservation.capability_proof_matches()
+                    || write_context.is_lock_lost()
                     || _object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
                     || _decommission_object_lock_guard
                         .as_ref()
@@ -3761,6 +3768,7 @@ mod tests {
     use crate::layout::endpoints::SetupType;
     use crate::services::notification_sys::install_remote_version_state_fleet_proof_for_test;
     use crate::set_disk::core::io_primitives::{ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, rename_fanout_barrier};
+    use crate::set_disk::get_lock_acquire_timeout;
     // No-locker helpers resolve to the isolated-context variants (see
     // `hermetic_set_disks_isolated`); the guard-based tests build through
     // `hermetic_set_disks_with_lockers`, which stays on the bootstrap context
@@ -7122,6 +7130,54 @@ mod tests {
         .await;
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    #[serial]
+    async fn borrowed_complete_keeps_namespace_owner_after_caller_cancel() {
+        let (_dirs, disks, set) = hermetic_set_disks(4).await;
+        let bucket = "multipart-borrowed-cancel";
+        let object = "object";
+        make_bucket_on_all(&disks, bucket).await;
+        let (upload_id, parts) =
+            stage_upload_with_create_opts(&set, bucket, object, &[0x53; 4096], &ObjectOptions::default()).await;
+        let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::AfterRename);
+        let writer = Arc::clone(&set);
+        let complete = tokio::spawn(async move {
+            let outer_guard = crate::object_api::WriteCommitGuard::acquire(
+                &writer.new_ns_lock(bucket, object).await.expect("namespace wrapper"),
+                get_lock_acquire_timeout(),
+            )
+            .await
+            .expect("outer MPU namespace guard");
+            let mut opts = ObjectOptions::default();
+            opts.add_write_commit_guard(&outer_guard);
+            writer
+                .complete_multipart_upload(bucket, object, &upload_id, parts, &opts)
+                .await
+        });
+        tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+            .await
+            .expect("borrowed completion reaches committed cleanup");
+        complete.abort();
+        assert!(complete.await.expect_err("completion caller canceled").is_cancelled());
+        let mut probe = Box::pin(set.acquire_write_lock_diag("borrowed_complete_probe", bucket, object));
+        assert!(
+            futures::poll!(probe.as_mut()).is_pending(),
+            "borrowed completion retains the real outer namespace owner"
+        );
+        barrier.release();
+        drop(
+            tokio::time::timeout(Duration::from_secs(30), probe)
+                .await
+                .expect("completion owner finishes cleanup")
+                .expect("next writer obtains namespace"),
+        );
+        let info = set
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("borrowed MPU result remains readable");
+        assert_eq!(info.size, 4096);
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     #[serial]
     async fn cancelled_complete_keeps_upload_lock_through_tail_cleanup() {

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

@@ -281,7 +281,7 @@ use super::super::GetObjectMetadataCacheEntry;
 use super::super::GetObjectMetadataCacheKey;
 #[cfg(test)]
 use super::super::capacity_scope_from_disks;
-#[cfg(all(test, feature = "test-util"))]
+#[cfg(test)]
 use super::super::get_lock_acquire_timeout;
 use crate::bucket::lifecycle::{
     tier_delete_journal::{TierDeleteDispatchAuthorization, record_tier_delete_journal_backend_identity},
@@ -320,10 +320,13 @@ use crate::disk::format::FormatV3;
 use crate::disk::new_disk;
 use crate::disk::{DataDirDeleteStatus, OldCurrentSize};
 use crate::error::is_err_invalid_upload_id;
+#[cfg(test)]
+use crate::object_api::WriteCompletion;
+use crate::object_api::write_commit_context::WriteCommitContext;
 use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppressed};
 use crate::object_api::{
     NamespaceLockFence, ReplicationStatusWritebackCondition, ReplicationStatusWritebackMode,
-    SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, WriteCompletion,
+    SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY,
 };
 use crate::services::notification_sys::RemoteVersionStateFleetProofToken;
 use crate::services::tier::tier::{TierConfigMgr, TierDestinationId, TierOperationLease, tier_destination_id_from_metadata};
@@ -3499,6 +3502,7 @@ impl SetDisks {
         opts: &ObjectOptions,
         mut publication_fence: Option<RemoteTuplePublicationFence>,
     ) -> Result<(ObjectInfo, Option<OldCurrentSize>)> {
+        let write_context = WriteCommitContext::from_options(opts, bucket, object, publication_fence.is_some())?;
         let protect_write = opts.shard_integrity_write_enabled();
         let source_bucket_incarnation_id = match opts.expected_bucket_incarnation_id {
             Some(incarnation_id) => Some(incarnation_id),
@@ -4026,7 +4030,7 @@ impl SetDisks {
                         self.pool_index,
                         bucket,
                         object,
-                        opts.no_lock || object_lock_guard.is_some(),
+                        !write_context.acquires_namespace() || object_lock_guard.is_some(),
                         DecommissionCapacityAdmission::Mutation,
                     )
                     .await?;
@@ -4034,7 +4038,8 @@ impl SetDisks {
                 decommission_target_lock_covered = target_lock_covered;
                 decommission_capacity_guard = capacity_guard;
             }
-            if publication_fence.is_some() || (!opts.no_lock && object_lock_guard.is_none() && !decommission_target_lock_covered)
+            if publication_fence.is_some()
+                || (write_context.acquires_namespace() && object_lock_guard.is_none() && !decommission_target_lock_covered)
             {
                 #[cfg(any(test, feature = "test-util"))]
                 pause_put_object_commit(bucket, object, PutObjectCommitPause::BeforeNamespace).await;
@@ -4382,10 +4387,11 @@ 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_owns_namespace_guard = commit_object_lock_guard.is_some()
+            let commit_owns_namespace_guard = write_context.has_borrowed_owner()
+                || 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
+            let commit_allows_early_ack = write_context.allows_early_ack(opts.write_completion)
                 && !(opts.data_movement && opts.has_decommission_capacity_reservation())
                 && commit_owns_namespace_guard
                 && commit_scanner_publication_scope.is_none();
@@ -4412,6 +4418,7 @@ impl SetDisks {
             tmp_cleanup_owned = true;
 
             let commit = move |cancellation: Option<CancellationToken>| async move {
+                let write_context = write_context;
                 let mut _object_lock_guard = commit_object_lock_guard;
                 let mut _decommission_object_lock_guard = commit_decommission_object_lock_guard;
                 let mut _publication_guard = commit_publication_guard;
@@ -4427,6 +4434,7 @@ impl SetDisks {
                     pause_put_object_commit(&commit_bucket, &commit_object, PutObjectCommitPause::BeforeQuotaRename).await;
                     if quota_reservation.is_lock_lost()
                         || !quota_reservation.capability_proof_matches()
+                        || write_context.is_lock_lost()
                         || _object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
                         || _decommission_object_lock_guard
                             .as_ref()
@@ -4486,6 +4494,7 @@ impl SetDisks {
                     }
                     if quota_reservation.is_lock_lost()
                         || !quota_reservation.capability_proof_matches()
+                        || write_context.is_lock_lost()
                         || _object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
                         || _decommission_object_lock_guard
                             .as_ref()
@@ -4671,6 +4680,7 @@ impl SetDisks {
                             guard_release_rx,
                             (
                                 object_lock_guard,
+                                write_context,
                                 publication_guard,
                                 bucket_lifecycle_guard,
                                 decommission_object_lock_guard,
@@ -4691,6 +4701,7 @@ impl SetDisks {
                             },
                             move |(
                                 object_lock_guard,
+                                write_context,
                                 publication_guard,
                                 bucket_lifecycle_guard,
                                 decommission_object_lock_guard,
@@ -4698,6 +4709,7 @@ impl SetDisks {
                             ),
                                   targets| async move {
                                 drop(object_lock_guard);
+                                drop(write_context);
                                 drop(publication_guard);
                                 drop(bucket_lifecycle_guard);
                                 cleanup_set
@@ -19817,6 +19829,16 @@ mod put_object_tmp_cleanup_tests {
             .await
             .expect("committed object must remain readable despite the failed tail");
         assert_eq!(info.size, TEST_OBJECT_SIZE as i64);
+        let mut read = set
+            .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("committed body must remain readable despite the failed tail");
+        let mut body = Vec::new();
+        read.stream
+            .read_to_end(&mut body)
+            .await
+            .expect("committed body must stream completely after the failed tail");
+        assert_eq!(body, vec![b'1'; TEST_OBJECT_SIZE]);
     }
 
     #[tokio::test]
@@ -20002,6 +20024,93 @@ mod put_object_tmp_cleanup_tests {
         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 borrowed_write_context_rejects_wrong_namespace_and_cache_flag() {
+        let (_dirs, _disks, set) = hermetic_set_disks(4).await;
+        let bucket = "borrowed-write-target";
+        let object = "owned-object";
+        let guard = crate::object_api::WriteCommitGuard::acquire(
+            &set.new_ns_lock(bucket, object).await.expect("namespace wrapper"),
+            get_lock_acquire_timeout(),
+        )
+        .await
+        .expect("outer write lock");
+        let mut opts = ObjectOptions::default();
+        opts.add_write_commit_guard(&guard);
+        assert!(matches!(
+            WriteCommitContext::from_options(&opts, bucket, "other-object", false),
+            Err(Error::InvalidArgument(..))
+        ));
+        assert!(matches!(
+            WriteCommitContext::from_options(&opts, bucket, "other-object", true),
+            Err(Error::InvalidArgument(..))
+        ));
+        let fake = ObjectOptions {
+            no_lock: true,
+            metadata_cache_safe: true,
+            ..Default::default()
+        };
+        assert!(matches!(
+            WriteCommitContext::from_options(&fake, bucket, object, false),
+            Err(Error::InvalidArgument(..))
+        ));
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn borrowed_tail_drained_put_keeps_owner_after_caller_cancel() {
+        let (_dirs, disks, set) = hermetic_set_disks(4).await;
+        let bucket = "borrowed-put-cancel";
+        let object = "borrowed-put-cancel-object";
+        for disk in &disks {
+            disk.make_volume(bucket).await.expect("bucket volume");
+        }
+        let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+        let tasks = rename_fanout_barrier::observe_tasks(object);
+        let writer = Arc::clone(&set);
+        let put = tokio::spawn(async move {
+            let guard = crate::object_api::WriteCommitGuard::acquire(
+                &writer.new_ns_lock(bucket, object).await.expect("namespace wrapper"),
+                get_lock_acquire_timeout(),
+            )
+            .await
+            .expect("outer write guard");
+            let mut opts = ObjectOptions {
+                write_completion: WriteCompletion::TailDrained,
+                ..Default::default()
+            };
+            opts.add_write_commit_guard(&guard);
+            let mut reader = PutObjReader::from_vec(vec![b'b'; TEST_OBJECT_SIZE]);
+            writer.put_object(bucket, object, &mut reader, &opts).await
+        });
+        tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+            .await
+            .expect("borrowed commit reaches rename phase");
+        wait_for_paused_tail_metadata_quorum(&disks, bucket, object).await;
+        assert!(!put.is_finished(), "full-tail borrowed write remains pending");
+        put.abort();
+        assert!(put.await.expect_err("caller was canceled").is_cancelled());
+        let mut probe = Box::pin(set.acquire_write_lock_diag("borrowed_cancel_probe", bucket, object));
+        assert!(
+            futures::poll!(probe.as_mut()).is_pending(),
+            "commit must retain the real borrowed namespace owner"
+        );
+        barrier.release();
+        drop(
+            tokio::time::timeout(Duration::from_secs(30), probe)
+                .await
+                .expect("borrowed owner drains")
+                .expect("next writer acquires"),
+        );
+        assert_eq!(tasks.running(), 0, "all borrowed rename tasks are reaped before releasing the owner");
+        let info = set
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("committed borrowed write remains readable");
+        assert_eq!(info.size, TEST_OBJECT_SIZE as i64);
+    }
+
     #[tokio::test]
     #[serial_test::serial(capacity_dirty_scope)]
     async fn no_lock_put_waits_for_rename_tail_under_outer_guard() {
@@ -20013,27 +20122,23 @@ mod put_object_tmp_cleanup_tests {
                 disk.make_volume(bucket).await.expect("bucket volume should be created");
             }
 
-            let outer_guard = set_disks
-                .acquire_write_lock_diag("outer_no_lock_put", bucket, object)
-                .await
-                .expect("the outer caller should hold the namespace guard");
+            let outer_guard = crate::object_api::WriteCommitGuard::acquire(
+                &set_disks.new_ns_lock(bucket, object).await.expect("outer namespace wrapper"),
+                get_lock_acquire_timeout(),
+            )
+            .await
+            .expect("the outer caller should hold the namespace guard");
             let rename_tasks = rename_fanout_barrier::observe_tasks(object);
             let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+            let mut opts = ObjectOptions {
+                write_completion: WriteCompletion::TailDrained,
+                ..Default::default()
+            };
+            opts.add_write_commit_guard(&outer_guard);
             let put_store = Arc::clone(&set_disks);
             let put = tokio::spawn(async move {
                 let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
-                put_store
-                    .put_object(
-                        bucket,
-                        object,
-                        &mut reader,
-                        &ObjectOptions {
-                            no_lock: true,
-                            write_completion: WriteCompletion::TailDrained,
-                            ..Default::default()
-                        },
-                    )
-                    .await
+                put_store.put_object(bucket, object, &mut reader, &opts).await
             });
 
             tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())

crates/ecstore/src/set_disk/read.rs

@@ -113,7 +113,28 @@ use super::GetCodecStreamingObjectClass;
 use super::GetCodecStreamingRollout;
 #[cfg(test)]
 use super::classify_get_codec_streaming_object_class;
-use super::core::io_primitives::*;
+use super::core::io_primitives::{
+    BitrotReaderSetup, BitrotReaderSetupAttribution, BitrotReaderSetupMode, DeferredReaderReopener, EVENT_SET_DISK_READ,
+    GetCodecStreamingReaderBuildOutcome, MetadataCacheLookup, ObjectBitrotReader, ReadRepairAdmissionSubmitter,
+    ReadRepairHealSubmission, SLOW_OBJECT_READ_LOG_THRESHOLD, codec_streaming_reader_setup_fallback_reason,
+    create_bitrot_readers_until_quorum_with_preference, create_data_block_bitrot_readers, resolved_read_repair_version_id,
+    send_read_repair_heal_request, shard_read_costs_for_disks, submit_read_repair_heal, submit_read_repair_heal_with_submitter,
+};
+#[cfg(test)]
+use super::core::io_primitives::{
+    ENV_RUSTFS_GET_CODEC_STREAMING_DATA_BLOCKS_FIRST_READER_SETUP, ENV_RUSTFS_GET_DATA_BLOCKS_FIRST_READER_SETUP,
+    MultipartCodecStreamingReader, ReadRepairAdmissionFuture, ReadRepairAdmissionOutcome, collect_read_multiple_results,
+    collect_read_parts_results, create_bitrot_readers_until_quorum, create_bitrot_readers_until_quorum_all_shards,
+    release_read_repair_heal_reservation, reserve_read_repair_heal, resolve_read_part_from_responses, shard_read_cost_for_disk,
+    shard_read_cost_for_endpoint,
+};
+#[cfg(test)]
+use super::core::metadata_quorum::{MetadataEarlyStopDecision, MetadataQuorumAccumulator};
+#[cfg(test)]
+use super::core::metadata_read::{
+    MetadataFanoutDiagnostics, MetadataFanoutObservation, metadata_early_stop_permitted, should_allow_metadata_early_stop,
+};
+use super::core::metadata_read::{late_materialization_candidate_is_safe, non_inline_data_read_early_stop_allowed};
 #[cfg(test)]
 use super::get_codec_streaming_config_cached_core;
 #[cfg(test)]
@@ -607,11 +628,11 @@ impl SetDisks {
         };
 
         // Early-stop for safe metadata reads is handled inside
-        // read_all_fileinfo_observed (see read_all_fileinfo_early_stop in
-        // core/io_primitives.rs); unsafe requests and callers that opt out
+        // read_metadata_observed (see read_all_fileinfo_early_stop in
+        // core/metadata_read.rs); unsafe requests and callers that opt out
         // (allow_early_stop=false) fall back to full-wait.
-        let (mut parts_metadata, errs, metadata_fanout_diagnostics) = if allow_read_version_coalescing {
-            Self::read_all_fileinfo_observed_for_get_object(
+        let metadata_read = if allow_read_version_coalescing {
+            Self::read_metadata_for_get_object(
                 &disks,
                 "",
                 bucket,
@@ -624,7 +645,7 @@ impl SetDisks {
             )
             .await?
         } else {
-            Self::read_all_fileinfo_observed(
+            Self::read_metadata_observed(
                 &disks,
                 "",
                 bucket,
@@ -638,13 +659,14 @@ impl SetDisks {
             )
             .await?
         };
+        let metadata_fanout_complete = metadata_read.is_complete();
+        let (mut parts_metadata, errs, metadata_fanout_diagnostics) = metadata_read.into_legacy();
         let metadata_metrics_path = if crate::bucket::utils::is_meta_bucketname(bucket) {
             GET_OBJECT_PATH_INTERNAL_META
         } else {
             GET_OBJECT_PATH_LEGACY_DUPLEX
         };
         metadata_fanout_diagnostics.record(metadata_metrics_path);
-        let metadata_fanout_complete = metadata_fanout_diagnostics.total_responses() >= disks.len();
         // warn!("get_object_fileinfo parts_metadata {:?}", &parts_metadata);
         // warn!("get_object_fileinfo {}/{} errs {:?}", bucket, object, &errs);
 
@@ -2117,7 +2139,7 @@ impl SetDisks {
         expected: &LateMetadataIdentity,
         metrics_path: &'static str,
     ) -> Result<(FileInfo, Vec<FileInfo>, Vec<Option<DiskStore>>)> {
-        let (mut parts_metadata, errs, diagnostics) = SetDisks::read_all_fileinfo_observed(
+        let metadata_read = SetDisks::read_metadata_observed(
             fallback_disks,
             "",
             bucket,
@@ -2130,6 +2152,7 @@ impl SetDisks {
             expected.parity_blocks,
         )
         .await?;
+        let (mut parts_metadata, errs, diagnostics) = metadata_read.into_legacy();
         diagnostics.record(metrics_path);
 
         let (read_quorum, write_quorum) = SetDisks::object_quorum_from_meta(&parts_metadata, &errs, expected.parity_blocks)

crates/ecstore/src/store/list_objects.rs

@@ -354,6 +354,15 @@ fn ensure_producer_limit_state(options: &mut ListPathOptions) -> Arc<AtomicBool>
     state
 }
 
+fn list_path_folds_common_prefixes(options: &ListPathOptions) -> bool {
+    !options.versioned
+        && (options.include_directories
+            || options
+                .separator
+                .as_deref()
+                .is_some_and(|separator| !separator.is_empty() && separator != SLASH_SEPARATOR))
+}
+
 async fn can_skip_hidden_prefix_check(options: &ListPathOptions) -> bool {
     if options.recursive || options.incl_deleted || options.versioned {
         return false;
@@ -2440,6 +2449,12 @@ where
             let suffix = key.trim_start_matches(prefix);
             if let Some((common_prefix, _)) = suffix.split_once(separator) {
                 let common_prefix = format!("{prefix}{common_prefix}{separator}");
+                if marker.is_some_and(|marker| common_prefix.as_str() <= marker) {
+                    if collect_stats {
+                        stats.skipped_keys += 1;
+                    }
+                    continue;
+                }
                 if prefix_set.insert(common_prefix.clone()) {
                     if collect_stats {
                         stats.common_prefixes += 1;
@@ -2542,6 +2557,10 @@ fn list_objects_from_metadata_snapshot_candidates(
             let suffix = object.name.trim_start_matches(prefix);
             if let Some((common_prefix, _)) = suffix.split_once(separator) {
                 let common_prefix = format!("{prefix}{common_prefix}{separator}");
+                if marker.is_some_and(|marker| common_prefix.as_str() <= marker) {
+                    stats.skipped_keys += 1;
+                    continue;
+                }
                 if prefix_set.insert(common_prefix.clone()) {
                     stats.common_prefixes += 1;
                     visible_entries.push(VerifiedIndexVisibleEntry::Prefix(common_prefix));
@@ -4962,6 +4981,8 @@ async fn gather_results(
     let gather_started = list_metrics_enabled.then(std::time::Instant::now);
     let mut scanned_entries = 0usize;
     let mut candidate_entries = 0usize;
+    let mut previous_prefix = String::new();
+    let folds_prefixes = list_path_folds_common_prefixes(&opts);
 
     while let Some(mut entry) = recv.recv().await {
         scanned_entries += 1;
@@ -5001,6 +5022,34 @@ async fn gather_results(
             continue;
         }
 
+        // Version LIST projections omit cleanup-only free versions. They must
+        // not consume the lookahead budget before a later visible version.
+        // Cleanup walks consume the raw set stream without this collector.
+        if opts.versioned
+            && is_object
+            && entry
+                .xl_meta()
+                .is_ok_and(|meta| !meta.versions.is_empty() && meta.versions.iter().all(|version| version.header.free_version()))
+        {
+            continue;
+        }
+
+        if folds_prefixes
+            && let Some(separator) = opts.separator.as_deref().filter(|separator| !separator.is_empty())
+            && let Some(offset) = entry.name[opts.prefix.len()..].find(separator)
+        {
+            let common_prefix = &entry.name[..opts.prefix.len() + offset + separator.len()];
+            if common_prefix == previous_prefix
+                || opts.marker.as_deref().is_some_and(|marker| {
+                    (!opts.include_marker && common_prefix <= marker) || (opts.include_marker && common_prefix < marker)
+                })
+            {
+                continue;
+            }
+            previous_prefix.clear();
+            previous_prefix.push_str(common_prefix);
+        }
+
         // TODO(backlog): integrate lifecycle evaluation during object listing
 
         entries.push(Some(entry));
@@ -7023,7 +7072,9 @@ impl SetDisks {
         );
 
         let limit = {
-            if opts.limit > 0 && opts.stop_disk_at_limit {
+            // A raw scan budget can be consumed entirely by keys folding into
+            // one common prefix. Let the logical collector stop these walks.
+            if opts.limit > 0 && opts.stop_disk_at_limit && !list_path_folds_common_prefixes(&opts) {
                 opts.limit + 4 + (opts.limit / 16)
             } else {
                 0
@@ -7248,6 +7299,9 @@ fn calc_common_counter(infos: &[DiskInfo], read_quorum: usize) -> u64 {
 
 // list_path_raw
 
+#[cfg(test)]
+mod differential_tests;
+
 #[cfg(test)]
 mod test {
     use super::{
@@ -7926,6 +7980,117 @@ mod test {
         assert!(cancel.is_cancelled());
     }
 
+    #[tokio::test]
+    async fn list_path_gather_results_counts_common_prefixes_before_page_limit() {
+        for separator in ["/", "-"] {
+            for extra_object in [false, true] {
+                let prefix_name = format!("a{separator}");
+                let first = if separator == "/" {
+                    prefix_name.clone()
+                } else {
+                    format!("{prefix_name}first")
+                };
+                let mut input = vec![test_object_meta_entry(&first)];
+                if separator == "/" {
+                    input.push(test_dir_meta_entry(&prefix_name));
+                } else {
+                    input.push(test_object_meta_entry(&format!("{prefix_name}second")));
+                }
+                input.push(test_object_meta_entry("b"));
+                if extra_object {
+                    input.push(test_object_meta_entry("c"));
+                }
+                let (entry_tx, entry_rx) = mpsc::channel(input.len());
+                let (result_tx, mut result_rx) = mpsc::channel(1);
+                for entry in input {
+                    entry_tx.send(entry).await.expect("delimiter candidates should queue");
+                }
+                drop(entry_tx);
+                let state = gather_results(
+                    CancellationToken::new(),
+                    ListPathOptions {
+                        bucket: "bucket".to_owned(),
+                        separator: Some(separator.to_owned()),
+                        recursive: separator != "/",
+                        include_directories: separator == "/",
+                        limit: 3,
+                        ..Default::default()
+                    },
+                    entry_rx,
+                    result_tx,
+                )
+                .await
+                .expect("delimiter collection should succeed");
+                let result = result_rx.recv().await.expect("delimiter page should arrive");
+                let mut expected = vec![first, "b".to_owned()];
+                if extra_object {
+                    expected.push("c".to_owned());
+                }
+                assert_eq!(
+                    result
+                        .entries
+                        .expect("delimiter entries should exist")
+                        .entries()
+                        .into_iter()
+                        .map(|entry| entry.name.clone())
+                        .collect::<Vec<_>>(),
+                    expected,
+                    "delimiter={separator}, extra={extra_object}"
+                );
+                assert_eq!(result.err.is_none(), extra_object, "n=max must reach EOF; max+1 must retain lookahead");
+                assert_eq!(
+                    state,
+                    if extra_object {
+                        GatherResultsState::LimitReached
+                    } else {
+                        GatherResultsState::InputClosed
+                    }
+                );
+            }
+        }
+        for include_marker in [false, true] {
+            let (entry_tx, entry_rx) = mpsc::channel(2);
+            let (result_tx, mut result_rx) = mpsc::channel(1);
+            for name in ["a/child", "b"] {
+                entry_tx
+                    .send(test_object_meta_entry(name))
+                    .await
+                    .expect("marker candidates should queue");
+            }
+            drop(entry_tx);
+            gather_results(
+                CancellationToken::new(),
+                ListPathOptions {
+                    bucket: "bucket".to_owned(),
+                    separator: Some("/".to_owned()),
+                    recursive: true,
+                    include_directories: true,
+                    marker: Some("a/".to_owned()),
+                    include_marker,
+                    limit: 3,
+                    ..Default::default()
+                },
+                entry_rx,
+                result_tx,
+            )
+            .await
+            .expect("logical marker collection should succeed");
+            let entries = result_rx
+                .recv()
+                .await
+                .expect("logical marker page should arrive")
+                .entries
+                .expect("logical marker entries should exist");
+            let projected = ObjectInfo::from_meta_cache_entries_sorted_infos(&entries, "bucket", "", Some("/".to_owned())).await;
+            let names: Vec<_> = projected.into_iter().map(|object| object.name).collect();
+            assert_eq!(
+                names,
+                if include_marker { vec!["a/", "b"] } else { vec!["b"] },
+                "a folded prefix equal to the marker must follow include_marker={include_marker}"
+            );
+        }
+    }
+
     #[test]
     fn list_versions_pagination_scan_limit_boundaries() {
         for has_version_marker in [false, true] {
@@ -8977,6 +9142,29 @@ mod test {
         assert_eq!(result.info.objects[0].etag.as_deref(), Some("etag-z"));
         assert_eq!(result.info.prefixes, vec!["photos/2026/nested/".to_string()]);
         assert!(!result.info.is_truncated);
+
+        let resumed = list_objects_from_metadata_snapshot_candidates(
+            "bucket",
+            "photos/2026/",
+            Some("photos/2026/nested/"),
+            &Some("/".to_string()),
+            1,
+            &objects,
+        );
+        assert!(
+            resumed.info.prefixes.is_empty(),
+            "metadata-fast cursors must not replay an emitted prefix"
+        );
+        assert_eq!(
+            resumed
+                .info
+                .objects
+                .iter()
+                .map(|object| object.name.as_str())
+                .collect::<Vec<_>>(),
+            ["photos/2026/z.jpg"]
+        );
+        assert!(!resumed.info.is_truncated);
     }
 
     #[test]
@@ -10089,6 +10277,23 @@ mod test {
         assert_eq!(result.prefixes, vec!["photos/2026/archive/".to_string()]);
         assert!(result.is_truncated);
         assert_eq!(result.next_marker.as_deref(), Some("photos/2026/archive/"));
+
+        let resumed = list_objects_from_verified_index_candidates(
+            "photos/2026/",
+            result.next_marker.as_deref(),
+            &Some("/".to_string()),
+            1,
+            &candidates,
+            |key| async move { Ok(Some(test_live_object_info(&key, "live-etag"))) },
+        )
+        .await
+        .expect("verified prefix continuation should succeed");
+        assert!(resumed.prefixes.is_empty(), "verified cursors must not replay an emitted prefix");
+        assert_eq!(
+            resumed.objects.iter().map(|object| object.name.as_str()).collect::<Vec<_>>(),
+            ["photos/2026/d.jpg"]
+        );
+        assert!(!resumed.is_truncated);
     }
 
     #[tokio::test]

crates/ecstore/src/store/object.rs

@@ -1038,8 +1038,23 @@ impl fmt::Display for ObjectLockDiagMode {
     }
 }
 
+enum ObjectLockOwner {
+    Read(rustfs_lock::NamespaceLockGuard),
+    Write(Arc<rustfs_lock::NamespaceLockGuard>),
+}
+
+impl ObjectLockOwner {
+    fn as_ref(&self) -> &rustfs_lock::NamespaceLockGuard {
+        match self {
+            Self::Read(guard) => guard,
+            Self::Write(guard) => guard,
+        }
+    }
+}
+
 pub(crate) struct ObjectLockDiagGuard {
-    guard: rustfs_lock::NamespaceLockGuard,
+    guard: ObjectLockOwner,
+    write_key: Option<(String, String)>,
     #[cfg(test)]
     test_namespace_lock_fence: Option<NamespaceLockFence>,
     enabled: bool,
@@ -1063,7 +1078,11 @@ impl ObjectLockDiagGuard {
         mode: ObjectLockDiagMode,
     ) -> Self {
         Self {
-            guard,
+            guard: match mode {
+                ObjectLockDiagMode::Read => ObjectLockOwner::Read(guard),
+                ObjectLockDiagMode::Write => ObjectLockOwner::Write(Arc::new(guard)),
+            },
+            write_key: None,
             #[cfg(test)]
             test_namespace_lock_fence: None,
             enabled,
@@ -1082,15 +1101,20 @@ impl ObjectLockDiagGuard {
         self
     }
 
+    fn with_write_key(mut self, bucket: &str, object: &str) -> Self {
+        self.write_key = Some((bucket.to_owned(), object.to_owned()));
+        self
+    }
+
     pub(crate) fn lock_lost_signal(&self) -> Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>> {
-        match &self.guard {
+        match self.guard.as_ref() {
             rustfs_lock::NamespaceLockGuard::Standard(guard) => Some(guard.lock_lost()),
             rustfs_lock::NamespaceLockGuard::Fast(_) => None,
         }
     }
 
     pub(crate) fn is_lock_lost(&self) -> bool {
-        self.guard.is_lock_lost() || {
+        self.guard.as_ref().is_lock_lost() || {
             #[cfg(test)]
             {
                 self.test_namespace_lock_fence
@@ -1105,6 +1129,13 @@ impl ObjectLockDiagGuard {
     }
 
     pub(crate) fn add_namespace_lock_fence(&self, opts: &mut ObjectOptions) {
+        if let Some((bucket, object)) = self.write_key.as_ref()
+            && let ObjectLockOwner::Write(guard) = &self.guard
+        {
+            let no_lock = opts.no_lock;
+            opts.add_owned_write_lock(Arc::clone(guard), bucket, object);
+            opts.no_lock = no_lock;
+        }
         opts.ensure_namespace_lock_fence();
         if let Some(signal) = self.lock_lost_signal() {
             opts.add_namespace_lock_lost_signal(signal);
@@ -1301,7 +1332,7 @@ impl RestoreAcceptGuard {
     /// refresh lost quorum). Callers must check this before committing the
     /// restore-status write the guard exists to serialize.
     pub fn is_lock_lost(&self) -> bool {
-        self.0.guard.is_lock_lost()
+        self.0.guard.as_ref().is_lock_lost()
     }
 
     pub fn add_namespace_lock_fence(&self, opts: &mut ObjectOptions) {
@@ -1322,7 +1353,7 @@ impl RestoreWorkerGuard {
 
 impl Drop for ObjectLockDiagGuard {
     fn drop(&mut self) {
-        if !self.enabled || self.guard.is_released() {
+        if !self.enabled || self.guard.as_ref().is_released() {
             return;
         }
 
@@ -3029,7 +3060,8 @@ impl ECStore {
             owner,
             ObjectLockDiagMode::Write,
         )
-        .with_attempt(attempt))
+        .with_attempt(attempt)
+        .with_write_key(bucket, object))
     }
 
     async fn acquire_object_write_lock_if_needed(
@@ -3044,11 +3076,9 @@ impl ECStore {
         }
 
         let guard = self.acquire_object_write_lock(op, bucket, object).await?;
-        if let Some(signal) = guard.lock_lost_signal() {
-            opts.add_namespace_lock_lost_signal(signal);
-        }
-        opts.ensure_namespace_lock_fence();
+        guard.add_namespace_lock_fence(opts);
         opts.no_lock = true;
+        opts.write_completion = crate::object_api::WriteCompletion::TailDrained;
 
         Ok(Some(guard))
     }
@@ -3187,7 +3217,10 @@ impl ECStore {
         let store = Arc::clone(self);
         let write = async move {
             let object = "buckets/.scanner-pause-backlog.json";
-            let mut opts = ObjectOptions::default();
+            let mut opts = ObjectOptions {
+                write_completion: crate::object_api::WriteCompletion::TailDrained,
+                ..Default::default()
+            };
             // Match migration: fixed object namespace -> durable pool metadata ->
             // actual replica namespace. The replica need not be the hash-routed set.
             let object_guard = if store.single_pool() {
@@ -3727,15 +3760,18 @@ impl ECStore {
                     acquire_start.elapsed(),
                     diag_enabled,
                 );
-                guards.push(ObjectLockDiagGuard::new(
-                    guard,
-                    diag_enabled,
-                    op,
-                    diag_enabled.then(|| bucket.to_string()),
-                    diag_enabled.then(|| object.to_string()),
-                    owner,
-                    ObjectLockDiagMode::Write,
-                ));
+                guards.push(
+                    ObjectLockDiagGuard::new(
+                        guard,
+                        diag_enabled,
+                        op,
+                        diag_enabled.then(|| bucket.to_string()),
+                        diag_enabled.then(|| object.to_string()),
+                        owner,
+                        ObjectLockDiagMode::Write,
+                    )
+                    .with_write_key(bucket, object),
+                );
                 locked_sets.push(Arc::clone(set));
             }
         }
@@ -3826,7 +3862,8 @@ impl ECStore {
                 diag_enabled.then(|| object.to_string()),
                 owner,
                 ObjectLockDiagMode::Write,
-            );
+            )
+            .with_write_key(bucket, object);
             #[cfg(test)]
             let guard = {
                 let mut guard = guard;
@@ -4571,11 +4608,13 @@ impl ECStore {
                     shard_integrity_write_mode: dst_opts.shard_integrity_write_mode,
                     version_id: dst_opts.version_id.clone(),
                     no_lock: dst_opts.no_lock,
+                    write_completion: dst_opts.write_completion,
                     mod_time: dst_opts.mod_time,
                     http_preconditions: dst_opts.http_preconditions.clone(),
                     expected_current_version_id: dst_opts.expected_current_version_id.clone(),
                     expected_bucket_incarnation_id: dst_opts.expected_bucket_incarnation_id,
                     namespace_lock_fence: dst_opts.namespace_lock_fence.clone(),
+                    write_lock_context: dst_opts.write_lock_context.clone(),
                     bucket_lifecycle_lock_fence: dst_opts.bucket_lifecycle_lock_fence.clone(),
                     object_lock_config_snapshot: dst_opts.object_lock_config_snapshot.clone(),
                     quota_admission: dst_opts.quota_admission,
@@ -4614,11 +4653,13 @@ impl ECStore {
                         shard_integrity_write_mode: dst_opts.shard_integrity_write_mode,
                         version_id: dst_opts.version_id.clone(),
                         no_lock: dst_opts.no_lock,
+                        write_completion: dst_opts.write_completion,
                         mod_time: dst_opts.mod_time,
                         http_preconditions: dst_opts.http_preconditions.clone(),
                         expected_current_version_id: dst_opts.expected_current_version_id.clone(),
                         expected_bucket_incarnation_id: dst_opts.expected_bucket_incarnation_id,
                         namespace_lock_fence: dst_opts.namespace_lock_fence.clone(),
+                        write_lock_context: dst_opts.write_lock_context.clone(),
                         bucket_lifecycle_lock_fence: dst_opts.bucket_lifecycle_lock_fence.clone(),
                         object_lock_config_snapshot: dst_opts.object_lock_config_snapshot.clone(),
                         quota_admission: dst_opts.quota_admission,
@@ -4667,11 +4708,13 @@ impl ECStore {
             shard_integrity_write_mode: dst_opts.shard_integrity_write_mode,
             version_id: dst_opts.version_id.clone(),
             no_lock: dst_opts.no_lock,
+            write_completion: dst_opts.write_completion,
             mod_time: dst_opts.mod_time,
             http_preconditions: dst_opts.http_preconditions.clone(),
             expected_current_version_id: dst_opts.expected_current_version_id.clone(),
             expected_bucket_incarnation_id: dst_opts.expected_bucket_incarnation_id,
             namespace_lock_fence: dst_opts.namespace_lock_fence.clone(),
+            write_lock_context: dst_opts.write_lock_context.clone(),
             bucket_lifecycle_lock_fence: dst_opts.bucket_lifecycle_lock_fence.clone(),
             object_lock_config_snapshot: dst_opts.object_lock_config_snapshot.clone(),
             quota_admission: dst_opts.quota_admission,
@@ -8721,6 +8764,108 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn borrowed_copy_tail_drained_waits_for_rename_tail() {
+        use crate::object_api::{WriteCommitGuard, WriteCompletion};
+        use crate::set_disk::{ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, rename_fanout_barrier};
+
+        temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
+            let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
+            let (_dirs, original) = make_local_set_disks_with_ctx(4, 2, Arc::clone(&ctx)).await;
+            let store = Arc::new(new_prepared_reader_test_store_with_ctx(&[original], ctx).await);
+            let bucket = RUSTFS_META_BUCKET;
+            let source = "copy-tail-drained-source";
+            let target = "copy-tail-drained-target";
+            let payload = vec![0x64; 4097];
+            let set = store.pools[0].get_disks_by_key(target);
+            let mut source_info = set
+                .put_object(
+                    bucket,
+                    source,
+                    &mut PutObjReader::from_vec(payload.clone()),
+                    &ObjectOptions {
+                        write_completion: WriteCompletion::TailDrained,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("seed copy source");
+            source_info.put_object_reader = Some(PutObjReader::from_vec(payload.clone()));
+
+            let barrier = rename_fanout_barrier::arm(target, 0, rename_fanout_barrier::PHASE_RENAME);
+            let rename_tasks = rename_fanout_barrier::observe_tasks(target);
+            let copy_store = Arc::clone(&store);
+            let mut copy = tokio::spawn(async move {
+                let lock = copy_store
+                    .new_ns_lock(bucket, target)
+                    .await
+                    .expect("target namespace wrapper");
+                let guard = WriteCommitGuard::acquire(&lock, get_lock_acquire_timeout())
+                    .await
+                    .expect("real Copy target namespace owner");
+                let mut options = ObjectOptions {
+                    write_completion: WriteCompletion::TailDrained,
+                    ..Default::default()
+                };
+                options.add_write_commit_guard(&guard);
+                copy_store
+                    .handle_copy_object(bucket, source, bucket, target, &mut source_info, &ObjectOptions::default(), &options)
+                    .await
+            });
+            let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
+            tokio::time::timeout_at(deadline, barrier.wait_until_paused())
+                .await
+                .expect("copy reaches the paused rename checkpoint");
+            tokio::time::timeout_at(deadline, async {
+                loop {
+                    if let Ok(info) = set
+                        .get_object_info(
+                            bucket,
+                            target,
+                            &ObjectOptions {
+                                no_lock: true,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                        && info.size == 4097
+                        && rename_tasks.running() == 1
+                    {
+                        break;
+                    }
+                    tokio::task::yield_now().await;
+                }
+            })
+            .await
+            .expect("copy commits a metadata quorum while one rename remains paused");
+            assert!(
+                tokio::time::timeout(Duration::from_secs(1), &mut copy).await.is_err(),
+                "Copy TailDrained must remain pending after metadata quorum"
+            );
+            barrier.release();
+            let result = tokio::time::timeout_at(deadline, copy)
+                .await
+                .expect("copy drains its rename tail")
+                .expect("copy task completes")
+                .expect("copy succeeds");
+            assert_eq!(result.size, 4097);
+            assert_eq!(rename_tasks.running(), 0);
+            let mut reader = store
+                .get_object_reader(bucket, target, None, HeaderMap::new(), &ObjectOptions::default())
+                .await
+                .expect("read copied target");
+            let mut actual = Vec::new();
+            reader
+                .stream
+                .read_to_end(&mut actual)
+                .await
+                .expect("read complete copied bytes");
+            assert_eq!(actual, payload);
+        })
+        .await;
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn object_lock_diagnostics_report_historical_get_blocked_by_put_commit() {

crates/ecstore/tests/fixtures/list_namespace_keys.json

@@ -0,0 +1,14 @@
+[
+  "a",
+  "a/",
+  "a/b",
+  "a/b/child",
+  "a0",
+  "b",
+  "b/leaf",
+  "quote\"&<",
+  "space key",
+  "z",
+  "é",
+  "中/文"
+]

crates/filemeta/src/metacache.rs

@@ -618,6 +618,7 @@ impl MetaCacheEntries {
         }
 
         let mut dir_exists = 0;
+        let mut directory = None;
         let mut selected = None;
 
         params.candidates.clear();
@@ -633,7 +634,7 @@ impl MetaCacheEntries {
             }
             if entry.is_dir() {
                 dir_exists += 1;
-                selected = Some(entry.clone());
+                directory = Some(entry.clone());
                 debug!(entry = %entry.name, "metacache resolve observed directory candidate");
                 continue;
             }
@@ -664,6 +665,10 @@ impl MetaCacheEntries {
             }
         }
 
+        // A prefix and an explicit directory-marker object can share a name.
+        // Keep directory evidence separate so its arrival cannot displace an
+        // object that meets the metadata quorum.
+        let directory = directory.filter(|_| dir_exists >= params.dir_quorum);
         let Some(selected) = selected else {
             debug!(
                 bucket = %params.bucket,
@@ -671,19 +676,9 @@ impl MetaCacheEntries {
                 obj_quorum = params.obj_quorum,
                 "metacache resolve could not select any candidate entry"
             );
-            return None;
+            return directory;
         };
 
-        if selected.is_dir() && dir_exists >= params.dir_quorum {
-            debug!(
-                entry = %selected.name,
-                dir_exists,
-                dir_quorum = params.dir_quorum,
-                "metacache resolve selected directory candidate after quorum reconciliation"
-            );
-            return Some(selected);
-        }
-
         // If we would never be able to reach the required object quorum.
         if objs_valid < params.obj_quorum {
             debug!(
@@ -691,7 +686,7 @@ impl MetaCacheEntries {
                 obj_quorum = params.obj_quorum,
                 "metacache resolve did not have enough valid object candidates to satisfy quorum"
             );
-            return None;
+            return directory;
         }
 
         if objs_agree == objs_valid {
@@ -724,7 +719,7 @@ impl MetaCacheEntries {
 
         let Some(cached) = selected.cached else {
             debug!(selected = %selected.name, "metacache resolve could not merge because selected candidate had no cached metadata");
-            return None;
+            return directory;
         };
 
         let versions = if enforce_write_quorum {
@@ -744,7 +739,7 @@ impl MetaCacheEntries {
                 requested_versions = params.requested_versions,
                 "metacache resolve produced no merged versions after reconciliation"
             );
-            return None;
+            return directory;
         }
 
         let merged_cached = FileMeta {
@@ -761,7 +756,7 @@ impl MetaCacheEntries {
                     error = ?e,
                     "metacache resolve failed to marshal merged metadata entry"
                 );
-                return None;
+                return directory;
             }
         };
 
@@ -2534,6 +2529,57 @@ mod tests {
         assert!(resolved.is_none());
     }
 
+    #[test]
+    fn resolve_directory_marker_and_prefix_preserve_quorum_in_both_orders() {
+        let modified = OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
+        let mut object = metacache_entry_with_erasure(modified, "marker-etag", 2, 2);
+        object.name = "folder/".to_string();
+        let directory = metacache_dir_entry("folder/");
+        for enforce_write_quorum in [false, true] {
+            for (object_copies, directory_copies) in [(3, 1), (3, 3), (2, 3), (0, 3), (2, 2)] {
+                for reverse in [false, true] {
+                    let mut entries = vec![Some(object.clone()); object_copies];
+                    entries.extend(vec![Some(directory.clone()); directory_copies]);
+                    if reverse {
+                        entries.reverse();
+                    }
+                    let entries = MetaCacheEntries(entries);
+                    let params = MetadataResolutionParams {
+                        dir_quorum: 3,
+                        obj_quorum: 3,
+                        requested_versions: 1,
+                        bucket: "bucket".to_string(),
+                        strict: true,
+                        ..Default::default()
+                    };
+                    let resolved = if enforce_write_quorum {
+                        entries.resolve_with_write_quorum(params)
+                    } else {
+                        entries.resolve(params)
+                    };
+                    let context = format!(
+                        "objects={object_copies}, directories={directory_copies}, reverse={reverse}, write={enforce_write_quorum}"
+                    );
+                    if object_copies >= 3 {
+                        let resolved = resolved.expect("metadata at quorum must shadow a same-named prefix");
+                        assert!(resolved.is_object(), "{context}");
+                        assert_eq!(
+                            resolved
+                                .to_fileinfo("bucket")
+                                .expect("marker metadata should decode")
+                                .mod_time,
+                            Some(modified)
+                        );
+                    } else if directory_copies >= 3 {
+                        assert!(resolved.expect("prefix at quorum must remain a fallback").is_dir(), "{context}");
+                    } else {
+                        assert!(resolved.is_none(), "{context}");
+                    }
+                }
+            }
+        }
+    }
+
     fn build_hashmap_cache(update_size: usize) -> Arc<Cache<HashMap<usize, usize>>> {
         let generation = Arc::new(AtomicUsize::new(0));
         Arc::new(Cache::new(

crates/lock/src/namespace/mod.rs

@@ -57,6 +57,11 @@ impl NamespaceLockWrapper {
         &self.owner
     }
 
+    /// The exact namespace key used by this acquisition wrapper.
+    pub fn resource(&self) -> &ObjectKey {
+        &self.resource
+    }
+
     /// Logical namespace label for diagnostics; local namespaces may share a manager.
     pub fn namespace(&self) -> &str {
         self.lock.namespace()

docs/architecture/compat-cleanup-register.md

@@ -11,6 +11,8 @@
 
 ## Open Items
 
+- `backlog-2253-metadata-observation-adapter` metadata observations: preserve aligned metadata/error slices for existing layout, shard and exact-version PUT, MPU and delete consumers while read/heal scheduling uses typed disk observations. Pending slots acquire empty placeholders only in this adapter. Remove after those remaining consumers use typed observations; delete the adapter and old facade in a separate cleanup.
+
 - `connect-894` pending Connect heartbeats: replay the exact preceding producer-capability list, with or without jobs, when upgrading to the memory-service capability. Preserve request ID, sequence, and persisted body rather than inserting the new capability into a retry. Remove after upgrades from the pre-service-memory capability set are unsupported.
 
 - `backlog-2539` administrator erasure-set scope: decode historical pending intents whose empty bucket list omitted the all-buckets marker. Normalize that representation to the explicit scope before replay or checkpoint comparison. Remove after all pre-marker administrator ErasureSet intents are retired; deployment verification must confirm that no such pending records remain on coordinator disks.

docs/architecture/ecstore-module-split-plan.md

@@ -83,3 +83,39 @@ A candidate is ready for code movement only when all of these hold:
 - old facade names have compatibility tests or explicit deprecation coverage;
 - focused tests cover the changed owner path before any full gate is attempted;
 - rollback preserves object IO, quorum, lifecycle/replication queues, scanner repair, notification/audit events, and metadata compatibility.
+
+## Metadata read ownership and identity
+
+`set_disk/core/metadata_read.rs` owns disk/RPC scheduling, coalescing, timeouts,
+hedging and cancellation. Its result retains each disk slot as pending, a
+successful `FileInfo`, or a typed `DiskError`. Pending includes both an
+unscheduled disk and a response not yet observed; it contributes no vote and
+does not establish absence. Task IDs retain the disk slot when a task panics.
+The existing full-wait and early-stop gates, coalescer admission, cancellation
+drain and fallback policy remain at this scheduling boundary.
+
+`set_disk/core/metadata_quorum.rs` consumes those observations without IO.
+`SetDisks` remains the sole owner of topology, locks and caches. Read and heal
+callers use the typed result; `MetadataReadResult::into_legacy` is the explicit
+adapter for existing quorum/layout and shard-reader helpers. It moves successful
+metadata without cloning and creates empty placeholders only at that adapter.
+Old full-wait facade callers retain the same aligned metadata/error slices.
+
+These identities have different purposes and must not become one generic hash:
+
+| Decision | Equal fields | Deliberately different or separately checked |
+| --- | --- | --- |
+| Quorum grouping (`file_info_quorum_hash`) | Size, deletion/restore/transition state, timestamps, version and data-directory UUIDs, checksums, mode/writer version, normalized metadata, target delete-marker versions and part identities. Payloads also include data/parity counts and distribution. | Object name and volume come from the request; per-disk erasure index and shard bytes are not an object identity. Canonical delete markers and zero-size objects do not add payload geometry. |
+| Early-stop strict match (`metadata_early_stop_candidate_matches`) | Request name/volume, version/latest/deletion state, transition fields, size/time, mode/writer version, complete metadata/replication state/parts/checksum, version-history fields, data directory, algorithm/block size/data/parity/distribution. | Per-disk erasure index and data bytes differ. A non-inline read reserve additionally proves distinct indexes mapped to their disk slots; inline reads verify shard identity and bitrot/body content before stopping. |
+| Late materialization (`LateMetadataIdentity`) | Request name/volume, algorithm, block size, legacy-checksum mode and quorum hash. | Each late shard must match the original distribution at its disk slot. A changed identity or insufficient matching shards fails read quorum; it cannot mix generations into a body. |
+
+`metadata_observation_all_four_slot_states_and_arrival_orders_preserve_reduction`
+checks all 625 four-disk success/missing/corrupt/offline/pending combinations and
+24 arrival orders with early-stop on and off. The observation regressions also
+cover a newer pending response, same-time different directories, duplicated
+shard indexes, null versions, canonical delete markers and invalid success
+payloads. Existing bounded-fanout, legacy inline bitrot, cache, late-metadata and
+GET/Range regressions continue to exercise the real scheduling and reader paths;
+the pure matrix does not replace those checks. Their call counters bound actual
+disk reads, including canceled tails, without adding a scheduling policy or a
+per-request metadata clone.

docs/architecture/unified-object-generation.md

@@ -82,7 +82,7 @@ A single filesystem rename does not atomically commit a sidecar plus `xl.meta`.
 
 A laggard need not contain the predecessor. Recovery fetches the chosen decision and its verified metadata, reconstructs or validates its shards under [erasure-coding.md](erasure-coding.md), and installs that state. An empty replacement uses a fresh disk incarnation and cannot reuse old preparation receipts. A disk whose durable state claims a later decision than the supplied one rejects the operation; a conflicting same-revision digest is quarantined. Never erase a divergent disk merely because it is in the minority.
 
-`rollback_committed_rename_std` and `rollback_inline_metadata_commit_std` in `crates/ecstore/src/disk/local/commit.rs`, and `restore_metadata_backup` in `crates/ecstore/src/disk/local.rs`, must become decision-aware before strict mode includes them. The permitted rollback is limited to a transaction's unchosen private preparation, or restoration proven by recovery to be necessary before any newer local effect. A chosen operation is repaired forward. Neither a client timeout nor a rename-tail error permits reverting an acknowledged decision.
+`RenameCommitContext::rollback_non_inline_std` and `RenameCommitContext::rollback_inline_std` in `crates/ecstore/src/disk/local/commit.rs`, and `restore_metadata_backup` in `crates/ecstore/src/disk/local.rs`, must become decision-aware before strict mode includes them. The permitted rollback is limited to a transaction's unchosen private preparation, or restoration proven by recovery to be necessary before any newer local effect. A chosen operation is repaired forward. Neither a client timeout nor a rename-tail error permits reverting an acknowledged decision.
 
 `RenameConvergence` remains a post-publication repair signal. `PartialCommit`, `SignatureDivergent`, and `Unknown` do not decide which transaction won. Keep their diagnostics and quorum accounting; resolve authority first. Early ACK may still precede minority-tail completion after the two quorum conditions hold. Tests that inspect all disks must synchronize the tail or assert the permitted minority residue separately.
 

docs/testing/README.md

@@ -21,6 +21,8 @@ Pick the lowest layer that can prove the change; add a higher-layer test only wh
 
 Every script named above is indexed with status and wiring in [`scripts/README.md`](../../scripts/README.md). Fixed GHSA advisories map to named regression tests in [security-regressions.md](security-regressions.md).
 
+The [ECStore core invariant gates](ecstore-core-invariants.md) map required native commit, rollback, lock, read, multipart and metadata regressions to the existing CI run and distinguish them from process and network fault evidence.
+
 The [scanner checkpoint fixture](scanner-checkpoint-fixture.md) diagnoses retained subtree coverage across budget interruption, persistence, reload, and plan invalidation.
 
 The [scanner cache cost profile](scanner-cache-cost.md) separates clone, subtree copy, encoding, and counted save costs without changing production cache behavior.

docs/testing/ecstore-core-invariants.md

@@ -0,0 +1,92 @@
+# ECStore core regression gates
+
+The required native tests and pinned fixture hashes live in
+`.config/ecstore-required-tests.json`. `Test and Lint` checks that each test
+was selected by the existing `ci` run, is not ignored, and has a fresh nonempty
+JUnit report. The check does not start another workspace test run.
+
+## Native invariant coverage
+
+The names below identify tests in `rustfs-ecstore` unless a different crate is
+shown. The manifest contains their complete module paths.
+
+| Boundary | Required test | Oracle |
+|---|---|---|
+| Write quorum | `inline_put_direct_commit_accepts_exact_quorum_and_rejects_quorum_minus_one` | W commits; W-1 fails without exposing a fresh object. |
+| Metadata rollback | `write_unique_file_info_reverts_metadata_when_write_quorum_fails` | A failed write preserves the previous metadata. |
+| Rollback inspection | `rename_rollback_incomplete_preserves_overwrite_data_dirs_and_staging` | Real failed undo records rejected inspection admission and retains incomplete receipts, old shards, backups, and staging across reopen. |
+| Cancelled undo | `rename_rollback_incomplete_cancelled_task_is_not_success` | A cancelled undo remains incomplete. |
+| Fault schedule | `commit_fault_schedule_keeps_multiple_disk_phases_independent` | Two armed phases remain independently paused and are released by their own handles. |
+| Fault checkpoint owner | `commit_fault_schedule_rejects_duplicate_checkpoint_without_disarming_owner` | A duplicate checkpoint cannot replace the current handle. |
+| Completion equivalence | `rename_rollback_incomplete_matches_early_ack_and_full_wait_after_reopen` | Both completion policies retain the same recovery residue. |
+| Physical undo ownership | `rename_rollback_children_keep_namespace_ownership_after_coordinator_panic` | A coordinator panic cannot retire a running physical undo. |
+| Late tail failure | `rename_data_early_ack_post_mutation_tail_error_never_rolls_back_commit` | A late error cannot reverse an acknowledged quorum commit. |
+| Control-plane completion | `tail_drained_put_waits_for_tail_and_allows_immediate_cas` | TailDrained waits for the tail before a same-key CAS. |
+| Borrowed lock | `no_lock_put_waits_for_rename_tail_under_outer_guard` | A borrowed write does not reacquire the same key and waits for its tail. |
+| Borrowed capability | `borrowed_write_context_rejects_wrong_namespace_and_cache_flag` | A different namespace or cache flag cannot manufacture a write owner. |
+| Borrowed cancellation | `borrowed_tail_drained_put_keeps_owner_after_caller_cancel` and `borrowed_complete_keeps_namespace_owner_after_caller_cancel` | The actual namespace owner survives caller cancellation until physical publication drains. |
+| Copy completion | `borrowed_copy_tail_drained_waits_for_rename_tail` | Copy preserves an explicit TailDrained policy after a metadata quorum commits. |
+| Pool metadata owner | `pool_meta_cas_context_retains_actual_owner_for_exact_target` and `pool_meta_cas_rejects_released_owner` | CAS retains its real owner and rejects a released guard. |
+| Catalog publication | `rustfs`: `commit_backend_borrowed_put_preserves_owner_authorization_and_precondition` | An authorized write under the actual outer owner avoids recursive acquisition, preserves IfAbsent, and leaves the original bytes after conflict. |
+| Stale writer | `put_object_no_lock_aborts_after_outer_namespace_lock_loss` | A lost outer fence rejects publication. |
+| Cancelled writer | `tail_drained_put_owned_commit_survives_waiter_cancellation` | Cancelling the waiter does not cancel the commit owner. |
+| GET reconstruction | `blackbox_get_restores_body_after_one_shard_file_is_removed` | The complete returned body equals the original bytes. |
+| Range reconstruction | `blackbox_range_read_restores_exact_slice_with_one_offline_disk` | The requested byte slice is reconstructed exactly. |
+| Transitioned Range | `transitioned_compressed_object_range_get_returns_plaintext_slice` | A transitioned compressed object returns plaintext Range bytes. |
+| MPU cancellation | `cancelled_complete_keeps_upload_lock_through_tail_cleanup` | Upload lock ownership survives caller cancellation and tail cleanup. |
+| MPU abort/complete | `abort_and_complete_linearize_for_plain_sse_and_legacy_layouts` | Abort and complete have one outcome across the supported layouts. |
+| MPU delete quorum | `abort_enforces_delete_write_quorum_boundary` | Abort requires the existing delete write quorum. |
+| MPU physical publication | `complete_multipart_timeout_keeps_namespace_owner_until_physical_publication` | Timeout cannot retire an active physical publication. |
+| Restore failures | `multipart_restore_aborts_every_post_create_failure` | Every injected post-create failure aborts the remote upload. |
+| Partial LIST metadata | `rustfs-filemeta`: `resolve_with_write_quorum_slack_keeps_partial_latest_hidden_during_merge` | A partial newer version remains hidden. |
+| Metadata observations | `metadata_observation_all_four_slot_states_and_arrival_orders_preserve_reduction` and its companion tests | Pending, absent, corrupt, offline and successful disk slots retain the existing quorum decisions across all four-slot arrival orders. |
+| LIST backend parity | `list_objects_shared_corpus_*` | One independent namespace oracle checks actual disk, set, pool, store and index paths, including version visibility, page boundaries, faults and cancellation. |
+| LIST logical budget | `list_path_gather_results_counts_common_prefixes_before_page_limit` | Repeated common prefixes consume one page slot while max+1 retains a continuation. |
+| Directory-marker metadata | `rustfs-filemeta`: `resolve_directory_marker_and_prefix_preserve_quorum_in_both_orders` | Quorum object metadata survives same-named prefix candidates; a fallback directory still requires its own quorum. |
+| MinIO metadata | `rustfs-filemeta`: `parses_real_minio_object_xlmeta` | Independently captured MinIO metadata is readable. |
+| Legacy metadata | `rustfs-filemeta`: `test_issue_2288_legacy_xlmeta_compatibility` | The fixed legacy fixture decodes with its expected object fields. |
+| Legacy null version | `rustfs-filemeta`: `test_into_fileinfo_reads_legacy_nil_uuid_inline_key` | Legacy nil UUID inline data retains null-version semantics. |
+| Corrupt part arrays | `rustfs-filemeta`: `crc_valid_but_part_arrays_corrupt_into_fileinfo_errors_not_panics` | CRC-valid inconsistent arrays return an error rather than panic. |
+
+The same manifest retains the S3, Azure and GCS source-contract tests for ODM.
+The static LIST keys in `crates/ecstore/tests/fixtures/list_namespace_keys.json`
+are also consumed by ODM's `list_through_static_namespace_boundary_matrix`.
+The ECStore query replay uses seed `0xec5707`; the ODM matrix retains seed
+`0xec5706`.
+Fixture provenance is recorded in
+[`crates/ecstore/tests/fixtures/minio/README.md`](../../crates/ecstore/tests/fixtures/minio/README.md).
+These metadata fixtures do not establish compatibility for an external legacy
+erasure-shard corpus.
+
+## Selection and evidence
+
+After the normal `ci` test run, inspect its actual selection:
+
+```bash
+cargo nextest list --profile ci --all --exclude e2e_test --message-format json > core-test-listing.json
+python3 scripts/check_test_wiring.py --check-core core-test-listing.json
+```
+
+`python3 scripts/check_test_wiring.py --self-test` exercises missing, ignored,
+filtered and malformed listing cases, and missing or altered required fixtures.
+Those cases must fail; an optional legacy fixture test that returns early is
+not evidence for a required fixture.
+
+The `test-and-lint` artifact contains the native log, listing, JUnit and
+execution receipt. A native test that reopens a disk in the same process does
+not establish process-crash or power-loss durability.
+
+## Process and network faults
+
+Use the existing `e2e-full`, `e2e-nightly` and `e2e-distributed` lanes for real
+server-process restart, fresh-drive and network-fault scenarios. Their
+membership remains pinned by `.config/e2e-*-selection.txt`; the lane checks
+selection before execution. See [distributed-e2e.md](distributed-e2e.md) for
+the named scenarios, platform requirements and failure artifacts.
+
+Record the source commit, selected test names, actual fault reached, process
+logs, metadata observations and JUnit for each run. A capability early return,
+setup failure, missing report or abstract protocol model cannot establish
+that the fault ran. Native ownership tests do not prove the durable ordered
+authority described in
+[unified-object-generation.md](../architecture/unified-object-generation.md).

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

@@ -58,7 +58,7 @@ Fault injection is explicit and deterministic: local disk mocks for unit tests,
 
 `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.
+The object tests reuse `rename_fanout_barrier::arm(object, disk_slot, phase)` and `observe_tasks(object)`. The fixture uses typed rename, rollback, cleanup, read-version and hedge-timer phases; multiple disk/phase checkpoints can be armed for one object. A duplicate live checkpoint is rejected instead of replacing its owner. `wait_until_paused_before(deadline)` reports the exact unreached checkpoint on timeout. 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 |
 |---|---|

rustfs/src/admin/handlers/site_replication.rs

@@ -7358,32 +7358,36 @@ where
 {
     let store =
         current_object_store_handle().ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()))?;
-    with_config_object_write_lock(store, SITE_REPLICATION_JOIN_ADMISSION_LOCK_PATH.to_string(), move || async move {
-        let fresh = load_site_replication_state().await?;
-        let fresh_local_peer = local_peer_at_endpoint(local_endpoint.clone(), &fresh);
-        if join_request_is_superseded(&fresh, join_req.updated_at) {
-            let peer = fresh
-                .peers
-                .get(&fresh_local_peer.deployment_id)
-                .cloned()
-                .unwrap_or(fresh_local_peer);
-            return Ok(PeerJoinOutcome::Superseded(peer));
-        }
-
-        apply_iam(join_req.clone()).await?;
-
-        let incoming_updated_at = join_req.updated_at;
-        update_site_replication_state_when_changed(move |state| {
-            let local_peer = local_peer_at_endpoint(local_endpoint, state);
-            if join_request_is_superseded(state, incoming_updated_at) {
-                let peer = state.peers.get(&local_peer.deployment_id).cloned().unwrap_or(local_peer);
-                return Ok(StateCommit::Unchanged(PeerJoinOutcome::Superseded(peer)));
+    with_config_object_write_lock(
+        store,
+        SITE_REPLICATION_JOIN_ADMISSION_LOCK_PATH.to_string(),
+        move |_write_guard| async move {
+            let fresh = load_site_replication_state().await?;
+            let fresh_local_peer = local_peer_at_endpoint(local_endpoint.clone(), &fresh);
+            if join_request_is_superseded(&fresh, join_req.updated_at) {
+                let peer = fresh
+                    .peers
+                    .get(&fresh_local_peer.deployment_id)
+                    .cloned()
+                    .unwrap_or(fresh_local_peer);
+                return Ok(PeerJoinOutcome::Superseded(peer));
             }
-            apply_peer_join(state, &local_peer, join_req, defer_sync_state_enable);
-            Ok(StateCommit::Changed(PeerJoinOutcome::Applied(Box::new(state.clone()), local_peer)))
-        })
-        .await
-    })
+
+            apply_iam(join_req.clone()).await?;
+
+            let incoming_updated_at = join_req.updated_at;
+            update_site_replication_state_when_changed(move |state| {
+                let local_peer = local_peer_at_endpoint(local_endpoint, state);
+                if join_request_is_superseded(state, incoming_updated_at) {
+                    let peer = state.peers.get(&local_peer.deployment_id).cloned().unwrap_or(local_peer);
+                    return Ok(StateCommit::Unchanged(PeerJoinOutcome::Superseded(peer)));
+                }
+                apply_peer_join(state, &local_peer, join_req, defer_sync_state_enable);
+                Ok(StateCommit::Changed(PeerJoinOutcome::Applied(Box::new(state.clone()), local_peer)))
+            })
+            .await
+        },
+    )
     .await
     .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("lock site replication join admission failed: {e}")))?
 }

rustfs/src/admin/handlers/table_catalog/mod.rs

@@ -1726,8 +1726,18 @@ where
         object: &str,
         data: Vec<u8>,
         precondition: crate::table_catalog::TableCatalogPutPrecondition,
+        write_guards: Vec<crate::admin::storage_api::object::WriteCommitGuard>,
     ) -> crate::table_catalog::TableCatalogStoreResult<()> {
-        self.put_object(bucket, object, data, precondition).await
+        self.authorize(bucket, object, S3Action::PutObjectAction).await?;
+        self.ensure_observation_allowed(bucket, object)?;
+        let observation = TableCommitObservedObject {
+            identity: TableCommitObjectIdentity::ContentSha256(hex_sha256(&data, str::to_string)),
+            max_size: None,
+        };
+        self.backend
+            .put_object_unlocked(bucket, object, data, precondition, write_guards)
+            .await?;
+        self.record_observation(bucket, object, observation)
     }
 
     async fn delete_object(&self, bucket: &str, object: &str) -> crate::table_catalog::TableCatalogStoreResult<()> {

rustfs/src/admin/handlers/table_catalog/tests.rs

@@ -87,6 +87,80 @@ async fn call_load_table_handler(request: S3Request<Body>) -> S3Result<S3Respons
     RestLoadTableHandler {}.call(request, matched.params).await
 }
 
+#[tokio::test]
+async fn commit_backend_borrowed_put_preserves_owner_authorization_and_precondition() {
+    use crate::admin::storage_api::object::WriteCommitGuard;
+    use crate::storage::storage_api::contract::namespace::NamespaceLocking as _;
+    use crate::table_catalog::{EcStoreTableCatalogObjectBackend, StrongTableCatalogRuntime, TableCatalogPutPrecondition};
+    use std::time::Duration;
+
+    let (_temp_dir, _disk_paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+    let bucket = "catalog-borrowed-owner";
+    let object = "metadata.json";
+    store
+        .make_bucket(bucket, &MakeBucketOptions::default())
+        .await
+        .expect("create catalog object bucket");
+    let lock = store.new_ns_lock(bucket, object).await.expect("target namespace wrapper");
+    let guard = WriteCommitGuard::acquire(&lock, Duration::from_secs(5))
+        .await
+        .expect("hold actual target owner");
+    let backend = EcStoreTableCatalogObjectBackend::new_with_strong_runtime(store, StrongTableCatalogRuntime::default());
+    let authorized = Arc::new(tokio::sync::Mutex::new(Vec::new()));
+    let commit = TableCommitObjectBackend::test(backend.clone(), Arc::clone(&authorized), None);
+    let payload = br#"{"version":1}"#.to_vec();
+
+    tokio::time::timeout(
+        Duration::from_secs(5),
+        commit.put_object_unlocked(
+            bucket,
+            object,
+            payload.clone(),
+            TableCatalogPutPrecondition::IfAbsent,
+            vec![guard.clone()],
+        ),
+    )
+    .await
+    .expect("borrowed catalog write must not acquire the held target lock again")
+    .expect("borrowed catalog publication succeeds");
+    assert!(
+        authorized
+            .lock()
+            .await
+            .contains(&(object.to_string(), S3Action::PutObjectAction))
+    );
+    let stored = backend
+        .read_object_unlocked(bucket, object)
+        .await
+        .expect("read under outer write owner")
+        .expect("object exists");
+    assert_eq!(stored.data, payload);
+    assert!(!guard.is_lock_lost());
+    let conflict = tokio::time::timeout(
+        Duration::from_secs(5),
+        commit.put_object_unlocked(
+            bucket,
+            object,
+            b"replacement".to_vec(),
+            TableCatalogPutPrecondition::IfAbsent,
+            vec![guard],
+        ),
+    )
+    .await
+    .expect("conflicting borrowed write must also avoid recursive acquisition")
+    .expect_err("IfAbsent must reject an existing object");
+    assert!(matches!(conflict, crate::table_catalog::TableCatalogStoreError::Conflict(_)));
+    assert_eq!(
+        tokio::time::timeout(Duration::from_secs(5), backend.read_object(bucket, object))
+            .await
+            .expect("failed borrowed write releases its last namespace owner")
+            .expect("read after owner release")
+            .expect("original remains")
+            .data,
+        payload
+    );
+}
+
 #[tokio::test]
 async fn table_catalog_authentication_and_credentials_use_the_request_context() {
     let (_temp_dir, _disk_paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;

rustfs/src/admin/service/config.rs

@@ -1121,8 +1121,8 @@ mod tests {
             let transaction_order = transaction_order.clone();
             tokio::spawn(async move {
                 let transaction_store = store.clone();
-                with_admin_server_config_write_lock(store, move || async move {
-                    let mut config = read_admin_config_without_migrate_no_lock(transaction_store.clone()).await?;
+                with_admin_server_config_write_lock(store, move |write_guard| async move {
+                    let mut config = read_admin_config_without_migrate_no_lock(transaction_store.clone(), &write_guard).await?;
                     transaction_order.lock().expect("transaction order lock").push("notify-read");
                     notify_entered.notify_one();
                     release_notify.notified().await;
@@ -1135,7 +1135,7 @@ mod tests {
                         .entry(NOTIFY_WEBHOOK_SUB_SYS.to_string())
                         .or_default()
                         .insert("concurrent-notify".to_string(), target);
-                    save_admin_server_config_no_lock(transaction_store, &config).await?;
+                    save_admin_server_config_no_lock(transaction_store, &config, &write_guard).await?;
                     transaction_order.lock().expect("transaction order lock").push("notify-save");
                     Ok::<(), StorageError>(())
                 })
@@ -1154,9 +1154,9 @@ mod tests {
             let transaction_order = transaction_order.clone();
             tokio::spawn(async move {
                 let transaction_store = store.clone();
-                let transaction = with_admin_server_config_write_lock(store, move || async move {
+                let transaction = with_admin_server_config_write_lock(store, move |write_guard| async move {
                     let _ = oidc_entered_tx.send(());
-                    let mut config = read_admin_config_without_migrate_no_lock(transaction_store.clone()).await?;
+                    let mut config = read_admin_config_without_migrate_no_lock(transaction_store.clone(), &write_guard).await?;
                     transaction_order.lock().expect("transaction order lock").push("oidc-read");
 
                     let mut provider = KVS::new();
@@ -1170,7 +1170,7 @@ mod tests {
                         .entry(IDENTITY_OPENID_SUB_SYS.to_string())
                         .or_default()
                         .insert("concurrent-oidc".to_string(), provider);
-                    save_admin_server_config_no_lock(transaction_store, &config).await?;
+                    save_admin_server_config_no_lock(transaction_store, &config, &write_guard).await?;
                     transaction_order.lock().expect("transaction order lock").push("oidc-save");
                     Ok::<(), StorageError>(())
                 });
@@ -1449,9 +1449,9 @@ mod tests {
                 let (writer_entered_tx, mut writer_entered_rx) = tokio::sync::oneshot::channel();
                 let writer = tokio::spawn(async move {
                     let transaction_store = writer_store.clone();
-                    let transaction = with_admin_server_config_write_lock(writer_store, move || async move {
+                    let transaction = with_admin_server_config_write_lock(writer_store, move |write_guard| async move {
                         writer_entered_tx.send(()).expect("signal writer entry");
-                        save_admin_server_config_no_lock(transaction_store, &latest).await
+                        save_admin_server_config_no_lock(transaction_store, &latest, &write_guard).await
                     });
                     tokio::pin!(transaction);
                     let mut writer_polled_tx = Some(writer_polled_tx);

rustfs/src/admin/service/site_replication.rs

@@ -118,13 +118,13 @@ pub async fn reload_site_replication_runtime_state_for_context(context: Option<&
     // fan-out cannot overwrite a concurrent state writer. IO must be the
     // no-lock variants — the boundary already holds the object lock.
     let lock_store = store.clone();
-    with_site_replication_state_lock_on(lock_store, move || async move {
+    with_site_replication_state_lock_on(lock_store, move |write_guard| async move {
         match read_config_no_lock(store.clone(), SITE_REPLICATION_STATE_PATH).await {
             Ok(data) => {
                 if let Some(normalized) = normalize_site_replication_state_json(&data)
                     .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e))?
                 {
-                    save_config_no_lock(store, SITE_REPLICATION_STATE_PATH, normalized)
+                    save_config_no_lock(store, SITE_REPLICATION_STATE_PATH, normalized, &write_guard)
                         .await
                         .map_err(|e| {
                             S3Error::with_message(

rustfs/src/admin/storage_api.rs

@@ -826,8 +826,11 @@ pub(crate) async fn read_existing_admin_server_config_no_lock(api: Arc<ECStore>)
 }
 
 #[cfg(test)]
-pub(crate) async fn read_admin_config_without_migrate_no_lock(api: Arc<ECStore>) -> Result<rustfs_config::server_config::Config> {
-    ecstore_config::com::read_config_without_migrate_no_lock(api).await
+pub(crate) async fn read_admin_config_without_migrate_no_lock(
+    api: Arc<ECStore>,
+    guard: &crate::storage::storage_api::ecstore_object::WriteCommitGuard,
+) -> Result<rustfs_config::server_config::Config> {
+    ecstore_config::com::read_config_without_migrate_no_lock(api, guard).await
 }
 
 pub(crate) type AdminServerConfigSnapshot = ecstore_config::com::ServerConfigSnapshot;
@@ -850,14 +853,15 @@ pub(crate) async fn save_admin_server_config(api: Arc<ECStore>, cfg: &rustfs_con
 pub(crate) async fn save_admin_server_config_no_lock(
     api: Arc<ECStore>,
     cfg: &rustfs_config::server_config::Config,
+    guard: &crate::storage::storage_api::ecstore_object::WriteCommitGuard,
 ) -> Result<()> {
-    ecstore_config::com::save_server_config_no_lock(api, cfg).await
+    ecstore_config::com::save_server_config_no_lock(api, cfg, guard).await
 }
 
 #[cfg(test)]
 pub(crate) async fn with_admin_server_config_write_lock<F, Fut, T>(api: Arc<ECStore>, operation: F) -> Result<T>
 where
-    F: FnOnce() -> Fut + Send + 'static,
+    F: FnOnce(crate::storage::storage_api::ecstore_object::WriteCommitGuard) -> Fut + Send + 'static,
     Fut: std::future::Future<Output = T> + Send + 'static,
     T: Send + 'static,
 {
@@ -1033,6 +1037,7 @@ pub(crate) mod metrics {
 
 pub(crate) mod object {
     pub(crate) use crate::storage::storage_api::StorageObjectOptions;
+    pub(crate) use crate::storage::storage_api::ecstore_object::WriteCommitGuard;
 }
 
 pub(crate) mod rebalance {

rustfs/src/app/object/copy.rs

@@ -96,13 +96,17 @@ fn copy_namespace_lock_error(bucket: &str, object: &str, mode: &'static str, err
     }
 }
 
-async fn acquire_self_copy_namespace_lock<S>(store: &S, bucket: &str, object: &str) -> S3Result<NamespaceLockGuard>
+async fn acquire_self_copy_namespace_lock<S>(
+    store: &S,
+    bucket: &str,
+    object: &str,
+) -> S3Result<crate::app::storage_api::object_usecase::WriteCommitGuard>
 where
     S: NamespaceLocking<Error = EcstoreError, NamespaceLock = rustfs_lock::NamespaceLockWrapper> + ?Sized,
 {
     let object = encode_dir_object(object);
     let lock = store.new_ns_lock(bucket, &object).await.map_err(ApiError::from)?;
-    lock.get_write_lock(get_lock_acquire_timeout())
+    crate::app::storage_api::object_usecase::WriteCommitGuard::acquire(&lock, get_lock_acquire_timeout())
         .await
         .map_err(|err| ApiError::from(copy_namespace_lock_error(bucket, &object, "write", err)).into())
 }
@@ -420,7 +424,8 @@ impl DefaultObjectUsecase {
             None
         };
         if let Some(guard) = _self_copy_lock_guard.as_ref() {
-            dst_opts.add_namespace_lock_guard(guard);
+            dst_opts.add_write_commit_guard(guard);
+            dst_opts.write_completion = crate::app::storage_api::object_usecase::WriteCompletion::TailDrained;
         }
         dst_opts.expected_current_version_id = expected_current_version_id.clone();
 
@@ -1098,7 +1103,7 @@ mod tests {
         let payload = b"object whose key equals its bucket".to_vec();
         let mut reader = PutObjReader::from_vec(payload.clone());
         let setup_opts = ObjectOptions {
-            no_lock: true,
+            write_completion: crate::app::storage_api::object_usecase::WriteCompletion::TailDrained,
             ..Default::default()
         };
         store

rustfs/src/app/storage_api.rs

@@ -1181,6 +1181,7 @@ pub(crate) mod bucket_usecase {
 
 pub(crate) mod object_usecase {
     pub(crate) use super::storage_contracts::BUCKET_LIFECYCLE_LOCK_OBJECT;
+    pub(crate) use crate::storage::storage_api::ecstore_object::{WriteCommitGuard, WriteCompletion};
 
     pub(crate) mod object_cache {
         #[cfg(test)]

rustfs/src/on_demand_migration/list_through.rs

@@ -1113,22 +1113,11 @@ mod tests {
 
     #[test]
     fn list_through_static_namespace_boundary_matrix() {
-        let corpus = [
-            "a",
-            "a/",
-            "a/b",
-            "a/b/child",
-            "a0",
-            "b",
-            "b/leaf",
-            "quote\"&<",
-            "space key",
-            "z",
-            "é",
-            "中/文",
-        ];
+        let corpus: Vec<String> =
+            serde_json::from_str(include_str!("../../../crates/ecstore/tests/fixtures/list_namespace_keys.json"))
+                .expect("shared LIST namespace corpus should decode");
         for count in [0, 1, 3, 4, corpus.len()] {
-            let keys: Vec<String> = corpus[..count].iter().map(|key| (*key).to_string()).collect();
+            let keys = corpus[..count].to_vec();
             for placement in 0..3 {
                 let (local, source): (Vec<_>, Vec<_>) =
                     keys.iter()

rustfs/src/server/module_switch.rs

@@ -275,11 +275,11 @@ where
     // waiting for the local guard.
     serialize_module_switch_rmw(|| async move {
         let save_store = store.clone();
-        with_config_object_write_lock(store, MODULE_SWITCH_CONFIG_PATH.to_string(), move || async move {
+        with_config_object_write_lock(store, MODULE_SWITCH_CONFIG_PATH.to_string(), move |write_guard| async move {
             save_persisted_module_switches_inner(
                 config,
                 move |data| async move {
-                    save_config_no_lock(save_store, MODULE_SWITCH_CONFIG_PATH, data)
+                    save_config_no_lock(save_store, MODULE_SWITCH_CONFIG_PATH, data, &write_guard)
                         .await
                         .map_err(|e| format!("failed to save module switch config: {e}"))
                 },
@@ -631,14 +631,14 @@ mod tests {
                 async move {
                     writer_started_tx.send(()).expect("signal remote writer start");
                     let save_store = store.clone();
-                    with_config_object_write_lock(store, MODULE_SWITCH_CONFIG_PATH.to_string(), move || async move {
+                    with_config_object_write_lock(store, MODULE_SWITCH_CONFIG_PATH.to_string(), move |write_guard| async move {
                         writer_entered.store(true, AtomicOrdering::SeqCst);
                         let data = serde_json::to_vec(&PersistedModuleSwitches {
                             notify_enabled: false,
                             audit_enabled: false,
                         })
                         .expect("serialize disabled module state");
-                        save_config_no_lock(save_store, MODULE_SWITCH_CONFIG_PATH, data)
+                        save_config_no_lock(save_store, MODULE_SWITCH_CONFIG_PATH, data, &write_guard)
                             .await
                             .map_err(|err| err.to_string())?;
                         publish_order.lock().expect("module publish order lock").push("new-false");

rustfs/src/site_replication/hooks.rs

@@ -46,7 +46,7 @@ where
         store,
         SITE_REPLICATION_BUCKET_MUTATION_ADMISSION_LOCK_PATH.to_string(),
         move || async move {
-            with_config_object_write_lock(mutation_store, mutation_path, operation)
+            with_config_object_write_lock(mutation_store, mutation_path, move |_write_guard| operation())
                 .await
                 .map_err(|err| S3Error::from(ApiError::from(err)))
         },
@@ -68,9 +68,13 @@ where
     Fut: std::future::Future<Output = S3Result<T>> + Send + 'static,
     T: Send + 'static,
 {
-    with_config_object_write_lock(store, SITE_REPLICATION_BUCKET_MUTATION_ADMISSION_LOCK_PATH.to_string(), operation)
-        .await
-        .map_err(|err| S3Error::from(ApiError::from(err)))?
+    with_config_object_write_lock(
+        store,
+        SITE_REPLICATION_BUCKET_MUTATION_ADMISSION_LOCK_PATH.to_string(),
+        move |_write_guard| operation(),
+    )
+    .await
+    .map_err(|err| S3Error::from(ApiError::from(err)))?
 }
 
 #[derive(Debug, Default)]
@@ -978,9 +982,11 @@ pub(crate) async fn commit_site_replication_delete_bucket(intent: &SiteReplicati
         reservations: intent.reservations.clone(),
         displaced: Vec::new(),
     };
-    match with_config_object_write_lock(store, SITE_REPLICATION_REPAIR_EXECUTION_LOCK_PATH.to_string(), move || async move {
-        broadcast_site_replication_delete_bucket(&delivery_intent).await
-    })
+    match with_config_object_write_lock(
+        store,
+        SITE_REPLICATION_REPAIR_EXECUTION_LOCK_PATH.to_string(),
+        move |_write_guard| async move { broadcast_site_replication_delete_bucket(&delivery_intent).await },
+    )
     .await
     {
         Ok(result) => result,

rustfs/src/site_replication/repair.rs

@@ -169,10 +169,11 @@ pub(crate) async fn load_site_replication_repair_state_from_store(store: Arc<ECS
 pub(crate) async fn save_site_replication_repair_state_to_store(
     store: Arc<ECStore>,
     state: &SiteReplicationRepairState,
+    guard: &crate::storage_api::site_replication::WriteCommitGuard,
 ) -> S3Result<()> {
     let data = serde_json::to_vec(state)
         .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize repair state failed: {e}")))?;
-    save_config_no_lock(store, SITE_REPLICATION_REPAIR_STATE_PATH, data)
+    save_config_no_lock(store, SITE_REPLICATION_REPAIR_STATE_PATH, data, guard)
         .await
         .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("save repair state failed: {e}")))
 }
@@ -197,10 +198,10 @@ where
         current_object_store_handle().ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()))?;
     let read_store = store.clone();
     let save_store = store.clone();
-    with_config_object_write_lock(store, SITE_REPLICATION_REPAIR_STATE_PATH.to_string(), move || async move {
+    with_config_object_write_lock(store, SITE_REPLICATION_REPAIR_STATE_PATH.to_string(), move |write_guard| async move {
         let mut state = load_site_replication_repair_state_from_store(read_store).await?;
         let result = update(&mut state)?;
-        save_site_replication_repair_state_to_store(save_store, &state).await?;
+        save_site_replication_repair_state_to_store(save_store, &state, &write_guard).await?;
         Ok(result)
     })
     .await
@@ -709,9 +710,11 @@ pub(crate) async fn execute_site_replication_repair(
 ) -> S3Result<S3Response<(StatusCode, Body)>> {
     let store =
         current_object_store_handle().ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()))?;
-    with_config_object_write_lock(store, SITE_REPLICATION_REPAIR_EXECUTION_LOCK_PATH.to_string(), move || async move {
-        execute_site_replication_repair_locked(request).await
-    })
+    with_config_object_write_lock(
+        store,
+        SITE_REPLICATION_REPAIR_EXECUTION_LOCK_PATH.to_string(),
+        move |_write_guard| async move { execute_site_replication_repair_locked(request).await },
+    )
     .await
     .map_err(|_| {
         S3Error::with_message(S3ErrorCode::ClientTokenConflict, "another site replication repair is active".to_string())

rustfs/src/site_replication/retry.rs

@@ -1700,23 +1700,27 @@ async fn drain_site_replication_retry_queue_lightweight_inner() -> S3Result<()>
     // timestamp, and the locked reload below decides what may actually send.
     promote_reachable_deferred_retry_events(&runtime, &actionable, deferred).await?;
 
-    with_config_object_write_lock(store, SITE_REPLICATION_REPAIR_EXECUTION_LOCK_PATH.to_string(), move || async move {
-        let Some(runtime) = runtime_site_replication_targets().await? else {
-            return Ok(());
-        };
-        if runtime.state.pending_endpoint_refresh.is_some()
-            || runtime.state.pending_remove.is_some()
-            || runtime.state.pending_rotation.is_some()
-        {
-            return Ok(());
-        }
-        let now = OffsetDateTime::now_utc();
-        let (actionable, _) = lightweight_retry_drain_partition(&runtime.state, now);
-        if actionable.is_empty() {
-            return Ok(());
-        }
-        drain_site_replication_retry_queue_lightweight_locked(Arc::new(runtime), actionable, now).await
-    })
+    with_config_object_write_lock(
+        store,
+        SITE_REPLICATION_REPAIR_EXECUTION_LOCK_PATH.to_string(),
+        move |_write_guard| async move {
+            let Some(runtime) = runtime_site_replication_targets().await? else {
+                return Ok(());
+            };
+            if runtime.state.pending_endpoint_refresh.is_some()
+                || runtime.state.pending_remove.is_some()
+                || runtime.state.pending_rotation.is_some()
+            {
+                return Ok(());
+            }
+            let now = OffsetDateTime::now_utc();
+            let (actionable, _) = lightweight_retry_drain_partition(&runtime.state, now);
+            if actionable.is_empty() {
+                return Ok(());
+            }
+            drain_site_replication_retry_queue_lightweight_locked(Arc::new(runtime), actionable, now).await
+        },
+    )
     .await
     .map_err(ApiError::from)?
 }
@@ -1762,26 +1766,30 @@ pub(crate) async fn drain_site_replication_retry_queue_inner() -> S3Result<()> {
     // re-runs the dry-run. The lock elects one server to replay the queue;
     // after acquiring it, reload state so a settled event or deleted bucket
     // cannot be replayed from this admission snapshot.
-    with_config_object_write_lock(store, SITE_REPLICATION_REPAIR_EXECUTION_LOCK_PATH.to_string(), move || async move {
-        // Runtime and queue snapshots captured before this distributed lock
-        // are only admission hints. Another node may have settled the event,
-        // or a local bucket may have been deleted, while this node waited.
-        let Some(runtime) = runtime_site_replication_targets().await? else {
-            return Ok(());
-        };
-        if runtime.state.pending_endpoint_refresh.is_some()
-            || runtime.state.pending_remove.is_some()
-            || runtime.state.pending_rotation.is_some()
-        {
-            return Ok(());
-        }
-        let horizon = heavyweight_retry_drain_horizon(OffsetDateTime::now_utc());
-        let actionable = actionable_site_replication_retry_events(&runtime.state, horizon);
-        if actionable.is_empty() {
-            return Ok(());
-        }
-        drain_site_replication_retry_queue_locked(runtime, actionable).await
-    })
+    with_config_object_write_lock(
+        store,
+        SITE_REPLICATION_REPAIR_EXECUTION_LOCK_PATH.to_string(),
+        move |_write_guard| async move {
+            // Runtime and queue snapshots captured before this distributed lock
+            // are only admission hints. Another node may have settled the event,
+            // or a local bucket may have been deleted, while this node waited.
+            let Some(runtime) = runtime_site_replication_targets().await? else {
+                return Ok(());
+            };
+            if runtime.state.pending_endpoint_refresh.is_some()
+                || runtime.state.pending_remove.is_some()
+                || runtime.state.pending_rotation.is_some()
+            {
+                return Ok(());
+            }
+            let horizon = heavyweight_retry_drain_horizon(OffsetDateTime::now_utc());
+            let actionable = actionable_site_replication_retry_events(&runtime.state, horizon);
+            if actionable.is_empty() {
+                return Ok(());
+            }
+            drain_site_replication_retry_queue_locked(runtime, actionable).await
+        },
+    )
     .await
     .map_err(ApiError::from)?
 }

rustfs/src/site_replication/state.rs

@@ -382,7 +382,11 @@ pub(crate) async fn load_site_replication_state_no_lock(store: Arc<ECStore>) ->
 /// Persist-or-clear under an already-held state object lock. Normalizes the
 /// peer map exactly once (the historical persist path normalized twice with
 /// two full clones — P2-22).
-pub(crate) async fn persist_site_replication_state_no_lock(store: Arc<ECStore>, mut state: SiteReplicationState) -> S3Result<()> {
+pub(crate) async fn persist_site_replication_state_no_lock(
+    store: Arc<ECStore>,
+    mut state: SiteReplicationState,
+    guard: &crate::storage_api::site_replication::WriteCommitGuard,
+) -> S3Result<()> {
     state.peers = normalize_peer_map_by_identity(state.peers);
     if state.peers.len() <= 1 && state.pending_rotation.is_none() && state.pending_remove.is_none() {
         match delete_config_no_lock(store, SITE_REPLICATION_STATE_PATH).await {
@@ -392,7 +396,7 @@ pub(crate) async fn persist_site_replication_state_no_lock(store: Arc<ECStore>,
     } else {
         let data = serde_json::to_vec(&state)
             .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize state failed: {e}")))?;
-        save_config_no_lock(store, SITE_REPLICATION_STATE_PATH, data)
+        save_config_no_lock(store, SITE_REPLICATION_STATE_PATH, data, guard)
             .await
             .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("save state failed: {e}")))
     }
@@ -437,13 +441,13 @@ where
     F: FnOnce(SiteReplicationState) -> Fut + Send + 'static,
     Fut: std::future::Future<Output = S3Result<(T, Option<SiteReplicationState>)>> + Send + 'static,
 {
-    with_site_replication_state_lock(move || async move {
+    with_site_replication_state_lock(move |write_guard| async move {
         let store = current_object_store_handle()
             .ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()))?;
         let state = load_site_replication_state_no_lock(store.clone()).await?;
         let (result, changed) = transaction(state).await?;
         if let Some(state) = changed {
-            persist_site_replication_state_no_lock(store, state).await?;
+            persist_site_replication_state_no_lock(store, state, &write_guard).await?;
         }
         Ok(result)
     })
@@ -457,13 +461,13 @@ where
     T: Send + 'static,
     F: FnOnce(&mut SiteReplicationState) -> S3Result<StateCommit<T>> + Send + 'static,
 {
-    with_site_replication_state_lock(move || async move {
+    with_site_replication_state_lock(move |write_guard| async move {
         let store = current_object_store_handle()
             .ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()))?;
         let mut state = load_site_replication_state_no_lock(store.clone()).await?;
         match update(&mut state)? {
             StateCommit::Changed(result) => {
-                persist_site_replication_state_no_lock(store, state).await?;
+                persist_site_replication_state_no_lock(store, state, &write_guard).await?;
                 Ok(result)
             }
             StateCommit::Unchanged(result) => Ok(result),

rustfs/src/site_replication/state_lock.rs

@@ -54,7 +54,7 @@ pub(crate) const SITE_REPLICATION_STATE_PATH: &str = "config/site-replication/st
 pub(crate) async fn with_site_replication_state_lock<T, F, Fut>(operation: F) -> S3Result<T>
 where
     T: Send + 'static,
-    F: FnOnce() -> Fut + Send + 'static,
+    F: FnOnce(crate::storage_api::site_replication::WriteCommitGuard) -> Fut + Send + 'static,
     Fut: std::future::Future<Output = S3Result<T>> + Send + 'static,
 {
     let store = current_object_store_handle().ok_or_else(|| S3Error::with_message(S3ErrorCode::InternalError, "Not init"))?;
@@ -91,7 +91,7 @@ where
 pub(crate) async fn with_site_replication_state_lock_on<T, F, Fut>(store: Arc<ECStore>, operation: F) -> S3Result<T>
 where
     T: Send + 'static,
-    F: FnOnce() -> Fut + Send + 'static,
+    F: FnOnce(crate::storage_api::site_replication::WriteCommitGuard) -> Fut + Send + 'static,
     Fut: std::future::Future<Output = S3Result<T>> + Send + 'static,
 {
     with_config_object_write_lock(store, SITE_REPLICATION_STATE_PATH.to_string(), operation)

rustfs/src/storage/rpc/node_service.rs

@@ -9060,7 +9060,7 @@ mod tests {
                 "object",
                 &mut reader,
                 &ObjectOptions {
-                    no_lock: true,
+                    write_completion: crate::storage::storage_api::ecstore_object::WriteCompletion::TailDrained,
                     ..Default::default()
                 },
             ),

rustfs/src/storage/storage_api.rs

@@ -577,9 +577,9 @@ pub(crate) mod ecstore_object {
     pub(crate) use rustfs_ecstore::api::object::{
         EncryptionResolutionError, EncryptionResolutionErrorKind, GetObjectBodyCacheHook, GetObjectBodyCacheHookLookup,
         ObjectEncryptionResolver, ObjectMutationHook, PrepareSelectObjectSnapshotError, ReadEncryptionMaterial,
-        ReadEncryptionMode, ReadEncryptionRequest, SelectObjectSnapshot, WriteCompletion, get_object_body_cache_plaintext_len,
-        lookup_get_object_body_cache_hook, register_get_object_body_cache_hook, register_object_mutation_hook,
-        unregister_get_object_body_cache_hook, unregister_object_mutation_hook,
+        ReadEncryptionMode, ReadEncryptionRequest, SelectObjectSnapshot, WriteCommitGuard, WriteCompletion,
+        get_object_body_cache_plaintext_len, lookup_get_object_body_cache_hook, register_get_object_body_cache_hook,
+        register_object_mutation_hook, unregister_get_object_body_cache_hook, unregister_object_mutation_hook,
     };
 }
 
@@ -1296,8 +1296,13 @@ pub(crate) fn replication_queue_current_count() -> Option<i64> {
     get_global_replication_stats().and_then(|stats| stats.queue_current_count())
 }
 
-pub(crate) async fn save_config_no_lock(api: Arc<ECStore>, file: &str, data: Vec<u8>) -> Result<()> {
-    ecstore_config::com::save_config_no_lock(api, file, data).await
+pub(crate) async fn save_config_no_lock(
+    api: Arc<ECStore>,
+    file: &str,
+    data: Vec<u8>,
+    guard: &ecstore_object::WriteCommitGuard,
+) -> Result<()> {
+    ecstore_config::com::save_config_no_lock(api, file, data, guard).await
 }
 
 pub(crate) async fn delete_config_no_lock(api: Arc<ECStore>, file: &str) -> Result<()> {
@@ -1306,7 +1311,7 @@ pub(crate) async fn delete_config_no_lock(api: Arc<ECStore>, file: &str) -> Resu
 
 pub(crate) async fn with_config_object_write_lock<F, Fut, T>(api: Arc<ECStore>, object: String, operation: F) -> Result<T>
 where
-    F: FnOnce() -> Fut + Send + 'static,
+    F: FnOnce(ecstore_object::WriteCommitGuard) -> Fut + Send + 'static,
     Fut: std::future::Future<Output = T> + Send + 'static,
     T: Send + 'static,
 {

rustfs/src/storage_api.rs

@@ -289,6 +289,7 @@ pub(crate) mod site_replication {
     pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata::{
         BUCKET_REPLICATION_CONFIG, BUCKET_TARGETS_FILE, BUCKET_VERSIONING_CONFIG, BucketMetadata,
     };
+    pub(crate) use crate::storage::storage_api::ecstore_object::WriteCommitGuard;
 
     #[cfg(test)]
     pub(crate) use crate::storage::storage_api::ecstore_bucket::replication::merge_incoming_replication_config;
@@ -443,6 +444,7 @@ pub(crate) mod workload {
 }
 
 pub(crate) mod table {
+    pub(crate) use crate::storage::storage_api::ecstore_object::{WriteCommitGuard, WriteCompletion};
     pub(crate) mod contract {
         pub(crate) mod http {
             pub(crate) use super::super::super::storage_contracts::HTTPPreconditions;

rustfs/src/table_catalog/store/mod.rs

@@ -14,6 +14,7 @@
 
 use super::identifier::is_valid_table_metadata_file_name;
 use super::*;
+use crate::storage_api::table::{WriteCommitGuard, WriteCompletion};
 
 mod migration;
 mod object;
@@ -573,6 +574,7 @@ pub(crate) struct TableCatalogObjectMetadata {
 pub(crate) struct TableCatalogLockGuard {
     _guard: Box<dyn Send>,
     lock_lost: Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
+    write_commit_guard: Option<WriteCommitGuard>,
 }
 
 impl TableCatalogLockGuard {
@@ -584,6 +586,7 @@ impl TableCatalogLockGuard {
         Self {
             _guard: Box::new(guard),
             lock_lost: None,
+            write_commit_guard: None,
         }
     }
 
@@ -592,11 +595,25 @@ impl TableCatalogLockGuard {
         Self {
             _guard: Box::new(guard),
             lock_lost,
+            write_commit_guard: None,
         }
     }
 
+    fn write_namespace(guard: WriteCommitGuard) -> Self {
+        Self {
+            _guard: Box::new(()),
+            lock_lost: None,
+            write_commit_guard: Some(guard),
+        }
+    }
+
+    fn write_commit_guards(&self) -> Vec<WriteCommitGuard> {
+        self.write_commit_guard.iter().cloned().collect()
+    }
+
     pub(crate) fn is_lock_lost(&self) -> bool {
         self.lock_lost.as_ref().is_some_and(|signal| signal.is_lost())
+            || self.write_commit_guard.as_ref().is_some_and(WriteCommitGuard::is_lock_lost)
     }
 }
 
@@ -722,6 +739,7 @@ pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static {
         object: &str,
         data: Vec<u8>,
         precondition: TableCatalogPutPrecondition,
+        _write_guards: Vec<WriteCommitGuard>,
     ) -> TableCatalogStoreResult<()> {
         self.put_object(bucket, object, data, precondition).await
     }
@@ -1946,7 +1964,7 @@ where
         data: Vec<u8>,
         precondition: TableCatalogPutPrecondition,
     ) -> TableCatalogStoreResult<()> {
-        self.put_object_with_options(bucket, object, data, precondition, false).await
+        self.put_object_with_options(bucket, object, data, precondition, None).await
     }
 
     async fn put_object_unlocked(
@@ -1955,8 +1973,10 @@ where
         object: &str,
         data: Vec<u8>,
         precondition: TableCatalogPutPrecondition,
+        write_guards: Vec<WriteCommitGuard>,
     ) -> TableCatalogStoreResult<()> {
-        self.put_object_with_options(bucket, object, data, precondition, true).await
+        self.put_object_with_options(bucket, object, data, precondition, Some(write_guards))
+            .await
     }
 
     async fn delete_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<()> {
@@ -2043,11 +2063,10 @@ where
             .new_ns_lock(bucket, object)
             .await
             .map_err(|err| storage_error_to_catalog("create catalog table lock", err))?;
-        let guard = lock
-            .get_write_lock(get_lock_acquire_timeout())
+        let guard = WriteCommitGuard::acquire(&lock, get_lock_acquire_timeout())
             .await
             .map_err(|err| catalog_lock_acquisition_error("acquire catalog table lock", err))?;
-        Ok(TableCatalogLockGuard::namespace(guard))
+        Ok(TableCatalogLockGuard::write_namespace(guard))
     }
 
     async fn acquire_read_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<TableCatalogLockGuard> {
@@ -2120,14 +2139,22 @@ where
         object: &str,
         data: Vec<u8>,
         precondition: TableCatalogPutPrecondition,
-        no_lock: bool,
+        write_guards: Option<Vec<WriteCommitGuard>>,
     ) -> TableCatalogStoreResult<()> {
         let mut reader = PutObjReader::from_vec(data);
-        let opts = ObjectOptions {
+        let mut opts = ObjectOptions {
             http_preconditions: http_preconditions_for_catalog_put(precondition),
-            no_lock,
             ..Default::default()
         };
+        if let Some(write_guards) = write_guards {
+            if write_guards.is_empty() {
+                return Err(TableCatalogStoreError::Internal("catalog write has no namespace owner".to_string()));
+            }
+            for guard in &write_guards {
+                opts.add_write_commit_guard(guard);
+            }
+            opts.write_completion = WriteCompletion::TailDrained;
+        }
         self.store
             .put_object(bucket, object, &mut reader, &opts)
             .await

rustfs/src/table_catalog/store/object.rs

@@ -747,13 +747,16 @@ where
         object: &str,
         entry: &T,
         precondition: TableCatalogPutPrecondition,
+        write_guards: Vec<WriteCommitGuard>,
     ) -> TableCatalogStoreResult<()>
     where
         T: Serialize,
     {
         let data = serde_json::to_vec(entry)
             .map_err(|err| TableCatalogStoreError::Internal(format!("failed to serialize catalog entry {object}: {err}")))?;
-        self.backend.put_object_unlocked(bucket, object, data, precondition).await
+        self.backend
+            .put_object_unlocked(bucket, object, data, precondition, write_guards)
+            .await
     }
 
     async fn write_exact_entry_unlocked<T>(
@@ -762,11 +765,14 @@ where
         object: &str,
         entry: &T,
         precondition: TableCatalogPutPrecondition,
+        write_guards: Vec<WriteCommitGuard>,
     ) -> TableCatalogStoreResult<String>
     where
         T: DeserializeOwned + PartialEq + Serialize,
     {
-        let write_result = self.write_entry_unlocked(bucket, object, entry, precondition).await;
+        let write_result = self
+            .write_entry_unlocked(bucket, object, entry, precondition, write_guards)
+            .await;
         let current = self.read_entry_unlocked::<T>(bucket, object).await?;
         match current {
             Some((current, Some(etag))) if current == *entry => Ok(etag),
@@ -878,14 +884,21 @@ where
         intent: &mut TableRenameIntent,
         etag: String,
         state: TableRenameIntentState,
+        write_guards: Vec<WriteCommitGuard>,
     ) -> TableCatalogStoreResult<String> {
         if intent.state >= state {
             return Ok(etag);
         }
         intent.state = state;
         intent.updated_at = OffsetDateTime::now_utc().to_string();
-        self.write_exact_entry_unlocked(self.catalog_bucket(), object, intent, TableCatalogPutPrecondition::IfMatch(etag))
-            .await
+        self.write_exact_entry_unlocked(
+            self.catalog_bucket(),
+            object,
+            intent,
+            TableCatalogPutPrecondition::IfMatch(etag),
+            write_guards,
+        )
+        .await
     }
 
     async fn recover_active_table_rename(
@@ -1027,6 +1040,10 @@ where
                     &source_object,
                     &source_fence,
                     TableCatalogPutPrecondition::IfMatch(current_etag),
+                    _catalog_guards
+                        .iter()
+                        .flat_map(TableCatalogLockGuard::write_commit_guards)
+                        .collect(),
                 )
                 .await?;
             }
@@ -1039,7 +1056,13 @@ where
             }
         }
         intent_etag = self
-            .advance_table_rename_intent_unlocked(&intent_object, &mut intent, intent_etag, TableRenameIntentState::SourceFenced)
+            .advance_table_rename_intent_unlocked(
+                &intent_object,
+                &mut intent,
+                intent_etag,
+                TableRenameIntentState::SourceFenced,
+                _intent_guard.write_commit_guards(),
+            )
             .await?;
 
         let mut destination_fence = intent.destination.clone();
@@ -1055,6 +1078,10 @@ where
                     &destination_object,
                     &destination_fence,
                     TableCatalogPutPrecondition::IfAbsent,
+                    _catalog_guards
+                        .iter()
+                        .flat_map(TableCatalogLockGuard::write_commit_guards)
+                        .collect(),
                 )
                 .await?;
             }
@@ -1067,6 +1094,10 @@ where
                     &destination_object,
                     &destination_fence,
                     TableCatalogPutPrecondition::IfMatch(current_etag),
+                    _catalog_guards
+                        .iter()
+                        .flat_map(TableCatalogLockGuard::write_commit_guards)
+                        .collect(),
                 )
                 .await?;
             }
@@ -1084,6 +1115,7 @@ where
                 &mut intent,
                 intent_etag,
                 TableRenameIntentState::DestinationWritten,
+                _intent_guard.write_commit_guards(),
             )
             .await?;
 
@@ -1098,6 +1130,10 @@ where
                     &source_object,
                     &source_tombstone,
                     TableCatalogPutPrecondition::IfMatch(current_etag),
+                    _catalog_guards
+                        .iter()
+                        .flat_map(TableCatalogLockGuard::write_commit_guards)
+                        .collect(),
                 )
                 .await?;
             }
@@ -1115,6 +1151,7 @@ where
                 &mut intent,
                 intent_etag,
                 TableRenameIntentState::SourceTombstoned,
+                _intent_guard.write_commit_guards(),
             )
             .await?;
 
@@ -1139,6 +1176,10 @@ where
                 &index_object,
                 &destination_index,
                 TableCatalogPutPrecondition::IfMatch(intent.warehouse_index_etag.clone()),
+                _catalog_guards
+                    .iter()
+                    .flat_map(TableCatalogLockGuard::write_commit_guards)
+                    .collect(),
             )
             .await?;
         }
@@ -1148,6 +1189,7 @@ where
                 &mut intent,
                 intent_etag,
                 TableRenameIntentState::IndexPublished,
+                _intent_guard.write_commit_guards(),
             )
             .await?;
         match self
@@ -1161,6 +1203,10 @@ where
                     &destination_object,
                     &intent.destination,
                     TableCatalogPutPrecondition::IfMatch(current_etag),
+                    _catalog_guards
+                        .iter()
+                        .flat_map(TableCatalogLockGuard::write_commit_guards)
+                        .collect(),
                 )
                 .await?;
             }
@@ -1178,10 +1224,17 @@ where
                 &mut intent,
                 intent_etag,
                 TableRenameIntentState::DestinationPublished,
+                _intent_guard.write_commit_guards(),
             )
             .await?;
-        self.advance_table_rename_intent_unlocked(&intent_object, &mut intent, intent_etag, TableRenameIntentState::Completed)
-            .await?;
+        self.advance_table_rename_intent_unlocked(
+            &intent_object,
+            &mut intent,
+            intent_etag,
+            TableRenameIntentState::Completed,
+            _intent_guard.write_commit_guards(),
+        )
+        .await?;
 
         let Some((mut bucket_entry, bucket_etag)) = self.read_table_bucket_with_etag_unlocked(table_bucket).await? else {
             return Err(TableCatalogStoreError::NotFound(format!("table bucket {table_bucket}")));
@@ -1196,12 +1249,17 @@ where
             &bucket_object,
             &bucket_entry,
             TableCatalogPutPrecondition::IfMatch(bucket_etag),
+            _bucket_guard.write_commit_guards(),
         )
         .await?;
         Ok(())
     }
 
-    async fn write_warehouse_index_state_unlocked(&self, table_bucket: &str) -> TableCatalogStoreResult<()> {
+    async fn write_warehouse_index_state_unlocked(
+        &self,
+        table_bucket: &str,
+        write_guards: Vec<WriteCommitGuard>,
+    ) -> TableCatalogStoreResult<()> {
         let state = TableWarehouseIndexStateEntry {
             version: TABLE_WAREHOUSE_INDEX_STATE_VERSION,
             table_bucket: table_bucket.to_string(),
@@ -1212,6 +1270,7 @@ where
             &self.paths.warehouse_index_state_path(table_bucket),
             &state,
             TableCatalogPutPrecondition::Any,
+            write_guards,
         )
         .await
     }
@@ -1324,6 +1383,7 @@ where
             object,
             replacement,
             TableCatalogPutPrecondition::IfMatch(current_etag),
+            _guard.write_commit_guards(),
         )
         .await?;
         Ok(true)
@@ -1606,7 +1666,13 @@ where
             .await?
         else {
             return self
-                .write_entry_unlocked(self.catalog_bucket(), &object, &tombstone, TableCatalogPutPrecondition::IfAbsent)
+                .write_entry_unlocked(
+                    self.catalog_bucket(),
+                    &object,
+                    &tombstone,
+                    TableCatalogPutPrecondition::IfAbsent,
+                    _guard.write_commit_guards(),
+                )
                 .await;
         };
         validate_table_warehouse_index_entry_object(&self.paths, &object, &current)?;
@@ -1626,6 +1692,7 @@ where
             &object,
             &tombstone,
             TableCatalogPutPrecondition::IfMatch(current_etag),
+            _guard.write_commit_guards(),
         )
         .await
     }
@@ -1927,7 +1994,8 @@ where
             )
             .await?;
         }
-        self.write_warehouse_index_state_unlocked(table_bucket).await
+        self.write_warehouse_index_state_unlocked(table_bucket, _guard.write_commit_guards())
+            .await
     }
 
     async fn require_table_bucket(&self, table_bucket: &str) -> TableCatalogStoreResult<()> {
@@ -2178,7 +2246,13 @@ where
             ));
         }
         let result = self
-            .write_entry_unlocked(self.catalog_bucket(), &table_path, &entry, precondition)
+            .write_entry_unlocked(
+                self.catalog_bucket(),
+                &table_path,
+                &entry,
+                precondition,
+                _table_guard.write_commit_guards(),
+            )
             .await;
         if result.is_err() {
             self.rollback_table_warehouse_index_reservation(&entry, reservation, "table entry write failed")
@@ -2244,7 +2318,7 @@ where
                 "view creation publication fence was lost before catalog update".to_string(),
             ));
         }
-        self.write_entry_unlocked(self.catalog_bucket(), &view_path, &entry, precondition)
+        self.write_entry_unlocked(self.catalog_bucket(), &view_path, &entry, precondition, _view_guard.write_commit_guards())
             .await
     }
 
@@ -4799,8 +4873,14 @@ where
             }
             entry.updated_at = Some(next_table_catalog_update_time(current.updated_at.as_deref()));
         }
-        self.write_entry_unlocked(self.catalog_bucket(), &object, &entry, TableCatalogPutPrecondition::Any)
-            .await
+        self.write_entry_unlocked(
+            self.catalog_bucket(),
+            &object,
+            &entry,
+            TableCatalogPutPrecondition::Any,
+            _guard.write_commit_guards(),
+        )
+        .await
     }
 
     async fn create_namespace(&self, entry: NamespaceEntry) -> TableCatalogStoreResult<()> {
@@ -4849,8 +4929,14 @@ where
                 TableCatalogPutPrecondition::IfAbsent
             }
         };
-        self.write_entry_unlocked(self.catalog_bucket(), &object, &entry, precondition)
-            .await
+        self.write_entry_unlocked(
+            self.catalog_bucket(),
+            &object,
+            &entry,
+            precondition,
+            _namespace_guard.write_commit_guards(),
+        )
+        .await
     }
 
     async fn list_namespaces(&self, table_bucket: &str) -> TableCatalogStoreResult<Vec<NamespaceEntry>> {
@@ -4954,8 +5040,14 @@ where
         if before == next {
             return Ok(result);
         }
-        self.write_entry_unlocked(self.catalog_bucket(), &namespace_path, &next, precondition)
-            .await?;
+        self.write_entry_unlocked(
+            self.catalog_bucket(),
+            &namespace_path,
+            &next,
+            precondition,
+            _namespace_guard.write_commit_guards(),
+        )
+        .await?;
         Ok(result)
     }
 
@@ -5313,6 +5405,7 @@ where
                         &index_object,
                         &source_index,
                         TableCatalogPutPrecondition::IfAbsent,
+                        _index_guard.write_commit_guards(),
                     )
                     .await?
                 }
@@ -5338,6 +5431,7 @@ where
                 &intent_object,
                 &intent,
                 TableCatalogPutPrecondition::IfAbsent,
+                _intent_guard.write_commit_guards(),
             )
             .await?;
             bucket_entry.active_rename_id = Some(rename_id);
@@ -5347,6 +5441,7 @@ where
                 &bucket_object,
                 &bucket_entry,
                 TableCatalogPutPrecondition::IfMatch(bucket_etag),
+                _bucket_guard.write_commit_guards(),
             )
             .await?;
         }
@@ -5796,6 +5891,7 @@ where
                 &table_path,
                 &next,
                 TableCatalogPutPrecondition::IfMatch(current_etag),
+                _guard.write_commit_guards(),
             )
             .await;
         record_table_commit_cas_result(&request.operation, cas_started, &cas_result);
@@ -6091,6 +6187,7 @@ where
                 &view_path,
                 &next,
                 TableCatalogPutPrecondition::IfMatch(current_etag),
+                _guard.write_commit_guards(),
             )
             .await;
         if let Err(err) = write_result {