diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index e61ab96ed..50c55cb4f 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -141,6 +141,9 @@ 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 DECOMMISSION_CAPACITY_TARGET_OVER_BUDGET_PREFIX: &str = "target pool "; +const DECOMMISSION_CAPACITY_TARGET_OVER_BUDGET_SUFFIX: &str = " is over a temporary-copy budget: "; +const METRIC_DECOMMISSION_CAPACITY_TARGET_OVER_BUDGET: &str = "rustfs_decommission_capacity_target_over_budget"; 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 @@ -1063,6 +1066,16 @@ pub(crate) fn is_decommission_capacity_target_gate_busy(err: &Error) -> bool { decommission_capacity_target_gate_busy_index(err).is_some() } +fn decommission_capacity_target_over_budget_index(err: &Error) -> Option { + if let Error::DecommissionCapacityBlocked { message } = err { + let (target_pool_index, _) = message + .strip_prefix(DECOMMISSION_CAPACITY_TARGET_OVER_BUDGET_PREFIX)? + .split_once(DECOMMISSION_CAPACITY_TARGET_OVER_BUDGET_SUFFIX)?; + return target_pool_index.parse().ok(); + } + data_movement::data_movement_stage_source(err).and_then(decommission_capacity_target_over_budget_index) +} + /// How the decommission capacity path must treat a failure. /// /// Callers decide between retrying, ignoring, and propagating from this single @@ -1076,6 +1089,10 @@ pub(crate) enum DecommissionCapacityFailure { /// Another mutation's durable target capacity intent is still unresolved. /// Transient, but converged on a different retry budget than gate contention. IntentConflict, + /// The target pool went over a temporary-copy budget after this mutation + /// selected it. Target selection skips such a pool, so retrying the entry + /// places the object elsewhere; it shares the intent-conflict retry budget. + TargetOverBudget { target_pool_index: usize }, /// The mutation is no longer needed: the object or version disappeared, or /// an equivalent target already exists. Never counted as a failure. BenignContention, @@ -1100,6 +1117,9 @@ fn classify_decommission_capacity_failure(err: &Error) -> DecommissionCapacityFa if let Some(target_pool_index) = decommission_capacity_target_gate_busy_index(err) { return DecommissionCapacityFailure::TargetGateBusy { target_pool_index }; } + if let Some(target_pool_index) = decommission_capacity_target_over_budget_index(err) { + return DecommissionCapacityFailure::TargetOverBudget { target_pool_index }; + } if is_decommission_capacity_intent_conflict(err) { return DecommissionCapacityFailure::IntentConflict; } @@ -1116,7 +1136,10 @@ enum DecommissionCapacityRetryKind { fn decommission_capacity_retry_kind(err: &Error, intent_conflict_attempt: usize) -> Option { (intent_conflict_attempt < DECOMMISSION_CAPACITY_INTENT_CONFLICT_MAX_ATTEMPTS - && matches!(classify_decommission_capacity_failure(err), DecommissionCapacityFailure::IntentConflict)) + && matches!( + classify_decommission_capacity_failure(err), + DecommissionCapacityFailure::IntentConflict | DecommissionCapacityFailure::TargetOverBudget { .. } + )) .then_some(DecommissionCapacityRetryKind::IntentConflict) } @@ -1422,6 +1445,14 @@ fn release_decommission_capacity_reservation(info: &mut PoolDecommissionInfo, re "pool_index" => reservation.source_pool_index.to_string() ) .set(0.0); + for target in &reservation.targets { + metrics::gauge!( + METRIC_DECOMMISSION_CAPACITY_TARGET_OVER_BUDGET, + "pool_index" => reservation.source_pool_index.to_string(), + "target_pool_index" => target.pool_index.to_string() + ) + .set(0.0); + } true } @@ -1832,10 +1863,76 @@ fn observe_decommission_capacity_reservation( .record(reservation.prediction_error_bytes.unsigned_abs() as f64); } +/// Rejection of growth into a target pool where this allocation holds more +/// inflight or pending bytes than its temporary-copy budget. Selection skips +/// such a pool, so the failure is retryable (`TargetOverBudget`). +fn decommission_target_over_budget_error( + source_index: usize, + reservation: &DecommissionCapacityReservation, + target: &DecommissionCapacityTarget, + phase: &str, +) -> Error { + let temporary_budget = target.temporary_copy_budget(reservation.temporary_copies); + let detail = if target.inflight_physical_bytes > temporary_budget { + format!( + "source pool {source_index} target pool {} has {} inflight physical bytes, exceeding its {temporary_budget}-byte temporary-copy reservation during {phase}", + target.pool_index, target.inflight_physical_bytes + ) + } else { + format!( + "source pool {source_index} target pool {} has {} pending physical bytes, exceeding its {temporary_budget}-byte committed-copy reservation during {phase}", + target.pool_index, target.pending_physical_bytes + ) + }; + decommission_capacity_blocked_error(format!( + "{DECOMMISSION_CAPACITY_TARGET_OVER_BUDGET_PREFIX}{}{DECOMMISSION_CAPACITY_TARGET_OVER_BUDGET_SUFFIX}{detail}", + target.pool_index + )) +} + +/// Target pools where an active source's allocation holds more inflight or +/// pending bytes than its temporary-copy budget. Growth into them fails, so +/// target selection skips them. +fn decommission_over_budget_target_pools(meta: &PoolMeta) -> HashSet { + active_decommission_source_indices(meta) + .into_iter() + .filter_map(|source_index| { + meta.pools + .get(source_index)? + .decommission + .as_ref()? + .capacity_reservation + .as_ref() + }) + .flat_map(|reservation| { + reservation + .targets + .iter() + .filter(|target| target.exceeds_temporary_copy_budget(reservation.temporary_copies)) + .map(|target| target.pool_index) + }) + .collect() +} + +/// Check the active reservations against the target pools' real free space. +/// +/// Per target pool, the remaining reservations of every source must fit the +/// pool's free space plus the inflight and pending bytes already placed there +/// (the free-space check). That check credits those bytes back only because +/// they lie inside the reservation, which the temporary-copy budget of each +/// source allocation guarantees (the budget check). A budget violation +/// therefore makes only that pool's free-space check unreliable: the +/// free-space check is computed per pool, and no other pool's credit depends +/// on it. So the budget check guards growth, not the whole decommission: +/// `growth_pool` names the pool a mutation is about to place bytes in, and the +/// budget of every source allocation on that pool must hold. Finalize and the +/// per-entry check pass `None`; they place no new bytes, and failing them would +/// pause every decommission for one pool's ledger. fn ensure_decommission_capacity_reservations_available( meta: &PoolMeta, capacity_infos: &[DecommissionPoolCapacityInfo], phase: &'static str, + growth_pool: Option, ) -> Result<()> { let mut required_by_target = HashMap::::new(); let mut inflight_by_target = HashMap::::new(); @@ -1861,19 +1958,15 @@ fn ensure_decommission_capacity_reservations_available( ) .set(reservation.remaining_peak_physical_bytes() as f64); for target in &reservation.targets { - let temporary_budget = target.remaining_reserved_physical_bytes(reservation.temporary_copies) - / 1usize.saturating_add(reservation.temporary_copies); - if target.inflight_physical_bytes > temporary_budget { - return Err(decommission_capacity_blocked_error(format!( - "source pool {source_index} target pool {} has {} inflight physical bytes, exceeding its {temporary_budget}-byte temporary-copy reservation during {phase}", - target.pool_index, target.inflight_physical_bytes - ))); - } - if target.pending_physical_bytes > temporary_budget { - return Err(decommission_capacity_blocked_error(format!( - "source pool {source_index} target pool {} has {} pending physical bytes, exceeding its {temporary_budget}-byte committed-copy reservation during {phase}", - target.pool_index, target.pending_physical_bytes - ))); + let over_budget = target.exceeds_temporary_copy_budget(reservation.temporary_copies); + metrics::gauge!( + METRIC_DECOMMISSION_CAPACITY_TARGET_OVER_BUDGET, + "pool_index" => source_index.to_string(), + "target_pool_index" => target.pool_index.to_string() + ) + .set(if over_budget { 1.0 } else { 0.0 }); + if over_budget && growth_pool == Some(target.pool_index) { + return Err(decommission_target_over_budget_error(source_index, reservation, target, phase)); } let required = required_by_target.entry(target.pool_index).or_default(); *required = required.saturating_add(target.remaining_reserved_physical_bytes(reservation.temporary_copies)); @@ -2096,7 +2189,12 @@ fn grow_decommission_target_reservation( Ok(true) })(); if result.is_ok() { - if let Err(err) = ensure_decommission_capacity_reservations_available(meta, capacity_infos, "reservation growth") { + if let Err(err) = ensure_decommission_capacity_reservations_available( + meta, + capacity_infos, + "reservation growth", + Some(target_pool_index), + ) { *meta = original; return Err(err); } @@ -2175,8 +2273,7 @@ fn reserve_decommission_target_pending( if requested_physical_bytes <= target.pending_physical_bytes { return Ok(0); } - let budget = target.remaining_reserved_physical_bytes(reservation.temporary_copies) - / 1usize.saturating_add(reservation.temporary_copies); + let budget = target.temporary_copy_budget(reservation.temporary_copies); if requested_physical_bytes > budget { return Err(decommission_capacity_blocked_error(format!( "source pool {source_pool_index} target pool {target_pool_index} mutation intent size changed during retry" @@ -2196,8 +2293,7 @@ fn reserve_decommission_target_pending( } } } - let budget = target.remaining_reserved_physical_bytes(reservation.temporary_copies) - / 1usize.saturating_add(reservation.temporary_copies); + let budget = target.temporary_copy_budget(reservation.temporary_copies); if requested_physical_bytes > budget { return Err(decommission_capacity_blocked_error(format!( "source pool {source_pool_index} target pool {target_pool_index} mutation requires {requested_physical_bytes} physical bytes, but only {budget} committed-copy bytes remain" @@ -2345,8 +2441,8 @@ fn decommission_capacity_intent_adoption_plan( /// adoption never matches the derived id, so it never releases the marker: /// the bytes stay inflight until the reservation is released, and once later /// commits shrink that target's temporary-copy budget below its inflight -/// bytes, `ensure_decommission_capacity_reservations_available` rejects every -/// decommission mutation, not only those for that target. +/// bytes, that target pool stops accepting decommission growth +/// (`decommission_over_budget_target_pools`). pub(crate) fn decommission_adopted_intent_mutation_id(mutation_id: uuid::Uuid) -> uuid::Uuid { let mut hasher = Sha256::new(); hasher.update(b"rustfs-decommission-adopted-intent"); @@ -2512,8 +2608,7 @@ fn decommission_adopted_intent_split( let adopted = marker_bytes(decommission_adopted_intent_mutation_id(intent.mutation_id)); let recorded = staged.saturating_add(adopted); let other_inflight = target.inflight_physical_bytes.saturating_sub(recorded); - let temporary_budget = target.remaining_reserved_physical_bytes(reservation.temporary_copies) - / 1usize.saturating_add(reservation.temporary_copies); + let temporary_budget = target.temporary_copy_budget(reservation.temporary_copies); let completion = adopted.max(intent.physical_bytes); let total = staged .saturating_add(completion) @@ -2599,12 +2694,18 @@ fn release_decommission_target_mutation_markers( } /// The target's remaining allocation if it still fits one object of -/// `expected_data_bytes` with its temporary copies. +/// `expected_data_bytes` with its temporary copies. A pool over a +/// temporary-copy budget (`decommission_over_budget_target_pools`) accepts no +/// growth, so it is no candidate. fn decommission_capacity_target_candidate( reservation: &DecommissionCapacityReservation, target: &DecommissionCapacityTarget, expected_data_bytes: usize, + over_budget_pools: &HashSet, ) -> Option<(usize, usize)> { + if over_budget_pools.contains(&target.pool_index) { + return None; + } let expected_physical_bytes = capacity_target_physical_bytes(expected_data_bytes.max(1), target.layout).ok()?; let required_peak = expected_physical_bytes.saturating_mul(1usize.saturating_add(reservation.temporary_copies)); let remaining = target.remaining_reserved_physical_bytes(reservation.temporary_copies); @@ -2619,18 +2720,25 @@ fn decommission_capacity_target_candidate( /// another target would leave it behind as inflight bytes. Otherwise the /// target with the most remaining allocation wins. /// -/// For an adopted completion the fallback cannot help: the target stops -/// fitting the object only when its temporary-copy budget drops below the -/// object's bytes, which the adopted marker already holds as inflight, so -/// `ensure_decommission_capacity_reservations_available` fails before any -/// placement on another target is admitted. +/// The fallback never settles the adopted marker; it stays until the +/// reservation is released. It happens when the holding pool is over a +/// temporary-copy budget for any source, or no longer fits the object. For an +/// adopted completion the second implies the first: the pool stops fitting +/// the object only when this source's budget there drops below the object's +/// bytes, which the adopted marker already holds as inflight. Another source +/// over budget on the same pool also bypasses it; waiting for that source +/// instead would tie this decommission's progress to another one's ledger, +/// while the stranded marker costs only this source's growth on a pool that +/// accepts none. fn decommission_reserved_target_choice( reservation: &DecommissionCapacityReservation, mutation_id: Option, expected_data_bytes: usize, + over_budget_pools: &HashSet, ) -> Option { - let candidate = - |target: &DecommissionCapacityTarget| decommission_capacity_target_candidate(reservation, target, expected_data_bytes); + let candidate = |target: &DecommissionCapacityTarget| { + decommission_capacity_target_candidate(reservation, target, expected_data_bytes, over_budget_pools) + }; let adopted_mutation_id = mutation_id.map(decommission_adopted_intent_mutation_id); reservation .targets @@ -2659,6 +2767,7 @@ fn log_decommission_adopted_target_choice( reservation: &DecommissionCapacityReservation, owner: DecommissionCapacityOwner, chosen_pool_index: usize, + over_budget_pools: &HashSet, ) { let Some(mutation_id) = owner.mutation_id else { return; @@ -2698,8 +2807,13 @@ fn log_decommission_adopted_target_choice( pool_index = owner.source_pool_index, target_pool_index = chosen_pool_index, adopted_target_pools = ?adopted_target_pools, + reason = if adopted_target_pools.iter().any(|pool| over_budget_pools.contains(pool)) { + "over_budget" + } else { + "does_not_fit" + }, mutation_id = %mutation_id, - "Decommission retry no longer fits the target that holds its adopted completion" + "Decommission retry cannot return to the target that holds its adopted completion" ); } } @@ -2786,8 +2900,7 @@ fn record_decommission_target_consumption( .iter_mut() .find(|target| target.pool_index == target_pool_index) .ok_or_else(|| decommission_capacity_blocked_error("target allocation disappeared during target write"))?; - let remaining_target_bytes = target.remaining_reserved_physical_bytes(reservation.temporary_copies) - / 1usize.saturating_add(reservation.temporary_copies); + let remaining_target_bytes = target.temporary_copy_budget(reservation.temporary_copies); let remaining_total_bytes = reservation .predicted_physical_bytes .saturating_sub(reservation.consumed_target_physical_bytes); @@ -4513,8 +4626,8 @@ pub(crate) async fn pause_pool_activation_after_durable_save(pool: &Arc, f #[cfg(test)] struct PoolActivationStartProbeState { kind: PoolActivationStartKind, - preflight_side_effect_attempted: std::sync::atomic::AtomicBool, - attempted: std::sync::atomic::AtomicBool, + preflight_side_effect_attempted: AtomicBool, + attempted: AtomicBool, notify: tokio::sync::Notify, } @@ -4532,8 +4645,8 @@ impl PoolActivationStartProbe { pub(crate) fn install(kind: PoolActivationStartKind) -> Self { let state = Arc::new(PoolActivationStartProbeState { kind, - preflight_side_effect_attempted: std::sync::atomic::AtomicBool::new(false), - attempted: std::sync::atomic::AtomicBool::new(false), + preflight_side_effect_attempted: AtomicBool::new(false), + attempted: AtomicBool::new(false), notify: tokio::sync::Notify::new(), }); POOL_ACTIVATION_START_PROBES @@ -4989,9 +5102,9 @@ enum DecommissionEntryOutcome { /// 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::TargetGateBusy { .. } + | DecommissionCapacityFailure::IntentConflict + | DecommissionCapacityFailure::TargetOverBudget { .. } => Ok(DecommissionEntryOutcome::Deferred(err)), DecommissionCapacityFailure::BenignContention => Ok(DecommissionEntryOutcome::Complete), DecommissionCapacityFailure::Fatal => Err(err), } @@ -10330,10 +10443,7 @@ fn plan_exact_delete_capacity_reconciliations( .targets .iter() .find(|target| target.pool_index == target_pool_index) - .map(|target| { - target.remaining_reserved_physical_bytes(reservation.temporary_copies) - / 1usize.saturating_add(reservation.temporary_copies) - }) + .map(|target| target.temporary_copy_budget(reservation.temporary_copies)) .unwrap_or_default(); let remaining_total_physical_bytes = reservation .predicted_physical_bytes @@ -10446,6 +10556,17 @@ impl DecommissionCapacityTarget { .saturating_mul(1usize.saturating_add(temporary_copies)), ) } + + /// The inflight or pending bytes this allocation may hold on its target + /// pool at once: one object's share of its remaining reservation. + fn temporary_copy_budget(&self, temporary_copies: usize) -> usize { + self.remaining_reserved_physical_bytes(temporary_copies) / 1usize.saturating_add(temporary_copies) + } + + fn exceeds_temporary_copy_budget(&self, temporary_copies: usize) -> bool { + let temporary_budget = self.temporary_copy_budget(temporary_copies); + self.inflight_physical_bytes > temporary_budget || self.pending_physical_bytes > temporary_budget + } } #[derive(Debug, Clone, Serialize, Deserialize, Default)] @@ -10708,6 +10829,8 @@ struct DecommissionCapacityLockOrderBarrierState { target_gate_inline_retry_entries: AtomicUsize, target_gate_exact_reloads: AtomicUsize, target_gate_acquire_pause_target: AtomicUsize, + target_gate_acquire_pause_skips: AtomicUsize, + target_gate_acquire_pause_once: AtomicBool, target_gate_acquire_entered: tokio::sync::Notify, target_gate_acquire_release: tokio::sync::Notify, cancel_before_start_entered: tokio::sync::Notify, @@ -10753,6 +10876,8 @@ impl DecommissionCapacityLockOrderBarrier { 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_pause_skips: AtomicUsize::new(0), + target_gate_acquire_pause_once: AtomicBool::new(false), target_gate_acquire_entered: tokio::sync::Notify::new(), target_gate_acquire_release: tokio::sync::Notify::new(), cancel_before_start_entered: tokio::sync::Notify::new(), @@ -10859,6 +10984,17 @@ impl DecommissionCapacityLockOrderBarrier { .store(target_pool_index, Ordering::Release); } + /// Let `skips` acquisitions of the target gate pass, pause the next one, + /// and let every later one pass. + #[cfg(feature = "test-util")] + pub(crate) fn pause_one_target_gate_acquire_after(&self, target_pool_index: usize, skips: usize) { + self.state.target_gate_acquire_pause_skips.store(skips, Ordering::Release); + self.state.target_gate_acquire_pause_once.store(true, Ordering::Release); + self.state + .target_gate_acquire_pause_target + .store(target_pool_index, Ordering::Release); + } + #[cfg(feature = "test-util")] pub(crate) async fn wait_until_target_gate_acquire_paused(&self) { tokio::time::timeout(std::time::Duration::from_secs(30), self.state.target_gate_acquire_entered.notified()) @@ -10984,6 +11120,27 @@ async fn pause_decommission_capacity_before_target_gate_acquire(store_id: uuid:: }) .cloned(); if let Some(barrier) = barrier { + if barrier.target_gate_acquire_pause_once.load(Ordering::Acquire) { + let mut skips = barrier.target_gate_acquire_pause_skips.load(Ordering::Acquire); + while skips > 0 { + match barrier.target_gate_acquire_pause_skips.compare_exchange_weak( + skips, + skips - 1, + Ordering::AcqRel, + Ordering::Acquire, + ) { + Ok(_) => return, + Err(current) => skips = current, + } + } + if barrier + .target_gate_acquire_pause_target + .compare_exchange(target_pool_index, usize::MAX, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { + return; + } + } barrier.target_gate_acquire_entered.notify_one(); barrier.target_gate_acquire_release.notified().await; } @@ -12628,7 +12785,8 @@ impl ECStore { false } else { let data_bytes = expected_data_bytes.unwrap_or(1).max(1); - let grew = grow_decommission_target_reservation( + // Growth already checked every reservation for this target pool. + grow_decommission_target_reservation( &mut snapshot, &capacity_infos, source_pool_index, @@ -12636,9 +12794,7 @@ impl ECStore { data_bytes, mutation_id, OffsetDateTime::now_utc(), - )?; - ensure_decommission_capacity_reservations_available(&snapshot, &capacity_infos, "mutation")?; - grew + )? }; let target_layout = snapshot .pools @@ -12670,11 +12826,7 @@ impl ECStore { .iter() .find(|target| target.pool_index == target_pool_index) .map(|target| { - ( - target.remaining_reserved_physical_bytes(reservation.temporary_copies) - / 1usize.saturating_add(reservation.temporary_copies), - target.inflight_physical_bytes, - ) + (target.temporary_copy_budget(reservation.temporary_copies), target.inflight_physical_bytes) }) }) .unwrap_or_default(); @@ -12870,7 +13022,7 @@ impl ECStore { let capacity_result = if temporary_release { Ok(()) } else { - ensure_decommission_capacity_reservations_available(&snapshot, &capacity_infos, "mutation") + ensure_decommission_capacity_reservations_available(&snapshot, &capacity_infos, "mutation", None) }; if let Err(err) = &capacity_result { snapshot.mark_decommission_capacity_blocked(source_pool_index, err.to_string(), now)?; @@ -13685,7 +13837,7 @@ impl ECStore { { return Ok(()); } - ensure_decommission_capacity_reservations_available(&pool_meta, &capacity_infos, "migration") + ensure_decommission_capacity_reservations_available(&pool_meta, &capacity_infos, "migration", None) } /// Adopt the pending target intents that an earlier incarnation of this @@ -13851,12 +14003,15 @@ impl ECStore { .and_then(|info| info.capacity_reservation.as_ref()) .filter(|reservation| reservation.admits_owner(owner, OffsetDateTime::now_utc())) .ok_or_else(|| decommission_capacity_blocked_error("decommission target selection reservation is stale"))?; + // Growth into a pool whose ledger exceeds a temporary-copy budget + // fails, so place the object on another pool instead. + let over_budget = decommission_over_budget_target_pools(&pool_meta); if let Some(permitted_target_pool_index) = decommission_capacity_target_permit_index(self.id, owner) { if let Some((pool_index, _)) = reservation .targets .iter() .find(|target| target.pool_index == permitted_target_pool_index) - .and_then(|target| decommission_capacity_target_candidate(reservation, target, expected_data_bytes)) + .and_then(|target| decommission_capacity_target_candidate(reservation, target, expected_data_bytes, &over_budget)) { return Ok(pool_index); } @@ -13865,13 +14020,17 @@ impl ECStore { // different target so one mutation never holds two target gates. discard_decommission_capacity_target_permit_except(self.id, owner, None); } - if let Some(pool_index) = decommission_reserved_target_choice(reservation, owner.mutation_id, expected_data_bytes) { - log_decommission_adopted_target_choice(reservation, owner, pool_index); + if let Some(pool_index) = + decommission_reserved_target_choice(reservation, owner.mutation_id, expected_data_bytes, &over_budget) + { + log_decommission_adopted_target_choice(reservation, owner, pool_index, &over_budget); return Ok(pool_index); } if reservation.source_data_equivalent_bytes > 0 { + let mut skipped = over_budget.iter().copied().collect::>(); + skipped.sort_unstable(); return Err(decommission_capacity_blocked_error(format!( - "source pool {} has no target allocation for {expected_data_bytes} data bytes", + "source pool {} has no target allocation for {expected_data_bytes} data bytes; target pools over their temporary-copy budget: {skipped:?}", owner.source_pool_index ))); } @@ -13882,6 +14041,7 @@ impl ECStore { .filter(|capacity| { capacity.pool_index != owner.source_pool_index && !active_sources.contains(&capacity.pool_index) + && !over_budget.contains(&capacity.pool_index) && pool_meta .pools .get(capacity.pool_index) @@ -22529,6 +22689,106 @@ mod tests { .expect("a repeated temporary reconciliation stays idempotent"); } + /// Start a decommission of pool 0 on a three-pool store and strand + /// `stranded` inflight bytes on target 1, as a lost marker would. + async fn start_over_budget_test_decommission(store: &Arc, stranded: usize) { + // A 200-byte peak: target 1 reserves 198 bytes, target 2 the other 2. + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + set_decommission_capacity_info_overrides_for_test( + store.id, + vec![vec![ + DecommissionPoolCapacityInfo::for_test(0, layout, 0, 100, 100), + DecommissionPoolCapacityInfo::for_test(1, layout, 198, 198, 0), + DecommissionPoolCapacityInfo::for_test(2, layout, 2, 2, 0), + ]], + ); + store + .save_current_pool_meta_for_decommission_start(&[0], Vec::new()) + .await + .expect("the decommission reservation should be activated"); + let mut meta = store.pool_meta.read().await.clone(); + { + let reservation = meta.pools[0] + .decommission + .as_mut() + .and_then(|info| info.capacity_reservation.as_mut()) + .expect("the decommission reservation should be active"); + let target = reservation + .targets + .iter_mut() + .find(|target| target.pool_index == 1) + .expect("the reservation should allocate target 1"); + target.temporary_mutations.push(DecommissionCapacityTemporaryMutation { + mutation_id: uuid::Uuid::new_v4(), + physical_bytes: stranded, + }); + target.inflight_physical_bytes += stranded; + target.observed_physical_bytes += stranded; + reservation.inflight_target_physical_bytes += stranded; + reservation.observed_target_physical_bytes += stranded; + } + meta.save(store.pools.clone()).await.expect("persist the stranded marker"); + *store.pool_meta.write().await = meta; + } + + #[tokio::test] + #[serial_test::serial] + async fn target_permit_on_an_over_budget_pool_is_discarded() { + let (_temp_dirs, store, _other_store) = + crate::services::rebalance::test_three_pool_stores_with_isolated_node_contexts(None).await; + // Target 1's 99-byte budget is below the 100 stranded bytes. + start_over_budget_test_decommission(&store, 100).await; + let owner = leftover_intent_test_owner(&store, uuid::Uuid::new_v4()).await; + let guard = store + .acquire_decommission_capacity_target_guard(1) + .await + .expect("acquire target 1's gate for the permit"); + let _permit = install_decommission_capacity_target_permit(store.id, 1, owner, guard).expect("install the permit"); + assert_eq!(decommission_capacity_target_permit_index(store.id, owner), Some(1)); + + let selected = store + .select_decommission_capacity_target_pool(owner, 1) + .await + .expect("another target fits the object"); + assert_eq!(selected, 2, "a permit must not pin the mutation to an over-budget pool"); + assert_eq!( + decommission_capacity_target_permit_index(store.id, owner), + None, + "the permit on the over-budget pool is released" + ); + } + + #[tokio::test] + #[serial_test::serial] + async fn finalize_that_pushes_its_pool_over_budget_does_not_pause_the_decommission() { + let (_temp_dirs, store, _other_store) = + crate::services::rebalance::test_three_pool_stores_with_isolated_node_contexts(None).await; + start_over_budget_test_decommission(&store, 10).await; + let owner = leftover_intent_test_owner(&store, uuid::Uuid::new_v4()).await; + // A 90-byte commit shrinks target 1's budget to 9 bytes, below the 10 + // stranded bytes. + store + .run_decommission_capacity_admitted_mutation(1, Some(owner), Some(90), || async { Ok::<_, Error>(()) }) + .await + .expect("a commit that pushes its pool over budget still finalizes"); + let meta = store.pool_meta.read().await; + let info = meta.pools[0] + .decommission + .as_ref() + .expect("the decommission state should remain present"); + assert_eq!(info.capacity_blocked_reason, None, "finalize must not pause the decommission"); + assert_eq!(decommission_over_budget_target_pools(&meta), HashSet::from([1])); + drop(meta); + let next = leftover_intent_test_owner(&store, uuid::Uuid::new_v4()).await; + assert_eq!( + store + .select_decommission_capacity_target_pool(next, 1) + .await + .expect("another target fits the object"), + 2 + ); + } + #[tokio::test] #[serial_test::serial] async fn uncharged_equivalent_target_without_a_record_still_fails_closed() { @@ -25651,9 +25911,10 @@ pub(crate) fn fallback_free_capacity_dedup(disks: &[rustfs_madmin::Disk]) -> usi #[cfg(test)] mod pools_tests { - use std::collections::HashMap; + use std::collections::{HashMap, HashSet}; use super::DECOMMISSION_PROGRESS_SAVE_RETRY_BACKOFF; + use super::decommission_over_budget_target_pools; use super::persist_v3_pool_meta_for_test; use super::record_decommission_entry_error; use super::resolve_decommission_partial_listing_entry; @@ -32036,7 +32297,7 @@ mod pools_tests { reserve_decommission_target_pending(&mut meta, 0, 1, 10, mutation_id, now + Duration::seconds(1)) .expect("the target intent should be durable before its write"); let own_consumption = vec![initial[0], DecommissionPoolCapacityInfo::for_test(1, layout, 50, 60, 10)]; - ensure_decommission_capacity_reservations_available(&meta, &own_consumption, "restart") + ensure_decommission_capacity_reservations_available(&meta, &own_consumption, "restart", None) .expect("a persisted target intent must recognize its own write after restart"); resolve_decommission_target_pending(&mut meta, 0, 1, 10, mutation_id) .expect("the persisted target intent should resolve at commit"); @@ -32053,11 +32314,11 @@ mod pools_tests { now + Duration::seconds(1), ) .expect("the operation's own target consumption should be persisted"); - ensure_decommission_capacity_reservations_available(&meta, &own_consumption, "migration") + ensure_decommission_capacity_reservations_available(&meta, &own_consumption, "migration", None) .expect("the operation's own committed target bytes must not look like external capacity loss"); let dropped = vec![initial[0], DecommissionPoolCapacityInfo::for_test(1, layout, 39, 60, 21)]; - let err = ensure_decommission_capacity_reservations_available(&meta, &dropped, "migration") + let err = ensure_decommission_capacity_reservations_available(&meta, &dropped, "migration", None) .expect_err("runtime migration must stop after a sudden competing write consumes reserved capacity"); assert!( err.to_string() @@ -32066,6 +32327,367 @@ mod pools_tests { assert!(is_decommission_capacity_blocked_error(&err)); } + /// Leave inflight bytes on a source allocation that no mutation will + /// settle, as a lost marker would. + fn strand_decommission_test_marker(meta: &mut PoolMeta, source_pool_index: usize, target_pool_index: usize, bytes: usize) { + strand_decommission_test_marker_as(meta, source_pool_index, target_pool_index, bytes, uuid::Uuid::new_v4()); + } + + fn strand_decommission_test_marker_as( + meta: &mut PoolMeta, + source_pool_index: usize, + target_pool_index: usize, + bytes: usize, + mutation_id: uuid::Uuid, + ) { + let reservation = meta.pools[source_pool_index] + .decommission + .as_mut() + .and_then(|info| info.capacity_reservation.as_mut()) + .expect("the source reservation should be active"); + let target = reservation + .targets + .iter_mut() + .find(|target| target.pool_index == target_pool_index) + .expect("the source should allocate the target pool"); + target.temporary_mutations.push(DecommissionCapacityTemporaryMutation { + mutation_id, + physical_bytes: bytes, + }); + target.inflight_physical_bytes += bytes; + target.observed_physical_bytes += bytes; + reservation.inflight_target_physical_bytes += bytes; + reservation.observed_target_physical_bytes += bytes; + validate_decommission_capacity_reservation(Some(reservation)).expect("the stranded marker must stay loadable"); + } + + fn decommission_budget_test_meta(source_pool_indices: &[usize], capacity: &[DecommissionPoolCapacityInfo]) -> PoolMeta { + let now = OffsetDateTime::UNIX_EPOCH + Duration::minutes(2); + let mut meta = PoolMeta { + version: POOL_META_VERSION, + pools: capacity + .iter() + .map(|capacity| decommission_test_pool_status(capacity.pool_index, None)) + .collect(), + ..Default::default() + }; + for &source_pool_index in source_pool_indices { + meta.decommission(source_pool_index, capacity[source_pool_index].space) + .unwrap(); + } + reserve_decommission_start_target_capacity( + &mut meta, + source_pool_indices, + capacity, + uuid::Uuid::new_v4(), + 1, + now, + DECOMMISSION_CAPACITY_MODEL_VERSION, + ) + .expect("the target capacity should fit the reservations"); + meta + } + + #[test] + fn over_budget_target_pool_blocks_only_growth_into_that_pool() { + let now = OffsetDateTime::UNIX_EPOCH + Duration::minutes(3); + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + // A 60-byte peak: target 1 takes 40 bytes, target 2 the other 20. + let capacity = vec![ + DecommissionPoolCapacityInfo::for_test(0, layout, 0, 30, 30), + DecommissionPoolCapacityInfo::for_test(1, layout, 40, 40, 0), + DecommissionPoolCapacityInfo::for_test(2, layout, 100, 100, 0), + ]; + let mut meta = decommission_budget_test_meta(&[0], &capacity); + // Target 1's temporary-copy budget is 20 bytes. + strand_decommission_test_marker(&mut meta, 0, 1, 25); + assert_eq!(decommission_over_budget_target_pools(&meta), HashSet::from([1])); + let reservation = meta.pools[0] + .decommission + .as_ref() + .and_then(|info| info.capacity_reservation.as_ref()) + .expect("the source reservation should be active"); + assert_eq!( + decommission_reserved_target_choice(reservation, None, 1, &HashSet::new()), + Some(1), + "by remaining allocation alone, target 1 wins" + ); + assert_eq!( + decommission_reserved_target_choice(reservation, None, 1, &decommission_over_budget_target_pools(&meta)), + Some(2), + "selection skips the over-budget pool" + ); + + let err = ensure_decommission_capacity_reservations_available(&meta, &capacity, "mutation", Some(1)) + .expect_err("growth into the over-budget pool must fail"); + assert!( + err.to_string() + .contains("has 25 inflight physical bytes, exceeding its 20-byte temporary-copy reservation"), + "unexpected admission error: {err}" + ); + assert!(is_decommission_capacity_blocked_error(&err)); + ensure_decommission_capacity_reservations_available(&meta, &capacity, "mutation", Some(2)) + .expect("growth into another pool does not depend on the over-budget pool"); + ensure_decommission_capacity_reservations_available(&meta, &capacity, "migration", None) + .expect("finalize and the per-entry check must not pause the decommission for one pool"); + + let mut into_other = meta.clone(); + grow_decommission_target_reservation(&mut into_other, &capacity, 0, 2, 1, uuid::Uuid::new_v4(), now) + .expect("reservation growth into another pool proceeds"); + let mut into_over_budget = meta.clone(); + grow_decommission_target_reservation(&mut into_over_budget, &capacity, 0, 1, 1, uuid::Uuid::new_v4(), now) + .expect_err("reservation growth into the over-budget pool fails"); + assert_eq!( + into_over_budget.pools[0] + .decommission + .as_ref() + .and_then(|info| info.capacity_reservation.as_ref()), + meta.pools[0] + .decommission + .as_ref() + .and_then(|info| info.capacity_reservation.as_ref()), + "a rejected growth leaves the ledger unchanged" + ); + } + + #[test] + fn over_budget_target_pool_blocks_growth_for_every_source() { + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + // Two sources share target pool 2; each reserves a 60-byte peak there. + let capacity = vec![ + DecommissionPoolCapacityInfo::for_test(0, layout, 0, 30, 30), + DecommissionPoolCapacityInfo::for_test(1, layout, 0, 30, 30), + DecommissionPoolCapacityInfo::for_test(2, layout, 200, 200, 0), + ]; + let mut meta = decommission_budget_test_meta(&[0, 1], &capacity); + strand_decommission_test_marker(&mut meta, 1, 2, 35); + assert_eq!(decommission_over_budget_target_pools(&meta), HashSet::from([2])); + + // Source 0's own allocation is within budget, but the pool's free-space + // check also credits source 1's inflight bytes. + let err = ensure_decommission_capacity_reservations_available(&meta, &capacity, "mutation", Some(2)) + .expect_err("growth into a pool with any over-budget allocation must fail"); + assert!( + err.to_string().contains("source pool 1 target pool 2"), + "unexpected admission error: {err}" + ); + ensure_decommission_capacity_reservations_available(&meta, &capacity, "migration", None) + .expect("the per-entry check of either source must not pause"); + } + + /// Commit one mutation on a target as a successful capacity mutation does, + /// after the admission check for that target. + fn commit_decommission_budget_test_mutation( + meta: &mut PoolMeta, + capacity: &[DecommissionPoolCapacityInfo], + target_pool_index: usize, + bytes: usize, + ) { + let now = OffsetDateTime::UNIX_EPOCH + Duration::minutes(3); + let mutation_id = uuid::Uuid::new_v4(); + ensure_decommission_capacity_reservations_available(meta, capacity, "mutation", Some(target_pool_index)) + .expect("the mutation should be admitted into its target"); + reserve_decommission_target_pending(meta, 0, target_pool_index, bytes, mutation_id, now) + .expect("reserve the mutation's intent"); + resolve_decommission_target_pending(meta, 0, target_pool_index, bytes, mutation_id).expect("resolve the intent"); + record_decommission_target_consumption( + meta, + 0, + target_pool_index, + DecommissionTargetConsumption { + committed_data_bytes: bytes, + target_physical_bytes: bytes, + observed_physical_bytes: bytes, + }, + mutation_id, + now, + ) + .expect("charge the committed mutation"); + ensure_decommission_capacity_reservations_available(meta, capacity, "mutation", None) + .expect("finalize must not pause the decommission"); + } + + fn decommission_budget_test_remaining(meta: &PoolMeta, target_pool_index: usize) -> usize { + let reservation = meta.pools[0] + .decommission + .as_ref() + .and_then(|info| info.capacity_reservation.as_ref()) + .expect("the source reservation should be active"); + reservation + .targets + .iter() + .find(|target| target.pool_index == target_pool_index) + .map(|target| target.remaining_reserved_physical_bytes(reservation.temporary_copies)) + .expect("the source should allocate the target pool") + } + + #[test] + fn stranded_marker_overtaken_by_later_commits_stops_only_its_pool() { + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + // A 200-byte peak with one temporary copy: target 1 takes 180 bytes, + // target 2 the other 20. + let capacity = vec![ + DecommissionPoolCapacityInfo::for_test(0, layout, 0, 100, 100), + DecommissionPoolCapacityInfo::for_test(1, layout, 180, 180, 0), + DecommissionPoolCapacityInfo::for_test(2, layout, 1_000, 1_000, 0), + ]; + let mut meta = decommission_budget_test_meta(&[0], &capacity); + // The completion a resumed routine adopted on target 1, whose retry + // has not committed yet. + strand_decommission_test_marker(&mut meta, 0, 1, 10); + + // Two other mutations both land on target 1, which still has the + // larger remaining allocation each time. + assert!(decommission_budget_test_remaining(&meta, 1) > decommission_budget_test_remaining(&meta, 2)); + commit_decommission_budget_test_mutation(&mut meta, &capacity, 1, 75); + assert!(decommission_budget_test_remaining(&meta, 1) > decommission_budget_test_remaining(&meta, 2)); + commit_decommission_budget_test_mutation(&mut meta, &capacity, 1, 6); + + // Target 1's budget is now 9 bytes, below the 10 adopted bytes. + assert_eq!(decommission_budget_test_remaining(&meta, 1), 18); + assert_eq!(decommission_over_budget_target_pools(&meta), HashSet::from([1])); + ensure_decommission_capacity_reservations_available(&meta, &capacity, "migration", None) + .expect("the per-entry check must not pause the decommission"); + ensure_decommission_capacity_reservations_available(&meta, &capacity, "mutation", Some(2)) + .expect("admission into target 2 proceeds"); + let err = ensure_decommission_capacity_reservations_available(&meta, &capacity, "mutation", Some(1)) + .expect_err("target 1 no longer accepts growth"); + assert!( + err.to_string() + .contains("has 10 inflight physical bytes, exceeding its 9-byte temporary-copy reservation"), + "unexpected admission error: {err}" + ); + } + + #[test] + fn adopted_marker_left_by_a_commit_on_another_target_stops_only_its_pool() { + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + let capacity = vec![ + DecommissionPoolCapacityInfo::for_test(1, layout, 400, 400, 0), + DecommissionPoolCapacityInfo::for_test(2, layout, 400, 400, 0), + ]; + let interrupted = uuid::Uuid::new_v4(); + let mut meta = decommission_test_cleanup_meta( + DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION, + Vec::new(), + 0, + Some((interrupted, 10)), + ); + // Target 1 keeps a quarter of the 200-byte peak, target 2 the rest. + { + let reservation = meta.pools[0] + .decommission + .as_mut() + .and_then(|info| info.capacity_reservation.as_mut()) + .expect("the test reservation should be present"); + let mut second = reservation.targets[0].clone(); + reservation.targets[0].reserved_physical_bytes = reservation.peak_physical_bytes / 4; + second.pool_index = 2; + second.reserved_physical_bytes = reservation.peak_physical_bytes - reservation.peak_physical_bytes / 4; + second.observed_physical_bytes = 0; + second.inflight_physical_bytes = 0; + second.pending_physical_bytes = 0; + second.pending_mutation_id = None; + reservation.targets.push(second); + } + meta.pools.push(decommission_test_pool_status(2, None)); + adopt_intent_adoption_test_intents(&mut meta); + + // The existing-copy lookup routes the retry to target 2, where a newer + // version of the key lives, so it commits there without selection and + // leaves its adopted completion on target 1. + commit_intent_adoption_test_mutation(&mut meta, 2, interrupted, 10); + let target_1 = intent_adoption_test_reservation(&meta).targets[0].clone(); + assert_eq!(target_1.pool_index, 1); + assert!(decommission_target_has_mutation_marker(&target_1, interrupted)); + + // Another commit on target 1 shrinks its budget to 9 bytes, below the + // 10 adopted bytes. + commit_intent_adoption_test_mutation(&mut meta, 1, uuid::Uuid::new_v4(), 16); + assert_eq!(decommission_over_budget_target_pools(&meta), HashSet::from([1])); + ensure_decommission_capacity_reservations_available(&meta, &capacity, "migration", None) + .expect("the per-entry check must not pause the decommission"); + ensure_decommission_capacity_reservations_available(&meta, &capacity, "mutation", Some(2)) + .expect("admission into target 2 proceeds"); + ensure_decommission_capacity_reservations_available(&meta, &capacity, "mutation", Some(1)) + .expect_err("target 1 no longer accepts growth"); + } + + #[test] + fn sticky_retry_bypasses_a_pool_another_source_pushed_over_budget() { + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + // Source 0 reserves 50 bytes on target 2 and 10 on target 3; source 1 + // reserves its 60 bytes on target 3. + let capacity = vec![ + DecommissionPoolCapacityInfo::for_test(0, layout, 0, 30, 30), + DecommissionPoolCapacityInfo::for_test(1, layout, 0, 30, 30), + DecommissionPoolCapacityInfo::for_test(2, layout, 50, 50, 0), + DecommissionPoolCapacityInfo::for_test(3, layout, 200, 200, 0), + ]; + let mut meta = decommission_budget_test_meta(&[0, 1], &capacity); + // Source 0's interrupted mutation left a 4-byte adopted completion on + // target 3, within source 0's own 5-byte budget there. + let interrupted = uuid::Uuid::new_v4(); + strand_decommission_test_marker_as(&mut meta, 0, 3, 4, decommission_adopted_intent_mutation_id(interrupted)); + let source_0 = meta.pools[0] + .decommission + .as_ref() + .and_then(|info| info.capacity_reservation.as_ref()) + .expect("source 0 should hold a reservation") + .clone(); + assert_eq!( + decommission_reserved_target_choice(&source_0, Some(interrupted), 2, &decommission_over_budget_target_pools(&meta)), + Some(3), + "the retry returns to its adopted target while no allocation there is over budget" + ); + + // Source 1 goes over its budget on the shared target 3. + strand_decommission_test_marker(&mut meta, 1, 3, 35); + assert_eq!(decommission_over_budget_target_pools(&meta), HashSet::from([3])); + assert!( + source_0 + .targets + .iter() + .all(|target| !target.exceeds_temporary_copy_budget(source_0.temporary_copies)), + "source 0's own allocations stay within budget" + ); + // Growth into target 3 now fails for every source, so the retry is + // placed on target 2 and leaves the adopted marker behind. + assert_eq!( + decommission_reserved_target_choice(&source_0, Some(interrupted), 2, &decommission_over_budget_target_pools(&meta)), + Some(2) + ); + } + + #[test] + fn free_space_shortfall_still_pauses_every_mutation() { + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + let capacity = vec![ + DecommissionPoolCapacityInfo::for_test(0, layout, 0, 30, 30), + DecommissionPoolCapacityInfo::for_test(1, layout, 40, 40, 0), + DecommissionPoolCapacityInfo::for_test(2, layout, 100, 100, 0), + ]; + let mut meta = decommission_budget_test_meta(&[0], &capacity); + // Within target 2's 10-byte budget, so only the free-space check can + // fail here. + strand_decommission_test_marker(&mut meta, 0, 2, 8); + assert!(decommission_over_budget_target_pools(&meta).is_empty()); + // A competing write left target 2 short of its remaining reservation; + // that still pauses every mutation, exactly as before. + let shrunk = vec![ + capacity[0], + capacity[1], + DecommissionPoolCapacityInfo::for_test(2, layout, 4, 100, 96), + ]; + let err = ensure_decommission_capacity_reservations_available(&meta, &shrunk, "migration", None) + .expect_err("a pool without room for its reservation must still pause the decommission"); + assert!( + err.to_string() + .contains("target pool 2 reservation requires 20 physical bytes"), + "unexpected free-space error: {err}" + ); + } + #[test] fn decommission_target_capacity_reuses_only_matching_pending_intent() { let now = OffsetDateTime::UNIX_EPOCH + Duration::minutes(2); @@ -33234,8 +33856,7 @@ mod pools_tests { fn intent_adoption_test_temporary_budget(meta: &PoolMeta) -> usize { let reservation = intent_adoption_test_reservation(meta); - reservation.targets[0].remaining_reserved_physical_bytes(reservation.temporary_copies) - / (1 + reservation.temporary_copies) + reservation.targets[0].temporary_copy_budget(reservation.temporary_copies) } /// A mutation's staged and adopted marker bytes together. @@ -33340,18 +33961,23 @@ mod pools_tests { let reservation = intent_adoption_test_reservation(&meta); assert_eq!( - decommission_reserved_target_choice(reservation, None, 10), + decommission_reserved_target_choice(reservation, None, 10, &HashSet::new()), Some(2), "without an adopted completion the larger allocation wins" ); assert_eq!( - decommission_reserved_target_choice(reservation, Some(interrupted), 10), + decommission_reserved_target_choice(reservation, Some(interrupted), 10, &HashSet::new()), Some(1), "the retry must go back to the target that holds its adopted completion" ); // The rule cannot help once that target no longer fits the object. assert_eq!( - decommission_reserved_target_choice(reservation, Some(interrupted), reservation.peak_physical_bytes / 4), + decommission_reserved_target_choice( + reservation, + Some(interrupted), + reservation.peak_physical_bytes / 4, + &HashSet::new() + ), Some(2) ); @@ -33392,7 +34018,7 @@ mod pools_tests { // budget to 40 bytes, which the completion counted once still fits. commit_intent_adoption_test_mutation(&mut meta, 1, uuid::Uuid::new_v4(), 60); assert_eq!(intent_adoption_test_temporary_budget(&meta), 40); - ensure_decommission_capacity_reservations_available(&meta, &capacity_infos, "migration") + ensure_decommission_capacity_reservations_available(&meta, &capacity_infos, "mutation", Some(1)) .expect("later admission must not trip over a completion counted twice"); commit_intent_adoption_test_mutation(&mut meta, 1, mutation_id, 30); @@ -33497,7 +34123,7 @@ mod pools_tests { staged, Some((completion, staged)), ); - ensure_decommission_capacity_reservations_available(&meta, &capacity_infos, "migration") + ensure_decommission_capacity_reservations_available(&meta, &capacity_infos, "mutation", Some(1)) .expect("the interrupted completion's ledger fits its budget"); let plan = decommission_capacity_intent_adoption_plan(&meta, 0, generation) @@ -33512,7 +34138,7 @@ mod pools_tests { Some(budget), "the overlapping staged parts and completion are counted up to the budget" ); - ensure_decommission_capacity_reservations_available(&meta, &capacity_infos, "migration") + ensure_decommission_capacity_reservations_available(&meta, &capacity_infos, "mutation", Some(1)) .expect("adoption must not push inflight bytes past the temporary-copy budget"); } @@ -33565,7 +34191,7 @@ mod pools_tests { other_staged, Some((completion, completion_bytes)), ); - ensure_decommission_capacity_reservations_available(&meta, &capacity_infos, "migration") + ensure_decommission_capacity_reservations_available(&meta, &capacity_infos, "mutation", Some(1)) .expect("the interrupted ledger fits its budget"); let mut finalized = meta.clone(); @@ -33593,8 +34219,8 @@ mod pools_tests { .expect("adopt the interrupted completion"); assert_eq!( - ensure_decommission_capacity_reservations_available(&adopted, &capacity_infos, "migration").is_ok(), - ensure_decommission_capacity_reservations_available(&finalized, &capacity_infos, "migration").is_ok(), + ensure_decommission_capacity_reservations_available(&adopted, &capacity_infos, "mutation", Some(1)).is_ok(), + ensure_decommission_capacity_reservations_available(&finalized, &capacity_infos, "mutation", Some(1)).is_ok(), "adoption must reach the same budget verdict as the lost finalize (other staged {other_staged}, completion {completion_bytes}, budget {budget})" ); } diff --git a/crates/ecstore/src/core/pools_test.rs b/crates/ecstore/src/core/pools_test.rs index 9fa6758fc..59e468b45 100644 --- a/crates/ecstore/src/core/pools_test.rs +++ b/crates/ecstore/src/core/pools_test.rs @@ -224,7 +224,7 @@ mod decommission_lock_order_tests { use std::future::Future; use std::sync::{ Arc, Condvar, Mutex as StdMutex, - atomic::{AtomicUsize, Ordering}, + atomic::{AtomicBool, AtomicUsize, Ordering}, }; use std::time::Duration; use tokio::io::AsyncReadExt; @@ -507,6 +507,333 @@ mod decommission_lock_order_tests { assert_lock_order_with_tail(mutation, false).await; } + /// Leave inflight bytes on the active decommission's allocation for + /// `target_pool_index` that no mutation will settle, as a lost marker + /// would, and persist them. + async fn strand_decommission_target_marker(store: &Arc, target_pool_index: usize, bytes: usize) { + let mut meta = store.pool_meta.read().await.clone(); + let reservation = meta.pools[0] + .decommission + .as_mut() + .and_then(|info| info.capacity_reservation.as_mut()) + .expect("the decommission reservation should be active"); + let target = reservation + .targets + .iter_mut() + .find(|target| target.pool_index == target_pool_index) + .expect("the reservation should allocate the target pool"); + target + .temporary_mutations + .push(crate::core::pools::DecommissionCapacityTemporaryMutation { + mutation_id: uuid::Uuid::new_v4(), + physical_bytes: bytes, + }); + target.inflight_physical_bytes += bytes; + target.observed_physical_bytes += bytes; + reservation.inflight_target_physical_bytes += bytes; + reservation.observed_target_physical_bytes += bytes; + meta.save(store.pools.clone()) + .await + .expect("persist the stranded target marker"); + *store.pool_meta.write().await = meta; + } + + /// Start a decommission of `bucket` on a three-pool store where target 1 + /// holds most of the allocation and `stranded` inflight bytes of a marker + /// no mutation will settle. + async fn start_over_budget_routine_decommission(store: &Arc, bucket: &str, stranded: usize) { + // Of the 128 MiB peak, target 1 takes 100 MiB and target 2 the other + // 28 MiB; target 1's temporary-copy budget is 50 MiB. + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + set_decommission_capacity_info_overrides_for_test( + store.id, + vec![vec![ + DecommissionPoolCapacityInfo::for_test(0, layout, 0, 64 << 20, 64 << 20), + DecommissionPoolCapacityInfo::for_test(1, layout, 100 << 20, 100 << 20, 0), + DecommissionPoolCapacityInfo::for_test(2, layout, 1 << 30, 1 << 30, 0), + ]], + ); + store + .save_current_pool_meta_for_decommission_start( + &[0], + vec![crate::core::pools::DecomBucketInfo { + name: bucket.to_string(), + prefix: String::new(), + }], + ) + .await + .expect("activate the decommission"); + strand_decommission_target_marker(store, 1, stranded).await; + } + + /// Pause one of `object`'s own target-1 gate acquisitions, after `skips` + /// of them, by arming the pause when the routine reads that object. + fn pause_object_target_gate_acquire( + barrier: Arc, + bucket: String, + object: String, + skips: usize, + ) -> DecommissionTestFaultGuard { + let armed = Arc::new(AtomicBool::new(false)); + DecommissionTestFaultGuard::install(Arc::new(move |stage, entry_bucket, entry_object, _, _| { + if stage == "object_read" && entry_bucket == bucket && entry_object == object && !armed.swap(true, Ordering::AcqRel) { + barrier.pause_one_target_gate_acquire_after(1, skips); + } + false + })) + } + + /// Another mutation commits `bytes` on target 1, as a concurrent entry + /// would. + async fn commit_other_mutation_on_target_1(store: &Arc, bytes: usize) { + let owner = decommission_capacity_owner(&*store.pool_meta.read().await).with_mutation_id(uuid::Uuid::new_v4()); + store + .run_decommission_capacity_admitted_mutation(1, Some(owner), Some(bytes), || async { + Ok::<_, crate::error::Error>(()) + }) + .await + .expect("the other mutation commits on target 1"); + } + + async fn assert_target_1_over_budget(store: &Arc) { + let target = killed_leader_target(store, 1).await; + let budget = target + .reserved_physical_bytes + .saturating_sub(target.consumed_physical_bytes * 2) + / 2; + assert!( + target.inflight_physical_bytes > budget, + "target 1 must be over its {budget}-byte budget: {target:?}" + ); + } + + async fn run_routine_over_a_target_going_over_budget( + store: &Arc, + barrier: &DecommissionCapacityLockOrderBarrier, + other_commit_bytes: usize, + before_other_commit: impl std::future::Future, + ) { + let routine = tokio::spawn({ + let store = Arc::clone(store); + async move { + let canceler = crate::core::pools::DecommissionCanceler::new(CancellationToken::new()); + store.decommission_cancelers.write().await[0] = Some(canceler.clone()); + store + .do_decommission_in_routine(canceler, 0, Arc::new(tokio::sync::Semaphore::new(2))) + .await + } + }); + // The object selected target 1 and waits to acquire its gate. + barrier.wait_until_target_gate_acquire_paused().await; + before_other_commit.await; + commit_other_mutation_on_target_1(store, other_commit_bytes).await; + assert_target_1_over_budget(store).await; + barrier.release_target_gate_acquire(); + routine + .await + .expect("the decommission routine task should join") + .expect("the decommission routine should return"); + let local = store.pool_meta.read().await; + let info = local.pools[0] + .decommission + .as_ref() + .expect("the decommission state should remain present"); + assert!( + info.complete && !info.failed && !info.canceled && info.capacity_blocked_reason.is_none(), + "a target that went over budget after selection must not pause the decommission: complete={}, failed={}, canceled={}, blocked_reason={:?}", + info.complete, + info.failed, + info.canceled, + info.capacity_blocked_reason + ); + assert_eq!(info.items_decommission_failed, 0); + } + + #[test] + #[serial_test::serial] + fn decommission_reselects_a_target_that_went_over_budget_after_selection() { + run_large_stack_async_test("decommission-reselect-over-budget-target", || async { + let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await; + let bucket = test_bucket("reselect-over-budget"); + let object = "late.bin"; + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create the reselection bucket"); + let mut source = PutObjReader::from_vec(b"selected target 1 before it went over budget".to_vec()); + store.pools[0] + .put_object(&bucket, object, &mut source, &ObjectOptions::default()) + .await + .expect("write the source object"); + // 45 of target 1's 50 MiB budget are stranded. + start_over_budget_routine_decommission(&store, &bucket, 45 << 20).await; + let barrier = Arc::new(DecommissionCapacityLockOrderBarrier::install(store.id, store.id)); + barrier.disable_owner_pause(); + let _fault = pause_object_target_gate_acquire(Arc::clone(&barrier), bucket.clone(), object.to_string(), 0); + + // A 6 MiB commit shrinks target 1's budget to 44 MiB. + run_routine_over_a_target_going_over_budget(&store, &barrier, 6 << 20, async {}).await; + store.pools[2] + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await + .expect("the entry must be placed on target 2 after target 1 went over budget"); + assert!( + store.pools[1] + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await + .is_err() + ); + }); + } + + #[test] + #[serial_test::serial] + fn decommission_restarts_a_multipart_upload_pinned_to_a_target_that_went_over_budget() { + run_large_stack_async_test("decommission-reselect-over-budget-multipart", || async { + let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await; + let bucket = test_bucket("reselect-over-budget-mpu"); + let object = "late-multipart.bin"; + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create the multipart reselection bucket"); + let source_upload = new_multipart_upload(&store, 0, &bucket, object, ObjectOptions::default()) + .await + .expect("create the source multipart upload"); + let source_opts = multipart_options(&store, &bucket, ObjectOptions::default()).await; + let mut source_parts = Vec::with_capacity(2); + for (part_num, part_body) in [(1, vec![0x31; 5 << 20]), (2, vec![0x32; 1])] { + let part = store.pools[0] + .put_object_part( + &bucket, + object, + &source_upload.upload_id, + part_num, + &mut PutObjReader::from_vec(part_body), + &source_opts, + ) + .await + .expect("stage the source multipart part"); + source_parts.push(crate::storage_api_contracts::multipart::CompletePart { + part_num: part.part_num, + etag: part.etag, + ..Default::default() + }); + } + let source_info = store.pools[0] + .clone() + .complete_multipart_upload(&bucket, object, &source_upload.upload_id, source_parts, &source_opts) + .await + .expect("complete the source multipart object"); + assert!(source_info.is_multipart()); + // 40 of target 1's 50 MiB budget are stranded. + start_over_budget_routine_decommission(&store, &bucket, 40 << 20).await; + let barrier = Arc::new(DecommissionCapacityLockOrderBarrier::install(store.id, store.id)); + barrier.disable_owner_pause(); + // Skip the upload creation and part 1; pause before part 2. + let _fault = pause_object_target_gate_acquire(Arc::clone(&barrier), bucket.clone(), object.to_string(), 2); + + // A 12 MiB commit shrinks target 1's budget to 38 MiB while the + // upload, pinned to target 1, waits to upload part 2. + let store_for_check = Arc::clone(&store); + run_routine_over_a_target_going_over_budget(&store, &barrier, 12 << 20, async move { + let target = killed_leader_target(&store_for_check, 1).await; + assert!( + target.temporary_mutations.len() >= 2, + "the upload must hold its staging record on target 1 next to the stranded marker: {:?}", + target.temporary_mutations + ); + }) + .await; + let migrated = store.pools[2] + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await + .expect("the restarted upload must complete on target 2"); + assert!(migrated.is_multipart()); + assert!( + store.pools[1] + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await + .is_err() + ); + }); + } + + #[test] + #[serial_test::serial] + fn decommission_completes_on_other_pools_while_one_pool_exceeds_its_temporary_budget() { + run_large_stack_async_test("decommission-over-budget-target-pool", || async { + let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await; + let bucket = test_bucket("over-budget-target-pool"); + let object = "over-budget-target-pool.bin"; + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create the over-budget bucket"); + let mut source = PutObjReader::from_vec(b"migrates around the over-budget pool".to_vec()); + store.pools[0] + .put_object(&bucket, object, &mut source, &ObjectOptions::default()) + .await + .expect("write the source object before decommission starts"); + // Of the 128 MiB peak, target 1 takes 100 MiB and target 2 the + // other 28 MiB; both stay far above the fixture and the + // `.rustfs.sys` metadata the routine also migrates. + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + set_decommission_capacity_info_overrides_for_test( + store.id, + vec![vec![ + DecommissionPoolCapacityInfo::for_test(0, layout, 0, 64 << 20, 64 << 20), + DecommissionPoolCapacityInfo::for_test(1, layout, 100 << 20, 100 << 20, 0), + DecommissionPoolCapacityInfo::for_test(2, layout, 1 << 30, 1 << 30, 0), + ]], + ); + store + .save_current_pool_meta_for_decommission_start( + &[0], + vec![crate::core::pools::DecomBucketInfo { + name: bucket.clone(), + prefix: String::new(), + }], + ) + .await + .expect("activate the decommission"); + // Target 1's temporary-copy budget is 50 MiB. + strand_decommission_target_marker(&store, 1, 60 << 20).await; + + let canceler = crate::core::pools::DecommissionCanceler::new(CancellationToken::new()); + store.decommission_cancelers.write().await[0] = Some(canceler.clone()); + store + .do_decommission_in_routine(canceler, 0, Arc::new(tokio::sync::Semaphore::new(1))) + .await + .expect("the decommission routine should return"); + let local = store.pool_meta.read().await; + let info = local.pools[0] + .decommission + .as_ref() + .expect("the decommission state should remain present"); + assert!( + info.complete && !info.failed && !info.canceled, + "one over-budget pool must not pause the decommission: complete={}, failed={}, canceled={}, blocked_reason={:?}", + info.complete, + info.failed, + info.canceled, + info.capacity_blocked_reason + ); + drop(local); + store.pools[2] + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await + .expect("the object must migrate to the pool within its budget"); + assert!( + store.pools[1] + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await + .is_err(), + "target selection must skip the over-budget pool" + ); + }); + } + #[tokio::test] #[serial_test::serial] async fn no_active_external_commit_probe_blocks_decommission_activation_until_commit_owns_it() {