fix(ecstore): fence rebalance and decommission activation

crates/ecstore/src/core/pools.rs

@@ -840,7 +840,7 @@ fn resolve_start_decommission_pool_meta_reload_result(result: Result<()>) -> Res
     resolve_decommission_pool_meta_reload_result(result, "start_decommission")
 }
 
-fn decommission_rebalance_meta_lock_error(err: rustfs_lock::LockError) -> Error {
+fn activation_rebalance_meta_lock_error(err: rustfs_lock::LockError) -> Error {
     match err {
         rustfs_lock::LockError::QuorumNotReached { required, achieved } => Error::NamespaceLockQuorumUnavailable {
             mode: "write",
@@ -850,12 +850,12 @@ fn decommission_rebalance_meta_lock_error(err: rustfs_lock::LockError) -> Error
             achieved,
         },
         other => Error::other(format!(
-            "failed to acquire rebalance metadata write lock before decommission start on {RUSTFS_META_BUCKET}/{REBAL_META_NAME}: {other}"
+            "failed to acquire rebalance activation lock on {RUSTFS_META_BUCKET}/{REBAL_META_NAME}: {other}"
         )),
     }
 }
 
-fn decommission_pool_meta_lock_error(err: rustfs_lock::LockError) -> Error {
+fn activation_pool_meta_lock_error(err: rustfs_lock::LockError) -> Error {
     match err {
         rustfs_lock::LockError::QuorumNotReached { required, achieved } => Error::NamespaceLockQuorumUnavailable {
             mode: "write",
@@ -865,11 +865,35 @@ fn decommission_pool_meta_lock_error(err: rustfs_lock::LockError) -> Error {
             achieved,
         },
         other => Error::other(format!(
-            "failed to acquire pool metadata write lock before decommission start on {RUSTFS_META_BUCKET}/{POOL_META_NAME}: {other}"
+            "failed to acquire pool activation lock on {RUSTFS_META_BUCKET}/{POOL_META_NAME}: {other}"
         )),
     }
 }
 
+pub(crate) async fn acquire_pool_rebalance_activation_locks<S>(
+    pool: Arc<S>,
+) -> Result<(rustfs_lock::NamespaceLockGuard, rustfs_lock::NamespaceLockGuard)>
+where
+    S: crate::storage_api_contracts::namespace::NamespaceLocking<
+            Error = Error,
+            NamespaceLock = rustfs_lock::NamespaceLockWrapper,
+        >,
+{
+    // Activation lock order is always pool.bin -> rebalance.bin.
+    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
+        .map_err(activation_pool_meta_lock_error)?;
+    let rebalance_meta_lock = pool.new_ns_lock(RUSTFS_META_BUCKET, REBAL_META_NAME).await?;
+    let rebalance_meta_guard = rebalance_meta_lock
+        .get_write_lock(get_lock_acquire_timeout())
+        .await
+        .map_err(activation_rebalance_meta_lock_error)?;
+
+    Ok((pool_meta_guard, rebalance_meta_guard))
+}
+
 fn rollback_decommission_pool_meta(pool_meta: &mut PoolMeta, previous_pool_meta: PoolMeta) {
     *pool_meta = previous_pool_meta;
 }
@@ -1675,7 +1699,7 @@ impl PoolMeta {
         self.load_no_lock(pool).await
     }
 
-    async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()>
+    pub(crate) async fn load_no_lock<S>(&mut self, pool: Arc<S>) -> Result<()>
     where
         S: EcstoreObjectIO,
     {
@@ -2501,16 +2525,7 @@ impl ECStore {
             .first()
             .cloned()
             .ok_or_else(|| Error::other("decommission start rebalance metadata load failed: no storage pools available"))?;
-        let pool_meta_lock = rebalance_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
-            .map_err(decommission_pool_meta_lock_error)?;
-        let ns_lock = rebalance_pool.new_ns_lock(RUSTFS_META_BUCKET, REBAL_META_NAME).await?;
-        let _guard = ns_lock
-            .get_write_lock(get_lock_acquire_timeout())
-            .await
-            .map_err(decommission_rebalance_meta_lock_error)?;
+        let (_pool_meta_guard, _rebalance_meta_guard) = acquire_pool_rebalance_activation_locks(rebalance_pool.clone()).await?;
 
         let mut rebalance_meta = RebalanceMeta::new();
         match rebalance_meta
@@ -5265,8 +5280,9 @@ pub(crate) fn fallback_free_capacity_dedup(disks: &[rustfs_madmin::Disk]) -> usi
 mod pools_tests {
     use super::{
         DECOMMISSION_PROGRESS_SAVE_INTERVAL, DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD, DecomBucketInfo,
-        DecommissionStartPoolState, DecommissionTerminalState, ListCallback, PoolDecommissionInfo, PoolMeta, PoolSpaceInfo,
-        PoolStatus, apply_decommission_status_space_info, bind_decommission_cancelers, bind_missing_decommission_cancelers,
+        DecommissionStartPoolState, DecommissionTerminalState, ListCallback, POOL_META_NAME, PoolDecommissionInfo, PoolMeta,
+        PoolSpaceInfo, PoolStatus, REBAL_META_NAME, acquire_pool_rebalance_activation_locks,
+        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,
@@ -5304,15 +5320,45 @@ mod pools_tests {
     use crate::services::rebalance::{RebalStatus, RebalanceInfo, RebalanceMeta, RebalanceStats};
     use rustfs_filemeta::{FileInfo, FileInfoVersions, MetaCacheEntry, ObjectPartInfo};
     use rustfs_rio::Index;
+    use std::future::Future;
     use std::sync::{
-        Arc,
+        Arc, Mutex,
         atomic::{AtomicBool, AtomicUsize, Ordering},
     };
+    use std::task::{Context, Poll};
     use std::time::Duration as StdDuration;
     use time::{Duration, OffsetDateTime};
     use tokio::sync::Semaphore;
     use tokio_util::sync::CancellationToken;
 
+    #[derive(Debug)]
+    struct ActivationLockRecorder {
+        lock_manager: Arc<rustfs_lock::GlobalLockManager>,
+        owner: &'static str,
+        resources: Mutex<Vec<String>>,
+    }
+
+    #[async_trait::async_trait]
+    impl crate::storage_api_contracts::namespace::NamespaceLocking for ActivationLockRecorder {
+        type Error = Error;
+        type NamespaceLock = rustfs_lock::NamespaceLockWrapper;
+
+        async fn new_ns_lock(&self, bucket: &str, object: &str) -> crate::error::Result<Self::NamespaceLock> {
+            self.resources
+                .lock()
+                .expect("activation lock recorder should not be poisoned")
+                .push(object.to_string());
+            Ok(rustfs_lock::NamespaceLockWrapper::new(
+                rustfs_lock::NamespaceLock::with_local_manager(
+                    "activation-lock-test".to_string(),
+                    Arc::clone(&self.lock_manager),
+                ),
+                rustfs_lock::ObjectKey::new(bucket, object),
+                self.owner.to_string(),
+            ))
+        }
+    }
+
     fn noop_decommission_list_callback() -> ListCallback {
         Arc::new(|_| Box::pin(async {}))
     }
@@ -5343,6 +5389,48 @@ mod pools_tests {
         }
     }
 
+    #[tokio::test]
+    async fn test_activation_fence_uses_one_lock_order_and_serializes_callers() {
+        let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+        let first = Arc::new(ActivationLockRecorder {
+            lock_manager: Arc::clone(&manager),
+            owner: "first",
+            resources: Mutex::new(Vec::new()),
+        });
+        let second = Arc::new(ActivationLockRecorder {
+            lock_manager: manager,
+            owner: "second",
+            resources: Mutex::new(Vec::new()),
+        });
+
+        let first_guards = acquire_pool_rebalance_activation_locks(first.clone())
+            .await
+            .expect("first activation should acquire both locks");
+        assert_eq!(
+            *first
+                .resources
+                .lock()
+                .expect("activation lock recorder should not be poisoned"),
+            vec![POOL_META_NAME.to_string(), REBAL_META_NAME.to_string()]
+        );
+
+        let mut second_acquire = Box::pin(acquire_pool_rebalance_activation_locks(second.clone()));
+        let mut context = Context::from_waker(futures::task::noop_waker_ref());
+        assert!(matches!(second_acquire.as_mut().poll(&mut context), Poll::Pending));
+
+        drop(first_guards);
+        second_acquire
+            .await
+            .expect("second activation should acquire both locks after the first releases them");
+        assert_eq!(
+            *second
+                .resources
+                .lock()
+                .expect("activation lock recorder should not be poisoned"),
+            vec![POOL_META_NAME.to_string(), REBAL_META_NAME.to_string()]
+        );
+    }
+
     #[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

@@ -16,6 +16,7 @@ use super::{
     RebalStatus, RebalanceInfo, RebalanceMeta, RebalanceStats, RebalanceStopPropagationRecord,
     encode_rebalance_stop_propagation_record,
 };
+use crate::core::pools::{PoolMeta, acquire_pool_rebalance_activation_locks, pool_meta_has_active_decommission};
 use crate::error::{Error, Result};
 use crate::object_api::ObjectOptions;
 use crate::set_disk::get_lock_acquire_timeout;
@@ -29,6 +30,38 @@ use time::OffsetDateTime;
 use tracing::{debug, info};
 use uuid::Uuid;
 
+fn ensure_rebalance_activation_pool_meta_allowed(meta: &PoolMeta) -> Result<()> {
+    if pool_meta_has_active_decommission(meta) {
+        return Err(Error::DecommissionAlreadyRunning);
+    }
+
+    Ok(())
+}
+
+async fn merge_and_save_rebalance_meta_no_lock<S>(pool: Arc<S>, local_snapshot: &RebalanceMeta, stage: &str) -> Result<()>
+where
+    S: EcstoreObjectIO,
+{
+    let opts = ObjectOptions {
+        no_lock: true,
+        ..Default::default()
+    };
+    let mut merged = RebalanceMeta::new();
+    match merged.load_with_opts(pool.clone(), opts.clone()).await {
+        Ok(()) => {
+            if merge_rebalance_meta(&mut merged, local_snapshot) == RebalanceMetaMergeOutcome::RejectedActiveConflict {
+                return Err(Error::RebalanceAlreadyRunning);
+            }
+        }
+        Err(Error::ConfigNotFound) => {
+            merged = local_snapshot.clone();
+        }
+        Err(err) => return Err(Error::other(format!("rebalance meta load before save failed during {stage}: {err}"))),
+    }
+
+    merged.save_with_opts(pool, opts).await
+}
+
 pub(super) fn validate_rebalance_disk_stats_coverage(disk_stats: &[DiskStat]) -> Result<()> {
     for (idx, disk_stat) in disk_stats.iter().enumerate() {
         if disk_stat.total_space == 0 {
@@ -90,24 +123,53 @@ impl ECStore {
             .await
             .map_err(rebalance_meta_lock_error)?;
 
-        let opts = ObjectOptions {
-            no_lock: true,
-            ..Default::default()
-        };
-        let mut merged = RebalanceMeta::new();
-        match merged.load_with_opts(pool.clone(), opts.clone()).await {
-            Ok(()) => {
-                if merge_rebalance_meta(&mut merged, local_snapshot) == RebalanceMetaMergeOutcome::RejectedActiveConflict {
-                    return Err(Error::RebalanceAlreadyRunning);
-                }
-            }
-            Err(Error::ConfigNotFound) => {
-                merged = local_snapshot.clone();
-            }
-            Err(err) => return Err(Error::other(format!("rebalance meta load before save failed during {stage}: {err}"))),
+        merge_and_save_rebalance_meta_no_lock(pool, local_snapshot, stage).await
+    }
+
+    async fn save_rebalance_activation_meta_with_merge<S>(
+        &self,
+        pool: Arc<S>,
+        local_snapshot: &RebalanceMeta,
+        stage: &str,
+    ) -> Result<()>
+    where
+        S: EcstoreObjectIO + StorageNamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
+    {
+        let (_pool_meta_guard, _rebalance_meta_guard) = acquire_pool_rebalance_activation_locks(pool.clone()).await?;
+        let mut pool_meta = PoolMeta::default();
+        pool_meta.load_no_lock(pool.clone()).await?;
+        ensure_rebalance_activation_pool_meta_allowed(&pool_meta)?;
+
+        merge_and_save_rebalance_meta_no_lock(pool, local_snapshot, stage).await
+    }
+
+    pub(super) async fn fence_rebalance_worker_activation<S>(&self, pool: Arc<S>, expected_id: &str) -> Result<bool>
+    where
+        S: EcstoreObjectIO + StorageNamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
+    {
+        let (_pool_meta_guard, _rebalance_meta_guard) = acquire_pool_rebalance_activation_locks(pool.clone()).await?;
+        let mut pool_meta = PoolMeta::default();
+        pool_meta.load_no_lock(pool.clone()).await?;
+        ensure_rebalance_activation_pool_meta_allowed(&pool_meta)?;
+
+        let mut persisted = RebalanceMeta::new();
+        persisted
+            .load_with_opts(
+                pool,
+                ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await?;
+        if persisted.id != expected_id {
+            return Err(Error::other(format!(
+                "rebalance metadata changed before worker activation: expected {expected_id}, found {}",
+                persisted.id
+            )));
         }
 
-        merged.save_with_opts(pool, opts).await
+        Ok(is_rebalance_conflicting_with_decommission(&persisted))
     }
 
     #[tracing::instrument(skip_all)]
@@ -301,7 +363,8 @@ impl ECStore {
 
         let pool = clone_first_arc(&self.pools, "init_rebalance_meta: no pools available")?;
         resolve_rebalance_meta_save_result(
-            self.save_rebalance_meta_with_merge(pool, &meta, "init_rebalance_meta").await,
+            self.save_rebalance_activation_meta_with_merge(pool, &meta, "init_rebalance_meta")
+                .await,
             "init_rebalance_meta",
         )?;
 
@@ -379,7 +442,7 @@ impl ECStore {
             }
         }
 
-        self.start_rebalance().await
+        self.start_rebalance_under_gate().await
     }
 
     pub async fn rollback_rebalance_start_for_id(self: &Arc<Self>, expected_id: Option<&str>, start_error: String) -> Result<()> {
@@ -620,6 +683,31 @@ impl ECStore {
 mod tests {
     use super::*;
 
+    #[test]
+    fn rebalance_activation_rejects_persisted_decommission_despite_idle_local_snapshot() {
+        let persisted = PoolMeta {
+            pools: vec![crate::core::pools::PoolStatus {
+                id: 0,
+                cmd_line: "pool-0".to_string(),
+                last_update: OffsetDateTime::UNIX_EPOCH,
+                decommission: Some(crate::core::pools::PoolDecommissionInfo {
+                    start_time: Some(OffsetDateTime::UNIX_EPOCH),
+                    ..Default::default()
+                }),
+            }],
+            ..Default::default()
+        };
+
+        let err = ensure_rebalance_activation_pool_meta_allowed(&persisted)
+            .expect_err("persisted decommission must block a stale rebalance admission");
+        assert!(matches!(err, Error::DecommissionAlreadyRunning));
+    }
+
+    #[test]
+    fn rebalance_activation_allows_persisted_idle_pool_meta() {
+        assert!(ensure_rebalance_activation_pool_meta_allowed(&PoolMeta::default()).is_ok());
+    }
+
     #[test]
     fn pool_rebalance_status_ignores_non_participating_pool_state() {
         let meta = RebalanceMeta {

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

@@ -42,6 +42,11 @@ pub(super) fn source_cleanup_defer_attempt(deferred_attempts: &mut HashMap<Strin
 impl ECStore {
     #[tracing::instrument(skip_all)]
     pub async fn start_rebalance(self: &Arc<Self>) -> Result<()> {
+        let _start_guard = self.start_gate.lock().await;
+        self.start_rebalance_under_gate().await
+    }
+
+    pub(super) async fn start_rebalance_under_gate(self: &Arc<Self>) -> Result<()> {
         info!(
             event = EVENT_REBALANCE_STATE,
             component = LOG_COMPONENT_ECSTORE,
@@ -49,8 +54,16 @@ impl ECStore {
             state = "starting",
             "Starting rebalance"
         );
+        let expected_id = {
+            let rebalance_meta = self.rebalance_meta.read().await;
+            rebalance_meta.as_ref().ok_or(Error::ConfigNotFound)?.id.clone()
+        };
+        let pool = clone_first_arc(self.pools.as_slice(), "start_rebalance: no pools available")?;
+        if !self.fence_rebalance_worker_activation(pool, &expected_id).await? {
+            return Ok(());
+        }
+
         let decommission_running = self.is_decommission_running().await;
-        // let rebalance_meta = self.rebalance_meta.read().await;
 
         let cancel_tx = CancellationToken::new();
         let rx = cancel_tx.clone();