diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 8803e075f..e61ab96ed 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -1040,8 +1040,12 @@ fn is_decommission_capacity_intent_conflict(err: &Error) -> bool { return message.contains("unresolved target capacity intent") || message.contains("pending capacity intent belongs to another mutation"); } - data_movement::data_movement_stage_source(err).is_some_and(is_decommission_capacity_intent_conflict) - || err.to_string().contains("unresolved target capacity intent") + if data_movement::data_movement_stage_source(err).is_some_and(is_decommission_capacity_intent_conflict) { + return true; + } + let rendered = err.to_string(); + rendered.contains("unresolved target capacity intent") + || rendered.contains("pending capacity intent belongs to another mutation") } pub(crate) fn decommission_capacity_target_gate_busy_index(err: &Error) -> Option { @@ -2248,6 +2252,316 @@ fn resolve_decommission_target_pending( Ok(()) } +/// A pending target intent that an earlier incarnation of a decommission +/// routine left in its still-active capacity reservation. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct DecommissionCapacityLeftoverIntent { + target_pool_index: usize, + mutation_id: uuid::Uuid, + physical_bytes: usize, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct DecommissionCapacityIntentAdoptionPlan { + model_version: u16, + operation_id: uuid::Uuid, + generation: u64, + intents: Vec, +} + +/// Plan which pending target intents a resumed decommission routine adopts. +/// +/// The caller must be a routine for `source_pool_index` that has not started +/// any worker yet, so no intent in the reservation belongs to a mutation of +/// this incarnation. Outside its target write window, a pending intent is only +/// a crash marker: the previous incarnation was killed before finalizing the +/// mutation, or a writer failed after persisting it and has not retried yet. +/// Adoption holds the target fence while it applies, so no write is in flight, +/// and it keeps the intent's mutation identity and bytes as an inflight +/// marker, so the mutation's later retry settles it exactly. The persisted +/// decommission start time and the reservation identity pin the plan to the +/// same decommission. Intents without an object identity are left in place +/// because nothing can settle them. +fn decommission_capacity_intent_adoption_plan( + meta: &PoolMeta, + source_pool_index: usize, + decommission_generation: OffsetDateTime, +) -> Result> { + ensure_decommission_generation(meta, source_pool_index, decommission_generation)?; + let Some(reservation) = meta + .pools + .get(source_pool_index) + .and_then(|pool| pool.decommission.as_ref()) + .and_then(|info| info.capacity_reservation.as_ref()) + .filter(|reservation| reservation.active()) + else { + return Ok(None); + }; + if reservation.source_pool_index != source_pool_index { + return Err(decommission_capacity_blocked_error( + "decommission intent adoption found a mismatched capacity source", + )); + } + if !matches!( + reservation.model_version, + DECOMMISSION_CAPACITY_LEGACY_MODEL_VERSION | DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION + ) { + return Err(Error::DecommissionCapacity(format!( + "decommission intent adoption found unsupported capacity lock model {}", + reservation.model_version + ))); + } + let mut intents = reservation + .targets + .iter() + .filter(|target| target.pending_physical_bytes > 0) + .filter_map(|target| { + let mutation_id = target.pending_mutation_id.filter(|mutation_id| !mutation_id.is_nil())?; + Some(DecommissionCapacityLeftoverIntent { + target_pool_index: target.pool_index, + mutation_id, + physical_bytes: target.pending_physical_bytes, + }) + }) + .collect::>(); + if intents.is_empty() { + return Ok(None); + } + intents.sort_unstable_by_key(|intent| intent.target_pool_index); + Ok(Some(DecommissionCapacityIntentAdoptionPlan { + model_version: reservation.model_version, + operation_id: reservation.operation_id, + generation: reservation.generation, + intents, + })) +} + +/// Id of the inflight marker that holds a mutation's adopted completion. +/// +/// The marker is a separate entry, derived deterministically from the +/// mutation id, so that a cleanup releasing the mutation's staged bytes cannot +/// drop it. Only the mutation's commit, its equivalent-target reconciliation, +/// or a cleanup that proved the exact target absent settles it. A node without +/// 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. +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"); + hasher.update(mutation_id.as_bytes()); + let digest = hasher.finalize(); + let mut bytes = [0; 16]; + bytes.copy_from_slice(&digest[..16]); + uuid::Uuid::from_bytes(bytes) +} + +/// Move each planned leftover intent out of its target's exclusive pending +/// slot into the mutation's adopted-completion marker. +/// +/// The marker keeps the mutation identity and the bytes that mutation may +/// have placed on the target. Its own retry settles it through the existing +/// paths (a committed write or an equivalent target records consumption and +/// releases it; a cleanup that proves the target absent releases it), while +/// other mutations can use the target again instead of waiting on a slot that +/// no write holds. +fn adopt_decommission_capacity_leftover_intents( + meta: &mut PoolMeta, + source_pool_index: usize, + decommission_generation: OffsetDateTime, + plan: &DecommissionCapacityIntentAdoptionPlan, + now: OffsetDateTime, +) -> Result<()> { + if decommission_capacity_intent_adoption_plan(meta, source_pool_index, decommission_generation)?.as_ref() != Some(plan) { + return Err(decommission_capacity_blocked_error( + "decommission capacity owner or leftover target intents changed before adoption", + )); + } + for intent in &plan.intents { + let split = decommission_adopted_intent_split(meta, source_pool_index, intent)?; + resolve_decommission_target_pending( + meta, + source_pool_index, + intent.target_pool_index, + intent.physical_bytes, + intent.mutation_id, + )?; + let pool_count = meta.pools.len(); + let pool = meta + .pools + .get_mut(source_pool_index) + .ok_or_else(|| invalid_decommission_pool_index_error(pool_count, source_pool_index))?; + let reservation = pool + .decommission + .as_mut() + .and_then(|info| info.capacity_reservation.as_mut()) + .filter(|reservation| reservation.active()) + .ok_or_else(|| decommission_capacity_blocked_error("active reservation disappeared during intent adoption"))?; + let model_version = reservation.model_version; + let target = reservation + .targets + .iter_mut() + .find(|target| target.pool_index == intent.target_pool_index) + .ok_or_else(|| decommission_capacity_blocked_error("target allocation disappeared during intent adoption"))?; + // A target-fence ledger keeps a zero-byte staged marker as the + // multipart discovery record; the legacy ledger cannot store one. + set_decommission_target_marker_bytes( + target, + intent.mutation_id, + split.staged, + model_version == DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION, + ); + set_decommission_target_marker_bytes( + target, + decommission_adopted_intent_mutation_id(intent.mutation_id), + split.adopted, + false, + ); + target.inflight_physical_bytes = target.inflight_physical_bytes.saturating_add(split.growth); + target.observed_physical_bytes = target.observed_physical_bytes.saturating_add(split.growth); + reservation.inflight_target_physical_bytes = reservation.inflight_target_physical_bytes.saturating_add(split.growth); + reservation.observed_target_physical_bytes = reservation.observed_target_physical_bytes.saturating_add(split.growth); + reservation.prediction_error_bytes = + signed_capacity_difference(reservation.observed_target_physical_bytes, reservation.consumed_target_physical_bytes); + renew_decommission_capacity_reservation(reservation, now, true); + pool.last_update = pool.last_update.max(now); + } + Ok(()) +} + +/// Replace a marker's bytes, dropping the entry at zero unless a zero-byte +/// entry that already exists should stay as a discovery record. +fn set_decommission_target_marker_bytes( + target: &mut DecommissionCapacityTarget, + mutation_id: uuid::Uuid, + physical_bytes: usize, + keep_existing_zero_byte_entry: bool, +) { + match target + .temporary_mutations + .iter() + .position(|mutation| mutation.mutation_id == mutation_id) + { + Some(index) if physical_bytes == 0 && !keep_existing_zero_byte_entry => { + target.temporary_mutations.remove(index); + } + Some(index) => target.temporary_mutations[index].physical_bytes = physical_bytes, + None if physical_bytes > 0 => target.temporary_mutations.push(DecommissionCapacityTemporaryMutation { + mutation_id, + physical_bytes, + }), + None => {} + } +} + +/// How an adopted intent splits between the mutation's staged marker and its +/// adopted-completion marker. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct DecommissionAdoptedIntentSplit { + staged: usize, + adopted: usize, + growth: usize, +} + +/// A multipart data movement stages parts (the staged marker) and then +/// reserves its completion for the whole object (the intent). The completion +/// moves the staged parts into the object, so the two overlap: committed or +/// not, the target holds about the larger of them. An intent left by a part +/// upload instead adds to the parts staged before it, and that part was +/// admitted only while all staged bytes plus the part fit the target's +/// temporary-copy budget. The mutation's total therefore counts the sum while +/// it fits that budget and otherwise stops at the budget, never below the +/// larger of the two; it then exceeds the budget only when finalizing the +/// completion would have, because finalize charges the same bytes as +/// consumption, which shrinks the budget by that amount. +/// +/// A mutation killed twice leaves an adopted completion and then a new +/// intent for the same object version. Both reserve that same completion, so +/// they merge by their maximum before the staged parts are added. +/// +/// The adopted marker takes at least the completion's bytes, so it never +/// reaches zero and a cleanup that frees staged bytes cannot drop it. The +/// staged marker keeps the rest of the total and never grows. +fn decommission_adopted_intent_split( + meta: &PoolMeta, + source_pool_index: usize, + intent: &DecommissionCapacityLeftoverIntent, +) -> Result { + let reservation = meta + .pools + .get(source_pool_index) + .and_then(|pool| pool.decommission.as_ref()) + .and_then(|info| info.capacity_reservation.as_ref()) + .filter(|reservation| reservation.active()) + .ok_or_else(|| decommission_capacity_blocked_error("active reservation disappeared during intent adoption"))?; + let target = reservation + .targets + .iter() + .find(|target| target.pool_index == intent.target_pool_index) + .ok_or_else(|| decommission_capacity_blocked_error("target allocation disappeared during intent adoption"))?; + let marker_bytes = |mutation_id: uuid::Uuid| { + target + .temporary_mutations + .iter() + .find(|mutation| mutation.mutation_id == mutation_id) + .map(|mutation| mutation.physical_bytes) + .unwrap_or_default() + }; + let staged = marker_bytes(intent.mutation_id); + 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 completion = adopted.max(intent.physical_bytes); + let total = staged + .saturating_add(completion) + .min(temporary_budget.saturating_sub(other_inflight)) + .max(recorded.max(completion)); + let staged_after = total.saturating_sub(completion).min(staged); + Ok(DecommissionAdoptedIntentSplit { + staged: staged_after, + adopted: total.saturating_sub(staged_after), + growth: total.saturating_sub(recorded), + }) +} + +#[cfg(test)] +static DECOMMISSION_INTENT_ADOPTION_TEST_FAILURES: std::sync::OnceLock>> = + std::sync::OnceLock::new(); + +/// Make the next intent adoption of `store_id` fail with an ordinary error. +#[cfg(all(test, feature = "test-util"))] +pub(crate) fn fail_next_decommission_intent_adoption_for_test(store_id: uuid::Uuid) { + DECOMMISSION_INTENT_ADOPTION_TEST_FAILURES + .get_or_init(Default::default) + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(store_id); +} + +#[cfg(test)] +fn take_decommission_intent_adoption_test_failure(store_id: uuid::Uuid) -> bool { + DECOMMISSION_INTENT_ADOPTION_TEST_FAILURES + .get_or_init(Default::default) + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(&store_id) +} + +/// A routine that cannot settle its predecessor's intents, for example while +/// a lock quorum is still forming after a restart, pauses without a terminal +/// transition so the next incarnation retries the adoption. +fn decommission_intent_adoption_failure(err: Error) -> Error { + if is_err_operation_canceled(&err) || is_decommission_capacity_blocked_error(&err) { + return err; + } + decommission_capacity_blocked_error(format!( + "failed to adopt target capacity intents left by a previous decommission incarnation: {err}" + )) +} + fn release_decommission_target_temporary_mutation( target: &mut DecommissionCapacityTarget, mutation_id: uuid::Uuid, @@ -2270,6 +2584,135 @@ fn release_decommission_target_temporary_mutation( (released, released > 0 || removed) } +/// Release both of a mutation's markers once its commit is settled: the bytes +/// it staged and the completion a resumed routine adopted for it. +fn release_decommission_target_mutation_markers( + target: &mut DecommissionCapacityTarget, + mutation_id: uuid::Uuid, +) -> (usize, bool) { + let (staged, staged_changed) = release_decommission_target_temporary_mutation(target, mutation_id, usize::MAX); + // An adopted marker is never zero bytes, so releasing it always changes + // the ledger. + let (adopted, _) = + release_decommission_target_temporary_mutation(target, decommission_adopted_intent_mutation_id(mutation_id), usize::MAX); + (staged.saturating_add(adopted), staged_changed || adopted > 0) +} + +/// The target's remaining allocation if it still fits one object of +/// `expected_data_bytes` with its temporary copies. +fn decommission_capacity_target_candidate( + reservation: &DecommissionCapacityReservation, + target: &DecommissionCapacityTarget, + expected_data_bytes: usize, +) -> Option<(usize, usize)> { + 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); + (required_peak <= remaining).then_some((target.pool_index, remaining)) +} + +/// Choose the reserved target for a mutation's next attempt. +/// +/// A retry goes back to a target that holds the mutation's adopted +/// completion while that target still fits the object: only a commit or an +/// equivalent target there settles the adopted marker, so committing on +/// 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. +fn decommission_reserved_target_choice( + reservation: &DecommissionCapacityReservation, + mutation_id: Option, + expected_data_bytes: usize, +) -> Option { + let candidate = + |target: &DecommissionCapacityTarget| decommission_capacity_target_candidate(reservation, target, expected_data_bytes); + let adopted_mutation_id = mutation_id.map(decommission_adopted_intent_mutation_id); + reservation + .targets + .iter() + .filter(|target| { + target + .temporary_mutations + .iter() + .any(|mutation| Some(mutation.mutation_id) == adopted_mutation_id) + }) + .find_map(candidate) + .or_else(|| { + reservation + .targets + .iter() + .filter_map(candidate) + .max_by_key(|(_, remaining)| *remaining) + }) + .map(|(pool_index, _)| pool_index) +} + +/// Record whether a mutation with an adopted completion went back to the +/// target that holds it, or had to be placed elsewhere and so leaves the +/// adopted marker behind. +fn log_decommission_adopted_target_choice( + reservation: &DecommissionCapacityReservation, + owner: DecommissionCapacityOwner, + chosen_pool_index: usize, +) { + let Some(mutation_id) = owner.mutation_id else { + return; + }; + let adopted_mutation_id = decommission_adopted_intent_mutation_id(mutation_id); + let adopted_target_pools = reservation + .targets + .iter() + .filter(|target| { + target + .temporary_mutations + .iter() + .any(|mutation| mutation.mutation_id == adopted_mutation_id) + }) + .map(|target| target.pool_index) + .collect::>(); + if adopted_target_pools.is_empty() { + return; + } + if adopted_target_pools.contains(&chosen_pool_index) { + info!( + event = EVENT_DECOMMISSION_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + state = "adopted_target_reused", + pool_index = owner.source_pool_index, + target_pool_index = chosen_pool_index, + mutation_id = %mutation_id, + "Decommission retry returns to the target that holds its adopted completion" + ); + } else { + warn!( + event = EVENT_DECOMMISSION_ENTRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + state = "adopted_target_bypassed", + pool_index = owner.source_pool_index, + target_pool_index = chosen_pool_index, + adopted_target_pools = ?adopted_target_pools, + mutation_id = %mutation_id, + "Decommission retry no longer fits the target that holds its adopted completion" + ); + } +} + +/// Whether a target still holds a marker of this mutation, staged or adopted. +fn decommission_target_has_mutation_marker(target: &DecommissionCapacityTarget, mutation_id: uuid::Uuid) -> bool { + let adopted = decommission_adopted_intent_mutation_id(mutation_id); + target + .temporary_mutations + .iter() + .any(|mutation| mutation.mutation_id == mutation_id || mutation.mutation_id == adopted) +} + fn settle_decommission_target_non_growing_replacement( meta: &mut PoolMeta, source_pool_index: usize, @@ -2296,7 +2739,7 @@ fn settle_decommission_target_non_growing_replacement( .iter_mut() .find(|target| target.pool_index == target_pool_index) .ok_or_else(|| decommission_capacity_blocked_error("target allocation disappeared during target replacement"))?; - let (released, changed) = release_decommission_target_temporary_mutation(target, mutation_id, usize::MAX); + let (released, changed) = release_decommission_target_mutation_markers(target, mutation_id); reservation.inflight_target_physical_bytes = reservation.inflight_target_physical_bytes.saturating_sub(released); if changed { renew_decommission_capacity_reservation(reservation, now, true); @@ -2352,7 +2795,7 @@ fn record_decommission_target_consumption( target.consumed_physical_bytes = target.consumed_physical_bytes.saturating_add(consumed); target.observed_physical_bytes = target.observed_physical_bytes.saturating_add(observed_physical_bytes); let has_scoped_temporary_mutations = !target.temporary_mutations.is_empty(); - let (released_inflight, _) = release_decommission_target_temporary_mutation(target, mutation_id, usize::MAX); + let (released_inflight, _) = release_decommission_target_mutation_markers(target, mutation_id); let released_inflight = if released_inflight == 0 && !has_scoped_temporary_mutations { let released = target.inflight_physical_bytes.min(consumed); target.inflight_physical_bytes = target.inflight_physical_bytes.saturating_sub(released); @@ -2512,6 +2955,19 @@ fn release_decommission_target_inflight( .ok_or_else(|| decommission_capacity_blocked_error("target allocation disappeared during target cleanup"))?; let has_temporary_mutations = !target.temporary_mutations.is_empty(); let pending_belongs_to_mutation = target.pending_mutation_id == Some(mutation_id); + // An adopted completion is the mutation's published intent moved out of + // the pending slot; it keeps that intent's delayed-release exemption while + // every marker on the target is this mutation's. A staged marker with + // bytes is released below, so the exemption never applies to it. + let adopted_mutation_id = decommission_adopted_intent_mutation_id(mutation_id); + let holds_only_own_adopted_completion = target + .temporary_mutations + .iter() + .any(|mutation| mutation.mutation_id == adopted_mutation_id) + && target + .temporary_mutations + .iter() + .all(|mutation| mutation.mutation_id == adopted_mutation_id || mutation.mutation_id == mutation_id); let zero_ledger_cleanup = confirmed_absent && model_version == DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION && !has_temporary_mutations @@ -2533,13 +2989,21 @@ fn release_decommission_target_inflight( } else { released }; + // Only a cleanup that proved the exact target absent shows that an adopted + // completion never committed. Any other cleanup leaves it for the commit or + // the equivalent-target reconciliation to charge. + let released_adopted = if confirmed_absent && clear_pending { + release_decommission_target_temporary_mutation(target, adopted_mutation_id, usize::MAX).0 + } else { + 0 + }; let pending_matches = clear_pending && confirmed_absent && pending_belongs_to_mutation; let published_pending_delayed_release = confirmed_absent && model_version == DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION && !clear_pending - && !has_temporary_mutations - && pending_belongs_to_mutation; + && ((!has_temporary_mutations && pending_belongs_to_mutation) || holds_only_own_adopted_completion); if released == 0 + && released_adopted == 0 && released_physical_bytes > 0 && !zero_ledger_cleanup && !published_pending_delayed_release @@ -2549,7 +3013,9 @@ fn release_decommission_target_inflight( "temporary target cleanup released bytes that cannot be attributed to its mutation", )); } - reservation.inflight_target_physical_bytes = reservation.inflight_target_physical_bytes.saturating_sub(released); + reservation.inflight_target_physical_bytes = reservation + .inflight_target_physical_bytes + .saturating_sub(released.saturating_add(released_adopted)); let cleared_pending = if pending_matches { let cleared = target.pending_physical_bytes; target.pending_physical_bytes = 0; @@ -2559,7 +3025,7 @@ fn release_decommission_target_inflight( } else { 0 }; - let changed = released > 0 || temporary_mutation_changed || cleared_pending > 0; + let changed = released > 0 || released_adopted > 0 || temporary_mutation_changed || cleared_pending > 0; if changed { renew_decommission_capacity_reservation(reservation, now, true); pool.last_update = pool.last_update.max(now); @@ -2895,8 +3361,16 @@ fn is_decommission_set_local_usage_cache(bucket: &str, object: &str) -> bool { name.strip_suffix(".bkp").unwrap_or(name) == DATA_USAGE_CACHE_NAME } -fn with_decommission_entry_context(stage: &str, bucket: &str, object: &str, err: E) -> Error { - Error::other(format!("decommission entry {stage} failed for bucket {bucket} object {object}: {err}")) +/// Keep the typed cause reachable so the entry retry loop and the routine +/// classify a wrapped capacity block or intent conflict by type. +fn with_decommission_entry_context(stage: &str, bucket: &str, object: &str, err: E) -> Error +where + E: std::error::Error + Send + Sync + 'static, +{ + data_movement::data_movement_context_error( + format!("decommission entry {stage} failed for bucket {bucket} object {object}: {err}"), + err, + ) } #[cfg(test)] @@ -2940,19 +3414,34 @@ async fn load_decommission_entry_exact_versions( fn resolve_decommission_check_after_list_result(list_result: Result<()>, entry_error: Option) -> Result<()> { match list_result { Ok(()) => entry_error.map_or(Ok(()), Err), - Err(list_err) => resolve_decommission_listing_error(Some(list_err), entry_error).map_or(Ok(()), Err), + Err(list_err) => decommission_root_cause(Some(list_err), entry_error).map_or(Ok(()), Err), } } -fn resolve_decommission_listing_error(listing_error: Option, entry_error: Option) -> Option { - match (listing_error, entry_error) { - (Some(listing_error), Some(entry_error)) if is_err_operation_canceled(&listing_error) => Some(entry_error), - (Some(listing_error), Some(entry_error)) if is_err_operation_canceled(&entry_error) => Some(listing_error), - (Some(listing_error), _) => Some(listing_error), - (None, entry_error) => entry_error, +/// The first failure of concurrent decommission work cancels its siblings, +/// which then report `OperationCanceled`, possibly before the failing task +/// returns. Keep the failure that caused the cancellation so the routine +/// classifies the real cause: a capacity block pauses instead of ending the +/// decommission as failed. Between two real failures the earlier one stays. +fn decommission_root_cause(first: Option, next: Option) -> Option { + match (first, next) { + (Some(first), Some(next)) if is_err_operation_canceled(&first) && !is_err_operation_canceled(&next) => Some(next), + (Some(first), _) => Some(first), + (None, next) => next, } } +/// The failure a bucket reports once its listing workers and entries +/// finished. An entry failure cancels the listing workers, so a canceled +/// listing must not hide the entry error that caused it. +fn decommission_bucket_failure(listing_results: Vec>, entry_error: Option) -> Option { + let listing_error = listing_results + .into_iter() + .filter_map(Result::err) + .fold(None, |first, err| decommission_root_cause(first, Some(err))); + decommission_root_cause(listing_error, entry_error) +} + fn decommission_unresolved_listing_error(entry: &DecommissionUnresolvedEntry) -> Error { Error::other(format!( "decommission listing could not resolve metadata for {bucket}/{object} on pool {pool_index} set {set_index} ({candidate_count} candidate(s), {disk_error_count} disk error(s))", @@ -4760,9 +5249,7 @@ where while let Some(result) = active.next().await { if let Err(err) = result { rx.cancel(); - if first_err.is_none() { - first_err = Some(err); - } + first_err = decommission_root_cause(first_err, Some(err)); continue; } @@ -11626,7 +12113,16 @@ impl ECStore { .acquire_pool_meta_write_guard(&mut save_guard, "decommission equivalent target reconciliation failed") .await?; let source_pool_index = owner.source_pool_index; - let (target_layout, target_pending_physical_bytes, target_consumed_physical_bytes) = { + let mutation_id = owner.mutation_id.ok_or_else(|| { + decommission_capacity_blocked_error("equivalent target reconciliation mutation identity is missing") + })?; + let ( + target_layout, + target_pending_physical_bytes, + target_pending_mutation_id, + target_consumed_physical_bytes, + has_temporary_mutation, + ) = { let reservation = snapshot .pools .get(source_pool_index) @@ -11645,13 +12141,38 @@ impl ECStore { .iter() .find(|target| target.pool_index == target_pool_index) .ok_or_else(|| decommission_capacity_blocked_error("equivalent target reconciliation allocation is missing"))?; - (target.layout, target.pending_physical_bytes, target.consumed_physical_bytes) + ( + target.layout, + target.pending_physical_bytes, + target.pending_mutation_id, + target.consumed_physical_bytes, + decommission_target_has_mutation_marker(target, mutation_id), + ) }; - let mutation_id = owner.mutation_id.ok_or_else(|| { - decommission_capacity_blocked_error("equivalent target reconciliation mutation identity is missing") - })?; let expected_target_physical_bytes = capacity_target_physical_bytes(expected_data_bytes.max(1), target_layout)?; - if target_pending_physical_bytes == 0 { + // A target holds one pending intent at a time, so a pending intent of + // another mutation means this mutation has none here; that intent is + // not this mutation's to check or resolve. Without its own pending + // intent, this mutation's inflight marker is the only unsettled record + // of it: either bytes it staged or an intent a previous routine + // incarnation left and the resumed routine adopted. The equivalent + // committed target proves the commit, so settle that marker into + // consumption exactly as the lost finalize would have. With neither, + // the commit was already charged. + let owns_pending = target_pending_physical_bytes > 0 && target_pending_mutation_id == Some(mutation_id); + let settles_marker_only = has_temporary_mutation && !owns_pending; + if !owns_pending && !has_temporary_mutation { + debug!( + event = EVENT_DECOMMISSION_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + state = "equivalent_target_already_charged", + pool_index = source_pool_index, + target_pool_index, + other_pending_intent = target_pending_physical_bytes > 0, + consumption_covers_object = target_consumed_physical_bytes >= expected_target_physical_bytes, + "Decommission equivalent target has no unsettled capacity record" + ); if target_consumed_physical_bytes >= expected_target_physical_bytes { let mut pool_meta = self.pool_meta.write().await; ensure_pool_meta_write_fence(&pool_meta_guard, "equivalent target idempotent publication")?; @@ -11669,19 +12190,20 @@ impl ECStore { "equivalent target has no pending capacity intent to reconcile", )); } - if target_pending_physical_bytes < expected_target_physical_bytes { - return Err(decommission_capacity_blocked_error( - "equivalent target pending capacity is smaller than the committed object", - )); + if !settles_marker_only { + if target_pending_physical_bytes < expected_target_physical_bytes { + return Err(decommission_capacity_blocked_error( + "equivalent target pending capacity is smaller than the committed object", + )); + } + resolve_decommission_target_pending( + &mut snapshot, + source_pool_index, + target_pool_index, + expected_target_physical_bytes, + mutation_id, + )?; } - - resolve_decommission_target_pending( - &mut snapshot, - source_pool_index, - target_pool_index, - expected_target_physical_bytes, - mutation_id, - )?; record_decommission_target_consumption( &mut snapshot, source_pool_index, @@ -11834,13 +12356,13 @@ impl ECStore { target.layout, target.pending_physical_bytes, target.pending_mutation_id, - target - .temporary_mutations - .iter() - .any(|mutation| mutation.mutation_id == mutation_id), + decommission_target_has_mutation_marker(target, mutation_id), ) }; - if pending_physical_bytes == 0 && !has_temporary_mutation { + // A target holds one pending intent at a time, so another mutation's + // pending intent means this mutation has none here to settle. + let owns_pending = pending_physical_bytes > 0 && pending_mutation_id == Some(mutation_id); + if !owns_pending && !has_temporary_mutation { // A prior successful attempt already settled both durable halves. ensure_pool_meta_write_fence(&pool_meta_guard, "equivalent temporary target reconciliation fence failed")?; if let Some(target_guard) = target_guard.as_ref() { @@ -11852,18 +12374,13 @@ impl ECStore { } return Ok(()); } - if pending_physical_bytes > 0 && pending_mutation_id != Some(mutation_id) { - return Err(decommission_capacity_blocked_error( - "equivalent temporary target pending intent belongs to another mutation", - )); - } let expected_target_physical_bytes = capacity_target_physical_bytes(expected_data_bytes.max(1), target_layout)?; - if pending_physical_bytes > 0 && pending_physical_bytes != expected_target_physical_bytes { + if owns_pending && pending_physical_bytes != expected_target_physical_bytes { return Err(decommission_capacity_blocked_error( "equivalent temporary target pending capacity does not match the committed checkpoint", )); } - if pending_physical_bytes > 0 { + if owns_pending { resolve_decommission_target_pending( &mut snapshot, source_pool_index, @@ -11945,10 +12462,7 @@ impl ECStore { true } else { (target.pending_physical_bytes > 0 && target.pending_mutation_id == Some(mutation_id)) - || target - .temporary_mutations - .iter() - .any(|mutation| mutation.mutation_id == mutation_id) + || decommission_target_has_mutation_marker(target, mutation_id) } }) } @@ -13174,6 +13688,108 @@ impl ECStore { ensure_decommission_capacity_reservations_available(&pool_meta, &capacity_infos, "migration") } + /// Adopt the pending target intents that an earlier incarnation of this + /// decommission routine left behind. Call it before this incarnation + /// starts any worker; `decommission_capacity_intent_adoption_plan` explains + /// why adopting them cannot take over a write in flight. + /// + /// Lock order: start gate, target gates, save gate, distributed pool + /// metadata fence, then pool_meta, matching terminal transitions. Holding + /// each target gate, or for the legacy model the pool metadata write fence + /// that legacy mutations keep through target I/O, proves that no mutation + /// is inside its target write window while the plan is applied. + async fn adopt_previous_incarnation_decommission_capacity_intents( + &self, + idx: usize, + generation: OffsetDateTime, + ) -> Result<()> { + let local_plan = { + let pool_meta = self.pool_meta.read().await; + decommission_capacity_intent_adoption_plan(&pool_meta, idx, generation)? + }; + if local_plan.is_none() { + return Ok(()); + } + #[cfg(test)] + if take_decommission_intent_adoption_test_failure(self.id) { + return Err(Error::other("injected decommission intent adoption failure")); + } + let _start_guard = self.start_gate.lock().await; + let plan = { + let save_guard = self.pool_meta_save_gate.lock().await; + let (_read_guard, snapshot) = self + .acquire_pool_meta_read_guard(&save_guard, "decommission intent adoption planning failed") + .await?; + decommission_capacity_intent_adoption_plan(&snapshot, idx, generation)? + }; + let Some(plan) = plan else { + return Ok(()); + }; + let mut target_guards = Vec::with_capacity(plan.intents.len()); + if plan.model_version == DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION { + for intent in &plan.intents { + let guard = self + .acquire_decommission_capacity_target_guard_retried(intent.target_pool_index) + .await?; + ensure_decommission_capacity_target_fence(&guard, intent.target_pool_index, "leftover intent adoption")?; + target_guards.push((intent.target_pool_index, guard)); + } + } + let mut save_guard = self.pool_meta_save_gate.lock().await; + let (pool_meta_guard, mut snapshot) = self + .acquire_pool_meta_write_guard(&mut save_guard, "decommission intent adoption failed") + .await?; + adopt_decommission_capacity_leftover_intents(&mut snapshot, idx, generation, &plan, OffsetDateTime::now_utc())?; + let outcome = snapshot + .save_no_lock_armed( + self.pools.clone(), + &mut save_guard, + PoolMetaPersistenceFence::Owned(Arc::clone(&pool_meta_guard)), + &[idx], + ) + .await?; + // Log as soon as the adopted ledger is durable: a fence lost after this + // point leaves nothing to adopt next time. A save that ends as + // TransactionUnknown can still apply the adoption without this event, + // so the rollback rule must not depend on it. + warn!( + event = EVENT_DECOMMISSION_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_POOLS, + state = "capacity_intents_adopted", + pool_index = idx, + operation_id = %plan.operation_id, + generation = plan.generation, + adopted_intents = plan.intents.len(), + adopted_physical_bytes = plan + .intents + .iter() + .fold(0usize, |total, intent| total.saturating_add(intent.physical_bytes)), + "Decommission adopted target capacity intents left by a previous incarnation" + ); + ensure_pool_meta_write_fence(&pool_meta_guard, "decommission intent adoption save failed")?; + for (target_pool_index, guard) in &target_guards { + ensure_decommission_capacity_target_fence(guard, *target_pool_index, "leftover intent adoption save")?; + } + { + let mut pool_meta = self.pool_meta.write().await; + publish_decommission_capacity_update(&mut pool_meta, &outcome.committed, idx)?; + } + ensure_pool_meta_write_fence(&pool_meta_guard, "decommission intent adoption save failed")?; + for (target_pool_index, guard) in &target_guards { + ensure_decommission_capacity_target_fence(guard, *target_pool_index, "leftover intent adoption publication")?; + } + outcome.disarm(); + Ok(()) + } + + #[cfg(feature = "test-util")] + pub async fn adopt_previous_incarnation_decommission_capacity_intents_for_test(&self, idx: usize) -> Result<()> { + let generation = self.active_decommission_generation(idx).await?; + self.adopt_previous_incarnation_decommission_capacity_intents(idx, generation) + .await + } + async fn decommission_capacity_owner_for_worker( &self, idx: usize, @@ -13235,18 +13851,12 @@ 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"))?; - let candidate = |target: &DecommissionCapacityTarget| { - 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); - (required_peak <= remaining).then_some((target.pool_index, remaining)) - }; 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(candidate) + .and_then(|target| decommission_capacity_target_candidate(reservation, target, expected_data_bytes)) { return Ok(pool_index); } @@ -13255,12 +13865,8 @@ 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, _)) = reservation - .targets - .iter() - .filter_map(candidate) - .max_by_key(|(_, remaining)| *remaining) - { + 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); return Ok(pool_index); } if reservation.source_data_equivalent_bytes > 0 { @@ -14907,11 +15513,7 @@ impl ECStore { .ok_or_else(|| decommission_capacity_blocked_error("scanner pause backlog retirement reservation is stale"))?; let mutation_id = decommission_capacity_version_mutation_id(owner, RUSTFS_META_BUCKET, version); if reservation.targets.iter().any(|target| { - target.pending_mutation_id == Some(mutation_id) - || target - .temporary_mutations - .iter() - .any(|mutation| mutation.mutation_id == mutation_id) + target.pending_mutation_id == Some(mutation_id) || decommission_target_has_mutation_marker(target, mutation_id) }) { return Err(decommission_capacity_blocked_error( "scanner pause backlog retirement has an unresolved target capacity intent", @@ -16527,21 +17129,16 @@ impl ECStore { "Decommission waiting for workers" ); - let mut listing_worker_error = None; + let mut listing_results = Vec::with_capacity(listing_workers.len()); for (set_id, worker) in listing_workers { - if let Err(err) = resolve_decommission_listing_worker_result(set_id, worker.await) { + let result = resolve_decommission_listing_worker_result(set_id, worker.await); + if result.is_err() { rx.cancel(); - if listing_worker_error.is_none() { - listing_worker_error = Some(err); - } } + listing_results.push(result); } - if let Some(err) = listing_worker_error { - return Err(err); - } - - if let Some(err) = entry_error.lock().await.clone() { + if let Some(err) = decommission_bucket_failure(listing_results, entry_error.lock().await.clone()) { return Err(err); } @@ -16657,9 +17254,18 @@ impl ECStore { } return Ok(()); } - let result = self - .decommission_in_background(rx.clone(), idx, generation, entry_budget) - .await; + // Settle the previous incarnation's unfinished target intents before + // any worker of this incarnation can create an intent of its own. + let result = match self + .adopt_previous_incarnation_decommission_capacity_intents(idx, generation) + .await + { + Ok(()) => { + self.decommission_in_background(rx.clone(), idx, generation, entry_budget) + .await + } + Err(err) => Err(decommission_intent_adoption_failure(err)), + }; if let Err(err) = &result && (is_decommission_capacity_blocked_error(err) || is_decommission_target_capacity_error(err)) @@ -21567,6 +22173,387 @@ mod tests { ); } + async fn start_leftover_intent_test_decommission( + store: &Arc, + decom_buckets: Vec, + expected_model_version: u16, + ) { + // The routine also migrates `.rustfs.sys` bucket metadata, so keep the + // reservation far above every fixture 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, 64 << 20, 64 << 20), + DecommissionPoolCapacityInfo::for_test(1, layout, 1 << 30, 1 << 30, 0), + ]], + ); + store + .save_current_pool_meta_for_decommission_start(&[0], decom_buckets) + .await + .expect("the decommission reservation should be activated"); + let pool_meta = store.pool_meta.read().await; + let reservation = pool_meta.pools[0] + .decommission + .as_ref() + .and_then(|info| info.capacity_reservation.as_ref()) + .expect("the decommission reservation should be active"); + assert_eq!(reservation.model_version, expected_model_version); + } + + async fn leftover_intent_test_owner(store: &Arc, mutation_id: uuid::Uuid) -> DecommissionCapacityOwner { + let pool_meta = store.pool_meta.read().await; + let reservation = pool_meta.pools[0] + .decommission + .as_ref() + .and_then(|info| info.capacity_reservation.as_ref()) + .expect("the decommission reservation should be active"); + DecommissionCapacityOwner { + source_pool_index: 0, + operation_id: reservation.operation_id, + generation: reservation.generation, + owner_nonce: reservation.owner_nonce, + mutation_id: Some(mutation_id), + } + } + + async fn leftover_intent_test_target(store: &Arc, target_pool_index: usize) -> DecommissionCapacityTarget { + store.pool_meta.read().await.pools[0] + .decommission + .as_ref() + .and_then(|info| info.capacity_reservation.as_ref()) + .and_then(|reservation| { + reservation + .targets + .iter() + .find(|target| target.pool_index == target_pool_index) + }) + .cloned() + .expect("the reservation should allocate the target pool") + } + + /// Leave the durable trace of a target mutation whose leader died after it + /// persisted the capacity intent but before it finalized the mutation. + async fn leave_unfinalized_decommission_target_intent( + store: &Arc, + target_pool_index: usize, + mutation_id: uuid::Uuid, + expected_data_bytes: usize, + ) { + let owner = leftover_intent_test_owner(store, mutation_id).await; + store + .run_decommission_capacity_admitted_mutation_with_capacity_lease( + target_pool_index, + Some(owner), + Some(expected_data_bytes), + |_| async { Err::<(), _>(Error::other("leader killed before finalizing the target mutation")) }, + ) + .await + .expect_err("the interrupted target mutation must not report success"); + let target = leftover_intent_test_target(store, target_pool_index).await; + assert_eq!( + target.pending_mutation_id, + Some(mutation_id), + "the interrupted mutation must leave its durable capacity intent" + ); + let mut persisted = PoolMeta::default(); + persisted + .load_no_lock_from_replicas(store.pools.clone()) + .await + .expect("the interrupted intent should be durable"); + assert_eq!( + persisted.pools[0] + .decommission + .as_ref() + .and_then(|info| info.capacity_reservation.as_ref()) + .and_then(|reservation| reservation + .targets + .iter() + .find(|target| target.pool_index == target_pool_index)) + .and_then(|target| target.pending_mutation_id), + Some(mutation_id) + ); + } + + async fn decommission_intent_adoption_never_takes_a_live_mutation_intent_case(legacy: bool) { + let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await; + let _proof_guard = legacy.then(crate::services::notification_sys::without_decommission_target_fence_fleet_proof_for_test); + start_leftover_intent_test_decommission( + &store, + Vec::new(), + if legacy { + DECOMMISSION_CAPACITY_LEGACY_MODEL_VERSION + } else { + DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION + }, + ) + .await; + let generation = store + .active_decommission_generation(0) + .await + .expect("the decommission generation should be active"); + + let live = uuid::Uuid::new_v4(); + let live_owner = leftover_intent_test_owner(&store, live).await; + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let live_store = Arc::clone(&store); + let live_mutation = tokio::spawn(async move { + live_store + .run_decommission_capacity_admitted_mutation_with_capacity_lease(1, Some(live_owner), Some(16), |_| async { + entered_tx.send(()).expect("the live mutation entry should be observed"); + release_rx.await.expect("the live mutation should be released"); + Ok(()) + }) + .await + }); + tokio::time::timeout(std::time::Duration::from_secs(30), entered_rx) + .await + .expect("the live mutation should enter its target write") + .expect("the live mutation should report entry"); + assert_eq!(leftover_intent_test_target(&store, 1).await.pending_mutation_id, Some(live)); + + if legacy { + // A legacy mutation keeps the pool metadata write fence through its + // target I/O, so adoption cannot even plan until it finalizes. + let adoption_store = Arc::clone(&store); + let mut adoption = tokio::spawn(async move { + adoption_store + .adopt_previous_incarnation_decommission_capacity_intents(0, generation) + .await + }); + tokio::time::timeout(std::time::Duration::from_millis(500), &mut adoption) + .await + .expect_err("adoption must wait while the live legacy mutation holds the metadata fence"); + let target = leftover_intent_test_target(&store, 1).await; + assert_eq!(target.pending_mutation_id, Some(live), "the live intent must stay with its mutation"); + assert!(target.temporary_mutations.is_empty()); + release_tx.send(()).expect("release the live mutation"); + live_mutation + .await + .expect("the live mutation task should not panic") + .expect("the live mutation should finalize its own intent"); + tokio::time::timeout(std::time::Duration::from_secs(30), adoption) + .await + .expect("adoption should proceed once the live mutation finalizes") + .expect("the adoption task should not panic") + .expect("a finalized live mutation leaves nothing to adopt"); + } else { + let err = store + .adopt_previous_incarnation_decommission_capacity_intents(0, generation) + .await + .expect_err("adoption must not take an intent whose mutation still holds the target fence"); + assert!(is_decommission_capacity_target_gate_busy(&err), "unexpected adoption error: {err}"); + let target = leftover_intent_test_target(&store, 1).await; + assert_eq!(target.pending_mutation_id, Some(live), "the live intent must stay with its mutation"); + assert!(target.temporary_mutations.is_empty()); + release_tx.send(()).expect("release the live mutation"); + live_mutation + .await + .expect("the live mutation task should not panic") + .expect("the live mutation should finalize its own intent"); + } + let target = leftover_intent_test_target(&store, 1).await; + assert_eq!(target.pending_mutation_id, None); + assert!(target.temporary_mutations.is_empty(), "a live intent must never become an adopted marker"); + assert_eq!(target.consumed_physical_bytes, 16); + store + .adopt_previous_incarnation_decommission_capacity_intents(0, generation) + .await + .expect("a finalized live mutation leaves nothing to adopt"); + assert_eq!(leftover_intent_test_target(&store, 1).await, target); + } + + #[tokio::test] + #[serial_test::serial] + async fn decommission_intent_adoption_never_takes_a_live_mutation_intent() { + decommission_intent_adoption_never_takes_a_live_mutation_intent_case(false).await; + } + + #[tokio::test] + #[serial_test::serial] + async fn legacy_decommission_intent_adoption_never_takes_a_live_mutation_intent() { + decommission_intent_adoption_never_takes_a_live_mutation_intent_case(true).await; + } + + #[tokio::test] + #[serial_test::serial] + async fn adopted_marker_settles_while_another_mutation_holds_the_pending_slot() { + let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await; + start_leftover_intent_test_decommission(&store, Vec::new(), DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION).await; + let generation = store + .active_decommission_generation(0) + .await + .expect("the decommission generation should be active"); + let adopted = uuid::Uuid::new_v4(); + leave_unfinalized_decommission_target_intent(&store, 1, adopted, 16).await; + store + .adopt_previous_incarnation_decommission_capacity_intents(0, generation) + .await + .expect("the resumed routine should adopt the leftover intent"); + // A mutation of the resumed incarnation failed after persisting its + // own intent and has not retried yet. + let waiting = uuid::Uuid::new_v4(); + leave_unfinalized_decommission_target_intent(&store, 1, waiting, 8).await; + + let owner = leftover_intent_test_owner(&store, adopted).await; + store + .reconcile_decommission_capacity_after_equivalent_target(owner, 1, 16) + .await + .expect("another mutation's pending slot must not block settling an adopted marker"); + let target = leftover_intent_test_target(&store, 1).await; + assert_eq!(target.consumed_physical_bytes, 16); + assert!( + !decommission_target_has_mutation_marker(&target, adopted), + "the adopted marker should be settled: {:?}", + target.temporary_mutations + ); + assert_eq!(target.pending_mutation_id, Some(waiting), "the other mutation keeps its pending intent"); + assert_eq!(target.pending_physical_bytes, 8); + } + + #[tokio::test] + #[serial_test::serial] + async fn settled_equivalent_target_does_not_wait_for_another_mutations_pending_slot() { + let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await; + start_leftover_intent_test_decommission(&store, Vec::new(), DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION).await; + // An earlier attempt committed and charged this copy, for example a + // delete marker whose source cleanup never ran. + let settled = uuid::Uuid::new_v4(); + let settled_owner = leftover_intent_test_owner(&store, settled).await; + store + .run_decommission_capacity_admitted_mutation_with_capacity_lease(1, Some(settled_owner), Some(1), |_| async { + Ok(()) + }) + .await + .expect("charge the settled copy"); + // Another mutation holds the slot between a failed write and its retry. + let waiting = uuid::Uuid::new_v4(); + leave_unfinalized_decommission_target_intent(&store, 1, waiting, 8).await; + let before = leftover_intent_test_target(&store, 1).await; + + store + .reconcile_decommission_capacity_after_equivalent_target(settled_owner, 1, 0) + .await + .expect("an equivalent copy that is already charged has nothing to wait for"); + store + .reconcile_decommission_capacity_after_equivalent_temporary_target(settled_owner, 1, 1) + .await + .expect("an equivalent checkpoint that is already settled has nothing to wait for"); + assert_eq!( + leftover_intent_test_target(&store, 1).await, + before, + "neither reconciliation may resolve the other mutation's intent or charge the copy again" + ); + } + + #[tokio::test] + #[serial_test::serial] + async fn equivalent_target_settles_an_adopted_intent_into_consumption() { + let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await; + start_leftover_intent_test_decommission(&store, Vec::new(), DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION).await; + let generation = store + .active_decommission_generation(0) + .await + .expect("the decommission generation should be active"); + let leftover = uuid::Uuid::new_v4(); + leave_unfinalized_decommission_target_intent(&store, 1, leftover, 16).await; + store + .adopt_previous_incarnation_decommission_capacity_intents(0, generation) + .await + .expect("the resumed routine should adopt the leftover intent"); + let adopted = leftover_intent_test_target(&store, 1).await; + assert_eq!(adopted.pending_mutation_id, None); + assert_eq!( + adopted.consumed_physical_bytes, 0, + "adoption must not assume the interrupted write committed" + ); + assert_eq!(adopted.inflight_physical_bytes, 16); + + // The interrupted mutation's retry finds its committed copy on the + // target and settles the adopted marker exactly once. + let owner = leftover_intent_test_owner(&store, leftover).await; + store + .reconcile_decommission_capacity_after_equivalent_target(owner, 1, 16) + .await + .expect("an equivalent committed target settles the adopted marker"); + let settled = leftover_intent_test_target(&store, 1).await; + assert!(settled.temporary_mutations.is_empty()); + assert_eq!(settled.inflight_physical_bytes, 0); + assert_eq!(settled.consumed_physical_bytes, 16); + store + .reconcile_decommission_capacity_after_equivalent_target(owner, 1, 16) + .await + .expect("a repeated equivalent reconciliation stays idempotent"); + assert_eq!(leftover_intent_test_target(&store, 1).await.consumed_physical_bytes, 16); + } + + #[tokio::test] + #[serial_test::serial] + async fn equivalent_temporary_target_settles_an_adopted_checkpoint_intent() { + let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await; + start_leftover_intent_test_decommission(&store, Vec::new(), DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION).await; + let generation = store + .active_decommission_generation(0) + .await + .expect("the decommission generation should be active"); + // A checkpoint replacement persisted its intent and failed before + // finalizing; the resumed routine adopts that intent. + let checkpoint = uuid::Uuid::new_v4(); + leave_unfinalized_decommission_target_intent(&store, 1, checkpoint, 16).await; + store + .adopt_previous_incarnation_decommission_capacity_intents(0, generation) + .await + .expect("the resumed routine should adopt the checkpoint intent"); + let adopted = leftover_intent_test_target(&store, 1).await; + assert_eq!(adopted.pending_mutation_id, None); + assert!(decommission_target_has_mutation_marker(&adopted, checkpoint)); + + // Multipart discovery and the checkpoint gate see the adopted marker + // as the mutation's temporary state. + let owner = leftover_intent_test_owner(&store, checkpoint).await; + assert!(store.has_decommission_capacity_temporary_mutation_state(1, owner).await); + + store + .reconcile_decommission_capacity_after_equivalent_temporary_target(owner, 1, 16) + .await + .expect("the committed checkpoint settles the adopted marker"); + let settled = leftover_intent_test_target(&store, 1).await; + assert!(!decommission_target_has_mutation_marker(&settled, checkpoint)); + assert_eq!(settled.inflight_physical_bytes, 0); + assert_eq!(settled.consumed_physical_bytes, 0, "a non-growing replacement charges nothing"); + assert!(!store.has_decommission_capacity_temporary_mutation_state(1, owner).await); + store + .reconcile_decommission_capacity_after_equivalent_temporary_target(owner, 1, 16) + .await + .expect("a repeated temporary reconciliation stays idempotent"); + } + + #[tokio::test] + #[serial_test::serial] + async fn uncharged_equivalent_target_without_a_record_still_fails_closed() { + let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await; + start_leftover_intent_test_decommission(&store, Vec::new(), DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION).await; + // Another mutation holds the slot, nothing was charged on the target, + // and this mutation has neither a pending intent nor a marker there. + let waiting = uuid::Uuid::new_v4(); + leave_unfinalized_decommission_target_intent(&store, 1, waiting, 8).await; + let before = leftover_intent_test_target(&store, 1).await; + assert_eq!(before.consumed_physical_bytes, 0); + let unrecorded_owner = leftover_intent_test_owner(&store, uuid::Uuid::new_v4()).await; + + let err = store + .reconcile_decommission_capacity_after_equivalent_target(unrecorded_owner, 1, 16) + .await + .expect_err("a commit the target never charged must not be assumed charged"); + assert!( + err.to_string() + .contains("equivalent target has no pending capacity intent to reconcile"), + "unexpected reconciliation error: {err}" + ); + assert_eq!(leftover_intent_test_target(&store, 1).await, before); + } + fn pool_meta_replica_test_meta(cmd_line: &str) -> PoolMeta { PoolMeta { version: POOL_META_VERSION, @@ -22688,6 +23675,24 @@ mod tests { ); } + /// The entry context must not hide a busy target gate either. + #[test] + fn decommission_entry_context_keeps_a_busy_target_gate_deferred() { + let gate_busy = decommission_capacity_blocked_error(format!( + "{DECOMMISSION_CAPACITY_TARGET_GATE_BUSY_PREFIX}1{DECOMMISSION_CAPACITY_TARGET_GATE_BUSY_SUFFIX}" + )); + let wrapped = with_decommission_entry_context("delete_marker_copy", "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(_)) + )); + } + /// A stage wrapper must not hide the underlying capacity class. #[test] fn decommission_capacity_failure_classification_sees_through_a_stage_wrapper() { @@ -23433,6 +24438,34 @@ mod tests { ); } + #[test] + fn decommission_capacity_conflict_stays_classified_through_entry_context_and_clone() { + let err = with_decommission_entry_context( + "delete_marker_copy", + "bucket", + "object", + decommission_capacity_blocked_error( + "source pool 0 target pool 1 pending capacity intent belongs to another mutation", + ), + ); + // A shared entry error slot hands out clones, which keep only the + // rendered message of an I/O error. + let cloned = err.clone(); + assert!(is_decommission_capacity_blocked_error(&err)); + assert_eq!( + decommission_capacity_retry_kind(&err, 0), + Some(DecommissionCapacityRetryKind::IntentConflict), + "a wrapped conflict must keep the bounded intent-conflict retry" + ); + // The routine pauses on either capacity predicate; a clone is still + // recognized as a capacity block and as an intent conflict. + assert!(is_decommission_capacity_blocked_error(&cloned) || is_decommission_target_capacity_error(&cloned)); + assert!(is_decommission_capacity_intent_conflict(&cloned)); + assert!(!is_decommission_capacity_blocked_error(&Error::other( + "decommission entry update_after failed" + ))); + } + #[test] fn decommission_target_capacity_error_rejects_unrelated_errors() { assert!(!is_decommission_target_capacity_error(&Error::SlowDown)); @@ -24623,7 +25656,6 @@ mod pools_tests { use super::DECOMMISSION_PROGRESS_SAVE_RETRY_BACKOFF; use super::persist_v3_pool_meta_for_test; use super::record_decommission_entry_error; - use super::resolve_decommission_listing_error; use super::resolve_decommission_partial_listing_entry; use super::{ DECOMMISSION_CAPACITY_LEGACY_MODEL_VERSION, DECOMMISSION_CAPACITY_MODEL_VERSION, DECOMMISSION_CAPACITY_RELEASE_CANCELED, @@ -24684,13 +25716,20 @@ mod pools_tests { }; use super::{ 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, + DecommissionCapacityLeftoverIntent, DecommissionCapacityOwner, DecommissionCapacityReleaseProof, + DecommissionCapacityReservation, DecommissionCapacityTemporaryMutation, adopt_decommission_capacity_leftover_intents, + decommission_adopted_intent_mutation_id, decommission_capacity_blocked_error, decommission_capacity_intent_adoption_plan, + decommission_capacity_mutation_id, decommission_capacity_target_gate_retry_exhausted, + decommission_intent_adoption_failure, 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, + validate_decommission_capacity_reservation, + }; + use super::{ + decommission_bucket_failure, decommission_reserved_target_choice, decommission_root_cause, + decommission_target_has_mutation_marker, settle_decommission_target_non_growing_replacement, }; use crate::bucket::lifecycle::{ DurableIlmRecordCheckpoint, @@ -27935,6 +28974,63 @@ mod pools_tests { assert_eq!(running.load(Ordering::SeqCst), 0); } + #[test] + fn decommission_bucket_failure_reports_the_cause_behind_canceled_listings() { + let capacity = || decommission_capacity_blocked_error("pending capacity intent belongs to another mutation"); + // A listing worker canceled by another set's capacity-blocked entry. + let err = decommission_bucket_failure(vec![Ok(()), Err(Error::OperationCanceled)], Some(capacity())) + .expect("the bucket must fail"); + assert!(is_decommission_capacity_blocked_error(&err), "unexpected bucket failure: {err:?}"); + // A canceled listing worker ahead of a capacity-blocked one. + let err = decommission_bucket_failure(vec![Err(Error::OperationCanceled), Err(capacity())], None) + .expect("the bucket must fail"); + assert!(is_decommission_capacity_blocked_error(&err), "unexpected bucket failure: {err:?}"); + // The first real failure stays the reported cause. + let err = decommission_bucket_failure(vec![Err(Error::SlowDown)], Some(capacity())).expect("the bucket must fail"); + assert!(matches!(err, Error::SlowDown), "unexpected bucket failure: {err:?}"); + assert!(decommission_bucket_failure(vec![Ok(()), Ok(())], None).is_none()); + assert!(matches!( + decommission_root_cause(Some(Error::OperationCanceled), Some(Error::OperationCanceled)), + Some(Error::OperationCanceled) + )); + } + + #[tokio::test] + async fn test_run_decommission_buckets_bounded_reports_the_failure_behind_a_sibling_cancellation() { + let rx = CancellationToken::new(); + let buckets = ["failing", "sibling"] + .into_iter() + .map(|name| DecomBucketInfo { + name: name.to_string(), + ..Default::default() + }) + .collect::>(); + + let err = run_decommission_buckets_bounded(rx, buckets, 2, |bucket, rx| { + Box::pin(async move { + if bucket.name == "failing" { + // A failing bucket cancels the shared token from inside, + // then still drains its own workers before returning. + rx.cancel(); + tokio::time::sleep(StdDuration::from_millis(50)).await; + Err(decommission_capacity_blocked_error( + "source pool 0 target pool 1 pending capacity intent belongs to another mutation", + )) + } else { + rx.cancelled().await; + Err(Error::OperationCanceled) + } + }) + }) + .await + .expect_err("a failed bucket must fail the phase"); + + assert!( + is_decommission_capacity_blocked_error(&err), + "the routine must see the capacity block, not the sibling's cancellation: {err:?}" + ); + } + #[tokio::test] async fn test_run_decommission_buckets_bounded_cancels_and_stops_launching_after_failure() { let rx = CancellationToken::new(); @@ -29755,11 +30851,11 @@ mod pools_tests { #[test] fn test_resolve_decommission_listing_error_preserves_real_listing_failure() { - let err = resolve_decommission_listing_error(Some(Error::SlowDown), Some(Error::OperationCanceled)) + let err = decommission_root_cause(Some(Error::SlowDown), Some(Error::OperationCanceled)) .expect("listing failure should be returned"); assert!(matches!(err, Error::SlowDown)); - let err = resolve_decommission_listing_error(Some(Error::OperationCanceled), Some(Error::SlowDown)) + let err = decommission_root_cause(Some(Error::OperationCanceled), Some(Error::SlowDown)) .expect("entry failure should be returned"); assert!(matches!(err, Error::SlowDown)); } @@ -31737,6 +32833,133 @@ mod pools_tests { assert!(reservation.targets[0].temporary_mutations.is_empty()); } + #[test] + fn cleanup_without_an_absence_proof_keeps_an_adopted_completion() { + let generation = OffsetDateTime::UNIX_EPOCH; + let now = OffsetDateTime::UNIX_EPOCH + Duration::seconds(1); + let completion = uuid::Uuid::new_v4(); + let mut meta = decommission_test_cleanup_meta( + DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION, + Vec::new(), + 0, + Some((completion, 40)), + ); + let plan = decommission_capacity_intent_adoption_plan(&meta, 0, generation) + .expect("plan adoption") + .expect("the interrupted completion should be planned"); + adopt_decommission_capacity_leftover_intents(&mut meta, 0, generation, &plan, now) + .expect("adopt the interrupted completion"); + let adopted = decommission_adopted_intent_mutation_id(completion); + let adopted_bytes = |meta: &PoolMeta| { + intent_adoption_test_reservation(meta).targets[0] + .temporary_mutations + .iter() + .find(|mutation| mutation.mutation_id == adopted) + .map(|mutation| mutation.physical_bytes) + }; + assert_eq!(adopted_bytes(&meta), Some(40)); + + // Whatever the pool-wide free-space delta reads, a cleanup that did not + // prove the exact target absent cannot settle the adopted completion. + for released_physical_bytes in [0, 40, 1 << 20] { + let mut target_present = meta.clone(); + release_decommission_target_inflight( + &mut target_present, + 0, + 1, + released_physical_bytes, + completion, + DecommissionCapacityReleaseProof::confirmed_absence(false), + now, + ) + .expect("a cleanup that found the target should complete"); + assert_eq!( + adopted_bytes(&target_present), + Some(40), + "free-space delta {released_physical_bytes} must not settle the adopted completion" + ); + validate_decommission_capacity_reservation(Some(intent_adoption_test_reservation(&target_present))) + .expect("the ledger must remain loadable"); + } + + // As for a published pending intent, freed bytes another mutation's + // marker could own stay unattributable. + let mut foreign_present = meta.clone(); + { + let reservation = foreign_present.pools[0] + .decommission + .as_mut() + .and_then(|info| info.capacity_reservation.as_mut()) + .expect("adoption should keep the reservation"); + reservation.targets[0] + .temporary_mutations + .push(DecommissionCapacityTemporaryMutation { + mutation_id: uuid::Uuid::new_v4(), + physical_bytes: 20, + }); + reservation.targets[0].inflight_physical_bytes += 20; + reservation.inflight_target_physical_bytes += 20; + } + let err = release_decommission_target_inflight( + &mut foreign_present, + 0, + 1, + 20, + completion, + DecommissionCapacityReleaseProof::confirmed_absence(false), + now, + ) + .expect_err("a delta another mutation could own must fail closed"); + assert!(is_decommission_capacity_blocked_error(&err), "unexpected cleanup error: {err:?}"); + + // Only a proven-absent target means nothing committed. + let mut target_absent = meta; + release_decommission_target_inflight( + &mut target_absent, + 0, + 1, + 40, + completion, + DecommissionCapacityReleaseProof::confirmed_absence(true), + now, + ) + .expect("a cleanup that proved the target absent should release the adopted completion"); + assert_eq!(adopted_bytes(&target_absent), None); + assert_eq!(intent_adoption_test_reservation(&target_absent).inflight_target_physical_bytes, 0); + } + + #[test] + fn cleanup_without_an_absence_proof_still_releases_staged_bytes() { + let staged = uuid::Uuid::new_v4(); + let mut meta = decommission_test_cleanup_meta( + DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION, + vec![DecommissionCapacityTemporaryMutation { + mutation_id: staged, + physical_bytes: 30, + }], + 30, + None, + ); + // The cleanup finished within the free-space sampling interval, so the + // observed delta is zero; the staged uploads are still confirmed gone. + release_decommission_target_inflight( + &mut meta, + 0, + 1, + 0, + staged, + DecommissionCapacityReleaseProof::confirmed_absence(false), + OffsetDateTime::UNIX_EPOCH + Duration::seconds(1), + ) + .expect("a cleanup that found the target should release staged bytes"); + let reservation = intent_adoption_test_reservation(&meta); + assert!( + reservation.targets[0].temporary_mutations.is_empty(), + "staged bytes must not survive a cleanup that confirmed the uploads absent" + ); + assert_eq!(reservation.inflight_target_physical_bytes, 0); + } + #[test] fn published_pending_ignores_a_delayed_release_observation() { let mutation_id = uuid::Uuid::new_v4(); @@ -31876,6 +33099,612 @@ mod pools_tests { assert_eq!(after, &before); } + fn intent_adoption_test_reservation(meta: &PoolMeta) -> &DecommissionCapacityReservation { + meta.pools[0] + .decommission + .as_ref() + .and_then(|info| info.capacity_reservation.as_ref()) + .expect("intent adoption test reservation should be present") + } + + #[test] + fn resumed_decommission_adopts_leftover_target_intent_as_mutation_marker() { + let generation = OffsetDateTime::UNIX_EPOCH; + let now = OffsetDateTime::UNIX_EPOCH + Duration::seconds(1); + for model_version in [ + DECOMMISSION_CAPACITY_LEGACY_MODEL_VERSION, + DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION, + ] { + let leftover = uuid::Uuid::new_v4(); + let waiting = uuid::Uuid::new_v4(); + let mut meta = decommission_test_cleanup_meta(model_version, Vec::new(), 0, Some((leftover, 11))); + let err = reserve_decommission_target_pending(&mut meta.clone(), 0, 1, 5, waiting, now) + .expect_err("an unadopted leftover intent blocks every other mutation on its target"); + assert!(err.to_string().contains("unresolved target capacity intent")); + + let plan = decommission_capacity_intent_adoption_plan(&meta, 0, generation) + .expect("the same decommission should plan adoption") + .expect("the leftover intent should be planned"); + assert_eq!( + plan.intents, + vec![DecommissionCapacityLeftoverIntent { + target_pool_index: 1, + mutation_id: leftover, + physical_bytes: 11, + }] + ); + adopt_decommission_capacity_leftover_intents(&mut meta, 0, generation, &plan, now) + .expect("the resumed routine should adopt its leftover intent"); + + let reservation = intent_adoption_test_reservation(&meta); + validate_decommission_capacity_reservation(Some(reservation)).expect("the adopted ledger must remain loadable"); + let target = &reservation.targets[0]; + assert_eq!(target.pending_physical_bytes, 0); + assert_eq!(target.pending_mutation_id, None); + assert_eq!( + target.temporary_mutations, + vec![DecommissionCapacityTemporaryMutation { + mutation_id: decommission_adopted_intent_mutation_id(leftover), + physical_bytes: 11, + }] + ); + assert_eq!(target.inflight_physical_bytes, 11); + assert_eq!(reservation.pending_target_physical_bytes, 0); + assert_eq!(reservation.inflight_target_physical_bytes, 11); + assert_eq!( + reservation.consumed_target_physical_bytes, 0, + "adoption must not assume the interrupted write committed" + ); + assert!( + decommission_capacity_intent_adoption_plan(&meta, 0, generation) + .expect("planning after adoption should succeed") + .is_none() + ); + assert_eq!( + reserve_decommission_target_pending(&mut meta.clone(), 0, 1, 5, waiting, now) + .expect("another mutation of the resumed routine may use the target"), + 5 + ); + + // The interrupted mutation's own retry re-reserves and settles its + // adopted marker exactly once. + assert_eq!( + reserve_decommission_target_pending(&mut meta, 0, 1, 11, leftover, now) + .expect("the interrupted mutation should be able to retry"), + 11 + ); + resolve_decommission_target_pending(&mut meta, 0, 1, 11, leftover).expect("the retry should resolve its intent"); + record_decommission_target_consumption( + &mut meta, + 0, + 1, + DecommissionTargetConsumption { + committed_data_bytes: 11, + target_physical_bytes: 11, + observed_physical_bytes: 0, + }, + leftover, + now, + ) + .expect("the retry should record its committed bytes"); + let reservation = intent_adoption_test_reservation(&meta); + validate_decommission_capacity_reservation(Some(reservation)).expect("the settled ledger must remain loadable"); + assert!(reservation.targets[0].temporary_mutations.is_empty()); + assert_eq!(reservation.inflight_target_physical_bytes, 0); + assert_eq!(reservation.consumed_target_physical_bytes, 11); + } + } + + #[test] + fn decommission_intent_adoption_merges_into_the_mutation_staging_marker() { + let generation = OffsetDateTime::UNIX_EPOCH; + let leftover = uuid::Uuid::new_v4(); + let mut meta = decommission_test_cleanup_meta( + DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION, + vec![DecommissionCapacityTemporaryMutation { + mutation_id: leftover, + physical_bytes: 7, + }], + 7, + Some((leftover, 11)), + ); + let plan = decommission_capacity_intent_adoption_plan(&meta, 0, generation) + .expect("plan adoption") + .expect("the leftover multipart completion intent should be planned"); + adopt_decommission_capacity_leftover_intents(&mut meta, 0, generation, &plan, OffsetDateTime::UNIX_EPOCH) + .expect("adopt the leftover multipart completion intent"); + let reservation = intent_adoption_test_reservation(&meta); + validate_decommission_capacity_reservation(Some(reservation)).expect("the merged ledger must remain loadable"); + assert_eq!( + reservation.targets[0].temporary_mutations, + vec![ + DecommissionCapacityTemporaryMutation { + mutation_id: leftover, + physical_bytes: 7, + }, + DecommissionCapacityTemporaryMutation { + mutation_id: decommission_adopted_intent_mutation_id(leftover), + physical_bytes: 11, + }, + ], + "the staged parts keep their own marker beside the adopted completion" + ); + assert_eq!(reservation.inflight_target_physical_bytes, 18); + } + + 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) + } + + /// A mutation's staged and adopted marker bytes together. + fn intent_adoption_test_marker(meta: &PoolMeta, mutation_id: uuid::Uuid) -> Option { + let adopted = decommission_adopted_intent_mutation_id(mutation_id); + intent_adoption_test_reservation(meta).targets[0] + .temporary_mutations + .iter() + .filter(|mutation| mutation.mutation_id == mutation_id || mutation.mutation_id == adopted) + .map(|mutation| mutation.physical_bytes) + .reduce(usize::saturating_add) + } + + /// Adopt every pending intent of the test reservation as a resumed routine + /// would, keeping the ledger loadable. + fn adopt_intent_adoption_test_intents(meta: &mut PoolMeta) { + let generation = OffsetDateTime::UNIX_EPOCH; + let plan = decommission_capacity_intent_adoption_plan(meta, 0, generation) + .expect("plan adoption") + .expect("a pending intent should be planned"); + adopt_decommission_capacity_leftover_intents(meta, 0, generation, &plan, generation + Duration::seconds(1)) + .expect("adopt the pending intents"); + validate_decommission_capacity_reservation(Some(intent_adoption_test_reservation(meta))) + .expect("the adopted ledger must remain loadable"); + } + + /// Finalize a mutation's commit on a target exactly as a successful + /// capacity mutation does: reserve its intent, then resolve and charge it. + fn commit_intent_adoption_test_mutation( + meta: &mut PoolMeta, + target_pool_index: usize, + mutation_id: uuid::Uuid, + bytes: usize, + ) { + let now = OffsetDateTime::UNIX_EPOCH + Duration::seconds(2); + reserve_decommission_target_pending(meta, 0, target_pool_index, bytes, mutation_id, now) + .expect("reserve the retry intent"); + resolve_decommission_target_pending(meta, 0, target_pool_index, bytes, mutation_id).expect("resolve the retry 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 retry"); + validate_decommission_capacity_reservation(Some(intent_adoption_test_reservation(meta))) + .expect("the committed ledger must remain loadable"); + } + + fn assert_no_intent_adoption_test_record(meta: &PoolMeta, mutation_id: uuid::Uuid) { + let reservation = intent_adoption_test_reservation(meta); + for target in &reservation.targets { + assert_ne!(target.pending_mutation_id, Some(mutation_id)); + assert!( + !decommission_target_has_mutation_marker(target, mutation_id), + "target pool {} still holds a marker of the mutation: {:?}", + target.pool_index, + target.temporary_mutations + ); + assert_eq!(target.inflight_physical_bytes, 0); + } + assert_eq!(reservation.inflight_target_physical_bytes, 0); + } + + #[test] + fn retry_returns_to_the_target_that_holds_its_adopted_completion() { + 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)), + ); + // Split the allocation so that target 2 has three times the remaining + // allocation of target 1, where the leader died before committing. + { + 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)); + validate_decommission_capacity_reservation(Some(intent_adoption_test_reservation(&meta))) + .expect("the split ledger must be loadable"); + adopt_intent_adoption_test_intents(&mut meta); + + let reservation = intent_adoption_test_reservation(&meta); + assert_eq!( + decommission_reserved_target_choice(reservation, None, 10), + Some(2), + "without an adopted completion the larger allocation wins" + ); + assert_eq!( + decommission_reserved_target_choice(reservation, Some(interrupted), 10), + 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), + Some(2) + ); + + commit_intent_adoption_test_mutation(&mut meta, 1, interrupted, 10); + assert_no_intent_adoption_test_record(&meta, interrupted); + } + + #[test] + fn adopting_the_same_completion_twice_counts_it_once() { + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + let capacity_infos = vec![DecommissionPoolCapacityInfo::for_test(1, layout, 400, 400, 0)]; + let mutation_id = uuid::Uuid::new_v4(); + let mut meta = decommission_test_cleanup_meta( + DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION, + Vec::new(), + 0, + Some((mutation_id, 30)), + ); + adopt_intent_adoption_test_intents(&mut meta); + // The resumed routine's retry reserves the same completion again and + // is killed again before finalizing; the next incarnation adopts it. + reserve_decommission_target_pending(&mut meta, 0, 1, 30, mutation_id, OffsetDateTime::UNIX_EPOCH) + .expect("reserve the second attempt"); + adopt_intent_adoption_test_intents(&mut meta); + + let reservation = intent_adoption_test_reservation(&meta); + let target = &reservation.targets[0]; + assert_eq!(target.pending_mutation_id, None); + assert_eq!( + intent_adoption_test_marker(&meta, mutation_id), + Some(30), + "both adoptions reserved the same completion" + ); + assert_eq!(target.inflight_physical_bytes, 30); + assert_eq!(reservation.inflight_target_physical_bytes, 30); + + // Another mutation commits 60 bytes and shrinks the temporary-copy + // 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") + .expect("later admission must not trip over a completion counted twice"); + + commit_intent_adoption_test_mutation(&mut meta, 1, mutation_id, 30); + assert_no_intent_adoption_test_record(&meta, mutation_id); + } + + #[test] + fn adoption_keeps_a_zero_byte_staging_record_only_in_the_target_fence_model() { + for model_version in [ + DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION, + DECOMMISSION_CAPACITY_LEGACY_MODEL_VERSION, + ] { + let completion = uuid::Uuid::new_v4(); + let budget = + intent_adoption_test_temporary_budget(&decommission_test_cleanup_meta(model_version, Vec::new(), 0, None)); + // The completion alone fills the budget, so the staged bytes move + // into the adopted marker and the staged marker drops to zero. + let staged = budget * 6 / 10; + let mut meta = decommission_test_cleanup_meta( + model_version, + vec![DecommissionCapacityTemporaryMutation { + mutation_id: completion, + physical_bytes: staged, + }], + staged, + Some((completion, budget)), + ); + adopt_intent_adoption_test_intents(&mut meta); + + let target = &intent_adoption_test_reservation(&meta).targets[0]; + let staged_entry = target + .temporary_mutations + .iter() + .find(|mutation| mutation.mutation_id == completion); + if model_version == DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION { + assert_eq!( + staged_entry.map(|mutation| mutation.physical_bytes), + Some(0), + "the target-fence ledger keeps the zero-byte multipart discovery record" + ); + } else { + assert!(staged_entry.is_none(), "the legacy ledger cannot store a zero-byte marker"); + } + assert_eq!(intent_adoption_test_marker(&meta, completion), Some(budget)); + assert_eq!(target.inflight_physical_bytes, budget); + } + } + + #[test] + fn non_growing_replacement_settles_an_adopted_checkpoint_intent() { + let checkpoint = uuid::Uuid::new_v4(); + let mut meta = decommission_test_cleanup_meta( + DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION, + Vec::new(), + 0, + Some((checkpoint, 16)), + ); + adopt_intent_adoption_test_intents(&mut meta); + assert_eq!(intent_adoption_test_marker(&meta, checkpoint), Some(16)); + + assert!( + settle_decommission_target_non_growing_replacement( + &mut meta, + 0, + 1, + checkpoint, + OffsetDateTime::UNIX_EPOCH + Duration::seconds(2) + ) + .expect("the checkpoint replacement settles") + ); + assert_no_intent_adoption_test_record(&meta, checkpoint); + assert_eq!( + intent_adoption_test_reservation(&meta).targets[0].consumed_physical_bytes, + 0, + "a non-growing replacement charges nothing" + ); + validate_decommission_capacity_reservation(Some(intent_adoption_test_reservation(&meta))) + .expect("the settled ledger must remain loadable"); + } + + #[test] + fn adopted_multipart_completion_stays_within_the_temporary_copy_budget() { + let generation = OffsetDateTime::UNIX_EPOCH; + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + let capacity_infos = vec![DecommissionPoolCapacityInfo::for_test(1, layout, 400, 400, 0)]; + let completion = uuid::Uuid::new_v4(); + let budget = intent_adoption_test_temporary_budget(&decommission_test_cleanup_meta( + DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION, + Vec::new(), + 0, + None, + )); + // The parts staged 60% of the budget and the completion reserved the + // same object again; their sum exceeds the budget. + let staged = budget * 6 / 10; + let mut meta = decommission_test_cleanup_meta( + DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION, + vec![DecommissionCapacityTemporaryMutation { + mutation_id: completion, + physical_bytes: staged, + }], + staged, + Some((completion, staged)), + ); + ensure_decommission_capacity_reservations_available(&meta, &capacity_infos, "migration") + .expect("the interrupted completion's ledger fits its budget"); + + let plan = decommission_capacity_intent_adoption_plan(&meta, 0, generation) + .expect("plan adoption") + .expect("the interrupted completion should be planned"); + adopt_decommission_capacity_leftover_intents(&mut meta, 0, generation, &plan, OffsetDateTime::UNIX_EPOCH) + .expect("adopt the interrupted completion"); + validate_decommission_capacity_reservation(Some(intent_adoption_test_reservation(&meta))) + .expect("the adopted ledger must remain loadable"); + assert_eq!( + intent_adoption_test_marker(&meta, completion), + Some(budget), + "the overlapping staged parts and completion are counted up to the budget" + ); + ensure_decommission_capacity_reservations_available(&meta, &capacity_infos, "migration") + .expect("adoption must not push inflight bytes past the temporary-copy budget"); + } + + #[test] + fn adopted_part_upload_keeps_its_bytes_on_top_of_earlier_parts() { + let generation = OffsetDateTime::UNIX_EPOCH; + let upload = uuid::Uuid::new_v4(); + let mut meta = decommission_test_cleanup_meta( + DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION, + vec![DecommissionCapacityTemporaryMutation { + mutation_id: upload, + physical_bytes: 30, + }], + 30, + Some((upload, 20)), + ); + let plan = decommission_capacity_intent_adoption_plan(&meta, 0, generation) + .expect("plan adoption") + .expect("the interrupted part upload should be planned"); + adopt_decommission_capacity_leftover_intents(&mut meta, 0, generation, &plan, OffsetDateTime::UNIX_EPOCH) + .expect("adopt the interrupted part upload"); + assert_eq!( + intent_adoption_test_marker(&meta, upload), + Some(50), + "a part admitted within the budget adds to the parts staged before it" + ); + } + + #[test] + fn adopted_intent_exceeds_the_budget_only_where_its_finalize_would() { + let generation = OffsetDateTime::UNIX_EPOCH; + let now = OffsetDateTime::UNIX_EPOCH + Duration::seconds(1); + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + let capacity_infos = vec![DecommissionPoolCapacityInfo::for_test(1, layout, 400, 400, 0)]; + let staging = uuid::Uuid::new_v4(); + let completion = uuid::Uuid::new_v4(); + let budget = intent_adoption_test_temporary_budget(&decommission_test_cleanup_meta( + DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION, + Vec::new(), + 0, + None, + )); + for (other_staged, completion_bytes) in [(budget / 4, budget / 2), (budget / 2, budget * 6 / 10)] { + let meta = decommission_test_cleanup_meta( + DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION, + vec![DecommissionCapacityTemporaryMutation { + mutation_id: staging, + physical_bytes: other_staged, + }], + other_staged, + Some((completion, completion_bytes)), + ); + ensure_decommission_capacity_reservations_available(&meta, &capacity_infos, "migration") + .expect("the interrupted ledger fits its budget"); + + let mut finalized = meta.clone(); + resolve_decommission_target_pending(&mut finalized, 0, 1, completion_bytes, completion) + .expect("finalize resolves the completion intent"); + record_decommission_target_consumption( + &mut finalized, + 0, + 1, + DecommissionTargetConsumption { + committed_data_bytes: completion_bytes, + target_physical_bytes: completion_bytes, + observed_physical_bytes: 0, + }, + completion, + now, + ) + .expect("finalize records the completion"); + + let mut adopted = meta.clone(); + let plan = decommission_capacity_intent_adoption_plan(&adopted, 0, generation) + .expect("plan adoption") + .expect("the interrupted completion should be planned"); + adopt_decommission_capacity_leftover_intents(&mut adopted, 0, generation, &plan, now) + .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(), + "adoption must reach the same budget verdict as the lost finalize (other staged {other_staged}, completion {completion_bytes}, budget {budget})" + ); + } + } + + #[test] + fn decommission_intent_adoption_is_pinned_to_the_same_decommission() { + let generation = OffsetDateTime::UNIX_EPOCH; + let now = OffsetDateTime::UNIX_EPOCH + Duration::seconds(1); + let leftover = uuid::Uuid::new_v4(); + let meta = + decommission_test_cleanup_meta(DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION, Vec::new(), 0, Some((leftover, 11))); + + let err = decommission_capacity_intent_adoption_plan(&meta, 0, generation + Duration::seconds(1)) + .expect_err("a routine for another decommission start must not plan adoption"); + assert!(matches!(err, Error::OperationCanceled), "unexpected planning error: {err}"); + + let plan = decommission_capacity_intent_adoption_plan(&meta, 0, generation) + .expect("plan adoption") + .expect("the leftover intent should be planned"); + let intent_changes: [fn(&mut DecommissionCapacityReservation); 4] = [ + |reservation| reservation.operation_id = uuid::Uuid::new_v4(), + |reservation| reservation.generation += 1, + |reservation| reservation.targets[0].pending_mutation_id = Some(uuid::Uuid::new_v4()), + |reservation| { + reservation.targets[0].pending_physical_bytes += 1; + reservation.pending_target_physical_bytes += 1; + }, + ]; + for change in intent_changes { + let mut changed = meta.clone(); + change( + changed.pools[0] + .decommission + .as_mut() + .and_then(|info| info.capacity_reservation.as_mut()) + .expect("intent adoption test reservation should be present"), + ); + let before = intent_adoption_test_reservation(&changed).clone(); + let err = adopt_decommission_capacity_leftover_intents(&mut changed, 0, generation, &plan, now) + .expect_err("a reservation or intent that changed after planning must not be adopted"); + assert!(err.to_string().contains("changed before adoption"), "unexpected adoption error: {err}"); + assert_eq!(intent_adoption_test_reservation(&changed), &before); + } + + let mut unidentified = meta; + unidentified.pools[0] + .decommission + .as_mut() + .and_then(|info| info.capacity_reservation.as_mut()) + .expect("intent adoption test reservation should be present") + .targets[0] + .pending_mutation_id = None; + assert!( + decommission_capacity_intent_adoption_plan(&unidentified, 0, generation) + .expect("planning should tolerate an unidentified intent") + .is_none(), + "an intent without a mutation identity cannot be settled and stays fail-closed" + ); + } + + #[test] + fn decommission_intent_adoption_failure_pauses_instead_of_failing_the_routine() { + assert!(matches!( + decommission_intent_adoption_failure(Error::OperationCanceled), + Error::OperationCanceled + )); + let busy = decommission_intent_adoption_failure(decommission_capacity_blocked_error("target gate busy")); + assert_eq!(busy.to_string(), decommission_capacity_blocked_error("target gate busy").to_string()); + let quorum = decommission_intent_adoption_failure(Error::NamespaceLockQuorumUnavailable { + mode: "write", + bucket: RUSTFS_META_BUCKET.to_string(), + object: "decommission/capacity-target/1".to_string(), + required: 2, + achieved: 1, + }); + assert!( + is_decommission_capacity_blocked_error(&quorum), + "a lock failure while adopting must pause the routine, not mark the decommission failed: {quorum}" + ); + } + + #[test] + fn decommission_intent_adoption_does_not_bypass_a_live_intent_of_the_same_incarnation() { + let generation = OffsetDateTime::UNIX_EPOCH; + let now = OffsetDateTime::UNIX_EPOCH + Duration::seconds(1); + let leftover = uuid::Uuid::new_v4(); + let mut meta = + decommission_test_cleanup_meta(DECOMMISSION_CAPACITY_TARGET_FENCE_MODEL_VERSION, Vec::new(), 0, Some((leftover, 11))); + let plan = decommission_capacity_intent_adoption_plan(&meta, 0, generation) + .expect("plan adoption") + .expect("the leftover intent should be planned"); + adopt_decommission_capacity_leftover_intents(&mut meta, 0, generation, &plan, now).expect("adopt the leftover intent"); + + // Adoption happens once, before workers start. An intent a worker of + // this incarnation then persists keeps excluding other mutations. + let live = uuid::Uuid::new_v4(); + let other = uuid::Uuid::new_v4(); + reserve_decommission_target_pending(&mut meta, 0, 1, 5, live, now).expect("a live mutation reserves the target"); + let err = reserve_decommission_target_pending(&mut meta.clone(), 0, 1, 5, other, now) + .expect_err("a live intent must still reject a different mutation"); + assert!(err.to_string().contains("unresolved target capacity intent")); + let err = resolve_decommission_target_pending(&mut meta.clone(), 0, 1, 5, other) + .expect_err("a different mutation must not resolve a live intent"); + assert!(err.to_string().contains("belongs to another mutation")); + let err = adopt_decommission_capacity_leftover_intents(&mut meta.clone(), 0, generation, &plan, now) + .expect_err("a stale plan must not adopt the live intent"); + assert!(err.to_string().contains("changed before adoption")); + } + #[test] fn pool_meta_v2_round_trip_preserves_each_supported_capacity_lock_model() { for model_version in [ diff --git a/crates/ecstore/src/core/pools_test.rs b/crates/ecstore/src/core/pools_test.rs index ae9548f05..9fa6758fc 100644 --- a/crates/ecstore/src/core/pools_test.rs +++ b/crates/ecstore/src/core/pools_test.rs @@ -7749,4 +7749,610 @@ mod decommission_lock_order_tests { .await; }); } + + async fn killed_leader_target( + store: &Arc, + target_pool_index: usize, + ) -> crate::core::pools::DecommissionCapacityTarget { + store.pool_meta.read().await.pools[0] + .decommission + .as_ref() + .and_then(|info| info.capacity_reservation.as_ref()) + .and_then(|reservation| { + reservation + .targets + .iter() + .find(|target| target.pool_index == target_pool_index) + }) + .cloned() + .expect("the reservation should allocate the target pool") + } + + async fn run_resumed_decommission_routine(store: &Arc) { + 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 resumed decommission routine should return"); + } + + /// Kill the decommission leader inside the target write window of a real + /// single-PUT migration: before the target commit, or after it but before + /// the capacity finalize. Returns the bucket and the interrupted mutation. + async fn kill_leader_inside_target_put( + store: &Arc, + committed: bool, + object: &str, + ) -> (String, uuid::Uuid) { + // The routine also migrates `.rustfs.sys` bucket metadata, so keep the + // reservation far above every fixture object. + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + kill_leader_inside_target_put_with_capacity( + store, + committed, + object, + vec![ + DecommissionPoolCapacityInfo::for_test(0, layout, 0, 64 << 20, 64 << 20), + DecommissionPoolCapacityInfo::for_test(1, layout, 1 << 30, 1 << 30, 0), + ], + ) + .await + } + + /// Like `kill_leader_inside_target_put`, with the capacity snapshot the + /// decommission reserves from. The selection must route the killed + /// migration to target pool 1. + async fn kill_leader_inside_target_put_with_capacity( + store: &Arc, + committed: bool, + object: &str, + capacity: Vec, + ) -> (String, uuid::Uuid) { + let bucket = test_bucket("killed-leader-put"); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create the killed-leader bucket"); + let mut source = PutObjReader::from_vec(b"source survives the leader kill".to_vec()); + store.pools[0] + .put_object(&bucket, object, &mut source, &ObjectOptions::default()) + .await + .expect("write the source object before decommission starts"); + let incarnation = store + .bucket_incarnation_id(&bucket) + .await + .expect("load the killed-leader bucket incarnation"); + set_decommission_capacity_info_overrides_for_test(store.id, vec![capacity]); + store + .save_current_pool_meta_for_decommission_start( + &[0], + vec![crate::core::pools::DecomBucketInfo { + name: bucket.clone(), + prefix: String::new(), + }], + ) + .await + .expect("activate the killed-leader decommission"); + let owner = decommission_capacity_owner(&*store.pool_meta.read().await); + let source_reader = store.pools[0] + .get_object_reader( + &bucket, + object, + None, + HeaderMap::new(), + &ObjectOptions { + no_lock: true, + data_movement: true, + raw_data_movement_read: true, + ..Default::default() + }, + ) + .await + .expect("read the source object for the interrupted migration"); + let pause = if committed { + PutObjectCommitPause::AfterRenameHandoff + } else { + PutObjectCommitPause::BeforeQuotaRename + }; + let barrier = PutObjectCommitBarrier::install(&bucket, object, pause); + let mut migration = tokio::spawn({ + let migration_store = Arc::clone(store); + let migration_bucket = bucket.clone(); + async move { + data_movement::migrate_decommission_object( + migration_store, + 0, + migration_bucket, + source_reader, + Some(incarnation), + "killed_leader_put", + Some(owner), + ) + .await + } + }); + tokio::select! { + () = barrier.wait_until_paused() => {} + result = &mut migration => { + panic!("the migration finished before the leader could be killed in its target write window: {result:?}"); + } + } + // The leader dies inside the target write window. + migration.abort(); + let _ = migration.await; + barrier.release(); + drop(barrier); + + let interrupted = killed_leader_target(store, 1) + .await + .pending_mutation_id + .expect("the killed mutation must leave its durable capacity intent"); + let target_copy = store.pools[1] + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await; + assert_eq!( + target_copy.is_ok(), + committed, + "the kill window must match the requested commit state: {target_copy:?}" + ); + (bucket, interrupted) + } + + async fn assert_resumed_decommission_settled( + store: &Arc, + bucket: &str, + object: &str, + interrupted: uuid::Uuid, + ) { + 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, + "the resumed decommission must complete: complete={}, failed={}, canceled={}, blocked_reason={:?}", + info.complete, + info.failed, + info.canceled, + info.capacity_blocked_reason + ); + assert_eq!(info.items_decommission_failed, 0); + drop(local); + let target = killed_leader_target(store, 1).await; + assert_eq!(target.pending_mutation_id, None); + assert_eq!(target.pending_physical_bytes, 0); + let adopted = crate::core::pools::decommission_adopted_intent_mutation_id(interrupted); + assert!( + target + .temporary_mutations + .iter() + .all(|mutation| mutation.mutation_id != interrupted && mutation.mutation_id != adopted), + "the interrupted mutation's retry must settle its markers: {:?}", + target.temporary_mutations + ); + assert!(target.consumed_physical_bytes > 0); + store.pools[1] + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("the resumed decommission should migrate the source object"); + let source = store.pools[0] + .get_object_info(bucket, object, &ObjectOptions::default()) + .await; + assert!( + source + .as_ref() + .err() + .is_some_and(|err| crate::error::is_err_object_not_found(err) || crate::error::is_err_version_not_found(err)), + "the source copy should be cleaned after migration: {source:?}" + ); + } + + async fn resumed_decommission_settles_a_killed_leader_put_case(committed: bool, legacy: bool) { + let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await; + let _proof_guard = legacy.then(crate::services::notification_sys::without_decommission_target_fence_fleet_proof_for_test); + let object = "killed-leader.bin"; + let (bucket, interrupted) = kill_leader_inside_target_put(&store, committed, object).await; + + // The restarted leader resumes the same decommission. Its retry of the + // interrupted object either finds the committed copy and reconciles it + // or rewrites it, and either way settles the adopted marker. + run_resumed_decommission_routine(&store).await; + assert_resumed_decommission_settled(&store, &bucket, object, interrupted).await; + } + + #[test] + #[serial_test::serial] + fn resumed_decommission_settles_a_leader_killed_after_the_target_commit() { + run_large_stack_async_test("resumed-decommission-killed-after-commit", || { + resumed_decommission_settles_a_killed_leader_put_case(true, false) + }); + } + + #[test] + #[serial_test::serial] + fn resumed_decommission_settles_a_leader_killed_before_the_target_commit() { + run_large_stack_async_test("resumed-decommission-killed-before-commit", || { + resumed_decommission_settles_a_killed_leader_put_case(false, false) + }); + } + + #[test] + #[serial_test::serial] + fn resumed_legacy_decommission_settles_a_leader_killed_after_the_target_commit() { + run_large_stack_async_test("resumed-legacy-decommission-killed-after-commit", || { + resumed_decommission_settles_a_killed_leader_put_case(true, true) + }); + } + + #[test] + #[serial_test::serial] + fn resumed_legacy_decommission_settles_a_leader_killed_before_the_target_commit() { + run_large_stack_async_test("resumed-legacy-decommission-killed-before-commit", || { + resumed_decommission_settles_a_killed_leader_put_case(false, true) + }); + } + + /// Move part of one target's remaining allocation to another target and + /// persist it, so that the next selection by remaining allocation alone + /// would prefer `to`. + async fn shift_decommission_target_allocation(store: &Arc, from: usize, to: 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"); + for target in &mut reservation.targets { + if target.pool_index == from { + target.reserved_physical_bytes -= bytes; + } else if target.pool_index == to { + target.reserved_physical_bytes += bytes; + } + } + meta.save(store.pools.clone()) + .await + .expect("persist the shifted target allocation"); + *store.pool_meta.write().await = meta; + } + + async fn resumed_decommission_returns_an_uncommitted_retry_to_its_adopted_target_case(legacy: bool) { + let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_three_pool_stores(None).await; + let _proof_guard = legacy.then(crate::services::notification_sys::without_decommission_target_fence_fleet_proof_for_test); + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + // Target 1 holds most of the allocation when the leader is killed + // before its target commit there. + let object = "sticky-adopted-target.bin"; + let (bucket, interrupted) = kill_leader_inside_target_put_with_capacity( + &store, + false, + object, + 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), + ], + ) + .await; + // Before the resume, target 2 has the larger remaining allocation. + shift_decommission_target_allocation(&store, 1, 2, 80 << 20).await; + assert!( + killed_leader_target(&store, 2).await.reserved_physical_bytes + > killed_leader_target(&store, 1).await.reserved_physical_bytes + ); + + run_resumed_decommission_routine(&store).await; + 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, + "the resumed decommission must complete: complete={}, failed={}, canceled={}, blocked_reason={:?}", + info.complete, + info.failed, + info.canceled, + info.capacity_blocked_reason + ); + assert_eq!(info.items_decommission_failed, 0); + drop(local); + let adopted = crate::core::pools::decommission_adopted_intent_mutation_id(interrupted); + for target_pool_index in [1, 2] { + let target = killed_leader_target(&store, target_pool_index).await; + assert_ne!(target.pending_mutation_id, Some(interrupted)); + assert!( + target + .temporary_mutations + .iter() + .all(|mutation| mutation.mutation_id != interrupted && mutation.mutation_id != adopted), + "target pool {target_pool_index} must not keep a marker of the interrupted mutation: {:?}", + target.temporary_mutations + ); + } + assert!( + killed_leader_target(&store, 1).await.consumed_physical_bytes > 0, + "the retry's commit must be charged on the target that held its adopted completion" + ); + store.pools[1] + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await + .expect("the retry must commit on the target that holds its adopted completion"); + let source = store.pools[0] + .get_object_info(&bucket, object, &ObjectOptions::default()) + .await; + assert!( + source + .as_ref() + .err() + .is_some_and(|err| crate::error::is_err_object_not_found(err) || crate::error::is_err_version_not_found(err)), + "the source copy should be cleaned after migration: {source:?}" + ); + } + + #[test] + #[serial_test::serial] + fn resumed_decommission_returns_an_uncommitted_retry_to_its_adopted_target() { + run_large_stack_async_test("resumed-decommission-sticky-adopted-target", || { + resumed_decommission_returns_an_uncommitted_retry_to_its_adopted_target_case(false) + }); + } + + #[test] + #[serial_test::serial] + fn resumed_legacy_decommission_returns_an_uncommitted_retry_to_its_adopted_target() { + run_large_stack_async_test("resumed-legacy-decommission-sticky-adopted-target", || { + resumed_decommission_returns_an_uncommitted_retry_to_its_adopted_target_case(true) + }); + } + + #[test] + #[serial_test::serial] + fn failed_intent_adoption_pauses_and_the_next_incarnation_completes() { + run_large_stack_async_test("failed-intent-adoption-pauses", || async { + let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(None).await; + let object = "adoption-failure.bin"; + let (bucket, interrupted) = kill_leader_inside_target_put(&store, true, object).await; + + crate::core::pools::fail_next_decommission_intent_adoption_for_test(store.id); + run_resumed_decommission_routine(&store).await; + 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, + "an adoption failure must pause the routine, not end the decommission: complete={}, failed={}, canceled={}", + info.complete, + info.failed, + info.canceled + ); + assert!( + info.capacity_blocked_reason + .as_deref() + .is_some_and(|reason| reason.contains("failed to adopt target capacity intents")), + "the pause must name the adoption failure: {:?}", + info.capacity_blocked_reason + ); + drop(local); + assert_eq!(killed_leader_target(&store, 1).await.pending_mutation_id, Some(interrupted)); + + run_resumed_decommission_routine(&store).await; + assert_resumed_decommission_settled(&store, &bucket, object, interrupted).await; + }); + } + + #[test] + #[serial_test::serial] + fn adopted_multipart_completion_survives_a_background_abort_until_reconciled() { + run_large_stack_async_test( + "adopted-multipart-completion-background-abort", + adopted_multipart_completion_survives_a_background_abort_until_reconciled_case, + ); + } + + async fn adopted_multipart_completion_survives_a_background_abort_until_reconciled_case() { + let (_temp_dirs, store, _other_store) = + test_three_pool_stores_with_three_disk_sets_with_isolated_node_contexts(None).await; + let bucket = test_bucket("adopted-mpu"); + let object = "adopted-multipart-completion.bin"; + let first_part = vec![0x61; 5 * 1024 * 1024]; + let second_part = vec![0x62; 1]; + let body_len = first_part.len() + second_part.len(); + let source_version = uuid::Uuid::new_v4().to_string(); + let source_mod_time = time::OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(43); + + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create the adopted multipart bucket"); + let incarnation = store + .bucket_incarnation_id(&bucket) + .await + .expect("load the adopted multipart bucket incarnation"); + let source_opts = ObjectOptions { + versioned: true, + version_id: Some(source_version.clone()), + mod_time: Some(source_mod_time), + expected_bucket_incarnation_id: Some(incarnation), + ..Default::default() + }; + let source_upload = new_multipart_upload(&store, 0, &bucket, object, source_opts.clone()) + .await + .expect("create the source multipart upload"); + let source_multipart_opts = multipart_options(&store, &bucket, source_opts).await; + let mut source_parts = Vec::with_capacity(2); + for (part_num, part_body) in [(1, first_part.as_slice()), (2, second_part.as_slice())] { + let mut part_reader = PutObjReader::from_vec(part_body.to_vec()); + let part = store.pools[0] + .put_object_part( + &bucket, + object, + &source_upload.upload_id, + part_num, + &mut part_reader, + &source_multipart_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_multipart_opts) + .await + .expect("complete the source multipart object"); + assert!(source_info.is_multipart(), "the source must take the multipart migration path"); + + let layout = DecommissionErasureLayout { data: 1, parity: 0 }; + let target_total = body_len.saturating_mul(4); + set_decommission_capacity_info_overrides_for_test( + store.id, + vec![vec![ + DecommissionPoolCapacityInfo::for_test(0, layout, 0, body_len, body_len), + DecommissionPoolCapacityInfo::for_test(1, layout, 0, target_total, target_total), + DecommissionPoolCapacityInfo::for_test(2, layout, target_total, target_total, 0), + ]], + ); + store + .save_current_pool_meta_for_decommission_start(&[0], Vec::new()) + .await + .expect("activate the multipart capacity reservation"); + let owner = decommission_capacity_owner(&*store.pool_meta.read().await); + let source_reader = store.pools[0] + .get_object_reader( + &bucket, + object, + None, + HeaderMap::new(), + &ObjectOptions { + versioned: true, + version_id: Some(source_version.clone()), + no_lock: true, + data_movement: true, + raw_data_movement_read: true, + ..Default::default() + }, + ) + .await + .expect("read the source multipart object for migration"); + + // The leader dies after the completion published the target and before + // the completion's capacity finalize. + let barrier = MultipartCommitBarrier::install(&bucket, object, MultipartCommitPause::AfterObjectPublication); + let mut migration = tokio::spawn({ + let migration_store = Arc::clone(&store); + let migration_bucket = bucket.clone(); + async move { + data_movement::migrate_decommission_object( + migration_store, + 0, + migration_bucket, + source_reader, + Some(incarnation), + "adopted_multipart_completion", + Some(owner), + ) + .await + } + }); + tokio::select! { + () = barrier.wait_until_paused() => {} + result = &mut migration => { + panic!("the multipart migration finished before its completion could be interrupted: {result:?}"); + } + } + migration.abort(); + let _ = migration.await; + barrier.release(); + drop(barrier); + + let interrupted = decommission_capacity_mutation_id( + owner, + &bucket, + object, + Some(source_version.as_str()), + false, + Some(source_mod_time), + ); + let leftover = killed_leader_target(&store, 2).await; + assert_eq!(leftover.pending_mutation_id, Some(interrupted)); + let completion_bytes = leftover.pending_physical_bytes; + assert!(completion_bytes > 0); + + store + .adopt_previous_incarnation_decommission_capacity_intents_for_test(0) + .await + .expect("the resumed routine should adopt the interrupted completion"); + let adopted = crate::core::pools::decommission_adopted_intent_mutation_id(interrupted); + let adopted_marker = killed_leader_target(&store, 2) + .await + .temporary_mutations + .iter() + .find(|mutation| mutation.mutation_id == adopted) + .map(|mutation| mutation.physical_bytes) + .expect("the interrupted completion should become its adopted-completion marker"); + assert!(adopted_marker >= completion_bytes); + + // The failed attempt's cleanup runs after the resume and finds the + // published target instead of a staged upload. + let mut cleanup_opts = ObjectOptions { + data_movement: true, + src_pool_idx: 0, + versioned: true, + version_id: Some(source_version.clone()), + mod_time: Some(source_mod_time), + expected_bucket_incarnation_id: Some(incarnation), + ..Default::default() + }; + let interrupted_owner = owner.with_mutation_id(interrupted); + interrupted_owner.apply_to(&mut cleanup_opts); + let multipart_fence = store + .acquire_decommission_multipart_mutation_fence(interrupted_owner) + .await + .expect("fence the background multipart cleanup"); + multipart_fence.add_namespace_lock_fence(&mut cleanup_opts); + store + .reconcile_multipart_uploads_for_data_movement( + 2, + &bucket, + object, + &data_movement::data_movement_upload_identity_from_options(&cleanup_opts), + &cleanup_opts, + ) + .await + .expect("the background cleanup should prove the published target"); + drop(multipart_fence); + assert_eq!( + killed_leader_target(&store, 2) + .await + .temporary_mutations + .iter() + .find(|mutation| mutation.mutation_id == adopted) + .map(|mutation| mutation.physical_bytes), + Some(adopted_marker), + "a cleanup that proves the committed target must not release the adopted completion" + ); + + // The interrupted mutation's retry finds the published target. + store + .reconcile_decommission_capacity_after_equivalent_target(interrupted_owner, 2, body_len) + .await + .expect("the equivalent target settles the adopted completion"); + let settled = killed_leader_target(&store, 2).await; + assert_eq!(settled.consumed_physical_bytes, completion_bytes); + assert!( + settled + .temporary_mutations + .iter() + .all(|mutation| mutation.mutation_id != interrupted && mutation.mutation_id != adopted) + ); + assert_eq!(settled.inflight_physical_bytes, 0); + } } diff --git a/crates/ecstore/src/services/rebalance/mod.rs b/crates/ecstore/src/services/rebalance/mod.rs index 1c32c07b3..e7fb7234f 100644 --- a/crates/ecstore/src/services/rebalance/mod.rs +++ b/crates/ecstore/src/services/rebalance/mod.rs @@ -100,6 +100,17 @@ pub(crate) async fn test_two_pool_stores( test_pool_stores_with_contexts(rebalance_meta, false, 2, 2).await } +#[cfg(all(test, feature = "test-util"))] +pub(crate) async fn test_three_pool_stores( + rebalance_meta: Option, +) -> ( + Vec, + std::sync::Arc, + std::sync::Arc, +) { + test_pool_stores_with_contexts(rebalance_meta, false, 3, 2).await +} + #[cfg(test)] pub(crate) async fn test_two_pool_stores_with_isolated_node_contexts( rebalance_meta: Option, diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 3786aa60d..c3965ae89 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -8395,6 +8395,307 @@ mod tests { shutdown.cancel(); } + /// Seed a versioned key with one data version and a newer delete marker on + /// the source pool, the shape whose delete marker takes the copy path. + async fn seed_decommission_delete_marker_key( + store: &Arc, + bucket: &str, + object: &str, + ) -> (uuid::Uuid, uuid::Uuid) { + store + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("create the delete marker bucket"); + let data_version = uuid::Uuid::new_v4(); + let marker_version = uuid::Uuid::new_v4(); + let base_time = OffsetDateTime::now_utc(); + seed_decommission_source( + store, + bucket, + object, + b"delete marker data".to_vec(), + &ObjectOptions { + versioned: true, + version_id: Some(data_version.to_string()), + mod_time: Some(base_time), + ..Default::default() + }, + ) + .await; + store.pools[0] + .delete_object( + bucket, + object, + ObjectOptions { + versioned: true, + version_id: Some(marker_version.to_string()), + delete_marker: true, + mod_time: Some(base_time + time::Duration::seconds(1)), + ..Default::default() + }, + ) + .await + .expect("seed source delete marker"); + (data_version, marker_version) + } + + async fn assert_decommission_delete_marker_key_migrated( + store: &Arc, + bucket: &str, + object: &str, + data_version: uuid::Uuid, + marker_version: uuid::Uuid, + ) { + let marker_opts = ObjectOptions { + versioned: true, + version_id: Some(marker_version.to_string()), + ..Default::default() + }; + let data_opts = ObjectOptions { + versioned: true, + version_id: Some(data_version.to_string()), + ..Default::default() + }; + let target_marker = store.pools[1] + .get_object_info(bucket, object, &marker_opts) + .await + .expect("target delete marker must exist"); + assert!(target_marker.delete_marker); + assert_eq!( + read_decommission_target_body(store, bucket, object, &data_opts).await, + b"delete marker data" + ); + assert_decommission_source_absent(store, bucket, object, &marker_opts).await; + assert_decommission_source_absent(store, bucket, object, &data_opts).await; + } + + /// The Linux CI run of the S00 matrix: without any kill, the delete marker + /// copy found the target capacity gate busy. It must wait for the gate like + /// an ordinary object copy and complete once the gate is released. + #[test] + #[serial_test::serial(storage_class_env)] + fn decommission_delete_marker_copy_completes_after_waiting_for_a_busy_target_gate() { + let handle = std::thread::Builder::new() + .name("decommission_delete_marker_copy_with_busy_target_gate".to_string()) + .stack_size(32 * 1024 * 1024) + .spawn(|| { + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .worker_threads(2) + .build() + .expect("test runtime should build"); + runtime.block_on(async { + let temp_dir = tempfile::tempdir().expect("create delete marker gate contention store dir"); + let (_ctx, store, shutdown) = without_storage_class_env(build_isolated_test_store( + temp_dir.path(), + "decommission-delete-marker-gate-contention", + &[4, 4], + )) + .await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + + let bucket = format!("decom-marker-gate-{}", uuid::Uuid::new_v4()); + let object = "marker.bin"; + let (data_version, marker_version) = seed_decommission_delete_marker_key(&store, &bucket, object).await; + mark_test_pool_decommissioning(&store, 0).await; + let incarnation = Some(store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation")); + + // Another writer holds the target pool's capacity gate. + use crate::storage_api_contracts::namespace::NamespaceLocking as _; + let target_lock = store + .new_ns_lock( + crate::disk::RUSTFS_META_BUCKET, + &format!("{}/1", crate::core::pools::DECOMMISSION_CAPACITY_TARGET_LOCK_PREFIX), + ) + .await + .expect("create the target capacity gate probe"); + let target_guard = target_lock + .get_write_lock(std::time::Duration::from_secs(30)) + .await + .expect("hold the target capacity gate"); + + let entry = tokio::spawn({ + let store = Arc::clone(&store); + let bucket = bucket.clone(); + async move { + run_decommission_entry_retry_test( + &store, + CancellationToken::new(), + &bucket, + object, + incarnation, + Arc::new(AtomicUsize::new(0)), + ) + .await + } + }); + tokio::time::sleep(std::time::Duration::from_millis(1500)).await; + assert!(!entry.is_finished(), "the entry must wait while the target gate is held"); + drop(target_guard); + entry + .await + .expect("the entry should join") + .expect("the delete marker copy must complete once the target gate is released"); + + assert_decommission_delete_marker_key_migrated(&store, &bucket, object, data_version, marker_version).await; + shutdown.cancel(); + }); + }) + .expect("spawn delete marker gate contention test thread"); + if let Err(payload) = handle.join() { + std::panic::resume_unwind(payload); + } + } + + /// The S00 leader-kill run: both versions of a key were committed and + /// charged on the target, the source cleanup never ran, and on the retry a + /// concurrent mutation holds the target's pending slot between a failed + /// write and its own retry. The delete marker copy must not fail the entry + /// on that slot, and the entry completes once the slot is released. + #[test] + #[serial_test::serial(storage_class_env)] + fn decommission_delete_marker_copy_completes_while_another_mutation_holds_the_target_slot() { + let handle = std::thread::Builder::new() + .name("decommission_delete_marker_copy_with_foreign_pending_slot".to_string()) + .stack_size(32 * 1024 * 1024) + .spawn(|| { + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .worker_threads(2) + .build() + .expect("test runtime should build"); + runtime.block_on(async { + let temp_dir = tempfile::tempdir().expect("create delete marker slot contention store dir"); + let (_ctx, store, shutdown) = without_storage_class_env(build_isolated_test_store( + temp_dir.path(), + "decommission-delete-marker-slot-contention", + &[4, 4], + )) + .await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + + let bucket = format!("decom-marker-slot-{}", uuid::Uuid::new_v4()); + let object = "marker.bin"; + let (data_version, marker_version) = seed_decommission_delete_marker_key(&store, &bucket, object).await; + mark_test_pool_decommissioning(&store, 0).await; + let incarnation = Some(store.bucket_incarnation_id(&bucket).await.expect("bucket incarnation")); + + // The first pass commits and charges both versions on the + // target, then loses its fence before the source cleanup. + let loss_hook = crate::store::object::DecommissionMutationFenceLossHook::install( + &bucket, + object, + crate::store::object::DecommissionMutationFenceTestPhase::SourceCleanup, + ); + let barrier = crate::data_movement::SourceCleanupDeleteBarrier::install(&bucket, object); + let first_pass = tokio::spawn({ + let store = Arc::clone(&store); + let bucket = bucket.clone(); + async move { + run_decommission_entry_retry_test( + &store, + CancellationToken::new(), + &bucket, + object, + incarnation, + Arc::new(AtomicUsize::new(0)), + ) + .await + } + }); + barrier.wait_until_paused().await; + loss_hook.mark_lost(); + barrier.release(); + drop(barrier); + let _ = first_pass.await.expect("the first pass should join"); + drop(loss_hook); + let data_opts = ObjectOptions { + versioned: true, + version_id: Some(data_version.to_string()), + ..Default::default() + }; + store.pools[0] + .get_object_info(&bucket, object, &data_opts) + .await + .expect("the lost source cleanup must retain the source"); + + // Another mutation persisted its intent on the same target, + // failed, and has not retried yet. + let owner = { + let pool_meta = store.pool_meta.read().await; + let reservation = pool_meta.pools[0] + .decommission + .as_ref() + .and_then(|info| info.capacity_reservation.as_ref()) + .expect("the decommission reservation should be active"); + crate::core::pools::DecommissionCapacityOwner { + source_pool_index: 0, + operation_id: reservation.operation_id, + generation: reservation.generation, + owner_nonce: reservation.owner_nonce, + mutation_id: Some(uuid::Uuid::new_v4()), + } + }; + store + .run_decommission_capacity_admitted_mutation_with_capacity_lease(1, Some(owner), Some(8), |_| async { + Err::<(), _>(StorageError::other("the competing write failed after persisting its intent")) + }) + .await + .expect_err("the competing mutation must leave its intent pending"); + + let retry = tokio::spawn({ + let store = Arc::clone(&store); + let bucket = bucket.clone(); + async move { + run_decommission_entry_retry_test( + &store, + CancellationToken::new(), + &bucket, + object, + incarnation, + Arc::new(AtomicUsize::new(0)), + ) + .await + } + }); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + // The competing mutation retries and finalizes its intent, + // waiting out the entry's own short target gate holds. + let mut finalize_attempt = 0; + loop { + match store + .run_decommission_capacity_admitted_mutation_with_capacity_lease(1, Some(owner), Some(8), |_| async { + Ok(()) + }) + .await + { + Ok(()) => break, + Err(err) + if crate::core::pools::is_decommission_capacity_target_gate_busy(&err) + && finalize_attempt < 50 => + { + finalize_attempt += 1; + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + Err(err) => panic!("the competing mutation should finalize its own intent: {err:?}"), + } + } + retry + .await + .expect("the retried entry should join") + .expect("the entry must complete instead of failing on another mutation's pending slot"); + + assert_decommission_delete_marker_key_migrated(&store, &bucket, object, data_version, marker_version).await; + + shutdown.cancel(); + }); + }) + .expect("spawn delete marker slot contention test thread"); + if let Err(payload) = handle.join() { + std::panic::resume_unwind(payload); + } + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[serial_test::serial(storage_class_env)] async fn reverse_decommission_reuses_fixed_target_fence_for_put_and_multipart() { diff --git a/crates/scanner/src/scanner/backlog.rs b/crates/scanner/src/scanner/backlog.rs index e63bd77dd..6722c7a22 100644 --- a/crates/scanner/src/scanner/backlog.rs +++ b/crates/scanner/src/scanner/backlog.rs @@ -3769,6 +3769,52 @@ mod tests { }); } + #[test] + #[serial_test::serial] + fn native_retirement_waits_for_an_adopted_target_intent() { + run_native_retirement_test(async || { + let (_root, store) = native_retirement_store().await; + let seeded = ScannerPauseBacklogController::claim(Arc::clone(&store), unix_now()) + .await + .expect("seed the real native record on every set"); + drop(seeded); + let (source_bytes, _) = native_replica_bytes(&store.pools[0].disk_set[0]).await; + store + .prepare_scanner_pause_backlog_retirement_for_test(0, source_bytes.len() * 2) + .await + .expect("activate native retirement"); + store + .stage_scanner_pause_backlog_retirement_intent_for_test(0, 0) + .await + .expect("retain a real target intent after failure in the admitted operation"); + // A resumed decommission routine moves the leftover intent into + // the mutation's adopted marker. + store + .adopt_previous_incarnation_decommission_capacity_intents_for_test(0) + .await + .expect("adopt the leftover retirement intent"); + { + let meta = store.pool_meta.read().await; + let reservation = meta.pools[0] + .decommission + .as_ref() + .unwrap() + .capacity_reservation + .as_ref() + .unwrap(); + assert_eq!(reservation.pending_target_physical_bytes, 0); + assert!(reservation.inflight_target_physical_bytes > 0); + } + let blocked = store + .retire_scanner_pause_backlog_for_test(0, 0) + .await + .expect_err("an adopted target intent still blocks retirement"); + assert!(blocked.to_string().contains("unresolved target capacity intent"), "{blocked}"); + assert_eq!(native_replica_bytes(&store.pools[0].disk_set[0]).await.0, source_bytes); + shutdown_native_retirement_store(store).await; + }); + } + fn observation(now: u64, paused: bool, pending: u64) -> ScannerPauseBacklogObservation { ScannerPauseBacklogObservation { now_unix_secs: now, diff --git a/scripts/error-other-format-baseline.txt b/scripts/error-other-format-baseline.txt index 2151ae440..4d077ca42 100644 --- a/scripts/error-other-format-baseline.txt +++ b/scripts/error-other-format-baseline.txt @@ -24,7 +24,7 @@ 10|crates/ecstore/src/cluster/rpc/remote_disk.rs 6|crates/ecstore/src/config/com.rs 14|crates/ecstore/src/config/storageclass.rs -178|crates/ecstore/src/core/pools.rs +177|crates/ecstore/src/core/pools.rs 6|crates/ecstore/src/data_movement/mod.rs 2|crates/ecstore/src/data_usage/local_snapshot.rs 12|crates/ecstore/src/data_usage/mod.rs