diff --git a/crates/ecstore/src/core/pools_test.rs b/crates/ecstore/src/core/pools_test.rs index b8430c1b4..5418522cf 100644 --- a/crates/ecstore/src/core/pools_test.rs +++ b/crates/ecstore/src/core/pools_test.rs @@ -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, bucket: &str, object: &str) { + async fn put_source_multipart_object( + store: &Arc, + 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, bucket: &str, object: &str, body: Vec) { @@ -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) { + 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, store: Arc, diff --git a/crates/ecstore/src/store/multipart.rs b/crates/ecstore/src/store/multipart.rs index 0be6f918c..a039e36c7 100644 --- a/crates/ecstore/src/store/multipart.rs +++ b/crates/ecstore/src/store/multipart.rs @@ -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 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 { - 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, 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::()), + 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