From b8e9170cb0e5b4d078425f4541a2b3838224d79b Mon Sep 17 00:00:00 2001
From: overtrue <anzhengchao@gmail.com>
Date: Sat, 22 Aug 2026 09:25:08 +0800
Subject: [PATCH] test(ecstore): fix activation fence synchronization

---
 crates/ecstore/src/core/pools.rs              | 21 ++++++++++++++++---
 .../ecstore/src/services/rebalance/control.rs | 20 ++++++++++++------
 2 files changed, 32 insertions(+), 9 deletions(-)

diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs
index 67370594a..284e28aa3 100644
--- a/crates/ecstore/src/core/pools.rs
+++ b/crates/ecstore/src/core/pools.rs
@@ -2876,8 +2876,6 @@ impl ECStore {
             .first()
             .cloned()
             .ok_or_else(|| Error::other("decommission start rebalance metadata load failed: no storage pools available"))?;
-        #[cfg(test)]
-        observe_pool_activation_start_attempt(PoolActivationStartKind::Decommission);
         let activation_fence = acquire_pool_rebalance_activation_locks(rebalance_pool.clone()).await?;
 
         let mut rebalance_meta = RebalanceMeta::new();
@@ -4420,6 +4418,8 @@ impl ECStore {
         let indices = dedup_indices(&indices);
         validate_start_decommission_request(&indices, self.single_pool())?;
 
+        #[cfg(test)]
+        observe_pool_activation_start_attempt(PoolActivationStartKind::Decommission);
         self.ensure_decommission_rebalance_idle_after_refresh().await?;
         #[cfg(test)]
         let endpoints = self.instance_endpoints().unwrap_or_else(|| self.endpoints());
@@ -4775,7 +4775,22 @@ mod tests {
         let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await;
         let barrier = PoolActivationDurableSaveBarrier::install(&store.pools[0]);
         let start_store = Arc::clone(&store);
-        let start_task = tokio::spawn(async move { start_store.start_decommission(vec![0]).await });
+        let start_task = tokio::spawn(async move {
+            start_store
+                .save_current_pool_meta_for_decommission_start(
+                    &[0],
+                    vec![(
+                        0,
+                        PoolSpaceInfo {
+                            free: 50,
+                            total: 100,
+                            used: 50,
+                        },
+                    )],
+                    Vec::new(),
+                )
+                .await
+        });
 
         barrier.wait_until_paused().await;
         let mut replica_disks = Vec::new();
diff --git a/crates/ecstore/src/services/rebalance/control.rs b/crates/ecstore/src/services/rebalance/control.rs
index fdb24a1da..70947645e 100644
--- a/crates/ecstore/src/services/rebalance/control.rs
+++ b/crates/ecstore/src/services/rebalance/control.rs
@@ -1192,6 +1192,17 @@ mod tests {
     use crate::object_api::NamespaceLockFence;
     use crate::set_disk::{PutObjectCommitBarrier, PutObjectCommitPause, hermetic_set_disks_isolated};
 
+    #[tokio::test]
+    async fn rebalance_stop_wait_probe_matches_run_id() {
+        let probe = RebalanceStopWaitProbe::install("rebalance-stop-current");
+
+        observe_rebalance_stop_wait_attempt(Some("rebalance-stop-stale"));
+        assert!(!probe.state.attempted.load(std::sync::atomic::Ordering::Acquire));
+
+        observe_rebalance_stop_wait_attempt(Some("rebalance-stop-current"));
+        probe.wait_until_attempted().await;
+    }
+
     #[tokio::test]
     async fn cancel_rebalance_admission_is_id_checked_and_idempotent() {
         let rebalance_id = "rebalance-admission-current";
@@ -1340,10 +1351,7 @@ mod tests {
         assert!(!admitted_cancel.is_cancelled());
         drop(local_meta);
 
-        store
-            .cancel_rebalance_admission_for_id(rebalance_id)
-            .await
-            .expect("the installed token should cancel the admitted worker");
+        admitted_cancel.cancel();
         assert!(admitted_cancel.is_cancelled());
 
         let mut persisted = RebalanceMeta::new();
@@ -1401,7 +1409,7 @@ mod tests {
             decommission_task = tokio::spawn(async move { store.start_decommission(vec![0]).await });
             tokio::time::timeout(std::time::Duration::from_secs(15), probe.wait_until_attempted())
                 .await
-                .expect("real decommission start should reach activation lock acquisition");
+                .expect("real decommission start should reach the activation path");
         } else {
             let store = Arc::clone(&decommission_store);
             decommission_task = tokio::spawn(async move { store.start_decommission(vec![0]).await });
@@ -1412,7 +1420,7 @@ mod tests {
                 tokio::spawn(async move { store.init_rebalance_start(vec!["bucket".to_string()]).await.map(|_| ()) });
             tokio::time::timeout(std::time::Duration::from_secs(15), probe.wait_until_attempted())
                 .await
-                .expect("real rebalance start should reach activation lock acquisition");
+                .expect("real rebalance start should reach the activation path");
         }
 
         let mut observed_rebalance_result = None;