fix(ecstore): fence deletes against decommission commits (#6363)

* fix(ecstore): fence deletes against decommission commits

* fix(ecstore): preserve decommission target write locks

* fix(ecstore): preserve delete markers during source cleanup

* fix(ecstore): route batch delete markers to active pools

* fix(ecstore): preserve batch delete pool errors

* fix(ecstore): retain source-set lock during cleanup

* test(ecstore): exercise decommission delete fences

* test(ecstore): finish decommission delete fence scenario

* fix(ecstore): reuse fixed fence for reverse decommission

* fix(ecstore): fence decommission commit loss

* fix(ecstore): annotate batch delete fallback

* test(ecstore): fix decommission fence fixtures

* fix(ecstore): unblock decommission delete fences

* fix(ecstore): preserve distributed decommission set locks

* fix(ecstore): match decommission lock backend domain

* test(ecstore): align decommission fence barriers

* fix(ecstore): satisfy delete fence lint checks

* fix(rebalance): preserve access-denied delete errors

crates/ecstore/src/set_disk/mod.rs

@@ -735,8 +735,12 @@ pub(crate) use core::io_primitives::disk_call_counters;
 mod ctx;
 mod metadata;
 mod ops;
+#[cfg(test)]
+pub(crate) use ops::multipart::NewMultipartUploadCommitObservation;
 #[cfg(any(test, feature = "test-util"))]
 pub use ops::multipart::{MultipartCommitBarrier, MultipartCommitPause};
+#[cfg(test)]
+pub(crate) use ops::object::DeleteObjectCommitBarrier;
 #[cfg(feature = "test-util")]
 pub(crate) use ops::object::TransitionCleanupStoreBarrier as SetDiskTransitionCleanupStoreBarrier;
 pub(crate) use ops::object::body_cache_plaintext_len;
@@ -3025,6 +3029,16 @@ pub struct SetDisks {
     storage_class_config_override: Arc<std::sync::RwLock<Option<Arc<storageclass::Config>>>>,
 }
 
+// DistributedLock sends the raw ObjectKey to its clients; LockRegistry clones
+// each endpoint's canonical Arc, so an exact Arc set identifies the lock domain.
+pub(crate) fn same_distributed_lock_domain(left: &[Arc<dyn LockClient>], right: &[Arc<dyn LockClient>]) -> bool {
+    left.iter()
+        .all(|left_client| right.iter().any(|right_client| Arc::ptr_eq(left_client, right_client)))
+        && right
+            .iter()
+            .all(|right_client| left.iter().any(|left_client| Arc::ptr_eq(left_client, right_client)))
+}
+
 const ERASURE_CACHE_MAX_ENTRIES: usize = 32;
 
 #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
@@ -3600,6 +3614,15 @@ impl SetDisks {
         &self.ctx
     }
 
+    /// Whether both sets' namespace-lock implementations cover the same object key.
+    pub(crate) async fn shares_namespace_lock_domain(&self, other: &Self) -> bool {
+        match (self.ctx.is_dist_erasure().await, other.ctx.is_dist_erasure().await) {
+            (false, false) => Arc::ptr_eq(&self.local_lock_manager, &other.local_lock_manager),
+            (true, true) => same_distributed_lock_domain(&self.lockers, &other.lockers),
+            _ => false,
+        }
+    }
+
     /// The lock manager this set actually uses (test-only; Phase 5 Slice 3).
     #[cfg(test)]
     pub(crate) fn local_lock_manager_for_test(&self) -> &Arc<rustfs_lock::GlobalLockManager> {
@@ -4584,11 +4607,11 @@ fn should_preserve_delete_replication_state(opts: &ObjectOptions) -> bool {
 }
 
 fn should_force_delete_marker_for_missing_version(opts: &ObjectOptions) -> bool {
-    opts.delete_marker || (opts.versioned && opts.version_id.is_none() && !opts.data_movement)
+    opts.delete_marker || ((opts.versioned || opts.version_suspended) && opts.version_id.is_none() && !opts.data_movement)
 }
 
 fn resolve_delete_version_state(opts: &ObjectOptions, goi: &ObjectInfo, version_found: bool) -> (bool, bool) {
-    let mut mark_delete = goi.version_id.is_some() || (opts.versioned && opts.version_id.is_none());
+    let mut mark_delete = goi.version_id.is_some() || ((opts.versioned || opts.version_suspended) && opts.version_id.is_none());
     let mut delete_marker = opts.versioned;
 
     if opts.version_id.is_some() {

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

@@ -32,6 +32,8 @@ use crate::crash_inject::{self, CrashPoint};
 use crate::multipart_listing::paginate_multipart_listing;
 use futures::{StreamExt, stream};
 use std::future::Future;
+#[cfg(test)]
+use std::sync::atomic::AtomicBool;
 #[cfg(any(test, feature = "test-util"))]
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::Duration;
@@ -65,6 +67,7 @@ impl StaleMultipartCleanupGuard {
 #[cfg(any(test, feature = "test-util"))]
 #[derive(Clone, Copy, PartialEq, Eq)]
 pub enum MultipartCommitPause {
+    NewUploadBeforeLockLost,
     PutPartBeforeLockAcquire,
     PutPartBeforeLockLost,
     PutPartAfterRename,
@@ -156,6 +159,72 @@ impl Drop for MultipartCommitBarrier {
     }
 }
 
+#[cfg(test)]
+struct NewMultipartUploadCommitObservationState {
+    bucket: String,
+    object: String,
+    committed: AtomicBool,
+}
+
+#[cfg(test)]
+pub(crate) struct NewMultipartUploadCommitObservation {
+    state: Arc<NewMultipartUploadCommitObservationState>,
+}
+
+#[cfg(test)]
+static NEW_MULTIPART_UPLOAD_COMMIT_OBSERVATION: std::sync::OnceLock<
+    std::sync::Mutex<Option<Arc<NewMultipartUploadCommitObservationState>>>,
+> = std::sync::OnceLock::new();
+
+#[cfg(test)]
+impl NewMultipartUploadCommitObservation {
+    pub(crate) fn install(bucket: &str, object: &str) -> Self {
+        let state = Arc::new(NewMultipartUploadCommitObservationState {
+            bucket: bucket.to_string(),
+            object: object.to_string(),
+            committed: AtomicBool::new(false),
+        });
+        let mut slot = NEW_MULTIPART_UPLOAD_COMMIT_OBSERVATION
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("new multipart upload commit observation mutex should not poison");
+        assert!(slot.is_none(), "new multipart upload commit observation must be unique");
+        *slot = Some(Arc::clone(&state));
+        Self { state }
+    }
+
+    pub(crate) fn committed(&self) -> bool {
+        self.state.committed.load(Ordering::Acquire)
+    }
+}
+
+#[cfg(test)]
+impl Drop for NewMultipartUploadCommitObservation {
+    fn drop(&mut self) {
+        let mut slot = NEW_MULTIPART_UPLOAD_COMMIT_OBSERVATION
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("new multipart upload commit observation mutex should not poison");
+        if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) {
+            *slot = None;
+        }
+    }
+}
+
+#[cfg(test)]
+fn observe_new_multipart_upload_commit(bucket: &str, object: &str) {
+    let state = NEW_MULTIPART_UPLOAD_COMMIT_OBSERVATION
+        .get_or_init(|| std::sync::Mutex::new(None))
+        .lock()
+        .expect("new multipart upload commit observation mutex should not poison")
+        .as_ref()
+        .filter(|state| state.bucket == bucket && state.object == object)
+        .cloned();
+    if let Some(state) = state {
+        state.committed.store(true, Ordering::Release);
+    }
+}
+
 #[cfg(any(test, feature = "test-util"))]
 async fn pause_multipart_commit(bucket: &str, object: &str, pause: MultipartCommitPause) {
     let barrier = {
@@ -1615,6 +1684,30 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
         let upload_path = Self::get_multipart_upload_dir(bucket, object, upload_uuid.as_str(), opts.data_movement);
 
+        #[cfg(any(test, feature = "test-util"))]
+        pause_multipart_commit(bucket, object, MultipartCommitPause::NewUploadBeforeLockLost).await;
+        if _object_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) {
+            return Err(StorageError::NamespaceLockQuorumUnavailable {
+                mode: "new_multipart_upload_commit",
+                bucket: bucket.to_string(),
+                object: object.to_string(),
+                required: 1,
+                achieved: 0,
+            });
+        }
+        if opts
+            .namespace_lock_fence
+            .as_ref()
+            .is_some_and(NamespaceLockFence::is_lock_lost)
+        {
+            return Err(StorageError::NamespaceLockQuorumUnavailable {
+                mode: "new_multipart_upload_outer_lock",
+                bucket: bucket.to_string(),
+                object: object.to_string(),
+                required: 1,
+                achieved: 0,
+            });
+        }
         ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
         Self::write_unique_file_info(
             &shuffle_disks,
@@ -1626,6 +1719,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         )
         .await
         .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
+        #[cfg(test)]
+        observe_new_multipart_upload_commit(bucket, object);
 
         // evalDisks
 

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

@@ -2497,6 +2497,7 @@ impl SetDisks {
                         })
                         .await?,
                     );
+                    notify_put_object_commit_namespace_acquired(bucket, object);
                 }
                 #[cfg(not(any(test, feature = "test-util")))]
                 {
@@ -4644,6 +4645,7 @@ struct PutObjectCommitBarrierState {
     arrived: tokio::sync::Notify,
     release: tokio::sync::Notify,
     namespace_pending: tokio::sync::Notify,
+    namespace_acquired: std::sync::atomic::AtomicBool,
 }
 
 #[cfg(any(test, feature = "test-util"))]
@@ -4665,6 +4667,7 @@ impl PutObjectCommitBarrier {
             arrived: tokio::sync::Notify::new(),
             release: tokio::sync::Notify::new(),
             namespace_pending: tokio::sync::Notify::new(),
+            namespace_acquired: std::sync::atomic::AtomicBool::new(false),
         });
         let mut slot = PUT_OBJECT_COMMIT_BARRIER
             .get_or_init(|| std::sync::Mutex::new(Vec::new()))
@@ -4699,6 +4702,10 @@ impl PutObjectCommitBarrier {
             .await
             .expect("put object should wait for the namespace lock after leaving the commit barrier");
     }
+
+    pub fn namespace_acquired(&self) -> bool {
+        self.state.namespace_acquired.load(std::sync::atomic::Ordering::Acquire)
+    }
 }
 
 #[cfg(any(test, feature = "test-util"))]
@@ -4755,6 +4762,22 @@ fn notify_put_object_commit_namespace_pending(bucket: &str, object: &str) {
     }
 }
 
+#[cfg(any(test, feature = "test-util"))]
+fn notify_put_object_commit_namespace_acquired(bucket: &str, object: &str) {
+    let barrier = PUT_OBJECT_COMMIT_BARRIER
+        .get_or_init(|| std::sync::Mutex::new(Vec::new()))
+        .lock()
+        .expect("put object commit barrier mutex should not poison")
+        .iter()
+        .find(|barrier| {
+            barrier.bucket == bucket && barrier.object == object && barrier.pause == PutObjectCommitPause::BeforeNamespace
+        })
+        .cloned();
+    if let Some(barrier) = barrier {
+        barrier.namespace_acquired.store(true, std::sync::atomic::Ordering::Release);
+    }
+}
+
 #[cfg(test)]
 struct DeleteObjectCommitBarrierState {
     bucket: String,
@@ -4764,7 +4787,7 @@ struct DeleteObjectCommitBarrierState {
 }
 
 #[cfg(test)]
-struct DeleteObjectCommitBarrier {
+pub(crate) struct DeleteObjectCommitBarrier {
     state: Arc<DeleteObjectCommitBarrierState>,
 }
 
@@ -4774,7 +4797,7 @@ static DELETE_OBJECT_COMMIT_BARRIER: std::sync::OnceLock<std::sync::Mutex<Option
 
 #[cfg(test)]
 impl DeleteObjectCommitBarrier {
-    fn install(bucket: &str, object: &str) -> Self {
+    pub(crate) fn install(bucket: &str, object: &str) -> Self {
         let state = Arc::new(DeleteObjectCommitBarrierState {
             bucket: bucket.to_string(),
             object: object.to_string(),
@@ -4790,13 +4813,13 @@ impl DeleteObjectCommitBarrier {
         Self { state }
     }
 
-    async fn wait_until_paused(&self) {
+    pub(crate) async fn wait_until_paused(&self) {
         tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
             .await
             .expect("delete object should reach the deterministic commit barrier");
     }
 
-    fn release(&self) {
+    pub(crate) fn release(&self) {
         self.state.release.notify_one();
     }
 }
@@ -5918,6 +5941,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             if dobj.version_id.is_none() && (version_suspended || versioned) {
                 vr.mod_time = Some(OffsetDateTime::now_utc());
                 vr.deleted = true;
+                vr.mark_deleted = true;
                 if versioned {
                     vr.version_id = Some(Uuid::new_v4());
                 }
@@ -11804,6 +11828,7 @@ mod transition_upload_integrity_tests {
                 crate::data_movement::SourceCleanupBucketFence {
                     expected_incarnation_id: None,
                     lifecycle_guard: Some(&bucket_guard),
+                    ..Default::default()
                 },
                 "test_data_movement",
             )