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

fix(decommission): finish multipart null versions superseded on target (#8774)

Browse Source 
* fix(decommission): finish multipart null versions superseded on target

A decommission entry for an unversioned multipart source whose key
already holds a newer null version on the target settles as superseded
and aborts its staged upload. The cleanup's target proof saw the newer
object, could not match it to the staged identity, and failed closed
with a capacity-blocked error, so the routine paused, the source was
never cleaned up, and the error read like a minimum-free-drive block.

The staged upload is already proven absent and the target lookup is
exact: a different version id, or a null target with another
modification time (one null slot per key), is a different object, so
the staged copy holds nothing on the target and the pending intent is
cleared. Only a versioned target with the staged version id but another
modification time stays fail-closed. The remaining identity-proof
failures use a typed error without capacity wording.

Add three-pool decommission regressions for restored-transitioned and
plain multipart null sources superseded on the target.

* test(decommission): pin multipart target proof failures as terminal

Document the accepted residual of the null-slot proof: a metadata-only
self-copy keeps a published but uncharged staged copy's data while
moving its modification time, so its marker is released and one
object's bytes stay uncharged.

Add a three-pool routine test where the target already holds an
unowned null object with the source's exact identity: the cleanup's
ownership-proof failure fails the decommission with no capacity-blocked
reason instead of pausing it.
This commit is contained in:
Chris
2026-10-22 09:22:24 +09:00
committed by GitHub
parent 831029e5b6
commit ffd94cfe1b
2 changed files with 374 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/core/pools_test.rs
+237 -1
View File
@@ -764,7 +764,11 @@ mod decommission_lock_order_tests {
}
/// A two-part source object: a 5 MiB part and a 1-byte part.
async fn put_source_multipart_object(store: &Arc<crate::store::ECStore>, bucket: &str, object: &str) {
async fn put_source_multipart_object(
store: &Arc<crate::store::ECStore>,
bucket: &str,
object: &str,
) -> crate::object_api::ObjectInfo {
let source_upload = new_multipart_upload(store, 0, bucket, object, ObjectOptions::default())
.await
.expect("create the source multipart upload");
@@ -794,6 +798,7 @@ mod decommission_lock_order_tests {
.await
.expect("complete the source multipart object");
assert!(source_info.is_multipart());
source_info
}
async fn put_source_object(store: &Arc<crate::store::ECStore>, bucket: &str, object: &str, body: Vec<u8>) {
@@ -2033,6 +2038,237 @@ mod decommission_lock_order_tests {
run_large_stack_async_test("decommission-superseded-restored-null-version", run_superseded_restored_null_version_case);
}
/// Every capacity marker of the decommission's target reservations is
/// settled: no pending intent and no temporary mutation remain.
async fn assert_decommission_capacity_settled(store: &Arc<crate::store::ECStore>) {
let pool_meta = store.pool_meta.read().await;
let info = pool_meta.pools[0]
.decommission
.as_ref()
.expect("the decommission state should remain present");
assert!(
info.capacity_blocked_reason.is_none(),
"the decommission must not pause: {:?}",
info.capacity_blocked_reason
);
let reservation = info
.capacity_reservation
.as_ref()
.expect("the capacity reservation should remain present");
assert_eq!(reservation.pending_target_physical_bytes, 0);
assert_eq!(reservation.inflight_target_physical_bytes, 0);
for target in &reservation.targets {
assert!(
target.pending_mutation_id.is_none(),
"target {} keeps a pending intent",
target.pool_index
);
assert_eq!(target.pending_physical_bytes, 0);
assert_eq!(target.inflight_physical_bytes, 0);
assert!(
target.temporary_mutations.is_empty(),
"target {} keeps temporary mutations: {:?}",
target.pool_index,
target.temporary_mutations
);
}
}
/// A newer unversioned object written to the target replaces a multipart
/// null source. The multipart copy settles as superseded and aborts its
/// staged upload; that cleanup must accept the newer target as proof the
/// staged copy holds nothing there, instead of failing closed as a
/// capacity block that never lets the entry finish (rustfs/backlog#2873).
async fn run_superseded_multipart_null_version_case(tiered: bool) {
let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await;
let bucket = test_bucket("superseded-multipart");
let object = "multipart.bin";
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create the unversioned bucket");
let written = put_source_multipart_object(&store, &bucket, object).await;
let backend = if tiered {
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;
store.pools[0]
.transition_object(
&bucket,
object,
&ObjectOptions {
transition: crate::bucket::lifecycle::lifecycle::TransitionOptions {
status: TRANSITION_PENDING.to_string(),
tier: tier_name,
etag: written.etag.clone().expect("the multipart source should have an ETag"),
..Default::default()
},
mod_time: written.mod_time,
..Default::default()
},
)
.await
.expect("transition the multipart source before the decommission");
restore_source_version(&store, &bucket, object, false, &written).await;
Some(backend)
} else {
None
};
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_eq!(source.parts.len(), 2, "the source must move through the multipart copy");
assert!(!source.is_remote(), "the 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 multipart entry should finish");
let left = exact_versions(&store, 0, &bucket, object).await;
if let Some(backend) = backend.as_ref() {
// The newer target references no remote object, so the source
// cleanup keeps the tier free version that owns the remote delete.
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);
assert!(backend.contains(&source.transitioned_objname).await);
assert_eq!(backend.remove_count().await, 0);
} else {
assert!(left.is_empty(), "the superseded source must be cleaned up: {left:?}");
}
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].tier_free_version());
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 target_set = store.pools[1].get_disks_by_key(object);
let incarnation = store
.bucket_incarnation_id(&bucket)
.await
.expect("load the bucket incarnation");
let staged_identity = format!(
"v1:none:{}",
source
.mod_time
.expect("the source has a modification time")
.unix_timestamp_nanos()
);
assert!(
target_set
.data_movement_multipart_upload_ids(&bucket, object, Some(incarnation), &staged_identity)
.await
.expect("scan the target for the staged upload")
.is_empty(),
"the staged upload must be aborted"
);
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_decommission_capacity_settled(&store).await;
}
#[test]
#[serial_test::serial]
fn decommission_finishes_a_restored_multipart_null_version_superseded_on_target() {
run_large_stack_async_test("decommission-superseded-restored-multipart", || {
run_superseded_multipart_null_version_case(true)
});
}
/// The target already holds a null object with the multipart source's
/// exact identity but without the data-movement ownership proof. The copy
/// cannot publish over it, and its cleanup cannot tell whether that object
/// is the staged copy. That identity-proof failure must fail the
/// decommission with an accurate error, not pause it as a capacity block
/// that only a restart could leave (rustfs/backlog#2873).
#[test]
#[serial_test::serial]
fn decommission_fails_instead_of_pausing_on_an_unowned_exact_multipart_target() {
run_large_stack_async_test("decommission-unowned-exact-multipart-target", || async {
let (_temp_dirs, store, _other_store) = test_three_pool_stores_with_isolated_node_contexts(None).await;
let bucket = test_bucket("unowned-exact-multipart");
let object = "multipart.bin";
store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.expect("create the unversioned bucket");
let written = put_source_multipart_object(&store, &bucket, object).await;
let unowned = b"same identity, written without a data movement".to_vec();
store.pools[1]
.put_object(
&bucket,
object,
&mut PutObjReader::from_vec(unowned.clone()),
&ObjectOptions {
mod_time: written.mod_time,
..Default::default()
},
)
.await
.expect("write the unowned exact-identity object on the target");
let target = exact_versions(&store, 1, &bucket, object).await;
assert_eq!(target.len(), 1);
assert!(target[0].version_id.is_none());
assert_eq!(target[0].mod_time, written.mod_time, "the target must carry the staged identity");
start_transitioned_entry_decommission(&store, &bucket).await;
let _ = spawn_decommission_routine(&store, 1)
.await
.expect("the decommission routine task should join");
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 && !info.complete && info.capacity_blocked_reason.is_none(),
"an identity-proof failure must fail the decommission, not pause it as a capacity block: complete={}, failed={}, blocked_reason={:?}",
info.complete,
info.failed,
info.capacity_blocked_reason
);
drop(local);
assert_eq!(exact_versions(&store, 0, &bucket, object).await.len(), 1, "the source must stay in place");
let mut reader = store.pools[1]
.get_object_reader(&bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
.await
.expect("the unowned target object should stay readable");
let mut body = Vec::new();
reader
.stream
.read_to_end(&mut body)
.await
.expect("drain the unowned target object");
assert_eq!(body, unowned);
});
}
#[test]
#[serial_test::serial]
fn decommission_finishes_a_multipart_null_version_superseded_on_target() {
run_large_stack_async_test("decommission-superseded-multipart", || run_superseded_multipart_null_version_case(false));
}
struct ExpiredAndMovedVersions {
_temp_dirs: Vec<tempfile::TempDir>,
store: Arc<crate::store::ECStore>,
crates/ecstore/src/store/multipart.rs
+137 -26
View File
@@ -68,33 +68,74 @@ async fn pause_orphan_multipart_cleanup(bucket: &str) {
}
}
/// Why a decommission multipart cleanup cannot decide whether its staged
/// identity reached the target pool. These are identity proofs, not free-space
/// decisions, so they stay out of the capacity-blocked error and its minimum
/// free drive wording.
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub(crate) enum DecommissionMultipartTargetProofError {
#[error("decommission multipart cleanup cannot prove exact target absence without a modification time")]
MissingModTime,
#[error("decommission multipart cleanup exact target version id is invalid")]
InvalidVersionId,
#[error("decommission multipart cleanup found the staged version id with a different modification time")]
AmbiguousVersion,
#[error("decommission multipart cleanup found the exact staged target without its ownership proof")]
UnownedExactTarget,
}
impl From<DecommissionMultipartTargetProofError> for Error {
fn from(err: DecommissionMultipartTargetProofError) -> Self {
Error::other(err)
}
}
/// Decide whether the cleanup of a staged decommission upload may clear the
/// mutation's pending capacity intent, from the exact-version lookup of the
/// staged identity on the target pool.
///
/// The staged upload itself is already proven absent. `true` means its
/// identity is not published on the target either: no target, or a target that
/// is a different object. A different version id is a different object. A null
/// version has one slot per key, so a null target with another modification
/// time replaced whatever was there: a newer unversioned write that superseded
/// the source, or an object the staged copy never reached. In those cases the
/// staged copy holds no bytes on the target. `false` means the exact staged
/// identity is published and owned, so its pending intent stays for the commit
/// to charge. A versioned target with the staged version id but another
/// modification time cannot be told apart and fails closed.
///
/// Known residual: a metadata-only self-copy (CopyObject with REPLACE) keeps
/// an object's data and only moves its modification time. If the staged copy
/// was already published but not yet charged (an adopted completion, or a
/// pending intent left by an ambiguous commit) and is then rewritten that way,
/// this cleanup releases its marker and one object's bytes stay uncharged. That
/// is a bounded estimate error with no data loss; failing closed here instead
/// stalled the decommission for good, so no fence is added for it.
fn decommission_multipart_target_clear_pending(opts: &ObjectOptions, target: Option<&ObjectInfo>) -> Result<bool> {
let expected_mod_time = opts.mod_time.ok_or_else(|| Error::DecommissionCapacityBlocked {
message: "multipart cleanup cannot prove exact target absence without a modification time".to_string(),
})?;
let expected_mod_time = opts.mod_time.ok_or(DecommissionMultipartTargetProofError::MissingModTime)?;
let expected_version_id = opts
.version_id
.as_deref()
.map(Uuid::parse_str)
.transpose()
.map_err(|err| Error::DecommissionCapacityBlocked {
message: format!("multipart cleanup exact target version is invalid: {err}"),
})?
.map_err(|_| DecommissionMultipartTargetProofError::InvalidVersionId)?
.filter(|version_id| !version_id.is_nil());
let Some(target) = target else {
return Ok(true);
};
if target.version_id.filter(|version_id| !version_id.is_nil()) != expected_version_id
|| target.mod_time != Some(expected_mod_time)
{
return Err(Error::DecommissionCapacityBlocked {
message: "multipart cleanup found a target but cannot prove the exact staged identity is absent".to_string(),
});
let target_version_id = target.version_id.filter(|version_id| !version_id.is_nil());
if target_version_id != expected_version_id {
return Ok(true);
}
if target.mod_time != Some(expected_mod_time) {
if expected_version_id.is_none() {
return Ok(true);
}
return Err(DecommissionMultipartTargetProofError::AmbiguousVersion.into());
}
if !crate::data_movement::is_owned_data_movement_target(target) {
return Err(Error::DecommissionCapacityBlocked {
message: "multipart cleanup found an exact target without its ownership proof".to_string(),
});
return Err(DecommissionMultipartTargetProofError::UnownedExactTarget.into());
}
Ok(false)
}
@@ -1372,34 +1413,104 @@ mod tests {
.expect("an exact owned target should preserve pending capacity")
);
let assert_proof_error = |result: Result<bool>, expected: DecommissionMultipartTargetProofError| {
let err = result.expect_err("an unprovable target identity must fail closed");
assert!(
!matches!(err, Error::DecommissionCapacityBlocked { .. }),
"a target identity proof is not a capacity block: {err}"
);
assert!(!err.to_string().contains("minimum free drive"), "{err}");
let Error::Io(io_err) = &err else {
panic!("the target proof error must stay typed: {err:?}");
};
assert_eq!(
io_err
.get_ref()
.and_then(|inner| inner.downcast_ref::<DecommissionMultipartTargetProofError>()),
Some(&expected)
);
};
let missing_identity = ObjectOptions {
mod_time: None,
..opts.clone()
};
assert!(matches!(
assert_proof_error(
decommission_multipart_target_clear_pending(&missing_identity, None),
Err(Error::DecommissionCapacityBlocked { .. })
));
DecommissionMultipartTargetProofError::MissingModTime,
);
let mismatched_target = ObjectInfo {
let other_version_target = ObjectInfo {
version_id: Some(Uuid::new_v4()),
mod_time: Some(mod_time),
..Default::default()
};
assert!(matches!(
decommission_multipart_target_clear_pending(&opts, Some(&mismatched_target)),
Err(Error::DecommissionCapacityBlocked { .. })
));
assert!(
decommission_multipart_target_clear_pending(&opts, Some(&other_version_target))
.expect("another version is not the staged identity")
);
let rewritten_version_target = ObjectInfo {
version_id: Some(version_id),
mod_time: Some(mod_time + time::Duration::seconds(1)),
..Default::default()
};
assert_proof_error(
decommission_multipart_target_clear_pending(&opts, Some(&rewritten_version_target)),
DecommissionMultipartTargetProofError::AmbiguousVersion,
);
let unowned_target = ObjectInfo {
version_id: Some(version_id),
mod_time: Some(mod_time),
..Default::default()
};
assert!(matches!(
assert_proof_error(
decommission_multipart_target_clear_pending(&opts, Some(&unowned_target)),
Err(Error::DecommissionCapacityBlocked { .. })
));
DecommissionMultipartTargetProofError::UnownedExactTarget,
);
}
/// A null version has one slot per key: a null target with another
/// modification time is a different object, such as a newer unversioned
/// write that superseded the source, so the staged copy holds nothing on
/// the target (rustfs/backlog#2873).
#[test]
fn decommission_multipart_cleanup_clears_pending_for_a_replaced_null_target() {
let mod_time = time::OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(7);
for version_id in [None, Some(Uuid::nil().to_string())] {
let opts = ObjectOptions {
version_id: version_id.clone(),
mod_time: Some(mod_time),
..Default::default()
};
for target_version_id in [None, Some(Uuid::nil())] {
for target_mod_time in [mod_time + time::Duration::seconds(1), mod_time - time::Duration::seconds(1)] {
let replaced = ObjectInfo {
version_id: target_version_id,
mod_time: Some(target_mod_time),
..Default::default()
};
assert!(
decommission_multipart_target_clear_pending(&opts, Some(&replaced))
.expect("a replaced null target is not the staged identity"),
"version_id={version_id:?} target_version_id={target_version_id:?} target_mod_time={target_mod_time}"
);
}
}
let mut metadata = HashMap::new();
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_DATA_MOVED, "true".to_string());
rustfs_utils::http::insert_str(&mut metadata, rustfs_utils::http::SUFFIX_DATA_MOVED_TAGS, "v1:".to_string());
let staged = ObjectInfo {
mod_time: Some(mod_time),
user_defined: Arc::new(metadata),
..Default::default()
};
assert!(
!decommission_multipart_target_clear_pending(&opts, Some(&staged))
.expect("the exact owned null target keeps its pending intent")
);
}
}
/// Models a single pool's `list_multipart_uploads`: returns uploads strictly