fix(ecstore): fence stale rebalance workers

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

@@ -46,7 +46,44 @@ pub(super) enum RebalanceWorkerActivationFence {
 }
 
 pub(super) struct RebalanceRunGuard {
-    _guard: tokio::sync::OwnedRwLockReadGuard<()>,
+    _local_guard: tokio::sync::OwnedRwLockReadGuard<()>,
+    persisted_guard: rustfs_lock::NamespaceLockGuard,
+}
+
+impl RebalanceRunGuard {
+    pub(super) fn ensure_held(&self, stage: &str) -> Result<()> {
+        if self.persisted_guard.is_lock_lost() {
+            return Err(Error::other(format!("rebalance distributed run fence lost during {stage}")));
+        }
+        Ok(())
+    }
+}
+
+async fn acquire_persisted_rebalance_run_guard<S>(
+    pool: Arc<S>,
+    expected_id: &str,
+    stage: &str,
+) -> Result<rustfs_lock::NamespaceLockGuard>
+where
+    S: EcstoreObjectIO + StorageNamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
+{
+    let ns_lock = pool.new_ns_lock(crate::disk::RUSTFS_META_BUCKET, REBAL_META_NAME).await?;
+    let guard = ns_lock
+        .get_read_lock(get_lock_acquire_timeout())
+        .await
+        .map_err(|err| rebalance_meta_lock_error(err, "read"))?;
+    let mut opts = ObjectOptions {
+        no_lock: true,
+        ..Default::default()
+    };
+    opts.add_namespace_lock_guard(&guard);
+    let mut persisted = RebalanceMeta::new();
+    persisted.load_with_opts(pool, opts).await?;
+    ensure_rebalance_worker_active(Some(&persisted), expected_id, stage)?;
+    if guard.is_lock_lost() {
+        return Err(Error::other(format!("rebalance distributed run fence lost during {stage}")));
+    }
+    Ok(guard)
 }
 
 async fn merge_and_save_rebalance_meta_no_lock<S>(
@@ -180,6 +217,7 @@ impl ECStore {
     }
 
     pub(super) async fn rebalance_run_guard(&self, expected_id: &str, stage: &str) -> Result<RebalanceRunGuard> {
+        // Runtime fence order is activation_gate -> rebalance.bin.
         let activation_gate = {
             let meta = self.rebalance_meta.read().await;
             ensure_rebalance_worker_active(meta.as_ref(), expected_id, stage)?;
@@ -203,7 +241,12 @@ impl ECStore {
             )));
         }
         drop(meta);
-        Ok(RebalanceRunGuard { _guard: guard })
+        let pool = clone_first_arc(self.pools.as_slice(), "rebalance run fence: no pools available")?;
+        let persisted_guard = acquire_persisted_rebalance_run_guard(pool, expected_id, stage).await?;
+        Ok(RebalanceRunGuard {
+            _local_guard: guard,
+            persisted_guard,
+        })
     }
 
     pub(super) async fn save_rebalance_meta_with_merge<S>(
@@ -247,7 +290,7 @@ impl ECStore {
         let guard = ns_lock
             .get_write_lock(get_lock_acquire_timeout())
             .await
-            .map_err(rebalance_meta_lock_error)?;
+            .map_err(|err| rebalance_meta_lock_error(err, "write"))?;
         let mut opts = ObjectOptions {
             no_lock: true,
             ..Default::default()
@@ -984,6 +1027,43 @@ mod tests {
         assert_eq!(after.pool_stats[0].bytes, 1);
     }
 
+    #[tokio::test]
+    async fn persisted_rebalance_run_fence_rejects_stale_active_node_after_remote_stop() {
+        let (_temp_dirs, _disk_stores, set_disks) = hermetic_set_disks_isolated(4).await;
+        let active = RebalanceMeta {
+            id: "rebalance-a".to_string(),
+            cancel: Some(tokio_util::sync::CancellationToken::new()),
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                info: RebalanceInfo {
+                    status: RebalStatus::Started,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+        active
+            .save(set_disks.clone())
+            .await
+            .expect("active rebalance metadata should be saved");
+        ensure_rebalance_worker_active(Some(&active), active.id.as_str(), "stale node precondition")
+            .expect("the stale node snapshot should still look active locally");
+
+        let mut stopped = active.clone();
+        stopped.stopped_at = Some(OffsetDateTime::now_utc());
+        stopped.pool_stats[0].info.stopping = true;
+        stopped
+            .save(set_disks.clone())
+            .await
+            .expect("remote stop should replace persisted metadata");
+
+        let err = acquire_persisted_rebalance_run_guard(set_disks, active.id.as_str(), "cross-node stale snapshot")
+            .await
+            .expect_err("persisted stop must fence a node that missed stop propagation");
+        assert!(err.to_string().contains("inactive rebalance worker rejected"));
+    }
+
     #[test]
     fn pool_rebalance_status_ignores_non_participating_pool_state() {
         let meta = RebalanceMeta {

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

@@ -56,6 +56,7 @@ impl ECStore {
         // Persisted stats can complete a pool on restart, so source cleanup must resolve first.
         let run_guard = self.rebalance_run_guard(expected_id, "rebalance source cleanup").await?;
         let cleanup_result = cleanup.await;
+        run_guard.ensure_held("rebalance source cleanup")?;
         drop(run_guard);
         let cleanup_result = resolve_rebalance_entry_cleanup_delete_result(cleanup_result, bucket, object);
         let RebalanceEntryCleanupResult::Completed { warning } = cleanup_result else {
@@ -168,7 +169,7 @@ impl ECStore {
             let run_guard = self
                 .rebalance_run_guard(rebalance_id.as_ref(), "rebalance lifecycle mutation")
                 .await?;
-            let expired_by_lifecycle = crate::core::pools::should_skip_lifecycle_for_data_movement(
+            let lifecycle_result = crate::core::pools::should_skip_lifecycle_for_data_movement(
                 self.clone(),
                 &bucket,
                 version,
@@ -177,8 +178,9 @@ impl ECStore {
                 true,
                 &crate::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc::Rebal,
             )
-            .await?;
-            drop(run_guard);
+            .await;
+            run_guard.ensure_held("rebalance lifecycle mutation")?;
+            let expired_by_lifecycle = lifecycle_result?;
             if expired_by_lifecycle {
                 expired += 1;
                 // The lifecycle expiry above physically deleted this version from the source set.
@@ -196,6 +198,7 @@ impl ECStore {
                     reason = "expired_by_lifecycle",
                     "Skipped rebalance version"
                 );
+                drop(run_guard);
                 continue;
             }
 
@@ -213,6 +216,7 @@ impl ECStore {
                     reason = "last_delete_marker_without_replication",
                     "Skipped rebalance version"
                 );
+                drop(run_guard);
                 continue;
             }
 
@@ -222,7 +226,6 @@ impl ECStore {
                 let store = self.clone();
                 async move {
                     store
-                        .clone()
                         .rebalance_object(src_pool_idx, bucket, rd, expected_bucket_incarnation_id)
                         .await
                 }
@@ -233,9 +236,6 @@ impl ECStore {
                 let store = self.clone();
                 async move { store.delete_object(&bucket, &object, opts).await }
             };
-            let run_guard = self
-                .rebalance_run_guard(rebalance_id.as_ref(), "rebalance version migration")
-                .await?;
             let result = migrate_entry_version(
                 &RebalanceMigrationBackend::new(set.as_ref(), self.as_ref()),
                 bucket.clone(),
@@ -249,6 +249,7 @@ impl ECStore {
                 &mut delete_marker,
             )
             .await;
+            run_guard.ensure_held("rebalance version migration")?;
             drop(run_guard);
 
             if result.ignored {
@@ -678,31 +679,21 @@ mod tests {
 
     #[tokio::test]
     async fn rebalance_stats_wait_for_source_cleanup_result() {
-        let endpoint_pools: crate::layout::endpoints::EndpointServerPools = Vec::new().into();
-        let store = Arc::new(ECStore {
-            id: uuid::Uuid::new_v4(),
-            disk_map: std::collections::HashMap::new(),
-            pools: Vec::new(),
-            peer_sys: crate::cluster::rpc::S3PeerSys::new(&endpoint_pools),
-            pool_meta: tokio::sync::RwLock::new(crate::core::pools::PoolMeta::default()),
-            rebalance_meta: tokio::sync::RwLock::new(Some(RebalanceMeta {
-                pool_stats: vec![RebalanceStats {
-                    participating: true,
-                    info: RebalanceInfo {
-                        start_time: Some(OffsetDateTime::now_utc()),
-                        status: RebalStatus::Started,
-                        ..Default::default()
-                    },
+        let rebalance_id = "rebalance-test";
+        let (_temp_dirs, store) = crate::services::rebalance::test_store_with_persisted_rebalance_meta(RebalanceMeta {
+            id: rebalance_id.to_string(),
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                info: RebalanceInfo {
+                    start_time: Some(OffsetDateTime::now_utc()),
+                    status: RebalStatus::Started,
                     ..Default::default()
-                }],
+                },
                 ..Default::default()
-            })),
-            decommission_cancelers: tokio::sync::RwLock::new(Vec::new()),
-            start_gate: tokio::sync::Mutex::new(()),
-            pool_meta_save_gate: tokio::sync::Mutex::new(()),
-            ctx: crate::runtime::instance::bootstrap_ctx(),
-            bucket_fence_registry: std::sync::Arc::default(),
-        });
+            }],
+            ..Default::default()
+        })
+        .await;
         let mut version = FileInfo::new("object.bin", 4, 2);
         version.name = "object.bin".to_string();
         version.size = 128;
@@ -713,7 +704,7 @@ mod tests {
         let finish_store = Arc::clone(&store);
         let finish = tokio::spawn(async move {
             finish_store
-                .finish_rebalance_entry_after_cleanup(0, "bucket", "object.bin", &[&version], "", async move {
+                .finish_rebalance_entry_after_cleanup(0, "bucket", "object.bin", &[&version], rebalance_id, async move {
                     cleanup_released.await.expect("cleanup release sender should remain alive");
                     Ok(ObjectInfo::default())
                 })
@@ -760,7 +751,7 @@ mod tests {
             meta.as_mut().expect("rebalance metadata should exist").pool_stats[0].bytes = 0;
         }
         let warning_result = store
-            .finish_rebalance_entry_after_cleanup(0, "bucket", "object.bin", &[&warning_version], "", async {
+            .finish_rebalance_entry_after_cleanup(0, "bucket", "object.bin", &[&warning_version], rebalance_id, async {
                 Err(Error::SlowDown.into())
             })
             .await
@@ -777,7 +768,7 @@ mod tests {
             meta.as_mut().expect("rebalance metadata should exist").pool_stats[0].bytes = 0;
         }
         let deferred = store
-            .finish_rebalance_entry_after_cleanup(0, "bucket", "object.bin", &[&warning_version], "", async {
+            .finish_rebalance_entry_after_cleanup(0, "bucket", "object.bin", &[&warning_version], rebalance_id, async {
                 Err(data_movement::SourceCleanupError::SourceChanged)
             })
             .await

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

@@ -53,5 +53,31 @@ pub use types::{
 };
 use types::{RebalanceBucketConfigs, RebalanceBucketOutcome, RebalanceEntryOutcome};
 
+#[cfg(test)]
+pub(crate) async fn test_store_with_persisted_rebalance_meta(
+    meta: RebalanceMeta,
+) -> (Vec<tempfile::TempDir>, std::sync::Arc<crate::store::ECStore>) {
+    let ctx = std::sync::Arc::new(crate::runtime::instance::InstanceContext::new());
+    let (temp_dirs, pool) = crate::core::sets::make_local_two_set_sets_with_ctx(ctx.clone()).await;
+    meta.save(pool.clone())
+        .await
+        .expect("rebalance test metadata should be persisted");
+    let endpoint_pools: crate::layout::endpoints::EndpointServerPools = vec![pool.endpoints.clone()].into();
+    let store = std::sync::Arc::new(crate::store::ECStore {
+        id: uuid::Uuid::new_v4(),
+        disk_map: std::collections::HashMap::new(),
+        pools: vec![pool],
+        peer_sys: crate::cluster::rpc::S3PeerSys::new_with_instance_ctx(&endpoint_pools, ctx.clone()),
+        pool_meta: tokio::sync::RwLock::new(crate::core::pools::PoolMeta::default()),
+        rebalance_meta: tokio::sync::RwLock::new(Some(meta)),
+        decommission_cancelers: tokio::sync::RwLock::new(vec![None]),
+        start_gate: tokio::sync::Mutex::new(()),
+        pool_meta_save_gate: tokio::sync::Mutex::new(()),
+        ctx,
+        bucket_fence_registry: std::sync::Arc::default(),
+    });
+    (temp_dirs, store)
+}
+
 #[cfg(test)]
 mod rebalance_unit_tests;

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

@@ -33,8 +33,8 @@ use super::migration::{
     rebalance_delete_marker_opts,
 };
 use super::runtime::{
-    RebalanceLocalActivationOutcome, commit_local_rebalance_worker_activation, should_fail_repeated_rebalance_bucket_defer,
-    source_cleanup_defer_attempt,
+    RebalanceLocalActivationOutcome, commit_local_rebalance_worker_activation, resolve_rebalance_pre_spawn_result,
+    should_fail_repeated_rebalance_bucket_defer, source_cleanup_defer_attempt,
 };
 use super::worker::{
     RebalanceEntryCleanupResult, ensure_rebalance_listing_disks_available, is_transient_rebalance_error,
@@ -2736,6 +2736,54 @@ fn test_stopped_activation_state_prevents_worker_token_commit() {
     assert!(meta.cancel.is_none(), "stopped rebalance must not receive a worker token");
 }
 
+#[test]
+fn test_rebalance_start_save_failure_rolls_back_local_worker_token() {
+    let activation_token = tokio_util::sync::CancellationToken::new();
+    let observer = activation_token.clone();
+    let mut meta = RebalanceMeta {
+        id: "rebalance-a".to_string(),
+        pool_stats: vec![RebalanceStats {
+            participating: true,
+            info: RebalanceInfo {
+                status: RebalStatus::Started,
+                ..Default::default()
+            },
+            ..Default::default()
+        }],
+        ..Default::default()
+    };
+    assert_eq!(
+        commit_local_rebalance_worker_activation(&mut meta, "rebalance-a", activation_token.clone())
+            .expect("active metadata should accept the worker token"),
+        RebalanceLocalActivationOutcome::Started
+    );
+
+    let err = resolve_rebalance_pre_spawn_result(
+        Some(&mut meta),
+        "rebalance-a",
+        &activation_token,
+        Err(Error::NamespaceLockQuorumUnavailable {
+            mode: "write",
+            bucket: crate::disk::RUSTFS_META_BUCKET.to_string(),
+            object: super::REBAL_META_NAME.to_string(),
+            required: 3,
+            achieved: 2,
+        }),
+    )
+    .expect_err("metadata save failure must abort local worker activation");
+
+    assert!(matches!(
+        err,
+        Error::NamespaceLockQuorumUnavailable {
+            required: 3,
+            achieved: 2,
+            ..
+        }
+    ));
+    assert!(observer.is_cancelled(), "the failed activation token must be canceled");
+    assert!(meta.cancel.is_none(), "a retry must not see a phantom active worker token");
+}
+
 #[tokio::test]
 async fn test_old_worker_cannot_mutate_replacement_rebalance_state() {
     let meta = RebalanceMeta {
@@ -2809,7 +2857,7 @@ async fn test_stop_waits_for_active_rebalance_migration_guard() {
         }],
         ..Default::default()
     };
-    let store = test_store_with_rebalance_meta(meta);
+    let (_temp_dirs, store) = super::test_store_with_persisted_rebalance_meta(meta).await;
     let run_guard = store
         .rebalance_run_guard("rebalance-a", "rebalance remote-tier migration")
         .await
@@ -2830,7 +2878,7 @@ async fn test_stop_waits_for_active_rebalance_migration_guard() {
 
     drop(run_guard);
     stop.await
-        .expect_err("empty test store should fail only after committing the local stop state");
+        .expect("stop should persist after the side-effect fence is released");
     assert!(
         store
             .rebalance_meta

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

@@ -64,6 +64,36 @@ pub(super) fn commit_local_rebalance_worker_activation(
     Ok(RebalanceLocalActivationOutcome::Started)
 }
 
+pub(super) fn rollback_local_rebalance_worker_activation(
+    meta: Option<&mut super::RebalanceMeta>,
+    expected_id: &str,
+    activation_token: &CancellationToken,
+) -> bool {
+    let Some(meta) = meta else {
+        return false;
+    };
+    if meta.id != expected_id || meta.cancel.as_ref() != Some(activation_token) {
+        return false;
+    }
+    if let Some(cancel) = meta.cancel.take() {
+        cancel.cancel();
+        return true;
+    }
+    false
+}
+
+pub(super) fn resolve_rebalance_pre_spawn_result(
+    meta: Option<&mut super::RebalanceMeta>,
+    expected_id: &str,
+    activation_token: &CancellationToken,
+    result: Result<()>,
+) -> Result<()> {
+    if result.is_err() {
+        rollback_local_rebalance_worker_activation(meta, expected_id, activation_token);
+    }
+    result
+}
+
 impl ECStore {
     #[tracing::instrument(skip_all)]
     pub async fn start_rebalance(self: &Arc<Self>) -> Result<()> {
@@ -84,7 +114,10 @@ impl ECStore {
             Arc::from(rebalance_meta.as_ref().ok_or(Error::ConfigNotFound)?.id.as_str())
         };
         let pool = clone_first_arc(self.pools.as_slice(), "start_rebalance: no pools available")?;
-        let activation_fence = match self.fence_rebalance_worker_activation(pool, expected_id.as_ref()).await? {
+        let activation_fence = match self
+            .fence_rebalance_worker_activation(pool.clone(), expected_id.as_ref())
+            .await?
+        {
             RebalanceWorkerActivationFence::Ready(fence) => fence,
             RebalanceWorkerActivationFence::NotStartedTerminal => return Ok(()),
         };
@@ -129,12 +162,15 @@ impl ECStore {
         drop(activation_fence);
 
         if let Some(meta) = meta_to_save {
-            let pool = clone_first_arc(self.pools.as_slice(), "start_rebalance: no pools available")?;
-            resolve_rebalance_meta_save_result(
+            let save_result = resolve_rebalance_meta_save_result(
                 self.save_rebalance_meta_with_merge(pool, &meta, "start_rebalance complete pools at goal")
                     .await,
                 "start_rebalance complete pools at goal",
-            )?;
+            );
+            if save_result.is_err() {
+                let mut rebalance_meta = self.rebalance_meta.write().await;
+                return resolve_rebalance_pre_spawn_result(rebalance_meta.as_mut(), expected_id.as_ref(), &rx, save_result);
+            }
         }
 
         if activation_outcome != RebalanceLocalActivationOutcome::Started {
@@ -156,6 +192,8 @@ impl ECStore {
         };
 
         if !participants.iter().any(|participating| *participating) {
+            let mut rebalance_meta = self.rebalance_meta.write().await;
+            rollback_local_rebalance_worker_activation(rebalance_meta.as_mut(), expected_id.as_ref(), &rx);
             debug!(
                 event = EVENT_REBALANCE_STATE,
                 component = LOG_COMPONENT_ECSTORE,
@@ -225,6 +263,8 @@ impl ECStore {
         }
 
         if workers_started == 0 {
+            let mut rebalance_meta = self.rebalance_meta.write().await;
+            rollback_local_rebalance_worker_activation(rebalance_meta.as_mut(), expected_id.as_ref(), &rx);
             debug!(
                 event = EVENT_REBALANCE_STATE,
                 component = LOG_COMPONENT_ECSTORE,

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

@@ -100,17 +100,17 @@ pub(super) fn resolve_rebalance_meta_save_result(result: Result<()>, stage: &str
     result.map_err(|err| Error::other(format!("rebalance meta save failed during {stage}: {err}")))
 }
 
-pub(super) fn rebalance_meta_lock_error(err: rustfs_lock::LockError) -> Error {
+pub(super) fn rebalance_meta_lock_error(err: rustfs_lock::LockError, mode: &'static str) -> Error {
     match err {
         rustfs_lock::LockError::QuorumNotReached { required, achieved } => Error::NamespaceLockQuorumUnavailable {
-            mode: "write",
+            mode,
             bucket: crate::disk::RUSTFS_META_BUCKET.to_string(),
             object: REBAL_META_NAME.to_string(),
             required,
             achieved,
         },
         other => Error::other(format!(
-            "failed to acquire rebalance metadata write lock on {}/{}: {other}",
+            "failed to acquire rebalance metadata {mode} lock on {}/{}: {other}",
             crate::disk::RUSTFS_META_BUCKET,
             REBAL_META_NAME
         )),