fix(decommission): single-flight start, recheck pool state under gate

Review of 3a5276c5b:

- A start re-checks the pool state under the start gate before it
  cancels routines, so a start racing another one can no longer cancel
  the worker the other start just spawned.
- Only one start runs per instance at a time: another request, including
  the retry of an operator whose request timed out, is rejected as
  already running instead of stacking a second detached start and its
  preflight heals. The start runs through `run_detached_mutation`, and a
  failure nobody waits for any more is logged.
- Tests: the proof test pins its drop after the save and requires a
  worker that persists a cancellation; the retry test really drops the
  first caller; new tests cover the under-gate recheck and a peer reload
  failure after the caller is gone. That last one shows the start
  rollback cannot clear a fresh start because the pool metadata
  stale-write guard refuses it; the gap is asserted and documented and
  tracked in rustfs/backlog#2873.
- The start tests run on their own large-stack thread: the io_uring lane
  (`uring_`, which also selects `during_`) runs them with libtest's 2 MiB
  default stack, where the proof test overflowed.
- Docs: single-flight starts, status and cancel during preflight, the
  rollback gap and its recovery, and the release requirement.

Refs rustfs/backlog#2873

crates/ecstore/src/core/pools.rs

@@ -100,7 +100,7 @@ use std::sync::{
 use time::{Duration, OffsetDateTime};
 use tokio::sync::{OwnedSemaphorePermit, Semaphore, mpsc};
 use tokio_util::sync::CancellationToken;
-use tracing::{Instrument as _, debug, error, info, warn};
+use tracing::{debug, error, info, warn};
 
 const LOG_COMPONENT_ECSTORE: &str = "ecstore";
 const LOG_SUBSYSTEM_POOLS: &str = "pools";
@@ -381,6 +381,38 @@ impl Drop for DecommissionCancelerGuard {
     }
 }
 
+/// The instance's single decommission start slot, released on drop.
+struct DecommissionStartSlot(Arc<crate::runtime::instance::InstanceContext>);
+
+impl DecommissionStartSlot {
+    fn claim(ctx: &Arc<crate::runtime::instance::InstanceContext>) -> Option<Self> {
+        ctx.try_begin_decommission_start().then(|| Self(Arc::clone(ctx)))
+    }
+}
+
+impl Drop for DecommissionStartSlot {
+    fn drop(&mut self) {
+        self.0.end_decommission_start();
+    }
+}
+
+/// Lets a detached start see whether its caller still waits for the result.
+#[derive(Default)]
+struct DecommissionStartCaller(Arc<AtomicBool>);
+
+impl DecommissionStartCaller {
+    fn waiting(&self) -> Arc<AtomicBool> {
+        self.0.store(true, Ordering::Release);
+        Arc::clone(&self.0)
+    }
+}
+
+impl Drop for DecommissionStartCaller {
+    fn drop(&mut self) {
+        self.0.store(false, Ordering::Release);
+    }
+}
+
 fn dedup_indices(indices: &[usize]) -> Vec<usize> {
     let mut seen = HashSet::with_capacity(indices.len());
     let mut output = Vec::with_capacity(indices.len());
@@ -10850,10 +10882,14 @@ pub(crate) fn set_decommission_start_gate_wait_for_test(store_id: uuid::Uuid, wa
 #[cfg(test)]
 #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
 pub(crate) enum DecommissionStartPausePoint {
+    /// After the unlocked pool-state check, before the activation start gate.
+    StartGate,
     /// Just before the wait for the activation movement writer.
     Activation,
     /// After the start is persisted, where the peers are asked to reload.
     PeerReload,
+    /// In a spawned worker once its pool is promoted, before any work.
+    WorkerPromoted,
 }
 
 #[cfg(test)]
@@ -10883,6 +10919,28 @@ pub(crate) fn install_decommission_start_pause_for_test(
     pause
 }
 
+#[cfg(test)]
+static DECOMMISSION_START_RELOAD_FAILURES: std::sync::OnceLock<std::sync::Mutex<HashSet<uuid::Uuid>>> =
+    std::sync::OnceLock::new();
+
+/// Makes the next start of a store fail its peer reload after the save.
+#[cfg(test)]
+pub(crate) fn fail_next_decommission_start_reload_for_test(store_id: uuid::Uuid) {
+    DECOMMISSION_START_RELOAD_FAILURES
+        .get_or_init(|| std::sync::Mutex::new(HashSet::new()))
+        .lock()
+        .expect("decommission start reload failure should not be poisoned")
+        .insert(store_id);
+}
+
+#[cfg(test)]
+fn take_decommission_start_reload_failure_for_test(store_id: uuid::Uuid) -> Option<Error> {
+    DECOMMISSION_START_RELOAD_FAILURES
+        .get()
+        .and_then(|failures| failures.lock().ok()?.remove(&store_id).then_some(()))
+        .map(|()| Error::other("injected peer pool metadata reload failure"))
+}
+
 #[cfg(test)]
 async fn pause_decommission_start_for_test(store_id: uuid::Uuid, point: DecommissionStartPausePoint) {
     let pause = DECOMMISSION_START_PAUSES
@@ -15029,16 +15087,44 @@ impl ECStore {
     /// The start runs in its own task. Once begun it must reach its end: it
     /// either fails before persisting anything, or persists the pools as
     /// active, has the peers reload pool metadata and spawns the local
-    /// workers, rolling back if the reload fails. A caller that stops waiting,
+    /// workers, attempting a rollback if the reload fails. A caller that stops waiting,
     /// such as a forwarded start whose internode RPC timed out, only gives up
     /// on the answer; dropping the start itself could leave a pool persisted
     /// as active with no worker and with peers still on the old metadata.
+    ///
+    /// One start runs at a time on an instance: another request while a start
+    /// is running, including its preflight, is rejected as already running
+    /// rather than queueing a second detached start behind it.
     #[tracing::instrument(skip(self, rx))]
     pub async fn decommission(self: &Arc<Self>, rx: CancellationToken, indices: Vec<usize>) -> Result<()> {
+        let Some(start_slot) = DecommissionStartSlot::claim(&self.ctx) else {
+            return Err(Error::DecommissionAlreadyRunning);
+        };
+        let caller = DecommissionStartCaller::default();
+        let caller_waiting = caller.waiting();
         let store = Arc::clone(self);
-        tokio::spawn(async move { store.run_decommission_start(rx, indices).await }.in_current_span())
-            .await
-            .map_err(Error::other)?
+        self.run_detached_mutation(async move {
+            let _start_slot = start_slot;
+            // Keep the detached task's root future small; the start's own
+            // state lives behind this box.
+            let result = Box::pin(store.run_decommission_start(rx, indices)).await;
+            if let Err(err) = &result
+                && !caller_waiting.load(Ordering::Acquire)
+            {
+                // Nobody receives this error any more; record it here.
+                warn!(
+                    event = EVENT_DECOMMISSION_STATE,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_POOLS,
+                    state = "start_failed",
+                    stage = "caller_gone",
+                    error = %err,
+                    "Decommission start failed after its caller stopped waiting"
+                );
+            }
+            result
+        })
+        .await?
     }
 
     async fn run_decommission_start(self: Arc<Self>, rx: CancellationToken, indices: Vec<usize>) -> Result<()> {
@@ -17583,6 +17669,8 @@ impl ECStore {
                 return Err(err);
             }
         };
+        #[cfg(test)]
+        pause_decommission_start_for_test(self.id, DecommissionStartPausePoint::WorkerPromoted).await;
         if rx.is_cancelled() {
             let already_canceled = {
                 let pool_meta = self.pool_meta.read().await;
@@ -18191,11 +18279,17 @@ impl ECStore {
             }
         }
 
+        #[cfg(test)]
+        pause_decommission_start_for_test(self.id, DecommissionStartPausePoint::StartGate).await;
         let _start_guard = budget
             .acquire(DECOMMISSION_ACTIVATION_START_GATE, self.start_gate.lock())
             .await?;
         self.ensure_decommission_rebalance_idle_after_refresh_under_start_gate(budget)
             .await?;
+        // Another start may have activated these pools since the unlocked
+        // check above. Reject it here, before cancelling routines below would
+        // stop that start's freshly spawned worker.
+        ensure_decommission_start_pool_states(&*self.pool_meta.read().await, &indices)?;
 
         let all_capacity_infos = self.get_decommission_all_pool_capacity_infos_with_owned_usage().await?;
         let target_fence_proof_available =
@@ -18238,9 +18332,17 @@ impl ECStore {
 
         #[cfg(test)]
         pause_decommission_start_for_test(self.id, DecommissionStartPausePoint::PeerReload).await;
-        if let Some(notification_sys) = runtime_sources::notification_sys()
-            && let Err(err) = resolve_start_decommission_pool_meta_reload_result(notification_sys.reload_pool_meta().await)
-        {
+        let notification_sys = runtime_sources::notification_sys();
+        let reload_result = match notification_sys.as_ref() {
+            Some(notification_sys) => {
+                resolve_start_decommission_pool_meta_reload_result(notification_sys.reload_pool_meta().await)
+            }
+            None => Ok(()),
+        };
+        #[cfg(test)]
+        let reload_result =
+            reload_result.and_then(|()| take_decommission_start_reload_failure_for_test(self.id).map_or(Ok(()), Err));
+        if let Err(err) = reload_result {
             warn!(
                 event = EVENT_DECOMMISSION_STATE,
                 component = LOG_COMPONENT_ECSTORE,
@@ -18270,10 +18372,12 @@ impl ECStore {
                 )));
             }
 
-            if let Err(rollback_reload_err) = resolve_decommission_pool_meta_reload_result(
-                notification_sys.reload_pool_meta().await,
-                "start_decommission_rollback",
-            ) {
+            if let Some(notification_sys) = notification_sys.as_ref()
+                && let Err(rollback_reload_err) = resolve_decommission_pool_meta_reload_result(
+                    notification_sys.reload_pool_meta().await,
+                    "start_decommission_rollback",
+                )
+            {
                 error!(
                     event = EVENT_DECOMMISSION_STATE,
                     component = LOG_COMPONENT_ECSTORE,
@@ -20689,302 +20793,454 @@ mod tests {
         assert_eq!(timed_out_gate(&err), Some(("b", std::time::Duration::from_millis(100))));
     }
 
-    #[tokio::test]
+    #[test]
     #[serial_test::serial]
-    async fn decommission_start_wait_for_a_held_start_gate_is_bounded() {
-        let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
-        let baseline = load_pool_meta_replicas(store.pools.clone(), true)
-            .await
-            .expect("baseline pool metadata should be readable");
-        set_decommission_start_gate_wait_for_test(store.id, std::time::Duration::from_millis(500));
+    fn decommission_start_wait_for_a_held_start_gate_is_bounded() {
+        run_decommission_start_test("decommission_start_wait_for_a_held_start_gate_is_bounded", || async {
+            let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
+            let baseline = load_pool_meta_replicas(store.pools.clone(), true)
+                .await
+                .expect("baseline pool metadata should be readable");
+            set_decommission_start_gate_wait_for_test(store.id, std::time::Duration::from_millis(500));
 
-        // Another start, a rebalance transition, or intent adoption after a
-        // leader restart holds the start gate.
-        let held_start_gate = store.start_gate.lock().await;
-        let err = tokio::time::timeout(std::time::Duration::from_secs(10), store.start_decommission(vec![0]))
-            .await
-            .expect("a start must not hang behind a held start gate")
-            .expect_err("a start must fail while the start gate stays held");
-        assert_eq!(
-            timed_out_gate(&err).map(|(gate, _)| gate),
-            Some(crate::services::rebalance::REBALANCE_REFRESH_START_GATE),
-            "{err:?}"
-        );
-        drop(held_start_gate);
-        assert_start_left_no_trace(&store, &baseline).await;
+            // Another start, a rebalance transition, or intent adoption after a
+            // leader restart holds the start gate.
+            let held_start_gate = store.start_gate.lock().await;
+            let err = tokio::time::timeout(std::time::Duration::from_secs(10), store.start_decommission(vec![0]))
+                .await
+                .expect("a start must not hang behind a held start gate")
+                .expect_err("a start must fail while the start gate stays held");
+            assert_eq!(
+                timed_out_gate(&err).map(|(gate, _)| gate),
+                Some(crate::services::rebalance::REBALANCE_REFRESH_START_GATE),
+                "{err:?}"
+            );
+            drop(held_start_gate);
+            assert_start_left_no_trace(&store, &baseline).await;
+        });
     }
 
-    #[tokio::test]
+    #[test]
     #[serial_test::serial]
-    async fn decommission_start_wait_for_an_in_flight_rebalance_entry_is_bounded() {
-        let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
-        let baseline = load_pool_meta_replicas(store.pools.clone(), true)
-            .await
-            .expect("baseline pool metadata should be readable");
-        set_decommission_start_gate_wait_for_test(store.id, std::time::Duration::from_millis(500));
-        let activation_gate = {
-            let mut rebalance_meta = store.rebalance_meta.write().await;
-            let meta = rebalance_meta.insert(RebalanceMeta::default());
-            Arc::clone(&meta.activation_gate)
+    fn decommission_start_wait_for_an_in_flight_rebalance_entry_is_bounded() {
+        run_decommission_start_test("decommission_start_wait_for_an_in_flight_rebalance_entry_is_bounded", || async {
+            let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
+            let baseline = load_pool_meta_replicas(store.pools.clone(), true)
+                .await
+                .expect("baseline pool metadata should be readable");
+            set_decommission_start_gate_wait_for_test(store.id, std::time::Duration::from_millis(500));
+            let activation_gate = {
+                let mut rebalance_meta = store.rebalance_meta.write().await;
+                let meta = rebalance_meta.insert(RebalanceMeta::default());
+                Arc::clone(&meta.activation_gate)
+            };
+
+            let in_flight_entry = activation_gate.read_owned().await;
+            let err = tokio::time::timeout(std::time::Duration::from_secs(10), store.start_decommission(vec![0]))
+                .await
+                .expect("a start must not hang behind an in-flight rebalance entry")
+                .expect_err("a start must fail while the rebalance activation gate stays held");
+            assert_eq!(
+                timed_out_gate(&err).map(|(gate, _)| gate),
+                Some(crate::services::rebalance::REBALANCE_REFRESH_ACTIVATION_GATE),
+                "{err:?}"
+            );
+            drop(in_flight_entry);
+            assert_start_left_no_trace(&store, &baseline).await;
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn decommission_entry_wait_is_bounded() {
+        run_decommission_start_test("decommission_entry_wait_is_bounded", || async {
+            let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
+            let baseline = load_pool_meta_replicas(store.pools.clone(), true)
+                .await
+                .expect("baseline pool metadata should be readable");
+            set_decommission_start_gate_wait_for_test(store.id, std::time::Duration::from_millis(500));
+
+            let stuck_side_effect = store.ctx.data_movement_operation_gate().read_owned().await;
+            let err =
+                tokio::time::timeout(std::time::Duration::from_secs(10), store.decommission(CancellationToken::new(), vec![0]))
+                    .await
+                    .expect("the admin and RPC start entry must not hang behind a stuck movement side effect")
+                    .expect_err("the start entry must fail while the movement gate stays busy");
+            assert_eq!(
+                timed_out_gate(&err).map(|(gate, _)| gate),
+                Some(crate::services::rebalance::REBALANCE_REFRESH_MOVEMENT_GATE),
+                "{err:?}"
+            );
+            assert_start_left_no_trace(&store, &baseline).await;
+            drop(stuck_side_effect);
+        });
+    }
+
+    /// Runs a test that drives real decommission starts on its own thread and
+    /// current-thread runtime, so the detached start tasks it spawns are polled
+    /// with the stack headroom the crate's other large-stack storage tests use,
+    /// whatever stack the test harness thread has.
+    fn run_decommission_start_test<C, F>(name: &str, case: C)
+    where
+        C: FnOnce() -> F + Send + 'static,
+        F: std::future::Future<Output = ()> + 'static,
+    {
+        const STACK_SIZE: usize = if cfg!(debug_assertions) {
+            8 * rustfs_config::DEFAULT_THREAD_STACK_SIZE
+        } else if cfg!(target_os = "macos") {
+            2 * rustfs_config::DEFAULT_THREAD_STACK_SIZE
+        } else {
+            rustfs_config::DEFAULT_THREAD_STACK_SIZE
         };
-
-        let in_flight_entry = activation_gate.read_owned().await;
-        let err = tokio::time::timeout(std::time::Duration::from_secs(10), store.start_decommission(vec![0]))
-            .await
-            .expect("a start must not hang behind an in-flight rebalance entry")
-            .expect_err("a start must fail while the rebalance activation gate stays held");
-        assert_eq!(
-            timed_out_gate(&err).map(|(gate, _)| gate),
-            Some(crate::services::rebalance::REBALANCE_REFRESH_ACTIVATION_GATE),
-            "{err:?}"
-        );
-        drop(in_flight_entry);
-        assert_start_left_no_trace(&store, &baseline).await;
+        std::thread::Builder::new()
+            .name(name.to_string())
+            .stack_size(STACK_SIZE)
+            .spawn(move || {
+                tokio::runtime::Builder::new_current_thread()
+                    .enable_all()
+                    .build()
+                    .expect("large-stack decommission start test runtime should build")
+                    .block_on(case());
+            })
+            .expect("large-stack decommission start test thread should spawn")
+            .join()
+            .expect("large-stack decommission start test should complete");
     }
 
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn decommission_entry_wait_is_bounded() {
-        let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
-        let baseline = load_pool_meta_replicas(store.pools.clone(), true)
-            .await
-            .expect("baseline pool metadata should be readable");
-        set_decommission_start_gate_wait_for_test(store.id, std::time::Duration::from_millis(500));
-
-        let stuck_side_effect = store.ctx.data_movement_operation_gate().read_owned().await;
-        let err = tokio::time::timeout(std::time::Duration::from_secs(10), store.decommission(CancellationToken::new(), vec![0]))
-            .await
-            .expect("the admin and RPC start entry must not hang behind a stuck movement side effect")
-            .expect_err("the start entry must fail while the movement gate stays busy");
-        assert_eq!(
-            timed_out_gate(&err).map(|(gate, _)| gate),
-            Some(crate::services::rebalance::REBALANCE_REFRESH_MOVEMENT_GATE),
-            "{err:?}"
-        );
-        assert_start_left_no_trace(&store, &baseline).await;
-        drop(stuck_side_effect);
+    fn persisted_decommission(selection: &PoolMetaSelection) -> Option<&PoolDecommissionInfo> {
+        selection.meta.pools[0].decommission.as_ref()
     }
 
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn decommission_start_survives_a_caller_dropped_during_peer_reload() {
-        let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
-        let baseline = load_pool_meta_replicas(store.pools.clone(), true)
-            .await
-            .expect("baseline pool metadata should be readable");
-        let reload = install_decommission_start_pause_for_test(store.id, DecommissionStartPausePoint::PeerReload);
-        let rx = CancellationToken::new();
-
-        // A forwarded start whose RPC times out has its handler future
-        // dropped; here that happens while the peers reload pool metadata,
-        // after the start is persisted and before its worker is spawned.
-        {
-            let start = store.decommission(rx.clone(), vec![0]);
-            tokio::pin!(start);
-            tokio::select! {
-                _ = reload.reached.notified() => {}
-                result = &mut start => panic!("the start finished before its peer reload: {result:?}"),
-            }
-        }
-        reload.proceed.notify_one();
-
-        // Either nothing was persisted, or the start ran to its end: a local
-        // worker owns the pool, or one already took it to a terminal state.
-        let outcome = tokio::time::timeout(std::time::Duration::from_secs(10), async {
+    /// Polls `pool.bin` until `done` holds, for up to ten seconds.
+    async fn wait_for_persisted_pool_meta(
+        store: &ECStore,
+        mut done: impl FnMut(&PoolMetaSelection) -> bool,
+    ) -> std::result::Result<PoolMetaSelection, tokio::time::error::Elapsed> {
+        tokio::time::timeout(std::time::Duration::from_secs(10), async {
             loop {
                 let persisted = load_pool_meta_replicas(store.pools.clone(), true)
                     .await
                     .expect("pool metadata should stay readable");
-                if persisted.canonical == baseline.canonical {
-                    return "not persisted";
-                }
-                let terminal = persisted.meta.pools[0]
-                    .decommission
-                    .as_ref()
-                    .is_some_and(|info| info.complete || info.failed || info.canceled);
-                if terminal || store.has_active_local_decommission_worker().await {
-                    return "worker started";
+                if done(&persisted) {
+                    return persisted;
                 }
                 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
             }
         })
-        .await;
-        rx.cancel();
-        outcome.expect("a dropped start must not leave the pool persisted as active without a worker");
+        .await
     }
 
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn decommission_start_retried_while_the_first_runs_starts_once() {
-        let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
-        set_decommission_start_gate_wait_for_test(store.id, std::time::Duration::from_millis(300));
-        let reload = install_decommission_start_pause_for_test(store.id, DecommissionStartPausePoint::PeerReload);
-        let rx = CancellationToken::new();
+    /// Drives `decommission` until the given pause point, then drops it the
+    /// way a timed-out forwarding RPC drops its handler.
+    async fn drop_decommission_caller_at(store: &Arc<ECStore>, rx: &CancellationToken, pause: &DecommissionStartPause) {
+        let start = store.decommission(rx.clone(), vec![0]);
+        tokio::pin!(start);
+        tokio::select! {
+            _ = pause.reached.notified() => {}
+            result = &mut start => panic!("the start finished before its pause point: {result:?}"),
+        }
+    }
 
-        let first = tokio::spawn({
-            let store = Arc::clone(&store);
-            let rx = rx.clone();
-            async move { store.decommission(rx, vec![0]).await }
+    #[test]
+    #[serial_test::serial]
+    fn decommission_start_survives_a_caller_dropped_during_peer_reload() {
+        run_decommission_start_test("decommission_start_survives_a_caller_dropped_during_peer_reload", || async {
+            let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
+            let baseline = load_pool_meta_replicas(store.pools.clone(), true)
+                .await
+                .expect("baseline pool metadata should be readable");
+            let reload = install_decommission_start_pause_for_test(store.id, DecommissionStartPausePoint::PeerReload);
+            let worker = install_decommission_start_pause_for_test(store.id, DecommissionStartPausePoint::WorkerPromoted);
+            let rx = CancellationToken::new();
+
+            // The caller goes away while the peers reload pool metadata, after
+            // the start is persisted and before its worker is spawned.
+            drop_decommission_caller_at(&store, &rx, &reload).await;
+            let persisted = load_pool_meta_replicas(store.pools.clone(), true)
+                .await
+                .expect("pool metadata should be readable");
+            assert_ne!(
+                persisted.canonical, baseline.canonical,
+                "the caller must be dropped after the start is persisted"
+            );
+            reload.proceed.notify_one();
+
+            // Only a running worker persists a cancellation.
+            tokio::time::timeout(std::time::Duration::from_secs(10), worker.reached.notified())
+                .await
+                .expect("the detached start should spawn its worker");
+            rx.cancel();
+            worker.proceed.notify_one();
+            wait_for_persisted_pool_meta(&store, |persisted| persisted_decommission(persisted).is_some_and(|info| info.canceled))
+                .await
+                .expect("a dropped start must still hand the pool to a worker that honors cancellation");
         });
-        tokio::time::timeout(std::time::Duration::from_secs(30), reload.reached.notified())
-            .await
-            .expect("the first start should reach its peer reload");
-        let started = load_pool_meta_replicas(store.pools.clone(), true)
-            .await
-            .expect("pool metadata should be readable")
-            .meta
-            .pools[0]
-            .decommission
-            .as_ref()
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn decommission_start_is_single_flight_after_its_caller_drops() {
+        run_decommission_start_test("decommission_start_is_single_flight_after_its_caller_drops", || async {
+            let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
+            let reload = install_decommission_start_pause_for_test(store.id, DecommissionStartPausePoint::PeerReload);
+            let worker = install_decommission_start_pause_for_test(store.id, DecommissionStartPausePoint::WorkerPromoted);
+            let rx = CancellationToken::new();
+
+            drop_decommission_caller_at(&store, &rx, &reload).await;
+            let started = persisted_decommission(
+                &load_pool_meta_replicas(store.pools.clone(), true)
+                    .await
+                    .expect("pool metadata should be readable"),
+            )
             .and_then(|info| info.start_time)
             .expect("the first start is persisted before its peer reload");
 
-        // A retry while the first start still holds the start gate gives up
-        // within its budget instead of starting a second time.
-        let retry = tokio::time::timeout(std::time::Duration::from_secs(10), store.decommission(rx.clone(), vec![0]))
-            .await
-            .expect("a retry must not hang behind the running start")
-            .expect_err("a retry must not start while the first start runs");
-        assert_eq!(
-            timed_out_gate(&retry).map(|(gate, _)| gate),
-            Some(crate::services::rebalance::REBALANCE_REFRESH_START_GATE),
-            "{retry:?}"
-        );
+            // The retry of an operator whose request timed out is turned away
+            // at once while the detached start still runs; it neither waits nor
+            // runs its own preflight.
+            let retry = tokio::time::timeout(std::time::Duration::from_secs(1), store.decommission(rx.clone(), vec![0]))
+                .await
+                .expect("a retry must be answered at once while a start runs")
+                .expect_err("a retry must not start while the first start runs");
+            assert!(matches!(retry, Error::DecommissionAlreadyRunning), "{retry:?}");
 
-        reload.proceed.notify_one();
-        tokio::time::timeout(std::time::Duration::from_secs(30), first)
-            .await
-            .expect("the first start should finish")
-            .expect("first start task should not panic")
-            .expect("the first start should succeed");
+            reload.proceed.notify_one();
+            tokio::time::timeout(std::time::Duration::from_secs(10), worker.reached.notified())
+                .await
+                .expect("the detached start should spawn its worker");
+            assert!(store.has_active_local_decommission_worker().await);
 
-        // Once it has finished, a retry sees the pool already moving. Give
-        // it room to wait out the new worker's brief gate holds.
-        set_decommission_start_gate_wait_for_test(store.id, std::time::Duration::from_secs(10));
-        let late = store
-            .decommission(rx.clone(), vec![0])
-            .await
-            .expect_err("a pool that is already decommissioning must not start again");
-        assert!(timed_out_gate(&late).is_none(), "{late:?}");
-        let persisted = load_pool_meta_replicas(store.pools.clone(), true)
-            .await
-            .expect("pool metadata should be readable");
-        assert_eq!(
-            persisted.meta.pools[0].decommission.as_ref().and_then(|info| info.start_time),
-            Some(started),
-            "the pool is started exactly once"
-        );
-        rx.cancel();
-    }
-
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn decommission_start_rebalance_refresh_wait_is_bounded() {
-        let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
-        let baseline = load_pool_meta_replicas(store.pools.clone(), true)
-            .await
-            .expect("baseline pool metadata should be readable");
-        set_decommission_start_gate_wait_for_test(store.id, std::time::Duration::from_millis(500));
-
-        let stuck_side_effect = store.ctx.data_movement_operation_gate().read_owned().await;
-        let err = tokio::time::timeout(std::time::Duration::from_secs(10), store.start_decommission(vec![0]))
-            .await
-            .expect("a start must not hang behind a stuck movement side effect")
-            .expect_err("a start must fail while the movement gate stays busy");
-        let (gate, waited) = timed_out_gate(&err).unwrap_or_else(|| panic!("a busy gate must be a lock timeout: {err:?}"));
-        assert_eq!(gate, crate::services::rebalance::REBALANCE_REFRESH_MOVEMENT_GATE);
-        assert!(waited >= std::time::Duration::from_millis(500), "{waited:?}");
-        assert_start_left_no_trace(&store, &baseline).await;
-        drop(stuck_side_effect);
-    }
-
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn decommission_admin_rebalance_precheck_wait_is_bounded() {
-        let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
-        set_decommission_start_gate_wait_for_test(store.id, std::time::Duration::from_millis(500));
-
-        let stuck_side_effect = store.ctx.data_movement_operation_gate().read_owned().await;
-        let err = tokio::time::timeout(std::time::Duration::from_secs(10), store.load_rebalance_meta_for_decommission_start())
-            .await
-            .expect("the admin pre-check must not hang behind a stuck movement side effect")
-            .expect_err("the admin pre-check must fail while the movement gate stays busy");
-        assert_eq!(
-            timed_out_gate(&err).map(|(gate, _)| gate),
-            Some(crate::services::rebalance::REBALANCE_REFRESH_MOVEMENT_GATE),
-            "{err:?}"
-        );
-        drop(stuck_side_effect);
-    }
-
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn decommission_activation_wait_is_bounded() {
-        let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
-        let baseline = load_pool_meta_replicas(store.pools.clone(), true)
-            .await
-            .expect("baseline pool metadata should be readable");
-        set_decommission_start_gate_wait_for_test(store.id, std::time::Duration::from_millis(500));
-        let pause = install_decommission_start_pause_for_test(store.id, DecommissionStartPausePoint::Activation);
-
-        let start = tokio::spawn({
-            let store = Arc::clone(&store);
-            async move { store.start_decommission(vec![0]).await }
+            // Once it has finished, a retry finds the pool already moving.
+            let late = store
+                .decommission(rx.clone(), vec![0])
+                .await
+                .expect_err("a pool that is already decommissioning must not start again");
+            assert!(matches!(late, Error::DecommissionAlreadyRunning), "{late:?}");
+            assert!(store.has_active_local_decommission_worker().await);
+            assert_eq!(
+                persisted_decommission(
+                    &load_pool_meta_replicas(store.pools.clone(), true)
+                        .await
+                        .expect("pool metadata should be readable"),
+                )
+                .and_then(|info| info.start_time),
+                Some(started),
+                "the pool is started exactly once"
+            );
+            rx.cancel();
+            worker.proceed.notify_one();
         });
-        tokio::time::timeout(std::time::Duration::from_secs(30), pause.reached.notified())
-            .await
-            .expect("the start should reach its activation wait");
-        let stuck_side_effect = store.ctx.data_movement_operation_gate().read_owned().await;
-        pause.proceed.notify_one();
-
-        let err = tokio::time::timeout(std::time::Duration::from_secs(10), start)
-            .await
-            .expect("the activation must not hang behind a stuck movement side effect")
-            .expect("start task should not panic")
-            .expect_err("the activation must fail while the movement gate stays busy");
-        assert_eq!(
-            timed_out_gate(&err).map(|(gate, _)| gate),
-            Some(DECOMMISSION_ACTIVATION_MOVEMENT_GATE),
-            "{err:?}"
-        );
-        assert_start_left_no_trace(&store, &baseline).await;
-        drop(stuck_side_effect);
     }
 
-    #[tokio::test]
+    #[test]
     #[serial_test::serial]
-    async fn decommission_start_gate_budget_survives_slow_preflight() {
-        let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
-        set_decommission_start_gate_wait_for_test(store.id, std::time::Duration::from_millis(300));
-        let pause = install_decommission_start_pause_for_test(store.id, DecommissionStartPausePoint::Activation);
+    fn decommission_start_rechecks_pool_state_before_cancelling_routines() {
+        run_decommission_start_test("decommission_start_rechecks_pool_state_before_cancelling_routines", || async {
+            let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
+            let gate = install_decommission_start_pause_for_test(store.id, DecommissionStartPausePoint::StartGate);
+            let rx = CancellationToken::new();
 
-        let start = tokio::spawn({
-            let store = Arc::clone(&store);
-            async move { store.start_decommission(vec![0]).await }
+            // Start A passes the unlocked pool-state check and waits for the gate.
+            let start_a = tokio::spawn({
+                let store = Arc::clone(&store);
+                async move { store.start_decommission(vec![0]).await }
+            });
+            tokio::time::timeout(std::time::Duration::from_secs(30), gate.reached.notified())
+                .await
+                .expect("start A should reach the start gate");
+
+            // Start B activates the pool meanwhile; its worker stays running.
+            let worker = install_decommission_start_pause_for_test(store.id, DecommissionStartPausePoint::WorkerPromoted);
+            store
+                .decommission(rx.clone(), vec![0])
+                .await
+                .expect("start B should activate the pool");
+            tokio::time::timeout(std::time::Duration::from_secs(10), worker.reached.notified())
+                .await
+                .expect("start B should spawn its worker");
+            assert!(store.has_active_local_decommission_worker().await);
+
+            gate.proceed.notify_one();
+            let err = tokio::time::timeout(std::time::Duration::from_secs(10), start_a)
+                .await
+                .expect("start A should finish")
+                .expect("start A task should not panic")
+                .expect_err("start A must be rejected once the pool is active");
+            assert!(matches!(err, Error::DecommissionAlreadyRunning), "{err:?}");
+            assert!(
+                store.has_active_local_decommission_worker().await,
+                "a rejected start must not cancel the running start's worker"
+            );
+            rx.cancel();
+            worker.proceed.notify_one();
         });
-        tokio::time::timeout(std::time::Duration::from_secs(30), pause.reached.notified())
-            .await
-            .expect("the start should reach its activation wait");
-        // Preflight work (bucket heals, capacity scans) outlasts the whole
-        // budget, then a short movement side effect holds the gate briefly.
-        tokio::time::sleep(std::time::Duration::from_millis(400)).await;
-        let brief_side_effect = store.ctx.data_movement_operation_gate().read_owned().await;
-        pause.proceed.notify_one();
-        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
-        drop(brief_side_effect);
+    }
 
-        tokio::time::timeout(std::time::Duration::from_secs(30), start)
-            .await
-            .expect("the start should finish")
-            .expect("start task should not panic")
-            .expect("a brief gate wait after slow preflight must not exhaust the budget");
-        assert!(
-            store.pool_meta.read().await.pools[0]
-                .decommission
-                .as_ref()
+    #[test]
+    #[serial_test::serial]
+    fn decommission_start_finishes_its_reload_failure_after_its_caller_drops() {
+        run_decommission_start_test("decommission_start_finishes_its_reload_failure_after_its_caller_drops", || async {
+            let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
+            let reload = install_decommission_start_pause_for_test(store.id, DecommissionStartPausePoint::PeerReload);
+            fail_next_decommission_start_reload_for_test(store.id);
+            let rx = CancellationToken::new();
+
+            drop_decommission_caller_at(&store, &rx, &reload).await;
+            assert!(
+                persisted_decommission(
+                    &load_pool_meta_replicas(store.pools.clone(), true)
+                        .await
+                        .expect("pool metadata should be readable"),
+                )
                 .is_some_and(PoolDecommissionInfo::has_decommission_state),
-            "the start should be published"
-        );
+                "the caller must be dropped after the start is persisted"
+            );
+            reload.proceed.notify_one();
+
+            // The detached start runs its reload-failure handling to the
+            // end and releases the start slot, without spawning a worker.
+            tokio::time::timeout(std::time::Duration::from_secs(10), async {
+                while !store.ctx.try_begin_decommission_start() {
+                    tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+                }
+                store.ctx.end_decommission_start();
+            })
+            .await
+            .expect("the detached start should finish and release its slot");
+            assert!(!store.has_active_local_decommission_worker().await);
+
+            // Known gap, tracked in rustfs/backlog#2873: the rollback save
+            // is refused by the pool metadata stale-write guard, which
+            // never lets a persisted active decommission be cleared, so
+            // the pool stays active without a worker. Update this when the
+            // start rollback can clear its own activation.
+            let persisted = load_pool_meta_replicas(store.pools.clone(), true)
+                .await
+                .expect("pool metadata should stay readable");
+            assert!(
+                persisted_decommission(&persisted).is_some_and(PoolDecommissionInfo::has_decommission_state),
+                "the stale-write guard currently keeps the rollback from clearing the start"
+            );
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn decommission_start_rebalance_refresh_wait_is_bounded() {
+        run_decommission_start_test("decommission_start_rebalance_refresh_wait_is_bounded", || async {
+            let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
+            let baseline = load_pool_meta_replicas(store.pools.clone(), true)
+                .await
+                .expect("baseline pool metadata should be readable");
+            set_decommission_start_gate_wait_for_test(store.id, std::time::Duration::from_millis(500));
+
+            let stuck_side_effect = store.ctx.data_movement_operation_gate().read_owned().await;
+            let err = tokio::time::timeout(std::time::Duration::from_secs(10), store.start_decommission(vec![0]))
+                .await
+                .expect("a start must not hang behind a stuck movement side effect")
+                .expect_err("a start must fail while the movement gate stays busy");
+            let (gate, waited) = timed_out_gate(&err).unwrap_or_else(|| panic!("a busy gate must be a lock timeout: {err:?}"));
+            assert_eq!(gate, crate::services::rebalance::REBALANCE_REFRESH_MOVEMENT_GATE);
+            assert!(waited >= std::time::Duration::from_millis(500), "{waited:?}");
+            assert_start_left_no_trace(&store, &baseline).await;
+            drop(stuck_side_effect);
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn decommission_admin_rebalance_precheck_wait_is_bounded() {
+        run_decommission_start_test("decommission_admin_rebalance_precheck_wait_is_bounded", || async {
+            let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
+            set_decommission_start_gate_wait_for_test(store.id, std::time::Duration::from_millis(500));
+
+            let stuck_side_effect = store.ctx.data_movement_operation_gate().read_owned().await;
+            let err =
+                tokio::time::timeout(std::time::Duration::from_secs(10), store.load_rebalance_meta_for_decommission_start())
+                    .await
+                    .expect("the admin pre-check must not hang behind a stuck movement side effect")
+                    .expect_err("the admin pre-check must fail while the movement gate stays busy");
+            assert_eq!(
+                timed_out_gate(&err).map(|(gate, _)| gate),
+                Some(crate::services::rebalance::REBALANCE_REFRESH_MOVEMENT_GATE),
+                "{err:?}"
+            );
+            drop(stuck_side_effect);
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn decommission_activation_wait_is_bounded() {
+        run_decommission_start_test("decommission_activation_wait_is_bounded", || async {
+            let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
+            let baseline = load_pool_meta_replicas(store.pools.clone(), true)
+                .await
+                .expect("baseline pool metadata should be readable");
+            set_decommission_start_gate_wait_for_test(store.id, std::time::Duration::from_millis(500));
+            let pause = install_decommission_start_pause_for_test(store.id, DecommissionStartPausePoint::Activation);
+
+            let start = tokio::spawn({
+                let store = Arc::clone(&store);
+                async move { store.start_decommission(vec![0]).await }
+            });
+            tokio::time::timeout(std::time::Duration::from_secs(30), pause.reached.notified())
+                .await
+                .expect("the start should reach its activation wait");
+            let stuck_side_effect = store.ctx.data_movement_operation_gate().read_owned().await;
+            pause.proceed.notify_one();
+
+            let err = tokio::time::timeout(std::time::Duration::from_secs(10), start)
+                .await
+                .expect("the activation must not hang behind a stuck movement side effect")
+                .expect("start task should not panic")
+                .expect_err("the activation must fail while the movement gate stays busy");
+            assert_eq!(
+                timed_out_gate(&err).map(|(gate, _)| gate),
+                Some(DECOMMISSION_ACTIVATION_MOVEMENT_GATE),
+                "{err:?}"
+            );
+            assert_start_left_no_trace(&store, &baseline).await;
+            drop(stuck_side_effect);
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn decommission_start_gate_budget_survives_slow_preflight() {
+        run_decommission_start_test("decommission_start_gate_budget_survives_slow_preflight", || async {
+            let (_temp_dirs, store, _other_store) = startable_decommission_store().await;
+            set_decommission_start_gate_wait_for_test(store.id, std::time::Duration::from_millis(300));
+            let pause = install_decommission_start_pause_for_test(store.id, DecommissionStartPausePoint::Activation);
+
+            let start = tokio::spawn({
+                let store = Arc::clone(&store);
+                async move { store.start_decommission(vec![0]).await }
+            });
+            tokio::time::timeout(std::time::Duration::from_secs(30), pause.reached.notified())
+                .await
+                .expect("the start should reach its activation wait");
+            // Preflight work (bucket heals, capacity scans) outlasts the whole
+            // budget, then a short movement side effect holds the gate briefly.
+            tokio::time::sleep(std::time::Duration::from_millis(400)).await;
+            let brief_side_effect = store.ctx.data_movement_operation_gate().read_owned().await;
+            pause.proceed.notify_one();
+            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+            drop(brief_side_effect);
+
+            tokio::time::timeout(std::time::Duration::from_secs(30), start)
+                .await
+                .expect("the start should finish")
+                .expect("start task should not panic")
+                .expect("a brief gate wait after slow preflight must not exhaust the budget");
+            assert!(
+                store.pool_meta.read().await.pools[0]
+                    .decommission
+                    .as_ref()
+                    .is_some_and(PoolDecommissionInfo::has_decommission_state),
+                "the start should be published"
+            );
+        });
     }
 
     #[tokio::test]

crates/ecstore/src/runtime/instance.rs

@@ -217,6 +217,10 @@ pub struct InstanceContext {
     /// Readers are held across one publication commit; movement transitions
     /// take the writer at their durable state commit boundary.
     data_movement_operation_gate: Arc<RwLock<()>>,
+    /// Set while an operator decommission start runs on this instance. A
+    /// start runs detached from its caller, so this keeps abandoned requests
+    /// and their retries from stacking concurrent starts and preflight heals.
+    decommission_start_in_flight: AtomicBool,
     /// Remote scanner publication leases own a read guard until explicit
     /// release or bounded expiry.  Keeping the guard in storage-owned state
     /// makes a remote movement transition wait on the same fence as a local
@@ -290,6 +294,7 @@ impl InstanceContext {
             bucket_metadata_sys: std::sync::Mutex::new(None),
             background_cancel_token: OnceLock::new(),
             data_movement_operation_gate: Arc::new(RwLock::new(())),
+            decommission_start_in_flight: AtomicBool::new(false),
             scanner_publication_leases: Arc::new(Mutex::new(HashMap::new())),
             data_movement_operation_epoch: AtomicU64::new(0),
             data_movement_operation_epoch_exhausted: AtomicBool::new(false),
@@ -349,6 +354,18 @@ impl InstanceContext {
         Arc::clone(&self.data_movement_operation_gate)
     }
 
+    /// Claim the single decommission start slot; `false` when a start is
+    /// already running. Release it with [`Self::end_decommission_start`].
+    pub(crate) fn try_begin_decommission_start(&self) -> bool {
+        self.decommission_start_in_flight
+            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
+            .is_ok()
+    }
+
+    pub(crate) fn end_decommission_start(&self) {
+        self.decommission_start_in_flight.store(false, Ordering::Release);
+    }
+
     pub(crate) async fn install_scanner_publication_lease(
         &self,
         token: Uuid,

docs/architecture/decommission-compatibility.md

@@ -24,7 +24,7 @@ Deterministic request rejections (unsupported single-pool operations, missing or
 
 The local-leader-prefix rule keeps the active worker on the leader for the pool being moved while still allowing a request to contain later targets whose leaders are different nodes. Later queued targets are recovered or promoted by the leader that owns that target.
 
-Start, cancel (`POST /v3/pools/cancel`), and clear (`POST /v3/pools/clear`) requests may arrive on any cluster node. When the target pool's first endpoint is remote, RustFS forwards the operation over the authenticated internode RPC channel to that endpoint; the receiving node still enforces the local-leader rule before mutating decommission state. On the leader a start runs in its own task (`ECStore::decommission`), so a caller that stops waiting, such as a forwarding node whose RPC timed out, only loses the answer: the start cannot stop between recording the pool as active in `pool.bin` and having the peers reload and the local worker start, or rolling back.
+Start, cancel (`POST /v3/pools/cancel`), and clear (`POST /v3/pools/clear`) requests may arrive on any cluster node. When the target pool's first endpoint is remote, RustFS forwards the operation over the authenticated internode RPC channel to that endpoint; the receiving node still enforces the local-leader rule before mutating decommission state. On the leader a start runs in its own task (`ECStore::decommission`), so a caller that stops waiting, such as a forwarding node whose RPC timed out, only loses the answer: the start cannot stop between recording the pool as active in `pool.bin` and having the peers reload and the local worker start, or rolling back. One start runs per leader at a time; another start is rejected as already running until it finishes. These guarantees need the leader to run a release that includes them.
 
 ### Persisted Metadata Shape
 

docs/operations/cluster-lifecycle-operations.md

@@ -176,7 +176,7 @@ Contract: [decommission-compatibility.md](../architecture/decommission-compatibi
 - Targets are validated as a batch before anything is persisted: unknown, duplicate, active, queued, or completed targets are rejected; failed or canceled targets must be cleared first unless they hold unresolved listing entries, in which case a new start is the retry path (`DecommissionStartPoolState::Retryable` versus `Blocked` in `crates/ecstore/src/core/pools.rs`).
 - Requests go to any node and are forwarded to the pool leader (§1).
 - In-flight data movement (an object copy or source cleanup, another start, a scanner publication lease) can hold the start, rebalance activation, and data-movement gates. The leader's rebalance pre-check and the start itself each spend at most 20 seconds in total waiting for those gates; preflight work such as bucket heals does not count, and a free gate is always taken. When the budget runs out the start persists nothing and fails with a lock timeout that names the gate: a request served by the leader answers `503 ServiceUnavailable` with `Retry-After`, while a forwarded request reaches the client as a generic internal error because the internode RPC carries only the message. Preflight bucket heals wait for the movement gate without this budget.
-- A start keeps running on the leader after its caller stops waiting, for example a forwarded start that hits the internode RPC timeout (`RUSTFS_INTERNODE_RPC_TIMEOUT_SECS`, 30 seconds by default) or an admin client that disconnects. It either fails before persisting anything, or records the pool as active in `pool.bin`, has the other nodes reload pool metadata, and starts the worker, rolling back if that reload fails. Check `decommission/status` before retrying: a retry that arrives while the first start is still running waits for it within its gate budget, then either reports the busy gate or is rejected because the pool is already active.
+- A start keeps running on the leader after its caller stops waiting, for example a forwarded start that hits the internode RPC timeout (`RUSTFS_INTERNODE_RPC_TIMEOUT_SECS`, 30 seconds by default) or an admin client that disconnects; this needs a leader running a release that includes it. Only one start runs on a leader at a time: until it finishes, another start request, including a retry, is rejected as already running. That includes its preflight (bucket heals, capacity scans), while `decommission/status` does not yet show the pool active and a cancel reports that the pool is not decommissioning. A start either fails before persisting anything, or records the pool as active in `pool.bin`, has the other nodes reload pool metadata, and starts the worker. If that reload fails the leader tries to roll the start back. Today that rollback cannot clear a fresh start, because the pool metadata stale-write guard refuses to clear a persisted active decommission (rustfs/backlog#2873). When the rollback fails (`rollback_failed` or `rollback_partial` in the leader log), the pool can stay active without a worker, or nodes can disagree about it: the leader's decommission watchdog resumes a missing worker within about 30 seconds once no other decommission worker runs on that node, restarting the leader resumes it otherwise, and other nodes keep their current view until they reload pool metadata, for example on restart. Check `decommission/status` before retrying.
 - Status reads (`pools/list`, `pools/status`, `decommission/status`) never wait for data movement. While movement holds the gates, a node answers from the persisted `pool.bin` and `rebalance.bin` merged into its in-memory view without publishing them; with the gates free it also publishes the refresh, as before.
 
 ```text