fix(decommission): detach the leader-side start from its caller

9ac4c10de carried only the reproduction (the testable entry point, the
peer-reload pause point and the proof test). This adds the fix it
describes: `ECStore::decommission` spawns the start as its own task and
awaits only its result, so a caller that stops waiting, such as a
forwarded start whose internode RPC timed out, can no longer drop the
start between persisting the pool as active and having the peers reload
and the local worker start.

Also prove a retry while the first start runs cannot start the pool
twice, update the operator and architecture docs for the new behavior,
and shrink the error(format!) ratchet baseline for the removed helper.

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::{debug, error, info, warn};
+use tracing::{Instrument as _, debug, error, info, warn};
 
 const LOG_COMPONENT_ECSTORE: &str = "ecstore";
 const LOG_SUBSYSTEM_POOLS: &str = "pools";
@@ -119,7 +119,7 @@ const DECOMMISSION_TARGET_FLEET_PROOF_EXPIRED: &str =
 /// indefinitely. It stays below the default internode RPC timeout (30 s) so
 /// gate waits alone cannot push a forwarded start past that timeout; the
 /// start's other work is not bounded by it, and a forwarded start whose RPC
-/// times out is dropped wherever it is (see the decommission operator docs).
+/// times out keeps running on the leader (see `ECStore::decommission`).
 const DECOMMISSION_START_GATE_WAIT: std::time::Duration = std::time::Duration::from_secs(20);
 pub(crate) const DECOMMISSION_ACTIVATION_START_GATE: &str = "start gate (decommission activation)";
 pub(crate) const DECOMMISSION_ACTIVATION_MOVEMENT_GATE: &str = "data movement gate (decommission activation)";
@@ -15024,8 +15024,24 @@ impl ECStore {
         Ok(())
     }
 
+    /// Start decommissioning `indices` from this leader.
+    ///
+    /// 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,
+    /// 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.
     #[tracing::instrument(skip(self, rx))]
     pub async fn decommission(self: &Arc<Self>, rx: CancellationToken, indices: Vec<usize>) -> Result<()> {
+        let store = Arc::clone(self);
+        tokio::spawn(async move { store.run_decommission_start(rx, indices).await }.in_current_span())
+            .await
+            .map_err(Error::other)?
+    }
+
+    async fn run_decommission_start(self: Arc<Self>, rx: CancellationToken, indices: Vec<usize>) -> Result<()> {
         let indices = dedup_indices(&indices);
 
         info!(
@@ -15050,7 +15066,7 @@ impl ECStore {
             .start_decommission_inner(indices, Some((&rx, local_indices.as_slice())), &budget)
             .await?;
         drop(spawn_decommission_index_cancelers(
-            Arc::clone(self),
+            Arc::clone(&self),
             rx,
             index_cancelers,
             Arc::new(Semaphore::new(decommission_entry_concurrency_limit())),
@@ -20797,6 +20813,70 @@ mod tests {
         outcome.expect("a dropped start must not leave the pool persisted as active without a worker");
     }
 
+    #[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();
+
+        let first = tokio::spawn({
+            let store = Arc::clone(&store);
+            let rx = rx.clone();
+            async move { store.decommission(rx, vec![0]).await }
+        });
+        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()
+            .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:?}"
+        );
+
+        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");
+
+        // 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() {

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.
+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.
 
 ### 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 whose caller stops waiting is dropped on the leader wherever it was. This happens to a forwarded start that hits the internode RPC timeout (`RUSTFS_INTERNODE_RPC_TIMEOUT_SECS`, 30 seconds by default) and may happen when an admin client disconnects. Either nothing was persisted, or `pool.bin` already records the pool as active although the leader had not yet started its worker or told the other nodes to reload pool metadata. Check `decommission/status` before retrying: a pool that is not active can be started again, while a new start of an active pool is rejected. For an active pool, the leader's decommission watchdog resumes the missing worker within about 30 seconds once no other decommission worker is running on that node; if progress still does not advance, restart the leader (the pool's first endpoint), whose startup resumes the decommission. Nodes that missed the reload keep their previous view of the pool until they reload pool metadata, for example on restart. Making the leader-side start cancel-safe is tracked in rustfs/backlog#2873.
+- 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.
 - 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

scripts/error-other-format-baseline.txt

@@ -24,7 +24,7 @@
 10|crates/ecstore/src/cluster/rpc/remote_disk.rs
 6|crates/ecstore/src/config/com.rs
 14|crates/ecstore/src/config/storageclass.rs
-177|crates/ecstore/src/core/pools.rs
+176|crates/ecstore/src/core/pools.rs
 6|crates/ecstore/src/data_movement/mod.rs
 2|crates/ecstore/src/data_usage/local_snapshot.rs
 12|crates/ecstore/src/data_usage/mod.rs