fix(decommission): keep status reads and starts off the movement gate

The data-movement operation gate is a write-preferring RwLock. A scanner
publication lease stores a read guard on the leasing node, and the
coordinator's lease-scoped rename, delete and validation on that node
queued a second read. A movement writer queued in between (a decommission
start, a status refresh, a peer reload) then waited for the lease while
the coordinator's second read waited for the writer, so both stalled
until the 60 s lease TTL. Admin status reads also took the movement
writer, so every status call waited for the slowest in-flight copy or
cleanup and stalled new movement reads while it queued.

- Lease-scoped target admission shares the read admission the lease
  already holds instead of queueing a new read.
- Pool and rebalance status reads publish under the gates only when they
  are free; otherwise they merge the persisted metadata into a private
  copy and publish nothing.
- A decommission start bounds its start, activation and movement gate
  waits with one 30 s budget and fails with a retryable lock timeout
  before persisting anything; the admin handler maps it to 503 with
  Retry-After.

Refs rustfs/backlog#2873

crates/ecstore/src/api/mod.rs

@@ -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,
+        DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolMeta, PoolStatus, get_total_usable_capacity,
-        is_pool_activation_fleet_proof_error, path2_bucket_object, path2_bucket_object_with_base_path,
+        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;
 }

crates/ecstore/src/core/pools.rs

@@ -111,6 +111,14 @@ 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";
+/// Bound on how long an operator decommission start waits 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.
+const DECOMMISSION_START_GATE_WAIT: std::time::Duration = std::time::Duration::from_secs(30);
+const DECOMMISSION_START_GATE_RESOURCE: &str = "decommission start gate";
+pub(crate) const DATA_MOVEMENT_GATE_RESOURCE: &str = "data movement operation gate";
 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 +3918,42 @@ pub(crate) async fn acquire_pool_activation_fleet_proof(
         .ok_or_else(|| Error::other(POOL_ACTIVATION_FLEET_PROOF_REQUIRED))
 }
 
+/// One wait budget shared by every gate a single operator start acquires.
+/// Expiry is a retryable lock timeout raised before the start changes any
+/// state, so a busy gate fails the request instead of holding it open.
+#[derive(Clone, Copy, Debug)]
+pub(crate) struct GateWaitDeadline {
+    at: tokio::time::Instant,
+    budget: std::time::Duration,
+}
+
+impl GateWaitDeadline {
+    pub(crate) fn after(budget: std::time::Duration) -> Self {
+        Self {
+            at: tokio::time::Instant::now() + budget,
+            budget,
+        }
+    }
+
+    pub(crate) async fn acquire<F: std::future::Future>(self, resource: &'static str, acquire: F) -> Result<F::Output> {
+        tokio::time::timeout_at(self.at, acquire)
+            .await
+            .map_err(|_| Error::Lock(rustfs_lock::LockError::timeout(resource, self.budget)))
+    }
+}
+
+/// Wait for `acquire` within `deadline`, or without bound when there is none.
+pub(crate) async fn acquire_gate_until<F: std::future::Future>(
+    deadline: Option<GateWaitDeadline>,
+    resource: &'static str,
+    acquire: F,
+) -> Result<F::Output> {
+    match deadline {
+        Some(deadline) => deadline.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 +10187,19 @@ 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);
+}
+
 #[cfg(any(test, feature = "test-util"))]
 type DecommissionCapacityInfoOverrides =
     std::sync::Mutex<HashMap<uuid::Uuid, std::collections::VecDeque<Vec<DecommissionPoolCapacityInfo>>>>;
@@ -13009,17 +13066,43 @@ impl ECStore {
         Ok(())
     }
 
-    async fn ensure_decommission_rebalance_idle_after_refresh(&self) -> Result<()> {
+    fn decommission_start_gate_deadline(&self) -> GateWaitDeadline {
-        self.load_rebalance_meta().await?;
+        #[cfg(test)]
+        if let Some(wait) = DECOMMISSION_START_GATE_WAIT_OVERRIDES
+            .get()
+            .and_then(|overrides| overrides.lock().ok()?.get(&self.id).copied())
+        {
+            return GateWaitDeadline::after(wait);
+        }
+        GateWaitDeadline::after(DECOMMISSION_START_GATE_WAIT)
+    }
+
+    /// Refresh rebalance metadata ahead of an operator decommission start,
+    /// within the same bounded gate wait the start itself uses.
+    pub async fn load_rebalance_meta_for_decommission_start(&self) -> Result<()> {
+        let deadline = self.decommission_start_gate_deadline();
+        let _start_guard = deadline
+            .acquire(DECOMMISSION_START_GATE_RESOURCE, self.start_gate.lock())
+            .await?;
+        self.load_rebalance_meta_under_start_gate_until(Some(deadline)).await
+    }
+
+    async fn ensure_decommission_rebalance_idle_after_refresh(&self, deadline: GateWaitDeadline) -> Result<()> {
+        let _start_guard = deadline
+            .acquire(DECOMMISSION_START_GATE_RESOURCE, self.start_gate.lock())
+            .await?;
+        self.ensure_decommission_rebalance_idle_after_refresh_under_start_gate(deadline)
+            .await
+    }
+
+    async fn ensure_decommission_rebalance_idle_after_refresh_under_start_gate(&self, deadline: GateWaitDeadline) -> Result<()> {
+        self.load_rebalance_meta_under_start_gate_until(Some(deadline)).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<()> {
+    /// Pool status for an admin read, computed from a snapshot returned by
-        self.load_rebalance_meta_under_start_gate().await?;
+    /// [`ECStore::pool_status_meta_snapshot`].
-        ensure_decommission_not_rebalancing(self.is_rebalance_conflicting_with_decommission().await)
+    pub async fn status(&self, pool_meta: &PoolMeta, idx: usize) -> Result<PoolStatus> {
-    }
-
-    pub async fn status(&self, idx: usize) -> Result<PoolStatus> {
         let capacity_infos = self.get_decommission_all_pool_capacity_infos().await?;
         let space_info = capacity_infos
             .iter()
@@ -13027,8 +13110,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 +13119,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 movement_guard = movement_gate.write_owned().await;
-        let persisted = self.load_runtime_pool_meta("refresh pool status metadata failed").await?;
+        self.publish_pool_status_refresh(movement_guard).await
+    }
 
-        let active_workers = {
+    /// Pool metadata for an admin status read. It never waits for data
-            let cancelers = self.decommission_cancelers.read().await;
+    /// movement: a movement side effect holds a read on the movement gate for
-            cancelers
+    /// as long as it runs, and a queued writer would also stall every new
-                .iter()
+    /// movement read behind it. With the gate free, the refresh publishes as
-                .map(|canceler| canceler.as_ref().is_some_and(DecommissionCanceler::is_active))
+    /// before. Otherwise the persisted snapshot is merged into a private copy
-                .collect::<Vec<_>>()
+    /// 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 +13162,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 +14192,13 @@ impl ECStore {
         );
         validate_start_decommission_request(&indices, self.single_pool())?;
 
-        self.ensure_decommission_rebalance_idle_after_refresh().await?;
+        let deadline = self.decommission_start_gate_deadline();
+        self.ensure_decommission_rebalance_idle_after_refresh(deadline).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())), deadline)
             .await?;
         drop(spawn_decommission_index_cancelers(
             store,
@@ -17148,27 +17261,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(|_| ())
+        let deadline = self.decommission_start_gate_deadline();
-    }
+        self.start_decommission_inner(indices, None, deadline).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
     }
 
+    /// Every gate this start waits for shares `deadline`; on expiry 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])>,
+        deadline: GateWaitDeadline,
     ) -> 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(deadline).await?;
         #[cfg(test)]
         let endpoints = self.instance_endpoints().unwrap_or_else(|| self.endpoints());
         #[cfg(not(test))]
@@ -17210,8 +17318,10 @@ impl ECStore {
             }
         }
 
-        let _start_guard = self.start_gate.lock().await;
+        let _start_guard = deadline
-        self.ensure_decommission_rebalance_idle_after_refresh_under_start_gate()
+            .acquire(DECOMMISSION_START_GATE_RESOURCE, self.start_gate.lock())
+            .await?;
+        self.ensure_decommission_rebalance_idle_after_refresh_under_start_gate(deadline)
             .await?;
 
         let all_capacity_infos = self.get_decommission_all_pool_capacity_infos_with_owned_usage().await?;
@@ -17221,7 +17331,10 @@ 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;
+        // 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 = deadline.acquire(DATA_MOVEMENT_GATE_RESOURCE, movement_gate.write()).await?;
 
         let index_cancelers = if let Some((rx, local_indices)) = reservation {
             // Lock order matches terminal transitions: movement gate, then
@@ -19523,6 +19636,120 @@ 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"
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn decommission_start_fails_retryably_instead_of_waiting_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 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");
+        assert!(
+            matches!(&err, Error::Lock(rustfs_lock::LockError::Timeout { resource, .. }) if resource == DATA_MOVEMENT_GATE_RESOURCE),
+            "a busy gate must surface as a retryable lock timeout: {err:?}"
+        );
+        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");
+
+        // A queued start writer must not keep movement reads waiting either:
+        // once the start gives up, a new movement read is admitted at once.
+        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);
+        drop(stuck_side_effect);
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn decommission_v1_start_preflights_reject_before_metadata_writes() {

crates/ecstore/src/runtime/instance.rs

@@ -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
+        let (expires_at, movement_generation, namespace_generation, purpose, permit) = leases.get(&token).map(|entry| {
-            .get(&token)
+            (
-            .map(|entry| (entry.expires_at, entry.movement_generation, entry.namespace_generation, entry.purpose))?;
+                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) {

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

@@ -18,8 +18,9 @@ use super::{
     encode_rebalance_stop_propagation_record,
 };
 use crate::core::pools::{
-    PoolMeta, PoolRebalanceActivationFence, PoolRebalanceWorkerFence, acquire_pool_activation_fleet_proof,
+    DATA_MOVEMENT_GATE_RESOURCE, GateWaitDeadline, PoolMeta, PoolRebalanceActivationFence, PoolRebalanceWorkerFence,
-    acquire_pool_rebalance_activation_locks, acquire_pool_rebalance_worker_locks, pool_meta_has_active_decommission,
+    acquire_gate_until, 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,8 @@ use time::OffsetDateTime;
 use tracing::{debug, info};
 use uuid::Uuid;
 
+const REBALANCE_ACTIVATION_GATE_RESOURCE: &str = "rebalance activation gate";
+
 #[cfg(test)]
 static FAIL_NEXT_REBALANCE_ACTIVATION_SAVE: std::sync::Mutex<Option<String>> = std::sync::Mutex::new(None);
 
@@ -383,6 +386,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 +429,15 @@ impl ECStore {
         &self,
         expected_id: Option<&str>,
         stage: &str,
+    ) -> Result<Option<tokio::sync::OwnedRwLockWriteGuard<()>>> {
+        self.rebalance_activation_write_guard_until(expected_id, stage, None).await
+    }
+
+    async fn rebalance_activation_write_guard_until(
+        &self,
+        expected_id: Option<&str>,
+        stage: &str,
+        deadline: Option<GateWaitDeadline>,
     ) -> Result<Option<tokio::sync::OwnedRwLockWriteGuard<()>>> {
         let activation_gate = {
             let meta = self.rebalance_meta.read().await;
@@ -426,7 +447,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_until(deadline, REBALANCE_ACTIVATION_GATE_RESOURCE, gate.write_owned()).await?),
             None => None,
         })
     }
@@ -713,6 +734,12 @@ impl ECStore {
     }
 
     pub(crate) async fn load_rebalance_meta_under_start_gate(&self) -> Result<()> {
+        self.load_rebalance_meta_under_start_gate_until(None).await
+    }
+
+    /// `deadline` bounds the waits for the activation and movement writers;
+    /// on expiry the load fails before it reads or publishes anything.
+    pub(crate) async fn load_rebalance_meta_under_start_gate_until(&self, deadline: Option<GateWaitDeadline>) -> Result<()> {
         let mut meta = RebalanceMeta::new();
         debug!(
             event = EVENT_REBALANCE_STATE,
@@ -722,9 +749,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_until(None, "load rebalance metadata", deadline)
+            .await?;
         let movement_gate = self.ctx.data_movement_operation_gate();
-        let _movement_guard = movement_gate.write().await;
+        let _movement_guard = acquire_gate_until(deadline, DATA_MOVEMENT_GATE_RESOURCE, 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 +806,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<()> {
+    pub async fn rebalance_status_meta_snapshot(&self) -> Result<Option<RebalanceMeta>> {
-        let _start_guard = self.start_gate.lock().await;
+        // Transition order is start_gate -> activation_gate -> movement gate -> rebalance_meta.
-        let pool = clone_first_arc(&self.pools, "refresh_rebalance_status_meta: no pools available")?;
+        if let Ok(_start_guard) = self.start_gate.try_lock() {
-        let mut persisted = RebalanceMeta::new();
+            let activation_gate = self
-        let _activation_guard = self
+                .rebalance_meta
-            .rebalance_activation_write_guard(None, "refresh rebalance metadata")
+                .read()
-            .await?;
+                .await
-        let movement_gate = self.ctx.data_movement_operation_gate();
+                .as_ref()
-        let _movement_guard = movement_gate.write().await;
+                .map(|meta| Arc::clone(&meta.activation_gate));
-        let loaded = persisted.load(pool).await;
+            let activation_guard = match activation_gate {
-        match loaded {
+                Some(gate) => gate.try_write_owned().ok().map(Some),
-            Ok(()) => {
+                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();
                 }
-            }
+                return Ok(rebalance_meta.clone());
-            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}")));
             }
         }
 
-        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 +1287,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 +1451,62 @@ 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_read_recovers_after_metadata_lock_timeout() {
         let id = "rebalance-status-lock-retry";
@@ -1422,7 +1526,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! {

crates/ecstore/src/store/mod.rs

@@ -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<()>> {
+    ) -> Result<Arc<tokio::sync::OwnedRwLockReadGuard<()>>> {
-        let operation_gate = self.ctx.data_movement_operation_gate();
+        // Share the read admission the lease already holds instead of queueing
-        let operation_guard = operation_gate.read_owned().await;
+        // 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());

crates/ecstore/src/store/peer.rs

@@ -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
         };

docs/architecture/scanner-usage-publication.md

@@ -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.
 

docs/operations/cluster-lifecycle-operations.md

@@ -175,6 +175,8 @@ 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).
+- A start never waits indefinitely for in-flight data movement (an object copy or source cleanup, another start, a scanner publication lease) to release the start, rebalance activation, and data-movement gates: the leader's rebalance pre-check and the activation each allow those waits 30 seconds. If a gate stays busy, the leader fails the start with a lock timeout naming the gate and persists nothing; served by the leader itself, the request answers `503 ServiceUnavailable` with `Retry-After`. Retry the same request.
+- 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

rustfs/src/admin/handlers/pools.rs

@@ -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,6 +337,13 @@ 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)
 }
 
@@ -772,10 +786,17 @@ impl Operation for StartDecommission {
                 .map_err(ApiError::from)
                 .map_err(|err| contextualize_admin_pool_api_error(err, "start decommission", &pool_context))?;
         } else {
-            store
+            store.load_rebalance_meta_for_decommission_start().await.map_err(|err| {
-                .load_rebalance_meta()
+                if is_lock_timeout(&err) {
-                .await
+                    S3Error::from(contextualize_admin_pool_api_error(
-                .map_err(|e| s3_error!(InternalError, "failed to refresh rebalance metadata before decommission start: {}", e))?;
+                        decommission_start_api_error(err),
+                        "start decommission",
+                        format!("pools {:?}", pools_indices),
+                    ))
+                } else {
+                    s3_error!(InternalError, "failed to refresh rebalance metadata before decommission start: {}", err)
+                }
+            })?;
             let decommission_running = store.is_decommission_running().await;
             let rebalance_running = store.is_rebalance_started().await;
             if decommission_running {
@@ -1254,6 +1275,25 @@ 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 operation gate",
+            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 operation gate"), "{}", 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_contextualize_admin_pool_api_error_preserves_source() {
         let err = contextualize_admin_pool_api_error(

rustfs/src/app/admin_usecase.rs

@@ -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,
+    DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolMeta, PoolStatus, RebalStatus, RebalanceMeta,
-    get_total_usable_capacity_free,
+    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<()> {
+    async fn rebalance_status_snapshot(store: &ECStore) -> AdminUsecaseResult<Option<RebalanceMeta>> {
-        store.refresh_rebalance_status_meta().await.map_err(|err| {
+        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<()> {
+    async fn pool_status_snapshot(store: &ECStore) -> AdminUsecaseResult<PoolMeta> {
-        store.refresh_pool_status_meta().await.map_err(|err| {
+        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 pool_meta = Self::pool_status_snapshot(store.as_ref()).await?;
-        let status = store.status(idx).await.map_err(ApiError::from)?;
+        let status = store.status(&pool_meta, idx).await.map_err(ApiError::from)?;
-        Self::refresh_rebalance_status_snapshot(store.as_ref()).await?;
+        let rebalance_meta = Self::rebalance_status_snapshot(store.as_ref()).await?;
-        let rebalance_status = store.pool_rebalance_status(idx).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 pool_meta = Self::pool_status_snapshot(store.as_ref()).await?;
-        let status = store.status(idx).await.map_err(ApiError::from)?;
+        let status = store.status(&pool_meta, idx).await.map_err(ApiError::from)?;
-        Self::refresh_rebalance_status_snapshot(store.as_ref()).await?;
+        let rebalance_meta = Self::rebalance_status_snapshot(store.as_ref()).await?;
-        let rebalance_status = store.pool_rebalance_status(idx).await;
+        let rebalance_status = ECStore::pool_rebalance_status(rebalance_meta.as_ref(), idx);
         Ok(Self::decommission_pool_status_from_status(status, rebalance_status))
     }
 

rustfs/src/app/storage_api.rs

@@ -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)

rustfs/src/storage/rpc/node_service/disk.rs

@@ -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,

rustfs/src/storage/storage_api.rs

@@ -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,
+        DecommissionUnresolvedEntry, PoolDecommissionInfo, PoolMeta, PoolStatus, get_total_usable_capacity,
-        is_pool_activation_fleet_proof_error, is_reserved_or_invalid_bucket,
+        get_total_usable_capacity_free, is_pool_activation_fleet_proof_error, is_reserved_or_invalid_bucket,
     };
 }