fix(rebalance): preserve committed activation recovery

crates/ecstore/src/api/mod.rs

@@ -444,6 +444,7 @@ pub mod rebalance {
     #[cfg(feature = "test-util")]
     pub mod test_util {
         pub use crate::services::rebalance::entry::test_util::PausedRebalanceEntryTestFixture;
+        pub use crate::services::rebalance::test_store_with_persisted_rebalance_meta;
     }
 }
 

crates/ecstore/src/core/pools.rs

@@ -28,7 +28,9 @@ use crate::bucket::{
     metadata_sys,
 };
 use crate::cache_value::metacache_set::{ListPathRawOptions, list_path_raw};
-use crate::config::com::{CONFIG_PREFIX, read_config, read_config_no_lock, save_config, save_config_with_opts};
+use crate::config::com::{
+    CONFIG_PREFIX, read_config, read_config_no_lock, save_config, save_config_with_opts, save_config_with_opts_quiet,
+};
 use crate::data_movement;
 use crate::data_movement::backpressure::{self, DataMovementOperation};
 use crate::data_usage::DATA_USAGE_CACHE_NAME;
@@ -1997,9 +1999,10 @@ impl PoolMeta {
 
         // Pool zero is canonical. Once its save succeeds, later writes only
         // replicate committed state and must not reuse the admission fence.
+        // Failed replicas remain repairable by the next full pool metadata save.
         drop(activation_fence);
-        for pool in pools {
-            save_config_with_opts(
+        for (pool_index, pool) in pools.enumerate() {
+            if let Err(err) = save_config_with_opts_quiet(
                 pool,
                 POOL_META_NAME,
                 data.clone(),
@@ -2009,7 +2012,18 @@ impl PoolMeta {
                     ..Default::default()
                 },
             )
-            .await?;
+            .await
+            {
+                warn!(
+                    event = EVENT_DECOMMISSION_STATE,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_POOLS,
+                    pool_index = pool_index + 1,
+                    state = "activation_replica_repair_pending",
+                    error = %err,
+                    "Decommission activation replica repair pending"
+                );
+            }
         }
 
         Ok(())
@@ -4755,6 +4769,65 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn decommission_activation_adopts_canonical_commit_after_replica_failure() {
+        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 });
+
+        barrier.wait_until_paused().await;
+        let mut replica_disks = Vec::new();
+        for set in &store.pools[1].disk_set {
+            let mut disks = set.disks.write().await;
+            let saved = std::mem::take(&mut *disks);
+            *disks = vec![None; saved.len()];
+            replica_disks.push(saved);
+        }
+        barrier.release_after_fence_loss();
+
+        tokio::time::timeout(std::time::Duration::from_secs(30), start_task)
+            .await
+            .expect("decommission activation should finish after its canonical commit")
+            .expect("decommission activation task should not panic")
+            .expect("a replica save failure must not report the committed activation as failed");
+
+        for (set, disks) in store.pools[1].disk_set.iter().zip(replica_disks) {
+            *set.disks.write().await = disks;
+        }
+
+        let local = store.pool_meta.read().await;
+        assert!(pool_meta_has_active_decommission(&local));
+        drop(local);
+
+        let mut canonical = PoolMeta::default();
+        canonical
+            .load_no_lock(store.pools[0].clone())
+            .await
+            .expect("the canonical committed decommission metadata should remain readable");
+        assert!(pool_meta_has_active_decommission(&canonical));
+
+        let mut replica = PoolMeta::default();
+        replica
+            .load_no_lock(store.pools[1].clone())
+            .await
+            .expect("the stale replica metadata should remain readable after disks recover");
+        assert!(!pool_meta_has_active_decommission(&replica));
+
+        let worker_cancel = CancellationToken::new();
+        store
+            .spawn_decommission_routines(Arc::clone(&store), worker_cancel.clone(), vec![0])
+            .await
+            .expect("the committed activation should admit its decommission worker");
+        let admitted_cancel = store.decommission_cancelers.read().await[0]
+            .clone()
+            .expect("the admitted decommission worker should have a cancellation token");
+        assert!(!admitted_cancel.is_cancelled());
+        worker_cancel.cancel();
+        assert!(admitted_cancel.is_cancelled());
+    }
+
     #[test]
     fn ensure_pool_not_left_in_cmdline_after_decommission_allows_active_pool() {
         assert!(ensure_pool_not_left_in_cmdline_after_decommission(0, "http://node{1...4}/disk{1...4}", false).is_ok());

crates/ecstore/src/runtime/sources.rs

@@ -119,8 +119,8 @@ pub(crate) fn endpoint_erasure_set_count() -> Option<usize> {
     endpoint_pools().map(|endpoints| endpoints.es_count())
 }
 
-pub(crate) fn endpoint_pool_is_local(pool_index: usize) -> bool {
-    get_global_endpoints()
+pub(crate) fn endpoint_pool_is_local(endpoints: &EndpointServerPools, pool_index: usize) -> bool {
+    endpoints
         .as_ref()
         .get(pool_index)
         .is_some_and(|pool| pool.endpoints.as_ref().first().is_some_and(|endpoint| endpoint.is_local))

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

@@ -536,6 +536,41 @@ impl ECStore {
         self.load_rebalance_meta_under_start_gate().await
     }
 
+    /// Cancels local admission before refreshing the persisted stop target under the start gate.
+    pub async fn prepare_rebalance_stop(&self) -> Result<Option<String>> {
+        let _start_guard = self.start_gate.lock().await;
+
+        {
+            let mut rebalance_meta = self.rebalance_meta.write().await;
+            if let Some(meta) = rebalance_meta.as_mut()
+                && is_rebalance_conflicting_with_decommission(meta)
+            {
+                meta.cancel
+                    .get_or_insert_with(tokio_util::sync::CancellationToken::new)
+                    .cancel();
+                #[cfg(any(test, feature = "test-util"))]
+                observe_rebalance_stop_wait_attempt(Some(meta.id.as_str()));
+            }
+        }
+
+        self.load_rebalance_meta_under_start_gate().await?;
+
+        let mut rebalance_meta = self.rebalance_meta.write().await;
+        let Some(meta) = rebalance_meta.as_mut() else {
+            return Ok(None);
+        };
+        if !is_rebalance_conflicting_with_decommission(meta) {
+            return Ok(None);
+        }
+        if meta.id.is_empty() {
+            return Err(Error::other("active rebalance metadata has no activation id"));
+        }
+        meta.cancel
+            .get_or_insert_with(tokio_util::sync::CancellationToken::new)
+            .cancel();
+        Ok(Some(meta.id.clone()))
+    }
+
     pub(crate) async fn load_rebalance_meta_under_start_gate(&self) -> Result<()> {
         let mut meta = RebalanceMeta::new();
         debug!(
@@ -1298,11 +1333,19 @@ mod tests {
             .expect("the committed worker candidate should remain installed locally");
         assert_eq!(local.id, rebalance_id);
         assert_eq!(local.pool_stats[0].info.status, RebalStatus::Completed);
-        if let Some(cancel) = local.cancel.as_ref() {
-            cancel.cancel();
-        }
+        let admitted_cancel = local
+            .cancel
+            .clone()
+            .expect("the committed worker candidate should install its cancellation token");
+        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");
+        assert!(admitted_cancel.is_cancelled());
+
         let mut persisted = RebalanceMeta::new();
         persisted
             .load(store.pools[0].clone())

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

@@ -54,7 +54,7 @@ pub use types::{
 use types::{RebalanceBucketConfigs, RebalanceBucketOutcome, RebalanceEntryOutcome};
 
 #[cfg(any(test, feature = "test-util"))]
-pub(crate) async fn test_store_with_persisted_rebalance_meta(
+pub 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());

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

@@ -272,6 +272,11 @@ impl ECStore {
             return Ok(());
         }
 
+        #[cfg(test)]
+        let endpoints = self.instance_endpoints().unwrap_or_else(|| self.endpoints());
+        #[cfg(not(test))]
+        let endpoints = self.endpoints();
+
         let mut workers_started = 0usize;
         for (idx, participating) in participants.iter().enumerate() {
             if !*participating {
@@ -287,7 +292,7 @@ impl ECStore {
                 continue;
             }
 
-            if !runtime_sources::endpoint_pool_is_local(idx) {
+            if !runtime_sources::endpoint_pool_is_local(&endpoints, idx) {
                 debug!(
                     event = EVENT_REBALANCE_STATE,
                     component = LOG_COMPONENT_ECSTORE,