mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-08 21:25:59 +00:00
fix(ecstore): preserve online writes during pool retirement (#7472)
* fix(ecstore): reconcile identical scanner backlog replicas * fix(ecstore): type invalid decommission requests * chore(ecstore): tighten typed-error ratchet baseline * fix(ecstore): fence late writes to retiring pools * fix(ecstore): share healthy pool capacity during decommission * fix: allow active multipart uploads to drain --------- Co-authored-by: overtrue <anzhengchao@gmail.com>
This commit is contained in:
cxymds
@@ -840,15 +840,21 @@ fn is_decommission_start_active_pool(pool: &PoolStatus) -> bool {
|
||||
decommission_start_pool_state(Some(pool)) == DecommissionStartPoolState::Active
|
||||
}
|
||||
|
||||
fn invalid_decommission_request(reason: impl Into<String>) -> Error {
|
||||
Error::InvalidArgument("decommission".to_string(), "pool-state".to_string(), reason.into())
|
||||
}
|
||||
|
||||
fn ensure_decommission_start_allowed(state: DecommissionStartPoolState) -> Result<()> {
|
||||
match state {
|
||||
DecommissionStartPoolState::Missing => Err(Error::other("failed to start decommission: target pool was not found")),
|
||||
DecommissionStartPoolState::Missing => {
|
||||
Err(invalid_decommission_request("failed to start decommission: target pool was not found"))
|
||||
}
|
||||
DecommissionStartPoolState::Active | DecommissionStartPoolState::Retryable => Ok(()),
|
||||
DecommissionStartPoolState::Decommissioning => Err(StorageError::DecommissionAlreadyRunning),
|
||||
DecommissionStartPoolState::Decommissioned => {
|
||||
Err(Error::other("failed to start decommission: target pool is already decommissioned"))
|
||||
}
|
||||
DecommissionStartPoolState::Blocked => Err(Error::other(
|
||||
DecommissionStartPoolState::Decommissioned => Err(invalid_decommission_request(
|
||||
"failed to start decommission: target pool is already decommissioned",
|
||||
)),
|
||||
DecommissionStartPoolState::Blocked => Err(invalid_decommission_request(
|
||||
"failed to start decommission: target pool decommission is blocked; clear failed or canceled metadata before starting again",
|
||||
)),
|
||||
}
|
||||
@@ -865,7 +871,7 @@ fn ensure_decommission_start_keeps_active_pool(meta: &PoolMeta, indices: &[usize
|
||||
.filter(|idx| meta.pools.get(**idx).is_some_and(is_decommission_start_active_pool))
|
||||
.count();
|
||||
if active_count.saturating_sub(active_target_count) == 0 {
|
||||
return Err(Error::other(
|
||||
return Err(invalid_decommission_request(
|
||||
"failed to start decommission: at least one active pool must remain after decommission start",
|
||||
));
|
||||
}
|
||||
@@ -1751,8 +1757,53 @@ fn ensure_decommission_capacity_reservations_available(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn ensure_external_decommission_target_admission(meta: &PoolMeta, target_pool_index: usize, phase: &'static str) -> Result<()> {
|
||||
if active_decommission_source_indices(meta).into_iter().any(|source_pool_index| {
|
||||
#[derive(Clone, Copy)]
|
||||
pub(crate) enum DecommissionCapacityAdmission {
|
||||
Mutation,
|
||||
ExistingMultipart,
|
||||
ScannerBacklog,
|
||||
BatchDelete,
|
||||
Heal,
|
||||
}
|
||||
|
||||
impl DecommissionCapacityAdmission {
|
||||
fn phase(self) -> &'static str {
|
||||
match self {
|
||||
Self::Mutation => "mutation",
|
||||
Self::ExistingMultipart => "existing_multipart",
|
||||
Self::ScannerBacklog => "scanner_backlog",
|
||||
Self::BatchDelete => "batch_delete",
|
||||
Self::Heal => "heal",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn ensure_external_decommission_target_admission(
|
||||
meta: &PoolMeta,
|
||||
target_pool_index: usize,
|
||||
admission: DecommissionCapacityAdmission,
|
||||
) -> Result<()> {
|
||||
let phase = admission.phase();
|
||||
// Pool selection may predate retirement or use a stale node-local snapshot.
|
||||
// Recheck publication against the fenced durable state. Repair and pure
|
||||
// capacity release retain their separate admission contracts.
|
||||
if matches!(admission, DecommissionCapacityAdmission::ScannerBacklog)
|
||||
&& !meta.scanner_pause_backlog_pool_writable(target_pool_index)
|
||||
{
|
||||
return Err(Error::SlowDown);
|
||||
}
|
||||
let active_sources = active_decommission_source_indices(meta);
|
||||
if meta.is_suspended(target_pool_index) {
|
||||
let active_source = active_sources.contains(&target_pool_index);
|
||||
if !matches!(
|
||||
admission,
|
||||
DecommissionCapacityAdmission::Heal | DecommissionCapacityAdmission::ScannerBacklog
|
||||
) && !(matches!(admission, DecommissionCapacityAdmission::ExistingMultipart) && active_source)
|
||||
{
|
||||
return Err(Error::SlowDown);
|
||||
}
|
||||
}
|
||||
if active_sources.into_iter().any(|source_pool_index| {
|
||||
meta.pools
|
||||
.get(source_pool_index)
|
||||
.and_then(|pool| pool.decommission.as_ref())
|
||||
@@ -1762,6 +1813,13 @@ fn ensure_external_decommission_target_admission(meta: &PoolMeta, target_pool_in
|
||||
metrics::counter!(METRIC_DECOMMISSION_CAPACITY_CONFLICTS_TOTAL, "phase" => phase).increment(1);
|
||||
return Err(Error::SlowDown);
|
||||
}
|
||||
// Migration reservations budget the mover, not exclusive ownership of a
|
||||
// healthy pool. Foreground publication shares its actual disk capacity;
|
||||
// migration must retain the source if its capacity or target write fails.
|
||||
// Repair keeps its separate, conservative reservation admission contract.
|
||||
if !matches!(admission, DecommissionCapacityAdmission::Heal) {
|
||||
return Ok(());
|
||||
}
|
||||
let reserved = active_decommission_target_reservations(meta)
|
||||
.get(&target_pool_index)
|
||||
.copied()
|
||||
@@ -4268,7 +4326,7 @@ fn should_retry_decommission_cancel_reload(changed: bool, already_canceled: bool
|
||||
|
||||
fn ensure_decommission_cancel_allowed(pool_present: bool, decommission_present: bool, terminal: bool) -> Result<()> {
|
||||
if !pool_present {
|
||||
return Err(Error::other("failed to cancel decommission: target pool was not found"));
|
||||
return Err(invalid_decommission_request("failed to cancel decommission: target pool was not found"));
|
||||
}
|
||||
|
||||
if !decommission_present || terminal {
|
||||
@@ -4287,7 +4345,7 @@ fn ensure_decommission_clear_allowed(
|
||||
unresolved_entries: usize,
|
||||
) -> Result<()> {
|
||||
if !pool_present {
|
||||
return Err(Error::other("failed to clear decommission: target pool was not found"));
|
||||
return Err(invalid_decommission_request("failed to clear decommission: target pool was not found"));
|
||||
}
|
||||
|
||||
if !decommission_present {
|
||||
@@ -4303,7 +4361,7 @@ fn ensure_decommission_clear_allowed(
|
||||
}
|
||||
|
||||
if unresolved_entries > 0 {
|
||||
return Err(Error::other(format!(
|
||||
return Err(invalid_decommission_request(format!(
|
||||
"failed to clear decommission: {unresolved_entries} unresolved listing entries must be reconciled by retrying decommission"
|
||||
)));
|
||||
}
|
||||
@@ -4313,7 +4371,7 @@ fn ensure_decommission_clear_allowed(
|
||||
|
||||
fn ensure_decommission_terminal_operation_supported(single_pool: bool, operation: &str) -> Result<()> {
|
||||
if single_pool {
|
||||
return Err(Error::other(format!(
|
||||
return Err(invalid_decommission_request(format!(
|
||||
"failed to {operation}: single pool deployments do not support decommission"
|
||||
)));
|
||||
}
|
||||
@@ -4323,7 +4381,9 @@ fn ensure_decommission_terminal_operation_supported(single_pool: bool, operation
|
||||
|
||||
fn validate_start_decommission_request(indices: &[usize], single_pool: bool) -> Result<()> {
|
||||
if indices.is_empty() {
|
||||
return Err(Error::other("failed to start decommission: no target pools were provided"));
|
||||
return Err(invalid_decommission_request(
|
||||
"failed to start decommission: no target pools were provided",
|
||||
));
|
||||
}
|
||||
|
||||
ensure_decommission_terminal_operation_supported(single_pool, "start decommission")
|
||||
@@ -7016,6 +7076,15 @@ impl PoolMeta {
|
||||
.is_some_and(is_decommission_suspended)
|
||||
}
|
||||
|
||||
pub(crate) fn scanner_pause_backlog_pool_writable(&self, idx: usize) -> bool {
|
||||
self.pools.get(idx).is_some_and(|pool| {
|
||||
!pool
|
||||
.decommission
|
||||
.as_ref()
|
||||
.is_some_and(|info| info.has_decommission_state() && !info.failed && !info.canceled)
|
||||
})
|
||||
}
|
||||
|
||||
fn mark_decommission_progress_saved(&mut self) {
|
||||
for pool in &mut self.pools {
|
||||
if let Some(info) = pool.decommission.as_mut() {
|
||||
@@ -10033,10 +10102,10 @@ impl ECStore {
|
||||
pub(crate) async fn acquire_external_decommission_capacity_fence(
|
||||
&self,
|
||||
target_pool_indices: &[usize],
|
||||
phase: &'static str,
|
||||
admission: DecommissionCapacityAdmission,
|
||||
) -> Result<rustfs_lock::NamespaceLockGuard> {
|
||||
Ok(self
|
||||
.acquire_external_decommission_capacity_fence_with_active_source(target_pool_indices, phase)
|
||||
.acquire_external_decommission_capacity_fence_with_active_source(target_pool_indices, admission)
|
||||
.await?
|
||||
.0)
|
||||
}
|
||||
@@ -10044,14 +10113,14 @@ impl ECStore {
|
||||
pub(crate) async fn acquire_external_decommission_capacity_fence_with_active_source(
|
||||
&self,
|
||||
target_pool_indices: &[usize],
|
||||
phase: &'static str,
|
||||
admission: DecommissionCapacityAdmission,
|
||||
) -> Result<(rustfs_lock::NamespaceLockGuard, bool)> {
|
||||
let save_guard = self.pool_meta_save_gate.lock().await;
|
||||
let (pool_meta_guard, snapshot) = self
|
||||
.acquire_pool_meta_read_guard(&save_guard, "target capacity admission failed")
|
||||
.await?;
|
||||
for target_pool_index in target_pool_indices.iter().copied() {
|
||||
ensure_external_decommission_target_admission(&snapshot, target_pool_index, phase)?;
|
||||
ensure_external_decommission_target_admission(&snapshot, target_pool_index, admission)?;
|
||||
}
|
||||
let has_active_source = pool_meta_has_active_decommission(&snapshot);
|
||||
drop(save_guard);
|
||||
@@ -10073,7 +10142,9 @@ impl ECStore {
|
||||
let admissions = target_pool_indices
|
||||
.iter()
|
||||
.copied()
|
||||
.map(|target_pool_index| ensure_external_decommission_target_admission(&snapshot, target_pool_index, "heal"))
|
||||
.map(|target_pool_index| {
|
||||
ensure_external_decommission_target_admission(&snapshot, target_pool_index, DecommissionCapacityAdmission::Heal)
|
||||
})
|
||||
.collect();
|
||||
drop(save_guard);
|
||||
Ok((pool_meta_guard, admissions))
|
||||
@@ -10747,7 +10818,7 @@ impl ECStore {
|
||||
));
|
||||
}
|
||||
let Some((owner, model_version)) = admitted_owner else {
|
||||
ensure_external_decommission_target_admission(&snapshot, target_pool_index, "mutation")?;
|
||||
ensure_external_decommission_target_admission(&snapshot, target_pool_index, DecommissionCapacityAdmission::Mutation)?;
|
||||
drop(save_guard);
|
||||
let capacity_lease = read_guard.lock_lost_signal();
|
||||
return operation.take().expect("capacity-admitted operation should run once")(capacity_lease).await;
|
||||
@@ -20911,17 +20982,17 @@ mod pools_tests {
|
||||
DecommissionStartPoolState, DecommissionTargetConsumption, DecommissionTerminalState, DecommissionUnresolvedEntry,
|
||||
ListCallback, POOL_META_GENERATION_VERSION, POOL_META_IDENTITY_NAME, POOL_META_NAME, POOL_META_V1_VERSION,
|
||||
POOL_META_VERSION, PoolDecommissionInfo, PoolMeta, PoolMetaCasToken, PoolMetaPersistenceFence, PoolSpaceInfo, PoolStatus,
|
||||
QueuedDecommissionEntry, REBAL_META_NAME, acquire_pool_rebalance_activation_locks, apply_decommission_status_space_info,
|
||||
await_decommission_worker, bind_decommission_cancelers, bind_missing_decommission_cancelers,
|
||||
build_decommission_capacity_reservation, build_decommission_capacity_reservation_with_model,
|
||||
cancel_decommission_canceler, clamp_decommission_entry_concurrency, classify_decommission_terminal_state,
|
||||
count_decommission_item, decommission_cancel_signal_result, decommission_durable_ilm_receipt_path,
|
||||
decommission_durable_ilm_receipt_run_prefix, decommission_durable_ilm_receipt_run_token,
|
||||
decommission_entry_queue_capacity, decommission_item_size, decommission_meta_bucket_options,
|
||||
decommission_physical_pool_capacity, decommission_retry_backoff_delay, decommission_start_pool_state,
|
||||
decommission_unresolved_listing_error, dedup_indices, default_decommission_bucket_concurrency,
|
||||
default_decommission_entry_concurrency, drain_decommission_entry_queue, enqueue_decommission_entry,
|
||||
ensure_decommission_cancel_allowed, ensure_decommission_capacity_reservations_available,
|
||||
QueuedDecommissionEntry, REBAL_META_NAME, acquire_pool_rebalance_activation_locks, active_decommission_source_indices,
|
||||
apply_decommission_status_space_info, await_decommission_worker, bind_decommission_cancelers,
|
||||
bind_missing_decommission_cancelers, build_decommission_capacity_reservation,
|
||||
build_decommission_capacity_reservation_with_model, cancel_decommission_canceler, clamp_decommission_entry_concurrency,
|
||||
classify_decommission_terminal_state, count_decommission_item, decommission_cancel_signal_result,
|
||||
decommission_durable_ilm_receipt_path, decommission_durable_ilm_receipt_run_prefix,
|
||||
decommission_durable_ilm_receipt_run_token, decommission_entry_queue_capacity, decommission_item_size,
|
||||
decommission_meta_bucket_options, decommission_physical_pool_capacity, decommission_retry_backoff_delay,
|
||||
decommission_start_pool_state, decommission_unresolved_listing_error, dedup_indices,
|
||||
default_decommission_bucket_concurrency, default_decommission_entry_concurrency, drain_decommission_entry_queue,
|
||||
enqueue_decommission_entry, ensure_decommission_cancel_allowed, ensure_decommission_capacity_reservations_available,
|
||||
ensure_decommission_clear_allowed, ensure_decommission_generation, ensure_decommission_listing_disks_available,
|
||||
ensure_decommission_not_rebalancing, ensure_decommission_start_allowed, ensure_decommission_start_keeps_active_pool,
|
||||
ensure_decommission_start_local_leader, ensure_decommission_start_pool_states,
|
||||
@@ -20957,12 +21028,13 @@ mod pools_tests {
|
||||
with_decommission_entry_context,
|
||||
};
|
||||
use super::{
|
||||
DecommissionCapacityOwner, DecommissionCapacityReleaseProof, DecommissionCapacityReservation,
|
||||
DecommissionCapacityTemporaryMutation, decommission_capacity_mutation_id, 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,
|
||||
DecommissionCapacityAdmission, DecommissionCapacityOwner, DecommissionCapacityReleaseProof,
|
||||
DecommissionCapacityReservation, DecommissionCapacityTemporaryMutation, decommission_capacity_mutation_id,
|
||||
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,
|
||||
};
|
||||
use crate::bucket::lifecycle::{
|
||||
DurableIlmRecordCheckpoint,
|
||||
@@ -25116,6 +25188,25 @@ mod pools_tests {
|
||||
assert!(!pool_meta_has_active_decommission(&terminal_meta));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_decommission_request_rejections_preserve_invalid_argument_type() {
|
||||
for result in [
|
||||
ensure_decommission_start_allowed(DecommissionStartPoolState::Missing),
|
||||
ensure_decommission_start_allowed(DecommissionStartPoolState::Decommissioned),
|
||||
ensure_decommission_start_allowed(DecommissionStartPoolState::Blocked),
|
||||
ensure_decommission_cancel_allowed(false, false, false),
|
||||
ensure_decommission_clear_allowed(false, false, false, false, false, 0),
|
||||
ensure_decommission_clear_allowed(true, true, false, true, false, 1),
|
||||
ensure_decommission_terminal_operation_supported(true, "cancel decommission"),
|
||||
validate_start_decommission_request(&[], false),
|
||||
validate_start_decommission_request(&[0], true),
|
||||
ensure_decommission_start_keeps_active_pool(&PoolMeta::default(), &[]),
|
||||
] {
|
||||
let err = result.expect_err("invalid lifecycle requests must be rejected before mutation");
|
||||
assert!(matches!(&err, Error::InvalidArgument(_, _, reason) if !reason.is_empty()), "{err:?}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_ensure_decommission_start_allowed_rejects_missing_pool() {
|
||||
let err =
|
||||
@@ -25818,7 +25909,7 @@ mod pools_tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ordinary_write_admission_cannot_race_into_a_reserved_target() {
|
||||
fn ordinary_write_admission_shares_a_reserved_target_without_becoming_its_owner() {
|
||||
let now = OffsetDateTime::UNIX_EPOCH + Duration::minutes(2);
|
||||
let layout = DecommissionErasureLayout { data: 1, parity: 0 };
|
||||
let capacity_infos = vec![
|
||||
@@ -25842,13 +25933,18 @@ mod pools_tests {
|
||||
)
|
||||
.expect("the decommission reservation should fit");
|
||||
|
||||
assert!(
|
||||
matches!(
|
||||
ensure_external_decommission_target_admission(&meta, 1, "ordinary_put"),
|
||||
Err(Error::SlowDown)
|
||||
),
|
||||
"an ordinary write must not consume a target reservation"
|
||||
);
|
||||
for admission in [
|
||||
DecommissionCapacityAdmission::Mutation,
|
||||
DecommissionCapacityAdmission::BatchDelete,
|
||||
DecommissionCapacityAdmission::ScannerBacklog,
|
||||
] {
|
||||
ensure_external_decommission_target_admission(&meta, 1, admission)
|
||||
.expect("a healthy target must remain writable while sharing capacity with migration");
|
||||
}
|
||||
assert!(matches!(
|
||||
ensure_external_decommission_target_admission(&meta, 1, DecommissionCapacityAdmission::Heal),
|
||||
Err(Error::SlowDown)
|
||||
));
|
||||
let rebalance_opts = ObjectOptions {
|
||||
data_movement: true,
|
||||
src_pool_idx: 0,
|
||||
@@ -25875,6 +25971,119 @@ mod pools_tests {
|
||||
let mut decommission_opts = rebalance_opts;
|
||||
expected_owner.apply_to(&mut decommission_opts);
|
||||
assert_eq!(DecommissionCapacityOwner::from_options(&decommission_opts), Some(expected_owner));
|
||||
|
||||
meta.pools[0]
|
||||
.decommission
|
||||
.as_mut()
|
||||
.expect("active source")
|
||||
.capacity_reservation = None;
|
||||
for admission in [
|
||||
DecommissionCapacityAdmission::Mutation,
|
||||
DecommissionCapacityAdmission::BatchDelete,
|
||||
DecommissionCapacityAdmission::ScannerBacklog,
|
||||
] {
|
||||
assert!(
|
||||
matches!(ensure_external_decommission_target_admission(&meta, 1, admission), Err(Error::SlowDown)),
|
||||
"shared capacity must not bypass an active source's missing durable ledger"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn external_decommission_admission_fences_suspended_sources_but_preserves_repair() {
|
||||
let now = OffsetDateTime::UNIX_EPOCH + Duration::minutes(2);
|
||||
let layout = DecommissionErasureLayout { data: 1, parity: 0 };
|
||||
let capacity_infos = vec![
|
||||
DecommissionPoolCapacityInfo::for_test(0, layout, 0, 30, 30),
|
||||
DecommissionPoolCapacityInfo::for_test(1, layout, 100, 100, 0),
|
||||
];
|
||||
let mut active = PoolMeta {
|
||||
version: POOL_META_VERSION,
|
||||
pools: vec![decommission_test_pool_status(0, None), decommission_test_pool_status(1, None)],
|
||||
..Default::default()
|
||||
};
|
||||
active
|
||||
.decommission(0, capacity_infos[0].space)
|
||||
.expect("start the source admission fixture");
|
||||
reserve_decommission_start_target_capacity(
|
||||
&mut active,
|
||||
&[0],
|
||||
&capacity_infos,
|
||||
uuid::Uuid::new_v4(),
|
||||
1,
|
||||
now,
|
||||
DECOMMISSION_CAPACITY_MODEL_VERSION,
|
||||
)
|
||||
.expect("the active source must have a valid reservation to isolate its write fence");
|
||||
|
||||
for (state, queued, failed, canceled, complete) in [
|
||||
("running", false, false, false, false),
|
||||
("queued", true, false, false, false),
|
||||
("failed", false, true, false, false),
|
||||
("canceled", false, false, true, false),
|
||||
("completed", false, false, false, true),
|
||||
] {
|
||||
let mut meta = active.clone();
|
||||
let info = meta.pools[0].decommission.as_mut().expect("the source fixture must exist");
|
||||
info.queued = queued;
|
||||
info.failed = failed;
|
||||
info.canceled = canceled;
|
||||
info.complete = complete;
|
||||
if queued || failed || canceled || complete {
|
||||
info.start_time = None;
|
||||
}
|
||||
for admission in [
|
||||
DecommissionCapacityAdmission::Mutation,
|
||||
DecommissionCapacityAdmission::BatchDelete,
|
||||
] {
|
||||
assert!(
|
||||
matches!(ensure_external_decommission_target_admission(&meta, 0, admission), Err(Error::SlowDown)),
|
||||
"{state} source must reject new publication until its decommission metadata is cleared"
|
||||
);
|
||||
}
|
||||
let existing_multipart =
|
||||
ensure_external_decommission_target_admission(&meta, 0, DecommissionCapacityAdmission::ExistingMultipart);
|
||||
if active_decommission_source_indices(&meta).contains(&0) {
|
||||
existing_multipart
|
||||
.unwrap_or_else(|err| panic!("{state} source must allow an existing multipart upload to drain: {err}"));
|
||||
} else {
|
||||
assert!(
|
||||
matches!(existing_multipart, Err(Error::SlowDown)),
|
||||
"{state} terminal source must reject an existing multipart publication"
|
||||
);
|
||||
}
|
||||
ensure_external_decommission_target_admission(&meta, 0, DecommissionCapacityAdmission::Heal)
|
||||
.unwrap_or_else(|err| panic!("{state} source repair must retain its capacity-only admission: {err}"));
|
||||
let scanner_result =
|
||||
ensure_external_decommission_target_admission(&meta, 0, DecommissionCapacityAdmission::ScannerBacklog);
|
||||
assert_eq!(
|
||||
meta.scanner_pause_backlog_pool_writable(0),
|
||||
failed || canceled,
|
||||
"{state} scanner selection"
|
||||
);
|
||||
if failed || canceled {
|
||||
scanner_result.unwrap_or_else(|err| panic!("{state} scanner membership repair must remain writable: {err}"));
|
||||
} else {
|
||||
assert!(
|
||||
matches!(scanner_result, Err(Error::SlowDown)),
|
||||
"{state} scanner publication must reject its source"
|
||||
);
|
||||
}
|
||||
meta.pools[0].decommission = None;
|
||||
ensure_external_decommission_target_admission(&meta, 0, DecommissionCapacityAdmission::Mutation)
|
||||
.unwrap_or_else(|err| panic!("cleared {state} source must become writable again: {err}"));
|
||||
ensure_external_decommission_target_admission(&meta, 0, DecommissionCapacityAdmission::ScannerBacklog)
|
||||
.unwrap_or_else(|err| panic!("cleared {state} scanner source must rejoin membership: {err}"));
|
||||
}
|
||||
assert!(!active.scanner_pause_backlog_pool_writable(active.pools.len()));
|
||||
assert!(matches!(
|
||||
ensure_external_decommission_target_admission(
|
||||
&active,
|
||||
active.pools.len(),
|
||||
DecommissionCapacityAdmission::ScannerBacklog
|
||||
),
|
||||
Err(Error::SlowDown)
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -570,6 +570,465 @@ mod decommission_lock_order_tests {
|
||||
.expect("decommission activation should commit after the probe release");
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial_test::serial]
|
||||
fn staged_external_put_rechecks_retiring_source_on_another_node() {
|
||||
run_large_stack_current_thread_async_test("staged-retiring-source", || async {
|
||||
let (_temp_dirs, store, other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await;
|
||||
let bucket = test_bucket("staged-source");
|
||||
let object = "selected-before-retirement.bin";
|
||||
let original = b"original source object";
|
||||
store
|
||||
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("create staged source bucket");
|
||||
store.pools[0]
|
||||
.put_object(&bucket, object, &mut PutObjReader::from_vec(original.to_vec()), &ObjectOptions::default())
|
||||
.await
|
||||
.expect("seed the source selected before retirement");
|
||||
|
||||
let layout = DecommissionErasureLayout { data: 1, parity: 0 };
|
||||
set_decommission_capacity_info_overrides_for_test(
|
||||
other_store.id,
|
||||
vec![vec![
|
||||
DecommissionPoolCapacityInfo::for_test(0, layout, 0, 1024, 1024),
|
||||
DecommissionPoolCapacityInfo::for_test(1, layout, 4096, 4096, 0),
|
||||
DecommissionPoolCapacityInfo::for_test(2, layout, 0, 4096, 4096),
|
||||
]],
|
||||
);
|
||||
let barrier = DecommissionCapacityLockOrderBarrier::install(store.id, store.id);
|
||||
barrier.pause_external_object_commit_phase();
|
||||
let put_store = Arc::clone(&store);
|
||||
let put_bucket = bucket.clone();
|
||||
let put = tokio::spawn(async move {
|
||||
put_store
|
||||
.put_object(
|
||||
&put_bucket,
|
||||
object,
|
||||
&mut PutObjReader::from_vec(b"must not replace a retiring source".to_vec()),
|
||||
&ObjectOptions::default(),
|
||||
)
|
||||
.await
|
||||
});
|
||||
tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_external_object_commit_phase_started())
|
||||
.await
|
||||
.expect("public PUT must stage before its decommission commit probe");
|
||||
assert!(!store.pool_meta.read().await.is_suspended(0));
|
||||
other_store
|
||||
.save_current_pool_meta_for_decommission_start(&[0], Vec::new())
|
||||
.await
|
||||
.expect("the other node should activate retirement before the staged PUT commits");
|
||||
assert!(other_store.pool_meta.read().await.is_suspended(0));
|
||||
assert!(
|
||||
!store.pool_meta.read().await.is_suspended(0),
|
||||
"the writer's local snapshot must remain stale to exercise the durable admission probe"
|
||||
);
|
||||
barrier.release_external_object_commit_phase();
|
||||
let result = tokio::time::timeout(Duration::from_secs(30), put)
|
||||
.await
|
||||
.expect("staged PUT must finish after the commit probe is released")
|
||||
.expect("staged PUT must not panic");
|
||||
assert!(
|
||||
matches!(result, Err(crate::error::Error::SlowDown)),
|
||||
"a staged PUT must retry pool selection instead of committing to a newly retiring source: {result:?}"
|
||||
);
|
||||
let mut reader = store.pools[0]
|
||||
.get_object_reader(&bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
|
||||
.await
|
||||
.expect("the original source must remain readable after admission rejects the replacement");
|
||||
let mut body = Vec::new();
|
||||
reader
|
||||
.stream
|
||||
.read_to_end(&mut body)
|
||||
.await
|
||||
.expect("read the full retained source body");
|
||||
assert_eq!(body, original);
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial_test::serial]
|
||||
fn reserved_target_shares_business_io_and_retains_source_after_capacity_loss() {
|
||||
run_large_stack_current_thread_async_test("shared-decommission-capacity", || async {
|
||||
for lose_capacity in [false, true] {
|
||||
let (_temp_dirs, store, other_store) =
|
||||
test_three_pool_stores_with_three_disk_sets_with_isolated_node_contexts(None).await;
|
||||
let bucket = test_bucket("shared-capacity");
|
||||
let object = "migrating-source.bin";
|
||||
let business_object = "business-write.bin";
|
||||
let multipart_object = "business-multipart.bin";
|
||||
let source_body = vec![0x35; 256 * 1024];
|
||||
let business_body = vec![0x57; 64 * 1024];
|
||||
store
|
||||
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||
.await
|
||||
.expect("create shared-capacity bucket");
|
||||
store.pools[0]
|
||||
.put_object(
|
||||
&bucket,
|
||||
object,
|
||||
&mut PutObjReader::from_vec(source_body.clone()),
|
||||
&ObjectOptions::default(),
|
||||
)
|
||||
.await
|
||||
.expect("seed the retiring source");
|
||||
store.pools[2]
|
||||
.put_object(
|
||||
&bucket,
|
||||
business_object,
|
||||
&mut PutObjReader::from_vec(b"previous business value".to_vec()),
|
||||
&ObjectOptions::default(),
|
||||
)
|
||||
.await
|
||||
.expect("pin the public overwrite to the migration target");
|
||||
let multipart_opts = ObjectOptions {
|
||||
expected_bucket_incarnation_id: Some(
|
||||
store
|
||||
.bucket_incarnation_id(&bucket)
|
||||
.await
|
||||
.expect("load the multipart bucket identity"),
|
||||
),
|
||||
..Default::default()
|
||||
};
|
||||
let routing_upload = new_multipart_upload(&store, 2, &bucket, multipart_object, multipart_opts.clone())
|
||||
.await
|
||||
.expect("pin subsequent public multipart creation to the migration target");
|
||||
let layout = DecommissionErasureLayout { data: 1, parity: 0 };
|
||||
let target_total = source_body.len() * 8;
|
||||
let capacities = vec![
|
||||
DecommissionPoolCapacityInfo::for_test(0, layout, 0, source_body.len() * 2, source_body.len() * 2),
|
||||
DecommissionPoolCapacityInfo::for_test(1, layout, 0, target_total, target_total),
|
||||
DecommissionPoolCapacityInfo::for_test(2, layout, target_total, target_total, 0),
|
||||
];
|
||||
set_decommission_capacity_info_overrides_for_test(store.id, vec![capacities.clone()]);
|
||||
store
|
||||
.save_current_pool_meta_for_decommission_start(&[0], Vec::new())
|
||||
.await
|
||||
.expect("activate source retirement");
|
||||
*other_store.pool_meta.write().await = store.pool_meta.read().await.clone();
|
||||
let before = other_store.pool_meta.read().await.clone();
|
||||
let reservation = before.pools[0]
|
||||
.decommission
|
||||
.as_ref()
|
||||
.expect("active source")
|
||||
.capacity_reservation
|
||||
.as_ref()
|
||||
.expect("durable reservation");
|
||||
assert_eq!(
|
||||
reservation.model_version, 2,
|
||||
"exercise migration I/O outside the global metadata write lock"
|
||||
);
|
||||
assert_eq!(reservation.targets[0].pool_index, 2);
|
||||
|
||||
let barrier =
|
||||
crate::set_disk::rename_fanout_barrier::arm(object, 0, crate::set_disk::rename_fanout_barrier::PHASE_RENAME);
|
||||
let migration_store = Arc::clone(&store);
|
||||
let migration_bucket = bucket.clone();
|
||||
let migration = tokio::spawn(async move {
|
||||
migration_store
|
||||
.decommission_entry_for_test_with_bucket_incarnation(
|
||||
0,
|
||||
MetaCacheEntry {
|
||||
name: object.to_string(),
|
||||
..Default::default()
|
||||
},
|
||||
migration_bucket,
|
||||
migration_store.pools[0].get_disks_by_key(object),
|
||||
)
|
||||
.await
|
||||
});
|
||||
tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
|
||||
.await
|
||||
.expect("migration must reach target publication");
|
||||
assert!(!migration.is_finished());
|
||||
let mut pending = crate::core::pools::PoolMeta::default();
|
||||
pending
|
||||
.load_no_lock_from_replicas(other_store.pools.clone())
|
||||
.await
|
||||
.expect("read migration intent from the other node");
|
||||
let pending_reservation = pending.pools[0]
|
||||
.decommission
|
||||
.as_ref()
|
||||
.expect("active source")
|
||||
.capacity_reservation
|
||||
.as_ref()
|
||||
.expect("pending reservation")
|
||||
.clone();
|
||||
assert_eq!(pending_reservation.pending_target_physical_bytes, source_body.len());
|
||||
assert_eq!(pending_reservation.consumed_target_physical_bytes, 0);
|
||||
|
||||
tokio::time::timeout(
|
||||
Duration::from_secs(30),
|
||||
other_store.put_object(
|
||||
&bucket,
|
||||
business_object,
|
||||
&mut PutObjReader::from_vec(business_body.clone()),
|
||||
&ObjectOptions::default(),
|
||||
),
|
||||
)
|
||||
.await
|
||||
.expect("business PUT must finish without waiting for the migration target gate")
|
||||
.expect("a reserved healthy pool must accept ordinary PUT");
|
||||
tokio::time::timeout(Duration::from_secs(30), async {
|
||||
let upload = other_store
|
||||
.new_multipart_upload(&bucket, multipart_object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("the reserved target must accept public multipart creation");
|
||||
assert_ne!(upload.upload_id, routing_upload.upload_id);
|
||||
let lifecycle_guard = other_store
|
||||
.acquire_bucket_lifecycle_read_lock(&bucket)
|
||||
.await
|
||||
.expect("fence the exact-pool multipart placement check");
|
||||
let mut lookup_opts = multipart_opts.clone();
|
||||
lookup_opts.add_bucket_lifecycle_lock_guard(&lifecycle_guard);
|
||||
other_store.pools[2]
|
||||
.get_multipart_info(&bucket, multipart_object, &upload.upload_id, &lookup_opts)
|
||||
.await
|
||||
.expect("public multipart creation must actually select the reserved target");
|
||||
drop(lifecycle_guard);
|
||||
let mut final_part = None;
|
||||
for payload in [vec![0x18; business_body.len()], business_body.clone()] {
|
||||
final_part = Some(
|
||||
other_store
|
||||
.put_object_part(
|
||||
&bucket,
|
||||
multipart_object,
|
||||
&upload.upload_id,
|
||||
1,
|
||||
&mut PutObjReader::from_vec(payload),
|
||||
&ObjectOptions::default(),
|
||||
)
|
||||
.await
|
||||
.expect("the reserved target must accept UploadPart and replacement of the same part"),
|
||||
);
|
||||
}
|
||||
let part = final_part.expect("the replacement part must be present");
|
||||
Arc::clone(&other_store)
|
||||
.complete_multipart_upload(
|
||||
&bucket,
|
||||
multipart_object,
|
||||
&upload.upload_id,
|
||||
vec![crate::storage_api_contracts::multipart::CompletePart {
|
||||
part_num: part.part_num,
|
||||
etag: part.etag,
|
||||
..Default::default()
|
||||
}],
|
||||
&ObjectOptions::default(),
|
||||
)
|
||||
.await
|
||||
.expect("the reserved target must accept multipart completion");
|
||||
other_store
|
||||
.abort_multipart_upload(&bucket, multipart_object, &routing_upload.upload_id, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("ordinary multipart cleanup must not consume the migration's pending intent");
|
||||
})
|
||||
.await
|
||||
.expect("business multipart operations must finish while migration I/O is paused");
|
||||
assert!(!migration.is_finished(), "business publication must overlap paused migration I/O");
|
||||
let mut after_business = crate::core::pools::PoolMeta::default();
|
||||
after_business
|
||||
.load_no_lock_from_replicas(other_store.pools.clone())
|
||||
.await
|
||||
.expect("reload the shared-capacity ledger");
|
||||
assert_eq!(
|
||||
after_business.pools[0]
|
||||
.decommission
|
||||
.as_ref()
|
||||
.expect("active source")
|
||||
.capacity_reservation
|
||||
.as_ref(),
|
||||
Some(&pending_reservation),
|
||||
"ordinary PUT and multipart operations must not settle or consume the migration's pending identity"
|
||||
);
|
||||
|
||||
let mut after_capacity = capacities;
|
||||
// Capacity injection is deterministic; the object I/O and durable metadata use real temporary disks.
|
||||
let free = if lose_capacity {
|
||||
0
|
||||
} else {
|
||||
target_total - source_body.len() - business_body.len() * 2
|
||||
};
|
||||
after_capacity[2] = DecommissionPoolCapacityInfo::for_test(2, layout, free, target_total, target_total - free);
|
||||
set_decommission_capacity_info_overrides_for_test(store.id, vec![after_capacity]);
|
||||
barrier.release();
|
||||
drop(barrier);
|
||||
let migrated = tokio::time::timeout(Duration::from_secs(30), migration)
|
||||
.await
|
||||
.expect("migration must finish after publication resumes")
|
||||
.expect("migration task must not panic");
|
||||
if lose_capacity {
|
||||
let err =
|
||||
migrated.expect_err("capacity loss must prevent source cleanup, even after the target write commits");
|
||||
assert!(err.to_string().contains("capacity"), "unexpected migration error: {err}");
|
||||
} else {
|
||||
migrated.expect("shared-capacity migration should finish when space remains sufficient");
|
||||
}
|
||||
let mut persisted = crate::core::pools::PoolMeta::default();
|
||||
persisted
|
||||
.load_no_lock_from_replicas(other_store.pools.clone())
|
||||
.await
|
||||
.expect("reload finalized migration state");
|
||||
let info = persisted.pools[0].decommission.as_ref().expect("source state");
|
||||
let reservation = info.capacity_reservation.as_ref().expect("migration ledger");
|
||||
assert_eq!(reservation.pending_target_physical_bytes, 0);
|
||||
assert_eq!(
|
||||
reservation.consumed_target_physical_bytes,
|
||||
source_body.len(),
|
||||
"foreign writes must not count as committed source bytes"
|
||||
);
|
||||
assert_eq!(reservation.committed_data_bytes, source_body.len());
|
||||
assert_eq!(info.capacity_blocked_reason.is_some(), lose_capacity);
|
||||
for (pool, key, expected) in [
|
||||
(2, business_object, &business_body),
|
||||
(2, multipart_object, &business_body),
|
||||
(2, object, &source_body),
|
||||
] {
|
||||
let mut reader = other_store.pools[pool]
|
||||
.get_object_reader(&bucket, key, None, HeaderMap::new(), &ObjectOptions::default())
|
||||
.await
|
||||
.expect("all acknowledged target objects must remain readable");
|
||||
let mut actual = Vec::new();
|
||||
reader.read_to_end(&mut actual).await.expect("read the complete target body");
|
||||
assert_eq!(&actual, expected);
|
||||
}
|
||||
if lose_capacity {
|
||||
let mut source = other_store.pools[0]
|
||||
.get_object_reader(&bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
|
||||
.await
|
||||
.expect("capacity-blocked migration must retain its source");
|
||||
let mut actual = Vec::new();
|
||||
source
|
||||
.read_to_end(&mut actual)
|
||||
.await
|
||||
.expect("read the complete retained source");
|
||||
assert_eq!(actual, source_body);
|
||||
other_store
|
||||
.put_object(
|
||||
&bucket,
|
||||
business_object,
|
||||
&mut PutObjReader::from_vec(business_body.clone()),
|
||||
&ObjectOptions::default(),
|
||||
)
|
||||
.await
|
||||
.expect("a capacity-blocked migration must not itself make the healthy target read-only");
|
||||
} else {
|
||||
let err = other_store.pools[0]
|
||||
.get_object_info(&bucket, object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect_err("successful migration must clean the exact source");
|
||||
assert!(crate::error::is_err_object_not_found(&err));
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial_test::serial]
|
||||
fn mixed_batch_delete_admits_only_marker_destinations_during_retirement() {
|
||||
run_large_stack_current_thread_async_test("batch-marker-admission", || async {
|
||||
use crate::storage_api_contracts::object::ObjectToDelete;
|
||||
|
||||
for marker_target in [1, 2] {
|
||||
let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await;
|
||||
let bucket = test_bucket("batch-marker");
|
||||
store
|
||||
.make_bucket(
|
||||
&bucket,
|
||||
&MakeBucketOptions {
|
||||
versioning_enabled: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("create a versioned batch-delete bucket");
|
||||
let source_version = uuid::Uuid::new_v4();
|
||||
for (pool, object, version) in [
|
||||
(0, "purge-source", source_version),
|
||||
(marker_target, "mark-active", uuid::Uuid::new_v4()),
|
||||
] {
|
||||
store.pools[pool]
|
||||
.put_object(
|
||||
&bucket,
|
||||
object,
|
||||
&mut PutObjReader::from_vec(b"version to delete".to_vec()),
|
||||
&ObjectOptions {
|
||||
versioned: true,
|
||||
version_id: Some(version.to_string()),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("seed each exact batch-delete destination");
|
||||
}
|
||||
let layout = DecommissionErasureLayout { data: 1, parity: 0 };
|
||||
set_decommission_capacity_info_overrides_for_test(
|
||||
store.id,
|
||||
vec![vec![
|
||||
DecommissionPoolCapacityInfo::for_test(0, layout, 0, 1024, 1024),
|
||||
DecommissionPoolCapacityInfo::for_test(1, layout, 4096, 4096, 0),
|
||||
DecommissionPoolCapacityInfo::for_test(2, layout, 0, 4096, 4096),
|
||||
]],
|
||||
);
|
||||
store
|
||||
.save_current_pool_meta_for_decommission_start(&[0], Vec::new())
|
||||
.await
|
||||
.expect("reserve pool 1 while pool 0 retires and pool 2 remains unreserved");
|
||||
let (deleted, errors) = store
|
||||
.delete_objects(
|
||||
&bucket,
|
||||
vec![
|
||||
ObjectToDelete {
|
||||
object_name: "mark-active".to_string(),
|
||||
..Default::default()
|
||||
},
|
||||
ObjectToDelete {
|
||||
object_name: "purge-source".to_string(),
|
||||
version_id: Some(source_version),
|
||||
..Default::default()
|
||||
},
|
||||
],
|
||||
ObjectOptions::default(),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(errors.len(), 2);
|
||||
assert!(
|
||||
errors.iter().all(Option::is_none),
|
||||
"the unrelated retiring/reserved pools must not reject marker admission: {errors:?}"
|
||||
);
|
||||
assert_eq!(deleted.len(), 2);
|
||||
assert_eq!(deleted[0].object_name, "mark-active");
|
||||
assert!(deleted[0].delete_marker);
|
||||
assert!(
|
||||
deleted[0].version_id.is_none(),
|
||||
"a latest-version delete does not request an explicit version"
|
||||
);
|
||||
assert!(
|
||||
deleted[0].delete_marker_version_id.is_some(),
|
||||
"the newly created marker must have its own version identity"
|
||||
);
|
||||
assert_eq!(deleted[1].object_name, "purge-source");
|
||||
assert!(!deleted[1].delete_marker);
|
||||
assert_eq!(deleted[1].version_id, Some(source_version));
|
||||
assert!(
|
||||
matches!(
|
||||
store.pools[0]
|
||||
.get_object_info(
|
||||
&bucket,
|
||||
"purge-source",
|
||||
&ObjectOptions {
|
||||
version_id: Some(source_version.to_string()),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await,
|
||||
Err(crate::error::Error::ObjectNotFound(..) | crate::error::Error::VersionNotFound(..))
|
||||
),
|
||||
"an exact source deletion must retain its capacity-release path"
|
||||
);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial]
|
||||
async fn public_upload_part_holds_decommission_capacity_until_rename() {
|
||||
@@ -4499,6 +4958,470 @@ mod decommission_lock_order_tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial_test::serial]
|
||||
fn scanner_backlog_cas_keeps_fences_after_waiter_cancellation_until_rename_drains() {
|
||||
run_large_stack_current_thread_async_test("scanner-backlog-canceled-waiter", async || {
|
||||
temp_env::async_with_vars([(crate::set_disk::ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
|
||||
let (_temp_dirs, writer, other) =
|
||||
test_three_pool_stores_with_three_disk_sets_with_isolated_node_contexts(None).await;
|
||||
let object = "buckets/.scanner-pause-backlog.json";
|
||||
let set_index = 1;
|
||||
let body = vec![0x37; 1024];
|
||||
assert!(
|
||||
!writer.pools[0].disk_set[0]
|
||||
.shares_namespace_lock_domain(&writer.pools[0].disk_set[set_index])
|
||||
.await
|
||||
);
|
||||
let rename_tasks = crate::set_disk::rename_fanout_barrier::observe_tasks(object);
|
||||
let tail =
|
||||
crate::set_disk::rename_fanout_barrier::arm(object, 0, crate::set_disk::rename_fanout_barrier::PHASE_RENAME);
|
||||
let put_store = Arc::clone(&writer);
|
||||
let put_body = body.clone();
|
||||
let mut put = tokio::spawn(async move {
|
||||
put_store
|
||||
.save_scanner_pause_backlog_replica(0, set_index, put_body, Default::default())
|
||||
.await
|
||||
});
|
||||
tokio::time::timeout(Duration::from_secs(30), tail.wait_until_paused())
|
||||
.await
|
||||
.expect("the native write must reach its held rename");
|
||||
tokio::time::timeout(Duration::from_secs(30), async {
|
||||
while rename_tasks.running() != 1 {
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("the other disks must reach quorum before canceling the waiter");
|
||||
assert!(
|
||||
tokio::time::timeout(Duration::from_millis(100), &mut put).await.is_err(),
|
||||
"native replica publication must await the entire rename tail"
|
||||
);
|
||||
put.abort();
|
||||
assert!(put.await.expect_err("the scanner waiter must be canceled").is_cancelled());
|
||||
|
||||
let capacity_lock = other
|
||||
.new_ns_lock(RUSTFS_META_BUCKET, POOL_META_NAME)
|
||||
.await
|
||||
.expect("capacity lock probe");
|
||||
let object_lock = other
|
||||
.new_ns_lock(RUSTFS_META_BUCKET, object)
|
||||
.await
|
||||
.expect("fixed object lock probe");
|
||||
let mut capacity_probe = tokio::spawn(async move { capacity_lock.get_write_lock(Duration::from_secs(30)).await });
|
||||
let mut object_probe = tokio::spawn(async move { object_lock.get_write_lock(Duration::from_secs(30)).await });
|
||||
for (label, probe) in [("capacity", &mut capacity_probe), ("fixed object", &mut object_probe)] {
|
||||
assert!(
|
||||
tokio::time::timeout(Duration::from_millis(100), probe).await.is_err(),
|
||||
"canceling the scanner waiter must retain its {label} fence while rename is pending"
|
||||
);
|
||||
}
|
||||
tail.release();
|
||||
drop(tail);
|
||||
for probe in [capacity_probe, object_probe] {
|
||||
drop(
|
||||
tokio::time::timeout(Duration::from_secs(30), probe)
|
||||
.await
|
||||
.expect("publication fence must drain after rename")
|
||||
.expect("lock probe must not panic")
|
||||
.expect("publication fence must eventually be released"),
|
||||
);
|
||||
}
|
||||
let mut reader = writer.pools[0].disk_set[set_index]
|
||||
.get_object_reader(RUSTFS_META_BUCKET, object, None, HeaderMap::new(), &ObjectOptions::default())
|
||||
.await
|
||||
.expect("canceled waiter must leave the committed replica readable");
|
||||
let mut actual = Vec::new();
|
||||
reader
|
||||
.read_to_end(&mut actual)
|
||||
.await
|
||||
.expect("read the full native replica after tail drain");
|
||||
assert_eq!(actual, body);
|
||||
})
|
||||
.await;
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial_test::serial]
|
||||
fn scanner_backlog_cas_rejects_lost_capacity_lease_before_publication() {
|
||||
run_large_stack_current_thread_async_test("scanner-backlog-lease-loss", async || {
|
||||
let (_temp_dirs, writer, other) = test_three_pool_stores_with_isolated_node_contexts(None).await;
|
||||
let object = "buckets/.scanner-pause-backlog.json";
|
||||
let body = b"native source before lease loss".to_vec();
|
||||
let original = writer
|
||||
.save_scanner_pause_backlog_replica(2, 1, body.clone(), Default::default())
|
||||
.await
|
||||
.expect("seed the exact native replica set");
|
||||
let (lossy, refresh_calls) = store_with_capacity_lease_loss(&other).await;
|
||||
let barrier = PutObjectCommitBarrier::install(RUSTFS_META_BUCKET, object, PutObjectCommitPause::BeforeQuotaRename);
|
||||
let put = tokio::spawn(async move {
|
||||
lossy
|
||||
.save_scanner_pause_backlog_replica(
|
||||
2,
|
||||
1,
|
||||
b"must not commit after lease loss".to_vec(),
|
||||
crate::storage_api_contracts::object::HTTPPreconditions {
|
||||
if_match: original.etag,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
});
|
||||
tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
|
||||
.await
|
||||
.expect("native CAS must reach its commit barrier");
|
||||
tokio::time::pause();
|
||||
tokio::task::yield_now().await;
|
||||
refresh_calls.arm();
|
||||
tokio::time::advance(Duration::from_secs(11)).await;
|
||||
tokio::task::yield_now().await;
|
||||
assert!(
|
||||
refresh_calls.load(Ordering::Acquire) > 0,
|
||||
"the durable metadata lease must lose refresh quorum"
|
||||
);
|
||||
barrier.release();
|
||||
tokio::time::resume();
|
||||
let err = tokio::time::timeout(Duration::from_secs(30), put)
|
||||
.await
|
||||
.expect("native CAS must finish after the barrier release")
|
||||
.expect("native CAS task must not panic")
|
||||
.expect_err("a lost outer capacity lease must reject native publication");
|
||||
assert!(
|
||||
matches!(err, crate::error::Error::NamespaceLockQuorumUnavailable { .. }),
|
||||
"unexpected lease error: {err}"
|
||||
);
|
||||
drop(barrier);
|
||||
let mut reader = writer.pools[2].disk_set[1]
|
||||
.get_object_reader(RUSTFS_META_BUCKET, object, None, HeaderMap::new(), &ObjectOptions::default())
|
||||
.await
|
||||
.expect("the preexisting replica must survive lease loss");
|
||||
let mut actual = Vec::new();
|
||||
reader.read_to_end(&mut actual).await.expect("read the full retained replica");
|
||||
assert_eq!(actual, body);
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial_test::serial]
|
||||
fn scanner_backlog_cas_rejects_a_retiring_source_on_a_stale_node() {
|
||||
run_large_stack_current_thread_async_test("scanner-backlog-source-fence", async || {
|
||||
let (_temp_dirs, store, writer) = test_three_pool_stores_with_three_disk_sets_with_isolated_node_contexts(None).await;
|
||||
let object = "buckets/.scanner-pause-backlog.json";
|
||||
let body = b"frozen native scanner replica".to_vec();
|
||||
let source_set_index = (writer.pools[0].get_disks_by_key(object).set_index + 1) % writer.pools[0].disk_set.len();
|
||||
assert_ne!(
|
||||
source_set_index,
|
||||
writer.pools[0].get_disks_by_key(object).set_index,
|
||||
"exercise a non-routed native set"
|
||||
);
|
||||
let original = writer
|
||||
.save_scanner_pause_backlog_replica(
|
||||
0,
|
||||
source_set_index,
|
||||
body.clone(),
|
||||
crate::storage_api_contracts::object::HTTPPreconditions {
|
||||
if_none_match: Some("*".to_string()),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("seed a native scanner replica before retirement");
|
||||
assert!(
|
||||
writer
|
||||
.scanner_pause_backlog_writable_set_disks()
|
||||
.await
|
||||
.iter()
|
||||
.any(|set| set.pool_index == 0)
|
||||
);
|
||||
|
||||
let layout = DecommissionErasureLayout { data: 1, parity: 0 };
|
||||
let target_total = body.len() * 8;
|
||||
set_decommission_capacity_info_overrides_for_test(
|
||||
store.id,
|
||||
vec![vec![
|
||||
DecommissionPoolCapacityInfo::for_test(0, layout, 0, body.len() * 2, body.len() * 2),
|
||||
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("another node durably retires the selected source");
|
||||
assert!(
|
||||
writer.pool_meta.read().await.pools[0].decommission.is_none(),
|
||||
"the writer must retain a stale snapshot"
|
||||
);
|
||||
assert!(
|
||||
writer
|
||||
.scanner_pause_backlog_writable_set_disks()
|
||||
.await
|
||||
.iter()
|
||||
.any(|set| set.pool_index == 0)
|
||||
);
|
||||
|
||||
let result = writer
|
||||
.save_scanner_pause_backlog_replica(
|
||||
0,
|
||||
source_set_index,
|
||||
b"late native scanner update".to_vec(),
|
||||
crate::storage_api_contracts::object::HTTPPreconditions {
|
||||
if_match: original.etag.clone(),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await;
|
||||
assert!(
|
||||
matches!(result, Err(crate::error::Error::SlowDown)),
|
||||
"late native source publication must fail: {result:?}"
|
||||
);
|
||||
let mut source = writer.pools[0].disk_set[source_set_index]
|
||||
.get_object_reader(RUSTFS_META_BUCKET, object, None, HeaderMap::new(), &ObjectOptions::default())
|
||||
.await
|
||||
.expect("the original source must remain readable");
|
||||
assert_eq!(source.object_info.etag, original.etag);
|
||||
let mut actual = Vec::new();
|
||||
source
|
||||
.read_to_end(&mut actual)
|
||||
.await
|
||||
.expect("read the entire retained source");
|
||||
assert_eq!(actual, body);
|
||||
|
||||
for set in &writer.pools[2].disk_set {
|
||||
let target_body = format!("surviving native scanner set {}", set.set_index).into_bytes();
|
||||
let committed = writer
|
||||
.save_scanner_pause_backlog_replica(
|
||||
2,
|
||||
set.set_index,
|
||||
target_body.clone(),
|
||||
crate::storage_api_contracts::object::HTTPPreconditions {
|
||||
if_none_match: Some("*".to_string()),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("every reserved healthy target set must still accept scanner replicas");
|
||||
let conflict = writer
|
||||
.save_scanner_pause_backlog_replica(
|
||||
2,
|
||||
set.set_index,
|
||||
b"must not bypass CAS".to_vec(),
|
||||
crate::storage_api_contracts::object::HTTPPreconditions {
|
||||
if_match: Some("stale-native-revision".to_string()),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect_err("capacity admission must retain the native writer's CAS");
|
||||
assert!(matches!(conflict, crate::error::Error::PreconditionFailed));
|
||||
let mut target = set
|
||||
.get_object_reader(RUSTFS_META_BUCKET, object, None, HeaderMap::new(), &ObjectOptions::default())
|
||||
.await
|
||||
.expect("read the actual replica set, not the hash-routed set");
|
||||
assert_eq!(target.object_info.etag, committed.etag);
|
||||
let mut actual = Vec::new();
|
||||
target
|
||||
.read_to_end(&mut actual)
|
||||
.await
|
||||
.expect("read the complete native target");
|
||||
assert_eq!(actual, target_body);
|
||||
}
|
||||
for (pool_index, set_index) in [(writer.pools.len(), 0), (0, writer.pools[0].disk_set.len())] {
|
||||
assert!(matches!(
|
||||
writer
|
||||
.save_scanner_pause_backlog_replica(pool_index, set_index, Vec::new(), Default::default())
|
||||
.await,
|
||||
Err(crate::error::Error::InvalidArgument(_, _, _))
|
||||
));
|
||||
}
|
||||
store
|
||||
.decommission_cancel(0)
|
||||
.await
|
||||
.expect("cancel retirement before restoring native membership");
|
||||
writer
|
||||
.save_scanner_pause_backlog_replica(
|
||||
0,
|
||||
source_set_index,
|
||||
b"canceled source membership repair".to_vec(),
|
||||
crate::storage_api_contracts::object::HTTPPreconditions {
|
||||
if_match: original.etag,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("cancel must retain scanner's existing native membership repair contract");
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial_test::serial]
|
||||
fn scanner_backlog_native_replica_reconciles_capacity_and_cleans_source() {
|
||||
run_large_stack_current_thread_async_test("scanner-backlog-reconcile", async || {
|
||||
let (_temp_dirs, store, other_store) =
|
||||
test_three_pool_stores_with_three_disk_sets_with_isolated_node_contexts(None).await;
|
||||
let object = "buckets/.scanner-pause-backlog.json";
|
||||
let body = br#"{"schemaVersion":1,"generation":2}"#.to_vec();
|
||||
let old_body = br#"{"schemaVersion":1,"generation":1}"#.to_vec();
|
||||
let source_time = time::OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(20);
|
||||
let target_time = time::OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(10);
|
||||
for (pool_index, payload, mod_time) in [(0, body.clone(), source_time), (2, old_body, target_time)] {
|
||||
store.pools[pool_index]
|
||||
.put_object(
|
||||
RUSTFS_META_BUCKET,
|
||||
object,
|
||||
&mut PutObjReader::from_vec(payload),
|
||||
&ObjectOptions {
|
||||
max_parity: true,
|
||||
mod_time: Some(mod_time),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("seed native scanner replicas with independent write times");
|
||||
}
|
||||
let layout = DecommissionErasureLayout { data: 1, parity: 0 };
|
||||
let target_total = body.len() * 8;
|
||||
let capacities = vec![
|
||||
DecommissionPoolCapacityInfo::for_test(0, layout, 0, body.len() * 2, body.len() * 2),
|
||||
DecommissionPoolCapacityInfo::for_test(1, layout, 0, target_total, target_total),
|
||||
DecommissionPoolCapacityInfo::for_test(2, layout, target_total, target_total, 0),
|
||||
];
|
||||
set_decommission_capacity_info_overrides_for_test(store.id, vec![capacities.clone()]);
|
||||
store
|
||||
.save_current_pool_meta_for_decommission_start(&[0], Vec::new())
|
||||
.await
|
||||
.expect("activate the source reservation");
|
||||
let owner = decommission_capacity_owner(&*store.pool_meta.read().await);
|
||||
let source_reader = store.pools[0]
|
||||
.get_object_reader(
|
||||
RUSTFS_META_BUCKET,
|
||||
object,
|
||||
None,
|
||||
HeaderMap::new(),
|
||||
&ObjectOptions {
|
||||
no_lock: true,
|
||||
data_movement: true,
|
||||
raw_data_movement_read: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("read the frozen source replica");
|
||||
let conflict = data_movement::migrate_decommission_object(
|
||||
Arc::clone(&store),
|
||||
0,
|
||||
RUSTFS_META_BUCKET.to_string(),
|
||||
source_reader,
|
||||
None,
|
||||
"scanner_backlog_conflict",
|
||||
Some(owner),
|
||||
)
|
||||
.await
|
||||
.expect_err("a different older native ledger must retain its source and capacity intent");
|
||||
assert!(conflict.to_string().contains("Precondition failed"), "unexpected conflict: {conflict}");
|
||||
let mut persisted = crate::core::pools::PoolMeta::default();
|
||||
persisted
|
||||
.load_no_lock_from_replicas(store.pools.clone())
|
||||
.await
|
||||
.expect("reload the unresolved intent");
|
||||
assert_eq!(
|
||||
persisted.pools[0]
|
||||
.decommission
|
||||
.as_ref()
|
||||
.expect("source state")
|
||||
.capacity_reservation
|
||||
.as_ref()
|
||||
.expect("durable capacity")
|
||||
.pending_target_physical_bytes,
|
||||
body.len()
|
||||
);
|
||||
let previous = store.pools[2]
|
||||
.get_object_info(RUSTFS_META_BUCKET, object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("read the native writer's CAS revision");
|
||||
let replacement = store.pools[2]
|
||||
.put_object(
|
||||
RUSTFS_META_BUCKET,
|
||||
object,
|
||||
&mut PutObjReader::from_vec(body.clone()),
|
||||
&ObjectOptions {
|
||||
max_parity: true,
|
||||
mod_time: Some(target_time),
|
||||
http_preconditions: Some(crate::storage_api_contracts::object::HTTPPreconditions {
|
||||
if_match: previous.etag,
|
||||
..Default::default()
|
||||
}),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("native scanner CAS converges the payload without a migration marker");
|
||||
assert!(!data_movement::is_owned_data_movement_target(&replacement));
|
||||
*other_store.pool_meta.write().await = persisted;
|
||||
set_decommission_capacity_info_overrides_for_test(other_store.id, vec![capacities]);
|
||||
tokio::time::timeout(
|
||||
Duration::from_secs(30),
|
||||
other_store.decommission_entry_for_test(
|
||||
0,
|
||||
MetaCacheEntry {
|
||||
name: object.to_string(),
|
||||
..Default::default()
|
||||
},
|
||||
RUSTFS_META_BUCKET.to_string(),
|
||||
other_store.pools[0].get_disks_by_key(object),
|
||||
),
|
||||
)
|
||||
.await
|
||||
.expect("replica conflict recovery must be bounded")
|
||||
.expect("identical native replica should finish migration on the reloaded node");
|
||||
let mut reconciled = crate::core::pools::PoolMeta::default();
|
||||
reconciled
|
||||
.load_no_lock_from_replicas(other_store.pools.clone())
|
||||
.await
|
||||
.expect("reload reconciled capacity");
|
||||
let reservation = reconciled.pools[0]
|
||||
.decommission
|
||||
.as_ref()
|
||||
.expect("source state")
|
||||
.capacity_reservation
|
||||
.as_ref()
|
||||
.expect("reconciled capacity");
|
||||
assert_eq!(reservation.pending_target_physical_bytes, 0);
|
||||
assert_eq!(reservation.committed_data_bytes, body.len());
|
||||
assert_eq!(reservation.consumed_target_physical_bytes, body.len());
|
||||
assert!(reservation.targets.iter().all(|target| target.pending_mutation_id.is_none()));
|
||||
assert_eq!(
|
||||
other_store.pool_meta.read().await.pools[0]
|
||||
.decommission
|
||||
.as_ref()
|
||||
.expect("worker progress")
|
||||
.items_decommission_failed,
|
||||
0
|
||||
);
|
||||
let missing = other_store.pools[0]
|
||||
.get_object_info(RUSTFS_META_BUCKET, object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect_err("the source should be cleaned only after equivalent-target capacity reconciliation");
|
||||
assert!(crate::error::is_err_object_not_found(&missing));
|
||||
let mut target_reader = other_store.pools[2]
|
||||
.get_object_reader(RUSTFS_META_BUCKET, object, None, HeaderMap::new(), &ObjectOptions::default())
|
||||
.await
|
||||
.expect("the surviving replica should remain readable");
|
||||
assert_eq!(
|
||||
target_reader.object_info.mod_time,
|
||||
Some(target_time),
|
||||
"recovery must not overwrite the native target"
|
||||
);
|
||||
let mut actual = Vec::new();
|
||||
target_reader
|
||||
.read_to_end(&mut actual)
|
||||
.await
|
||||
.expect("read surviving ledger bytes");
|
||||
assert_eq!(actual, body);
|
||||
});
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial_test::serial]
|
||||
fn data_movement_equivalent_target_reconciles_published_capacity_after_restart() {
|
||||
|
||||
@@ -984,6 +984,24 @@ fn is_superseding_unversioned_data_movement_object(source: &ObjectInfo, target:
|
||||
.is_some_and(|(source_time, target_time)| target_time > source_time)
|
||||
}
|
||||
|
||||
fn is_equivalent_scanner_backlog_replica(source: &ObjectInfo, target: &ObjectInfo, compare_part_checksums: bool) -> bool {
|
||||
// Scanner publishes this exact payload to surviving sets with CAS. Each
|
||||
// set assigns its own write time; that timestamp is not a ledger generation.
|
||||
// Accept only an identical, known unversioned identity, never a different
|
||||
// record based on timestamp ordering or a similarly named user object.
|
||||
source.bucket == crate::disk::RUSTFS_META_BUCKET
|
||||
&& target.bucket == source.bucket
|
||||
&& source.name == "buckets/.scanner-pause-backlog.json"
|
||||
&& target.name == source.name
|
||||
&& is_unversioned_data_movement_object(source)
|
||||
&& is_unversioned_data_movement_object(target)
|
||||
&& !source.delete_marker
|
||||
&& source.mod_time.is_some()
|
||||
&& target.mod_time.is_some()
|
||||
&& source.etag.as_ref().is_some_and(|etag| !etag.is_empty())
|
||||
&& is_equivalent_data_movement_object_identity(source, target, false, compare_part_checksums)
|
||||
}
|
||||
|
||||
fn is_data_movement_upload_takeover_target(source: &ObjectInfo, target: &ObjectInfo, compare_part_checksums: bool) -> bool {
|
||||
let identity = data_movement_upload_identity(source);
|
||||
source.mod_time.is_some()
|
||||
@@ -1453,7 +1471,9 @@ fn resolve_data_movement_overwrite_resume_result_for(
|
||||
return Ok(true);
|
||||
}
|
||||
|
||||
Ok(matches!(err, Error::PreconditionFailed) && is_superseding_unversioned_data_movement_object(source, &target))
|
||||
Ok(matches!(err, Error::PreconditionFailed)
|
||||
&& (is_equivalent_scanner_backlog_replica(source, &target, compare_part_checksums)
|
||||
|| is_superseding_unversioned_data_movement_object(source, &target)))
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
@@ -3288,6 +3308,132 @@ mod tests {
|
||||
assert!(overwrite_resume_for_target(&source, source.clone()));
|
||||
}
|
||||
|
||||
fn scanner_backlog_replica_pair() -> (ObjectInfo, ObjectInfo) {
|
||||
let source = ObjectInfo {
|
||||
bucket: crate::disk::RUSTFS_META_BUCKET.to_string(),
|
||||
name: "buckets/.scanner-pause-backlog.json".to_string(),
|
||||
version_id: None,
|
||||
mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::SECOND),
|
||||
..overwrite_equivalence_source()
|
||||
};
|
||||
let target = ObjectInfo {
|
||||
mod_time: Some(OffsetDateTime::UNIX_EPOCH),
|
||||
..source.clone()
|
||||
};
|
||||
(source, target)
|
||||
}
|
||||
|
||||
fn scanner_backlog_precondition_resumes(source: &ObjectInfo, target: ObjectInfo) -> bool {
|
||||
resolve_data_movement_overwrite_resume_result_for(&Error::PreconditionFailed, Ok(Some(target)), source, 0, 1, true)
|
||||
.expect("scanner replica conflict should be adjudicated")
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_scanner_backlog_resume_accepts_identical_native_replica_with_older_write_time() {
|
||||
let (source, target) = scanner_backlog_replica_pair();
|
||||
assert!(!is_owned_data_movement_target(&target), "native scanner writes are not migration copies");
|
||||
assert!(!is_equivalent_data_movement_object(&source, &target));
|
||||
assert!(
|
||||
scanner_backlog_precondition_resumes(&source, target),
|
||||
"identical ledger payloads have replica-local write times, not distinct committed generations"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_scanner_backlog_resume_rejects_changed_payload_or_metadata() {
|
||||
let (source, target) = scanner_backlog_replica_pair();
|
||||
let mut different_etag = target.clone();
|
||||
different_etag.etag = Some("different-ledger-generation".to_string());
|
||||
let mut different_size = target.clone();
|
||||
different_size.size += 1;
|
||||
let mut different_checksum = target.clone();
|
||||
different_checksum.checksum = Some(Bytes::from_static(b"different-checksum"));
|
||||
let mut different_metadata = target.clone();
|
||||
Arc::make_mut(&mut different_metadata.user_defined).insert("x-amz-meta-key".to_string(), "different".to_string());
|
||||
let mut different_tags = target.clone();
|
||||
different_tags.user_tags = Arc::new("tag=changed".to_string());
|
||||
let mut different_parts = target.clone();
|
||||
Arc::make_mut(&mut different_parts.parts)[0].etag = "different-part".to_string();
|
||||
let mut different_tier = target;
|
||||
different_tier.transitioned_object.tier = "different-tier".to_string();
|
||||
for (label, different) in [
|
||||
("etag", different_etag),
|
||||
("size", different_size),
|
||||
("checksum", different_checksum),
|
||||
("metadata", different_metadata),
|
||||
("tags", different_tags),
|
||||
("parts", different_parts),
|
||||
("tier", different_tier),
|
||||
] {
|
||||
assert!(
|
||||
!scanner_backlog_precondition_resumes(&source, different),
|
||||
"replica-local timestamps do not authorize a changed {label}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_scanner_backlog_resume_rejects_other_namespaces_and_incomplete_identity() {
|
||||
let (source, target) = scanner_backlog_replica_pair();
|
||||
for (bucket, name) in [
|
||||
("user-bucket", "buckets/.scanner-pause-backlog.json"),
|
||||
(crate::disk::RUSTFS_META_BUCKET, "buckets/.scanner-pause-backlog.json.bkp"),
|
||||
(crate::disk::RUSTFS_META_BUCKET, "buckets/.usage-cache.bin"),
|
||||
] {
|
||||
let mut source = source.clone();
|
||||
let mut target = target.clone();
|
||||
for replica in [&mut source, &mut target] {
|
||||
replica.bucket = bucket.to_string();
|
||||
replica.name = name.to_string();
|
||||
}
|
||||
assert!(!scanner_backlog_precondition_resumes(&source, target), "out-of-scope key {bucket}/{name}");
|
||||
}
|
||||
for missing in ["etag", "empty-etag", "source-time", "target-time", "version", "delete-marker"] {
|
||||
let mut source = source.clone();
|
||||
let mut target = target.clone();
|
||||
match missing {
|
||||
"etag" => {
|
||||
source.etag = None;
|
||||
target.etag = None;
|
||||
}
|
||||
"empty-etag" => {
|
||||
source.etag = Some(String::new());
|
||||
target.etag = Some(String::new());
|
||||
}
|
||||
"source-time" => source.mod_time = None,
|
||||
"target-time" => target.mod_time = None,
|
||||
"version" => {
|
||||
source.version_id = Some(Uuid::from_u128(1));
|
||||
target.version_id = source.version_id;
|
||||
}
|
||||
"delete-marker" => {
|
||||
source.delete_marker = true;
|
||||
target.delete_marker = true;
|
||||
}
|
||||
_ => unreachable!("all identity variants are enumerated above"),
|
||||
}
|
||||
assert!(!scanner_backlog_precondition_resumes(&source, target), "unsupported identity: {missing}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_scanner_backlog_resume_requires_a_cross_pool_precondition_conflict() {
|
||||
let (source, target) = scanner_backlog_replica_pair();
|
||||
for (err, target_pool) in [
|
||||
(Error::PreconditionFailed, 0),
|
||||
(Error::SlowDown, 1),
|
||||
(
|
||||
Error::InvalidUploadID(source.bucket.clone(), source.name.clone(), "upload".to_string()),
|
||||
1,
|
||||
),
|
||||
] {
|
||||
assert!(
|
||||
!resolve_data_movement_overwrite_resume_result_for(&err, Ok(Some(target.clone())), &source, 0, target_pool, true)
|
||||
.expect("non-resumable conflict should return false")
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_data_movement_overwrite_resume_accepts_part_mod_time_drift() {
|
||||
let source = overwrite_equivalence_source();
|
||||
|
||||
@@ -20,6 +20,8 @@
|
||||
//! contract stays implemented `for SetDisks`, so its associated-type bounds are
|
||||
//! unchanged; method bodies are moved verbatim and runtime behavior is the same.
|
||||
|
||||
use crate::core::pools::DecommissionCapacityAdmission;
|
||||
|
||||
#[cfg(test)]
|
||||
use super::super::GetObjectMetadataCacheKey;
|
||||
#[cfg(test)]
|
||||
@@ -1809,7 +1811,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
let decommission_capacity_guard = if let Some(store) = opts.decommission_capacity_admission.as_ref() {
|
||||
Some(
|
||||
store
|
||||
.acquire_external_decommission_capacity_fence(&[self.pool_index], "mutation")
|
||||
.acquire_external_decommission_capacity_fence(
|
||||
&[self.pool_index],
|
||||
DecommissionCapacityAdmission::ExistingMultipart,
|
||||
)
|
||||
.await?,
|
||||
)
|
||||
} else {
|
||||
@@ -2345,6 +2350,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
bucket,
|
||||
object,
|
||||
opts.no_lock || object_lock_guard.is_some(),
|
||||
DecommissionCapacityAdmission::ExistingMultipart,
|
||||
)
|
||||
.await?;
|
||||
decommission_object_lock_guard = object_guard;
|
||||
@@ -3110,7 +3116,10 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
{
|
||||
decommission_capacity_guard = Some(
|
||||
store
|
||||
.acquire_external_decommission_capacity_fence(&[self.pool_index], "mutation")
|
||||
.acquire_external_decommission_capacity_fence(
|
||||
&[self.pool_index],
|
||||
DecommissionCapacityAdmission::ExistingMultipart,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -19,6 +19,8 @@
|
||||
//! bounds are unchanged, and the impls reach shared primitives through the
|
||||
//! SetDisks core (io_primitives) via inherent calls.
|
||||
|
||||
use crate::core::pools::DecommissionCapacityAdmission;
|
||||
|
||||
#[cfg(test)]
|
||||
use super::super::MetadataCacheInvalidationProbe;
|
||||
use super::super::{
|
||||
@@ -3905,6 +3907,7 @@ impl SetDisks {
|
||||
bucket,
|
||||
object,
|
||||
opts.no_lock || object_lock_guard.is_some(),
|
||||
DecommissionCapacityAdmission::Mutation,
|
||||
)
|
||||
.await?;
|
||||
decommission_object_lock_guard = object_guard;
|
||||
@@ -4102,7 +4105,7 @@ impl SetDisks {
|
||||
{
|
||||
decommission_capacity_guard = Some(
|
||||
store
|
||||
.acquire_external_decommission_capacity_fence(&[self.pool_index], "mutation")
|
||||
.acquire_external_decommission_capacity_fence(&[self.pool_index], DecommissionCapacityAdmission::Mutation)
|
||||
.await?,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -17,6 +17,7 @@ use super::{
|
||||
UpdateMetadataOpts, Uuid, X_AMZ_RESTORE, get_raw_etag, restore_operation_id_from_metadata,
|
||||
};
|
||||
use crate::bucket::lifecycle::lifecycle;
|
||||
use crate::core::pools::DecommissionCapacityAdmission;
|
||||
use rustfs_filemeta::RestoreStatusOps;
|
||||
use rustfs_utils::http::headers::{AMZ_RESTORE_EXPIRY_DAYS, AMZ_RESTORE_REQUEST_DATE};
|
||||
use s3s::dto::{RestoreStatus, Timestamp};
|
||||
@@ -160,7 +161,13 @@ impl SetDisks {
|
||||
let (decommission_object_lock_guard, decommission_target_lock_covered, mut decommission_capacity_guard) =
|
||||
if let Some(store) = opts.decommission_capacity_admission.as_ref() {
|
||||
store
|
||||
.acquire_external_decommission_commit_guards(self.pool_index, bucket, object, opts.no_lock)
|
||||
.acquire_external_decommission_commit_guards(
|
||||
self.pool_index,
|
||||
bucket,
|
||||
object,
|
||||
opts.no_lock,
|
||||
DecommissionCapacityAdmission::Mutation,
|
||||
)
|
||||
.await?
|
||||
} else {
|
||||
(None, false, None)
|
||||
@@ -178,7 +185,7 @@ impl SetDisks {
|
||||
{
|
||||
decommission_capacity_guard = Some(
|
||||
store
|
||||
.acquire_external_decommission_capacity_fence(&[self.pool_index], "mutation")
|
||||
.acquire_external_decommission_capacity_fence(&[self.pool_index], DecommissionCapacityAdmission::Mutation)
|
||||
.await?,
|
||||
);
|
||||
}
|
||||
@@ -264,7 +271,13 @@ impl SetDisks {
|
||||
let (decommission_object_lock_guard, decommission_target_lock_covered, mut decommission_capacity_guard) =
|
||||
if let Some(store) = opts.decommission_capacity_admission.as_ref() {
|
||||
store
|
||||
.acquire_external_decommission_commit_guards(self.pool_index, bucket, object, opts.no_lock)
|
||||
.acquire_external_decommission_commit_guards(
|
||||
self.pool_index,
|
||||
bucket,
|
||||
object,
|
||||
opts.no_lock,
|
||||
DecommissionCapacityAdmission::Mutation,
|
||||
)
|
||||
.await?
|
||||
} else {
|
||||
(None, false, None)
|
||||
@@ -282,7 +295,7 @@ impl SetDisks {
|
||||
{
|
||||
decommission_capacity_guard = Some(
|
||||
store
|
||||
.acquire_external_decommission_capacity_fence(&[self.pool_index], "mutation")
|
||||
.acquire_external_decommission_capacity_fence(&[self.pool_index], DecommissionCapacityAdmission::Mutation)
|
||||
.await?,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -671,9 +671,9 @@ mod tests {
|
||||
use crate::cluster::rpc::PeerS3Client;
|
||||
use crate::config::com::{delete_config, read_config_no_lock_preserve_empty_with_metadata, save_config};
|
||||
use crate::core::pools::{
|
||||
DecommissionCapacityLockOrderBarrier, DecommissionErasureLayout, DecommissionPoolCapacityInfo, POOL_META_IDENTITY_NAME,
|
||||
PoolDecommissionInfo, PoolMetaReplicaState, PoolStatus, initialized_pool_meta_identity_for_test,
|
||||
set_decommission_capacity_info_overrides_for_test,
|
||||
DecommissionCapacityAdmission, DecommissionCapacityLockOrderBarrier, DecommissionErasureLayout,
|
||||
DecommissionPoolCapacityInfo, POOL_META_IDENTITY_NAME, PoolDecommissionInfo, PoolMetaReplicaState, PoolStatus,
|
||||
initialized_pool_meta_identity_for_test, set_decommission_capacity_info_overrides_for_test,
|
||||
};
|
||||
use crate::core::sets::HealFormatAfterSaveBarrier;
|
||||
use crate::disk::error::Result as DiskResult;
|
||||
@@ -1149,7 +1149,7 @@ mod tests {
|
||||
let (temp_dir, store, shutdown) = multi_pool_heal_store().await;
|
||||
let target = remove_heal_test_format(&temp_dir, &store, 0, 3).await;
|
||||
let capacity_guard = store
|
||||
.acquire_external_decommission_capacity_fence(&[0], "heal")
|
||||
.acquire_external_decommission_capacity_fence(&[0], DecommissionCapacityAdmission::Heal)
|
||||
.await
|
||||
.expect("ordinary heal capacity fence should be acquired");
|
||||
|
||||
|
||||
@@ -4198,13 +4198,60 @@ mod tests {
|
||||
.await
|
||||
.expect("suspended source versions should be readable")
|
||||
.expect("suspended source must exist before worker convergence");
|
||||
assert_eq!(versions.versions.len(), 1, "DELETE must not add a marker to the retiring source");
|
||||
let source = &versions.versions[0];
|
||||
assert!(
|
||||
versions
|
||||
.versions
|
||||
.iter()
|
||||
.any(|version| !version.deleted && version.version_id.is_none_or(|version_id| version_id.is_nil())),
|
||||
"the source pool must retain its null data version while DELETE owns the fixed fence"
|
||||
!source.deleted && source.version_id.is_none_or(|version_id| version_id.is_nil()),
|
||||
"the source pool must retain its null data version until worker convergence"
|
||||
);
|
||||
assert_eq!(source.mod_time, Some(OffsetDateTime::UNIX_EPOCH + time::Duration::SECOND));
|
||||
|
||||
let mut reader = store.pools[0]
|
||||
.get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
|
||||
.await
|
||||
.expect("the retiring source must remain directly readable before worker convergence");
|
||||
let mut body = Vec::new();
|
||||
reader
|
||||
.stream
|
||||
.read_to_end(&mut body)
|
||||
.await
|
||||
.expect("read retained source bytes");
|
||||
assert_eq!(body, b"suspended source generation");
|
||||
}
|
||||
|
||||
async fn assert_suspended_null_delete_marker_visible(
|
||||
store: &Arc<crate::store::ECStore>,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
marker_mod_time: OffsetDateTime,
|
||||
) {
|
||||
let versions = store.pools[1]
|
||||
.get_disks_by_key(object)
|
||||
.load_file_info_versions_exact(bucket, object)
|
||||
.await
|
||||
.expect("healthy target versions should be readable")
|
||||
.expect("the healthy target must retain the DELETE marker");
|
||||
assert_eq!(versions.versions.len(), 1, "the target must contain only the null delete marker");
|
||||
let marker = &versions.versions[0];
|
||||
assert!(marker.deleted, "migration must not replace the DELETE marker with source data");
|
||||
assert!(marker.version_id.is_none_or(|version_id| version_id.is_nil()));
|
||||
assert_eq!(marker.size, 0);
|
||||
assert_eq!(marker.mod_time, Some(marker_mod_time), "migration must preserve the marker generation");
|
||||
assert!(marker_mod_time > OffsetDateTime::UNIX_EPOCH + time::Duration::SECOND);
|
||||
|
||||
let head_err = store
|
||||
.get_object_info(bucket, object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect_err("HEAD must observe the DELETE marker instead of the old null source");
|
||||
assert!(matches!(head_err, Error::ObjectNotFound(_, _)), "unexpected HEAD result: {head_err:?}");
|
||||
let get_err = match store
|
||||
.get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
|
||||
.await
|
||||
{
|
||||
Ok(_) => panic!("GET must not resurrect the deleted null source"),
|
||||
Err(err) => err,
|
||||
};
|
||||
assert!(matches!(get_err, Error::ObjectNotFound(_, _)), "unexpected GET result: {get_err:?}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -8042,7 +8089,7 @@ mod tests {
|
||||
write_suspended_decommission_source(&store, &bucket, object).await;
|
||||
mark_test_pool_decommissioning(&store, 0).await;
|
||||
|
||||
let delete_err = store
|
||||
let deleted = store
|
||||
.delete_object(
|
||||
&bucket,
|
||||
object,
|
||||
@@ -8052,12 +8099,12 @@ mod tests {
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect_err("capacity-reserved target must reject a concurrent suspended DELETE");
|
||||
assert!(
|
||||
matches!(delete_err, Error::SlowDown),
|
||||
"unexpected suspended DELETE result: {delete_err:?}"
|
||||
);
|
||||
.expect("a healthy reserved target must accept suspended DELETE");
|
||||
assert!(deleted.delete_marker);
|
||||
assert_eq!(deleted.version_id, Some(uuid::Uuid::nil()));
|
||||
let marker_mod_time = deleted.mod_time.expect("DELETE must return the marker generation");
|
||||
assert_suspended_null_source_present(&store, &bucket, object).await;
|
||||
assert_suspended_null_delete_marker_visible(&store, &bucket, object, marker_mod_time).await;
|
||||
|
||||
let source_set = store.pools[0].get_disks_by_key(object);
|
||||
let worker_store = Arc::clone(&store);
|
||||
@@ -8077,7 +8124,7 @@ mod tests {
|
||||
})
|
||||
.await
|
||||
.expect("suspended decommission worker should join")
|
||||
.expect("worker must migrate the fenced suspended source");
|
||||
.expect("worker must converge the old null source behind the newer DELETE marker");
|
||||
|
||||
assert_decommission_source_absent(
|
||||
&store,
|
||||
@@ -8089,10 +8136,7 @@ mod tests {
|
||||
},
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
read_decommission_target_body(&store, &bucket, object, &ObjectOptions::default()).await,
|
||||
b"suspended source generation"
|
||||
);
|
||||
assert_suspended_null_delete_marker_visible(&store, &bucket, object, marker_mod_time).await;
|
||||
shutdown.cancel();
|
||||
}
|
||||
|
||||
@@ -8125,7 +8169,7 @@ mod tests {
|
||||
},
|
||||
None,
|
||||
));
|
||||
let (_deleted, errors) = store
|
||||
let (deleted, errors) = store
|
||||
.delete_objects(
|
||||
&bucket,
|
||||
vec![ObjectToDelete {
|
||||
@@ -8139,10 +8183,23 @@ mod tests {
|
||||
)
|
||||
.await;
|
||||
assert!(
|
||||
matches!(errors.as_slice(), [Some(Error::SlowDown)]),
|
||||
matches!(errors.as_slice(), [None]),
|
||||
"unexpected suspended batch DELETE result: {errors:?}"
|
||||
);
|
||||
assert_eq!(deleted.len(), 1);
|
||||
assert!(deleted[0].delete_marker);
|
||||
assert_eq!(deleted[0].object_name, object);
|
||||
assert!(
|
||||
deleted[0]
|
||||
.delete_marker_version_id
|
||||
.is_none_or(|version_id| version_id.is_nil()),
|
||||
"batch DELETE must retain the native null version identity"
|
||||
);
|
||||
let marker_mod_time = deleted[0]
|
||||
.delete_marker_mtime
|
||||
.expect("batch DELETE must return the marker generation");
|
||||
assert_suspended_null_source_present(&store, &bucket, object).await;
|
||||
assert_suspended_null_delete_marker_visible(&store, &bucket, object, marker_mod_time).await;
|
||||
|
||||
let source_set = store.pools[0].get_disks_by_key(object);
|
||||
let worker_store = Arc::clone(&store);
|
||||
@@ -8162,7 +8219,7 @@ mod tests {
|
||||
})
|
||||
.await
|
||||
.expect("suspended batch decommission worker should join")
|
||||
.expect("worker must migrate the batch-fenced suspended source");
|
||||
.expect("worker must converge the old null source behind the newer batch DELETE marker");
|
||||
|
||||
assert_decommission_source_absent(
|
||||
&store,
|
||||
@@ -8174,10 +8231,7 @@ mod tests {
|
||||
},
|
||||
)
|
||||
.await;
|
||||
assert_eq!(
|
||||
read_decommission_target_body(&store, &bucket, object, &ObjectOptions::default()).await,
|
||||
b"suspended source generation"
|
||||
);
|
||||
assert_suspended_null_delete_marker_visible(&store, &bucket, object, marker_mod_time).await;
|
||||
shutdown.cancel();
|
||||
}
|
||||
|
||||
|
||||
@@ -762,13 +762,7 @@ impl ECStore {
|
||||
self.pools
|
||||
.iter()
|
||||
.enumerate()
|
||||
.filter(|(pool_index, _)| {
|
||||
!pool_meta.pools.get(*pool_index).is_some_and(|pool| {
|
||||
pool.decommission
|
||||
.as_ref()
|
||||
.is_some_and(|info| info.has_decommission_state() && !info.failed && !info.canceled)
|
||||
})
|
||||
})
|
||||
.filter(|(pool_index, _)| pool_meta.scanner_pause_backlog_pool_writable(*pool_index))
|
||||
.flat_map(|(_, pool)| pool.disk_set.iter().cloned())
|
||||
.collect()
|
||||
}
|
||||
|
||||
@@ -37,7 +37,7 @@ use crate::bucket::object_lock::objectlock_sys::{
|
||||
};
|
||||
use crate::bucket::replication::{DeleteReplicationConfigSnapshot, ReplicationObjectBridge};
|
||||
use crate::bucket::versioning::VersioningApi;
|
||||
use crate::core::pools::{DecommissionCapacityOwner, ensure_decommission_capacity_mutation_id};
|
||||
use crate::core::pools::{DecommissionCapacityAdmission, DecommissionCapacityOwner, ensure_decommission_capacity_mutation_id};
|
||||
use crate::disk::OldCurrentSize;
|
||||
use crate::object_api::{
|
||||
NamespaceLockFence, ObjectLockConfigSnapshot, ScannerPublicationCommitScopeGuard, ScannerPublicationCommitState,
|
||||
@@ -3060,6 +3060,68 @@ impl ECStore {
|
||||
)))
|
||||
}
|
||||
|
||||
/// Publish a native scanner replica without allowing stale pool selection
|
||||
/// to race retirement. Failed/canceled membership repair remains permitted.
|
||||
/// A canceled waiter cannot release publication fences from an in-flight write.
|
||||
pub async fn save_scanner_pause_backlog_replica(
|
||||
self: &Arc<Self>,
|
||||
pool_index: usize,
|
||||
set_index: usize,
|
||||
data: Vec<u8>,
|
||||
preconditions: crate::storage_api_contracts::object::HTTPPreconditions,
|
||||
) -> Result<ObjectInfo> {
|
||||
let set = self
|
||||
.pools
|
||||
.get(pool_index)
|
||||
.and_then(|pool| pool.disk_set.get(set_index))
|
||||
.ok_or_else(|| Error::InvalidArgument("scanner-backlog".into(), "replica".into(), "unknown pool or set".into()))?;
|
||||
let set = Arc::clone(set);
|
||||
let store = Arc::clone(self);
|
||||
let write = async move {
|
||||
let object = "buckets/.scanner-pause-backlog.json";
|
||||
let mut opts = ObjectOptions {
|
||||
max_parity: true,
|
||||
http_preconditions: Some(preconditions),
|
||||
write_completion: crate::object_api::WriteCompletion::TailDrained,
|
||||
..Default::default()
|
||||
};
|
||||
// Match migration: fixed object namespace -> durable pool metadata ->
|
||||
// actual replica namespace. The replica need not be the hash-routed set.
|
||||
let object_guard = if store.single_pool() {
|
||||
None
|
||||
} else {
|
||||
Some(
|
||||
store
|
||||
.acquire_object_write_lock("scanner_backlog_replica", RUSTFS_META_BUCKET, object)
|
||||
.await?,
|
||||
)
|
||||
};
|
||||
let capacity_guard = if let Some(guard) = object_guard.as_ref() {
|
||||
guard.add_namespace_lock_fence(&mut opts);
|
||||
opts.no_lock = match store.pools.first().and_then(|pool| pool.disk_set.first()) {
|
||||
Some(fixed) => fixed.shares_namespace_lock_domain(&set).await,
|
||||
None => false,
|
||||
};
|
||||
let capacity_guard = store
|
||||
.acquire_external_decommission_capacity_fence(&[pool_index], DecommissionCapacityAdmission::ScannerBacklog)
|
||||
.await?;
|
||||
opts.add_namespace_lock_guard(&capacity_guard);
|
||||
Some(capacity_guard)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let result = set
|
||||
.put_object(RUSTFS_META_BUCKET, object, &mut PutObjReader::from_vec(data), &opts)
|
||||
.await;
|
||||
drop(capacity_guard);
|
||||
drop(object_guard);
|
||||
result
|
||||
};
|
||||
// The set layer may detach its rename owner, even for full-tail writes.
|
||||
// Keep these outer guards alive until that owner finishes if scanner exits.
|
||||
tokio::spawn(write).await.map_err(Error::from)?
|
||||
}
|
||||
|
||||
pub(super) async fn run_external_decommission_capacity_object_mutation<T, F, Fut>(
|
||||
&self,
|
||||
target_pool_idx: usize,
|
||||
@@ -3129,8 +3191,11 @@ impl ECStore {
|
||||
let (capacity_guard, has_active_decommission) = if capacity_releasing {
|
||||
self.acquire_decommission_capacity_release_fence_with_active_source().await?
|
||||
} else {
|
||||
self.acquire_external_decommission_capacity_fence_with_active_source(&[target_pool_idx], "mutation")
|
||||
.await?
|
||||
self.acquire_external_decommission_capacity_fence_with_active_source(
|
||||
&[target_pool_idx],
|
||||
DecommissionCapacityAdmission::Mutation,
|
||||
)
|
||||
.await?
|
||||
};
|
||||
let (capacity_guard, object_guard) = if has_active_decommission && !opts.no_lock {
|
||||
// Active migration acquires the object namespace before its capacity
|
||||
@@ -3146,7 +3211,7 @@ impl ECStore {
|
||||
let capacity_guard = if capacity_releasing {
|
||||
self.acquire_decommission_capacity_release_fence_with_active_source().await?.0
|
||||
} else {
|
||||
self.acquire_external_decommission_capacity_fence(&[target_pool_idx], "mutation")
|
||||
self.acquire_external_decommission_capacity_fence(&[target_pool_idx], DecommissionCapacityAdmission::Mutation)
|
||||
.await?
|
||||
};
|
||||
(capacity_guard, Some(guard))
|
||||
@@ -3176,9 +3241,10 @@ impl ECStore {
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
no_lock: bool,
|
||||
admission: DecommissionCapacityAdmission,
|
||||
) -> Result<(Option<ObjectLockDiagGuard>, bool, Option<rustfs_lock::NamespaceLockGuard>)> {
|
||||
let (capacity_guard, has_active_decommission) = self
|
||||
.acquire_external_decommission_capacity_fence_with_active_source(&[target_pool_idx], "mutation")
|
||||
.acquire_external_decommission_capacity_fence_with_active_source(&[target_pool_idx], admission)
|
||||
.await?;
|
||||
if !has_active_decommission {
|
||||
// Keep the read probe through the staged commit. This closes the
|
||||
@@ -3227,7 +3293,10 @@ impl ECStore {
|
||||
return operation(opts).await;
|
||||
}
|
||||
let (capacity_guard, has_active_decommission) = self
|
||||
.acquire_external_decommission_capacity_fence_with_active_source(&[target_pool_idx], "heal")
|
||||
.acquire_external_decommission_capacity_fence_with_active_source(
|
||||
&[target_pool_idx],
|
||||
DecommissionCapacityAdmission::Heal,
|
||||
)
|
||||
.await?;
|
||||
let (capacity_guard, object_guard) = if has_active_decommission && !opts.no_lock {
|
||||
// Active migration acquires the object namespace before its capacity
|
||||
@@ -3248,7 +3317,7 @@ impl ECStore {
|
||||
None => false,
|
||||
};
|
||||
let capacity_guard = self
|
||||
.acquire_external_decommission_capacity_fence(&[target_pool_idx], "heal")
|
||||
.acquire_external_decommission_capacity_fence(&[target_pool_idx], DecommissionCapacityAdmission::Heal)
|
||||
.await?;
|
||||
opts.no_lock = target_lock_covered;
|
||||
(capacity_guard, Some(guard))
|
||||
@@ -5093,9 +5162,14 @@ impl ECStore {
|
||||
}
|
||||
|
||||
let _capacity_fence = if !self.single_pool() && latest_marker_objects.iter().any(|creates_marker| *creates_marker) {
|
||||
let target_pool_indices = (0..self.pools.len()).collect::<Vec<_>>();
|
||||
// Only marker destinations can grow. Other pools participate in
|
||||
// exact deletion under the same metadata read fence and must not
|
||||
// be treated as publication targets merely because they retire.
|
||||
let mut target_pool_indices = marker_target_pool_indices.iter().flatten().copied().collect::<Vec<_>>();
|
||||
target_pool_indices.sort_unstable();
|
||||
target_pool_indices.dedup();
|
||||
match self
|
||||
.acquire_external_decommission_capacity_fence(&target_pool_indices, "batch_delete")
|
||||
.acquire_external_decommission_capacity_fence(&target_pool_indices, DecommissionCapacityAdmission::BatchDelete)
|
||||
.await
|
||||
{
|
||||
Ok(fence) => Some(fence),
|
||||
|
||||
@@ -23,9 +23,7 @@ use super::ScannerCycleOutcome;
|
||||
use crate::data_usage_define::DataUsageCacheRevision;
|
||||
use crate::storage_api::ScannerStorage;
|
||||
use crate::storage_api::owner::ObjectIO as _;
|
||||
use crate::{
|
||||
BUCKET_META_PREFIX, ECStore, EcstoreError, RUSTFS_META_BUCKET, ScannerObjectOptions, SetDisks, save_config_with_preconditions,
|
||||
};
|
||||
use crate::{BUCKET_META_PREFIX, ECStore, EcstoreError, RUSTFS_META_BUCKET, ScannerObjectOptions, SetDisks};
|
||||
use futures::future::join_all;
|
||||
use http::HeaderMap;
|
||||
use serde::{Deserialize, Serialize};
|
||||
@@ -1195,13 +1193,14 @@ where
|
||||
};
|
||||
let revision = revisions.get(&id).cloned();
|
||||
let data = data.clone();
|
||||
let storeapi = storeapi.clone();
|
||||
async move {
|
||||
let Some(revision) = revision else {
|
||||
return (id, Err("replica revision is unavailable".to_string()));
|
||||
};
|
||||
let result = save_config_with_preconditions(set, SCANNER_PAUSE_BACKLOG_PATH.as_str(), data, revision.preconditions())
|
||||
let result = storeapi
|
||||
.save_scanner_pause_backlog_replica(id.pool_index, id.set_index, data, revision.preconditions())
|
||||
.await
|
||||
.map(|_| ())
|
||||
.map_err(|err| err.to_string());
|
||||
(id, result)
|
||||
}
|
||||
|
||||
@@ -152,6 +152,49 @@ fn run_data_scanner_keeps_its_two_argument_api() {
|
||||
assert_run_data_scanner_signature(run_data_scanner);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn native_backlog_replica_writes_preserve_cas_across_writer_restart() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store_with_pool_count(false, 2).await;
|
||||
let now = scanner_pause_backlog_now();
|
||||
let mut stale = ScannerPauseBacklogController::claim(store.clone(), now)
|
||||
.await
|
||||
.expect("the original scanner writer must publish to both pools");
|
||||
let original = scanner_pause_backlog_status(store.clone()).await;
|
||||
assert!(original.durable);
|
||||
assert_eq!(original.healthy_replicas, 2);
|
||||
let restarted = restart_scanner_cycle_store_from(&store).await;
|
||||
let _replacement = ScannerPauseBacklogController::claim(restarted.clone(), now.saturating_add(1))
|
||||
.await
|
||||
.expect("the restarted scanner must claim the surviving native replicas");
|
||||
let claimed = scanner_pause_backlog_status(restarted.clone()).await;
|
||||
assert!(claimed.writer_epoch > original.writer_epoch);
|
||||
assert_eq!(claimed.healthy_replicas, 2);
|
||||
|
||||
stale
|
||||
.observe(ScannerPauseBacklogObservation {
|
||||
now_unix_secs: now.saturating_add(2),
|
||||
paused: true,
|
||||
movement_generation: store.scanner_data_movement_generation().saturating_add(1),
|
||||
movement_work_items: 1,
|
||||
pause_started_at_unix_secs: now.saturating_add(2),
|
||||
dirty_usage_buckets: 0,
|
||||
discovered_expiry_items: 0,
|
||||
discovered_transition_items: 0,
|
||||
})
|
||||
.await;
|
||||
let retained = scanner_pause_backlog_status(restarted.clone()).await;
|
||||
assert_eq!(retained.writer_epoch, claimed.writer_epoch);
|
||||
assert_eq!(retained.generation, claimed.generation);
|
||||
assert_eq!(retained.phase, ScannerPauseBacklogPhase::Idle);
|
||||
assert_eq!(retained.healthy_replicas, 2);
|
||||
assert_eq!(retained.stale_or_unavailable_replicas, 0);
|
||||
let _recovered = ScannerPauseBacklogController::claim(restarted.clone(), now.saturating_add(3))
|
||||
.await
|
||||
.expect("a fresh writer must still recover after the stale CAS failure");
|
||||
assert!(scanner_pause_backlog_status(restarted).await.error.is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn restarted_main_loop_completes_durable_pause_backlog_catch_up() {
|
||||
crate::scanner_io::clear_dirty_usage_buckets_for_tests();
|
||||
|
||||
@@ -355,6 +355,13 @@ pub(crate) trait ScannerStorage:
|
||||
async fn list_bucket_for_scanner(&self, opts: &storage_contracts::BucketOptions) -> EcstoreResultType<ScannerBucketListing>;
|
||||
fn all_set_disks(&self) -> Vec<Arc<EcstoreSetDisks>>;
|
||||
async fn scanner_pause_backlog_writable_set_disks(&self) -> Vec<Arc<EcstoreSetDisks>>;
|
||||
async fn save_scanner_pause_backlog_replica(
|
||||
self: Arc<Self>,
|
||||
pool_index: usize,
|
||||
set_index: usize,
|
||||
data: Vec<u8>,
|
||||
preconditions: storage_contracts::HTTPPreconditions,
|
||||
) -> EcstoreResultType<()>;
|
||||
#[cfg(test)]
|
||||
fn scanner_observed_probe_store_key(&self) -> usize;
|
||||
}
|
||||
@@ -417,6 +424,18 @@ impl ScannerStorage for EcstoreStore {
|
||||
EcstoreStore::scanner_pause_backlog_writable_set_disks(self).await
|
||||
}
|
||||
|
||||
async fn save_scanner_pause_backlog_replica(
|
||||
self: Arc<Self>,
|
||||
pool_index: usize,
|
||||
set_index: usize,
|
||||
data: Vec<u8>,
|
||||
preconditions: storage_contracts::HTTPPreconditions,
|
||||
) -> EcstoreResultType<()> {
|
||||
EcstoreStore::save_scanner_pause_backlog_replica(&self, pool_index, set_index, data, preconditions)
|
||||
.await
|
||||
.map(|_| ())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn scanner_observed_probe_store_key(&self) -> usize {
|
||||
std::ptr::from_ref(self).cast::<()>() as usize
|
||||
@@ -577,6 +596,20 @@ mod tests {
|
||||
Vec::new()
|
||||
}
|
||||
|
||||
async fn save_scanner_pause_backlog_replica(
|
||||
self: Arc<Self>,
|
||||
_pool_index: usize,
|
||||
_set_index: usize,
|
||||
_data: Vec<u8>,
|
||||
_preconditions: storage_contracts::HTTPPreconditions,
|
||||
) -> EcstoreResultType<()> {
|
||||
Err(EcstoreErrorType::InvalidArgument(
|
||||
"scanner-backlog".into(),
|
||||
"replica".into(),
|
||||
"fake storage has no writable replicas".into(),
|
||||
))
|
||||
}
|
||||
|
||||
fn scanner_observed_probe_store_key(&self) -> usize {
|
||||
0
|
||||
}
|
||||
|
||||
@@ -7,6 +7,8 @@
|
||||
|
||||
RustFS supports queued multi-pool decommission start requests on multi-pool deployments. The admin handler accepts the MinIO-compatible request shape, including comma-separated pool targets. An empty target list is rejected; single-pool deployments reject decommission because there is no destination pool; on multi-pool deployments one or more valid target pools are accepted as a single queued operation.
|
||||
|
||||
Deterministic request rejections (unsupported single-pool operations, missing or terminal targets, an empty start request, removing the last active pool, and clearing unresolved recovery entries) retain the typed `InvalidArgument` error and its actionable reason. Active-operation conflicts retain their existing `InvalidRequest` or `OperationAborted` contract. Storage, quorum, and fleet-proof failures are not converted into argument errors.
|
||||
|
||||
### Request Semantics
|
||||
|
||||
`POST /v3/pools/decommission` with comma-separated pool targets is a queue submission:
|
||||
@@ -15,7 +17,8 @@ RustFS supports queued multi-pool decommission start requests on multi-pool depl
|
||||
- reject duplicate target pools in the same request;
|
||||
- reject active or queued target pools;
|
||||
- reject completed decommission targets, because completion means the pool can be removed from the deployment configuration;
|
||||
- allow failed or canceled targets to be retried;
|
||||
- require failed or canceled targets to be cleared before restarting, except
|
||||
when unresolved listing entries require an explicit recovery retry;
|
||||
- persist queued metadata before starting workers;
|
||||
- start only the local-leader prefix of the queue on the receiving node.
|
||||
|
||||
@@ -56,6 +59,29 @@ Cancel separates active and queued behavior:
|
||||
|
||||
Cancel requests can be accepted on non-leader nodes as remote cancel intent; the leader observes the pending cancel and applies it to the active worker.
|
||||
|
||||
`queuedBuckets` retains the unfinished work inventory after cancellation. It is
|
||||
not evidence of active scheduling: `queued` is false and `startTime` is absent.
|
||||
Operators and tests must inspect the terminal flags, peer state and progress
|
||||
stability instead of requiring the historical inventory to be empty. A normal
|
||||
canceled entry remains blocked until clear; unresolved listing entries instead
|
||||
retain the explicit retry path that can re-observe or resolve those entries.
|
||||
|
||||
### Publication On Retiring Pools
|
||||
|
||||
Ordinary publication rechecks the selected pool against the durable pool metadata under its existing read fence. Selection may have happened before retirement, or on a node whose local pool state has not been refreshed. A staged new PUT must return `SlowDown` instead of publishing into a pool that has since become suspended. The staged input is not automatically replayed into another pool.
|
||||
|
||||
Running, queued, failed, canceled, and completed decommission states exclude the source from new ordinary publication, including new multipart uploads. Previously created multipart uploads retain their drain path while the source remains non-terminal; terminal source states reject further multipart publication. Failed and canceled entries become writable for new ordinary publication only after an allowed clear operation removes that state. This check does not change repair admission or the separate fence for operations that only release capacity.
|
||||
|
||||
For mixed batch deletes, only the pools selected to receive new delete markers are publication targets. Exact-version deletions on other pools remain protected by the same pool metadata read fence, without treating the retiring source or an unrelated reserved target as a destination for those markers.
|
||||
|
||||
### Shared Capacity On Healthy Targets
|
||||
|
||||
Ordinary publication into a healthy target is not rejected solely because that pool has an active decommission reservation. This follows the MinIO decommission write-routing contract: the retiring source stops accepting new writes, while the remaining pools share physical capacity between foreground requests and migration. A reservation remains a migration budget and recovery ledger, not an exclusive foreground-write quota. Repair retains its existing conservative reservation admission policy.
|
||||
|
||||
The existing durable metadata fence, valid active reservation checks, owner/mutation identity, pending-intent recovery, target write quorum and source-cleanup preflight remain required. Foreground writes do not acquire the mover's target I/O lock or settle its pending intent. Capacity estimates, including filesystem free-space deltas observed during migration, may include concurrent unrelated I/O; they are not proof of exclusive space or of a committed target object. Actual write failures and identity/quorum checks remain authoritative. Space loss can stop migration with the source retained, including after a target copy has committed. Capacity exhaustion can also fail foreground writes; this policy does not guarantee foreground priority or success. RustFS retains its existing capacity-blocked state and recovery behavior rather than changing terminal-state or retry semantics here.
|
||||
|
||||
The native regression overlaps public PUT and multipart create/part replacement/complete/abort operations with a paused target rename on another node context, checks that foreground publication leaves the pending migration ledger unchanged, and then checks both sufficient-capacity cleanup and injected capacity loss with byte-for-byte retained source and target data. Mixed batch deletion covers marker publication on both reserved and unreserved healthy targets together with exact-version removal on the retiring source. Capacity is injected deterministically; the object and metadata operations use real temporary disks, not a physical disk-exhaustion test.
|
||||
|
||||
### Status Response Shape
|
||||
|
||||
`GET /v3/pools/list` and `GET /v3/pools/status?pool=...` expose per-pool machine-readable decommission state. The `status` field can report `active`, `running`, `queued`, `complete`, `failed`, or `canceled`.
|
||||
@@ -70,6 +96,40 @@ When decommission metadata is present, `decommissionInfo` includes:
|
||||
|
||||
This makes queued pools and stalled metadata visible without requiring operators to inspect pool metadata files directly.
|
||||
|
||||
### Scanner Backlog Replica Conflicts
|
||||
|
||||
Native scanner CAS publication uses the storage-owned replica write path, not a
|
||||
direct write to a set selected from node-local pool state. On multi-pool stores,
|
||||
the fixed object namespace precedes the durable pool metadata read fence and
|
||||
the actual replica-set namespace. Admission excludes running, queued and
|
||||
completed sources; failed/canceled sources retain the scanner's existing
|
||||
membership-repair behavior. Missing pool metadata does not authorize a replica.
|
||||
Healthy reserved targets remain writable under the shared-capacity contract.
|
||||
|
||||
The replica writer retains both outer guards in an owned task and waits for the
|
||||
rename tail, including when its caller is canceled. Lock-loss signals remain
|
||||
attached to the set commit. This does not require every disk to succeed or alter
|
||||
write quorum/fsync policy. Replica writes for this one internal key serialize
|
||||
through its fixed namespace; ordinary PUT/GET do not enter this writer. The
|
||||
scanner still requires CAS success on every surviving set before acknowledging
|
||||
a ledger generation, and retains its partial-commit recovery protocol.
|
||||
Older scanner writers still use direct set CAS; this source-publication fence
|
||||
requires updating every scanner-capable node. No new on-disk or wire format is
|
||||
introduced.
|
||||
|
||||
The exact internal object `.rustfs.sys/buckets/.scanner-pause-backlog.json` is
|
||||
published with CAS to surviving sets. Its replica-local object modification
|
||||
times are not scanner ledger generations. A cross-pool migration receiving
|
||||
`PreconditionFailed` can therefore accept an existing unversioned replica with
|
||||
an identical known ETag, payload identity and metadata even when its write time
|
||||
differs. This exception does not apply to other keys, versioned objects, delete
|
||||
markers, missing identity evidence, or a different older ledger payload.
|
||||
|
||||
The source is still revalidated under its mutation fence before migration.
|
||||
Existing capacity-owner and mutation checks reconcile the pending intent before
|
||||
source cleanup; the replica exception does not clear an unknown intent, rewrite
|
||||
the native target, or change the scanner's committed-membership selection.
|
||||
|
||||
## MinIO Divergence Decisions
|
||||
|
||||
Behavior that is close to MinIO but not byte-for-byte identical. Changing either decision requires an operator compatibility note and updated characterization tests.
|
||||
|
||||
@@ -25,7 +25,7 @@
|
||||
10|crates/ecstore/src/cluster/rpc/remote_disk.rs
|
||||
6|crates/ecstore/src/config/com.rs
|
||||
14|crates/ecstore/src/config/storageclass.rs
|
||||
180|crates/ecstore/src/core/pools.rs
|
||||
178|crates/ecstore/src/core/pools.rs
|
||||
7|crates/ecstore/src/data_movement/mod.rs
|
||||
2|crates/ecstore/src/data_usage/local_snapshot.rs
|
||||
12|crates/ecstore/src/data_usage/mod.rs
|
||||
|
||||
Reference in New Issue
Block a user