gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-27 06:18:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(decommission): adopt target intents left by a killed leader (#8710)

Browse Source 
* 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
This commit is contained in:
Chris
2026-10-20 11:28:22 +09:00
committed by GitHub
parent 10a9b158d7
commit ea596b2e2d
6 changed files with 2885 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/core/pools.rs
+1920 -91
View File
File diff suppressed because it is too large Load Diff
crates/ecstore/src/core/pools_test.rs
+606
View File
@@ -7749,4 +7749,610 @@ mod decommission_lock_order_tests {
.await;
});
}
async fn killed_leader_target(
store: &Arc<crate::store::ECStore>,
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<crate::store::ECStore>) {
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<crate::store::ECStore>,
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<crate::store::ECStore>,
committed: bool,
object: &str,
capacity: Vec<DecommissionPoolCapacityInfo>,
) -> (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<crate::store::ECStore>,
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<crate::store::ECStore>, 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);
}
}
crates/ecstore/src/services/rebalance/mod.rs
+11
View File
@@ -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<RebalanceMeta>,
) -> (
Vec<tempfile::TempDir>,
std::sync::Arc<crate::store::ECStore>,
std::sync::Arc<crate::store::ECStore>,
) {
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<RebalanceMeta>,
crates/ecstore/src/store/init.rs
+301
View File
@@ -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<crate::store::ECStore>,
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<crate::store::ECStore>,
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() {
crates/scanner/src/scanner/backlog.rs
+46
View File
@@ -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,
scripts/error-other-format-baseline.txt
+1 -1
View File
@@ -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