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

fix(decommission): scope temporary-copy budget checks to the growth pool (#8721)

Browse Source
This commit is contained in:
Chris
2026-10-20 13:28:51 +09:00
committed by GitHub
parent da7fde0e62
commit e7e53f07b2
2 changed files with 1029 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/core/pools.rs
+701 -75
View File
File diff suppressed because it is too large Load Diff
crates/ecstore/src/core/pools_test.rs
+328 -1
View File
@@ -224,7 +224,7 @@ mod decommission_lock_order_tests {
use std::future::Future;
use std::sync::{
Arc, Condvar, Mutex as StdMutex,
atomic::{AtomicUsize, Ordering},
atomic::{AtomicBool, AtomicUsize, Ordering},
};
use std::time::Duration;
use tokio::io::AsyncReadExt;
@@ -507,6 +507,333 @@ mod decommission_lock_order_tests {
assert_lock_order_with_tail(mutation, false).await;
}
/// Leave inflight bytes on the active decommission's allocation for
/// `target_pool_index` that no mutation will settle, as a lost marker
/// would, and persist them.
async fn strand_decommission_target_marker(store: &Arc<crate::store::ECStore>, target_pool_index: usize, bytes: usize) {
let mut meta = store.pool_meta.read().await.clone();
let reservation = meta.pools[0]
.decommission
.as_mut()
.and_then(|info| info.capacity_reservation.as_mut())
.expect("the decommission reservation should be active");
let target = reservation
.targets
.iter_mut()
.find(|target| target.pool_index == target_pool_index)
.expect("the reservation should allocate the target pool");
target
.temporary_mutations
.push(crate::core::pools::DecommissionCapacityTemporaryMutation {
mutation_id: uuid::Uuid::new_v4(),
physical_bytes: bytes,
});
target.inflight_physical_bytes += bytes;
target.observed_physical_bytes += bytes;
reservation.inflight_target_physical_bytes += bytes;
reservation.observed_target_physical_bytes += bytes;
meta.save(store.pools.clone())
.await
.expect("persist the stranded target marker");
*store.pool_meta.write().await = meta;
}
/// Start a decommission of `bucket` on a three-pool store where target 1
/// holds most of the allocation and `stranded` inflight bytes of a marker
/// no mutation will settle.
async fn start_over_budget_routine_decommission(store: &Arc<crate::store::ECStore>, bucket: &str, stranded: usize) {
// Of the 128 MiB peak, target 1 takes 100 MiB and target 2 the other
// 28 MiB; target 1's temporary-copy budget is 50 MiB.
let layout = DecommissionErasureLayout { data: 1, parity: 0 };
set_decommission_capacity_info_overrides_for_test(
store.id,
vec![vec![
DecommissionPoolCapacityInfo::for_test(0, layout, 0, 64 << 20, 64 << 20),
DecommissionPoolCapacityInfo::for_test(1, layout, 100 << 20, 100 << 20, 0),
DecommissionPoolCapacityInfo::for_test(2, layout, 1 << 30, 1 << 30, 0),
]],
);
store
.save_current_pool_meta_for_decommission_start(
&[0],
vec![crate::core::pools::DecomBucketInfo {
name: bucket.to_string(),
prefix: String::new(),
}],
)
.await
.expect("activate the decommission");
strand_decommission_target_marker(store, 1, stranded).await;
}
/// Pause one of `object`'s own target-1 gate acquisitions, after `skips`
/// of them, by arming the pause when the routine reads that object.
fn pause_object_target_gate_acquire(
barrier: Arc<DecommissionCapacityLockOrderBarrier>,
bucket: String,
object: String,
skips: usize,
) -> DecommissionTestFaultGuard {
let armed = Arc::new(AtomicBool::new(false));
DecommissionTestFaultGuard::install(Arc::new(move |stage, entry_bucket, entry_object, _, _| {
if stage == "object_read" && entry_bucket == bucket && entry_object == object && !armed.swap(true, Ordering::AcqRel) {
barrier.pause_one_target_gate_acquire_after(1, skips);
}
false
}))
}
/// Another mutation commits `bytes` on target 1, as a concurrent entry
/// would.
async fn commit_other_mutation_on_target_1(store: &Arc<crate::store::ECStore>, bytes: usize) {
let owner = decommission_capacity_owner(&*store.pool_meta.read().await).with_mutation_id(uuid::Uuid::new_v4());
store
.run_decommission_capacity_admitted_mutation(1, Some(owner), Some(bytes), || async {
Ok::<_, crate::error::Error>(())
})
.await
.expect("the other mutation commits on target 1");
}
async fn assert_target_1_over_budget(store: &Arc<crate::store::ECStore>) {
let target = killed_leader_target(store, 1).await;
let budget = target
.reserved_physical_bytes
.saturating_sub(target.consumed_physical_bytes * 2)
/ 2;
assert!(
target.inflight_physical_bytes > budget,
"target 1 must be over its {budget}-byte budget: {target:?}"
);
}
async fn run_routine_over_a_target_going_over_budget(
store: &Arc<crate::store::ECStore>,
barrier: &DecommissionCapacityLockOrderBarrier,
other_commit_bytes: usize,
before_other_commit: impl std::future::Future<Output = ()>,
) {
let routine = tokio::spawn({
let store = Arc::clone(store);
async move {
let canceler = crate::core::pools::DecommissionCanceler::new(CancellationToken::new());
store.decommission_cancelers.write().await[0] = Some(canceler.clone());
store
.do_decommission_in_routine(canceler, 0, Arc::new(tokio::sync::Semaphore::new(2)))
.await
}
});
// The object selected target 1 and waits to acquire its gate.
barrier.wait_until_target_gate_acquire_paused().await;
before_other_commit.await;
commit_other_mutation_on_target_1(store, other_commit_bytes).await;
assert_target_1_over_budget(store).await;
barrier.release_target_gate_acquire();
routine
.await
.expect("the decommission routine task should join")
.expect("the decommission routine should return");
let local = store.pool_meta.read().await;
let info = local.pools[0]
.decommission
.as_ref()
.expect("the decommission state should remain present");
assert!(
info.complete && !info.failed && !info.canceled && info.capacity_blocked_reason.is_none(),
"a target that went over budget after selection must not pause the decommission: complete={}, failed={}, canceled={}, blocked_reason={:?}",
info.complete,
info.failed,
info.canceled,
info.capacity_blocked_reason
);
assert_eq!(info.items_decommission_failed, 0);
}
#[test]
#[serial_test::serial]
fn decommission_reselects_a_target_that_went_over_budget_after_selection() {
run_large_stack_async_test("decommission-reselect-over-budget-target", || async {
let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await;
let bucket = test_bucket("reselect-over-budget");
let object = "late.bin";
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create the reselection bucket");
let mut source = PutObjReader::from_vec(b"selected target 1 before it went over budget".to_vec());
store.pools[0]
.put_object(&bucket, object, &mut source, &ObjectOptions::default())
.await
.expect("write the source object");
// 45 of target 1's 50 MiB budget are stranded.
start_over_budget_routine_decommission(&store, &bucket, 45 << 20).await;
let barrier = Arc::new(DecommissionCapacityLockOrderBarrier::install(store.id, store.id));
barrier.disable_owner_pause();
let _fault = pause_object_target_gate_acquire(Arc::clone(&barrier), bucket.clone(), object.to_string(), 0);
// A 6 MiB commit shrinks target 1's budget to 44 MiB.
run_routine_over_a_target_going_over_budget(&store, &barrier, 6 << 20, async {}).await;
store.pools[2]
.get_object_info(&bucket, object, &ObjectOptions::default())
.await
.expect("the entry must be placed on target 2 after target 1 went over budget");
assert!(
store.pools[1]
.get_object_info(&bucket, object, &ObjectOptions::default())
.await
.is_err()
);
});
}
#[test]
#[serial_test::serial]
fn decommission_restarts_a_multipart_upload_pinned_to_a_target_that_went_over_budget() {
run_large_stack_async_test("decommission-reselect-over-budget-multipart", || async {
let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await;
let bucket = test_bucket("reselect-over-budget-mpu");
let object = "late-multipart.bin";
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create the multipart reselection bucket");
let source_upload = new_multipart_upload(&store, 0, &bucket, object, ObjectOptions::default())
.await
.expect("create the source multipart upload");
let source_opts = multipart_options(&store, &bucket, ObjectOptions::default()).await;
let mut source_parts = Vec::with_capacity(2);
for (part_num, part_body) in [(1, vec![0x31; 5 << 20]), (2, vec![0x32; 1])] {
let part = store.pools[0]
.put_object_part(
&bucket,
object,
&source_upload.upload_id,
part_num,
&mut PutObjReader::from_vec(part_body),
&source_opts,
)
.await
.expect("stage the source multipart part");
source_parts.push(crate::storage_api_contracts::multipart::CompletePart {
part_num: part.part_num,
etag: part.etag,
..Default::default()
});
}
let source_info = store.pools[0]
.clone()
.complete_multipart_upload(&bucket, object, &source_upload.upload_id, source_parts, &source_opts)
.await
.expect("complete the source multipart object");
assert!(source_info.is_multipart());
// 40 of target 1's 50 MiB budget are stranded.
start_over_budget_routine_decommission(&store, &bucket, 40 << 20).await;
let barrier = Arc::new(DecommissionCapacityLockOrderBarrier::install(store.id, store.id));
barrier.disable_owner_pause();
// Skip the upload creation and part 1; pause before part 2.
let _fault = pause_object_target_gate_acquire(Arc::clone(&barrier), bucket.clone(), object.to_string(), 2);
// A 12 MiB commit shrinks target 1's budget to 38 MiB while the
// upload, pinned to target 1, waits to upload part 2.
let store_for_check = Arc::clone(&store);
run_routine_over_a_target_going_over_budget(&store, &barrier, 12 << 20, async move {
let target = killed_leader_target(&store_for_check, 1).await;
assert!(
target.temporary_mutations.len() >= 2,
"the upload must hold its staging record on target 1 next to the stranded marker: {:?}",
target.temporary_mutations
);
})
.await;
let migrated = store.pools[2]
.get_object_info(&bucket, object, &ObjectOptions::default())
.await
.expect("the restarted upload must complete on target 2");
assert!(migrated.is_multipart());
assert!(
store.pools[1]
.get_object_info(&bucket, object, &ObjectOptions::default())
.await
.is_err()
);
});
}
#[test]
#[serial_test::serial]
fn decommission_completes_on_other_pools_while_one_pool_exceeds_its_temporary_budget() {
run_large_stack_async_test("decommission-over-budget-target-pool", || async {
let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await;
let bucket = test_bucket("over-budget-target-pool");
let object = "over-budget-target-pool.bin";
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create the over-budget bucket");
let mut source = PutObjReader::from_vec(b"migrates around the over-budget pool".to_vec());
store.pools[0]
.put_object(&bucket, object, &mut source, &ObjectOptions::default())
.await
.expect("write the source object before decommission starts");
// Of the 128 MiB peak, target 1 takes 100 MiB and target 2 the
// other 28 MiB; both stay far above the fixture and the
// `.rustfs.sys` metadata the routine also migrates.
let layout = DecommissionErasureLayout { data: 1, parity: 0 };
set_decommission_capacity_info_overrides_for_test(
store.id,
vec![vec![
DecommissionPoolCapacityInfo::for_test(0, layout, 0, 64 << 20, 64 << 20),
DecommissionPoolCapacityInfo::for_test(1, layout, 100 << 20, 100 << 20, 0),
DecommissionPoolCapacityInfo::for_test(2, layout, 1 << 30, 1 << 30, 0),
]],
);
store
.save_current_pool_meta_for_decommission_start(
&[0],
vec![crate::core::pools::DecomBucketInfo {
name: bucket.clone(),
prefix: String::new(),
}],
)
.await
.expect("activate the decommission");
// Target 1's temporary-copy budget is 50 MiB.
strand_decommission_target_marker(&store, 1, 60 << 20).await;
let canceler = crate::core::pools::DecommissionCanceler::new(CancellationToken::new());
store.decommission_cancelers.write().await[0] = Some(canceler.clone());
store
.do_decommission_in_routine(canceler, 0, Arc::new(tokio::sync::Semaphore::new(1)))
.await
.expect("the decommission routine should return");
let local = store.pool_meta.read().await;
let info = local.pools[0]
.decommission
.as_ref()
.expect("the decommission state should remain present");
assert!(
info.complete && !info.failed && !info.canceled,
"one over-budget pool must not pause the decommission: complete={}, failed={}, canceled={}, blocked_reason={:?}",
info.complete,
info.failed,
info.canceled,
info.capacity_blocked_reason
);
drop(local);
store.pools[2]
.get_object_info(&bucket, object, &ObjectOptions::default())
.await
.expect("the object must migrate to the pool within its budget");
assert!(
store.pools[1]
.get_object_info(&bucket, object, &ObjectOptions::default())
.await
.is_err(),
"target selection must skip the over-budget pool"
);
});
}
#[tokio::test]
#[serial_test::serial]
async fn no_active_external_commit_probe_blocks_decommission_activation_until_commit_owns_it() {