diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 4f5d402d9..e1008a48e 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -1350,6 +1350,7 @@ fn should_cleanup_decommission_source_entry(decommissioned: usize, total_version } const DECOMMISSION_FREE_VERSION_MIGRATED_REASON: &str = "tier_free_version_migrated"; +const DECOMMISSION_FREE_VERSION_CONSUMED_REASON: &str = "tier_free_version_already_consumed"; const DECOMMISSION_FREE_VERSION_RETAINED_REASON: &str = "tier_free_version_migration_failed"; const DECOMMISSION_FREE_VERSION_SWEEP_REASON: &str = "tier_free_version_unresolved_after_decommission"; const DECOMMISSION_FREE_VERSION_DISPOSITION_REASON: &str = "tier_free_version_disposition_recorded"; @@ -1357,6 +1358,7 @@ const DECOMMISSION_FREE_VERSION_DISPOSITION_REASON: &str = "tier_free_version_di #[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] struct DecommissionFreeVersionDisposition { migrated: usize, + consumed: usize, retained: usize, } @@ -1365,12 +1367,32 @@ impl DecommissionFreeVersionDisposition { self.migrated += 1; } + fn record_consumed(&mut self) { + self.consumed += 1; + } + fn record_retained(&mut self) { self.retained += 1; } fn total(self) -> usize { - self.migrated.saturating_add(self.retained) + self.migrated.saturating_add(self.consumed).saturating_add(self.retained) + } +} + +enum DecommissionFreeVersionAttempt { + Migrated, + Consumed, + CapacityFailure(Error), + Retry(Error), +} + +fn classify_decommission_free_version_attempt(result: Result<()>) -> DecommissionFreeVersionAttempt { + match result { + Ok(()) => DecommissionFreeVersionAttempt::Migrated, + Err(err) if is_decommission_copy_cleanup_safe_error(&err) => DecommissionFreeVersionAttempt::Consumed, + Err(err) if is_decommission_target_capacity_error(&err) => DecommissionFreeVersionAttempt::CapacityFailure(err), + Err(err) => DecommissionFreeVersionAttempt::Retry(err), } } @@ -3867,39 +3889,49 @@ impl ECStore { let version_id = version.version_id.map(|v| v.to_string()); let mut migration_error = None; let mut migrated = false; + let mut consumed = false; + let mut capacity_failure = false; for _ in 0..3 { - match run_decommission_side_effect(&rx, &operation_gate, || async { - self.decommission_tiered_object( - bucket.as_str(), - &version.name, - version, - &decommission_remote_tiered_opts(version, version_id.clone(), idx, expected_bucket_incarnation_id), - ) - .await - }) - .await - { - Ok(()) => { + match classify_decommission_free_version_attempt( + run_decommission_side_effect(&rx, &operation_gate, || async { + self.decommission_tiered_object( + bucket.as_str(), + &version.name, + version, + &decommission_remote_tiered_opts( + version, + version_id.clone(), + idx, + expected_bucket_incarnation_id, + ), + ) + .await + }) + .await, + ) { + DecommissionFreeVersionAttempt::Migrated => { migrated = true; migration_error = None; break; } - Err(err) if is_decommission_target_capacity_error(&err) => { - return Err(with_decommission_entry_context( - "decommission_tier_free_version", - bucket.as_str(), - version.name.as_str(), - err, - )); + DecommissionFreeVersionAttempt::Consumed => { + consumed = true; + migration_error = None; + break; } - Err(err) => migration_error = Some(err), + DecommissionFreeVersionAttempt::CapacityFailure(err) => { + capacity_failure = true; + migration_error = Some(err); + break; + } + DecommissionFreeVersionAttempt::Retry(err) => migration_error = Some(err), } } { let mut pool_meta = self.pool_meta.write().await; ensure_decommission_generation(&pool_meta, idx, generation)?; - if let Err(err) = count_decommission_item(&mut pool_meta, idx, 0, !migrated) { + if let Err(err) = count_decommission_item(&mut pool_meta, idx, 0, !migrated && !consumed) { return Err(with_decommission_entry_context( "count_decommission_item", bucket.as_str(), @@ -3909,10 +3941,14 @@ impl ECStore { } } - if migrated { + if migrated || consumed { decommissioned += 1; cleanup_preflight_allowed_missing.push(data_movement::source_cleanup_version_identity(version)); + } + if migrated { free_version_disposition.record_migrated(); + } else if consumed { + free_version_disposition.record_consumed(); } else { free_version_disposition.record_retained(); } @@ -3928,14 +3964,31 @@ impl ECStore { result = ?migration_error, reason = if migrated { DECOMMISSION_FREE_VERSION_MIGRATED_REASON + } else if consumed { + DECOMMISSION_FREE_VERSION_CONSUMED_REASON } else { DECOMMISSION_FREE_VERSION_RETAINED_REASON }, - state = if migrated { "free_version_migrated" } else { "free_version_retained" }, + state = if migrated { + "free_version_migrated" + } else if consumed { + "free_version_consumed" + } else { + "free_version_retained" + }, "Decommission free-version disposition recorded" ); - if !migrated { + if capacity_failure { + return Err(with_decommission_entry_context( + "decommission_tier_free_version", + bucket.as_str(), + version.name.as_str(), + migration_error.expect("capacity failure must retain its error"), + )); + } + + if !migrated && !consumed { break; } continue; @@ -4249,6 +4302,7 @@ impl ECStore { bucket = %bucket, object = %entry.name, free_versions_migrated = free_version_disposition.migrated, + free_versions_consumed = free_version_disposition.consumed, free_versions_retained = free_version_disposition.retained, free_versions_total = free_version_disposition.total(), reason = DECOMMISSION_FREE_VERSION_DISPOSITION_REASON, @@ -5555,6 +5609,21 @@ mod tests { ))); } + #[test] + fn decommission_free_version_attempt_treats_missing_source_as_consumed() { + let attempt = + classify_decommission_free_version_attempt(Err(Error::ObjectNotFound("bucket".to_string(), "object".to_string()))); + + assert!(matches!(attempt, DecommissionFreeVersionAttempt::Consumed)); + } + + #[test] + fn decommission_free_version_attempt_preserves_capacity_failure() { + let attempt = classify_decommission_free_version_attempt(Err(Error::DiskFull)); + + assert!(matches!(attempt, DecommissionFreeVersionAttempt::CapacityFailure(Error::DiskFull))); + } + #[test] fn decommission_delete_marker_copy_error_rejects_data_movement_overwrite() { let err = Error::DataMovementOverwriteErr("bucket".to_string(), "object".to_string(), "version".to_string()); diff --git a/crates/ecstore/src/data_movement/mod.rs b/crates/ecstore/src/data_movement/mod.rs index 112e7fed0..07eb1b5b0 100644 --- a/crates/ecstore/src/data_movement/mod.rs +++ b/crates/ecstore/src/data_movement/mod.rs @@ -1956,8 +1956,7 @@ mod tests { let mut free_version = cleanup_test_file_info("object.txt", Uuid::from_u128(2), "tier-cleanup"); free_version.deleted = true; free_version.set_tier_free_version(); - let mut expected = cleanup_test_versions(vec![migrated.clone()]); - expected.free_versions = vec![free_version.clone()]; + let expected = cleanup_test_versions(vec![migrated.clone(), free_version.clone()]); let current = cleanup_test_versions(vec![migrated]); let allowed_missing = vec![source_cleanup_version_identity(&free_version)]; diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index daf3f8175..269d32b7d 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -249,6 +249,7 @@ async fn inspect_decommission_tier_free_version_target( { matching_count += 1; let existing = existing.into_fileinfo(bucket, object, true)?; + existing.validate_for_metadata_read()?; if !existing.tier_free_version() || !crate::store::tiered_data_movement_source_matches(source, &existing)? { all_matching_versions_equivalent = false; } diff --git a/docs/architecture/decommission-compatibility.md b/docs/architecture/decommission-compatibility.md index 35cb0caeb..1d6fa95ea 100644 --- a/docs/architecture/decommission-compatibility.md +++ b/docs/architecture/decommission-compatibility.md @@ -198,6 +198,9 @@ boundary. 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 @@ -218,13 +221,15 @@ migrated unchanged rather than discarded: the lifecycle worker retains them if remote identity validation cannot make a delete request. Each migrated record emits `state = "free_version_migrated"` with reason -`tier_free_version_migrated`. Each failed record emits the retained state and -failure reason above. The entry also emits a disposition summary with migrated, -retained, and total counts. The final decommission sweep uses the exact loader, -counts free records still present, and emits one retained record/reason for each -unresolved free version before failing the sweep. This makes both successful -migration and retained cleanup obligations visible instead of silently omitting -free records. +`tier_free_version_migrated`. A record consumed before migration emits +`state = "free_version_consumed"` with reason +`tier_free_version_already_consumed`. Each failed record emits the retained state +and failure reason above. The entry also emits a disposition summary with +migrated, consumed, retained, and total counts. The final decommission sweep uses +the exact loader, counts free records still present, and emits one retained +record/reason for each unresolved free version before failing the sweep. This +makes successful migration, completed cleanup, and retained cleanup obligations +visible instead of silently omitting free records. No new S3-visible version or admin response field is needed: free versions remain internal and are never counted as user-visible versions. The structured