diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index ef3793625..448fbea2b 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/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, - 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; } diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index a2e742ac4..00a1ab106 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/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(self, resource: &'static str, acquire: F) -> Result { + 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( + deadline: Option, + resource: &'static str, + acquire: F, +) -> Result { + 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::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>>>; @@ -13009,17 +13066,43 @@ impl ECStore { Ok(()) } - async fn ensure_decommission_rebalance_idle_after_refresh(&self) -> Result<()> { - self.load_rebalance_meta().await?; + fn decommission_start_gate_deadline(&self) -> GateWaitDeadline { + #[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<()> { - self.load_rebalance_meta_under_start_gate().await?; - ensure_decommission_not_rebalancing(self.is_rebalance_conflicting_with_decommission().await) - } - - pub async fn status(&self, idx: usize) -> Result { + /// 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 { 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 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::>() - }; + /// 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 { + 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 { + 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) -> Result<()> { - self.start_decommission_inner(indices, None).await.map(|_| ()) - } - - async fn start_decommission_with_routines( - &self, - indices: Vec, - rx: &CancellationToken, - local_indices: &[usize], - ) -> Result> { - self.start_decommission_inner(indices, Some((rx, local_indices))).await + let deadline = self.decommission_start_gate_deadline(); + self.start_decommission_inner(indices, None, deadline).await.map(|_| ()) } + /// 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, reservation: Option<(&CancellationToken, &[usize])>, + deadline: GateWaitDeadline, ) -> Result> { 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; - self.ensure_decommission_rebalance_idle_after_refresh_under_start_gate() + 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?; 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() { diff --git a/crates/ecstore/src/runtime/instance.rs b/crates/ecstore/src/runtime/instance.rs index 4438665d6..e7dc1edb1 100644 --- a/crates/ecstore/src/runtime/instance.rs +++ b/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>; + pub(crate) struct ScannerPublicationLeaseEntry { pub(crate) expires_at: Instant, pub(crate) movement_generation: u64, pub(crate) namespace_generation: Option, 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, 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, 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) { diff --git a/crates/ecstore/src/services/rebalance/control.rs b/crates/ecstore/src/services/rebalance/control.rs index 0ee302157..a08a351a7 100644 --- a/crates/ecstore/src/services/rebalance/control.rs +++ b/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, - acquire_pool_rebalance_activation_locks, acquire_pool_rebalance_worker_locks, pool_meta_has_active_decommission, + DATA_MOVEMENT_GATE_RESOURCE, GateWaitDeadline, PoolMeta, PoolRebalanceActivationFence, PoolRebalanceWorkerFence, + 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> = std::sync::Mutex::new(None); @@ -383,6 +386,15 @@ fn merge_rebalance_status_refresh(current: &mut Option, 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, persisted: Option) -> 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) -> 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>> { + 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, ) -> Result>> { 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) -> 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<()> { - 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> { + // 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> { + 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 { 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! { diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index f28479a13..738bb2282 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/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> { + pub async fn acquire_scanner_publication_lease_guard( + &self, + token: Uuid, + ) -> Result>> { self.acquire_scanner_publication_lease_guard_with_target( token, None, @@ -1251,7 +1254,7 @@ impl ECStore { token: Uuid, volume: &str, path: &str, - ) -> Result> { + ) -> Result>> { 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, - ) -> Result> { - let operation_gate = self.ctx.data_movement_operation_gate(); - let operation_guard = operation_gate.read_owned().await; + ) -> Result>> { + // 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()); diff --git a/crates/ecstore/src/store/peer.rs b/crates/ecstore/src/store/peer.rs index be4560216..9e448a03c 100644 --- a/crates/ecstore/src/store/peer.rs +++ b/crates/ecstore/src/store/peer.rs @@ -243,11 +243,11 @@ impl ECStore { scanner_token: Option, ) -> DiskResult { let external_guard: Option> = 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, + ) } else { None }; diff --git a/docs/architecture/scanner-usage-publication.md b/docs/architecture/scanner-usage-publication.md index 4985fc2ec..81b2e6eb3 100644 --- a/docs/architecture/scanner-usage-publication.md +++ b/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. diff --git a/docs/operations/cluster-lifecycle-operations.md b/docs/operations/cluster-lifecycle-operations.md index 87f7c9ab9..9070cf74d 100644 --- a/docs/operations/cluster-lifecycle-operations.md +++ b/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=[,] # by command line diff --git a/rustfs/src/admin/handlers/pools.rs b/rustfs/src/admin/handlers/pools.rs index 22ec2a20d..9225259db 100644 --- a/rustfs/src/admin/handlers/pools.rs +++ b/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 - .load_rebalance_meta() - .await - .map_err(|e| s3_error!(InternalError, "failed to refresh rebalance metadata before decommission start: {}", e))?; + store.load_rebalance_meta_for_decommission_start().await.map_err(|err| { + 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) + } + })?; 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( diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index 469f550d7..0db2b82a1 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/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, - 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> { + 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 { + 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)) } diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 4666a403f..2fd0d716e 100644 --- a/rustfs/src/app/storage_api.rs +++ b/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) diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index 601d17bd1..c992e4d87 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/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), Err(err) => { return Ok(Response::new(DeleteResponse { success: false, diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 3e9114c80..e17adde67 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/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, - 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, }; }