test(decommission): prove the shared gate budget and each bounded wait

Second review of 652c56875..0ef02a485: - The operator doc and the budget comment claimed a forwarded start that outlives the internode RPC timeout still completes on the leader. The leader drops it wherever it is, possibly after pool.bin records the pool as active but before its worker starts or the peers reload. Document that state, the watchdog that resumes the worker, and the checks to run; the cancel-safety fix is tracked in rustfs/backlog#2873. - Prove the budget is shared across waits (paused clock: a 60 ms wait leaves 40 ms for the next gate). - Prove the rebalance-refresh start gate and activation gate waits and the decommission() entry are bounded and named. Refs rustfs/backlog#2873
fix(decommission): count only gate waiting in the start budget
2026-10-31 08:16:35 +00:00 · 2026-10-20 12:00:10 +09:00 · 2026-10-20 11:29:55 +09:00 · 2026-10-20 06:39:50 +09:00
14 changed files with 995 additions and 115 deletions
@@ -341,8 +341,9 @@ pub mod cache {

 pub mod capacity {
    pub use crate::core::pools::{
-        DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free,
-        is_pool_activation_fleet_proof_error, path2_bucket_object, path2_bucket_object_with_base_path,
+        DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolMeta, PoolStatus, get_total_usable_capacity,
+        get_total_usable_capacity_free, is_pool_activation_fleet_proof_error, path2_bucket_object,
+        path2_bucket_object_with_base_path,
    };
    pub use crate::store::utils::is_reserved_or_invalid_bucket;
 }
@@ -111,6 +111,18 @@ const POOL_ACTIVATION_FLEET_PROOF_REQUIRED: &str = "pool activation requires a l
 const POOL_ACTIVATION_FLEET_PROOF_EXPIRED: &str = "pool activation fleet capability proof expired before commit";
 const DECOMMISSION_TARGET_FLEET_PROOF_EXPIRED: &str =
    "decommission target fence fleet capability proof expired before reservation commit";
+/// Total time an operator decommission start may spend waiting for the gates
+/// that running data movement can hold: the start gate, the rebalance
+/// activation gate and the data-movement writer. A movement side effect keeps
+/// a read on the movement gate for as long as it runs, so a large or stalled
+/// copy would otherwise hold the start, and every reader queued behind it,
+/// 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).
+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)";
 const DECOMMISSION_STAGE_MIGRATE_OBJECT: &str = "migrate_object";
 const DECOMMISSION_STAGE_CLEANUP_PREFLIGHT: &str = "cleanup_preflight";
 const DECOMMISSION_STAGE_SOURCE_CLEANUP: &str = "source_cleanup";
@@ -3910,6 +3922,63 @@ pub(crate) async fn acquire_pool_activation_fleet_proof(
        .ok_or_else(|| Error::other(POOL_ACTIVATION_FLEET_PROOF_REQUIRED))
 }

+/// Gate-waiting budget shared by every gate a single operator start acquires.
+/// Only time spent actually waiting for a gate counts, so slow preflight work
+/// between the waits (bucket heals, capacity scans) cannot use it up, and a
+/// gate that is free when asked for is always taken. Expiry is a retryable
+/// lock timeout that reports the total time waited.
+#[derive(Debug)]
+pub(crate) struct GateWaitBudget {
+    budget: std::time::Duration,
+    waited_nanos: std::sync::atomic::AtomicU64,
+}
+
+impl GateWaitBudget {
+    pub(crate) fn new(budget: std::time::Duration) -> Self {
+        Self {
+            budget,
+            waited_nanos: std::sync::atomic::AtomicU64::new(0),
+        }
+    }
+
+    pub(crate) async fn acquire<F: std::future::Future>(&self, resource: &'static str, acquire: F) -> Result<F::Output> {
+        let remaining = self.budget.saturating_sub(self.waited());
+        let started = tokio::time::Instant::now();
+        // `timeout` polls `acquire` before its timer, so a free gate is taken
+        // even when the budget is already spent.
+        let acquired = tokio::time::timeout(remaining, acquire).await;
+        let waited = self.record_wait(started.elapsed());
+        acquired.map_err(|_| Error::Lock(rustfs_lock::LockError::timeout(resource, waited)))
+    }
+
+    fn waited(&self) -> std::time::Duration {
+        std::time::Duration::from_nanos(self.waited_nanos.load(std::sync::atomic::Ordering::Acquire))
+    }
+
+    fn record_wait(&self, elapsed: std::time::Duration) -> std::time::Duration {
+        let elapsed = u64::try_from(elapsed.as_nanos()).unwrap_or(u64::MAX);
+        let previous = self
+            .waited_nanos
+            .try_update(std::sync::atomic::Ordering::AcqRel, std::sync::atomic::Ordering::Acquire, |waited| {
+                Some(waited.saturating_add(elapsed))
+            })
+            .unwrap_or_else(|waited| waited);
+        std::time::Duration::from_nanos(previous.saturating_add(elapsed))
+    }
+}
+
+/// Wait for `acquire` within `budget`, or without bound when there is none.
+pub(crate) async fn acquire_gate_within<F: std::future::Future>(
+    budget: Option<&GateWaitBudget>,
+    resource: &'static str,
+    acquire: F,
+) -> Result<F::Output> {
+    match budget {
+        Some(budget) => budget.acquire(resource, acquire).await,
+        None => Ok(acquire.await),
+    }
+}
+
 pub fn is_pool_activation_fleet_proof_error(err: &Error) -> bool {
    // Save-stage helpers add context by formatting the original error, so the
    // marker may be nested in the display string. Restrict matching to the
@@ -10143,6 +10212,55 @@ impl DecommissionPoolCapacityInfo {
    }
 }

+#[cfg(test)]
+static DECOMMISSION_START_GATE_WAIT_OVERRIDES: std::sync::OnceLock<std::sync::Mutex<HashMap<uuid::Uuid, std::time::Duration>>> =
+    std::sync::OnceLock::new();
+
+#[cfg(test)]
+pub(crate) fn set_decommission_start_gate_wait_for_test(store_id: uuid::Uuid, wait: std::time::Duration) {
+    DECOMMISSION_START_GATE_WAIT_OVERRIDES
+        .get_or_init(|| std::sync::Mutex::new(HashMap::new()))
+        .lock()
+        .expect("decommission start gate wait override should not be poisoned")
+        .insert(store_id, wait);
+}
+
+/// Parks one decommission start of a store just before it waits for the
+/// activation movement writer, so a test can change gate state there.
+#[cfg(test)]
+#[derive(Default)]
+pub(crate) struct DecommissionActivationPause {
+    pub(crate) reached: tokio::sync::Notify,
+    pub(crate) proceed: tokio::sync::Notify,
+}
+
+#[cfg(test)]
+static DECOMMISSION_ACTIVATION_PAUSES: std::sync::OnceLock<
+    std::sync::Mutex<HashMap<uuid::Uuid, Arc<DecommissionActivationPause>>>,
+> = std::sync::OnceLock::new();
+
+#[cfg(test)]
+pub(crate) fn install_decommission_activation_pause_for_test(store_id: uuid::Uuid) -> Arc<DecommissionActivationPause> {
+    let pause = Arc::new(DecommissionActivationPause::default());
+    DECOMMISSION_ACTIVATION_PAUSES
+        .get_or_init(|| std::sync::Mutex::new(HashMap::new()))
+        .lock()
+        .expect("decommission activation pause should not be poisoned")
+        .insert(store_id, Arc::clone(&pause));
+    pause
+}
+
+#[cfg(test)]
+async fn pause_decommission_activation_for_test(store_id: uuid::Uuid) {
+    let pause = DECOMMISSION_ACTIVATION_PAUSES
+        .get()
+        .and_then(|pauses| pauses.lock().ok()?.remove(&store_id));
+    if let Some(pause) = pause {
+        pause.reached.notify_one();
+        pause.proceed.notified().await;
+    }
+}
+
 #[cfg(any(test, feature = "test-util"))]
 type DecommissionCapacityInfoOverrides =
    std::sync::Mutex<HashMap<uuid::Uuid, std::collections::VecDeque<Vec<DecommissionPoolCapacityInfo>>>>;
@@ -13009,17 +13127,37 @@ impl ECStore {
        Ok(())
    }

-    async fn ensure_decommission_rebalance_idle_after_refresh(&self) -> Result<()> {
-        self.load_rebalance_meta().await?;
+    fn decommission_start_gate_budget(&self) -> GateWaitBudget {
+        #[cfg(test)]
+        if let Some(wait) = DECOMMISSION_START_GATE_WAIT_OVERRIDES
+            .get()
+            .and_then(|overrides| overrides.lock().ok()?.get(&self.id).copied())
+        {
+            return GateWaitBudget::new(wait);
+        }
+        GateWaitBudget::new(DECOMMISSION_START_GATE_WAIT)
+    }
+
+    /// Refresh rebalance metadata ahead of an operator decommission start
+    /// (the leader-local admin pre-check), with the start's gate budget.
+    pub async fn load_rebalance_meta_for_decommission_start(&self) -> Result<()> {
+        self.load_rebalance_meta_within(Some(&self.decommission_start_gate_budget()))
+            .await
+    }
+
+    async fn ensure_decommission_rebalance_idle_after_refresh(&self, budget: &GateWaitBudget) -> Result<()> {
+        self.load_rebalance_meta_within(Some(budget)).await?;
        ensure_decommission_not_rebalancing(self.is_rebalance_conflicting_with_decommission().await)
    }

-    async fn ensure_decommission_rebalance_idle_after_refresh_under_start_gate(&self) -> Result<()> {
-        self.load_rebalance_meta_under_start_gate().await?;
+    async fn ensure_decommission_rebalance_idle_after_refresh_under_start_gate(&self, budget: &GateWaitBudget) -> Result<()> {
+        self.load_rebalance_meta_under_start_gate_within(Some(budget)).await?;
        ensure_decommission_not_rebalancing(self.is_rebalance_conflicting_with_decommission().await)
    }

-    pub async fn status(&self, idx: usize) -> Result<PoolStatus> {
+    /// Pool status for an admin read, computed from a snapshot returned by
+    /// [`ECStore::pool_status_meta_snapshot`].
+    pub async fn status(&self, pool_meta: &PoolMeta, idx: usize) -> Result<PoolStatus> {
        let capacity_infos = self.get_decommission_all_pool_capacity_infos().await?;
        let space_info = capacity_infos
            .iter()
@@ -13027,8 +13165,7 @@ impl ECStore {
            .map(|capacity| capacity.space)
            .ok_or_else(|| invalid_decommission_pool_index_error(self.pools.len(), idx))?;

-        let pool_meta = self.pool_meta.read().await;
-        let active_target_reservations = active_decommission_target_reservations(&pool_meta);
+        let active_target_reservations = active_decommission_target_reservations(pool_meta);

        let mut pool_info = get_by_index(pool_meta.pools.as_slice(), idx, "fetch decommission status")?.clone();
        if let Some(info) = pool_info.decommission.as_mut() {
@@ -13037,19 +13174,41 @@ impl ECStore {
        Ok(apply_decommission_status_space_info(pool_info, space_info))
    }

+    /// Publish persisted pool metadata into memory, waiting for in-flight
+    /// data movement to drain. Startup uses it before any movement runs.
    #[tracing::instrument(skip_all)]
    pub async fn refresh_pool_status_meta(&self) -> Result<()> {
        let movement_gate = self.ctx.data_movement_operation_gate();
-        let _movement_guard = movement_gate.write().await;
-        let persisted = self.load_runtime_pool_meta("refresh pool status metadata failed").await?;
+        let movement_guard = movement_gate.write_owned().await;
+        self.publish_pool_status_refresh(movement_guard).await
+    }

-        let active_workers = {
-            let cancelers = self.decommission_cancelers.read().await;
-            cancelers
-                .iter()
-                .map(|canceler| canceler.as_ref().is_some_and(DecommissionCanceler::is_active))
-                .collect::<Vec<_>>()
-        };
+    /// Pool metadata for an admin status read. It never waits for data
+    /// movement: a movement side effect holds a read on the movement gate for
+    /// as long as it runs, and a queued writer would also stall every new
+    /// movement read behind it. With the gate free, the refresh publishes as
+    /// before. Otherwise the persisted snapshot is merged into a private copy
+    /// with the same monotonic rule and nothing is published, so no movement
+    /// state changes while an operation that sampled the old state is running.
+    #[tracing::instrument(skip_all)]
+    pub async fn pool_status_meta_snapshot(&self) -> Result<PoolMeta> {
+        if let Ok(movement_guard) = self.ctx.data_movement_operation_gate().try_write_owned() {
+            self.publish_pool_status_refresh(movement_guard).await?;
+            return Ok(self.pool_meta.read().await.clone());
+        }
+
+        let persisted = self.load_runtime_pool_meta("read pool status metadata failed").await?;
+        let active_workers = self.decommission_worker_activity().await;
+        let mut snapshot = self.pool_meta.read().await.clone();
+        merge_pool_status_refresh(&mut snapshot, persisted, &active_workers);
+        Ok(snapshot)
+    }
+
+    /// Load and publish under the movement writer, so no movement read sees
+    /// the transition mid-operation and no transition interleaves the load.
+    async fn publish_pool_status_refresh(&self, _movement_guard: tokio::sync::OwnedRwLockWriteGuard<()>) -> Result<()> {
+        let persisted = self.load_runtime_pool_meta("refresh pool status metadata failed").await?;
+        let active_workers = self.decommission_worker_activity().await;

        let mut pool_meta = self.pool_meta.write().await;
        if merge_pool_status_refresh(&mut pool_meta, persisted, &active_workers) {
@@ -13058,6 +13217,14 @@ impl ECStore {
        Ok(())
    }

+    async fn decommission_worker_activity(&self) -> Vec<bool> {
+        let cancelers = self.decommission_cancelers.read().await;
+        cancelers
+            .iter()
+            .map(|canceler| canceler.as_ref().is_some_and(DecommissionCanceler::is_active))
+            .collect()
+    }
+
    async fn get_decommission_pool_capacity_info(
        &self,
        idx: usize,
@@ -14080,12 +14247,13 @@ impl ECStore {
        );
        validate_start_decommission_request(&indices, self.single_pool())?;

-        self.ensure_decommission_rebalance_idle_after_refresh().await?;
+        let budget = self.decommission_start_gate_budget();
+        self.ensure_decommission_rebalance_idle_after_refresh(&budget).await?;

        let store = require_decommission_store(runtime_sources::object_store_handle(), "start decommission")?;
        let local_indices = local_decommission_queue_prefix(&self.endpoints(), &indices)?;
        let index_cancelers = self
-            .start_decommission_with_routines(indices, &rx, local_indices.as_slice())
+            .start_decommission_inner(indices, Some((&rx, local_indices.as_slice())), &budget)
            .await?;
        drop(spawn_decommission_index_cancelers(
            store,
@@ -17148,27 +17316,22 @@ impl ECStore {

    #[tracing::instrument(skip(self))]
    pub async fn start_decommission(&self, indices: Vec<usize>) -> Result<()> {
-        self.start_decommission_inner(indices, None).await.map(|_| ())
-    }
-
-    async fn start_decommission_with_routines(
-        &self,
-        indices: Vec<usize>,
-        rx: &CancellationToken,
-        local_indices: &[usize],
-    ) -> Result<Vec<(usize, DecommissionCancelerGuard)>> {
-        self.start_decommission_inner(indices, Some((rx, local_indices))).await
+        let budget = self.decommission_start_gate_budget();
+        self.start_decommission_inner(indices, None, &budget).await.map(|_| ())
    }

+    /// Every gate this start waits for draws on `budget`; once it is spent the
+    /// start fails with a retryable lock timeout before anything is persisted.
    async fn start_decommission_inner(
        &self,
        indices: Vec<usize>,
        reservation: Option<(&CancellationToken, &[usize])>,
+        budget: &GateWaitBudget,
    ) -> Result<Vec<(usize, DecommissionCancelerGuard)>> {
        let indices = dedup_indices(&indices);
        validate_start_decommission_request(&indices, self.single_pool())?;

-        self.ensure_decommission_rebalance_idle_after_refresh().await?;
+        self.ensure_decommission_rebalance_idle_after_refresh(budget).await?;
        #[cfg(test)]
        let endpoints = self.instance_endpoints().unwrap_or_else(|| self.endpoints());
        #[cfg(not(test))]
@@ -17210,8 +17373,10 @@ impl ECStore {
            }
        }

-        let _start_guard = self.start_gate.lock().await;
-        self.ensure_decommission_rebalance_idle_after_refresh_under_start_gate()
+        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?;

        let all_capacity_infos = self.get_decommission_all_pool_capacity_infos_with_owned_usage().await?;
@@ -17221,7 +17386,14 @@ impl ECStore {
        // object operations can observe the signal and release read guards.
        self.cancel_decommission_routines(&indices).await;
        let movement_gate = self.ctx.data_movement_operation_gate();
-        let _movement_guard = movement_gate.write().await;
+        #[cfg(test)]
+        pause_decommission_activation_for_test(self.id).await;
+        // The writer waits for every in-flight movement read; bound the wait
+        // so one slow or stalled side effect fails this start instead of
+        // holding it, and every movement read queued behind it, open.
+        let _movement_guard = budget
+            .acquire(DECOMMISSION_ACTIVATION_MOVEMENT_GATE, movement_gate.write())
+            .await?;

        let index_cancelers = if let Some((rx, local_indices)) = reservation {
            // Lock order matches terminal transitions: movement gate, then
@@ -19523,6 +19695,366 @@ mod tests {
        assert!(!is_pool_activation_fleet_proof_error(&Error::ConfigNotFound));
    }

+    #[tokio::test]
+    #[serial_test::serial]
+    async fn pool_status_snapshot_does_not_wait_for_a_stuck_movement_side_effect() {
+        let (_temp_dirs, store, _other_store) = Box::pin(crate::services::rebalance::test_two_pool_stores(None)).await;
+        crate::services::rebalance::promote_test_pool_meta_to_v2(&store).await;
+        let in_memory = store.pool_meta.read().await.clone();
+        let mut persisted = in_memory.clone();
+        persisted.pools[1].last_update = in_memory.pools[1].last_update + Duration::minutes(1);
+        persisted.pools[1].decommission = Some(PoolDecommissionInfo {
+            items_decommissioned: 7,
+            ..Default::default()
+        });
+        persisted
+            .save(store.pools.clone())
+            .await
+            .expect("another node's newer pool status should be persisted");
+
+        // A movement side effect that never finishes keeps its read admission.
+        let stuck_side_effect = store.ctx.data_movement_operation_gate().read_owned().await;
+        let epoch = store.ctx.data_movement_operation_epoch();
+        let snapshot = tokio::time::timeout(std::time::Duration::from_secs(10), store.pool_status_meta_snapshot())
+            .await
+            .expect("a status read must not wait for in-flight data movement")
+            .expect("the persisted pool status should be readable");
+        assert_eq!(
+            snapshot.pools[1].decommission.as_ref().map(|info| info.items_decommissioned),
+            Some(7),
+            "the status read reports the persisted progress"
+        );
+        assert_eq!(
+            store.pool_meta.read().await.pools[1]
+                .decommission
+                .as_ref()
+                .map(|info| info.items_decommissioned),
+            in_memory.pools[1].decommission.as_ref().map(|info| info.items_decommissioned),
+            "a busy movement gate must keep the refresh from publishing"
+        );
+        assert_eq!(store.ctx.data_movement_operation_epoch(), epoch);
+
+        set_decommission_capacity_info_overrides_for_test(
+            store.id,
+            vec![vec![
+                DecommissionPoolCapacityInfo::for_test(0, DecommissionErasureLayout { data: 1, parity: 0 }, 0, 100, 100),
+                DecommissionPoolCapacityInfo::for_test(1, DecommissionErasureLayout { data: 1, parity: 0 }, 200, 200, 0),
+            ]],
+        );
+        let status = tokio::time::timeout(std::time::Duration::from_secs(10), store.status(&snapshot, 1))
+            .await
+            .expect("pool status must not wait for in-flight data movement")
+            .expect("pool status should be computed from the snapshot");
+        assert_eq!(status.decommission.map(|info| info.items_decommissioned), Some(7));
+
+        drop(stuck_side_effect);
+        store
+            .pool_status_meta_snapshot()
+            .await
+            .expect("an idle movement gate should allow the refresh to publish");
+        assert_eq!(
+            store.pool_meta.read().await.pools[1]
+                .decommission
+                .as_ref()
+                .map(|info| info.items_decommissioned),
+            Some(7)
+        );
+        assert!(
+            store.ctx.data_movement_operation_epoch() > epoch,
+            "a published refresh advances the epoch"
+        );
+    }
+
+    /// A two-pool store whose pool 0 can start decommissioning onto pool 1.
+    async fn startable_decommission_store() -> (Vec<tempfile::TempDir>, Arc<ECStore>, Arc<ECStore>) {
+        let (temp_dirs, store, other_store) =
+            Box::pin(crate::services::rebalance::test_two_pool_stores_with_isolated_node_contexts(None)).await;
+        crate::services::rebalance::promote_test_pool_meta_to_v2(&store).await;
+        let layout = DecommissionErasureLayout { data: 1, parity: 0 };
+        set_decommission_capacity_info_overrides_for_test(
+            store.id,
+            vec![vec![
+                DecommissionPoolCapacityInfo::for_test(0, layout, 0, 100, 100),
+                DecommissionPoolCapacityInfo::for_test(1, layout, 200, 200, 0),
+            ]],
+        );
+        (temp_dirs, store, other_store)
+    }
+
+    fn timed_out_gate(err: &Error) -> Option<(&str, std::time::Duration)> {
+        match err {
+            Error::Lock(rustfs_lock::LockError::Timeout { resource, timeout }) => Some((resource.as_str(), *timeout)),
+            _ => None,
+        }
+    }
+
+    async fn assert_start_left_no_trace(store: &ECStore, baseline: &PoolMetaSelection) {
+        assert!(
+            store
+                .pool_meta
+                .read()
+                .await
+                .pools
+                .iter()
+                .all(|pool| pool.decommission.is_none()),
+            "a timed-out start must not change in-memory pool state"
+        );
+        let after = load_pool_meta_replicas(store.pools.clone(), true)
+            .await
+            .expect("pool metadata should stay readable");
+        assert_eq!(after.canonical, baseline.canonical, "a timed-out start must not persist anything");
+        // The abandoned writer must not keep movement reads queued either.
+        let next_side_effect =
+            tokio::time::timeout(std::time::Duration::from_secs(1), store.ctx.data_movement_operation_gate().read_owned())
+                .await
+                .expect("movement reads must not stay queued behind an abandoned start");
+        drop(next_side_effect);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn gate_wait_budget_counts_only_time_spent_waiting() {
+        let budget = GateWaitBudget::new(std::time::Duration::from_millis(100));
+        let gate = Arc::new(tokio::sync::RwLock::new(()));
+
+        let held = Arc::clone(&gate).read_owned().await;
+        let err = budget
+            .acquire("test gate", gate.write())
+            .await
+            .expect_err("a held gate must time out once the budget is spent");
+        assert_eq!(
+            timed_out_gate(&err),
+            Some(("test gate", std::time::Duration::from_millis(100))),
+            "the timeout reports the time actually waited"
+        );
+        drop(held);
+
+        // Time spent outside gate waits does not count, and a free gate is
+        // taken even with nothing left to wait.
+        tokio::time::advance(std::time::Duration::from_secs(60)).await;
+        drop(
+            budget
+                .acquire("test gate", gate.write())
+                .await
+                .expect("a free gate is taken with the budget spent"),
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn gate_wait_budget_is_shared_across_waits() {
+        let budget = GateWaitBudget::new(std::time::Duration::from_millis(100));
+        let gate_a = Arc::new(tokio::sync::RwLock::new(()));
+        let gate_b = Arc::new(tokio::sync::RwLock::new(()));
+
+        let held_a = Arc::clone(&gate_a).read_owned().await;
+        tokio::spawn(async move {
+            tokio::time::sleep(std::time::Duration::from_millis(60)).await;
+            drop(held_a);
+        });
+        let started = tokio::time::Instant::now();
+        drop(
+            budget
+                .acquire("a", gate_a.write())
+                .await
+                .expect("gate a frees within the budget"),
+        );
+        assert_eq!(started.elapsed(), std::time::Duration::from_millis(60));
+
+        let _held_b = Arc::clone(&gate_b).read_owned().await;
+        let started = tokio::time::Instant::now();
+        let err = budget
+            .acquire("b", gate_b.write())
+            .await
+            .expect_err("gate b gets only what gate a left of the budget");
+        assert_eq!(started.elapsed(), std::time::Duration::from_millis(40));
+        assert_eq!(timed_out_gate(&err), Some(("b", std::time::Duration::from_millis(100))));
+    }
+
+    #[tokio::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));
+
+        // 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]
+    #[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)
+        };
+
+        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;
+    }
+
+    #[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);
+    }
+
+    #[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_activation_pause_for_test(store.id);
+
+        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);
+    }
+
+    #[tokio::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_activation_pause_for_test(store.id);
+
+        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]
    #[serial_test::serial]
    async fn decommission_v1_start_preflights_reject_before_metadata_writes() {
@@ -83,12 +83,16 @@ pub(crate) enum ScannerPublicationLeasePurpose {
    Observational,
 }

+/// Movement read admission granted with a publication lease. Lease-scoped
+/// target operations share it rather than queueing a second read on the gate.
+pub(crate) type ScannerPublicationLeasePermit = Arc<OwnedRwLockReadGuard<()>>;
+
 pub(crate) struct ScannerPublicationLeaseEntry {
    pub(crate) expires_at: Instant,
    pub(crate) movement_generation: u64,
    pub(crate) namespace_generation: Option<u64>,
    pub(crate) purpose: ScannerPublicationLeasePurpose,
-    pub(crate) _operation_guard: OwnedRwLockReadGuard<()>,
+    pub(crate) operation_guard: ScannerPublicationLeasePermit,
 }

 pub(crate) struct NamespaceCommitGuard {
@@ -365,7 +369,7 @@ impl InstanceContext {
                movement_generation,
                namespace_generation,
                purpose,
-                _operation_guard: operation_guard,
+                operation_guard: Arc::new(operation_guard),
            },
        );
        true
@@ -383,16 +387,35 @@ impl InstanceContext {
        &self,
        token: Uuid,
    ) -> Option<(u64, Option<u64>, ScannerPublicationLeasePurpose)> {
+        self.scanner_publication_lease_admission(token)
+            .await
+            .map(|(movement_generation, namespace_generation, purpose, _)| (movement_generation, namespace_generation, purpose))
+    }
+
+    /// Return the live lease's generations and purpose together with a share
+    /// of its movement read admission. The share keeps movement writers out
+    /// for as long as the caller holds it, even if the lease is released or
+    /// expires meanwhile, without queueing a new read behind a waiting writer.
+    pub(crate) async fn scanner_publication_lease_admission(
+        &self,
+        token: Uuid,
+    ) -> Option<(u64, Option<u64>, ScannerPublicationLeasePurpose, ScannerPublicationLeasePermit)> {
        let mut leases = self.scanner_publication_leases.lock().await;
        let now = Instant::now();
-        let (expires_at, movement_generation, namespace_generation, purpose) = leases
-            .get(&token)
-            .map(|entry| (entry.expires_at, entry.movement_generation, entry.namespace_generation, entry.purpose))?;
+        let (expires_at, movement_generation, namespace_generation, purpose, permit) = leases.get(&token).map(|entry| {
+            (
+                entry.expires_at,
+                entry.movement_generation,
+                entry.namespace_generation,
+                entry.purpose,
+                Arc::clone(&entry.operation_guard),
+            )
+        })?;
        if expires_at <= now {
            leases.remove(&token);
            return None;
        }
-        Some((movement_generation, namespace_generation, purpose))
+        Some((movement_generation, namespace_generation, purpose, permit))
    }

    pub(crate) async fn expire_scanner_publication_lease(&self, token: Uuid, expires_at: Instant) {
@@ -18,8 +18,9 @@ use super::{
    encode_rebalance_stop_propagation_record,
 };
 use crate::core::pools::{
-    PoolMeta, PoolRebalanceActivationFence, PoolRebalanceWorkerFence, acquire_pool_activation_fleet_proof,
-    acquire_pool_rebalance_activation_locks, acquire_pool_rebalance_worker_locks, pool_meta_has_active_decommission,
+    GateWaitBudget, PoolMeta, PoolRebalanceActivationFence, PoolRebalanceWorkerFence, acquire_gate_within,
+    acquire_pool_activation_fleet_proof, acquire_pool_rebalance_activation_locks, acquire_pool_rebalance_worker_locks,
+    pool_meta_has_active_decommission,
 };
 use crate::error::{Error, Result};
 use crate::object_api::ObjectOptions;
@@ -36,6 +37,10 @@ use time::OffsetDateTime;
 use tracing::{debug, info};
 use uuid::Uuid;

+pub(crate) const REBALANCE_REFRESH_START_GATE: &str = "start gate (rebalance metadata refresh)";
+pub(crate) const REBALANCE_REFRESH_ACTIVATION_GATE: &str = "rebalance activation gate (rebalance metadata refresh)";
+pub(crate) const REBALANCE_REFRESH_MOVEMENT_GATE: &str = "data movement gate (rebalance metadata refresh)";
+
 #[cfg(test)]
 static FAIL_NEXT_REBALANCE_ACTIVATION_SAVE: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);

@@ -383,6 +388,15 @@ fn merge_rebalance_status_refresh(current: &mut Option<RebalanceMeta>, persisted
    }
 }

+/// Apply a persisted status refresh (`None` when no rebalance was recorded).
+/// Returns whether the movement snapshot changed.
+fn merge_persisted_rebalance_status(current: &mut Option<RebalanceMeta>, persisted: Option<RebalanceMeta>) -> bool {
+    match persisted {
+        Some(persisted) => merge_rebalance_status_refresh(current, persisted),
+        None => clear_rebalance_status_refresh(current),
+    }
+}
+
 fn clear_rebalance_status_refresh(current: &mut Option<RebalanceMeta>) -> bool {
    if current.as_ref().is_none_or(|meta| !is_rebalance_actively_running(meta)) {
        current.take().is_some()
@@ -417,6 +431,15 @@ impl ECStore {
        &self,
        expected_id: Option<&str>,
        stage: &str,
+    ) -> Result<Option<tokio::sync::OwnedRwLockWriteGuard<()>>> {
+        self.rebalance_activation_write_guard_within(expected_id, stage, None).await
+    }
+
+    async fn rebalance_activation_write_guard_within(
+        &self,
+        expected_id: Option<&str>,
+        stage: &str,
+        budget: Option<&GateWaitBudget>,
    ) -> Result<Option<tokio::sync::OwnedRwLockWriteGuard<()>>> {
        let activation_gate = {
            let meta = self.rebalance_meta.read().await;
@@ -426,7 +449,7 @@ impl ECStore {
            meta.as_ref().map(|meta| Arc::clone(&meta.activation_gate))
        };
        Ok(match activation_gate {
-            Some(gate) => Some(gate.write_owned().await),
+            Some(gate) => Some(acquire_gate_within(budget, REBALANCE_REFRESH_ACTIVATION_GATE, gate.write_owned()).await?),
            None => None,
        })
    }
@@ -656,8 +679,14 @@ impl ECStore {

    #[tracing::instrument(skip_all)]
    pub async fn load_rebalance_meta(&self) -> Result<()> {
-        let _start_guard = self.start_gate.lock().await;
-        self.load_rebalance_meta_under_start_gate().await
+        self.load_rebalance_meta_within(None).await
+    }
+
+    /// `budget` bounds the waits for the start, activation and movement gates;
+    /// once it is spent the load fails before it reads or publishes anything.
+    pub(crate) async fn load_rebalance_meta_within(&self, budget: Option<&GateWaitBudget>) -> Result<()> {
+        let _start_guard = acquire_gate_within(budget, REBALANCE_REFRESH_START_GATE, self.start_gate.lock()).await?;
+        self.load_rebalance_meta_under_start_gate_within(budget).await
    }

    /// Cancels local admission before refreshing the persisted stop target under the start gate.
@@ -713,6 +742,12 @@ impl ECStore {
    }

    pub(crate) async fn load_rebalance_meta_under_start_gate(&self) -> Result<()> {
+        self.load_rebalance_meta_under_start_gate_within(None).await
+    }
+
+    /// `budget` bounds the waits for the activation and movement writers;
+    /// once it is spent the load fails before it reads or publishes anything.
+    pub(crate) async fn load_rebalance_meta_under_start_gate_within(&self, budget: Option<&GateWaitBudget>) -> Result<()> {
        let mut meta = RebalanceMeta::new();
        debug!(
            event = EVENT_REBALANCE_STATE,
@@ -722,9 +757,11 @@ impl ECStore {
            "Loading rebalance metadata"
        );
        let pool = clone_first_arc(&self.pools, "rebalanceMeta: no pools available")?;
-        let _activation_guard = self.rebalance_activation_write_guard(None, "load rebalance metadata").await?;
+        let _activation_guard = self
+            .rebalance_activation_write_guard_within(None, "load rebalance metadata", budget)
+            .await?;
        let movement_gate = self.ctx.data_movement_operation_gate();
-        let _movement_guard = movement_gate.write().await;
+        let _movement_guard = acquire_gate_within(budget, REBALANCE_REFRESH_MOVEMENT_GATE, movement_gate.write()).await?;
        let loaded = resolve_rebalance_meta_load_result(meta.load(pool).await)?;
        if loaded {
            let movement_changed = rebalance_movement_snapshot_changed(self.rebalance_meta.read().await.as_ref(), &meta);
@@ -777,36 +814,60 @@ impl ECStore {
        Ok(())
    }

+    /// Rebalance metadata for an admin status read. It never waits on the
+    /// start, activation or data-movement gates, which a pending start or an
+    /// in-flight movement side effect can hold for as long as that work runs.
+    /// With all three free, the refresh publishes as before. Otherwise the
+    /// persisted metadata is merged into a private copy and nothing is
+    /// published, so no movement transition becomes visible here without the
+    /// movement writer.
    #[tracing::instrument(skip_all)]
-    pub async fn refresh_rebalance_status_meta(&self) -> Result<()> {
-        let _start_guard = self.start_gate.lock().await;
-        let pool = clone_first_arc(&self.pools, "refresh_rebalance_status_meta: no pools available")?;
-        let mut persisted = RebalanceMeta::new();
-        let _activation_guard = self
-            .rebalance_activation_write_guard(None, "refresh rebalance metadata")
-            .await?;
-        let movement_gate = self.ctx.data_movement_operation_gate();
-        let _movement_guard = movement_gate.write().await;
-        let loaded = persisted.load(pool).await;
-        match loaded {
-            Ok(()) => {
+    pub async fn rebalance_status_meta_snapshot(&self) -> Result<Option<RebalanceMeta>> {
+        // Transition order is start_gate -> activation_gate -> movement gate -> rebalance_meta.
+        if let Ok(_start_guard) = self.start_gate.try_lock() {
+            let activation_gate = self
+                .rebalance_meta
+                .read()
+                .await
+                .as_ref()
+                .map(|meta| Arc::clone(&meta.activation_gate));
+            let activation_guard = match activation_gate {
+                Some(gate) => gate.try_write_owned().ok().map(Some),
+                None => Some(None),
+            };
+            if let Some(_activation_guard) = activation_guard
+                && let Ok(_movement_guard) = self.ctx.data_movement_operation_gate().try_write_owned()
+            {
+                let persisted = self.load_rebalance_status_meta().await?;
                let mut rebalance_meta = self.rebalance_meta.write().await;
-                if merge_rebalance_status_refresh(&mut rebalance_meta, persisted) {
+                if merge_persisted_rebalance_status(&mut rebalance_meta, persisted) {
                    self.ctx.advance_data_movement_operation_epoch();
                }
-            }
-            Err(Error::ConfigNotFound) => {
-                let mut rebalance_meta = self.rebalance_meta.write().await;
-                if clear_rebalance_status_refresh(&mut rebalance_meta) {
-                    self.ctx.advance_data_movement_operation_epoch();
-                }
-            }
-            Err(err) => {
-                return Err(Error::other(format!("rebalance metadata refresh failed during pool status: {err}")));
+                return Ok(rebalance_meta.clone());
            }
        }

-        Ok(())
+        let persisted = self.load_rebalance_status_meta().await?;
+        let mut snapshot = self.rebalance_meta.read().await.clone();
+        merge_persisted_rebalance_status(&mut snapshot, persisted);
+        Ok(snapshot)
+    }
+
+    /// Persisted rebalance metadata, or `None` when no rebalance was recorded.
+    async fn load_rebalance_status_meta(&self) -> Result<Option<RebalanceMeta>> {
+        let pool = clone_first_arc(&self.pools, "refresh_rebalance_status_meta: no pools available")?;
+        let mut persisted = RebalanceMeta::new();
+        match persisted.load(pool).await {
+            Ok(()) => Ok(Some(persisted)),
+            Err(Error::ConfigNotFound) => Ok(None),
+            Err(err) => Err(Error::other(format!("rebalance metadata refresh failed during pool status: {err}"))),
+        }
+    }
+
+    /// Rebalance status of one pool in a snapshot returned by
+    /// [`ECStore::rebalance_status_meta_snapshot`].
+    pub fn pool_rebalance_status(rebalance_meta: Option<&RebalanceMeta>, pool_index: usize) -> (RebalStatus, bool) {
+        pool_rebalance_status_from_meta(rebalance_meta, pool_index)
    }

    #[tracing::instrument(skip_all)]
@@ -1234,11 +1295,6 @@ impl ECStore {
        false
    }

-    pub async fn pool_rebalance_status(&self, pool_index: usize) -> (RebalStatus, bool) {
-        let rebalance_meta = self.rebalance_meta.read().await;
-        pool_rebalance_status_from_meta(rebalance_meta.as_ref(), pool_index)
-    }
-
    pub async fn current_rebalance_id(&self) -> Option<String> {
        let rebalance_meta = self.rebalance_meta.read().await;
        rebalance_meta
@@ -1403,6 +1459,117 @@ mod tests {
    use crate::object_api::NamespaceLockFence;
    use crate::set_disk::{PutObjectCommitBarrier, PutObjectCommitPause, hermetic_set_disks_isolated};

+    #[tokio::test]
+    async fn rebalance_status_snapshot_does_not_wait_for_a_pending_start_or_stuck_movement() {
+        let id = "rebalance-status-snapshot";
+        let (_temp_dirs, store) = crate::services::rebalance::test_store_with_persisted_rebalance_meta(RebalanceMeta {
+            id: id.to_string(),
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                info: RebalanceInfo {
+                    status: RebalStatus::Completed,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        })
+        .await;
+        // This node has not observed the persisted run yet.
+        *store.rebalance_meta.write().await = None;
+
+        // A start waiting on the movement writer holds the start gate, and an
+        // in-flight movement side effect holds its read admission.
+        let pending_start = store.start_gate.lock().await;
+        let stuck_side_effect = store.ctx.data_movement_operation_gate().read_owned().await;
+        let epoch = store.ctx.data_movement_operation_epoch();
+        let snapshot = tokio::time::timeout(std::time::Duration::from_secs(10), store.rebalance_status_meta_snapshot())
+            .await
+            .expect("a status read must not wait for a start or in-flight data movement")
+            .expect("persisted rebalance metadata should be readable");
+        assert_eq!(snapshot.as_ref().map(|meta| meta.id.as_str()), Some(id));
+        assert_eq!(
+            ECStore::pool_rebalance_status(snapshot.as_ref(), 0),
+            (RebalStatus::Completed, false),
+            "the status read reports the persisted run"
+        );
+        assert!(
+            store.rebalance_meta.read().await.is_none(),
+            "busy gates must keep the refresh from publishing"
+        );
+        assert_eq!(store.ctx.data_movement_operation_epoch(), epoch);
+
+        drop(stuck_side_effect);
+        drop(pending_start);
+        store
+            .rebalance_status_meta_snapshot()
+            .await
+            .expect("free gates should allow the refresh to publish");
+        assert_eq!(
+            store.rebalance_meta.read().await.as_ref().map(|meta| meta.id.clone()),
+            Some(id.to_string())
+        );
+        assert!(
+            store.ctx.data_movement_operation_epoch() > epoch,
+            "a published refresh advances the epoch"
+        );
+    }
+
+    #[tokio::test]
+    async fn rebalance_status_snapshot_does_not_wait_for_an_in_flight_rebalance_entry() {
+        let id = "rebalance-status-activation";
+        let (_temp_dirs, store) = crate::services::rebalance::test_store_with_persisted_rebalance_meta(RebalanceMeta {
+            id: id.to_string(),
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                info: RebalanceInfo {
+                    status: RebalStatus::Completed,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        })
+        .await;
+        // This node knows the run but not yet its per-pool state.
+        let activation_gate = {
+            let mut rebalance_meta = store.rebalance_meta.write().await;
+            let meta = rebalance_meta.as_mut().expect("the fixture installs the run in memory");
+            meta.pool_stats.clear();
+            Arc::clone(&meta.activation_gate)
+        };
+
+        // An in-flight rebalance entry holds the activation read gate.
+        let in_flight_entry = activation_gate.read_owned().await;
+        let epoch = store.ctx.data_movement_operation_epoch();
+        let snapshot = tokio::time::timeout(std::time::Duration::from_secs(10), store.rebalance_status_meta_snapshot())
+            .await
+            .expect("a status read must not wait for an in-flight rebalance entry")
+            .expect("persisted rebalance metadata should be readable");
+        assert_eq!(
+            snapshot.as_ref().map(|meta| meta.pool_stats.len()),
+            Some(1),
+            "the status read reports the persisted pool state"
+        );
+        assert!(
+            store
+                .rebalance_meta
+                .read()
+                .await
+                .as_ref()
+                .is_some_and(|meta| meta.pool_stats.is_empty()),
+            "a busy activation gate must keep the refresh from publishing"
+        );
+        assert_eq!(store.ctx.data_movement_operation_epoch(), epoch);
+
+        drop(in_flight_entry);
+        store
+            .rebalance_status_meta_snapshot()
+            .await
+            .expect("free gates should allow the refresh to publish");
+        assert_eq!(store.rebalance_meta.read().await.as_ref().map(|meta| meta.pool_stats.len()), Some(1));
+    }
+
    #[tokio::test]
    async fn rebalance_status_read_recovers_after_metadata_lock_timeout() {
        let id = "rebalance-status-lock-retry";
@@ -1422,7 +1589,7 @@ mod tests {
            .expect("hold metadata writer");
        let retried = Arc::new(tokio::sync::Notify::new());
        let refresh = super::super::worker::REBALANCE_METADATA_RETRY_PROBE
-            .scope(Arc::clone(&retried), store.refresh_rebalance_status_meta());
+            .scope(Arc::clone(&retried), store.rebalance_status_meta_snapshot());
        tokio::pin!(refresh);
        tokio::time::timeout(std::time::Duration::from_secs(30), async {
            tokio::select! {
@@ -40,6 +40,8 @@ pub(crate) const REBALANCE_SOURCE_CLEANUP_DEFERRED_ERROR_PREFIX: &str = "deferre
 const REBALANCE_CLEANUP_WARNING_ENTRY_LIMIT: usize = 10;

 mod control;
+#[cfg(test)]
+pub(crate) use control::{REBALANCE_REFRESH_ACTIVATION_GATE, REBALANCE_REFRESH_MOVEMENT_GATE, REBALANCE_REFRESH_START_GATE};
 mod entry;
 mod meta;
 mod migration;
@@ -1236,7 +1236,10 @@ impl ECStore {
    /// the complete rename/write operation.  A lease token is process-owned;
    /// restart, expiry, generation changes, or blocked movement all reject it
    /// before the target disk is touched.
-    pub async fn acquire_scanner_publication_lease_guard(&self, token: Uuid) -> Result<tokio::sync::OwnedRwLockReadGuard<()>> {
+    pub async fn acquire_scanner_publication_lease_guard(
+        &self,
+        token: Uuid,
+    ) -> Result<Arc<tokio::sync::OwnedRwLockReadGuard<()>>> {
        self.acquire_scanner_publication_lease_guard_with_target(
            token,
            None,
@@ -1251,7 +1254,7 @@ impl ECStore {
        token: Uuid,
        volume: &str,
        path: &str,
-    ) -> Result<tokio::sync::OwnedRwLockReadGuard<()>> {
+    ) -> Result<Arc<tokio::sync::OwnedRwLockReadGuard<()>>> {
        let target = format!(
            "{}/{}",
            crate::disk::BUCKET_META_PREFIX,
@@ -1268,18 +1271,24 @@ impl ECStore {
        observed_target: Option<(&str, &str)>,
        validation_only: bool,
        expected_purpose: Option<ScannerPublicationLeasePurpose>,
-    ) -> Result<tokio::sync::OwnedRwLockReadGuard<()>> {
-        let operation_gate = self.ctx.data_movement_operation_gate();
-        let operation_guard = operation_gate.read_owned().await;
+    ) -> Result<Arc<tokio::sync::OwnedRwLockReadGuard<()>>> {
+        // Share the read admission the lease already holds instead of queueing
+        // a new read. A movement writer may already be waiting for this very
+        // lease; on the write-preferring gate a new read would queue behind
+        // that writer, so the coordinator could not finish the publication
+        // that releases the lease and both would stall until lease expiry.
+        // The shared admission has kept movement writers out since the grant
+        // and outlives a release or expiry until the caller drops it.
+        let Some((lease_generation, _, purpose, operation_guard)) = self.ctx.scanner_publication_lease_admission(token).await
+        else {
+            return Err(Error::other("scanner publication lease is unknown or expired"));
+        };
        if self.ctx.data_movement_generation_exhausted() || self.ctx.data_movement_operation_epoch_exhausted() {
            return Err(Error::other("scanner publication lease generation is exhausted"));
        }
        if self.scanner_data_movement_snapshot_locked().await.1 {
            return Err(Error::other("scanner publication lease is blocked by data movement"));
        }
-        let Some((lease_generation, _, purpose)) = self.ctx.scanner_publication_lease_generations(token).await else {
-            return Err(Error::other("scanner publication lease is unknown or expired"));
-        };
        if expected_purpose.is_some_and(|expected| expected != purpose) {
            return Err(Error::other("scanner publication lease purpose does not match its request"));
        }
@@ -2670,6 +2679,54 @@ mod tests {
        assert!(!store.release_scanner_publication_lease(expiring_token).await);
    }

+    #[tokio::test]
+    async fn lease_scoped_target_admission_does_not_queue_behind_a_waiting_movement_writer() {
+        let store = build_store_with_ctx(Arc::new(InstanceContext::new()));
+        let (token, generation) = store
+            .acquire_scanner_publication_lease(0, crate::runtime::instance::SCANNER_PUBLICATION_LEASE_TTL)
+            .await
+            .expect("an idle store should grant a publication lease");
+
+        // A decommission start or a status refresh queues a movement writer
+        // behind the lease's read admission.
+        let gate = store.ctx.data_movement_operation_gate();
+        let mut movement_writer = tokio::spawn(gate.clone().write_owned());
+        tokio::time::timeout(Duration::from_secs(5), async {
+            // A queued writer claims every spare permit, so reads stop succeeding.
+            while gate.try_read().is_ok() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("the movement writer should queue behind the lease");
+
+        // The coordinator's validation and target rename for that lease must
+        // proceed; otherwise neither side progresses until the lease expires.
+        tokio::time::timeout(Duration::from_secs(5), store.validate_scanner_publication_lease(token, generation))
+            .await
+            .expect("lease validation must not queue behind the writer waiting for that lease")
+            .expect("a live lease validates");
+        let rename_guard = tokio::time::timeout(
+            Duration::from_secs(5),
+            store.acquire_scanner_publication_lease_guard_for_rename(token, RUSTFS_META_BUCKET, "buckets/.usage.v2.json"),
+        )
+        .await
+        .expect("a lease-scoped rename must not queue behind the writer waiting for that lease")
+        .expect("a live lease admits its target rename");
+
+        assert!(store.release_scanner_publication_lease(token).await);
+        tokio::task::yield_now().await;
+        assert!(
+            !movement_writer.is_finished(),
+            "the target rename keeps movement out after the lease is released"
+        );
+        drop(rename_guard);
+        tokio::time::timeout(Duration::from_secs(5), &mut movement_writer)
+            .await
+            .expect("the movement writer should proceed once the rename finishes")
+            .expect("movement writer task should not panic");
+    }
+
    #[tokio::test]
    async fn observational_lease_only_allows_observed_usage_rename_during_namespace_commits() {
        let ctx = Arc::new(InstanceContext::new());
@@ -243,11 +243,11 @@ impl ECStore {
        scanner_token: Option<Uuid>,
    ) -> DiskResult<RenameDataResp> {
        let external_guard: Option<Arc<dyn Send + Sync>> = if let Some(token) = scanner_token {
-            Some(Arc::new(
+            Some(
                self.acquire_scanner_publication_lease_guard_for_rename(token, destination.0, destination.1)
                    .await
-                    .map_err(|err| DiskError::other(err.to_string()))?,
-            ))
+                    .map_err(|err| DiskError::other(err.to_string()))? as Arc<dyn Send + Sync>,
+            )
        } else {
            None
        };
@@ -358,7 +358,11 @@ An unlimited walk budget can also leave a retryable authoritative publication
 waiting for leadership loss or a successful proof. Observational and
 authoritative publication both retain the movement fence, on this node and
 through the remote publication leases. Do not remove those fences to clear a
-status. Verify
+status. A lease-scoped target operation (validation, rename, or delete on the
+leasing node) shares the read admission the lease already holds rather than
+queueing a new read on the movement gate. The gate is write-preferring, so a
+new read would wait behind a movement writer that is itself waiting for the
+lease, and neither would progress until the lease expired (rustfs/backlog#2873). Verify
 that a deployment contains the relevant fixes by commit ancestry; a newer
 release name alone is not evidence that it includes current main.

@@ -175,6 +175,9 @@ Contract: [decommission-compatibility.md](../architecture/decommission-compatibi
 - Supported only on multi-pool, non-legacy (ellipsis) deployments; anything else returns `NotImplemented`. A rebalance in progress blocks start.
 - 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.
+- 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
 POST /rustfs/admin/v3/pools/decommission?pool=<a>[,<b>]      # by command line
@@ -322,6 +322,13 @@ fn contextualize_admin_pool_api_error(
    }
 }

+/// A lock timeout, such as a start giving up on a gate that in-flight data
+/// movement holds, is transient contention: report it as retryable and name
+/// the lock instead of folding it into a generic internal error.
+fn is_lock_timeout(err: &crate::storage_api::error::StorageError) -> bool {
+    matches!(err, crate::storage_api::error::StorageError::Lock(rustfs_lock::LockError::Timeout { .. }))
+}
+
 fn decommission_start_api_error(err: crate::storage_api::error::StorageError) -> ApiError {
    if crate::storage_api::capacity::is_pool_activation_fleet_proof_error(&err) {
        return ApiError {
@@ -330,9 +337,31 @@ fn decommission_start_api_error(err: crate::storage_api::error::StorageError) ->
            source: Some(Box::new(err)),
        };
    }
+    if is_lock_timeout(&err) {
+        return ApiError {
+            code: S3ErrorCode::ServiceUnavailable,
+            message: err.to_string(),
+            source: Some(Box::new(err)),
+        };
+    }
    ApiError::from(err)
 }

+/// The leader-local pre-check refreshes rebalance metadata before the start.
+/// A busy gate there is the same retryable 503 the start itself reports;
+/// any other refresh failure stays an internal error.
+fn decommission_start_rebalance_refresh_error(err: crate::storage_api::error::StorageError, pools_indices: &[usize]) -> S3Error {
+    if is_lock_timeout(&err) {
+        S3Error::from(contextualize_admin_pool_api_error(
+            decommission_start_api_error(err),
+            "start decommission",
+            format!("pools {pools_indices:?}"),
+        ))
+    } else {
+        s3_error!(InternalError, "failed to refresh rebalance metadata before decommission start: {}", err)
+    }
+}
+
 fn decommission_admin_not_initialized_error_with_audit(operation: &str, audit: PoolAuditContext<'_>) -> S3Error {
    error!(
        event = EVENT_ADMIN_REQUEST_FAILED,
@@ -773,9 +802,9 @@ impl Operation for StartDecommission {
                .map_err(|err| contextualize_admin_pool_api_error(err, "start decommission", &pool_context))?;
        } else {
            store
-                .load_rebalance_meta()
+                .load_rebalance_meta_for_decommission_start()
                .await
-                .map_err(|e| s3_error!(InternalError, "failed to refresh rebalance metadata before decommission start: {}", e))?;
+                .map_err(|err| decommission_start_rebalance_refresh_error(err, &pools_indices))?;
            let decommission_running = store.is_decommission_running().await;
            let rebalance_running = store.is_rebalance_started().await;
            if decommission_running {
@@ -1051,8 +1080,8 @@ mod pools_handler_tests {
        POOL_ACTIVATION_FLEET_PROOF_REQUIRED, Params, PoolAuditContext, S3ErrorCode, S3Request, StartDecommission,
        StatusDecommission, StatusPool, Uri, contextualize_admin_pool_api_error,
        decommission_admin_not_initialized_error_with_audit, decommission_peer_target, decommission_start_api_error,
-        has_duplicate_indices, parse_mutation_pool_query, parse_pool_idx_by_id, parse_status_pool_query,
-        pool_admin_missing_credentials_error, pool_admin_missing_credentials_error_with_request,
+        decommission_start_rebalance_refresh_error, has_duplicate_indices, parse_mutation_pool_query, parse_pool_idx_by_id,
+        parse_status_pool_query, pool_admin_missing_credentials_error, pool_admin_missing_credentials_error_with_request,
        pool_admin_pool_index_error_with_audit, pool_admin_pool_not_found_error_with_audit,
        pool_admin_pool_parse_error_with_audit, pool_admin_query_parse_error, pool_admin_query_parse_error_with_audit,
        validate_pool_mutation_leader, validate_start_decommission_guards,
@@ -1254,6 +1283,66 @@ mod pools_handler_tests {
        assert_eq!(unrelated.message, "We encountered an internal error, please try again.");
    }

+    #[test]
+    fn test_decommission_start_api_error_names_a_busy_gate_as_retryable() {
+        let err = crate::storage_api::error::StorageError::Lock(rustfs_lock::LockError::timeout(
+            "data movement gate (decommission activation)",
+            std::time::Duration::from_secs(30),
+        ));
+
+        let err = decommission_start_api_error(err);
+
+        assert_eq!(err.code, s3s::S3ErrorCode::ServiceUnavailable);
+        assert!(err.message.contains("data movement gate (decommission activation)"), "{}", err.message);
+        let err = s3s::S3Error::from(contextualize_admin_pool_api_error(err, "start decommission", "pools [0]"));
+        assert_eq!(
+            err.headers().and_then(|headers| headers.get(http::header::RETRY_AFTER)),
+            Some(&http::HeaderValue::from_static("1")),
+            "a busy gate is advertised as retryable"
+        );
+    }
+
+    #[test]
+    fn test_decommission_start_rebalance_refresh_error_keeps_busy_gates_retryable() {
+        let busy = decommission_start_rebalance_refresh_error(
+            crate::storage_api::error::StorageError::Lock(rustfs_lock::LockError::timeout(
+                "data movement gate (rebalance metadata refresh)",
+                std::time::Duration::from_secs(20),
+            )),
+            &[0, 1],
+        );
+        assert_eq!(busy.code(), &s3s::S3ErrorCode::ServiceUnavailable);
+        assert!(
+            busy.message()
+                .is_some_and(|message| message.contains("pools [0, 1]") && message.contains("rebalance metadata refresh")),
+            "{:?}",
+            busy.message()
+        );
+        assert_eq!(
+            busy.headers().and_then(|headers| headers.get(http::header::RETRY_AFTER)),
+            Some(&http::HeaderValue::from_static("1")),
+            "a busy gate is advertised as retryable"
+        );
+
+        let failed = decommission_start_rebalance_refresh_error(
+            crate::storage_api::error::StorageError::other("rebalance metadata unreadable"),
+            &[0],
+        );
+        assert_eq!(failed.code(), &s3s::S3ErrorCode::InternalError);
+        assert!(
+            failed
+                .message()
+                .is_some_and(|message| message.starts_with("failed to refresh rebalance metadata before decommission start")),
+            "{:?}",
+            failed.message()
+        );
+        assert!(
+            failed
+                .headers()
+                .is_none_or(|headers| headers.get(http::header::RETRY_AFTER).is_none())
+        );
+    }
+
    #[test]
    fn test_contextualize_admin_pool_api_error_preserves_source() {
        let err = contextualize_admin_pool_api_error(
@@ -16,8 +16,8 @@

 use super::storage_api::admin_usecase::admin::get_server_info;
 use super::storage_api::admin_usecase::capacity::{
-    DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolStatus, RebalStatus, get_total_usable_capacity,
-    get_total_usable_capacity_free,
+    DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolMeta, PoolStatus, RebalStatus, RebalanceMeta,
+    get_total_usable_capacity, get_total_usable_capacity_free,
 };
 use super::storage_api::admin_usecase::contract::StorageAdminApi;
 use super::storage_api::admin_usecase::contract::bucket::{BucketOperations as _, BucketOptions};
@@ -267,15 +267,15 @@ impl DefaultAdminUsecase {
        }
    }

-    async fn refresh_rebalance_status_snapshot(store: &ECStore) -> AdminUsecaseResult<()> {
-        store.refresh_rebalance_status_meta().await.map_err(|err| {
+    async fn rebalance_status_snapshot(store: &ECStore) -> AdminUsecaseResult<Option<RebalanceMeta>> {
+        store.rebalance_status_meta_snapshot().await.map_err(|err| {
            error!("refresh rebalance metadata for pool status failed: {:?}", err);
            ApiError::from(err)
        })
    }

-    async fn refresh_pool_status_snapshot(store: &ECStore) -> AdminUsecaseResult<()> {
-        store.refresh_pool_status_meta().await.map_err(|err| {
+    async fn pool_status_snapshot(store: &ECStore) -> AdminUsecaseResult<PoolMeta> {
+        store.pool_status_meta_snapshot().await.map_err(|err| {
            error!("refresh pool metadata for pool status failed: {:?}", err);
            ApiError::from(err)
        })
@@ -400,11 +400,11 @@ impl DefaultAdminUsecase {
            return Err(Self::app_error_default(S3ErrorCode::NotImplemented));
        };

-        Self::refresh_pool_status_snapshot(store.as_ref()).await?;
+        let pool_meta = Self::pool_status_snapshot(store.as_ref()).await?;

        let mut pool_statuses = Vec::new();
        for (idx, _) in endpoints.as_ref().iter().enumerate() {
-            let state = store.status(idx).await.map_err(ApiError::from)?;
+            let state = store.status(&pool_meta, idx).await.map_err(ApiError::from)?;
            pool_statuses.push(state);
        }

@@ -416,10 +416,10 @@ impl DefaultAdminUsecase {
            return Err(Self::app_error(S3ErrorCode::InternalError, "Not init"));
        };
        let pool_statuses = self.execute_list_pool_statuses().await?;
-        Self::refresh_rebalance_status_snapshot(store.as_ref()).await?;
+        let rebalance_meta = Self::rebalance_status_snapshot(store.as_ref()).await?;
        let mut items = Vec::with_capacity(pool_statuses.len());
        for status in pool_statuses {
-            let rebalance_status = store.pool_rebalance_status(status.id).await;
+            let rebalance_status = ECStore::pool_rebalance_status(rebalance_meta.as_ref(), status.id);
            items.push(Self::pool_list_item_from_status(status, rebalance_status));
        }
        Ok(items)
@@ -455,10 +455,10 @@ impl DefaultAdminUsecase {
            return Err(Self::app_error(S3ErrorCode::InternalError, "Not init"));
        };

-        Self::refresh_pool_status_snapshot(store.as_ref()).await?;
-        let status = store.status(idx).await.map_err(ApiError::from)?;
-        Self::refresh_rebalance_status_snapshot(store.as_ref()).await?;
-        let rebalance_status = store.pool_rebalance_status(idx).await;
+        let pool_meta = Self::pool_status_snapshot(store.as_ref()).await?;
+        let status = store.status(&pool_meta, idx).await.map_err(ApiError::from)?;
+        let rebalance_meta = Self::rebalance_status_snapshot(store.as_ref()).await?;
+        let rebalance_status = ECStore::pool_rebalance_status(rebalance_meta.as_ref(), idx);
        Ok(Self::pool_list_item_from_status(status, rebalance_status))
    }

@@ -473,10 +473,10 @@ impl DefaultAdminUsecase {
            return Err(Self::app_error_default(S3ErrorCode::NotImplemented));
        }
        let pool_statuses = self.execute_list_pool_statuses().await?;
-        Self::refresh_rebalance_status_snapshot(store.as_ref()).await?;
+        let rebalance_meta = Self::rebalance_status_snapshot(store.as_ref()).await?;
        let mut pools = Vec::with_capacity(pool_statuses.len());
        for status in pool_statuses {
-            let rebalance_status = store.pool_rebalance_status(status.id).await;
+            let rebalance_status = ECStore::pool_rebalance_status(rebalance_meta.as_ref(), status.id);
            pools.push(Self::decommission_pool_status_from_status(status, rebalance_status));
        }
        Ok(AdminDecommissionStatus { pools })
@@ -500,10 +500,10 @@ impl DefaultAdminUsecase {
            return Err(Self::app_error(S3ErrorCode::InternalError, "Not init"));
        };

-        Self::refresh_pool_status_snapshot(store.as_ref()).await?;
-        let status = store.status(idx).await.map_err(ApiError::from)?;
-        Self::refresh_rebalance_status_snapshot(store.as_ref()).await?;
-        let rebalance_status = store.pool_rebalance_status(idx).await;
+        let pool_meta = Self::pool_status_snapshot(store.as_ref()).await?;
+        let status = store.status(&pool_meta, idx).await.map_err(ApiError::from)?;
+        let rebalance_meta = Self::rebalance_status_snapshot(store.as_ref()).await?;
+        let rebalance_status = ECStore::pool_rebalance_status(rebalance_meta.as_ref(), idx);
        Ok(Self::decommission_pool_status_from_status(status, rebalance_status))
    }

@@ -67,8 +67,10 @@ pub(crate) mod admin {
 pub(crate) mod capacity {
    pub(crate) type DecommissionUnresolvedEntry = crate::storage::storage_api::ecstore_capacity::DecommissionUnresolvedEntry;
    pub(crate) type PoolDecommissionInfo = crate::storage::storage_api::ecstore_capacity::PoolDecommissionInfo;
+    pub(crate) type PoolMeta = crate::storage::storage_api::ecstore_capacity::PoolMeta;
    pub(crate) type PoolStatus = crate::storage::storage_api::ecstore_capacity::PoolStatus;
    pub(crate) type RebalStatus = crate::storage::storage_api::ecstore_rebalance::RebalStatus;
+    pub(crate) type RebalanceMeta = crate::storage::storage_api::ecstore_rebalance::RebalanceMeta;

    pub(crate) fn get_total_usable_capacity(disks: &[rustfs_madmin::Disk], info: &rustfs_madmin::StorageInfo) -> usize {
        crate::storage::storage_api::ecstore_capacity::get_total_usable_capacity(disks, info)
@@ -2228,7 +2228,7 @@ impl NodeService {
                        }));
                    };
                    match store.acquire_scanner_publication_lease_guard(token).await {
-                        Ok(guard) => Some(Arc::new(guard)),
+                        Ok(guard) => Some(guard as Arc<dyn Send + Sync>),
                        Err(err) => {
                            return Ok(Response::new(DeleteResponse {
                                success: false,
@@ -441,8 +441,8 @@ pub(crate) mod ecstore_bucket {

 pub(crate) mod ecstore_capacity {
    pub(crate) use rustfs_ecstore::api::capacity::{
-        DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free,
-        is_pool_activation_fleet_proof_error, is_reserved_or_invalid_bucket,
+        DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolMeta, PoolStatus, get_total_usable_capacity,
+        get_total_usable_capacity_free, is_pool_activation_fleet_proof_error, is_reserved_or_invalid_bucket,
    };
 }