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

fix(ecstore): leave no source free version for a moved tier version (#8732)

Browse Source 
* fix(ecstore): leave no source free version for a moved tier version

Data-movement source cleanup deletes each source version separately and
never set skip_tier_free_version, so every transitioned version that
decommission or rebalance had already copied to a target pool left a new
tier free version on the source. Decommission then failed its post-check
for any pool with tiered objects; rebalance left the record behind. The
record claims the remote object the target copy still reads and stays
blocked behind that live reference.

Decommission and rebalance now pass the identities of transitioned
versions whose copy committed on a target, and the cleanup skips the
free version for exactly those. A transitioned version that did not
move, such as one lifecycle expired during the move, still gets its free
version because that record is its only remote-delete owner.

Refs rustfs/backlog#2873, rustfs/backlog#2872

* fix(ecstore): keep the free version of a superseded moved source

Data movement treats a source null version as complete when a newer
unversioned object already sits on the target, without copying it.
That outcome returned the same Ok as a committed copy, so a restored
transitioned null version superseded this way was recorded as
transferred and lost its source free version, leaving its remote object
with no owner.

The copy paths now return DataMovementCopy::Committed or ::Superseded,
and decommission and rebalance record only committed copies of
transitioned versions. A superseded source keeps its free version.

Tests cover the superseded restored null version for rebalance and
decommission, a restored and a null transitioned version moving by
rebalance, an entry with one lifecycle-expired and one moved version,
the per-version skip with two versions of one key, and a directory
object. The tiered stale-target routine test from #8724 now expects the
decommission to complete. The ILM ownership contract and the
decommission notes describe the moved, superseded, expired and legacy
cases.

Refs rustfs/backlog#2873, rustfs/backlog#2872

* test(ecstore): build the decommission ILM fixtures without s3s

The superseded and expiry-mix decommission tests named s3s request and
lifecycle types directly, which added pools_test.rs to the ecstore
files that reference s3s and failed the footprint ratchet. Set the
restore request through ObjectOptions, and store the lifecycle rule as
bucket metadata XML and load it with get_expiry_configs, as the
decommission routine does.

* test(ecstore): pin restored decommission copies and multipart supersession

Decommission recorded a committed data copy of a restored transitioned
version without any test failing if the record were dropped; add a
routine-level decommission of a restored version that checks the post-
check passes, the source keeps nothing, and the target keeps the full
remote tuple. Add the multipart variant of the superseded restored null
version (rebalance), and a unit test that separates committed,
superseded and mismatched targets in the overwrite resume adjudication.
This commit is contained in:
Chris
2026-10-20 22:20:12 +09:00
committed by GitHub
parent 928a369445
commit 7cc4a9dbd0
12 changed files with 1407 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
+30
View File
@@ -1038,6 +1038,36 @@ pub struct ExpiryState {
recovery_notify: Arc<Notify>,
}
/// Keeps every lifecycle expiry task the process enqueues in its queue until
/// released, as if the expiry workers were busy.
#[cfg(all(test, feature = "test-util"))]
pub(crate) struct HeldExpiryQueue {
state: Arc<RwLock<ExpiryState>>,
previous: ExpiryState,
}
#[cfg(all(test, feature = "test-util"))]
impl HeldExpiryQueue {
pub(crate) async fn hold() -> Self {
let held = Arc::try_unwrap(ExpiryState::new_with_unconsumed_worker_channel(16))
.ok()
.expect("a new expiry state has a single owner")
.into_inner();
let state = runtime_sources::expiry_state_handle();
let previous = std::mem::replace(&mut *state.write().await, held);
Self { state, previous }
}
pub(crate) async fn pending_tasks(&self) -> usize {
self.state.read().await.pending_tasks()
}
/// Drops the held tasks unrun and restores the previous expiry state.
pub(crate) async fn release(self) {
*self.state.write().await = self.previous;
}
}
impl ExpiryState {
#[allow(clippy::new_ret_no_self)]
pub fn new() -> Arc<RwLock<Self>> {
crates/ecstore/src/core/pools.rs
+28 -2
View File
@@ -16155,6 +16155,7 @@ impl ECStore {
SCANNER_PAUSE_BACKLOG_PATH,
expected,
&[],
&[],
data_movement::SourceCleanupBucketFence {
expected_incarnation_id: None,
lifecycle_guard: None,
@@ -16491,6 +16492,7 @@ impl ECStore {
let mut free_version_disposition = DecommissionFreeVersionDisposition::default();
let mut cleanup_preflight_allowed_missing = Vec::new();
let mut cleanup_preflight_vanished = Vec::new();
let mut cleanup_transferred_tier_versions = Vec::new();
let mut entry_blocked = false;
for version in fivs.versions.iter() {
@@ -16693,6 +16695,7 @@ impl ECStore {
let mut cleanup_ignored = false;
let mut failure = false;
let mut error = None;
let mut copy = None;
if version.deleted {
let mut version_attempt = 1;
let mut target_busy_attempt: usize = 0;
@@ -16946,6 +16949,8 @@ impl ECStore {
Ok(_) => {
failure = false;
error = None;
// The tiered copy commits only the exact source tuple.
copy = Some(data_movement::DataMovementCopy::Committed);
}
Err(err) if is_decommission_copy_cleanup_safe_error(&err) => {
ignore = true;
@@ -17128,6 +17133,7 @@ impl ECStore {
migrate_result,
);
copy = migrate_result.as_ref().ok().copied();
if let Err(err) = migrate_result {
if is_decommission_copy_cleanup_safe_error(&err) {
ignore = true;
@@ -17243,6 +17249,11 @@ impl ECStore {
if should_count_decommission_version_complete(ignore, cleanup_ignored, failure) {
decommissioned += 1;
if copy == Some(data_movement::DataMovementCopy::Committed)
&& version.transition_status == rustfs_filemeta::TRANSITION_COMPLETE
{
cleanup_transferred_tier_versions.push(data_movement::source_cleanup_version_identity(version));
}
}
}
@@ -17305,6 +17316,7 @@ impl ECStore {
entry.name.as_str(),
&cleanup_expected,
&cleanup_preflight_allowed_missing,
&cleanup_transferred_tier_versions,
data_movement::SourceCleanupBucketFence {
expected_incarnation_id: expected_bucket_incarnation_id,
lifecycle_guard: bucket_incarnation_fence
@@ -17559,6 +17571,19 @@ impl ECStore {
entry: MetaCacheEntry,
bucket: String,
set: Arc<SetDisks>,
) -> Result<()> {
self.decommission_entry_for_test_with_lifecycle(idx, entry, bucket, set, None)
.await
}
#[cfg(all(test, feature = "test-util"))]
pub(crate) async fn decommission_entry_for_test_with_lifecycle(
self: &Arc<Self>,
idx: usize,
entry: MetaCacheEntry,
bucket: String,
set: Arc<SetDisks>,
lifecycle_config: Option<BucketLifecycleConfiguration>,
) -> Result<()> {
let expected_bucket_incarnation_id = if is_meta_bucketname(&bucket) {
None
@@ -17574,7 +17599,7 @@ impl ECStore {
entry,
bucket,
set,
None,
lifecycle_config,
None,
None,
expected_bucket_incarnation_id,
@@ -19830,6 +19855,7 @@ impl ECStore {
path,
&source_versions,
&[],
&[],
data_movement::SourceCleanupBucketFence::default(),
"decommission durable ILM final sweep",
)
@@ -20299,7 +20325,7 @@ impl ECStore {
rd: GetObjectReader,
expected_bucket_incarnation_id: Option<uuid::Uuid>,
capacity_owner: Option<DecommissionCapacityOwner>,
) -> Result<()> {
) -> Result<data_movement::DataMovementCopy> {
warn!("decommission_object: start {} {}", &bucket, &rd.object_info.name);
let object_name = rd.object_info.name.clone();
let mut migration = tokio::task::JoinSet::new();
crates/ecstore/src/core/pools_test.rs
+513 -31
View File
@@ -1108,37 +1108,7 @@ mod decommission_lock_order_tests {
put_source_object(&store, &bucket, anchor, b"anchor".to_vec()).await;
let subject = source_key_after(&store, anchor, "tiered");
let _tier = seed_source_transitioned_object(&store, &bucket, &subject, false).await;
start_over_budget_routine_decommission(&store, &bucket, 45 << 20).await;
let barrier = stale_target_barrier(&store);
let _fault = pause_object_target_gate_acquire(Arc::clone(&barrier), bucket.clone(), anchor.to_string(), 1);
let routine = spawn_decommission_routine(&store, 1);
barrier.wait_until_target_gate_acquire_paused().await;
assert_held_at_subject(&store, &bucket, anchor, &subject).await;
commit_other_mutation_on_target_1(&store, 6 << 20).await;
assert_target_1_over_budget(&store).await;
barrier.release_target_gate_acquire();
// The source cleanup of a migrated transitioned version leaves a
// tier free version behind, so the routine fails its post-check
// even without contention (rustfs/backlog#2873, defect D). Once
// that is fixed this flips: expect the routine to complete.
let err = routine
.await
.expect("the decommission routine task should join")
.expect_err("the post-check fails while the tier free version is left behind");
assert!(
err.to_string().contains("post-check failed") && err.to_string().contains("were found in bucket"),
"{err}"
);
let local = store.pool_meta.read().await;
let info = local.pools[0]
.decommission
.as_ref()
.expect("the decommission state should remain present");
assert!(info.failed, "the post-check failure marks the decommission failed");
assert_eq!(info.items_decommission_failed, 0);
assert!(info.capacity_blocked_reason.is_none(), "{:?}", info.capacity_blocked_reason);
drop(local);
assert_source_key_migrated_to_target_2(&store, &bucket, &subject).await;
run_stale_subject_case(&store, &bucket, anchor, &subject).await;
});
}
@@ -1480,6 +1450,518 @@ mod decommission_lock_order_tests {
});
}
/// A transitioned version moves to a target pool as tier metadata, so the
/// target copy becomes the live owner of the remote object. Removing the
/// version from the source must not append a tier free version there: it
/// would fail the post-check and claim a remote object the target still
/// reads (rustfs/backlog#2873, defect D).
#[test]
#[serial_test::serial]
fn decommission_of_a_transitioned_object_completes_without_a_source_free_version() {
run_large_stack_async_test("decommission-transitioned-source-cleanup", || async {
let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await;
let bucket = test_bucket("transitioned-cleanup");
let object = "transitioned.bin";
let payload = b"transitioned source bytes".to_vec();
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create the transitioned bucket");
let tier_name = format!("DECOMTIER{}", &uuid::Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
let backend = register_mock_tier(&store.pools[0].instance_ctx().tier_config_mgr(), &tier_name).await;
let source = store.pools[0]
.put_object(
&bucket,
object,
&mut PutObjReader::from_vec(payload.clone()),
&ObjectOptions {
versioned: true,
..Default::default()
},
)
.await
.expect("write the transitioned source");
store.pools[0]
.transition_object(
&bucket,
object,
&ObjectOptions {
versioned: true,
version_id: source.version_id.map(|version| version.to_string()),
transition: crate::bucket::lifecycle::lifecycle::TransitionOptions {
status: TRANSITION_PENDING.to_string(),
tier: tier_name.clone(),
etag: source.etag.clone().expect("the transitioned source should have an ETag"),
..Default::default()
},
mod_time: source.mod_time,
..Default::default()
},
)
.await
.expect("transition the source before the decommission");
let source_versions = store.pools[0]
.get_disks_by_key(object)
.load_file_info_versions_exact(&bucket, object)
.await
.expect("the source versions should decode")
.expect("the transitioned source should exist");
assert_eq!(source_versions.versions.len(), 1);
let source_version = source_versions.versions[0].clone();
assert_eq!(source_version.transition_status, rustfs_filemeta::TRANSITION_COMPLETE);
let remote_object = source_version.transitioned_objname.clone();
assert!(backend.contains(&remote_object).await, "the transition must upload the remote object");
let layout = DecommissionErasureLayout { data: 1, parity: 0 };
start_routine_decommission_with_capacity(
&store,
&bucket,
vec![
DecommissionPoolCapacityInfo::for_test(0, layout, 0, 64 << 20, 64 << 20),
DecommissionPoolCapacityInfo::for_test(1, layout, 100 << 20, 100 << 20, 0),
DecommissionPoolCapacityInfo::for_test(2, layout, 1 << 30, 1 << 30, 0),
],
0,
)
.await;
spawn_decommission_routine(&store, 1)
.await
.expect("the decommission routine task should join")
.expect("the decommission of a transitioned object should pass its post-check");
let local = store.pool_meta.read().await;
let info = local.pools[0]
.decommission
.as_ref()
.expect("the decommission state should remain present");
assert!(
info.complete && !info.failed && !info.canceled,
"complete={}, failed={}, canceled={}",
info.complete,
info.failed,
info.canceled
);
assert_eq!(info.items_decommission_failed, 0);
drop(local);
let left = store.pools[0]
.get_disks_by_key(object)
.load_file_info_versions_exact(&bucket, object)
.await
.expect("the source versions should decode");
assert!(
left.as_ref().is_none_or(|versions| versions.versions.is_empty()),
"the source must keep neither the version nor a tier free version: {:?}",
left.map(|versions| versions.versions)
);
let mut owners = Vec::new();
for target_pool_index in [1, 2] {
if let Some(versions) = store.pools[target_pool_index]
.get_disks_by_key(object)
.load_file_info_versions_exact(&bucket, object)
.await
.expect("the target versions should decode")
.filter(|versions| !versions.versions.is_empty())
{
owners.push((target_pool_index, versions));
}
}
assert_eq!(owners.len(), 1, "exactly one target pool must own the migrated version");
let (target_pool_index, target_versions) = owners.pop().expect("one owner");
assert_eq!(
target_versions.versions.len(),
1,
"the target must hold the version and no tier free version: {:?}",
target_versions.versions
);
let moved = &target_versions.versions[0];
assert!(!moved.tier_free_version());
assert_eq!(moved.version_id, source_version.version_id);
assert_eq!(moved.transition_status, rustfs_filemeta::TRANSITION_COMPLETE);
assert_eq!(moved.transition_tier, tier_name);
assert_eq!(moved.transitioned_objname, remote_object);
assert_eq!(moved.transition_version_id, source_version.transition_version_id);
assert!(backend.contains(&remote_object).await, "the target still reads the remote object");
assert_eq!(backend.remove_count().await, 0, "no remote delete may be issued");
assert_eq!(backend.exact_remove_count(), 0, "no remote delete may be issued");
let mut reader = store.pools[target_pool_index]
.get_object_reader(
&bucket,
object,
None,
HeaderMap::new(),
&ObjectOptions {
versioned: true,
version_id: source_version.version_id.map(|version| version.to_string()),
..Default::default()
},
)
.await
.expect("the migrated version should be readable from its tier");
let mut body = Vec::new();
reader
.stream
.read_to_end(&mut body)
.await
.expect("drain the migrated version");
assert_eq!(body, payload);
});
}
async fn exact_versions(
store: &Arc<crate::store::ECStore>,
pool_index: usize,
bucket: &str,
object: &str,
) -> Vec<rustfs_filemeta::FileInfo> {
store.pools[pool_index]
.get_disks_by_key(object)
.load_file_info_versions_exact(bucket, object)
.await
.expect("the pool versions should decode")
.map(|versions| versions.versions)
.unwrap_or_default()
}
/// Write a version to pool 0 and transition it to `tier_name`.
async fn put_transitioned_source_version(
store: &Arc<crate::store::ECStore>,
bucket: &str,
object: &str,
tier_name: &str,
versioned: bool,
mod_time: Option<time::OffsetDateTime>,
) -> crate::object_api::ObjectInfo {
let written = store.pools[0]
.put_object(
bucket,
object,
&mut PutObjReader::from_vec(format!("source bytes of {object}").into_bytes()),
&ObjectOptions {
versioned,
mod_time,
..Default::default()
},
)
.await
.expect("write the transitioned source");
store.pools[0]
.transition_object(
bucket,
object,
&ObjectOptions {
versioned,
version_id: written.version_id.map(|version| version.to_string()),
transition: crate::bucket::lifecycle::lifecycle::TransitionOptions {
status: TRANSITION_PENDING.to_string(),
tier: tier_name.to_string(),
etag: written.etag.clone().expect("the transitioned source should have an ETag"),
..Default::default()
},
mod_time: written.mod_time,
..Default::default()
},
)
.await
.expect("transition the source before the decommission");
written
}
/// Restore a transitioned source version, so that it keeps a local copy
/// next to its remote object and moves as data rather than tier metadata.
async fn restore_source_version(
store: &Arc<crate::store::ECStore>,
bucket: &str,
object: &str,
versioned: bool,
written: &crate::object_api::ObjectInfo,
) {
let mut restore_opts = ObjectOptions {
versioned,
version_id: written.version_id.map(|version| version.to_string()),
..Default::default()
};
restore_opts.transition.restore_request.days = Some(1);
Arc::clone(store)
.restore_transitioned_object(bucket, object, &restore_opts)
.await
.expect("restore the transitioned source");
}
async fn start_transitioned_entry_decommission(store: &Arc<crate::store::ECStore>, bucket: &str) {
let layout = DecommissionErasureLayout { data: 1, parity: 0 };
start_routine_decommission_with_capacity(
store,
bucket,
vec![
DecommissionPoolCapacityInfo::for_test(0, layout, 0, 64 << 20, 64 << 20),
DecommissionPoolCapacityInfo::for_test(1, layout, 100 << 20, 100 << 20, 0),
DecommissionPoolCapacityInfo::for_test(2, layout, 1 << 30, 1 << 30, 0),
],
0,
)
.await;
}
fn assert_same_remote_tuple(actual: &rustfs_filemeta::FileInfo, expected: &rustfs_filemeta::FileInfo) {
assert_eq!(actual.transition_tier, expected.transition_tier);
assert_eq!(actual.transitioned_objname, expected.transitioned_objname);
assert_eq!(actual.transition_version_id, expected.transition_version_id);
assert_eq!(
crate::services::tier::tier::tier_destination_id_from_metadata(&actual.metadata)
.expect("the copy's destination identity should decode"),
crate::services::tier::tier::tier_destination_id_from_metadata(&expected.metadata)
.expect("the source destination identity should decode"),
);
}
/// A restored source moves as data; the copy keeps its transition tuple,
/// so the target owns the remote object and the source cleanup must not
/// leave a tier free version (rustfs/backlog#2873, defect D).
#[test]
#[serial_test::serial]
fn decommission_of_a_restored_transitioned_version_moves_its_remote_owner() {
run_large_stack_async_test("decommission-restored-transitioned-version", || async {
let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await;
let bucket = test_bucket("restored-transitioned");
let object = "restored.bin";
store
.make_bucket(
&bucket,
&MakeBucketOptions {
versioning_enabled: true,
..Default::default()
},
)
.await
.expect("create the versioned bucket");
let tier_name = format!("DECOMTIER{}", &uuid::Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
let backend = register_mock_tier(&store.pools[0].instance_ctx().tier_config_mgr(), &tier_name).await;
let written = put_transitioned_source_version(&store, &bucket, object, &tier_name, true, None).await;
restore_source_version(&store, &bucket, object, true, &written).await;
let source_versions = exact_versions(&store, 0, &bucket, object).await;
assert_eq!(source_versions.len(), 1);
let source = source_versions[0].clone();
assert_eq!(source.transition_status, rustfs_filemeta::TRANSITION_COMPLETE);
assert!(!source.is_remote(), "a restored source keeps its local copy");
start_transitioned_entry_decommission(&store, &bucket).await;
spawn_decommission_routine(&store, 1)
.await
.expect("the decommission routine task should join")
.expect("the decommission of a restored transitioned version should pass its post-check");
let local = store.pool_meta.read().await;
let info = local.pools[0]
.decommission
.as_ref()
.expect("the decommission state should remain present");
assert!(
info.complete && !info.failed && !info.canceled,
"complete={}, failed={}, canceled={}",
info.complete,
info.failed,
info.canceled
);
assert_eq!(info.items_decommission_failed, 0);
drop(local);
let left = exact_versions(&store, 0, &bucket, object).await;
assert!(left.is_empty(), "the source must keep no version or tier free version: {left:?}");
let mut owners = Vec::new();
for target_pool_index in [1, 2] {
owners.extend(exact_versions(&store, target_pool_index, &bucket, object).await);
}
assert_eq!(owners.len(), 1, "one target must hold the version and no free version: {owners:?}");
let moved = &owners[0];
assert!(!moved.tier_free_version());
assert_eq!(moved.version_id, source.version_id);
assert_eq!(moved.transition_status, rustfs_filemeta::TRANSITION_COMPLETE);
assert_same_remote_tuple(moved, &source);
assert!(
backend.contains(&source.transitioned_objname).await,
"the remote object must stay in the tier"
);
assert_eq!(backend.remove_count().await, 0, "no remote delete may be issued");
assert_eq!(backend.exact_remove_count(), 0, "no remote delete may be issued");
});
}
/// A newer unversioned object written to the target replaces a restored
/// null source instead of receiving a copy of it. The source's remote
/// object is then referenced nowhere, so the source cleanup must keep a
/// tier free version to delete it (rustfs/backlog#2873, defect D).
async fn run_superseded_restored_null_version_case() {
let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await;
let bucket = test_bucket("superseded-restored");
let object = "restored.bin";
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create the unversioned bucket");
let tier_name = format!("DECOMTIER{}", &uuid::Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
let backend = register_mock_tier(&store.pools[0].instance_ctx().tier_config_mgr(), &tier_name).await;
let written = put_transitioned_source_version(&store, &bucket, object, &tier_name, false, None).await;
restore_source_version(&store, &bucket, object, false, &written).await;
let source_versions = exact_versions(&store, 0, &bucket, object).await;
assert_eq!(source_versions.len(), 1);
let source = source_versions[0].clone();
assert!(source.version_id.is_none(), "the source must be a null version");
assert!(!source.is_remote(), "a restored source keeps its local copy");
let newer = b"written to the target after the source".to_vec();
store.pools[1]
.put_object(&bucket, object, &mut PutObjReader::from_vec(newer.clone()), &ObjectOptions::default())
.await
.expect("write a newer null version on the target");
start_transitioned_entry_decommission(&store, &bucket).await;
store
.decommission_entry_for_test_with_bucket_incarnation(
0,
MetaCacheEntry {
name: object.to_string(),
..Default::default()
},
bucket.clone(),
store.pools[0].get_disks_by_key(object),
)
.await
.expect("the superseded entry should finish");
let left = exact_versions(&store, 0, &bucket, object).await;
assert_eq!(left.len(), 1, "the source must keep one tier free version: {left:?}");
assert!(left[0].tier_free_version());
assert_same_remote_tuple(&left[0], &source);
let target = exact_versions(&store, 1, &bucket, object).await;
assert_eq!(target.len(), 1, "the target keeps only its newer version: {target:?}");
assert!(target[0].transition_status.is_empty(), "the newer version references no remote object");
assert!(exact_versions(&store, 2, &bucket, object).await.is_empty());
let mut reader = store.pools[1]
.get_object_reader(&bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
.await
.expect("the newer target version should be readable");
let mut body = Vec::new();
reader.stream.read_to_end(&mut body).await.expect("drain the newer version");
assert_eq!(body, newer);
assert!(backend.contains(&source.transitioned_objname).await);
assert_eq!(backend.remove_count().await, 0);
}
#[test]
#[serial_test::serial]
fn decommission_keeps_a_free_version_for_a_restored_null_version_superseded_on_target() {
run_large_stack_async_test("decommission-superseded-restored-null-version", run_superseded_restored_null_version_case);
}
/// Of two transitioned versions in one entry, lifecycle expires the
/// noncurrent one while its queued expiry has not run yet, and the current
/// one moves. The source cleanup removes both, but only the expired
/// version keeps a tier free version: it is the only owner of that remote
/// object, while the target copy owns the other (rustfs/backlog#2873).
#[test]
#[serial_test::serial]
fn decommission_keeps_a_free_version_only_for_the_expired_version_of_an_entry() {
run_large_stack_async_test("decommission-expired-and-moved-versions", || async {
let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await;
let bucket = test_bucket("expired-and-moved");
let object = "versions.bin";
store
.make_bucket(
&bucket,
&MakeBucketOptions {
versioning_enabled: true,
..Default::default()
},
)
.await
.expect("create the versioned bucket");
let tier_name = format!("DECOMTIER{}", &uuid::Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
let backend = register_mock_tier(&store.pools[0].instance_ctx().tier_config_mgr(), &tier_name).await;
let now = time::OffsetDateTime::now_utc();
let expired =
put_transitioned_source_version(&store, &bucket, object, &tier_name, true, Some(now - time::Duration::days(10)))
.await;
let moved =
put_transitioned_source_version(&store, &bucket, object, &tier_name, true, Some(now - time::Duration::days(5)))
.await;
let source_versions = exact_versions(&store, 0, &bucket, object).await;
assert_eq!(source_versions.len(), 2);
let source = |version_id| {
source_versions
.iter()
.find(|version| version.version_id == version_id)
.cloned()
.expect("each written version should be loaded")
};
let (expired, moved) = (source(expired.version_id), source(moved.version_id));
assert_ne!(expired.transitioned_objname, moved.transitioned_objname);
let lifecycle_xml = r#"<?xml version="1.0" encoding="UTF-8"?>
<LifecycleConfiguration>
<Rule>
<ID>expire-noncurrent</ID>
<Status>Enabled</Status>
<Filter>
<Prefix></Prefix>
</Filter>
<NoncurrentVersionExpiration>
<NoncurrentDays>1</NoncurrentDays>
</NoncurrentVersionExpiration>
</Rule>
</LifecycleConfiguration>"#;
crate::bucket::metadata_sys::update_in(
&store.ctx,
&bucket,
crate::bucket::metadata::BUCKET_LIFECYCLE_CONFIG,
lifecycle_xml.as_bytes().to_vec(),
)
.await
.expect("store the noncurrent expiration rule");
// Load it the way the decommission routine does for each bucket.
let lifecycle = crate::bucket::lifecycle::get_expiry_configs(&store, &bucket)
.await
.expect("the bucket lifecycle configuration should load")
.lifecycle
.map(|config| (*config).clone())
.expect("the bucket should have a lifecycle configuration");
start_transitioned_entry_decommission(&store, &bucket).await;
let expiry_queue = crate::bucket::lifecycle::bucket_lifecycle_ops::HeldExpiryQueue::hold().await;
let result = store
.decommission_entry_for_test_with_lifecycle(
0,
MetaCacheEntry {
name: object.to_string(),
..Default::default()
},
bucket.clone(),
store.pools[0].get_disks_by_key(object),
Some(lifecycle),
)
.await;
let pending_expiries = expiry_queue.pending_tasks().await;
expiry_queue.release().await;
result.expect("the entry should finish");
assert_eq!(pending_expiries, 1, "the noncurrent version's expiry must stay queued");
let left = exact_versions(&store, 0, &bucket, object).await;
assert_eq!(left.len(), 1, "only the expired version keeps a tier free version: {left:?}");
assert!(left[0].tier_free_version());
assert_same_remote_tuple(&left[0], &expired);
let mut owners = Vec::new();
for target_pool_index in [1, 2] {
owners.extend(exact_versions(&store, target_pool_index, &bucket, object).await);
}
assert_eq!(owners.len(), 1, "only the current version moves: {owners:?}");
assert!(!owners[0].tier_free_version());
assert_eq!(owners[0].version_id, moved.version_id);
assert_same_remote_tuple(&owners[0], &moved);
for remote in [&expired.transitioned_objname, &moved.transitioned_objname] {
assert!(backend.contains(remote).await, "no remote object may be deleted by the move");
}
assert_eq!(backend.remove_count().await, 0);
});
}
#[tokio::test]
#[serial_test::serial]
async fn no_active_external_commit_probe_blocks_decommission_activation_until_commit_owns_it() {
crates/ecstore/src/data_movement/mod.rs
+192 -37
View File
@@ -1128,6 +1128,43 @@ pub(crate) fn source_cleanup_version_identity(version: &FileInfo) -> SourceClean
}
}
/// Source versions whose transitioned remote object a committed data-movement
/// copy now references from the target pool.
///
/// Removing a transitioned version from xl.meta normally appends a tier free
/// version that owns the remote delete. After a move the target copy is the
/// live reference to that remote object, so the source must not get an owner
/// of its own: it would fail the decommission post-check and could only be
/// consumed once the target copy is gone. A version that did not move keeps
/// its free version: one lifecycle expired during the move, or one a newer
/// unversioned target object superseded ([`DataMovementCopy::Superseded`]).
#[doc(hidden)]
#[derive(Clone)]
pub struct TransferredTierVersions(Arc<[SourceCleanupVersionIdentity]>);
impl TransferredTierVersions {
fn new(versions: &[SourceCleanupVersionIdentity]) -> Option<Self> {
(!versions.is_empty()).then(|| Self(versions.iter().cloned().map(Self::without_name).collect()))
}
// Every version of one cleanup belongs to the same key, but the mover and
// the delete name a directory object differently.
fn without_name(mut identity: SourceCleanupVersionIdentity) -> SourceCleanupVersionIdentity {
identity.name.clear();
identity
}
pub(crate) fn contains(&self, version: &FileInfo) -> bool {
self.0.contains(&Self::without_name(source_cleanup_version_identity(version)))
}
}
impl std::fmt::Debug for TransferredTierVersions {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_tuple("TransferredTierVersions").field(&self.0.len()).finish()
}
}
fn source_cleanup_versions_match_with_allowed_missing(
expected: &FileInfoVersions,
current: &FileInfoVersions,
@@ -1357,12 +1394,16 @@ async fn pause_source_cleanup_before_delete(bucket: &str, object: &str) {
}
}
/// `transferred_tier_versions` lists the transitioned versions whose copy
/// settled as [`DataMovementCopy::Committed`]; see [`TransferredTierVersions`].
#[allow(clippy::too_many_arguments)]
pub(crate) async fn cleanup_source_entry_if_unchanged(
set: Arc<SetDisks>,
bucket: &str,
object: &str,
expected: &FileInfoVersions,
allowed_missing: &[SourceCleanupVersionIdentity],
transferred_tier_versions: &[SourceCleanupVersionIdentity],
bucket_fence: SourceCleanupBucketFence<'_>,
op_label: &str,
) -> std::result::Result<ObjectInfo, SourceCleanupError> {
@@ -1408,6 +1449,7 @@ pub(crate) async fn cleanup_source_entry_if_unchanged(
delete_prefix: true,
delete_prefix_object: true,
data_movement: true,
data_movement_transferred_tier_versions: TransferredTierVersions::new(transferred_tier_versions),
no_lock: true,
expected_bucket_incarnation_id: bucket_fence.expected_incarnation_id,
..Default::default()
@@ -1463,6 +1505,19 @@ async fn find_data_movement_target_info(
}
}
/// How a data-movement copy settled on its target pool.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum DataMovementCopy {
/// The target holds this exact source version, committed by this copy or
/// found equivalent from an earlier attempt, so it also holds any remote
/// tier tuple of the version.
Committed,
/// A newer unversioned object on the target replaced the source. The
/// target holds no copy of the source and no reference to its remote tier
/// object.
Superseded,
}
#[allow(dead_code, reason = "resume adjudication asserted by this file's tests (backlog#1823)")]
fn resolve_data_movement_overwrite_resume_result(
err: &Error,
@@ -1471,16 +1526,18 @@ fn resolve_data_movement_overwrite_resume_result(
src_pool_idx: usize,
target_pool_idx: usize,
) -> Result<bool> {
resolve_data_movement_overwrite_resume_result_for(
Ok(resolve_data_movement_overwrite_resume_copy(
err,
target_result,
source,
src_pool_idx,
target_pool_idx,
data_movement_part_checksum_writer_enabled(),
)
)?
.is_some())
}
#[cfg(test)]
fn resolve_data_movement_overwrite_resume_result_for(
err: &Error,
target_result: Result<Option<ObjectInfo>>,
@@ -1489,30 +1546,47 @@ fn resolve_data_movement_overwrite_resume_result_for(
target_pool_idx: usize,
compare_part_checksums: bool,
) -> Result<bool> {
Ok(resolve_data_movement_overwrite_resume_copy(
err,
target_result,
source,
src_pool_idx,
target_pool_idx,
compare_part_checksums,
)?
.is_some())
}
fn resolve_data_movement_overwrite_resume_copy(
err: &Error,
target_result: Result<Option<ObjectInfo>>,
source: &ObjectInfo,
src_pool_idx: usize,
target_pool_idx: usize,
compare_part_checksums: bool,
) -> Result<Option<DataMovementCopy>> {
if scanner_backlog::is_scanner_pause_backlog(&source.bucket, &source.name)
|| !should_check_data_movement_overwrite_resume(err)
|| !should_check_data_movement_resume_target(src_pool_idx, target_pool_idx)
{
return Ok(false);
return Ok(None);
}
let Some(target) = target_result? else {
return Ok(false);
return Ok(None);
};
if is_equivalent_data_movement_object_identity(source, &target, true, compare_part_checksums) {
return Ok(true);
if is_equivalent_data_movement_object_identity(source, &target, true, compare_part_checksums)
|| (compare_part_checksums && is_legacy_data_movement_checksum_target(source, &target))
|| is_data_movement_upload_takeover_target(source, &target, compare_part_checksums)
{
return Ok(Some(DataMovementCopy::Committed));
}
if compare_part_checksums && is_legacy_data_movement_checksum_target(source, &target) {
return Ok(true);
}
if is_data_movement_upload_takeover_target(source, &target, compare_part_checksums) {
return Ok(true);
}
Ok(matches!(err, Error::PreconditionFailed) && is_superseding_unversioned_data_movement_object(source, &target))
Ok(
(matches!(err, Error::PreconditionFailed) && is_superseding_unversioned_data_movement_object(source, &target))
.then_some(DataMovementCopy::Superseded),
)
}
#[derive(Clone, Copy)]
@@ -1529,13 +1603,13 @@ async fn should_treat_data_movement_overwrite_as_complete(
err: &Error,
compare_part_checksums: bool,
capacity: DataMovementOverwriteCapacity,
) -> Result<bool> {
) -> Result<Option<DataMovementCopy>> {
if !should_check_data_movement_overwrite_resume(err) {
return Ok(false);
return Ok(None);
}
let (src_pool_idx, target_pool_idx) = pool_indices;
let equivalent = resolve_data_movement_overwrite_resume_result_for(
let copy = resolve_data_movement_overwrite_resume_copy(
err,
find_data_movement_target_info(store, target_pool_idx, bucket, object_info).await,
object_info,
@@ -1543,7 +1617,9 @@ async fn should_treat_data_movement_overwrite_as_complete(
target_pool_idx,
compare_part_checksums,
)?;
if equivalent && let Some(owner) = capacity.owner {
if copy.is_some()
&& let Some(owner) = capacity.owner
{
let expected_data_bytes = capacity
.expected_data_bytes
.ok_or_else(|| Error::other("equivalent data-movement target cannot reconcile unknown committed data size"))?;
@@ -1551,7 +1627,7 @@ async fn should_treat_data_movement_overwrite_as_complete(
.reconcile_decommission_capacity_after_equivalent_target(owner, target_pool_idx, expected_data_bytes)
.await?;
}
Ok(equivalent)
Ok(copy)
}
fn data_movement_part_stage_error(
@@ -1611,7 +1687,7 @@ pub(crate) async fn migrate_decommission_object(
source_bucket_incarnation_id: Option<uuid::Uuid>,
op_label: &str,
capacity_owner: Option<DecommissionCapacityOwner>,
) -> Result<()> {
) -> Result<DataMovementCopy> {
let source = rd.object_info.clone();
let mutation_fence = store
.acquire_decommission_object_mutation_fence(&bucket, &source.name)
@@ -1645,7 +1721,7 @@ pub(crate) async fn migrate_object(
rd: GetObjectReader,
source_bucket_incarnation_id: Option<uuid::Uuid>,
op_label: &str,
) -> Result<()> {
) -> Result<DataMovementCopy> {
migrate_object_with_lock_lost_signal(store, pool_idx, bucket, rd, source_bucket_incarnation_id, op_label, None).await
}
@@ -1658,7 +1734,7 @@ pub(crate) async fn migrate_object_with_lock_lost_signal(
source_bucket_incarnation_id: Option<uuid::Uuid>,
op_label: &str,
lock_lost_signal: Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
) -> Result<()> {
) -> Result<DataMovementCopy> {
migrate_object_inner(
store,
pool_idx,
@@ -1684,7 +1760,7 @@ async fn migrate_object_inner(
lock_lost_signal: Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
capacity_owner: Option<DecommissionCapacityOwner>,
mutation_fence: Option<DecommissionFixedReadAnchor>,
) -> Result<()> {
) -> Result<DataMovementCopy> {
if scanner_backlog::is_scanner_pause_backlog(&bucket, &rd.object_info.name) {
return Err(Error::other("scanner pause backlog requires native retirement handoff"));
}
@@ -1758,7 +1834,7 @@ async fn migrate_object_inner(
bucket.as_str(),
object_info.name.as_str()
);
return Ok(());
return Ok(DataMovementCopy::Committed);
}
let mut cleanup_opts = data_movement_abort_opts(
&object_info,
@@ -1816,7 +1892,7 @@ async fn migrate_object_inner(
}
};
let abort_multipart_flag = new_multipart_abort_flag();
let multipart_result: Result<()> = async {
let multipart_result: Result<DataMovementCopy> = async {
let mut parts = vec![CompletePart::default(); object_info.parts.len()];
let reader = Arc::new(Mutex::new(rd.stream));
for (i, part) in object_info.parts.iter().enumerate() {
@@ -1930,7 +2006,7 @@ async fn migrate_object_inner(
)
.await
{
if should_treat_data_movement_overwrite_as_complete(
if let Some(copy) = should_treat_data_movement_overwrite_as_complete(
store.as_ref(),
(pool_idx, target_pool_idx),
bucket.as_str(),
@@ -1949,7 +2025,7 @@ async fn migrate_object_inner(
bucket.as_str(),
object_info.name.as_str()
);
return Ok(());
return Ok(copy);
}
error!("{op_label}: complete_multipart_upload err {:?}", &err);
@@ -1963,11 +2039,13 @@ async fn migrate_object_inner(
}
mark_multipart_upload_completed(&abort_multipart_flag);
Ok(())
Ok(DataMovementCopy::Committed)
}
.await;
if multipart_result.is_ok() && should_abort_multipart_upload(&abort_multipart_flag) {
if let Ok(copy) = multipart_result
&& should_abort_multipart_upload(&abort_multipart_flag)
{
let mut abort_opts = data_movement_abort_opts(
&object_info,
pool_idx,
@@ -1991,9 +2069,9 @@ async fn migrate_object_inner(
)
.await;
match abort_result {
Ok(()) => return Ok(()),
Ok(()) => return Ok(copy),
Err(abort_err) if is_err_invalid_upload_id(&abort_err) => {
if should_treat_data_movement_overwrite_as_complete(
if let Some(copy) = should_treat_data_movement_overwrite_as_complete(
store.as_ref(),
(pool_idx, target_pool_idx),
bucket.as_str(),
@@ -2007,7 +2085,7 @@ async fn migrate_object_inner(
)
.await?
{
return Ok(());
return Ok(copy);
}
return Err(data_movement_stage_error(
op_label,
@@ -2100,7 +2178,7 @@ async fn migrate_object_inner(
return Err(primary_err);
}
return Ok(());
return multipart_result;
}
let mut data = data_movement_put_object_reader(bucket.as_str(), &object_info, rd, op_label)?;
@@ -2128,7 +2206,7 @@ async fn migrate_object_inner(
.await
.map_err(|err| data_movement_stage_error(op_label, "prepare_put_object", &bucket, &object_info.name, err))?;
if let Err(err) = put_result {
if should_treat_data_movement_overwrite_as_complete(
if let Some(copy) = should_treat_data_movement_overwrite_as_complete(
store.as_ref(),
(pool_idx, target_pool_idx),
bucket.as_str(),
@@ -2147,7 +2225,7 @@ async fn migrate_object_inner(
bucket.as_str(),
object_info.name.as_str()
);
return Ok(());
return Ok(copy);
}
error!("{op_label}: put_object err {:?}", &err);
@@ -2160,7 +2238,7 @@ async fn migrate_object_inner(
));
}
Ok(())
Ok(DataMovementCopy::Committed)
}
#[cfg(test)]
@@ -4357,6 +4435,83 @@ mod tests {
}
}
/// The resume adjudication tells a target that holds the source version
/// (equivalent, an old node's upload takeover, a legacy checksum target)
/// from a newer unversioned object that replaced it, and accepts neither
/// when the target's remote tier tuple differs (rustfs/backlog#2873).
#[test]
fn overwrite_resume_copy_separates_committed_from_superseded_targets() {
let classify = |err: &Error, target: &ObjectInfo, source: &ObjectInfo| {
resolve_data_movement_overwrite_resume_copy(err, Ok(Some(target.clone())), source, 0, 1, true)
.expect("the target should be evaluated")
};
let overwrite = Error::DataMovementOverwriteErr("bucket".to_string(), "object".to_string(), "version".to_string());
let mut source = overwrite_equivalence_source();
source.transitioned_object.status = rustfs_filemeta::TRANSITION_COMPLETE.to_string();
source.transitioned_object.tier = "WARM".to_string();
source.transitioned_object.name = "remote/object".to_string();
source.transitioned_object.version_id = "remote-version".to_string();
assert_eq!(classify(&overwrite, &source, &source), Some(DataMovementCopy::Committed));
let mut legacy = source.clone();
let mut legacy_parts = legacy.parts.as_ref().clone();
legacy_parts[0].checksums = None;
legacy.parts = Arc::new(legacy_parts);
let legacy_metadata = Arc::make_mut(&mut legacy.user_defined);
rustfs_utils::http::insert_str(legacy_metadata, SUFFIX_DATA_MOVED, "true".to_string());
rustfs_utils::http::insert_str(legacy_metadata, SUFFIX_DATA_MOVED_TAGS, "v1:tag=value".to_string());
assert_eq!(classify(&overwrite, &legacy, &source), Some(DataMovementCopy::Committed));
let mut takeover_source = source.clone();
let mut takeover_parts = takeover_source.parts.as_ref().clone();
takeover_parts[0].checksums = None;
takeover_source.parts = Arc::new(takeover_parts);
let mut takeover = ObjectInfo {
mod_time: OffsetDateTime::UNIX_EPOCH.checked_add(time::Duration::SECOND),
..takeover_source.clone()
};
rustfs_utils::http::insert_str(
Arc::make_mut(&mut takeover.user_defined),
SUFFIX_DATA_MOVEMENT_UPLOAD,
data_movement_upload_identity(&takeover_source),
);
let invalid_upload = Error::InvalidUploadID("bucket".to_string(), "object".to_string(), "upload-id".to_string());
assert_eq!(classify(&invalid_upload, &takeover, &takeover_source), Some(DataMovementCopy::Committed));
for (err, target, target_source) in [
(&overwrite, &source, &source),
(&overwrite, &legacy, &source),
(&invalid_upload, &takeover, &takeover_source),
] {
let mut other_remote = target.clone();
other_remote.transitioned_object.name = "remote/other".to_string();
assert_eq!(
classify(err, &other_remote, target_source),
None,
"a different remote tuple is not a copy"
);
}
let mut null_source = source;
null_source.version_id = None;
let newer_null = ObjectInfo {
etag: Some("etag-client-write".to_string()),
mod_time: OffsetDateTime::UNIX_EPOCH.checked_add(time::Duration::SECOND),
transitioned_object: Default::default(),
..null_source.clone()
};
assert_eq!(
classify(&Error::PreconditionFailed, &newer_null, &null_source),
Some(DataMovementCopy::Superseded)
);
assert_eq!(
classify(&overwrite, &newer_null, &null_source),
None,
"only a precondition conflict can be superseded"
);
}
#[test]
fn test_overwrite_error_rejects_newer_unversioned_target() {
let source = ObjectInfo {
crates/ecstore/src/object_api/types.rs
+7
View File
@@ -991,6 +991,12 @@ pub struct ObjectOptions {
pub data_movement: bool,
pub raw_data_movement_read: bool,
/// Data-movement source cleanup only: transitioned source versions whose
/// committed target copy now references their remote object. Deleting one
/// of them appends no tier free version. Storage-owned; callers outside
/// `ecstore` must leave it unset.
#[doc(hidden)]
pub data_movement_transferred_tier_versions: Option<crate::data_movement::TransferredTierVersions>,
/// Internal, in-memory protection context. Never populated from S3 metadata.
/// None selects the rollout default only for a new write, not for a rewrite.
#[doc(hidden)]
@@ -1170,6 +1176,7 @@ impl std::fmt::Debug for ObjectOptions {
.field("scanner_publication_commit_scope", &self.scanner_publication_commit_scope)
.field("data_movement", &self.data_movement)
.field("raw_data_movement_read", &self.raw_data_movement_read)
.field("data_movement_transferred_tier_versions", &self.data_movement_transferred_tier_versions)
.field("include_part_checksums", &self.include_part_checksums)
.field("src_pool_idx", &self.src_pool_idx)
.field("user_defined_count", &self.user_defined.len())
crates/ecstore/src/services/rebalance/entry.rs
+383 -1
View File
@@ -283,6 +283,7 @@ impl ECStore {
let mut expired: usize = 0;
let mut cleanup_preflight_allowed_missing = Vec::new();
let mut cleanup_preflight_vanished = Vec::new();
let mut cleanup_transferred_tier_versions = Vec::new();
let mut stats_updates = Vec::with_capacity(fivs.versions.len());
for version in fivs.versions.iter() {
ensure_rebalance_entry_active(&cancel)?;
@@ -476,6 +477,9 @@ impl ECStore {
if should_count_rebalance_version_complete(&result) {
rebalanced += 1;
}
if result.moved && !result.superseded && version.transition_status == rustfs_filemeta::TRANSITION_COMPLETE {
cleanup_transferred_tier_versions.push(data_movement::source_cleanup_version_identity(version));
}
}
if should_cleanup_rebalance_source_entry(rebalanced, fivs.versions.len(), expired) {
@@ -500,6 +504,7 @@ impl ECStore {
entry.name.as_str(),
&cleanup_expected,
&cleanup_preflight_allowed_missing,
&cleanup_transferred_tier_versions,
data_movement::SourceCleanupBucketFence {
expected_incarnation_id: bucket_configs.bucket_incarnation_id,
lifecycle_guard: bucket_incarnation_fence
@@ -595,7 +600,7 @@ impl ECStore {
rd: GetObjectReader,
expected_bucket_incarnation_id: Option<uuid::Uuid>,
lock_lost_signal: Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
) -> Result<()> {
) -> Result<data_movement::DataMovementCopy> {
data_movement::migrate_object_with_lock_lost_signal(
self,
pool_idx,
@@ -1761,6 +1766,383 @@ mod tests {
fixture.assert_moved().await;
}
/// Runs a case on a thread whose stack fits the transitioned copy path in
/// debug builds; libtest's 2 MiB default does not under plain `cargo test`.
fn run_large_stack_async_test<C, F>(name: &str, case: C)
where
C: FnOnce() -> F + Send + 'static,
F: std::future::Future<Output = ()> + 'static,
{
const STACK_SIZE: usize = if cfg!(debug_assertions) {
8 * rustfs_config::DEFAULT_THREAD_STACK_SIZE
} else if cfg!(target_os = "macos") {
2 * rustfs_config::DEFAULT_THREAD_STACK_SIZE
} else {
rustfs_config::DEFAULT_THREAD_STACK_SIZE
};
std::thread::Builder::new()
.name(name.to_string())
.stack_size(STACK_SIZE)
.spawn(move || {
let runtime = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.worker_threads(2)
.thread_stack_size(STACK_SIZE)
.build()
.expect("large-stack rebalance test runtime should build");
runtime.block_on(case());
})
.expect("large-stack rebalance test thread should spawn")
.join()
.expect("large-stack rebalance test thread should complete");
}
const TRANSITIONED_REBALANCE_ID: &str = "rebalance-transitioned-source-cleanup";
/// One transitioned source version on pool 0 of a two-pool store with an
/// active rebalance run. A restored source keeps its local copy, so the
/// rebalance moves it as data that keeps the transition tuple instead of
/// as tier metadata (rustfs/backlog#2873, defect D).
struct TransitionedRebalance {
_temp_dirs: Vec<tempfile::TempDir>,
store: Arc<ECStore>,
bucket: &'static str,
object: &'static str,
versioned: bool,
incarnation: uuid::Uuid,
payload: Vec<u8>,
source: FileInfo,
source_set: Arc<SetDisks>,
backend: crate::services::tier::test_util::MockWarmBackend,
}
impl TransitionedRebalance {
async fn new(versioned: bool, restored: bool) -> Self {
Self::with_source(versioned, restored, false).await
}
/// With `multipart`, the source is a two-part object, so the copy
/// settles on the multipart path rather than a single PUT.
async fn with_source(versioned: bool, restored: bool, multipart: bool) -> Self {
let (temp_dirs, store, _peer) =
crate::services::rebalance::test_two_pool_stores_with_isolated_node_contexts(None).await;
let bucket = "rebalance-transitioned";
let object = "transitioned.bin";
// An active run blocks the data-usage publication bucket creation needs.
Box::pin(store.make_bucket(
bucket,
&MakeBucketOptions {
versioning_enabled: versioned,
..Default::default()
},
))
.await
.expect("create bucket");
let incarnation = store.bucket_incarnation_id(bucket).await.expect("bucket incarnation");
let tier_name = format!("REBALTIER{}", &uuid::Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
let backend = crate::services::tier::test_util::register_mock_tier(&store.tier_config_mgr(), &tier_name).await;
let source_set = store.pools[0].get_disks_by_key(object);
let write_opts = ObjectOptions {
versioned,
expected_bucket_incarnation_id: Some(incarnation),
..Default::default()
};
let (payload, written) = if multipart {
// Set-level multipart writes with a bucket incarnation require
// the caller's lifecycle fence; stage these without one.
let write_opts = ObjectOptions {
versioned,
..Default::default()
};
let parts = [vec![0x31; 5 << 20], vec![0x32; 1]];
let upload = Box::pin(source_set.new_multipart_upload(bucket, object, &write_opts))
.await
.expect("create the source multipart upload");
let mut completed = Vec::with_capacity(parts.len());
for (part_num, body) in (1..).zip(parts.iter()) {
let part = Box::pin(source_set.put_object_part(
bucket,
object,
&upload.upload_id,
part_num,
&mut PutObjReader::from_vec(body.clone()),
&write_opts,
))
.await
.expect("stage the source part");
completed.push(CompletePart {
part_num: part.part_num,
etag: part.etag,
..Default::default()
});
}
let written = Box::pin(Arc::clone(&source_set).complete_multipart_upload(
bucket,
object,
&upload.upload_id,
completed,
&write_opts,
))
.await
.expect("complete the source multipart object");
(parts.concat(), written)
} else {
let payload = b"transitioned before the rebalance".to_vec();
let written =
Box::pin(source_set.put_object(bucket, object, &mut PutObjReader::from_vec(payload.clone()), &write_opts))
.await
.expect("seed the source version");
(payload, written)
};
let version_id = written.version_id.map(|version| version.to_string());
Box::pin(store.pools[0].transition_object(
bucket,
object,
&ObjectOptions {
versioned,
version_id: version_id.clone(),
transition: crate::bucket::lifecycle::lifecycle::TransitionOptions {
status: crate::bucket::lifecycle::lifecycle::TRANSITION_PENDING.to_string(),
tier: tier_name,
etag: written.etag.clone().expect("the source should have an ETag"),
..Default::default()
},
mod_time: written.mod_time,
..Default::default()
},
))
.await
.expect("transition the source before the rebalance");
if restored {
Box::pin(Arc::clone(&store).restore_transitioned_object(
bucket,
object,
&ObjectOptions {
versioned,
version_id,
transition: crate::bucket::lifecycle::lifecycle::TransitionOptions {
restore_request: s3s::dto::RestoreRequest {
days: Some(1),
..Default::default()
},
..Default::default()
},
..Default::default()
},
))
.await
.expect("restore the transitioned source");
}
let rebalance_meta = active_rebalance_meta(TRANSITIONED_REBALANCE_ID);
rebalance_meta
.save(Arc::clone(&store.pools[0]))
.await
.expect("persist the active rebalance run");
*store.rebalance_meta.write().await = Some(rebalance_meta);
let seeded = source_set
.load_file_info_versions_exact(bucket, object)
.await
.expect("read seeded versions")
.expect("the transitioned source must exist");
assert_eq!(seeded.versions.len(), 1);
let source = seeded.versions[0].clone();
assert_eq!(source.transition_status, TRANSITION_COMPLETE);
assert_eq!(source.version_id.is_some(), versioned);
assert_eq!(!source.is_remote(), restored, "a restored source keeps its local copy");
assert_eq!(source.parts.len() > 1, multipart, "unexpected source part layout: {:?}", source.parts);
assert!(
backend.metadata(&source.transitioned_objname).await.is_some(),
"the transition must upload the remote object"
);
Self {
_temp_dirs: temp_dirs,
store,
bucket,
object,
versioned,
incarnation,
payload,
source,
source_set,
backend,
}
}
async fn run_entry(&self) -> RebalanceEntryOutcome {
let entry = metacache_entry_from_source(&self.source_set, self.bucket, self.object).await;
Box::pin(Arc::clone(&self.store).rebalance_entry(
RebalanceEntryTarget {
bucket: self.bucket.to_string(),
pool_index: 0,
},
entry,
Arc::clone(&self.source_set),
Arc::new(RebalanceBucketConfigs {
bucket_incarnation_id: Some(self.incarnation),
..Default::default()
}),
Arc::from(TRANSITIONED_REBALANCE_ID),
CancellationToken::new(),
))
.await
.expect("rebalance entry")
}
async fn versions_on(&self, pool_index: usize) -> Vec<FileInfo> {
self.store.pools[pool_index]
.get_disks_by_key(self.object)
.load_file_info_versions_exact(self.bucket, self.object)
.await
.expect("read pool versions")
.map(|versions| versions.versions)
.unwrap_or_default()
}
fn assert_same_remote_tuple(&self, version: &FileInfo) {
assert_eq!(version.transition_tier, self.source.transition_tier);
assert_eq!(version.transitioned_objname, self.source.transitioned_objname);
assert_eq!(version.transition_version_id, self.source.transition_version_id);
assert_eq!(
crate::services::tier::tier::tier_destination_id_from_metadata(&version.metadata)
.expect("the copy's destination identity should decode"),
crate::services::tier::tier::tier_destination_id_from_metadata(&self.source.metadata)
.expect("the source destination identity should decode"),
);
}
async fn assert_remote_object_kept(&self) {
assert!(
self.backend.metadata(&self.source.transitioned_objname).await.is_some(),
"the remote object must stay in the tier"
);
assert_eq!(self.backend.remove_count().await, 0, "no remote delete may be issued");
assert_eq!(self.backend.exact_remove_count(), 0, "no remote delete may be issued");
}
/// The target owns the version and its remote object, and the source
/// keeps neither the version nor a tier free version.
async fn assert_moved(&self) {
let left = self.versions_on(0).await;
assert!(left.is_empty(), "the source must keep no version or tier free version: {left:?}");
let target = self.versions_on(1).await;
assert_eq!(target.len(), 1, "the target must hold the version and no tier free version: {target:?}");
let moved = &target[0];
assert!(!moved.tier_free_version());
assert_eq!(moved.version_id, self.source.version_id);
assert_eq!(moved.transition_status, TRANSITION_COMPLETE);
self.assert_same_remote_tuple(moved);
self.assert_remote_object_kept().await;
let mut reader = self.store.pools[1]
.get_object_reader(
self.bucket,
self.object,
None,
HeaderMap::new(),
&ObjectOptions {
versioned: self.versioned,
version_id: self.source.version_id.map(|version| version.to_string()),
..Default::default()
},
)
.await
.expect("the moved version should be readable");
let mut body = Vec::new();
reader.stream.read_to_end(&mut body).await.expect("drain the moved version");
assert_eq!(body, self.payload);
}
}
#[test]
#[serial_test::serial]
fn real_rebalance_of_a_transitioned_version_leaves_no_source_free_version() {
run_large_stack_async_test("rebalance-transitioned-version", || async {
let fixture = TransitionedRebalance::new(true, false).await;
assert_eq!(fixture.run_entry().await, RebalanceEntryOutcome::Completed);
fixture.assert_moved().await;
});
}
#[test]
#[serial_test::serial]
fn real_rebalance_of_a_transitioned_null_version_leaves_no_source_free_version() {
run_large_stack_async_test("rebalance-transitioned-null-version", || async {
let fixture = TransitionedRebalance::new(false, false).await;
assert_eq!(fixture.run_entry().await, RebalanceEntryOutcome::Completed);
fixture.assert_moved().await;
});
}
/// A restored source moves as data; the copy keeps its transition tuple,
/// so the target owns the remote object as well.
#[test]
#[serial_test::serial]
fn real_rebalance_of_a_restored_transitioned_version_moves_its_remote_owner() {
run_large_stack_async_test("rebalance-restored-transitioned-version", || async {
let fixture = TransitionedRebalance::new(true, true).await;
assert_eq!(fixture.run_entry().await, RebalanceEntryOutcome::Completed);
fixture.assert_moved().await;
});
}
/// A newer unversioned object written to the target replaces a restored
/// null source instead of receiving a copy of it. Nothing references the
/// source's remote object any more, so its source cleanup must keep a tier
/// free version to delete it.
async fn run_superseded_restored_null_version_case(multipart: bool) {
let fixture = TransitionedRebalance::with_source(false, true, multipart).await;
let newer = b"written to the target after the source".to_vec();
Box::pin(fixture.store.pools[1].get_disks_by_key(fixture.object).put_object(
fixture.bucket,
fixture.object,
&mut PutObjReader::from_vec(newer.clone()),
&ObjectOptions {
expected_bucket_incarnation_id: Some(fixture.incarnation),
..Default::default()
},
))
.await
.expect("write a newer null version on the target");
assert_eq!(fixture.run_entry().await, RebalanceEntryOutcome::Completed);
let left = fixture.versions_on(0).await;
assert_eq!(left.len(), 1, "the source must keep one tier free version: {left:?}");
assert!(left[0].tier_free_version());
fixture.assert_same_remote_tuple(&left[0]);
let target = fixture.versions_on(1).await;
assert_eq!(target.len(), 1, "the target keeps only its newer version: {target:?}");
assert!(!target[0].tier_free_version());
assert!(target[0].transition_status.is_empty(), "the newer version references no remote object");
let mut reader = fixture.store.pools[1]
.get_object_reader(fixture.bucket, fixture.object, None, HeaderMap::new(), &ObjectOptions::default())
.await
.expect("the newer target version should be readable");
let mut body = Vec::new();
reader.stream.read_to_end(&mut body).await.expect("drain the newer version");
assert_eq!(body, newer);
fixture.assert_remote_object_kept().await;
}
#[test]
#[serial_test::serial]
fn real_rebalance_keeps_a_free_version_for_a_restored_null_version_superseded_on_target() {
run_large_stack_async_test("rebalance-superseded-restored-null-version", || {
run_superseded_restored_null_version_case(false)
});
}
/// The same superseded source as a multipart object settles on the
/// multipart copy path.
#[test]
#[serial_test::serial]
fn real_rebalance_keeps_a_free_version_for_a_restored_multipart_null_version_superseded_on_target() {
run_large_stack_async_test("rebalance-superseded-restored-multipart", || {
run_superseded_restored_null_version_case(true)
});
}
/// A version the mover reports as not found while it still exists must not
/// be deleted with the entry; cleanup is deferred and the source retained.
#[tokio::test]
crates/ecstore/src/services/rebalance/migration.rs
+50 -30
View File
@@ -1,5 +1,6 @@
use super::worker::{is_transient_rebalance_error, rebalance_migration_retry_delay, sleep_rebalance_migration_retry};
use crate::bucket::replication::replication_state_from_filemeta;
use crate::data_movement::DataMovementCopy;
use crate::data_usage::DATA_USAGE_CACHE_NAME;
use crate::error::{Error, Result, is_err_object_not_found, is_err_version_not_found};
use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions};
@@ -15,6 +16,9 @@ use tokio::time::Duration;
#[derive(Debug, Default, Clone)]
pub(crate) struct MigrationVersionResult {
pub moved: bool,
/// The copy found the source replaced by a newer unversioned object on the
/// target, which holds no copy of the source version.
pub superseded: bool,
pub ignored: bool,
pub cleanup_ignored: bool,
pub failed: bool,
@@ -177,7 +181,7 @@ pub(crate) async fn migrate_entry_version<Backend, F, Fut, D, DFut>(
where
Backend: MigrationBackend + ?Sized,
F: FnMut(usize, String, GetObjectReader) -> Fut + Send,
Fut: Future<Output = Result<()>> + Send,
Fut: Future<Output = Result<DataMovementCopy>> + Send,
D: FnMut(String, String, ObjectOptions) -> DFut + Send,
DFut: Future<Output = Result<ObjectInfo>> + Send,
{
@@ -214,7 +218,7 @@ pub(super) async fn migrate_entry_version_with_retry_wait<Backend, F, Fut, D, DF
where
Backend: MigrationBackend + ?Sized,
F: FnMut(usize, String, GetObjectReader) -> Fut + Send,
Fut: Future<Output = Result<()>> + Send,
Fut: Future<Output = Result<DataMovementCopy>> + Send,
D: FnMut(String, String, ObjectOptions) -> DFut + Send,
DFut: Future<Output = Result<ObjectInfo>> + Send,
W: FnMut(Duration) -> WFut + Send,
@@ -253,7 +257,7 @@ async fn migrate_entry_version_with_retry_wait_and_incarnation<Backend, F, Fut,
where
Backend: MigrationBackend + ?Sized,
F: FnMut(usize, String, GetObjectReader) -> Fut + Send,
Fut: Future<Output = Result<()>> + Send,
Fut: Future<Output = Result<DataMovementCopy>> + Send,
D: FnMut(String, String, ObjectOptions) -> DFut + Send,
DFut: Future<Output = Result<ObjectInfo>> + Send,
W: FnMut(Duration) -> WFut + Send,
@@ -264,6 +268,7 @@ where
if ignore_data_usage_cache && bucket == crate::disk::RUSTFS_META_BUCKET && version.name.contains(DATA_USAGE_CACHE_NAME) {
return MigrationVersionResult {
moved: false,
superseded: false,
ignored: true,
cleanup_ignored: false,
failed: false,
@@ -285,6 +290,7 @@ where
if is_err_object_not_found(&err) || is_err_version_not_found(&err) {
return MigrationVersionResult {
moved: false,
superseded: false,
ignored: true,
cleanup_ignored: true,
failed: false,
@@ -295,6 +301,7 @@ where
return MigrationVersionResult {
moved: false,
superseded: false,
ignored: false,
cleanup_ignored: false,
failed: true,
@@ -305,6 +312,7 @@ where
return MigrationVersionResult {
moved: true,
superseded: false,
ignored: false,
cleanup_ignored: false,
failed: false,
@@ -328,6 +336,7 @@ where
if is_err_object_not_found(&err) || is_err_version_not_found(&err) {
return MigrationVersionResult {
moved: false,
superseded: false,
ignored: true,
cleanup_ignored: true,
failed: false,
@@ -338,6 +347,7 @@ where
return MigrationVersionResult {
moved: false,
superseded: false,
ignored: false,
cleanup_ignored: false,
failed: true,
@@ -348,6 +358,7 @@ where
return MigrationVersionResult {
moved: true,
superseded: false,
ignored: false,
cleanup_ignored: false,
failed: false,
@@ -373,6 +384,7 @@ where
if is_err_object_not_found(&err) || is_err_version_not_found(&err) {
return MigrationVersionResult {
moved: false,
superseded: false,
ignored: true,
cleanup_ignored: true,
failed: false,
@@ -388,6 +400,7 @@ where
if attempt + 1 >= max_attempts || !is_transient_rebalance_error(err) {
return MigrationVersionResult {
moved: false,
superseded: false,
ignored: false,
cleanup_ignored: false,
failed: true,
@@ -401,39 +414,45 @@ where
}
};
if let Err(err) = transfer(pool_index, bucket.clone(), rd).await {
if is_err_object_not_found(&err) || is_err_version_not_found(&err) {
return MigrationVersionResult {
moved: false,
ignored: true,
cleanup_ignored: true,
failed: false,
stage: Some("write_target"),
error: None,
};
}
let copy = match transfer(pool_index, bucket.clone(), rd).await {
Ok(copy) => copy,
Err(err) => {
if is_err_object_not_found(&err) || is_err_version_not_found(&err) {
return MigrationVersionResult {
moved: false,
superseded: false,
ignored: true,
cleanup_ignored: true,
failed: false,
stage: Some("write_target"),
error: None,
};
}
last_error = Some(err);
let Some(err) = last_error.as_ref() else {
last_error = Some(err);
let Some(err) = last_error.as_ref() else {
continue;
};
if attempt + 1 >= max_attempts || !is_transient_rebalance_error(err) {
return MigrationVersionResult {
moved: false,
superseded: false,
ignored: false,
cleanup_ignored: false,
failed: true,
stage: Some("write_target"),
error: last_error,
};
}
wait_retry(rebalance_migration_retry_delay(attempt, err)).await;
continue;
};
if attempt + 1 >= max_attempts || !is_transient_rebalance_error(err) {
return MigrationVersionResult {
moved: false,
ignored: false,
cleanup_ignored: false,
failed: true,
stage: Some("write_target"),
error: last_error,
};
}
wait_retry(rebalance_migration_retry_delay(attempt, err)).await;
continue;
}
};
return MigrationVersionResult {
moved: true,
superseded: copy == DataMovementCopy::Superseded,
ignored: false,
cleanup_ignored: false,
failed: false,
@@ -444,6 +463,7 @@ where
MigrationVersionResult {
moved: false,
superseded: false,
ignored: false,
cleanup_ignored: false,
failed: true,
crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs
+17 -17
View File
@@ -59,7 +59,7 @@ use super::{
};
use crate::bucket::replication::{ReplicationState, ReplicationStatusType, replication_state_to_filemeta};
use crate::data_movement;
use crate::data_movement::SourceCleanupError;
use crate::data_movement::{DataMovementCopy, SourceCleanupError};
use crate::data_usage::DATA_USAGE_CACHE_NAME;
use crate::disk::RUSTFS_META_BUCKET;
use crate::disk::error::DiskError;
@@ -286,7 +286,7 @@ async fn test_migrate_entry_version_remote_version_is_moved_without_transfer() {
let transfer_count = transfer_count.clone();
async move {
transfer_count.fetch_add(1, Ordering::SeqCst);
Ok(())
Ok(DataMovementCopy::Committed)
}
}
};
@@ -332,7 +332,7 @@ async fn test_migrate_entry_version_remote_not_found_is_cleanup_ignored() {
let transfer_count = transfer_count.clone();
async move {
transfer_count.fetch_add(1, Ordering::SeqCst);
Ok(())
Ok(DataMovementCopy::Committed)
}
}
};
@@ -372,7 +372,7 @@ async fn test_migrate_entry_version_remote_overwrite_is_not_ignored() {
))),
);
let version = version_remote();
let mut transfer = |_, _, _| async move { Ok(()) };
let mut transfer = |_, _, _| async move { Ok(DataMovementCopy::Committed) };
let result = migrate_entry_version(
&backend,
@@ -408,7 +408,7 @@ async fn test_migrate_entry_version_remote_failure_is_reported() {
let transfer_count = transfer_count.clone();
async move {
transfer_count.fetch_add(1, Ordering::SeqCst);
Ok(())
Ok(DataMovementCopy::Committed)
}
}
};
@@ -441,7 +441,7 @@ async fn test_migrate_entry_version_remote_failure_is_reported() {
async fn test_migrate_entry_version_deleted_version_routes_delete_through_store_and_moved() {
let backend = MigrationBackendSpy::new(None, None);
let version = version_deleted();
let mut transfer = |_, _, _| async move { Ok(()) };
let mut transfer = |_, _, _| async move { Ok(DataMovementCopy::Committed) };
// The delete marker must be routed through the store closure (cross-pool routing), never
// through the source SetDisks. Assert the closure is invoked and the source set is not.
let delete_calls = Arc::new(AtomicUsize::new(0));
@@ -483,7 +483,7 @@ async fn test_migrate_entry_version_deleted_version_routes_delete_through_store_
async fn test_migrate_entry_version_deleted_version_not_found_is_ignored() {
let backend = MigrationBackendSpy::new(None, None);
let version = version_deleted();
let mut transfer = |_, _, _| async move { Ok(()) };
let mut transfer = |_, _, _| async move { Ok(DataMovementCopy::Committed) };
let delete_calls = Arc::new(AtomicUsize::new(0));
let mut delete_marker = {
let delete_calls = delete_calls.clone();
@@ -522,7 +522,7 @@ async fn test_migrate_entry_version_deleted_version_not_found_is_ignored() {
async fn test_migrate_entry_version_deleted_version_overwrite_is_not_ignored() {
let backend = MigrationBackendSpy::new(None, None);
let version = version_deleted();
let mut transfer = |_, _, _| async move { Ok(()) };
let mut transfer = |_, _, _| async move { Ok(DataMovementCopy::Committed) };
let delete_calls = Arc::new(AtomicUsize::new(0));
let mut delete_marker = {
let delete_calls = delete_calls.clone();
@@ -567,7 +567,7 @@ async fn test_migrate_entry_version_reader_not_found_is_ignored() {
let backend =
MigrationBackendSpy::new(Some(Err(Error::ObjectNotFound("bucket".to_string(), "object.bin".to_string()))), None);
let version = version_normal();
let mut transfer = |_, _, _| async move { Ok(()) };
let mut transfer = |_, _, _| async move { Ok(DataMovementCopy::Committed) };
let result = migrate_entry_version(
&backend,
@@ -602,7 +602,7 @@ async fn test_migrate_entry_version_reader_retries_before_success() {
let transfer_count = transfer_count.clone();
async move {
transfer_count.fetch_add(1, Ordering::SeqCst);
Ok(())
Ok(DataMovementCopy::Committed)
}
}
};
@@ -691,7 +691,7 @@ async fn test_migrate_entry_version_reader_fails_after_retries() {
let transfer_count = transfer_count.clone();
async move {
transfer_count.fetch_add(1, Ordering::SeqCst);
Ok(())
Ok(DataMovementCopy::Committed)
}
}
};
@@ -730,7 +730,7 @@ async fn test_migrate_entry_version_zero_max_attempts_still_attempts_once() {
let transfer_count = transfer_count.clone();
async move {
transfer_count.fetch_add(1, Ordering::SeqCst);
Ok(())
Ok(DataMovementCopy::Committed)
}
}
};
@@ -773,7 +773,7 @@ async fn test_migrate_entry_version_transfer_retries_before_success() {
if attempt == 0 {
return Err(Error::SlowDown);
}
Ok(())
Ok(DataMovementCopy::Committed)
}
}
};
@@ -1002,7 +1002,7 @@ async fn test_migrate_entry_version_ignores_data_usage_cache_when_enabled() {
let transfer_count = transfer_count.clone();
async move {
transfer_count.fetch_add(1, Ordering::SeqCst);
Ok(())
Ok(DataMovementCopy::Committed)
}
}
};
@@ -1045,7 +1045,7 @@ async fn test_migrate_entry_version_data_usage_cache_moves_when_ignore_disabled(
let transfer_count = transfer_count.clone();
async move {
transfer_count.fetch_add(1, Ordering::SeqCst);
Ok(())
Ok(DataMovementCopy::Committed)
}
}
};
@@ -2186,7 +2186,7 @@ async fn test_rebalance_stage_wrapped_lock_timeout_retries_real_migration_loop()
let attempt = attempts.fetch_add(1, Ordering::SeqCst);
async move {
if succeeds_on_retry && attempt > 0 {
return Ok(());
return Ok(DataMovementCopy::Committed);
}
Err(data_movement::data_movement_stage_error_for_test(
"rebalance_object",
@@ -2455,7 +2455,7 @@ async fn test_migrate_entry_version_transfer_failure_reports_write_target_stage(
#[tokio::test]
async fn test_migrate_entry_version_reader_failure_reports_read_source_stage() {
let backend = AlwaysFailGetBackend::new();
let mut transfer = |_, _, _| async { Ok(()) };
let mut transfer = |_, _, _| async { Ok(DataMovementCopy::Committed) };
let version = version_normal();
let result = migrate_entry_version(
crates/ecstore/src/set_disk/ops/object.rs
+178
View File
@@ -10537,6 +10537,17 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
..Default::default()
};
delete_request.set_tier_free_version_id(&Uuid::new_v4().to_string());
// A moved transitioned version's remote object now
// belongs to its target copy; any other removed
// transitioned version keeps its free version.
if opts.data_movement
&& opts
.data_movement_transferred_tier_versions
.as_ref()
.is_some_and(|transferred| transferred.contains(version))
{
delete_request.set_skip_tier_free_version();
}
begin_scanner_publication_delete_mutation(scanner_publication_commit_scope.as_ref())?;
self.delete_object_version_with_purge(
bucket,
@@ -19223,6 +19234,7 @@ mod transition_upload_integrity_tests {
object,
&expected,
&[],
&[],
crate::data_movement::SourceCleanupBucketFence::default(),
"test_data_movement",
)
@@ -19293,6 +19305,7 @@ mod transition_upload_integrity_tests {
object,
&expected,
&[],
&[],
crate::data_movement::SourceCleanupBucketFence {
expected_incarnation_id: None,
lifecycle_guard: Some(&bucket_guard),
@@ -19324,6 +19337,171 @@ mod transition_upload_integrity_tests {
drop(distributed_setup);
}
/// Data-movement source cleanup appends no tier free version for a
/// transitioned version the mover copied to a target, and keeps the free
/// version, the only remote-delete owner, for one it did not copy (such as
/// a version lifecycle expired during the move).
#[tokio::test]
#[serial_test::serial]
async fn data_movement_cleanup_keeps_a_free_version_only_for_untransferred_tier_versions() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let object = "object.bin";
let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
for (bucket, transferred) in [
("data-movement-cleanup-transferred", true),
("data-movement-cleanup-expired", false),
] {
let payload = b"a transitioned source removed by data movement".repeat(64);
let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await;
set_disks
.transition_object(bucket, object, &transition_options(&original, tier_name.clone()))
.await
.expect("transition should commit");
let expected = set_disks
.load_file_info_versions_exact(bucket, object)
.await
.expect("source versions should be readable")
.expect("source versions should exist");
let source = expected.versions[0].clone();
assert_eq!(source.transition_status, TRANSITION_COMPLETE);
let transferred_tier_versions = if transferred {
vec![crate::data_movement::source_cleanup_version_identity(&source)]
} else {
Vec::new()
};
crate::data_movement::cleanup_source_entry_if_unchanged(
Arc::clone(&set_disks),
bucket,
object,
&expected,
&[],
&transferred_tier_versions,
crate::data_movement::SourceCleanupBucketFence::default(),
"test_data_movement",
)
.await
.expect("source cleanup should succeed");
let left = set_disks
.load_file_info_versions_exact(bucket, object)
.await
.expect("source versions should be readable")
.map(|versions| versions.versions)
.unwrap_or_default();
if transferred {
assert!(left.is_empty(), "a transferred version must leave no free version: {left:?}");
} else {
assert_eq!(left.len(), 1, "an untransferred version must keep its free version: {left:?}");
let free = &left[0];
assert!(free.tier_free_version());
assert_eq!(free.transition_tier, source.transition_tier);
assert_eq!(free.transitioned_objname, source.transitioned_objname);
assert_eq!(free.transition_version_id, source.transition_version_id);
}
assert!(
backend.contains(&source.transitioned_objname).await,
"source cleanup never deletes the remote object itself"
);
}
assert_eq!(backend.remove_count().await, 0);
assert_eq!(backend.exact_remove_count(), 0);
}
/// The skip is decided per version: of two transitioned versions of one
/// key, only the one reported as transferred leaves no free version. The
/// directory object is listed as `dir/`, while the cleanup deletes its
/// encoded key, so the match must not depend on the version's name.
#[tokio::test]
#[serial_test::serial]
async fn data_movement_cleanup_skips_the_free_version_per_transferred_version() {
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
for (bucket, object) in [
("data-movement-cleanup-per-version", "object.bin"),
("data-movement-cleanup-directory", "dir/"),
] {
for disk in &disk_stores {
disk.make_volume(bucket).await.expect("bucket volume should be created");
}
let disk_object = rustfs_utils::path::encode_dir_object(object);
let mut version_ids = Vec::new();
for body in [
b"first transitioned version".as_slice(),
b"second transitioned version".as_slice(),
] {
let original = set_disks
.put_object(
bucket,
&disk_object,
&mut PutObjReader::from_vec(body.repeat(64)),
&ObjectOptions {
versioned: true,
..Default::default()
},
)
.await
.expect("source version should be written");
let mut opts = transition_options(&original, tier_name.clone());
opts.versioned = true;
set_disks
.transition_object(bucket, &disk_object, &opts)
.await
.expect("transition should commit");
version_ids.push(original.version_id);
}
let expected = set_disks
.load_file_info_versions_exact(bucket, object)
.await
.expect("source versions should be readable")
.expect("source versions should exist");
assert_eq!(expected.versions.len(), 2);
let version = |version_id| {
expected
.versions
.iter()
.find(|version| version.version_id == version_id)
.cloned()
.expect("each written version should be loaded")
};
let (moved, kept) = (version(version_ids[0]), version(version_ids[1]));
assert_eq!(moved.transition_status, TRANSITION_COMPLETE);
assert_eq!(kept.transition_status, TRANSITION_COMPLETE);
assert_ne!(moved.transitioned_objname, kept.transitioned_objname);
crate::data_movement::cleanup_source_entry_if_unchanged(
Arc::clone(&set_disks),
bucket,
object,
&expected,
&[],
&[crate::data_movement::source_cleanup_version_identity(&moved)],
crate::data_movement::SourceCleanupBucketFence::default(),
"test_data_movement",
)
.await
.expect("source cleanup should succeed");
let left = set_disks
.load_file_info_versions_exact(bucket, object)
.await
.expect("source versions should be readable")
.map(|versions| versions.versions)
.unwrap_or_default();
assert_eq!(left.len(), 1, "only the untransferred version keeps a free version: {left:?}");
let free = &left[0];
assert!(free.tier_free_version());
assert_eq!(free.transitioned_objname, kept.transitioned_objname);
assert_eq!(free.transition_version_id, kept.transition_version_id);
for remote in [&moved.transitioned_objname, &kept.transitioned_objname] {
assert!(backend.contains(remote).await, "source cleanup never deletes the remote object itself");
}
}
assert_eq!(backend.remove_count().await, 0);
}
#[tokio::test]
#[serial_test::serial]
async fn partial_remote_acceptance_cleans_exact_candidate_and_preserves_source() {
crates/ecstore/src/store/init.rs
+2
View File
@@ -9168,6 +9168,7 @@ mod tests {
&cleanup_object,
&expected_source_versions,
&[],
&[],
crate::data_movement::SourceCleanupBucketFence {
object_mutation_fence: Some(&mutation_fence),
..Default::default()
@@ -9288,6 +9289,7 @@ mod tests {
object,
&expected_source_versions,
&[],
&[],
crate::data_movement::SourceCleanupBucketFence {
object_mutation_fence: Some(&mutation_fence),
..Default::default()
docs/architecture/decommission-compatibility.md
+5 -1
View File
@@ -166,7 +166,7 @@ A tier free version is an internal xl.meta record (`rustfs_filemeta::FREE_VERSIO
### Lifecycle And Consumers
Creation: a local delete that removes a version whose transition status is `complete` normally appends the record via `MetaObject::delete_version` → `init_free_version` (skipped only when `skip_tier_free_version` is set, as on data-movement copies). User-facing single and batch deletes always retain that historical owner when they actually remove a transitioned source; they do not create a tier journal, probe a fleet capability, or issue a peer mutation RPC. `TransitionVersionState::Unknown` and incomplete destination identities stay on the same conservative free-version path. Delete-marker creation on an Enabled bucket is unchanged and does not schedule remote deletion.
Creation: a local delete that removes a version whose transition status is `complete` normally appends the record via `MetaObject::delete_version` → `init_free_version` (skipped only when `skip_tier_free_version` is set: on data-movement copies, and on the data-movement source cleanup of a version whose copy committed on a target pool). User-facing single and batch deletes always retain that historical owner when they actually remove a transitioned source; they do not create a tier journal, probe a fleet capability, or issue a peer mutation RPC. `TransitionVersionState::Unknown` and incomplete destination identities stay on the same conservative free-version path. Delete-marker creation on an Enabled bucket is unchanged and does not schedule remote deletion.
Recursive prefix/delete-all cannot preserve per-object markers across its physical directory purge, so it requires a v6 recoverable journal for every transitioned visible source plus a durable dispatch manifest for the whole operation. It fails closed before mutation on legacy metadata or on any existing hidden tier free-version under the prefix. Its internal streaming walk discovers logical keys, then exact-loads every key from its authoritative set in every pool, including free versions; the S3 listing merge is never treated as a complete physical-owner inventory. Tier-operation leases stay held from that preflight through journal prepare and physical deletion. Once physical deletion starts, any error is mutation-ambiguous: authorized/dispatched journals remain for recovery to commit owners only after all physical sets prove both the source and the exact free-version identity absent; uncertain owners are retained. If a retry discovers a later transitioned source after the manifest reached `DispatchAuthorized`, it replays only the manifest's immutable predecessor set, completes that operation, and leaves the newcomer for a successor dispatch. Operators may retry after the legacy free-version worker has durably completed remote and local cleanup. Journal-less internal deletes and older nodes keep their established marker behavior.
@@ -178,6 +178,10 @@ The exact decommission inventory loader (`load_file_info_versions_exact` via `ge
The source record is physically removed only after the target write quorum has committed and the source cleanup preflight still matches the exact inventory. If the lifecycle worker has already completed the remote delete and removed the source record before decommission acquires the source lock, decommission records that identity as already consumed and treats the missing source record as safe. If target capacity, metadata validation, lock fencing, or quorum fails, the source record remains and the entry records `state = "free_version_retained"` with reason `tier_free_version_migration_failed`; the worker retries the operation on a later pass. A target record with the same version id is accepted only when its free-version identity matches; a conflicting ordinary version or different free record is an overwrite error. This makes retries idempotent and prevents a free record from replacing a user-visible version.
A transitioned version itself moves as tier metadata, or as a data copy that keeps its transition tuple when it is restored, so once the target commits, the target copy is the live reference to the remote object. Decommission and rebalance pass the identity of every transitioned version whose copy committed, or found an equivalent target with the same tuple (`DataMovementCopy::Committed`), to `cleanup_source_entry_if_unchanged`, and the source cleanup deletes exactly those versions with `skip_tier_free_version`: the source gets no free version, and the remote delete belongs to whatever later removes the target copy. MinIO reaches the same state because its source cleanup is a physical prefix delete that never creates a free version. A transitioned version that did not move is still deleted with a new free version, because that record is its only remote-delete owner: one that lifecycle expired during the move whose queued expiry has not run yet, and a null version that a newer unversioned object on the target superseded (`DataMovementCopy::Superseded`), since the target then holds neither a copy nor a reference to the remote object. That free version stays on the source, so the decommission post-check still fails unless the free-version worker consumes it first, as it did before the skip existed. Without the skip every moved transitioned version left a fresh free version on the source, so the decommission post-check failed and the record stayed blocked behind the target's live reference (rustfs/backlog#2873, defect D).
Free versions that earlier builds left on a source this way are not removed. Their decommission ends failed at the post-check; once the operator clears the failed entry (`POST /v3/pools/clear`) and starts it again, the migration loop moves those records to the target pool like any other free version. There they stay blocked behind the live target version until it is deleted, and then both records are consumed by one remote delete. During a rolling upgrade, a source cleanup that an older node executes still creates them.
`TransitionVersionState::Unknown` records are migrated unchanged rather than discarded; the lifecycle worker retains them if remote identity validation cannot make a delete request. Only an authorized recursive prefix/delete-all v6 transaction may use a per-source journal as the sole retry source; ordinary single/batch deletes never take that path, and a journal discovered alongside an older or fallback free-version never authorizes dropping the xl.meta record.
### Remote-Tuple Publication Fence
docs/architecture/ilm-tiering-persistence-contracts.md
+2 -1
View File
@@ -19,7 +19,7 @@ These are approved-target invariants. A protocol's explicitly labeled current ex
2. A remote candidate may be deleted only when authoritative metadata proves that no live local owner references it, or when a durable time/fence/owner proof shows that its creator can no longer publish that candidate.
3. The approved transition-commit contract preserves and rechecks source version ID, data directory, modification time, size, and ETag. The current recovery predicate is narrower, as documented below. Bucket incarnation, Object Lock, and replication admission remain independent gates.
4. Free-version cleanup preserves the bucket-lifecycle fence, exact tier generation lease, physical-set locks, authoritative all-pool scan, and before/after fence checks.
5. At most one live protocol owns remote DELETE. Ownership moves from transition transaction to committed `xl.meta`, then to an `xl.meta` free-version on ordinary deletion. Only an authorized v6 journal/manifest operation may replace the free-version owner for a destructive prefix operation.
5. At most one live protocol owns remote DELETE. Ownership moves from transition transaction to committed `xl.meta`, then to an `xl.meta` free-version on ordinary deletion. Once a data-movement copy of a transitioned version commits on a target pool, the target `xl.meta` owns the remote object and the source deletion appends no free-version; a source version that did not move keeps its free-version. Only an authorized v6 journal/manifest operation may replace the free-version owner for a destructive prefix operation.
6. A successful metadata write response is not the only durability proof. Every response-lost or quorum-uncertain write must converge through a strong read of the same key and exact identity/generation; an absent, conflicting, or unreadable readback remains unknown.
7. Unknown namespaces, schemas, states, illegal state edges, checksum failures, identity conflicts, truncated scans without a continuation marker, and mixed-version ambiguity fail closed. They are never cleanup authorization. Current transition path validation, manual-job uppercase alias handling, and transition/manual page-token exceptions are documented below as gaps against this target.
@@ -31,6 +31,7 @@ These are approved-target invariants. A protocol's explicitly labeled current ex
| Local transition commit is complete | Exact transitioned version in `xl.meta` | No | Recovery finds the transaction's logical bucket/object/version and requires the complete recorded source identity (version ID, data directory, modification time, size, and ETag), `TRANSITION_COMPLETE`, and the same remote object, tier, and remote version before removing only the transaction record |
| An ordinary delete removes that transitioned version | Hidden `xl.meta` free-version | Yes | Metadata quorum atomically removes the visible version and preserves its exact tier tuple in the free-version |
| PUT or materialized self-copy replaces a transitioned null version | Hidden `xl.meta` free-version | Yes, after replacement commit and complete physical reference checks | The coordinator supplies one cleanup UUID to every disk; replacement metadata and the old remote tuple are written in the same `xl.meta` commit |
| Decommission or rebalance copies a transitioned version to a target pool | Target `xl.meta` version with the same tier tuple | No | The copy committed (or an equivalent target was found) with the exact transition tuple; the source cleanup then deletes that version with `skip_tier_free_version`. A source superseded by a newer unversioned target object, or expired by lifecycle during the move, was not copied and is deleted with a free-version |
| A recursive prefix/delete-all operation cannot preserve per-object markers | v6 journal bound to an immutable single dispatch manifest or a chunk-parent-bound child manifest | Yes, but only after child/manifest completion and all-pool absence proof | `DispatchAuthorized`, exact local destructive mutation, every journal `Committed`, then child/manifest `Completed`; a chunk parent advances only after that child completion |
| Tier configuration mutation, manual job, or decommission receipt | Intent/admission/copy proof only | No | These records gate configuration, scheduling, or migration; they never become remote-object cleanup owners |