mirror of
https://github.com/rustfs/rustfs.git
synced 2026-11-08 11:33:41 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 3736d14501 |
@@ -2208,9 +2208,14 @@ fn ensure_decommission_capacity_reservations_available(
|
||||
continue;
|
||||
}
|
||||
metrics::counter!(METRIC_DECOMMISSION_CAPACITY_CONFLICTS_TOTAL, "phase" => phase).increment(1);
|
||||
return Err(decommission_capacity_blocked_error(format!(
|
||||
let detail = format!(
|
||||
"target pool {target_pool_index} reservation requires {required} physical bytes, but only {available} bytes remain during {phase}"
|
||||
)));
|
||||
);
|
||||
return Err(if growth_pool.is_some() {
|
||||
decommission_target_no_longer_fits_error(target_pool_index, detail)
|
||||
} else {
|
||||
decommission_capacity_blocked_error(detail)
|
||||
});
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
@@ -2341,6 +2346,13 @@ fn grow_decommission_target_reservation(
|
||||
let existing_remaining = existing_target
|
||||
.map(|target| target.remaining_reserved_physical_bytes(reservation.temporary_copies))
|
||||
.unwrap_or_default();
|
||||
let existing_reserved = existing_target
|
||||
.map(|target| target.reserved_physical_bytes)
|
||||
.unwrap_or_default();
|
||||
let allocation_limit = existing_target
|
||||
.map(|target| target.physical_free_at_reservation)
|
||||
.unwrap_or_default()
|
||||
.max(target_capacity.physical_free);
|
||||
if required_peak <= existing_remaining {
|
||||
return Ok(false);
|
||||
}
|
||||
@@ -2376,6 +2388,15 @@ fn grow_decommission_target_reservation(
|
||||
.predicted_physical_bytes
|
||||
.saturating_add(reservation.temporary_physical_bytes);
|
||||
let allocation_growth = reservation.remaining_peak_physical_bytes().saturating_sub(current_peak);
|
||||
let required_allocation = existing_reserved.saturating_add(allocation_growth);
|
||||
if required_allocation > allocation_limit {
|
||||
return Err(decommission_target_no_longer_fits_error(
|
||||
target_pool_index,
|
||||
format!(
|
||||
"target pool {target_pool_index} reservation requires {required_allocation} physical bytes, but its observed allocation limit is {allocation_limit} bytes"
|
||||
),
|
||||
));
|
||||
}
|
||||
if let Some(target) = reservation
|
||||
.targets
|
||||
.iter_mut()
|
||||
@@ -37890,10 +37911,9 @@ mod pools_tests {
|
||||
);
|
||||
let before = serde_json::to_value(&meta).expect("snapshot reservation before rejected growth");
|
||||
let shortage = vec![initial[0], DecommissionPoolCapacityInfo::for_test(1, layout, 43, 64, 21)];
|
||||
assert!(
|
||||
grow_decommission_target_reservation(&mut meta, &shortage, 0, 1, 32, uuid::Uuid::new_v4(), now).is_err(),
|
||||
"real free-space shortage still rejects growth"
|
||||
);
|
||||
let error = grow_decommission_target_reservation(&mut meta, &shortage, 0, 1, 32, uuid::Uuid::new_v4(), now)
|
||||
.expect_err("real free-space shortage still rejects growth");
|
||||
assert_eq!(super::decommission_capacity_stale_target_index(&error), Some(1));
|
||||
assert_eq!(serde_json::to_value(&meta).expect("snapshot after rejected growth"), before);
|
||||
}
|
||||
|
||||
|
||||
@@ -1616,7 +1616,7 @@ mod decommission_lock_order_tests {
|
||||
run_routine_to_a_deferral_pause(
|
||||
&store,
|
||||
1,
|
||||
"target pool 1 no longer fits the selected mutation: source pool 0 reservation estimate is exhausted",
|
||||
"target pool 1 no longer fits the selected mutation: target pool 1 reservation requires",
|
||||
)
|
||||
.await;
|
||||
});
|
||||
@@ -1647,7 +1647,7 @@ mod decommission_lock_order_tests {
|
||||
run_routine_to_a_deferral_pause(
|
||||
&store,
|
||||
1,
|
||||
"target pool 1 no longer fits the selected mutation: source pool 0 reservation estimate is exhausted",
|
||||
"target pool 1 no longer fits the selected mutation: target pool 1 reservation requires",
|
||||
)
|
||||
.await;
|
||||
});
|
||||
@@ -6062,6 +6062,12 @@ mod decommission_lock_order_tests {
|
||||
)
|
||||
.await
|
||||
.expect("read the next source object for the multipart capacity oracle");
|
||||
// Reconciled bytes remain charged. With no physical room for a new
|
||||
// reservation, a distinct mutation must fail instead of reusing them.
|
||||
let mut exhausted_capacity = capacity_snapshot();
|
||||
exhausted_capacity[2] = DecommissionPoolCapacityInfo::for_test(2, layout, 0, target_total, target_total);
|
||||
set_decommission_capacity_info_overrides_for_test(other_store.id, vec![exhausted_capacity]);
|
||||
|
||||
let next_result = tokio::time::timeout(
|
||||
Duration::from_secs(30),
|
||||
data_movement::migrate_decommission_object(
|
||||
@@ -7827,6 +7833,12 @@ mod decommission_lock_order_tests {
|
||||
.expect("same-mutation retry should publish a readable target");
|
||||
assert_eq!(retry_info.size, i64::try_from(body.len()).expect("test body size fits i64"));
|
||||
|
||||
// Reconciled bytes remain charged. With no physical room for a new
|
||||
// reservation, a distinct mutation must fail instead of reusing them.
|
||||
let mut exhausted_capacity = capacity_snapshot();
|
||||
exhausted_capacity[2] = DecommissionPoolCapacityInfo::for_test(2, layout, 0, target_total, target_total);
|
||||
set_decommission_capacity_info_overrides_for_test(other_store.id, vec![exhausted_capacity]);
|
||||
|
||||
let next_object = "data-movement-put-after-retry.bin";
|
||||
let next_version_id = uuid::Uuid::new_v4().to_string();
|
||||
let mut next_opts = ObjectOptions {
|
||||
@@ -7841,7 +7853,11 @@ mod decommission_lock_order_tests {
|
||||
let next_result = other_store
|
||||
.put_object_for_data_movement(RUSTFS_META_BUCKET, next_object, &mut next_data, &next_opts, None)
|
||||
.await;
|
||||
assert!(next_result.is_err(), "a different mutation must not reuse A's recovered capacity");
|
||||
let next_error = match next_result {
|
||||
Err(error) | Ok((_, Err(error))) => error,
|
||||
Ok((_, Ok(_))) => panic!("a different mutation must not reuse A's recovered capacity"),
|
||||
};
|
||||
assert!(next_error.to_string().contains("decommission_capacity_blocked"), "{next_error}");
|
||||
let next_target_err = other_store.pools[2]
|
||||
.get_object_info(
|
||||
RUSTFS_META_BUCKET,
|
||||
@@ -9929,6 +9945,12 @@ mod decommission_lock_order_tests {
|
||||
)
|
||||
.await
|
||||
.expect("read the next source object for the capacity oracle");
|
||||
// Reconciled bytes remain charged. With no physical room for a new
|
||||
// reservation, a distinct mutation must fail instead of reusing them.
|
||||
let mut exhausted_capacity = capacity_snapshot();
|
||||
exhausted_capacity[2] = DecommissionPoolCapacityInfo::for_test(2, layout, 0, target_total, target_total);
|
||||
set_decommission_capacity_info_overrides_for_test(other_store.id, vec![exhausted_capacity]);
|
||||
|
||||
let next_result = tokio::time::timeout(
|
||||
Duration::from_secs(30),
|
||||
data_movement::migrate_decommission_object(
|
||||
@@ -10190,11 +10212,20 @@ mod decommission_lock_order_tests {
|
||||
assert_eq!(reconciled_reservation.pending_target_physical_bytes, 0);
|
||||
assert_eq!(reconciled_reservation.consumed_target_physical_bytes, 1);
|
||||
|
||||
// Reconciled bytes remain charged. With no physical room for a new
|
||||
// reservation, a distinct mutation must fail instead of reusing them.
|
||||
let mut exhausted_capacity = capacity_snapshot();
|
||||
exhausted_capacity[2] = DecommissionPoolCapacityInfo::for_test(2, layout, 1, 2, 1);
|
||||
set_decommission_capacity_info_overrides_for_test(other_store.id, vec![exhausted_capacity]);
|
||||
|
||||
let next_after_reconcile = other_store
|
||||
.delete_object(&bucket, next_object, next_opts)
|
||||
.await
|
||||
.expect_err("the consumed exact-fit capacity must reject the next marker");
|
||||
assert!(next_after_reconcile.to_string().contains("decommission_capacity_blocked"));
|
||||
assert!(
|
||||
next_after_reconcile.to_string().contains("decommission_capacity_blocked"),
|
||||
"{next_after_reconcile}"
|
||||
);
|
||||
let next_target_err = other_store.pools[2]
|
||||
.get_object_info(
|
||||
&bucket,
|
||||
|
||||
Reference in New Issue
Block a user