From ea596b2e2d5ab80dcf6b5c8054918c5aa474a6ff Mon Sep 17 00:00:00 2001 From: Chris Date: Tue, 20 Oct 2026 11:28:22 +0900 Subject: [PATCH] fix(decommission): adopt target intents left by a killed leader (#8710) * fix(decommission): adopt target intents left by a killed leader A decommission leader killed after persisting a target capacity intent and before finalizing it left that intent in the exclusive pending slot of the still-active reservation. The resumed routine derives the same reservation identity, but every other mutation on that target was rejected as "another mutation" until the interrupted object happened to be retried on the same target, so the resumed decommission exhausted its retries and stalled. At routine start, before any worker runs, the resumed routine now adopts the reservation's pending intents: under the target fence (or the legacy pool metadata fence) it moves each one into a mutation-scoped inflight marker that keeps its identity and bytes. The interrupted mutation's retry settles the marker through the existing commit, temporary-release and equivalent-target paths, and an equivalent committed target now settles a mutation's own marker when no pending intent remains. Adoption is pinned to the persisted decommission start time and reservation identity, never runs mid-incarnation, and a failure pauses the routine instead of marking the decommission failed. * fix(decommission): keep adopted intents within budget and settle them once Address the adversarial review of the leftover intent adoption. - Merge an adopted intent into its mutation's staged marker by summing only while the target's temporary-copy budget allows, never below the larger of the two. A multipart completion moves its staged parts into the object, so summing them could exceed the budget and pause every later routine start; the adopted marker now exceeds the budget only where the completion's own finalize would have. Reuse resolve and record_inflight for the move. - Distinguish a temporary release that proved the exact committed target from one that could not prove the target identity. When the committed mutation no longer has a pending intent, release only the bytes the cleanup freed and keep the adopted completion for the equivalent-target reconciliation. - Let an equivalent target settle a mutation's own marker while another mutation holds the pending slot, instead of failing on that unrelated intent. - Cover real leader kills before and after the target commit at routine level for both lock models, an injected adoption failure that pauses and then completes, an adopted multipart completion surviving a background abort, and a live legacy intent. Routine tests reuse the large-stack test helper. * fix(decommission): pause instead of failing on target slot contention The S00 leader-kill run with the null-version fix merged ended the resumed decommission as failed. A re-run of a versioned key reconciled its delete marker against an already charged target copy while another worker of the same incarnation held the target's pending slot between a failed write and its retry. The reconciliation rejected that unrelated intent, the entry context wrapper flattened the error to a string the intent-conflict retry could not recognize, and the bucket runner reported a sibling bucket's OperationCanceled instead of the capacity block, so the routine marked the decommission failed. - An equivalent-target reconciliation, including the temporary checkpoint variant, no longer depends on another mutation's pending intent: a target holds one pending intent at a time, so this mutation has none to settle. - The entry context keeps the typed cause, and the capacity-block and intent-conflict classifiers also recognize the rendered message that an error clone keeps. - Concurrent bucket and listing failures keep the root cause over sibling cancellations, so a capacity block pauses the routine. * test(decommission): delete marker copy waits for a busy target gate The Linux run of the S00 matrix showed the delete marker copy meeting a busy target capacity gate without any leader kill. The copy already waits on that gate like an ordinary object copy; pin it with an entry-level test that holds the gate, checks the entry keeps waiting, and expects the key to migrate once the gate is released. Share the versioned delete marker fixture with the foreign pending slot test. * fix(decommission): keep an adopted completion unless the target is absent Close the remaining path where an adopted completion could lose its record before being charged. A multipart cleanup that found a target it could not prove as the exact owned copy still dropped the whole marker of a mutation without a pending intent, after which the equivalent-target reconciliation would treat the commit as already charged. Any cleanup that did not prove the exact target absent now releases only the bytes it freed when the mutation has no pending intent, and leaves the rest for the reconciliation. This makes the three-state release unnecessary, so the release proof goes back to the existing clear-pending flag. * chore(ci): shrink the error format ratchet baseline for pools.rs 9eae47ce8 removed one Error::other(format!) call site from crates/ecstore/src/core/pools.rs, so the ratchet requires the baseline to drop from 178 to 177. * fix(decommission): keep adopted completions in their own capacity marker An adopted completion shared the mutation's staged marker, so whether it survived a cleanup depended on the pool-wide free-space delta, and a cleanup that freed little space left staged bytes behind and repeated forever. Record each adopted intent under a mutation id derived from the interrupted mutation (decommission_adopted_intent_mutation_id) as a separate temporaryMutations entry. A commit or an equivalent-target reconciliation settles both markers, a temporary cleanup releases the staged marker as before, and only a cleanup that proved the exact target absent drops the adopted one. While the target holds only that adopted completion, a cleanup keeps the delayed-release exemption of the published pending intent it replaced. Older nodes see an ordinary temporary mutation they never release, which only over-reserves. Also log capacity_intents_adopted as soon as the adopted ledger is durable, merge the root-cause helpers into decommission_root_cause, drop the decommission_capacity_blocked: prefix fallback, and log the already-charged equivalent-target branch at debug level. Refs rustfs/backlog#2873, rustfs/backlog#2872 * fix(decommission): return uncommitted retries to their adopted target A retry that had not committed yet went to the reserved target with the most remaining allocation. When that was not the target holding the mutation's adopted completion, the commit settled only its own target and the adopted marker stayed inflight until the reservation was released. Prefer a reserved target that holds the mutation's adopted marker while it still fits the object (decommission_reserved_target_choice). Also cover the remaining settlement paths of the derived marker (non-growing replacement, equivalent temporary reconciliation and its discovery gate, scanner backlog retirement), repeated adoption of one mutation and the zero-byte staging record in both lock models, drop two conditions that always equal another one, and correct the adopted-id doc comment about nodes without adoption. Refs rustfs/backlog#2873, rustfs/backlog#2872 * fix(decommission): count a completion adopted twice only once A mutation killed twice leaves an adopted completion and then a new intent for the same object version. Adoption added the two, so an 11-byte object counted 22 inflight bytes and could trip the capacity check once later commits shrank the budget. Both reserve the same completion, so merge them by their maximum before adding staged parts; a single adoption is unchanged. Also assert in the three-pool sticky test that no entry failed, the retry is charged on its adopted target and the source is cleaned, run it for the legacy lock model too, stop saying operators rely on the adoption log before rollback, and document that the selection fallback cannot help an adopted completion. Refs rustfs/backlog#2873, rustfs/backlog#2872 * fix(decommission): log where an adopted retry is placed An e2e run could not show whether a resumed retry went back to the target holding its adopted completion. Target selection now logs adopted_target_reused when it does, and adopted_target_bypassed with the adopted target pools when the object no longer fits there and the adopted marker stays behind. Both carry the mutation id and the source and target pools. Refs rustfs/backlog#2873, rustfs/backlog#2872 --- crates/ecstore/src/core/pools.rs | 2011 +++++++++++++++++- crates/ecstore/src/core/pools_test.rs | 606 ++++++ crates/ecstore/src/services/rebalance/mod.rs | 11 + crates/ecstore/src/store/init.rs | 301 +++ crates/scanner/src/scanner/backlog.rs | 46 + scripts/error-other-format-baseline.txt | 2 +- 6 files changed, 2885 insertions(+), 92 deletions(-) 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