From d0ce2f758b3e898182b18638ea5b10fad4dfd7b4 Mon Sep 17 00:00:00 2001 From: cxymds Date: Tue, 22 Sep 2026 19:20:05 +0800 Subject: [PATCH] fix(ecstore): bound decommission target gate contention (#8061) * fix(ecstore): bound decommission target gate contention Target capacity gate contention during pool decommission escalated a per-object transient into a durable bucket pause: each contended object failed its bucket entry, which re-ran the whole bucket listing and amplified attempts on the same objects. - centralize the decommission capacity failure classification so gate contention, benign contention and fatal failures are decided once - retry target-gate contention inline (bounded, jittered) before a mutation is admitted, covering put, part, complete, new-multipart and abort - defer contended entries to the end of the round instead of failing the bucket entry, and require the deferred set to drain before a set completes - treat a missing object or version, an overwrite and a superseded upload id as benign contention that is neither counted as a failure nor escalated - use full-jitter exponential backoff, capped, for decommission retries - expose the capacity pause reason, the waiting reason and a cumulative pause count in the admin pool status, plus gate-retry and per-object attempt metrics Related: rustfs/backlog#2644 * fix(ecstore): correct deferred replay and metadata compatibility --- crates/ecstore/src/core/pools.rs | 1021 +++++++++++++++-- crates/ecstore/src/core/pools_test.rs | 7 + .../decommission-compatibility.md | 16 +- rustfs/src/app/admin_usecase.rs | 70 +- 4 files changed, 1026 insertions(+), 88 deletions(-) diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 636b3c04e..70a6369ce 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -46,8 +46,8 @@ use crate::disk::error::DiskError; use crate::disk::{BUCKET_META_PREFIX, DiskAPI, RUSTFS_META_BUCKET}; use crate::error::{Error, Result}; use crate::error::{ - StorageError, is_err_bucket_exists, is_err_bucket_not_found, is_err_object_not_found, is_err_operation_canceled, - is_err_version_not_found, + StorageError, is_err_bucket_exists, is_err_bucket_not_found, is_err_data_movement_overwrite, is_err_invalid_upload_id, + is_err_object_not_found, is_err_operation_canceled, is_err_version_not_found, }; use crate::layout::endpoints::EndpointServerPools; use crate::object_api::{DecommissionCapacityOptions, GetObjectReader, ObjectInfo, ObjectOptions}; @@ -74,6 +74,7 @@ use futures::{ stream::FuturesUnordered, }; use http::HeaderMap; +use rand::RngExt as _; #[cfg(test)] use rmp_serde::Deserializer; use rmp_serde::Serializer; @@ -136,9 +137,20 @@ const DECOMMISSION_CAPACITY_RELEASE_COMPLETED: &str = "completed"; pub(crate) const DECOMMISSION_CAPACITY_TARGET_LOCK_PREFIX: &str = "decommission/capacity-target"; const DECOMMISSION_CAPACITY_TARGET_LOCK_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(250); const DECOMMISSION_CAPACITY_TARGET_GATE_MAX_ATTEMPTS: usize = 12; +pub(crate) const DECOMMISSION_MUTATION_GATE_MAX_INLINE_ATTEMPTS: usize = 3; +static DECOMMISSION_OBJECT_ATTEMPTS_MAX: std::sync::Mutex = std::sync::Mutex::new(0); const DECOMMISSION_CAPACITY_TARGET_GATE_BUSY_PREFIX: &str = "target pool "; const DECOMMISSION_CAPACITY_TARGET_GATE_BUSY_SUFFIX: &str = " target capacity mutation gate is busy"; const METRIC_DECOMMISSION_CAPACITY_CONFLICTS_TOTAL: &str = "rustfs_decommission_capacity_conflicts_total"; +const METRIC_DECOMMISSION_CAPACITY_GATE_RETRIES_TOTAL: &str = "rustfs_decommission_capacity_gate_retries_total"; +/// Entry passes and repeated version-copy attempts within a listing and its +/// deferred replay. Inline gate waits have their own retry counter. +const METRIC_DECOMMISSION_OBJECT_ATTEMPTS_TOTAL: &str = "rustfs_decommission_object_attempts_total"; +/// Highest per-entry attempt count, including its deferred replay. A new +/// listing after a durable pause starts a new observation. +const METRIC_DECOMMISSION_OBJECT_ATTEMPTS_MAX: &str = "rustfs_decommission_object_attempts_max"; +/// Wait time spent inside a single target capacity gate retry. +const METRIC_DECOMMISSION_CAPACITY_GATE_WAIT_SECONDS: &str = "rustfs_decommission_capacity_gate_wait_seconds"; const METRIC_DECOMMISSION_CAPACITY_PREDICTED_BYTES: &str = "rustfs_decommission_capacity_predicted_physical_bytes"; const METRIC_DECOMMISSION_CAPACITY_RESERVED_BYTES: &str = "rustfs_decommission_capacity_reserved_physical_bytes"; const METRIC_DECOMMISSION_CAPACITY_PREDICTION_ERROR_BYTES: &str = "rustfs_decommission_capacity_prediction_error_bytes"; @@ -156,6 +168,8 @@ const DECOMMISSION_SCANNER_BACKLOG_HANDOFF_MAX_ATTEMPTS: usize = 12; const DECOMMISSION_SOURCE_CLEANUP_RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(100); pub(crate) const DECOMMISSION_VERSION_COPY_ATTEMPTS: usize = 3; const DECOMMISSION_COPY_RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(50); +const DECOMMISSION_CAPACITY_GATE_RETRY_BASE: std::time::Duration = std::time::Duration::from_millis(100); +const DECOMMISSION_RETRY_BACKOFF_CAP: std::time::Duration = std::time::Duration::from_millis(1500); const DECOMMISSION_SOURCE_CHANGED_EXHAUSTION_LIMIT: usize = 100; const DECOMMISSION_TERMINAL_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(1); const DECOMMISSION_CANCEL_TARGET_LOCK_MAX_ATTEMPTS: usize = 3; @@ -163,6 +177,33 @@ const DECOMMISSION_CANCEL_TARGET_LOCK_MAX_ATTEMPTS: usize = 3; fn decommission_capacity_target_gate_retry_exhausted(attempt: usize) -> bool { attempt.saturating_add(1) >= DECOMMISSION_CAPACITY_TARGET_GATE_MAX_ATTEMPTS } + +fn record_decommission_capacity_gate_retry(target_pool_index: usize) { + metrics::counter!( + METRIC_DECOMMISSION_CAPACITY_GATE_RETRIES_TOTAL, + "target_pool_index" => target_pool_index.to_string() + ) + .increment(1); +} + +/// Record one object-processing attempt plus the peak attempts any object +/// reached, so amplification is visible without reconstructing it from logs. +fn record_decommission_object_attempt(attempts: &mut usize) { + *attempts = attempts.saturating_add(1); + metrics::counter!(METRIC_DECOMMISSION_OBJECT_ATTEMPTS_TOTAL).increment(1); + let attempt = u64::try_from(*attempts).unwrap_or(u64::MAX); + // Serialize publication too: concurrent fetch_max + gauge.set can publish + // an older peak after a newer one. This guard never crosses an await. + let mut peak = DECOMMISSION_OBJECT_ATTEMPTS_MAX.lock().unwrap_or_else(|err| err.into_inner()); + if *peak < attempt { + *peak = attempt; + metrics::gauge!(METRIC_DECOMMISSION_OBJECT_ATTEMPTS_MAX).set(attempt as f64); + } +} + +fn record_decommission_capacity_gate_wait(delay: std::time::Duration) { + metrics::histogram!(METRIC_DECOMMISSION_CAPACITY_GATE_WAIT_SECONDS).record(delay.as_secs_f64()); +} const DECOMMISSION_DURABLE_ILM_RECEIPT_ROOT: &str = "decommission/ilm-receipts"; const DECOMMISSION_DURABLE_ILM_MANIFEST_ROOT: &str = "decommission/ilm-manifests"; const DECOMMISSION_DURABLE_ILM_RECEIPT_SCHEMA: &str = "v2"; @@ -1003,7 +1044,7 @@ fn is_decommission_capacity_intent_conflict(err: &Error) -> bool { || err.to_string().contains("unresolved target capacity intent") } -fn decommission_capacity_target_gate_busy_index(err: &Error) -> Option { +pub(crate) fn decommission_capacity_target_gate_busy_index(err: &Error) -> Option { if let Error::DecommissionCapacityBlocked { message } = err { return message .strip_prefix(DECOMMISSION_CAPACITY_TARGET_GATE_BUSY_PREFIX)? @@ -1014,10 +1055,56 @@ fn decommission_capacity_target_gate_busy_index(err: &Error) -> Option { data_movement::data_movement_stage_source(err).and_then(decommission_capacity_target_gate_busy_index) } -fn is_decommission_capacity_target_gate_busy(err: &Error) -> bool { +pub(crate) fn is_decommission_capacity_target_gate_busy(err: &Error) -> bool { decommission_capacity_target_gate_busy_index(err).is_some() } +/// How the decommission capacity path must treat a failure. +/// +/// Callers decide between retrying, ignoring, and propagating from this single +/// classification so the same failure shape cannot be treated differently at +/// different call sites. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum DecommissionCapacityFailure { + /// The target pool's capacity mutation gate is held by another mutation. + /// Transient: the mutation can be retried after a short backoff. + TargetGateBusy { target_pool_index: usize }, + /// Another mutation's durable target capacity intent is still unresolved. + /// Transient, but converged on a different retry budget than gate contention. + IntentConflict, + /// The mutation is no longer needed: the object or version disappeared, or + /// an equivalent target already exists. Never counted as a failure. + BenignContention, + /// A real failure that must propagate to the caller. + Fatal, +} + +/// Predicate for the benign-contention class. +/// +/// Only identity/overwrite shapes belong here: a missing object or version, an +/// equivalent target that already published the mutation, and an upload id that +/// a competing mutation already superseded. Everything else must stay fatal so a +/// genuine migration failure cannot be silently dropped. +fn is_decommission_benign_contention_error(err: &Error) -> bool { + if is_decommission_copy_cleanup_safe_error(err) || is_err_data_movement_overwrite(err) || is_err_invalid_upload_id(err) { + return true; + } + data_movement::data_movement_stage_source(err).is_some_and(is_decommission_benign_contention_error) +} + +fn classify_decommission_capacity_failure(err: &Error) -> DecommissionCapacityFailure { + if let Some(target_pool_index) = decommission_capacity_target_gate_busy_index(err) { + return DecommissionCapacityFailure::TargetGateBusy { target_pool_index }; + } + if is_decommission_capacity_intent_conflict(err) { + return DecommissionCapacityFailure::IntentConflict; + } + if is_decommission_benign_contention_error(err) { + return DecommissionCapacityFailure::BenignContention; + } + DecommissionCapacityFailure::Fatal +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum DecommissionCapacityRetryKind { IntentConflict, @@ -1025,7 +1112,7 @@ enum DecommissionCapacityRetryKind { fn decommission_capacity_retry_kind(err: &Error, intent_conflict_attempt: usize) -> Option { (intent_conflict_attempt < DECOMMISSION_CAPACITY_INTENT_CONFLICT_MAX_ATTEMPTS - && is_decommission_capacity_intent_conflict(err)) + && matches!(classify_decommission_capacity_failure(err), DecommissionCapacityFailure::IntentConflict)) .then_some(DecommissionCapacityRetryKind::IntentConflict) } @@ -4005,6 +4092,37 @@ fn resolve_decommission_listing_worker_result( worker_result.map_err(|err| Error::other(format!("decommission listing worker {set_idx} task join error: {err}")))? } +/// A bucket set may only finish once every deferred entry has been resolved. +/// +/// The deferred round is the last chance for a contended entry; leaving one +/// unresolved would mark the bucket done and drop the object. This is reported +/// as a capacity failure so the pool takes the existing durable-pause path +/// instead of a terminal failure. +fn ensure_decommission_deferred_drained(bucket: &str, set_idx: usize, deferred_remaining: usize) -> Result<()> { + if deferred_remaining == 0 { + return Ok(()); + } + Err(decommission_capacity_blocked_error(format!( + "decommission bucket {bucket} set {set_idx} finished with {deferred_remaining} deferred entries remaining" + ))) +} + +fn decommission_entry_budget_acquire_error(err: impl Display) -> Error { + Error::other(format!("decommission entry budget permit acquire failed: {err}")) +} + +/// Collapse a single-entry test invocation into a plain result. +/// +/// A deferred outcome means the entry still needs another round, which the test +/// callers do not run; surface it as the contention failure that caused it. +#[cfg(test)] +fn resolve_decommission_entry_test_outcome(outcome: Result) -> Result<()> { + match outcome? { + DecommissionEntryOutcome::Complete => Ok(()), + DecommissionEntryOutcome::Deferred(err) => Err(err), + } +} + fn should_retry_decommission_listing(err: &Error, attempt: usize, max_attempts: usize) -> bool { !is_err_bucket_not_found(err) && attempt + 1 < max_attempts } @@ -4016,8 +4134,32 @@ async fn wait_decommission_retry_backoff(rx: &CancellationToken, delay: std::tim } } +/// Exponential backoff ceiling for a 1-based `attempt`, capped so a long +/// contention streak cannot turn into a multi-minute stall. +fn decommission_retry_backoff_ceiling(base: std::time::Duration, attempt: usize) -> std::time::Duration { + let exponent = u32::try_from(attempt.saturating_sub(1)).unwrap_or(u32::MAX).min(31); + base.saturating_mul(1u32 << exponent).min(DECOMMISSION_RETRY_BACKOFF_CAP) +} + +/// Full-jitter exponential decommission retry backoff. +/// +/// Racing workers keep observing the same target capacity gate. A shared fixed +/// delay makes them collide again after every round, so each retry draws a +/// uniform delay from `[0, ceiling]` instead. fn decommission_retry_backoff_delay(base: std::time::Duration, attempt: usize) -> std::time::Duration { - base.saturating_mul(u32::try_from(attempt).unwrap_or(u32::MAX)) + let ceiling = decommission_retry_backoff_ceiling(base, attempt); + if ceiling.is_zero() { + return ceiling; + } + let ceiling_millis = u64::try_from(ceiling.as_millis()).unwrap_or(u64::MAX); + std::time::Duration::from_millis(rand::rng().random_range(0..=ceiling_millis)) +} + +/// Backoff for retries that wait on another mutation's target capacity gate. +pub(crate) fn decommission_capacity_gate_retry_delay(attempt: usize) -> std::time::Duration { + let delay = decommission_retry_backoff_delay(DECOMMISSION_CAPACITY_GATE_RETRY_BASE, attempt); + record_decommission_capacity_gate_wait(delay); + delay } #[cfg(test)] @@ -4223,6 +4365,43 @@ enum DecommissionEntryAttemptOutcome { SourceChanged, } +/// Compact retry state retained until the listing has drained. +#[derive(Debug)] +struct DeferredDecommissionEntry { + name: String, + attempts: usize, +} + +/// Terminal result of a decommission entry pass. +#[derive(Debug)] +enum DecommissionEntryOutcome { + Complete, + /// The entry is still contended on a transient capacity condition after its + /// attempt budget. It must be retried at the end of the current round + /// instead of failing the whole bucket. The carried error is the last + /// contention failure, which becomes terminal only if the deferred round + /// cannot make progress either. + Deferred(Error), +} + +/// Decide what a terminal entry-attempt failure means for the current round. +/// +/// Target-gate and capacity-intent contention is a property of the target pool, +/// so the entry is deferred to the end of the round instead of failing the +/// bucket. Benign contention means the mutation is no longer needed (the object +/// or version disappeared, or a competing mutation already superseded it): the +/// entry is done and must not be counted as a failure or escalated. Every other +/// failure stays terminal. +fn resolve_decommission_entry_failure(err: Error) -> Result { + match classify_decommission_capacity_failure(&err) { + DecommissionCapacityFailure::TargetGateBusy { .. } | DecommissionCapacityFailure::IntentConflict => { + Ok(DecommissionEntryOutcome::Deferred(err)) + } + DecommissionCapacityFailure::BenignContention => Ok(DecommissionEntryOutcome::Complete), + DecommissionCapacityFailure::Fatal => Err(err), + } +} + #[cfg(test)] pub(crate) type DecommissionTestFaultDecision = Arc bool + Send + Sync>; @@ -9815,6 +9994,8 @@ struct DecommissionCapacityLockOrderBarrierState { external_heal_target_lock_attempted: tokio::sync::Notify, target_gate_retry_entered: tokio::sync::Notify, target_gate_retry_entries: AtomicUsize, + target_gate_inline_retry_entered: tokio::sync::Notify, + target_gate_inline_retry_entries: AtomicUsize, target_gate_exact_reloads: AtomicUsize, target_gate_acquire_pause_target: AtomicUsize, target_gate_acquire_entered: tokio::sync::Notify, @@ -9858,6 +10039,8 @@ impl DecommissionCapacityLockOrderBarrier { external_heal_target_lock_attempted: tokio::sync::Notify::new(), target_gate_retry_entered: tokio::sync::Notify::new(), target_gate_retry_entries: AtomicUsize::new(0), + target_gate_inline_retry_entered: tokio::sync::Notify::new(), + target_gate_inline_retry_entries: AtomicUsize::new(0), target_gate_exact_reloads: AtomicUsize::new(0), target_gate_acquire_pause_target: AtomicUsize::new(usize::MAX), target_gate_acquire_entered: tokio::sync::Notify::new(), @@ -9949,6 +10132,11 @@ impl DecommissionCapacityLockOrderBarrier { .expect("decommission entries should observe target gate contention"); } + #[cfg(feature = "test-util")] + pub(crate) fn target_gate_inline_retries(&self) -> usize { + self.state.target_gate_inline_retry_entries.load(Ordering::Acquire) + } + #[cfg(feature = "test-util")] pub(crate) fn target_gate_exact_reloads(&self) -> usize { self.state.target_gate_exact_reloads.load(Ordering::Acquire) @@ -10139,6 +10327,21 @@ fn notify_decommission_target_gate_retry(store_id: uuid::Uuid) { } } +#[cfg(test)] +fn notify_decommission_target_gate_inline_retry(store_id: uuid::Uuid) { + let barrier = DECOMMISSION_CAPACITY_LOCK_ORDER_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("decommission capacity lock-order barrier should not be poisoned") + .as_ref() + .filter(|state| state.owner_store_id == store_id) + .cloned(); + if let Some(barrier) = barrier { + barrier.target_gate_inline_retry_entries.fetch_add(1, Ordering::AcqRel); + barrier.target_gate_inline_retry_entered.notify_one(); + } +} + #[cfg(test)] fn notify_decommission_target_gate_exact_reload(store_id: uuid::Uuid) { let barrier = DECOMMISSION_CAPACITY_LOCK_ORDER_BARRIER @@ -10720,6 +10923,50 @@ impl ECStore { } } + /// Acquire the target pool's capacity mutation gate, absorbing a transient + /// gate-busy result with a small jittered inline retry. + /// + /// The target gate is acquired before the admitted mutation reads or writes + /// any data, so a busy gate guarantees the mutation has not started and can + /// be retried in place. A bounded retry here keeps short contention from + /// escalating an otherwise healthy object migration to a durable pause. + async fn acquire_decommission_capacity_target_guard_retried( + &self, + target_pool_index: usize, + ) -> Result { + let mut gate_attempt = 0; + loop { + match self.acquire_decommission_capacity_target_guard(target_pool_index).await { + Ok(guard) => return Ok(guard), + Err(err) if is_decommission_capacity_target_gate_busy(&err) => { + gate_attempt += 1; + if gate_attempt >= DECOMMISSION_MUTATION_GATE_MAX_INLINE_ATTEMPTS { + return Err(err); + } + #[cfg(test)] + notify_decommission_target_gate_inline_retry(self.id); + record_decommission_capacity_gate_retry(target_pool_index); + let retry_delay = decommission_capacity_gate_retry_delay(gate_attempt); + warn!( + event = EVENT_DECOMMISSION_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + store_id = %self.id, + target_pool_index, + attempt = gate_attempt, + max_attempts = DECOMMISSION_MUTATION_GATE_MAX_INLINE_ATTEMPTS, + retry_delay_ms = retry_delay.as_millis(), + state = "capacity_gate_inline_retry", + error = %err, + "Decommission target capacity gate busy; retrying mutation inline" + ); + tokio::time::sleep(retry_delay).await; + } + Err(err) => return Err(err), + } + } + } + async fn acquire_decommission_capacity_owner_target_guard( &self, owner: DecommissionCapacityOwner, @@ -11588,7 +11835,10 @@ impl ECStore { DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION => { Some(match take_decommission_capacity_target_permit(self.id, target_pool_index, owner) { Some(guard) => guard, - None => self.acquire_decommission_capacity_target_guard(target_pool_index).await?, + None => { + self.acquire_decommission_capacity_target_guard_retried(target_pool_index) + .await? + } }) } version => { @@ -13612,6 +13862,7 @@ impl ECStore { source_changed_exhaustions: Arc, entry_budget: Arc, queue: Arc>>, + deferred: Arc>>, entry_error: Arc>>, ) { loop { @@ -13626,7 +13877,13 @@ impl ECStore { let Some(QueuedDecommissionEntry { entry, queue_permit }) = queued else { return; }; + // Only listing entries retain metadata that distinguishes prefixes + // from real directory objects. Deferred names are already objects. + if entry.is_dir() { + continue; + } let object_name = entry.name.clone(); + let mut attempts = 0; if entry_error.lock().await.is_some() { drop(queue_permit); @@ -13670,7 +13927,7 @@ impl ECStore { } { Ok(permit) => permit, Err(err) => { - let err = Error::other(format!("decommission entry budget permit acquire failed: {err}")); + let err = decommission_entry_budget_acquire_error(err); error!( event = EVENT_DECOMMISSION_ENTRY, component = LOG_COMPONENT_ECSTORE, @@ -13693,7 +13950,7 @@ impl ECStore { rx.clone(), idx, generation, - entry, + entry.clone(), bucket.clone(), set.clone(), lifecycle_config.clone(), @@ -13701,30 +13958,172 @@ impl ECStore { replication_config.clone(), expected_bucket_incarnation_id, Arc::clone(&source_changed_exhaustions), + &mut attempts, ) .await; drop(entry_budget_permit); - drop(queue_permit); - if let Err(err) = result { - error!( - event = EVENT_DECOMMISSION_ENTRY, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_POOLS, - pool_index = idx, - set_index = set_idx, - bucket = %bucket, - object = %object_name, - state = "entry_failed", - error = %err, - "Decommission entry failed" - ); - record_decommission_entry_error(&entry_error, &rx, err).await; - return; + match result { + Ok(DecommissionEntryOutcome::Complete) => drop(queue_permit), + Ok(DecommissionEntryOutcome::Deferred(err)) => { + // Release the outstanding permit before deferring: the + // listing drain waits for the full budget to return, and the + // listing has already finished for this round. + drop(queue_permit); + // Only the name and attempt count are retained. A contention streak + // would otherwise pin one metadata buffer per object, and the + // replay reloads the versions it needs from disk anyway. + deferred.lock().await.push(DeferredDecommissionEntry { + name: object_name, + attempts, + }); + info!( + event = EVENT_DECOMMISSION_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + pool_index = idx, + set_index = set_idx, + bucket = %bucket, + object = %entry.name, + state = "entry_deferred_to_round_end", + error = %err, + "Decommission entry deferred to the end of the current round" + ); + } + Err(err) => { + drop(queue_permit); + error!( + event = EVENT_DECOMMISSION_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + pool_index = idx, + set_index = set_idx, + bucket = %bucket, + object = %object_name, + state = "entry_failed", + error = %err, + "Decommission entry failed" + ); + record_decommission_entry_error(&entry_error, &rx, err).await; + return; + } } } } + /// Run one deferred entry after the round's listing drained. + /// + /// The worker loop owns deferral, so this path must never defer again: any + /// remaining contention is terminal for the round and fails the bucket. + #[allow(clippy::too_many_arguments)] + async fn replay_deferred_decommission_entry( + self: Arc, + rx: CancellationToken, + idx: usize, + set_idx: usize, + generation: OffsetDateTime, + mut deferred_entry: DeferredDecommissionEntry, + bucket: String, + set: Arc, + lifecycle_config: Option, + object_lock_config: Option, + replication_config: Option<(ReplicationConfiguration, OffsetDateTime)>, + expected_bucket_incarnation_id: Option, + source_changed_exhaustions: Arc, + entry_budget: Arc, + entry_error: Arc>>, + ) -> Result { + let object_name = deferred_entry.name; + let entry = MetaCacheEntry { + name: object_name.clone(), + ..Default::default() + }; + + if entry_error.lock().await.is_some() || rx.is_cancelled() { + return Ok(DecommissionEntryOutcome::Complete); + } + + if let Err(err) = self.ensure_decommission_generation_current(idx, generation).await { + if matches!(err, Error::OperationCanceled) { + rx.cancel(); + } else { + record_decommission_entry_error(&entry_error, &rx, err).await; + } + return Ok(DecommissionEntryOutcome::Complete); + } + + if let Err(err) = backpressure::wait_for_data_movement_admission(DataMovementOperation::Decommission, idx, &rx).await { + if matches!(err, Error::OperationCanceled) { + return Ok(DecommissionEntryOutcome::Complete); + } + error!( + event = EVENT_DECOMMISSION_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + pool_index = idx, + set_index = set_idx, + bucket = %bucket, + object = %object_name, + state = "entry_admission_failed", + error = %err, + "Decommission entry admission failed" + ); + record_decommission_entry_error(&entry_error, &rx, err).await; + return Ok(DecommissionEntryOutcome::Complete); + } + + let entry_budget_permit = match tokio::select! { + biased; + _ = rx.cancelled() => return Ok(DecommissionEntryOutcome::Complete), + permit = entry_budget.clone().acquire_owned() => permit, + } { + Ok(permit) => permit, + Err(err) => { + let err = decommission_entry_budget_acquire_error(err); + record_decommission_entry_error(&entry_error, &rx, err).await; + return Ok(DecommissionEntryOutcome::Complete); + } + }; + + let result = self + .decommission_entry( + rx.clone(), + idx, + generation, + entry, + bucket.clone(), + set, + lifecycle_config, + object_lock_config, + replication_config, + expected_bucket_incarnation_id, + source_changed_exhaustions, + &mut deferred_entry.attempts, + ) + .await; + drop(entry_budget_permit); + + let err = match result { + Ok(DecommissionEntryOutcome::Complete) => return Ok(DecommissionEntryOutcome::Complete), + Ok(outcome @ DecommissionEntryOutcome::Deferred(_)) => return Ok(outcome), + Err(err) => err, + }; + error!( + event = EVENT_DECOMMISSION_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + pool_index = idx, + set_index = set_idx, + bucket = %bucket, + object = %object_name, + state = "entry_failed", + error = %err, + "Decommission entry failed" + ); + record_decommission_entry_error(&entry_error, &rx, err).await; + Ok(DecommissionEntryOutcome::Complete) + } + #[allow(clippy::too_many_arguments)] async fn decommission_set( self: Arc, @@ -13748,6 +14147,7 @@ impl ECStore { let outstanding = Arc::new(Semaphore::new(outstanding_capacity)); let (tx, rx_queue) = mpsc::channel(queue_capacity); let queue = Arc::new(tokio::sync::Mutex::new(rx_queue)); + let deferred: Arc>> = Arc::new(tokio::sync::Mutex::new(Vec::new())); let mut entry_workers = tokio::task::JoinSet::new(); for _ in 0..worker_count { @@ -13761,6 +14161,7 @@ impl ECStore { let source_changed_exhaustions = Arc::clone(&source_changed_exhaustions); let queue = queue.clone(); let entry_budget = entry_budget.clone(); + let deferred = deferred.clone(); let entry_error = entry_error.clone(); entry_workers.spawn(async move { this.decommission_entry_worker( @@ -13777,6 +14178,7 @@ impl ECStore { source_changed_exhaustions, entry_budget, queue, + deferred, entry_error, ) .await; @@ -13907,7 +14309,62 @@ impl ECStore { if let Some(err) = entry_error.lock().await.clone() { return Err(err); } - listing_result + listing_result?; + + // The listing drained, so every entry either completed or was deferred. + // Replaying the deferred set is the last chance for this round; a + // bucket must not be reported complete while a deferred entry remains. + let deferred_entries = std::mem::take(&mut *deferred.lock().await); + let deferred_remaining = Arc::new(AtomicUsize::new(0)); + if !deferred_entries.is_empty() { + // The replay keeps the set's own worker bound: a whole-bucket + // deferral must not spawn one task per object. + futures::stream::iter(deferred_entries) + .for_each_concurrent(Some(worker_count), |deferred_entry| { + let this = self.clone(); + let rx = rx.clone(); + let bucket = bi.name.clone(); + let set = set.clone(); + let lifecycle_config = lifecycle_config.clone(); + let object_lock_config = object_lock_config.clone(); + let replication_config = replication_config.clone(); + let source_changed_exhaustions = Arc::clone(&source_changed_exhaustions); + let entry_budget = entry_budget.clone(); + let entry_error = entry_error.clone(); + let deferred_remaining = Arc::clone(&deferred_remaining); + async move { + let outcome = this + .replay_deferred_decommission_entry( + rx, + idx, + set_idx, + generation, + deferred_entry, + bucket, + set, + lifecycle_config, + object_lock_config, + replication_config, + expected_bucket_incarnation_id, + source_changed_exhaustions, + entry_budget, + entry_error, + ) + .await; + if matches!(outcome, Ok(DecommissionEntryOutcome::Deferred(_))) { + deferred_remaining.fetch_add(1, Ordering::Relaxed); + } + } + }) + .await; + if let Some(err) = entry_error.lock().await.clone() { + return Err(err); + } + } + let deferred_remaining = deferred_remaining.load(Ordering::Relaxed); + // A set must not report success while an entry is still deferred: the + // bucket would be marked done and the object silently dropped. + ensure_decommission_deferred_drained(bi.name.as_str(), set_idx, deferred_remaining) } async fn track_decommission_entry_progress_stage( @@ -13946,10 +14403,8 @@ impl ECStore { notify_decommission_target_gate_retry(self.id); let mut wait_attempt = target_busy_attempt; let target_guard = loop { - let retry_delay = decommission_retry_backoff_delay( - DECOMMISSION_SOURCE_CLEANUP_RETRY_DELAY, - wait_attempt.min(DECOMMISSION_CAPACITY_INTENT_CONFLICT_MAX_ATTEMPTS), - ); + let retry_delay = + decommission_capacity_gate_retry_delay(wait_attempt.min(DECOMMISSION_CAPACITY_INTENT_CONFLICT_MAX_ATTEMPTS)); if wait_decommission_retry_backoff(rx, retry_delay).await { decommission_cancel_signal_result(rx.is_cancelled())?; } @@ -14308,7 +14763,8 @@ impl ECStore { replication_config: Option<(ReplicationConfiguration, OffsetDateTime)>, expected_bucket_incarnation_id: Option, source_changed_exhaustions: Arc, - ) -> Result<()> { + attempts: &mut usize, + ) -> Result { let mut counted_versions = HashSet::new(); for entry_attempt in 1..=DECOMMISSION_ENTRY_MAX_ATTEMPTS { @@ -14332,6 +14788,7 @@ impl ECStore { scanner_backlog_handoff_attempt, source_changed_exhaustions.as_ref(), &mut counted_versions, + attempts, ) .await; let retry = result @@ -14383,7 +14840,7 @@ impl ECStore { } }; match attempt_result { - Ok(DecommissionEntryAttemptOutcome::Complete) => return Ok(()), + Ok(DecommissionEntryAttemptOutcome::Complete) => return Ok(DecommissionEntryOutcome::Complete), Ok(DecommissionEntryAttemptOutcome::SourceChanged) => { let retry_delay = decommission_retry_backoff_delay(DECOMMISSION_SOURCE_CLEANUP_RETRY_DELAY, entry_attempt); warn!( @@ -14403,7 +14860,43 @@ impl ECStore { decommission_cancel_signal_result(rx.is_cancelled())?; } } - Err(err) => return Err(err), + Err(err) => { + return match resolve_decommission_entry_failure(err) { + Ok(DecommissionEntryOutcome::Deferred(err)) => { + // Contention belongs to the target pool, not to this + // object: failing the entry would discard every + // in-flight object's progress for a condition the + // deferred round can still absorb. + warn!( + event = EVENT_DECOMMISSION_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + state = "entry_deferred", + pool_index = idx, + bucket = %bucket, + object = %entry.name, + attempts = DECOMMISSION_ENTRY_MAX_ATTEMPTS, + error = %err, + "Decommission entry deferred after exhausting its attempts on target capacity contention" + ); + Ok(DecommissionEntryOutcome::Deferred(err)) + } + Ok(DecommissionEntryOutcome::Complete) => { + warn!( + event = EVENT_DECOMMISSION_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + state = "entry_ignored", + pool_index = idx, + bucket = %bucket, + object = %entry.name, + "Decommission entry ignored because its mutation is no longer needed" + ); + Ok(DecommissionEntryOutcome::Complete) + } + Err(err) => Err(err), + }; + } } } @@ -14430,7 +14923,9 @@ impl ECStore { scanner_backlog_handoff_attempt: usize, source_changed_exhaustions: &AtomicUsize, counted_versions: &mut HashSet<(Option, bool)>, + attempts: &mut usize, ) -> Result { + record_decommission_object_attempt(attempts); debug!( event = EVENT_DECOMMISSION_ENTRY, component = LOG_COMPONENT_ECSTORE, @@ -14443,19 +14938,6 @@ impl ECStore { max_attempts = DECOMMISSION_ENTRY_MAX_ATTEMPTS, "Decommission entry started" ); - if entry.is_dir() { - debug!( - event = EVENT_DECOMMISSION_ENTRY, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_POOLS, - pool_index = idx, - bucket = %bucket, - object = %entry.name, - state = "skipped_directory", - "Decommission entry skipped directory" - ); - return Ok(DecommissionEntryAttemptOutcome::Complete); - } // Scanner caches describe their own erasure set and are rebuilt there. // Copying one onto another set can overwrite unrelated cache contents or // leave an unresolvable target-capacity intent after a conditional PUT. @@ -14590,6 +15072,9 @@ impl ECStore { let mut target_busy_attempt: usize = 0; let mut target_permit = None; while version_attempt <= DECOMMISSION_VERSION_COPY_ATTEMPTS { + if version_attempt > 1 || target_busy_attempt > 0 { + record_decommission_object_attempt(attempts); + } let result = self .run_guarded_decommission_side_effect(&rx, &operation_gate, || async { self.decommission_tiered_object( @@ -14773,6 +15258,9 @@ impl ECStore { let mut target_busy_attempt: usize = 0; let mut target_permit = None; while version_attempt <= DECOMMISSION_VERSION_COPY_ATTEMPTS { + if version_attempt > 1 || target_busy_attempt > 0 { + record_decommission_object_attempt(attempts); + } let result = self .run_guarded_decommission_side_effect(&rx, &operation_gate, || async { self.delete_object( @@ -14955,6 +15443,9 @@ impl ECStore { let mut target_busy_attempt: usize = 0; let mut target_permit = None; while version_attempt <= DECOMMISSION_VERSION_COPY_ATTEMPTS { + if version_attempt > 1 || target_busy_attempt > 0 { + record_decommission_object_attempt(attempts); + } if version.is_remote() { let result = self .run_guarded_decommission_side_effect(&rx, &operation_gate, || async { @@ -15595,20 +16086,23 @@ impl ECStore { pool_meta.version = version; } let generation = self.active_decommission_generation(idx).await?; - self.decommission_entry( - rx, - idx, - generation, - entry, - bucket, - set, - None, - None, - None, - expected_bucket_incarnation_id, - source_changed_exhaustions, + resolve_decommission_entry_test_outcome( + self.decommission_entry( + rx, + idx, + generation, + entry, + bucket, + set, + None, + None, + None, + expected_bucket_incarnation_id, + source_changed_exhaustions, + &mut 0, + ) + .await, ) - .await } #[cfg(all(test, feature = "test-util"))] @@ -15625,20 +16119,23 @@ impl ECStore { Some(self.bucket_incarnation_id_from_disk(&bucket).await?) }; let generation = self.active_decommission_generation(idx).await?; - self.decommission_entry( - CancellationToken::new(), - idx, - generation, - entry, - bucket, - set, - None, - None, - None, - expected_bucket_incarnation_id, - Arc::new(AtomicUsize::new(0)), + resolve_decommission_entry_test_outcome( + self.decommission_entry( + CancellationToken::new(), + idx, + generation, + entry, + bucket, + set, + None, + None, + None, + expected_bucket_incarnation_id, + Arc::new(AtomicUsize::new(0)), + &mut 0, + ) + .await, ) - .await } #[tracing::instrument(skip(self, rx))] @@ -21842,6 +22339,280 @@ mod tests { assert!(err.to_string().contains("failed to apply lifecycle expiry action")); } + /// The capacity failure classification is the single place that decides + /// between retrying, ignoring, and propagating a decommission failure. + #[test] + fn decommission_capacity_failure_classification_covers_each_class() { + let gate_busy = decommission_capacity_blocked_error(format!( + "{DECOMMISSION_CAPACITY_TARGET_GATE_BUSY_PREFIX}2{DECOMMISSION_CAPACITY_TARGET_GATE_BUSY_SUFFIX}" + )); + assert_eq!( + classify_decommission_capacity_failure(&gate_busy), + DecommissionCapacityFailure::TargetGateBusy { target_pool_index: 2 } + ); + + let intent_conflict = + decommission_capacity_blocked_error("decommission target mutation has an unresolved target capacity intent"); + assert_eq!( + classify_decommission_capacity_failure(&intent_conflict), + DecommissionCapacityFailure::IntentConflict + ); + + let missing_source = Error::VersionNotFound("bucket".to_string(), "object".to_string(), "version".to_string()); + assert_eq!( + classify_decommission_capacity_failure(&missing_source), + DecommissionCapacityFailure::BenignContention + ); + + let overwrite = Error::DataMovementOverwriteErr("bucket".to_string(), "object".to_string(), "v1".to_string()); + assert_eq!( + classify_decommission_capacity_failure(&overwrite), + DecommissionCapacityFailure::BenignContention + ); + + let invalid_upload = Error::InvalidUploadID("bucket".to_string(), "object".to_string(), "upload".to_string()); + assert_eq!( + classify_decommission_capacity_failure(&invalid_upload), + DecommissionCapacityFailure::BenignContention + ); + + assert_eq!( + classify_decommission_capacity_failure(&Error::SlowDown), + DecommissionCapacityFailure::Fatal + ); + assert_eq!( + classify_decommission_capacity_failure(&decommission_capacity_blocked_error( + "decommission target mutation reservation identity is stale" + )), + DecommissionCapacityFailure::Fatal + ); + } + + /// A stage wrapper must not hide the underlying capacity class. + #[test] + fn decommission_capacity_failure_classification_sees_through_a_stage_wrapper() { + let gate_busy = decommission_capacity_blocked_error(format!( + "{DECOMMISSION_CAPACITY_TARGET_GATE_BUSY_PREFIX}1{DECOMMISSION_CAPACITY_TARGET_GATE_BUSY_SUFFIX}" + )); + let wrapped = + data_movement::data_movement_stage_error_for_test("decommission_object", "put_object_part", "b", "o", gate_busy); + + assert_eq!( + classify_decommission_capacity_failure(&wrapped), + DecommissionCapacityFailure::TargetGateBusy { target_pool_index: 1 } + ); + assert!(matches!( + resolve_decommission_entry_failure(wrapped), + Ok(DecommissionEntryOutcome::Deferred(_)) + )); + } + + #[tokio::test] + #[serial_test::serial] + async fn decommission_deferred_directory_object_is_copied_and_removed() { + use crate::object_api::PutObjReader; + let (_temp_dirs, store, _other_store) = + Box::pin(crate::services::rebalance::test_two_pool_stores_with_isolated_node_contexts(None)).await; + let bucket = "deferred-directory-object"; + let object = "folder/"; + Box::pin(store.make_bucket(bucket, &MakeBucketOptions::default())) + .await + .expect("create bucket"); + let incarnation = store.bucket_incarnation_id(bucket).await.expect("bucket incarnation"); + Box::pin(store.pools[0].put_object( + bucket, + &encode_dir_object(object), + &mut PutObjReader::from_vec(Vec::new()), + &ObjectOptions { + expected_bucket_incarnation_id: Some(incarnation), + ..Default::default() + }, + )) + .await + .expect("seed directory object"); + assert!( + store.pools[0] + .get_disks_by_key(&encode_dir_object(object)) + .load_file_info_versions_exact(bucket, object) + .await + .expect("read seeded directory object") + .is_some() + ); + let retry_object = "retry.bin"; + Box::pin(store.pools[0].put_object( + bucket, + retry_object, + &mut PutObjReader::from_vec(b"retry contents".to_vec()), + &ObjectOptions { + expected_bucket_incarnation_id: Some(incarnation), + ..Default::default() + }, + )) + .await + .expect("seed retry object"); + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + set_decommission_capacity_info_overrides_for_test( + store.id, + vec![vec![ + DecommissionPoolCapacityInfo::for_test(0, layout, 0, 16_384, 16_384), + DecommissionPoolCapacityInfo::for_test(1, layout, 131_072, 131_072, 0), + ]], + ); + Box::pin(store.save_current_pool_meta_for_decommission_start(&[0], Vec::new())) + .await + .expect("activate reservation"); + let generation = store.active_decommission_generation(0).await.expect("active generation"); + let source_set = store.pools[0].get_disks_by_key(&encode_dir_object(object)); + let entry_error = Arc::new(tokio::sync::Mutex::new(None)); + let outcome = Box::pin(store.clone().replay_deferred_decommission_entry( + CancellationToken::new(), + 0, + 0, + generation, + DeferredDecommissionEntry { + name: object.into(), + attempts: 2, + }, + bucket.into(), + source_set.clone(), + None, + None, + None, + Some(incarnation), + Arc::new(AtomicUsize::new(0)), + Arc::new(Semaphore::new(1)), + entry_error.clone(), + )) + .await + .expect("replay deferred directory object"); + assert!(matches!(outcome, DecommissionEntryOutcome::Complete)); + assert!(entry_error.lock().await.is_none()); + assert!( + source_set + .load_file_info_versions_exact(bucket, object) + .await + .expect("read exact source versions") + .is_none(), + "source object must be removed" + ); + let target = store.pools[1] + .get_disks_by_key(&encode_dir_object(object)) + .load_file_info_versions_exact(bucket, object) + .await + .expect("read exact target versions") + .expect("directory object must reach target"); + assert_eq!(target.versions.len(), 1); + assert_eq!(target.versions[0].size, 0); + let faults = Arc::new(AtomicUsize::new(0)); + let observed = faults.clone(); + let _fault = DecommissionTestFaultGuard::install(Arc::new(move |stage, b, o, attempt, success| { + if stage == "object_read" && b == bucket && o == retry_object && attempt == 1 && success { + observed.fetch_add(1, Ordering::Relaxed); + return true; + } + false + })); + let mut attempts = 2; + let result = Box::pin(store.decommission_entry( + CancellationToken::new(), + 0, + generation, + MetaCacheEntry { + name: retry_object.into(), + ..Default::default() + }, + bucket.into(), + store.pools[0].get_disks_by_key(retry_object), + None, + None, + None, + Some(incarnation), + Arc::new(AtomicUsize::new(0)), + &mut attempts, + )) + .await + .expect("retry migration"); + assert!(matches!(result, DecommissionEntryOutcome::Complete)); + assert_eq!(faults.load(Ordering::Relaxed), 1, "inject one actual read failure"); + assert_eq!(attempts, 4, "carry two attempts, then count the entry pass and copy retry"); + } + + #[test] + fn decommission_object_attempt_metric_keeps_the_peak() { + let mut attempts = 0; + record_decommission_object_attempt(&mut attempts); + record_decommission_object_attempt(&mut attempts); + let mut deferred = DeferredDecommissionEntry { + name: "object".into(), + attempts, + }; + record_decommission_object_attempt(&mut deferred.attempts); + assert_eq!(deferred.attempts, 3); + assert!(*DECOMMISSION_OBJECT_ATTEMPTS_MAX.lock().expect("attempt peak lock") >= 3); + } + + /// Gate-wait durations are recorded on the same jittered path the retry + /// uses, so the histogram cannot drift from the actual wait. + #[test] + fn decommission_capacity_gate_retry_delay_is_bounded_by_its_ceiling() { + for attempt in 1..=6 { + let delay = decommission_capacity_gate_retry_delay(attempt); + assert!( + delay <= decommission_retry_backoff_ceiling(DECOMMISSION_CAPACITY_GATE_RETRY_BASE, attempt), + "gate wait must stay within its jittered ceiling" + ); + } + } + + /// A set that still holds a deferred entry must not report success. + #[test] + fn decommission_deferred_drain_invariant_rejects_unresolved_entries() { + assert!(ensure_decommission_deferred_drained("bucket-a", 2, 0).is_ok()); + + let err = ensure_decommission_deferred_drained("bucket-a", 2, 3) + .expect_err("a set with deferred entries remaining must not complete"); + let message = err.to_string(); + assert!(message.contains("bucket-a"), "{message}"); + assert!(message.contains('2'), "{message}"); + assert!(message.contains('3'), "{message}"); + } + + /// Contention defers, benign contention completes without counting a + /// failure, and fatal errors stay terminal. Folding either of the first two + /// into the fatal arm would pause a healthy bucket, and folding a fatal + /// error into a deferral would hide a real migration failure. + #[test] + fn decommission_entry_failure_resolution_covers_each_class() { + let gate_busy = decommission_capacity_blocked_error(format!( + "{DECOMMISSION_CAPACITY_TARGET_GATE_BUSY_PREFIX}1{DECOMMISSION_CAPACITY_TARGET_GATE_BUSY_SUFFIX}" + )); + let intent_conflict = + decommission_capacity_blocked_error("decommission target mutation has an unresolved target capacity intent"); + for err in [gate_busy, intent_conflict] { + let outcome = resolve_decommission_entry_failure(err.clone()) + .expect("transient capacity contention must defer, not fail the entry"); + match outcome { + DecommissionEntryOutcome::Deferred(retained) => assert_eq!(retained, err), + DecommissionEntryOutcome::Complete => panic!("contention must not complete the entry"), + } + } + + for benign in [ + Error::ObjectNotFound("bucket".to_string(), "object".to_string()), + Error::VersionNotFound("bucket".to_string(), "object".to_string(), "version".to_string()), + Error::DataMovementOverwriteErr("bucket".to_string(), "object".to_string(), "v1".to_string()), + Error::InvalidUploadID("bucket".to_string(), "object".to_string(), "upload".to_string()), + ] { + assert!( + matches!(resolve_decommission_entry_failure(benign.clone()), Ok(DecommissionEntryOutcome::Complete)), + "benign contention must complete without failing the entry: {benign}" + ); + } + + assert!(resolve_decommission_entry_failure(Error::SlowDown).is_err()); + assert!(resolve_decommission_entry_failure(Error::OperationCanceled).is_err()); + } + #[test] fn decommission_copy_cleanup_safe_error_accepts_missing_source_errors() { assert!(is_decommission_copy_cleanup_safe_error(&Error::ObjectNotFound( @@ -22228,6 +22999,69 @@ mod tests { assert!(!pool_meta.decommission_complete(0)); } + #[test] + fn decommission_capacity_reporting_preserves_existing_disk_layout() { + // Frozen V2/V3 reader contract from before capacity contention reporting. + #[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] + #[serde(deny_unknown_fields)] + struct PreContentionDecommissionInfo { + #[serde(rename = "startTime", with = "time::serde::rfc3339::option")] + pub start_time: Option, + #[serde(rename = "startSize")] + pub start_size: usize, + #[serde(rename = "totalSize")] + pub total_size: usize, + #[serde(rename = "currentSize")] + pub current_size: usize, + #[serde(rename = "complete")] + pub complete: bool, + #[serde(rename = "failed")] + pub failed: bool, + #[serde(rename = "canceled")] + pub canceled: bool, + #[serde(rename = "queued", default)] + pub queued: bool, + #[serde(rename = "queuedBuckets", default)] + pub queued_buckets: Vec, + #[serde(rename = "decommissionedBuckets", default)] + pub decommissioned_buckets: Vec, + #[serde(rename = "bucket", default)] + pub bucket: String, + #[serde(rename = "prefix", default)] + pub prefix: String, + #[serde(rename = "object", default)] + pub object: String, + #[serde(rename = "objectsDecommissioned")] + pub items_decommissioned: usize, + #[serde(rename = "objectsDecommissionedFailed")] + pub items_decommission_failed: usize, + #[serde(rename = "bytesDecommissioned")] + pub bytes_done: usize, + #[serde(rename = "bytesDecommissionedFailed")] + pub bytes_failed: usize, + #[serde(rename = "terminalReloadAttemptAt", with = "time::serde::rfc3339::option", default)] + pub terminal_reload_attempt_at: Option, + #[serde(rename = "terminalReloadFailures", default)] + pub terminal_reload_failures: Vec, + #[serde(rename = "unresolvedEntries", default)] + pub unresolved_entries: Vec, + #[serde(rename = "capacityReservation", default)] + pub capacity_reservation: Option, + #[serde(rename = "capacityBlockedReason", default)] + pub capacity_blocked_reason: Option, + } + let current = PersistedPoolDecommissionInfo { + capacity_blocked_reason: Some("target gate busy".into()), + ..Default::default() + }; + for bytes in [rmp_serde::to_vec(¤t), rmp_serde::to_vec_named(¤t)] { + let bytes = bytes.expect("serialize current decommission state"); + let old: PreContentionDecommissionInfo = + rmp_serde::from_slice(&bytes).expect("existing V2/V3 readers must accept new writes"); + assert_eq!(old.capacity_blocked_reason, current.capacity_blocked_reason); + } + } + #[test] fn pool_meta_persists_decommission_resume_queues() { let start_time = OffsetDateTime::now_utc(); @@ -23141,8 +23975,8 @@ mod pools_tests { classify_decommission_terminal_state, count_decommission_item, decommission_cancel_signal_result, decommission_durable_ilm_receipt_path, decommission_durable_ilm_receipt_run_prefix, decommission_durable_ilm_receipt_run_token, decommission_entry_queue_capacity, decommission_item_size, - decommission_meta_bucket_options, decommission_physical_pool_capacity, decommission_retry_backoff_delay, - decommission_start_pool_state, decommission_unresolved_listing_error, dedup_indices, + decommission_meta_bucket_options, decommission_physical_pool_capacity, decommission_retry_backoff_ceiling, + decommission_retry_backoff_delay, decommission_start_pool_state, decommission_unresolved_listing_error, dedup_indices, default_decommission_bucket_concurrency, default_decommission_entry_concurrency, drain_decommission_entry_queue, enqueue_decommission_entry, ensure_decommission_cancel_allowed, ensure_decommission_capacity_reservations_available, ensure_decommission_clear_allowed, ensure_decommission_generation, ensure_decommission_listing_disks_available, @@ -23181,14 +24015,14 @@ mod pools_tests { with_decommission_entry_context, }; use super::{ - DECOMMISSION_CAPACITY_TARGET_GATE_MAX_ATTEMPTS, DecommissionCapacityAdmission, DecommissionCapacityOwner, - DecommissionCapacityReleaseProof, DecommissionCapacityReservation, DecommissionCapacityTemporaryMutation, - decommission_capacity_mutation_id, decommission_capacity_target_gate_retry_exhausted, - ensure_decommission_target_owner_admission, ensure_exact_delete_capacity_namespace_fences, - ensure_external_decommission_target_admission, is_decommission_capacity_blocked_error, - plan_exact_delete_capacity_reconciliations, record_decommission_target_consumption, release_decommission_target_inflight, - reserve_decommission_target_pending, resolve_decommission_target_pending, - set_decommission_capacity_info_overrides_for_test, + DECOMMISSION_CAPACITY_TARGET_GATE_MAX_ATTEMPTS, DECOMMISSION_RETRY_BACKOFF_CAP, DecommissionCapacityAdmission, + DecommissionCapacityOwner, DecommissionCapacityReleaseProof, DecommissionCapacityReservation, + DecommissionCapacityTemporaryMutation, decommission_capacity_mutation_id, + decommission_capacity_target_gate_retry_exhausted, ensure_decommission_target_owner_admission, + ensure_exact_delete_capacity_namespace_fences, ensure_external_decommission_target_admission, + is_decommission_capacity_blocked_error, plan_exact_delete_capacity_reconciliations, + record_decommission_target_consumption, release_decommission_target_inflight, reserve_decommission_target_pending, + resolve_decommission_target_pending, set_decommission_capacity_info_overrides_for_test, }; use crate::bucket::lifecycle::{ DurableIlmRecordCheckpoint, @@ -27727,12 +28561,27 @@ mod pools_tests { } #[test] - fn test_decommission_retry_backoff_delay_grows_linearly() { + fn test_decommission_retry_backoff_ceiling_grows_exponentially_and_caps() { let base = StdDuration::from_millis(100); - assert_eq!(decommission_retry_backoff_delay(base, 1), base); - assert_eq!(decommission_retry_backoff_delay(base, 3), StdDuration::from_millis(300)); - assert_eq!(decommission_retry_backoff_delay(base, usize::MAX), base.saturating_mul(u32::MAX)); + assert_eq!(decommission_retry_backoff_ceiling(base, 1), StdDuration::from_millis(100)); + assert_eq!(decommission_retry_backoff_ceiling(base, 2), StdDuration::from_millis(200)); + assert_eq!(decommission_retry_backoff_ceiling(base, 3), StdDuration::from_millis(400)); + assert_eq!(decommission_retry_backoff_ceiling(base, 4), StdDuration::from_millis(800)); + assert_eq!(decommission_retry_backoff_ceiling(base, 5), DECOMMISSION_RETRY_BACKOFF_CAP); + assert_eq!(decommission_retry_backoff_ceiling(base, usize::MAX), DECOMMISSION_RETRY_BACKOFF_CAP); + } + + #[test] + fn test_decommission_retry_backoff_delay_stays_within_jittered_ceiling() { + let base = StdDuration::from_millis(100); + + for attempt in 1..=6 { + let ceiling = decommission_retry_backoff_ceiling(base, attempt); + let delay = decommission_retry_backoff_delay(base, attempt); + assert!(delay <= ceiling, "delay {delay:?} must not exceed ceiling {ceiling:?}"); + } + assert_eq!(decommission_retry_backoff_delay(StdDuration::ZERO, 3), StdDuration::ZERO); } #[test] diff --git a/crates/ecstore/src/core/pools_test.rs b/crates/ecstore/src/core/pools_test.rs index 6fc60bb04..b1d393c63 100644 --- a/crates/ecstore/src/core/pools_test.rs +++ b/crates/ecstore/src/core/pools_test.rs @@ -4253,6 +4253,13 @@ mod decommission_lock_order_tests { panic!("cancellation probe finished before observing target contention: {result:?}"); } } + // The mutation absorbs the gate contention inline before the outer wait + // loop ever sees it, and that inline budget is bounded. + assert_eq!( + retry_observer.target_gate_inline_retries(), + crate::core::pools::DECOMMISSION_MUTATION_GATE_MAX_INLINE_ATTEMPTS - 1, + "inline target-gate retries must stop at their bounded attempt budget" + ); tokio::time::sleep(Duration::from_millis(800)).await; assert_eq!( retry_observer.target_gate_exact_reloads(), diff --git a/docs/architecture/decommission-compatibility.md b/docs/architecture/decommission-compatibility.md index 54ba71752..1dd2590e8 100644 --- a/docs/architecture/decommission-compatibility.md +++ b/docs/architecture/decommission-compatibility.md @@ -92,7 +92,21 @@ When decommission metadata is present, `decommissionInfo` includes: - progress counters: `objectsDecommissioned`, `objectsDecommissionedFailed`, `bytesDecommissioned`, and `bytesDecommissionedFailed`; - current location: `bucket`, `prefix`, and `object`; - queue/history lists: `queuedBuckets` and `decommissionedBuckets`; -- `waitingReason`: `queued` for queued entries and `waiting_for_worker` when metadata exists but no worker has started. +- `waitingReason`: `capacity` while the pool is paused on target capacity, + `queued` for queued entries, and `waiting_for_worker` when metadata exists but + no worker has started; +- `capacityBlockedReason`: the persisted detail for the active capacity pause, + absent once the pause clears. + +A capacity pause is reported ahead of the worker states so operators can tell +"contending but progressing" from "genuinely out of target capacity". The existing `pool.bin` layout is unchanged; +cumulative pause history requires a separately versioned persistence contract. + +The object-attempt metrics count entry passes and repeated version-copy attempts +within a listing and its deferred replay. Inline gate waits use the gate-retry +counter. A new listing after a durable pause starts a new per-object observation; +the maximum is the highest observation in the process, not a persisted lifetime +attempt count. This makes queued pools and stalled metadata visible without requiring operators to inspect pool metadata files directly. diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index 3ba2d330e..469f550d7 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -43,6 +43,10 @@ use std::sync::Arc; use tracing::{debug, error, info, warn}; pub type AdminUsecaseResult = Result; + +/// `waitingReason` value reported while a pool is paused on target capacity. +const DECOMMISSION_WAITING_REASON_CAPACITY: &str = "capacity"; + pub const ADMIN_CLUSTER_SNAPSHOT_ROUTE: &str = "/rustfs/admin/v4/cluster/snapshot"; pub const ADMIN_EXTENSIONS_CATALOG_ROUTE: &str = "/rustfs/admin/v4/extensions/catalog"; pub const ADMIN_RUNTIME_CAPABILITIES_ROUTE: &str = "/rustfs/admin/v4/runtime/capabilities"; @@ -108,6 +112,8 @@ pub struct AdminPoolDecommissionInfo { pub bytes_failed: usize, #[serde(rename = "waitingReason")] pub waiting_reason: Option, + #[serde(rename = "capacityBlockedReason", skip_serializing_if = "Option::is_none")] + pub capacity_blocked_reason: Option, #[serde(rename = "unresolvedEntries", skip_serializing_if = "Vec::is_empty")] pub unresolved_entries: Vec, } @@ -622,12 +628,23 @@ impl DefaultAdminUsecase { bytes_done: info.bytes_done, bytes_failed: info.bytes_failed, waiting_reason, + capacity_blocked_reason: info.capacity_blocked_reason, unresolved_entries: info.unresolved_entries, } } + /// Why a pool is not currently making progress. + /// + /// A durable capacity pause is reported ahead of the worker states: it is + /// the actionable condition, and `capacityBlockedReason` carries the detail. fn decommission_waiting_reason(info: &PoolDecommissionInfo) -> Option<&'static str> { - if !info.has_decommission_state() || info.complete || info.failed || info.canceled || info.start_time.is_some() { + if !info.has_decommission_state() || info.complete || info.failed || info.canceled { + return None; + } + if info.capacity_blocked_reason.is_some() { + return Some(DECOMMISSION_WAITING_REASON_CAPACITY); + } + if info.start_time.is_some() { return None; } if info.queued { @@ -995,6 +1012,57 @@ mod tests { assert_eq!(item.rebalance_status, "started"); } + /// A durable capacity pause must be distinguishable from "slow but + /// progressing": the pause surfaces its own waiting reason, the persisted + /// detail. + #[test] + fn admin_pool_list_item_exposes_decommission_capacity_pause() { + let item = DefaultAdminUsecase::pool_list_item_from_status( + PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: Some(PoolDecommissionInfo { + start_time: Some(OffsetDateTime::UNIX_EPOCH), + total_size: 1_000, + current_size: 500, + capacity_blocked_reason: Some("target pool 1 target capacity mutation gate is busy".to_string()), + ..Default::default() + }), + }, + (RebalStatus::None, false), + ); + + let value = serde_json::to_value(item).expect("paused pool status should serialize"); + assert_eq!(value["decommissionInfo"]["waitingReason"], "capacity"); + assert_eq!( + value["decommissionInfo"]["capacityBlockedReason"], + "target pool 1 target capacity mutation gate is busy" + ); + } + + /// Once the pause clears, `waitingReason` must return to the worker states + /// without retaining a stale capacity reason. + #[test] + fn admin_pool_list_item_clears_capacity_pause() { + let item = DefaultAdminUsecase::pool_list_item_from_status( + PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: Some(PoolDecommissionInfo { + start_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }), + }, + (RebalStatus::None, false), + ); + + let value = serde_json::to_value(item).expect("resumed pool status should serialize"); + assert!(value["decommissionInfo"]["waitingReason"].is_null()); + assert!(value["decommissionInfo"].get("capacityBlockedReason").is_none()); + } + #[test] fn admin_pool_list_item_exposes_queued_decommission_state() { let item = DefaultAdminUsecase::pool_list_item_from_status(