test(ecstore): exercise lost rebalance commit fence

crates/ecstore/src/core/pools.rs

@@ -877,7 +877,11 @@ pub(crate) struct PoolRebalanceActivationFence {
 
 impl PoolRebalanceActivationFence {
     pub(crate) fn ensure_held(&self) -> Result<()> {
-        ensure_activation_locks_held(self.pool_meta_guard.is_lock_lost(), self.rebalance_meta_guard.is_lock_lost())
+        if self.pool_meta_guard.is_lock_lost() || self.rebalance_meta_guard.is_lock_lost() {
+            return Err(Error::other("activation lock lost before metadata commit or worker admission"));
+        }
+
+        Ok(())
     }
 
     pub(crate) fn add_namespace_lock_fence(&self, opts: &mut ObjectOptions) {
@@ -886,14 +890,6 @@ impl PoolRebalanceActivationFence {
     }
 }
 
-fn ensure_activation_locks_held(pool_lock_lost: bool, rebalance_lock_lost: bool) -> Result<()> {
-    if pool_lock_lost || rebalance_lock_lost {
-        return Err(Error::other("activation lock lost before metadata commit or worker admission"));
-    }
-
-    Ok(())
-}
-
 pub(crate) async fn acquire_pool_rebalance_activation_locks<S>(pool: Arc<S>) -> Result<PoolRebalanceActivationFence>
 where
     S: crate::storage_api_contracts::namespace::NamespaceLocking<
@@ -5346,7 +5342,7 @@ mod pools_tests {
         apply_decommission_status_space_info, bind_decommission_cancelers, bind_missing_decommission_cancelers,
         cancel_decommission_canceler, classify_decommission_terminal_state, count_decommission_item,
         decommission_cancel_signal_result, decommission_item_size, decommission_meta_bucket_options,
-        decommission_start_pool_state, dedup_indices, default_decommission_bucket_concurrency, ensure_activation_locks_held,
+        decommission_start_pool_state, dedup_indices, default_decommission_bucket_concurrency,
         ensure_decommission_cancel_allowed, ensure_decommission_clear_allowed, ensure_decommission_listing_disks_available,
         ensure_decommission_not_rebalancing, ensure_decommission_start_allowed, ensure_decommission_start_keeps_active_pool,
         ensure_decommission_start_local_leader, ensure_decommission_start_pool_states,
@@ -5492,16 +5488,6 @@ mod pools_tests {
         );
     }
 
-    #[test]
-    fn test_activation_fence_rejects_either_lost_guard_before_commit() {
-        assert!(ensure_activation_locks_held(false, false).is_ok());
-        for (pool_lost, rebalance_lost) in [(true, false), (false, true), (true, true)] {
-            let err = ensure_activation_locks_held(pool_lost, rebalance_lost)
-                .expect_err("either lost activation guard must fence the commit");
-            assert!(err.to_string().contains("activation lock lost"));
-        }
-    }
-
     #[test]
     fn test_apply_decommission_status_space_info_adds_idle_pool_usage() {
         let status = apply_decommission_status_space_info(

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

@@ -907,6 +907,9 @@ impl ECStore {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::disk::{DiskAPI as _, RUSTFS_META_BUCKET};
+    use crate::object_api::NamespaceLockFence;
+    use crate::set_disk::ops::object::hermetic_set_disks_support::hermetic_set_disks_isolated;
 
     #[test]
     fn rebalance_activation_rejects_persisted_decommission_despite_idle_local_snapshot() {
@@ -933,6 +936,60 @@ mod tests {
         assert!(ensure_rebalance_activation_pool_meta_allowed(&PoolMeta::default()).is_ok());
     }
 
+    #[tokio::test]
+    async fn rebalance_merge_save_does_not_commit_after_namespace_fence_loss() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_isolated(4).await;
+        for disk in &disk_stores {
+            disk.make_volume(RUSTFS_META_BUCKET)
+                .await
+                .expect("metadata volume should be created");
+        }
+
+        let persisted = RebalanceMeta {
+            id: "rebalance-a".to_string(),
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                bytes: 1,
+                info: RebalanceInfo {
+                    status: RebalStatus::Started,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+        persisted
+            .save(set_disks.clone())
+            .await
+            .expect("baseline rebalance metadata should be saved");
+
+        let mut update = persisted.clone();
+        update.pool_stats[0].bytes = 999;
+        let err = merge_and_save_rebalance_meta_no_lock(
+            set_disks.clone(),
+            &update,
+            "lost namespace fence test",
+            ObjectOptions {
+                no_lock: true,
+                namespace_lock_fence: Some(NamespaceLockFence::lost_for_test()),
+                ..Default::default()
+            },
+            None,
+            Some(persisted.id.as_str()),
+        )
+        .await
+        .expect_err("lost namespace fence must reject the metadata commit");
+        assert!(matches!(err, Error::NamespaceLockQuorumUnavailable { .. }));
+
+        let mut after = RebalanceMeta::new();
+        after
+            .load(set_disks)
+            .await
+            .expect("baseline rebalance metadata should remain readable");
+        assert_eq!(after.id, persisted.id);
+        assert_eq!(after.pool_stats[0].bytes, 1);
+    }
+
     #[test]
     fn pool_rebalance_status_ignores_non_participating_pool_state() {
         let meta = RebalanceMeta {

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

@@ -7613,7 +7613,7 @@ mod object_encryption_resolver_wiring_tests {
 }
 
 #[cfg(test)]
-pub(in crate::set_disk::ops) mod hermetic_set_disks_support {
+pub(crate) mod hermetic_set_disks_support {
     //! Shared hermetic `SetDisks` construction for the ops tests below: the
     //! `SetDisks` under test is built directly on formatted local disks (same
     //! pattern as the `ops/locking.rs` tests) so the tests stay hermetic — no
@@ -7678,9 +7678,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(
+    pub(crate) async fn hermetic_set_disks_isolated(disk_count: usize) -> (Vec<TempDir>, Vec<DiskStore>, Arc<SetDisks>) {
-        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
     }