fix(ecstore): fence rebalance and decommission activation (#6400)

* fix(ecstore): fence rebalance and decommission activation

* fix(ecstore): fence lost activation locks

* fix(ecstore): bind rebalance workers to activation id

* fix(ecstore): close rebalance activation races

* test(ecstore): exercise lost rebalance commit fence

* fix(ecstore): repair rebalance fence test wiring

* test(ecstore): reuse rebalance metadata fixture

* fix(ecstore): satisfy rebalance activation clippy checks

* fix(ecstore): fence stale rebalance workers

* fix(ecstore): commit rebalance activation after persistence

* fix(ecstore): fence rebalance commits and unblock stop

* test(ecstore): exercise real rebalance fences

* fix(rebalance): cancel admin stop before activation wait

* fix(ecstore): fence multipart staging on rebalance lock loss

* fix(ecstore): adopt activations after durable commit

* fix(rebalance): preserve committed activation recovery

* fix(rebalance): make prepared stop terminal-safe

* fix(ecstore): repair rebalance test imports

* fix(ecstore): repair rebalance entry runtime failures

* test(ecstore): fix activation fence synchronization

* test(ecstore): scope rebalance disk trait import

* test(ecstore): observe decommission lock attempt

* fix(ecstore): align activation fence test imports

* fix(ecstore): remove duplicate activation test import

* fix(ecstore): resolve CI clippy failures

* fix: satisfy activation merge lint gates

---------

Co-authored-by: houseme <housemecn@gmail.com>

crates/ecstore/src/set_disk/mod.rs

@@ -799,6 +799,9 @@ pub(crate) use core::io_primitives::disk_call_counters;
 mod ctx;
 mod metadata;
 mod ops;
+
+#[cfg(test)]
+pub(crate) use ops::hermetic_set_disks_isolated;
 #[cfg(test)]
 pub(crate) use ops::multipart::NewMultipartUploadCommitObservation;
 #[cfg(any(test, feature = "test-util"))]
@@ -1513,6 +1516,21 @@ mod prepared_get_object_metadata_tests {
 }
 
 impl SetDisks {
+    #[cfg(test)]
+    async fn pause_tiered_metadata_commit(bucket: &str, object: &str) {
+        let barrier = TIERED_METADATA_COMMIT_BARRIER
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("tiered metadata commit barrier should not be poisoned")
+            .as_ref()
+            .filter(|barrier| barrier.bucket == bucket && barrier.object == object)
+            .cloned();
+        if let Some(barrier) = barrier {
+            barrier.arrived.notify_one();
+            barrier.release.notified().await;
+        }
+    }
+
     pub(crate) async fn prepare_get_object_metadata(
         &self,
         bucket: &str,
@@ -4889,6 +4907,8 @@ impl SetDisks {
         )?;
         let fi = build_tiered_decommission_file_info(bucket, object, fi, layout);
         let write_quorum = layout.write_quorum;
+        #[cfg(test)]
+        Self::pause_tiered_metadata_commit(bucket, object).await;
         if _lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost())
             || opts
                 .namespace_lock_fence
@@ -4944,6 +4964,66 @@ impl SetDisks {
     }
 }
 
+#[cfg(test)]
+struct TieredMetadataCommitBarrierState {
+    bucket: String,
+    object: String,
+    arrived: tokio::sync::Notify,
+    release: tokio::sync::Notify,
+}
+
+#[cfg(test)]
+pub(crate) struct TieredMetadataCommitBarrier {
+    state: Arc<TieredMetadataCommitBarrierState>,
+}
+
+#[cfg(test)]
+static TIERED_METADATA_COMMIT_BARRIER: std::sync::OnceLock<std::sync::Mutex<Option<Arc<TieredMetadataCommitBarrierState>>>> =
+    std::sync::OnceLock::new();
+
+#[cfg(test)]
+impl TieredMetadataCommitBarrier {
+    pub(crate) fn install(bucket: &str, object: &str) -> Self {
+        let state = Arc::new(TieredMetadataCommitBarrierState {
+            bucket: bucket.to_string(),
+            object: object.to_string(),
+            arrived: tokio::sync::Notify::new(),
+            release: tokio::sync::Notify::new(),
+        });
+        let mut slot = TIERED_METADATA_COMMIT_BARRIER
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("tiered metadata commit barrier should not be poisoned");
+        assert!(slot.is_none(), "tiered metadata commit barrier must be unique");
+        *slot = Some(Arc::clone(&state));
+        Self { state }
+    }
+
+    pub(crate) async fn wait_until_paused(&self) {
+        tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
+            .await
+            .expect("tiered metadata write should reach its deterministic commit barrier");
+    }
+
+    pub(crate) fn release(&self) {
+        self.state.release.notify_one();
+    }
+}
+
+#[cfg(test)]
+impl Drop for TieredMetadataCommitBarrier {
+    fn drop(&mut self) {
+        self.state.release.notify_one();
+        let mut slot = TIERED_METADATA_COMMIT_BARRIER
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("tiered metadata commit barrier should not be poisoned");
+        if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) {
+            *slot = None;
+        }
+    }
+}
+
 #[derive(Debug, PartialEq, Eq)]
 struct ObjProps {
     successor_mod_time: Option<OffsetDateTime>,

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

@@ -25,3 +25,6 @@ pub(crate) mod list;
 pub(crate) mod locking;
 pub(crate) mod multipart;
 pub(crate) mod object;
+
+#[cfg(test)]
+pub(crate) use object::hermetic_set_disks_support::hermetic_set_disks_isolated;

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

@@ -86,6 +86,8 @@ struct MultipartCommitBarrierState {
     pause: MultipartCommitPause,
     expected_arrivals: usize,
     arrivals: AtomicUsize,
+    #[cfg(test)]
+    committed: AtomicBool,
     arrived: tokio::sync::Notify,
     release: tokio::sync::Semaphore,
 }
@@ -113,6 +115,8 @@ impl MultipartCommitBarrier {
             pause,
             expected_arrivals,
             arrivals: AtomicUsize::new(0),
+            #[cfg(test)]
+            committed: AtomicBool::new(false),
             arrived: tokio::sync::Notify::new(),
             release: tokio::sync::Semaphore::new(0),
         });
@@ -143,6 +147,11 @@ impl MultipartCommitBarrier {
     pub fn release(&self) {
         self.state.release.add_permits(self.state.expected_arrivals);
     }
+
+    #[cfg(test)]
+    pub(crate) fn commit_observed(&self) -> bool {
+        self.state.committed.load(Ordering::Acquire)
+    }
 }
 
 #[cfg(any(test, feature = "test-util"))]
@@ -263,6 +272,20 @@ async fn pause_multipart_commit(bucket: &str, object: &str, pause: MultipartComm
     }
 }
 
+#[cfg(test)]
+fn observe_multipart_commit(bucket: &str, object: &str, pause: MultipartCommitPause) {
+    let slot = MULTIPART_COMMIT_BARRIER
+        .get_or_init(|| std::sync::Mutex::new(None))
+        .lock()
+        .expect("multipart commit barrier mutex should not poison");
+    if let Some(barrier) = slot
+        .as_ref()
+        .filter(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause)
+    {
+        barrier.committed.store(true, Ordering::Release);
+    }
+}
+
 fn map_upload_id_metadata_error(bucket: &str, object: &str, upload_id: &str, err: DiskError) -> Error {
     if err == DiskError::FileNotFound {
         return StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned());
@@ -1337,6 +1360,19 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost).await;
             fence_commit_on_lock_loss(_upload_commit_guard.as_ref(), "put_object_part_commit", &upload_id_path)?;
             fence_commit_on_lock_loss(_part_commit_guard.as_ref(), "put_object_part_commit", &part_lock_path)?;
+            if opts
+                .namespace_lock_fence
+                .as_ref()
+                .is_some_and(NamespaceLockFence::is_lock_lost)
+            {
+                return Err(StorageError::NamespaceLockQuorumUnavailable {
+                    mode: "put_object_part_outer_lock",
+                    bucket: bucket.to_string(),
+                    object: object.to_string(),
+                    required: 1,
+                    achieved: 0,
+                });
+            }
             ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
 
             let _ = self
@@ -1357,6 +1393,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                     }),
                 )
                 .await?;
+            #[cfg(test)]
+            observe_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockLost);
 
             #[cfg(test)]
             pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartAfterRename).await;
@@ -1737,7 +1775,10 @@ 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);
+        {
+            observe_multipart_commit(bucket, object, MultipartCommitPause::NewUploadBeforeLockLost);
+            observe_new_multipart_upload_commit(bucket, object);
+        }
 
         // evalDisks
 

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

@@ -6662,6 +6662,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         };
 
         let find_vid = Uuid::new_v4();
+        #[cfg(test)]
+        pause_delete_object_commit(bucket, object).await;
 
         if mark_delete && (opts.versioned || opts.version_suspended) {
             if !delete_marker {
@@ -6730,8 +6732,6 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             dfi.set_skip_tier_free_version();
         }
 
-        #[cfg(test)]
-        pause_delete_object_commit(bucket, object).await;
         ensure_delete_commit_locks_held(_lock_guard.as_ref(), bucket, object, &opts)?;
         self.delete_object_version(bucket, object, &dfi, opts.delete_marker)
             .await
@@ -7904,9 +7904,7 @@ pub(in crate::set_disk::ops) mod hermetic_set_disks_support {
     /// for tests that never touch context-resolved services registered on the
     /// ambient context (tier config manager, expiry state, ...), because the
     /// isolated context starts every one of those cells fresh.
-    pub(in crate::set_disk::ops) async fn hermetic_set_disks_isolated(
-        disk_count: usize,
-    ) -> (Vec<TempDir>, Vec<DiskStore>, Arc<SetDisks>) {
+    pub(crate) async fn hermetic_set_disks_isolated(disk_count: usize) -> (Vec<TempDir>, Vec<DiskStore>, Arc<SetDisks>) {
         hermetic_set_disks_for_pool_with_default_parity_isolated(disk_count, 0, disk_count / 2).await
     }
 
@@ -12603,6 +12601,7 @@ mod transition_upload_integrity_tests {
                 crate::data_movement::SourceCleanupBucketFence {
                     expected_incarnation_id: None,
                     lifecycle_guard: Some(&bucket_guard),
+                    namespace_lock_lost_signal: None,
                     ..Default::default()
                 },
                 "test_data_movement",