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

fix(tier): drain cleanup before tier removal

Browse Source
This commit is contained in:
cxymds
2026-09-05 23:18:14 +08:00
parent cc5060ac20
commit 681a3228df
4 changed files with 612 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
+131 -1
View File
@@ -150,6 +150,18 @@ static XXHASH_SEED: u64 = 0;
static TIER_FREE_VERSION_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
static MANUAL_TRANSITION_JOB_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
#[cfg(test)]
#[derive(Default)]
struct FreeVersionPostRemoteDeleteTestBarrier {
arrived: Notify,
release: Notify,
}
#[cfg(test)]
tokio::task_local! {
static FREE_VERSION_POST_REMOTE_DELETE_TEST_BARRIER: Arc<FreeVersionPostRemoteDeleteTestBarrier>;
}
pub const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging";
#[allow(
dead_code,
@@ -910,6 +922,11 @@ async fn cleanup_free_version_exact(api: Arc<ECStore>, oi: &ObjectInfo, cancel:
})??;
}
}
#[cfg(test)]
if let Ok(barrier) = FREE_VERSION_POST_REMOTE_DELETE_TEST_BARRIER.try_with(Arc::clone) {
barrier.arrived.notify_one();
barrier.release.notified().await;
}
if !free_version_cleanup_fences_current(&topology_generation, &api, &bucket_guard, &object_guards, &lease, cancel, deadline) {
// Remote DELETE is idempotent, but a changed fence makes the local
// outcome ambiguous. Keep every marker for a fully fenced retry.
@@ -5831,7 +5848,7 @@ mod tests {
#[cfg(feature = "test-util")]
use crate::services::tier::test_util::register_mock_tier;
#[cfg(feature = "test-util")]
use crate::services::tier::tier::TierConfigMgr;
use crate::services::tier::tier::{TIER_DRIVER_TEST_FACTORY, TierConfigMgr, TierDriverTestFactory};
#[cfg(feature = "test-util")]
use crate::services::tier::warm_backend::{TransitionCandidateProbe, WarmBackend as _};
use crate::set_disk::{MultipartCommitBarrier, MultipartCommitPause};
@@ -7830,6 +7847,119 @@ mod tests {
}
}
#[cfg(feature = "test-util")]
#[tokio::test]
#[serial]
async fn tier_remove_waits_for_inflight_free_version_local_commit() {
let (disk_paths, ecstore) = setup_test_env().await;
let bucket = format!("tier-remove-free-version-{}", Uuid::new_v4());
let object = "free-version";
create_test_bucket(&ecstore, &bucket).await;
let (backend, identity_hex) = register_recovery_mock_tier(&ecstore).await;
let tier_manager = ecstore.tier_config_mgr();
{
let manager = tier_manager.read().await;
manager
.save_tiering_config(Arc::clone(&ecstore))
.await
.expect("mock tier configuration should persist before removal");
}
seed_recoverable_free_version(&disk_paths, &bucket, object, None, Some(identity_hex)).await;
let page = list_tier_free_versions(Arc::clone(&ecstore), 1, None, None, CancellationToken::new())
.await
.expect("seeded free version should be listed");
let oi = page
.items
.into_iter()
.next()
.expect("seeded free version should be recoverable");
backend
.set_put_remote_version(Some(oi.transitioned_object.version_id.clone()))
.await;
let seed_lease = TierConfigMgr::acquire_operation_lease(&tier_manager, "WARM")
.await
.expect("mock tier lease should be available");
seed_lease
.put(&oi.transitioned_object.name, ReaderImpl::Body(Bytes::from_static(b"body")), 4)
.await
.expect("remote free-version tuple should be seeded");
drop(seed_lease);
let barrier = Arc::new(super::FreeVersionPostRemoteDeleteTestBarrier::default());
let cleanup_barrier = Arc::clone(&barrier);
let cleanup_store = Arc::clone(&ecstore);
let cleanup_oi = oi.clone();
let cleanup = tokio::spawn(async move {
super::FREE_VERSION_POST_REMOTE_DELETE_TEST_BARRIER
.scope(cleanup_barrier, async move {
super::cleanup_free_version_exact(cleanup_store, &cleanup_oi, &CancellationToken::new()).await
})
.await
});
tokio::time::timeout(StdDuration::from_secs(30), barrier.arrived.notified())
.await
.expect("free-version cleanup should pause after the remote delete");
assert!(!backend.contains(&oi.transitioned_object.name).await);
let remove_manager = Arc::clone(&tier_manager);
let remove_store = Arc::clone(&ecstore);
let remove_backend = backend.clone();
let remove_driver_factory: TierDriverTestFactory = Arc::new(move |_| Ok(Box::new(remove_backend.clone())));
let mut remove = tokio::spawn(async move {
TIER_DRIVER_TEST_FACTORY
.scope(
remove_driver_factory,
TierConfigMgr::remove_and_save(&remove_manager, remove_store, "WARM", true),
)
.await
});
let prepared = tokio::time::timeout(StdDuration::from_secs(30), async {
loop {
match TierConfigMgr::acquire_operation_lease(&tier_manager, "WARM").await {
Ok(lease) => drop(lease),
Err(err) if TierConfigMgr::operation_lease_blocked_by_mutation(&err) => break,
Err(err) => panic!("tier remove should only block new operations while cleanup is paused: {err}"),
}
tokio::task::yield_now().await;
}
});
tokio::select! {
prepared = prepared => {
prepared.expect("tier remove should install its prepared admission fence");
}
result = &mut remove => {
panic!("tier remove finished before installing its prepared admission fence: {result:?}");
}
}
assert!(!remove.is_finished(), "tier remove must wait for the leased local cleanup commit");
barrier.release.notify_one();
tokio::time::timeout(StdDuration::from_secs(30), cleanup)
.await
.expect("free-version cleanup should finish after release")
.expect("free-version cleanup task should join")
.expect("free-version cleanup should keep its generation current");
tokio::time::timeout(StdDuration::from_secs(30), remove)
.await
.expect("tier remove should finish after local cleanup")
.expect("tier remove task should join")
.expect("tier remove should pass its fresh authoritative proof");
assert!(!tier_manager.read().await.is_tier_valid("WARM"));
for disk_path in &disk_paths {
assert!(
!fs::try_exists(disk_path.join(&bucket).join(object))
.await
.expect("post-removal free-version path check should succeed")
);
}
ecstore
.delete_bucket(&bucket, &DeleteBucketOptions::default())
.await
.expect("empty free-version test bucket should be removed");
}
#[cfg(feature = "test-util")]
#[tokio::test]
#[serial]
crates/ecstore/src/services/tier/tier.rs
+470 -57
View File
@@ -143,11 +143,12 @@ struct TierDriverBuildBarrier {
static TIER_DRIVER_BUILD_BARRIER: LazyLock<Mutex<Option<Arc<TierDriverBuildBarrier>>>> = LazyLock::new(|| Mutex::new(None));
#[cfg(test)]
type TierDriverTestFactory = Arc<dyn Fn(&TierConfig) -> std::result::Result<WarmBackendImpl, AdminError> + Send + Sync + 'static>;
pub(crate) type TierDriverTestFactory =
Arc<dyn Fn(&TierConfig) -> std::result::Result<WarmBackendImpl, AdminError> + Send + Sync + 'static>;
#[cfg(test)]
tokio::task_local! {
static TIER_DRIVER_TEST_FACTORY: TierDriverTestFactory;
pub(crate) static TIER_DRIVER_TEST_FACTORY: TierDriverTestFactory;
}
#[cfg(test)]
@@ -1377,27 +1378,42 @@ async fn ensure_no_authoritative_persisted_references<S>(
where
S: TierReferenceProofStore,
{
ensure_no_authoritative_persisted_references_with(api.clone(), TIER_DELETE_JOURNAL_PREFIX, |_object, data| {
let journal = decode_tier_delete_journal_entry(data).map_err(io::Error::other)?;
Ok((
journal.tier_name.clone(),
tier_persisted_reference_blocks_any_target(&journal.tier_name, journal.backend_identity, targets),
))
})
ensure_no_authoritative_persisted_references_with(
api.clone(),
TIER_DELETE_JOURNAL_PREFIX,
"tier-delete journal",
|_object, data| {
let journal = decode_tier_delete_journal_entry(data).map_err(io::Error::other)?;
Ok((
journal.tier_name.clone(),
tier_persisted_reference_blocks_any_target(&journal.tier_name, journal.backend_identity, targets),
))
},
)
.await?;
ensure_no_authoritative_persisted_references_with(api, TRANSITION_TRANSACTION_RECORD_PREFIX, |object, data| {
let transaction = decode_transition_transaction_record(object, data).map_err(io::Error::other)?;
Ok((
transaction.tier_name.clone(),
tier_persisted_reference_blocks_any_target(&transaction.tier_name, Some(transaction.backend_fingerprint), targets),
))
})
ensure_no_authoritative_persisted_references_with(
api,
TRANSITION_TRANSACTION_RECORD_PREFIX,
"transition transaction",
|object, data| {
let transaction = decode_transition_transaction_record(object, data).map_err(io::Error::other)?;
Ok((
transaction.tier_name.clone(),
tier_persisted_reference_blocks_any_target(
&transaction.tier_name,
Some(transaction.backend_fingerprint),
targets,
),
))
},
)
.await
}
async fn ensure_no_authoritative_persisted_references_with<S, F>(
api: Arc<S>,
prefix: &str,
reference_kind: &str,
blocks_target: F,
) -> std::result::Result<(), AdminError>
where
@@ -1426,7 +1442,7 @@ where
.map_err(tier_reference_proof_admin_error)?;
let (tier_name, blocks) = blocks_target(&object.name, &data).map_err(tier_reference_proof_admin_error)?;
if blocks {
return Err(tier_reference_proof_persisted_in_use_error(&tier_name, &object.name));
return Err(tier_reference_proof_persisted_in_use_error(&tier_name, reference_kind, &object.name));
}
}
if !page.is_truncated {
@@ -1488,16 +1504,21 @@ fn tier_persisted_reference_blocks_target(
fn tier_reference_proof_in_use_error(tier_name: &str, object: &ObjectInfo) -> AdminError {
let mut err = ERR_TIER_BACKEND_IN_USE.clone();
let reference_kind = if object.transitioned_object.free_version {
"free-version ownership"
} else {
"transitioned-object"
};
err.message = format!(
"Remote tier {tier_name} still has object references, for example {}/{}",
"Remote tier {tier_name} still has a {reference_kind} reference, for example {}/{}",
object.bucket, object.name
);
err
}
fn tier_reference_proof_persisted_in_use_error(tier_name: &str, object: &str) -> AdminError {
fn tier_reference_proof_persisted_in_use_error(tier_name: &str, reference_kind: &str, object: &str) -> AdminError {
let mut err = ERR_TIER_BACKEND_IN_USE.clone();
err.message = format!("Remote tier {tier_name} still has a persisted reference, for example {object}");
err.message = format!("Remote tier {tier_name} still has a {reference_kind} reference, for example {object}");
err
}
@@ -3685,13 +3706,17 @@ impl TierConfigMgr {
let manager = handle.read().await;
let runtime = tier_driver_runtime(handle, &manager);
let runtime = lock_unpoisoned(&runtime);
if !runtime
let prepared = runtime
.prepared_mutation_blocks
.values()
.any(|blocked_mutation_id| *blocked_mutation_id == mutation_id);
let committed = runtime
.committed_mutation_blocks
.values()
.any(|mutation_ids| mutation_ids.contains(&mutation_id))
{
.any(|mutation_ids| mutation_ids.contains(&mutation_id));
if !prepared && !committed {
let mut err = ERR_TIER_INVALID_CONFIG.clone();
err.message = "Remote tier committed mutation fence was not installed".to_string();
err.message = "Remote tier mutation fence was not installed".to_string();
return Err(err);
}
Ok(MutationBlockAllowance {
@@ -3898,14 +3923,6 @@ impl TierConfigMgr {
Self::begin_tier_transition_with_destinations(handle, manager, changed, replaced_destinations, mutation_block_allowance)
}
fn begin_tier_transition(
handle: &Arc<RwLock<Self>>,
manager: &mut Self,
changed: HashSet<String>,
) -> std::result::Result<TierPublishTransition, AdminError> {
Self::begin_tier_transition_with_destinations(handle, manager, changed, HashMap::new(), None)
}
fn begin_tier_transition_with_destinations(
handle: &Arc<RwLock<Self>>,
manager: &mut Self,
@@ -4176,7 +4193,7 @@ impl TierConfigMgr {
let mut config_lock = config_lock;
let coordinated_config_update = config_lock.is_some();
let mut update = Some(update);
let (mutation_kind, explicit_tier_name, mutation_force, current_for_targets, driver_tier, mut transition) =
let (mutation_kind, explicit_tier_name, mutation_force, current_for_targets, driver_tier, target_tiers) =
match mutation {
TierCandidateMutation::Prevalidated(prepared) => {
if version != prepared.version {
@@ -4192,9 +4209,8 @@ impl TierConfigMgr {
)));
}
candidate = prepared.candidate;
let validation_deadline = Instant::now() + TIER_REMOTE_VALIDATION_TIMEOUT;
let mut transition = {
let mut manager = handle.write().await;
let target_tiers = {
let manager = handle.read().await;
let mut target_tiers = changed_tier_names(&manager, &candidate);
if let Some(tier_name) = prepared.explicit_tier_name.as_ref()
&& (manager.tiers.contains_key(tier_name)
@@ -4203,8 +4219,7 @@ impl TierConfigMgr {
{
target_tiers.insert(tier_name.clone());
}
Self::begin_tier_transition(&handle, &mut manager, target_tiers)
.map_err(TierConfigUpdateError::Publish)?
target_tiers
};
(
prepared.kind,
@@ -4212,7 +4227,7 @@ impl TierConfigMgr {
prepared.force,
prepared.current,
prepared.driver_tier,
transition,
target_tiers,
)
}
mutation => {
@@ -4237,11 +4252,9 @@ impl TierConfigMgr {
last_refreshed_at: candidate.last_refreshed_at,
};
let validation_deadline = Instant::now() + TIER_REMOTE_VALIDATION_TIMEOUT;
let mut transition = {
let mut manager = handle.write().await;
let target_tiers = mutation.target_tiers(&manager, &candidate);
Self::begin_tier_transition(&handle, &mut manager, target_tiers)
.map_err(TierConfigUpdateError::Publish)?
let target_tiers = {
let manager = handle.read().await;
mutation.target_tiers(&manager, &candidate)
};
let driver_tier = apply_tier_candidate_mutation(mutation, &mut candidate, validation_deadline)
.await
@@ -4252,7 +4265,7 @@ impl TierConfigMgr {
mutation_force,
current_for_targets,
driver_tier,
transition,
target_tiers,
)
}
};
@@ -4275,28 +4288,83 @@ impl TierConfigMgr {
save_coordinator_tier_mutation_intent(api.clone(), coordinator_intent.as_ref())
.await
.map_err(TierConfigUpdateError::Save)?;
let mut blocked_target_tiers = target_tiers.clone();
if let Some(intent) = coordinator_intent.as_ref() {
TierConfigMgr::apply_prepared_mutation_intent_block(&handle, intent)
blocked_target_tiers.extend(intent.affected_targets.iter().map(|target| target.tier_name.clone()));
}
if let Some(intent) = coordinator_intent.as_ref() {
// `target_tiers` may include a stale local-only manager
// entry that is absent from the persisted proof
// snapshot. Fence that local transition under the same
// mutation ID as well; recovery may discard this
// process-local superset, which advances the revision
// and makes the deferred transition fail closed.
TierConfigMgr::apply_prepared_mutation_intent_block_for_tiers(&handle, intent, &blocked_target_tiers)
.await
.map_err(TierConfigUpdateError::Publish)?;
}
let prepared_mutation_block_allowance = match coordinator_intent.as_ref() {
Some(intent) => Some(
TierConfigMgr::mutation_block_allowance_for(&handle, intent.mutation_id)
.await
.map_err(TierConfigUpdateError::Publish)?,
),
None => None,
};
// A durable coordinator intent supplies the admission
// fence that lets us defer generation revocation. Keep the
// original early transition for no-intent paths (for
// example, reconciling a stale local manager to an
// idempotently removed persisted tier), where there is no
// Prepared record capable of blocking a new lease.
let (mut transition, deferred_target_tiers) = if coordinator_intent.is_some() {
(None, Some(target_tiers))
} else {
let transition = {
let mut manager = handle.write().await;
Self::begin_tier_transition_with_destinations(
&handle,
&mut manager,
target_tiers,
HashMap::new(),
None,
)
.map_err(TierConfigUpdateError::Publish)?
};
(Some(transition), None)
};
if coordinated_config_update {
drop(update.take());
drop(config_lock.take());
}
let drain_deadline = Instant::now() + TIER_REMOTE_VALIDATION_TIMEOUT;
if let Err(drain_error) = transition.wait_for_active_leases_until(drain_deadline).await {
if !abort_prepared_tier_mutation(&handle, api.clone(), coordinator_intent.as_ref(), Vec::new()).await {
// The durable Prepared block closes admission before the
// zero-reference proof, but deliberately leaves already
// issued generations current. In particular, an exact
// free-version cleanup that has completed remote DELETE
// must still be able to remove its local ownership marker;
// revoking its generation here would strand that marker
// and make this mutation reject its own interrupted work.
if let Some(intent) = coordinator_intent.as_ref()
&& let Err(drain_error) =
TierConfigMgr::wait_for_blocked_tier_operation_leases_for_tiers(&handle, &blocked_target_tiers).await
{
if !abort_prepared_tier_mutation(&handle, api.clone(), Some(intent), Vec::new()).await {
warn!(
event = "tier_mutation_abort",
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_TIER,
result = "prepared_intent_retained",
coordinator_intent = coordinator_intent.is_some(),
mutation_id = %intent.mutation_id,
"tier mutation lease drain failed and abort was incomplete"
);
}
return Err(TierConfigUpdateError::Publish(drain_error));
} else if let Some(transition) = transition.as_ref() {
let drain_deadline = Instant::now() + TIER_REMOTE_VALIDATION_TIMEOUT;
transition
.wait_for_active_leases_until(drain_deadline)
.await
.map_err(TierConfigUpdateError::Publish)?;
}
let prepared_peers = if let Some(intent) = coordinator_intent.as_ref() {
let peers = match remote_tier_mutation_peers().await {
@@ -4363,6 +4431,71 @@ impl TierConfigMgr {
}
return Err(TierConfigUpdateError::Publish(proof_error));
}
// No affected-tier lease can start after Prepared, and the
// existing set was drained above. It is now safe to revoke
// the generation for publication without invalidating a
// cleanup between its remote and local commit boundaries.
if transition.is_none() {
let target_tiers = deferred_target_tiers.ok_or_else(|| {
let mut err = ERR_TIER_INVALID_CONFIG.clone();
err.message = "Remote tier mutation lost its deferred transition targets".to_string();
TierConfigUpdateError::Publish(err)
})?;
let mut manager = handle.write().await;
transition = Some(
match Self::begin_tier_transition_with_destinations(
&handle,
&mut manager,
target_tiers,
HashMap::new(),
prepared_mutation_block_allowance.as_ref(),
) {
Ok(transition) => transition,
Err(transition_error) => {
drop(manager);
if !abort_prepared_tier_mutation(
&handle,
api.clone(),
coordinator_intent.as_ref(),
prepared_peers,
)
.await
{
warn!(
event = "tier_mutation_abort",
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_TIER,
result = "prepared_intent_retained",
coordinator_intent = coordinator_intent.is_some(),
"tier mutation publish transition failed and abort was incomplete"
);
}
return Err(TierConfigUpdateError::Publish(transition_error));
}
},
);
}
let mut transition = transition.ok_or_else(|| {
let mut err = ERR_TIER_INVALID_CONFIG.clone();
err.message = "Remote tier mutation lost its publish transition".to_string();
TierConfigUpdateError::Publish(err)
})?;
let drain_deadline = Instant::now() + TIER_REMOTE_VALIDATION_TIMEOUT;
if let Err(drain_error) = transition.wait_for_active_leases_until(drain_deadline).await {
drop(transition);
if !abort_prepared_tier_mutation(&handle, api.clone(), coordinator_intent.as_ref(), prepared_peers).await
{
warn!(
event = "tier_mutation_abort",
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_TIER,
result = "prepared_intent_retained",
coordinator_intent = coordinator_intent.is_some(),
"tier mutation publish drain failed and abort was incomplete"
);
}
return Err(TierConfigUpdateError::Publish(drain_error));
}
let candidate_digest = tier_config_candidate_digest(&candidate).map_err(TierConfigUpdateError::Save)?;
if coordinated_config_update {
config_lock = match Self::acquire_tier_config_write_lock(api.clone()).await {
@@ -5414,11 +5547,40 @@ impl TierConfigMgr {
handle: &Arc<RwLock<Self>>,
intent: &TierMutationIntent,
) -> std::result::Result<(), AdminError> {
let target_tiers = intent
.affected_targets
.iter()
.map(|target| target.tier_name.clone())
.collect();
Self::apply_prepared_mutation_intent_block_for_tiers(handle, intent, &target_tiers).await
}
async fn apply_prepared_mutation_intent_block_for_tiers(
handle: &Arc<RwLock<Self>>,
intent: &TierMutationIntent,
target_tiers: &HashSet<String>,
) -> std::result::Result<(), AdminError> {
if intent.state != TierMutationIntentState::Prepared {
return Ok(());
}
let manager = handle.read().await;
let runtime = tier_driver_runtime(handle, &manager);
let mut runtime = lock_unpoisoned(&runtime);
let mut prepared_mutation_blocks = runtime.prepared_mutation_blocks.clone();
Self::collect_prepared_mutation_intent_block(&mut prepared_mutation_blocks, intent)?;
for tier_name in target_tiers {
match prepared_mutation_blocks.entry(tier_name.clone()) {
Entry::Vacant(entry) => {
entry.insert(intent.mutation_id);
}
Entry::Occupied(entry) if *entry.get() == intent.mutation_id => {}
Entry::Occupied(_) => {
let mut err = ERR_TIER_BACKEND_IN_USE.clone();
err.message = format!("Remote tier {tier_name} already has another prepared mutation");
return Err(err);
}
}
}
if prepared_mutation_blocks == runtime.prepared_mutation_blocks {
return Ok(());
}
@@ -5434,6 +5596,18 @@ impl TierConfigMgr {
pub(crate) async fn wait_for_blocked_tier_operation_leases(
handle: &Arc<RwLock<Self>>,
intent: &TierMutationIntent,
) -> std::result::Result<(), AdminError> {
let target_tiers = intent
.affected_targets
.iter()
.map(|target| target.tier_name.clone())
.collect();
Self::wait_for_blocked_tier_operation_leases_for_tiers(handle, &target_tiers).await
}
async fn wait_for_blocked_tier_operation_leases_for_tiers(
handle: &Arc<RwLock<Self>>,
target_tiers: &HashSet<String>,
) -> std::result::Result<(), AdminError> {
let generations = {
let manager = handle.read().await;
@@ -5441,10 +5615,9 @@ impl TierConfigMgr {
return Ok(());
};
let runtime = lock_unpoisoned(&runtime);
intent
.affected_targets
target_tiers
.iter()
.filter_map(|target| runtime.generations.get(&target.tier_name).cloned())
.filter_map(|tier_name| runtime.generations.get(tier_name).cloned())
.collect::<Vec<_>>()
};
let drain = async {
@@ -14414,6 +14587,13 @@ mod tests {
.push(object);
}
fn remove_listed_version(&self, bucket: &str, object: &str) {
self.listed_versions
.lock()
.expect("tier reference fixture should not poison")
.retain(|version| version.bucket != bucket || version.name != object);
}
fn add_lifecycle_config(&self, bucket: &str, config: BucketLifecycleConfiguration) {
self.lifecycle_configs
.lock()
@@ -15682,6 +15862,152 @@ mod tests {
);
}
#[tokio::test]
async fn tier_remove_prepared_fence_allows_inflight_free_version_cleanup_to_finish() {
let store = Arc::new(CasConfigStore::default());
let tier = build_rustfs_tier("COLD-A");
let identity = tier_backend_identity(&tier).expect("test tier identity should encode");
let mut persisted = empty_mgr();
persisted.tiers.insert("COLD-A".to_string(), tier.clone_with_credentials());
persisted
.save_tiering_config_if_current(store.clone(), None)
.await
.expect("free-version drain fixture should persist");
let manager = TierConfigMgr::new();
{
let mut guard = manager.write().await;
guard.tiers.insert("COLD-A".to_string(), tier);
guard.tiers.insert("COLD-B".to_string(), build_rustfs_tier("COLD-B"));
guard
.replace_driver("COLD-A", Box::new(LeaseTestBackend::ready("cleanup")))
.expect("cleanup driver generation should install");
guard
.replace_driver("COLD-B", Box::new(LeaseTestBackend::ready("stale-local")))
.expect("stale local driver generation should install");
}
let cleanup_lease = TierConfigMgr::acquire_operation_lease(&manager, "COLD-A")
.await
.expect("in-flight cleanup lease should be available");
let stale_local_lease = TierConfigMgr::acquire_operation_lease(&manager, "COLD-B")
.await
.expect("stale local tier lease should be available");
let mut free_version = transitioned_tier_object("photos", "2026/free-version.jpg", "COLD-A", Some(identity));
free_version.transitioned_object.status = "pending".to_string();
free_version.transitioned_object.free_version = true;
store.add_listed_version(free_version);
let remove_manager = manager.clone();
let remove_store = store.clone();
let remove = tokio::spawn(async move {
TIER_MUTATION_TEST_PEERS
.scope(
Vec::new(),
TierConfigMgr::remove_and_save_with(&remove_manager, remove_store, "COLD-A", true),
)
.await
});
tokio::time::timeout(Duration::from_secs(1), async {
loop {
let guard = manager.read().await;
let runtime = registered_tier_driver_runtime(&guard).expect("runtime should remain registered");
let prepared = {
let runtime = lock_unpoisoned(&runtime);
runtime.prepared_mutation_blocks.contains_key("COLD-A")
&& runtime.prepared_mutation_blocks.contains_key("COLD-B")
};
if prepared {
break;
}
drop(guard);
tokio::task::yield_now().await;
}
})
.await
.expect("tier remove should install its durable prepared fence");
assert!(
cleanup_lease.is_current(&manager).await,
"the prepared fence must let the already leased cleanup finish its exact local marker deletion"
);
assert!(
stale_local_lease.is_current(&manager).await,
"the local superset fence must also let an already leased stale-manager operation finish"
);
let blocked = match TierConfigMgr::acquire_operation_lease(&manager, "COLD-A").await {
Ok(_) => panic!("the prepared fence must reject new tier operations"),
Err(err) => err,
};
assert!(TierConfigMgr::operation_lease_blocked_by_mutation(&blocked));
let stale_blocked = match TierConfigMgr::acquire_operation_lease(&manager, "COLD-B").await {
Ok(_) => panic!("the local superset fence must reject new stale-manager operations"),
Err(err) => err,
};
assert!(TierConfigMgr::operation_lease_blocked_by_mutation(&stale_blocked));
store.remove_listed_version("photos", "2026/free-version.jpg");
drop(cleanup_lease);
drop(stale_local_lease);
tokio::time::timeout(Duration::from_secs(5), remove)
.await
.expect("tier remove should finish after both in-flight operations release their leases")
.expect("tier remove task should join")
.expect("tier remove should pass once the in-flight cleanup removes its marker");
assert!(!manager.read().await.tiers.contains_key("COLD-A"));
assert!(!manager.read().await.tiers.contains_key("COLD-B"));
assert!(
!load_tier_config_for_update(store)
.await
.expect("removed tier config should reload")
.0
.tiers
.contains_key("COLD-A")
);
}
#[tokio::test]
async fn no_intent_stale_manager_removal_keeps_early_generation_drain() {
let store = Arc::new(CasConfigStore::default());
empty_mgr()
.save_tiering_config_if_current(store.clone(), None)
.await
.expect("empty persisted tier config should exist");
let manager = TierConfigMgr::new();
{
let mut guard = manager.write().await;
install_lease_backend(&mut guard, "COLD-A", LeaseTestBackend::ready("stale"));
}
let old = TierConfigMgr::acquire_operation_lease(&manager, "COLD-A")
.await
.expect("stale manager lease should be available");
let remove_manager = manager.clone();
let remove_store = store.clone();
let remove =
tokio::spawn(async move { TierConfigMgr::remove_and_save_with(&remove_manager, remove_store, "COLD-A", true).await });
tokio::time::timeout(Duration::from_secs(1), async {
while old.is_current(&manager).await {
tokio::task::yield_now().await;
}
})
.await
.expect("no-intent stale-manager reconciliation should revoke before its proof");
let blocked = match TierConfigMgr::acquire_operation_lease(&manager, "COLD-A").await {
Ok(_) => panic!("stale-manager reconciliation must not admit a new operation"),
Err(err) => err,
};
assert!(TierConfigMgr::operation_lease_blocked_by_mutation(&blocked));
drop(old);
remove
.await
.expect("stale-manager removal task should join")
.expect("stale-manager removal should converge to the persisted empty config");
assert!(!manager.read().await.tiers.contains_key("COLD-A"));
}
async fn assert_lifecycle_only_reference_obeys_force(clear: bool, force: bool) {
let store = Arc::new(CasConfigStore::default());
let tier = build_rustfs_tier("COLD-A");
@@ -16585,12 +16911,23 @@ mod tests {
.await
});
tokio::time::timeout(Duration::from_secs(1), async {
while old.is_current(&manager).await {
loop {
let guard = manager.read().await;
let runtime = registered_tier_driver_runtime(&guard).expect("runtime should remain registered");
let prepared = lock_unpoisoned(&runtime).prepared_mutation_blocks.contains_key("COLD-A");
if prepared {
break;
}
drop(guard);
tokio::task::yield_now().await;
}
})
.await
.expect("owned update should revoke before caller cancellation");
.expect("owned update should install its prepared fence before caller cancellation");
assert!(
old.is_current(&manager).await,
"an already leased operation must remain current until it can finish"
);
caller.abort();
drop(old);
@@ -16635,12 +16972,23 @@ mod tests {
.await
});
tokio::time::timeout(Duration::from_secs(1), async {
while old.is_current(&manager).await {
loop {
let guard = manager.read().await;
let runtime = registered_tier_driver_runtime(&guard).expect("runtime should remain registered");
let prepared = lock_unpoisoned(&runtime).prepared_mutation_blocks.contains_key("COLD-A");
if prepared {
break;
}
drop(guard);
tokio::task::yield_now().await;
}
})
.await
.expect("owned update should revoke before caller cancellation");
.expect("owned update should install its prepared fence before caller cancellation");
assert!(
old.is_current(&manager).await,
"the prepared fence must not invalidate an already leased operation"
);
caller.abort();
let config_file = tier_config_lock_path();
@@ -17026,6 +17374,71 @@ mod tests {
assert!(current.tiers.contains_key("COLD-B"));
}
#[tokio::test]
#[serial_test::serial]
async fn reference_proof_rejects_a_changed_prepared_fence_revision_before_publish() {
let manager = TierConfigMgr::new();
let store = Arc::new(CasConfigStore::default());
let mut persisted = empty_mgr();
persisted.tiers.insert("COLD-A".to_string(), build_rustfs_tier("COLD-A"));
persisted
.save_tiering_config_if_current(store.clone(), None)
.await
.expect("prepared-fence revision fixture should persist");
{
let mut guard = manager.write().await;
install_lease_backend(&mut guard, "COLD-A", LeaseTestBackend::ready("old"));
}
let barrier = tier_reference_proof_test_barrier();
let scoped_barrier = barrier.clone();
let update_manager = manager.clone();
let update_store = store.clone();
let update = tokio::spawn(async move {
TIER_REFERENCE_PROOF_TEST_BARRIER
.scope(
scoped_barrier,
TIER_MUTATION_TEST_PEERS.scope(
Vec::new(),
TierConfigMgr::update_candidate_with_config_lock(
&update_manager,
update_store,
TierCandidateMutation::Remove("COLD-A".to_string(), true),
),
),
)
.await
});
barrier.arrived.notified().await;
let unrelated = prepared_remove_intent("COLD-B", uuid::Uuid::from_u128(0x2237));
TierConfigMgr::apply_prepared_mutation_intent_block(&manager, &unrelated)
.await
.expect("an unrelated prepared fence should advance the runtime revision");
barrier.release.add_permits(1);
let err = update
.await
.expect("tier update task should join")
.expect_err("a reference proof cannot authorize publication across a fence revision change");
let TierConfigUpdateError::Publish(err) = err else {
panic!("the stale prepared-fence allowance should fail publication: {err:?}");
};
assert!(err.message.contains("changed before replacement"), "{err}");
assert!(manager.read().await.tiers.contains_key("COLD-A"));
assert!(
load_tier_config_for_update(store)
.await
.expect("rejected tier config should remain readable")
.0
.tiers
.contains_key("COLD-A")
);
TierConfigMgr::clear_prepared_mutation_intent_block(&manager, unrelated.mutation_id)
.await
.expect("unrelated test fence should clear");
}
#[tokio::test]
#[serial_test::serial]
async fn caller_cancellation_after_durable_prepare_does_not_hide_the_mutation() {