From a6cb34c7a4fdd14cedae417f510cedbf9f5b072a Mon Sep 17 00:00:00 2001 From: cxymds Date: Thu, 3 Sep 2026 18:38:30 +0800 Subject: [PATCH] fix: fence transition transaction recovery (#7095) --- crates/ecstore/src/api/mod.rs | 1 + .../src/bucket/lifecycle/tier_sweeper.rs | 1 + .../lifecycle/transition_transaction.rs | 476 +++++++++++++++--- crates/ecstore/src/set_disk/mod.rs | 2 + crates/ecstore/src/set_disk/ops/object.rs | 276 +++++++--- crates/ecstore/src/store/init.rs | 353 +++++++++++-- .../tests/lifecycle_integration_test.rs | 37 +- crates/scanner/tests/storage_api/mod.rs | 7 +- 8 files changed, 984 insertions(+), 169 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 7f265c564..46f91ebd5 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -78,6 +78,7 @@ pub mod bucket { #[cfg(feature = "test-util")] pub use crate::bucket::lifecycle::transition_transaction::{ TransitionTransactionRecoveryStats, recover_transition_transaction_records, + recover_transition_transaction_records_at, }; } diff --git a/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs b/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs index bfbefe2ae..6518451ff 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs @@ -575,6 +575,7 @@ pub(crate) async fn delete_confirmed_transition_candidate_exact_with_lease_idemp #[cfg(test)] static CONFIRMED_TRANSITION_EMPTY_GUARD_DISPATCHES: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); +#[cfg(test)] pub(crate) async fn delete_confirmed_transition_candidate_exact_with_manager_and_identity( obj_name: &str, rv_id: &str, diff --git a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs index 8c1c1e22c..87df9fe0d 100644 --- a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs +++ b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs @@ -25,14 +25,17 @@ use crate::bucket::lifecycle::durable_namespace::TRANSITION_TRANSACTION_NAMESPAC use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE; use crate::bucket::lifecycle::tier_sweeper::{ delete_confirmed_transition_candidate_exact_with_lease_idempotent, - delete_confirmed_transition_candidate_exact_with_manager_and_identity, delete_object_from_remote_tier_idempotent_with_manager_and_identity, }; use crate::disk::RUSTFS_META_BUCKET; use crate::error::{Error, Result as EcstoreResult}; use crate::object_api::ObjectOptions; use crate::services::tier::{tier::TierConfigMgr, warm_backend::TransitionCandidateProbe}; -use crate::storage_api_contracts::{list::ListOperations as _, object::ObjectOperations as _}; +use crate::storage_api_contracts::{ + list::ListOperations as _, + namespace::NamespaceLocking as _, + object::{HTTPPreconditions, ObjectOperations as _}, +}; use crate::store::ECStore; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; @@ -110,7 +113,7 @@ pub struct TransitionRemoteVersion { impl TransitionRemoteVersion { pub fn known_from_put_response(version_id: impl Into) -> Self { let version_id = version_id.into(); - if version_id.is_empty() || Uuid::parse_str(&version_id).is_ok_and(|parsed| parsed.is_nil()) { + if version_id.is_empty() { Self::unversioned() } else { Self::versioned(version_id) @@ -307,10 +310,16 @@ impl TransitionTransaction { if self.write_id.is_nil() { return Err(TransitionTransactionError::Corrupt("write_id is nil")); } + if self.not_after_unix_nanos <= 0 { + return Err(TransitionTransactionError::Corrupt("ownership deadline is not positive")); + } self.source.validate()?; if self.tier_name.is_empty() { return Err(TransitionTransactionError::Corrupt("tier name is empty")); } + if self.backend_fingerprint == [0; 32] { + return Err(TransitionTransactionError::Corrupt("backend fingerprint is empty")); + } if self.remote_object != canonical_transition_remote_object(self.deployment_id, &self.source.bucket, self.transaction_id, self.write_id)? { @@ -476,6 +485,18 @@ impl TransitionTransaction { Ok(()) } + fn has_same_immutable_identity(&self, other: &Self) -> bool { + self.deployment_id == other.deployment_id + && self.transaction_id == other.transaction_id + && self.owner_epoch == other.owner_epoch + && self.write_id == other.write_id + && self.source == other.source + && self.tier_name == other.tier_name + && self.backend_fingerprint == other.backend_fingerprint + && self.remote_object == other.remote_object + && self.not_after_unix_nanos == other.not_after_unix_nanos + } + fn validate_cleanup_proof(&self, proof: &TransitionCleanupProof) -> Result<()> { if proof.transaction_id != self.transaction_id || proof.write_id != self.write_id @@ -585,20 +606,91 @@ pub(crate) async fn save_transition_transaction_record( let object = transition_transaction_record_object_name(transaction.transaction_id).map_err(transition_transaction_store_error)?; let data = transaction.encode().map_err(transition_transaction_store_error)?; - config_boundary::save_config(api.clone(), &object, data.clone()).await?; + config_boundary::save_config_with_opts( + api.clone(), + &object, + data.clone(), + &ObjectOptions { + max_parity: true, + http_preconditions: Some(HTTPPreconditions { + if_none_match: Some("*".to_string()), + ..Default::default() + }), + ..Default::default() + }, + ) + .await?; // Box::pin: the durable-receipt state machine is large and sits on the // already-deep transition worker poll chain; keeping it inline overflows // the default 2 MiB tokio worker stack in debug builds. Box::pin(api.record_durable_ilm_decommission_progress(&object, &data)).await } +pub(crate) async fn save_transition_transaction_record_if_current( + api: Arc, + expected: &TransitionTransaction, + next: &TransitionTransaction, +) -> EcstoreResult<()> { + let object = transition_transaction_record_object_name(next.transaction_id).map_err(transition_transaction_store_error)?; + let revision_is_next = expected.revision.checked_add(1) == Some(next.revision); + let state_is_next = state_change_allowed(expected.state, next.state) + || matches!( + (expected.state, next.state), + ( + TransitionTransactionState::Uploaded + | TransitionTransactionState::UploadOutcomeUnknown + | TransitionTransactionState::LocalCommitStarted, + TransitionTransactionState::CleanupPending + ) + ); + let remote_version_is_monotonic = expected.remote_version.is_unknown() || expected.remote_version == next.remote_version; + if !expected.has_same_immutable_identity(next) || !revision_is_next || !state_is_next || !remote_version_is_monotonic { + return Err(Error::PreconditionFailed); + } + let (current, etag) = load_transition_transaction_record_with_etag(api.clone(), expected.transaction_id).await?; + if ¤t != expected { + return Err(Error::PreconditionFailed); + } + let data = next.encode().map_err(transition_transaction_store_error)?; + config_boundary::save_config_with_opts( + api.clone(), + &object, + data.clone(), + &ObjectOptions { + max_parity: true, + http_preconditions: Some(HTTPPreconditions { + if_match: Some(etag), + ..Default::default() + }), + ..Default::default() + }, + ) + .await?; + // Box::pin: see save_transition_transaction_record. + Box::pin(api.record_durable_ilm_decommission_progress(&object, &data)).await +} + pub(crate) async fn load_transition_transaction_record( api: Arc, transaction_id: Uuid, ) -> EcstoreResult { + load_transition_transaction_record_with_etag(api, transaction_id) + .await + .map(|(transaction, _)| transaction) +} + +async fn load_transition_transaction_record_with_etag( + api: Arc, + transaction_id: Uuid, +) -> EcstoreResult<(TransitionTransaction, String)> { let object = transition_transaction_record_object_name(transaction_id).map_err(transition_transaction_store_error)?; - let data = config_boundary::read_config(api, &object).await?; - TransitionTransaction::decode(transaction_id, &data).map_err(transition_transaction_store_error) + let (data, object_info) = config_boundary::read_config_with_metadata(api, &object, &ObjectOptions::default()).await?; + let etag = object_info + .etag + .filter(|etag| !etag.trim().is_empty()) + .ok_or_else(|| Error::other("transition transaction record is missing an ETag"))?; + let transaction = TransitionTransaction::decode(transaction_id, &data).map_err(transition_transaction_store_error)?; + Ok((transaction, etag)) } pub(crate) async fn delete_transition_transaction_record( @@ -607,10 +699,18 @@ pub(crate) async fn delete_transition_transaction_record( ) -> EcstoreResult<()> { let object = transition_transaction_record_object_name(transaction.transaction_id).map_err(transition_transaction_store_error)?; - let data = transaction.encode().map_err(transition_transaction_store_error)?; + let (current, etag) = match load_transition_transaction_record_with_etag(api.clone(), transaction.transaction_id).await { + Ok(record) => record, + Err(Error::ConfigNotFound) => return Ok(()), + Err(err) => return Err(err), + }; + if ¤t != transaction { + return Err(Error::PreconditionFailed); + } + let data = current.encode().map_err(transition_transaction_store_error)?; // Box::pin: see save_transition_transaction_record. Box::pin(api.record_durable_ilm_decommission_terminal(&object, &data)).await?; - match config_boundary::delete_config(api, &object).await { + match config_boundary::delete_config_if_match(api, &object, &etag).await { Ok(()) | Err(Error::ConfigNotFound) => Ok(()), Err(err) => Err(err), } @@ -637,6 +737,84 @@ pub enum TransitionTransactionRecoveryOutcome { Retained, } +#[cfg(test)] +#[derive(Default)] +struct TransitionRecoveryClaimBarrierState { + transaction_id: Uuid, + arrived: tokio::sync::Notify, + release: tokio::sync::Notify, +} + +#[cfg(test)] +pub(crate) struct TransitionRecoveryClaimBarrier { + state: Arc, +} + +#[cfg(test)] +static TRANSITION_RECOVERY_CLAIM_BARRIER: std::sync::OnceLock< + std::sync::Mutex>>, +> = std::sync::OnceLock::new(); + +#[cfg(test)] +impl TransitionRecoveryClaimBarrier { + pub(crate) fn install(transaction_id: Uuid) -> Self { + let state = Arc::new(TransitionRecoveryClaimBarrierState { + transaction_id, + ..Default::default() + }); + let mut slot = TRANSITION_RECOVERY_CLAIM_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("transition recovery claim barrier mutex should not poison"); + assert!( + slot.is_none(), + "transition recovery claim barrier must be installed by one test at a time" + ); + *slot = Some(Arc::clone(&state)); + drop(slot); + Self { state } + } + + pub(crate) async fn wait_until_paused(&self) { + tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified()) + .await + .expect("transition recovery should reach the cleanup claim CAS"); + } + + pub(crate) fn release(&self) { + self.state.release.notify_one(); + } +} + +#[cfg(test)] +impl Drop for TransitionRecoveryClaimBarrier { + fn drop(&mut self) { + self.state.release.notify_one(); + let mut slot = TRANSITION_RECOVERY_CLAIM_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("transition recovery claim barrier mutex should not poison"); + if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) { + *slot = None; + } + } +} + +#[cfg(test)] +async fn pause_before_transition_recovery_claim(transaction_id: Uuid) { + let barrier = TRANSITION_RECOVERY_CLAIM_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("transition recovery claim barrier mutex should not poison") + .as_ref() + .filter(|barrier| barrier.transaction_id == transaction_id) + .cloned(); + if let Some(barrier) = barrier { + barrier.arrived.notify_one(); + barrier.release.notified().await; + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] pub enum TransitionOperatorProbe { @@ -858,49 +1036,129 @@ pub async fn process_transition_transaction_record( transaction: &TransitionTransaction, ) -> EcstoreResult { transaction.validate().map_err(transition_transaction_store_error)?; - match transaction.state { + // Box the expanded recovery state machine so callers on Tokio's default + // worker stack do not inline its full future into an already-deep scan. + Box::pin(process_transition_transaction_record_at( + api, + transaction, + time::OffsetDateTime::now_utc().unix_timestamp_nanos(), + )) + .await +} + +async fn process_transition_transaction_record_at( + api: Arc, + observed: &TransitionTransaction, + now_unix_nanos: i128, +) -> EcstoreResult { + let record_name = + transition_transaction_record_object_name(observed.transaction_id).map_err(transition_transaction_store_error)?; + // The synthetic key avoids nesting the recovery lock with the config + // object's own I/O lock. Holding it across the bounded source proof and + // remote DELETE elects one destructive recovery worker across nodes. + let recovery_lock = api + .new_ns_lock(RUSTFS_META_BUCKET, &format!("{record_name}.recovery-lock")) + .await?; + let _recovery_guard = recovery_lock + .get_write_lock(crate::set_disk::get_lock_acquire_timeout()) + .await?; + let current = match load_transition_transaction_record(api.clone(), observed.transaction_id).await { + Ok(current) => current, + Err(Error::ConfigNotFound) => return Ok(TransitionTransactionRecoveryOutcome::RecordDeleted), + Err(err) => return Err(err), + }; + if ¤t != observed { + return Ok(TransitionTransactionRecoveryOutcome::Retained); + } + + match current.state { TransitionTransactionState::Uploaded => { - delete_transition_remote_candidate(api.clone(), transaction).await?; - delete_transition_transaction_record(api, transaction).await?; - Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted) - } - TransitionTransactionState::CleanupPending => match local_commit_matches_transaction(api.clone(), transaction).await { - Ok(true) => { - delete_transition_transaction_record(api, transaction).await?; - Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) + if transition_transaction_ownership_is_active(¤t, now_unix_nanos) { + return Ok(TransitionTransactionRecoveryOutcome::Retained); } - Ok(false) => { - delete_transition_remote_candidate(api.clone(), transaction).await?; - delete_transition_transaction_record(api, transaction).await?; - Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted) - } - Err(err) if transition_source_is_missing(&err) => { - delete_transition_remote_candidate(api.clone(), transaction).await?; - delete_transition_transaction_record(api, transaction).await?; - Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted) - } - Err(err) => Err(err), - }, - TransitionTransactionState::LocalCommitStarted => { - match local_commit_matches_transaction(api.clone(), transaction).await { - Ok(true) => { - delete_transition_transaction_record(api, transaction).await?; - Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) - } - Ok(false) => Ok(TransitionTransactionRecoveryOutcome::Retained), - Err(err) if transition_source_is_missing(&err) => Ok(TransitionTransactionRecoveryOutcome::Retained), + let mut cleanup = current.clone(); + cleanup + .mark_cleanup_pending( + current.fence(), + TransitionCleanupProof { + transaction_id: current.transaction_id, + write_id: current.write_id, + remote_object: current.remote_object.clone(), + remote_version: current.remote_version.clone(), + backend_fingerprint: current.backend_fingerprint, + decision: TransitionCleanupDecision::UploadAbortedBeforeLocalCommit, + }, + ) + .map_err(transition_transaction_store_error)?; + #[cfg(test)] + pause_before_transition_recovery_claim(current.transaction_id).await; + match save_transition_transaction_record_if_current(api.clone(), ¤t, &cleanup).await { + Ok(()) => recover_cleanup_pending(api, &cleanup).await, + Err(Error::PreconditionFailed) | Err(Error::ConfigNotFound) => Ok(TransitionTransactionRecoveryOutcome::Retained), Err(err) => Err(err), } } + TransitionTransactionState::CleanupPending => recover_cleanup_pending(api, ¤t).await, + TransitionTransactionState::LocalCommitStarted => match local_commit_matches_transaction(api.clone(), ¤t).await { + Ok(true) => { + delete_transition_transaction_record(api, ¤t).await?; + Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) + } + Ok(false) => Ok(TransitionTransactionRecoveryOutcome::Retained), + Err(err) if transition_source_is_missing(&err) => Ok(TransitionTransactionRecoveryOutcome::Retained), + Err(err) => Err(err), + }, TransitionTransactionState::AbortedNoRemote | TransitionTransactionState::Committed => { + delete_transition_transaction_record(api, ¤t).await?; + Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) + } + TransitionTransactionState::UploadOutcomeUnknown => { + if transition_transaction_ownership_is_active(¤t, now_unix_nanos) { + Ok(TransitionTransactionRecoveryOutcome::Retained) + } else { + recover_unknown_upload_outcome(api, ¤t).await + } + } + TransitionTransactionState::UploadStarted => Ok(TransitionTransactionRecoveryOutcome::Retained), + } +} + +fn transition_transaction_ownership_is_active(transaction: &TransitionTransaction, now_unix_nanos: i128) -> bool { + now_unix_nanos < i128::from(transaction.not_after_unix_nanos) +} + +async fn recover_cleanup_pending( + api: Arc, + transaction: &TransitionTransaction, +) -> EcstoreResult { + match local_commit_matches_transaction(api.clone(), transaction).await { + Ok(true) => { delete_transition_transaction_record(api, transaction).await?; Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) } - TransitionTransactionState::UploadOutcomeUnknown => recover_unknown_upload_outcome(api, transaction).await, - TransitionTransactionState::UploadStarted => Ok(TransitionTransactionRecoveryOutcome::Retained), + Ok(false) => delete_unreferenced_transition_candidate(api, transaction).await, + Err(err) if transition_source_is_missing(&err) => delete_unreferenced_transition_candidate(api, transaction).await, + Err(err) => Err(err), } } +async fn delete_unreferenced_transition_candidate( + api: Arc, + transaction: &TransitionTransaction, +) -> EcstoreResult { + let current = match load_transition_transaction_record(api.clone(), transaction.transaction_id).await { + Ok(current) => current, + Err(Error::ConfigNotFound) => return Ok(TransitionTransactionRecoveryOutcome::RecordDeleted), + Err(err) => return Err(err), + }; + if ¤t != transaction || current.state != TransitionTransactionState::CleanupPending { + return Ok(TransitionTransactionRecoveryOutcome::Retained); + } + delete_transition_remote_candidate(api.clone(), ¤t).await?; + delete_transition_transaction_record(api, ¤t).await?; + Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted) +} + async fn recover_unknown_upload_outcome( api: Arc, transaction: &TransitionTransaction, @@ -928,17 +1186,7 @@ async fn recover_unknown_upload_outcome( TransitionCandidateProbe::VersionedPresent(version_id) if Uuid::parse_str(&version_id).is_ok_and(|version_id| version_id.is_nil()) => { - delete_confirmed_transition_candidate_exact_with_manager_and_identity( - &transaction.remote_object, - &version_id, - &transaction.tier_name, - transaction.backend_fingerprint, - &api.tier_config_mgr(), - ) - .await - .map_err(Error::other)?; - delete_transition_transaction_record(api, transaction).await?; - Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted) + Ok(TransitionTransactionRecoveryOutcome::Retained) } TransitionCandidateProbe::VersionedPresent(version_id) => { cleanup_recovered_unknown_upload_candidate(api, transaction, TransitionRemoteVersion::versioned(version_id)).await @@ -968,10 +1216,11 @@ async fn cleanup_recovered_unknown_upload_candidate( }, ) .map_err(transition_transaction_store_error)?; - save_transition_transaction_record(api.clone(), &cleanup).await?; - delete_transition_remote_candidate(api.clone(), &cleanup).await?; - delete_transition_transaction_record(api, &cleanup).await?; - Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted) + match save_transition_transaction_record_if_current(api.clone(), transaction, &cleanup).await { + Ok(()) => recover_cleanup_pending(api, &cleanup).await, + Err(Error::PreconditionFailed) | Err(Error::ConfigNotFound) => Ok(TransitionTransactionRecoveryOutcome::Retained), + Err(err) => Err(err), + } } fn transition_source_is_missing(err: &Error) -> bool { @@ -986,13 +1235,7 @@ fn transition_source_is_missing(err: &Error) -> bool { } async fn local_commit_matches_transaction(api: Arc, transaction: &TransitionTransaction) -> EcstoreResult { - let opts = ObjectOptions { - version_id: transaction.source.version_id.map(|version_id| version_id.to_string()), - versioned: transaction.source.version_mode == TransitionSourceVersionMode::Versioned, - version_suspended: transaction.source.version_mode == TransitionSourceVersionMode::VersionSuspended, - metadata_cache_safe: false, - ..Default::default() - }; + let opts = transition_source_lookup_options(transaction); let object = api .get_object_info(&transaction.source.bucket, &transaction.source.object, &opts) .await?; @@ -1003,6 +1246,23 @@ async fn local_commit_matches_transaction(api: Arc, transaction: &Trans && transitioned.version_id == transaction.remote_version.tier_delete_version_id().unwrap_or_default()) } +fn transition_source_lookup_options(transaction: &TransitionTransaction) -> ObjectOptions { + ObjectOptions { + version_id: match transaction.source.version_mode { + TransitionSourceVersionMode::Versioned => transaction.source.version_id.map(|version_id| version_id.to_string()), + // Both modes identify the stored null version. Query it explicitly + // so a later versioning change cannot redirect the proof to a new latest version. + TransitionSourceVersionMode::Unversioned | TransitionSourceVersionMode::VersionSuspended => { + Some(Uuid::nil().to_string()) + } + }, + versioned: transaction.source.version_mode == TransitionSourceVersionMode::Versioned, + version_suspended: transaction.source.version_mode == TransitionSourceVersionMode::VersionSuspended, + metadata_cache_safe: false, + ..Default::default() + } +} + async fn delete_transition_remote_candidate(api: Arc, transaction: &TransitionTransaction) -> EcstoreResult<()> { let version_id = transaction.remote_version.tier_delete_version_id().unwrap_or_default(); let version_id_exact = transaction.remote_version.kind == TransitionRemoteVersionKind::Versioned; @@ -1023,6 +1283,25 @@ pub async fn recover_transition_transaction_records( api: Arc, limit: usize, marker: Option, +) -> EcstoreResult { + recover_transition_transaction_records_with_now(api, limit, marker, None).await +} + +#[cfg(any(test, feature = "test-util"))] +pub async fn recover_transition_transaction_records_at( + api: Arc, + limit: usize, + marker: Option, + now_unix_nanos: i128, +) -> EcstoreResult { + recover_transition_transaction_records_with_now(api, limit, marker, Some(now_unix_nanos)).await +} + +async fn recover_transition_transaction_records_with_now( + api: Arc, + limit: usize, + marker: Option, + now_unix_nanos: Option, ) -> EcstoreResult { if limit == 0 { return Err(Error::other("transition transaction recovery limit must be greater than zero")); @@ -1087,7 +1366,13 @@ pub async fn recover_transition_transaction_records( } }; - match process_transition_transaction_record(api.clone(), &transaction).await { + let recovery = match now_unix_nanos { + Some(now_unix_nanos) => { + Box::pin(process_transition_transaction_record_at(api.clone(), &transaction, now_unix_nanos)).await + } + None => process_transition_transaction_record(api.clone(), &transaction).await, + }; + match recovery { Ok( TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted | TransitionTransactionRecoveryOutcome::RecordDeleted, @@ -1332,6 +1617,34 @@ mod tests { .expect("expired unknown upload outcome should be eligible"); } + #[test] + fn transition_ownership_window_expires_at_not_after() { + let transaction = new_transaction(); + let deadline = i128::from(transaction.not_after_unix_nanos); + + assert!(transition_transaction_ownership_is_active(&transaction, deadline - 1)); + assert!(!transition_transaction_ownership_is_active(&transaction, deadline)); + } + + #[test] + fn null_transition_source_lookup_targets_the_exact_version_shape() { + for mode in [ + TransitionSourceVersionMode::Unversioned, + TransitionSourceVersionMode::VersionSuspended, + ] { + let mut transaction = new_transaction(); + transaction.source.version_id = None; + transaction.source.version_mode = mode; + + let opts = transition_source_lookup_options(&transaction); + + assert_eq!(opts.version_id, Some(Uuid::nil().to_string())); + assert!(!opts.versioned); + assert_eq!(opts.version_suspended, mode == TransitionSourceVersionMode::VersionSuspended); + assert!(!opts.metadata_cache_safe); + } + } + fn cleanup_proof(transaction: &TransitionTransaction, decision: TransitionCleanupDecision) -> TransitionCleanupProof { TransitionCleanupProof { transaction_id: transaction.transaction_id, @@ -1346,10 +1659,17 @@ mod tests { #[test] fn remote_version_distinguishes_unknown_unversioned_and_versioned() { assert_eq!(TransitionRemoteVersion::known_from_put_response("").tier_delete_version_id(), None); + let nil_version = Uuid::nil().to_string(); + let invalid_nil = TransitionRemoteVersion::known_from_put_response(nil_version.clone()); assert_eq!( - TransitionRemoteVersion::known_from_put_response(Uuid::nil().to_string()).tier_delete_version_id(), - None + invalid_nil.tier_delete_version_id(), + Some(nil_version.as_str()), + "a non-empty version must never be downgraded to an unversioned DELETE" ); + assert!(matches!( + invalid_nil.validate(), + Err(TransitionTransactionError::Corrupt("versioned remote version is nil uuid")) + )); let version_id = Uuid::new_v4().to_string(); assert_eq!( @@ -1468,6 +1788,34 @@ mod tests { }) .is_ok() ); + + assert!(matches!( + TransitionTransaction::new(TransitionTransactionInit { + deployment_id: Uuid::new_v4(), + transaction_id: Uuid::new_v4(), + owner_epoch: Uuid::new_v4(), + write_id: Uuid::new_v4(), + source: source_identity(TransitionSourceVersionMode::Unversioned), + tier_name: "warm-tier".to_string(), + backend_fingerprint: [0; 32], + not_after_unix_nanos: 1, + }), + Err(TransitionTransactionError::Corrupt("backend fingerprint is empty")) + )); + + assert!(matches!( + TransitionTransaction::new(TransitionTransactionInit { + deployment_id: Uuid::new_v4(), + transaction_id: Uuid::new_v4(), + owner_epoch: Uuid::new_v4(), + write_id: Uuid::new_v4(), + source: source_identity(TransitionSourceVersionMode::Unversioned), + tier_name: "warm-tier".to_string(), + backend_fingerprint: BACKEND_FINGERPRINT, + not_after_unix_nanos: 0, + }), + Err(TransitionTransactionError::Corrupt("ownership deadline is not positive")) + )); } #[test] diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 8437a8e75..9e0edd9b5 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -876,6 +876,8 @@ pub use ops::multipart::{MultipartCommitBarrier, MultipartCommitPause}; pub(crate) use ops::object::DeleteObjectCommitBarrier; #[cfg(any(test, feature = "test-util"))] pub(crate) use ops::object::TransitionCleanupStoreBarrier as SetDiskTransitionCleanupStoreBarrier; +#[cfg(test)] +pub(crate) use ops::object::TransitionUploadedCommitBarrier as SetDiskTransitionUploadedCommitBarrier; pub(crate) use ops::object::body_cache_plaintext_len; #[cfg(all(test, feature = "test-util"))] pub(crate) use ops::object::cleanup_rejected_transition_upload_durably; diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index ccd64d54e..a68849af7 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -266,9 +266,9 @@ use crate::bucket::lifecycle::{ transitioned_force_delete_journal_entry, }, transition_transaction::{ - TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction, - TransitionTransactionInit, TransitionTransactionState, delete_transition_transaction_record, - load_transition_transaction_record, save_transition_transaction_record, + TransitionCleanupDecision, TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, + TransitionTransaction, TransitionTransactionInit, TransitionTransactionState, delete_transition_transaction_record, + save_transition_transaction_record, save_transition_transaction_record_if_current, }, }; use crate::bucket::quota::reservation; @@ -5513,6 +5513,7 @@ pub(crate) struct TransitionUploadCleanup { object: String, candidate: Option, cleanup_api: Option>, + cleanup_transaction: Option, armed: bool, } @@ -5523,10 +5524,20 @@ impl TransitionUploadCleanup { object: object.to_string(), candidate: None, cleanup_api: None, + cleanup_transaction: None, armed: true, } } + fn set_cleanup_owner(&mut self, api: Option>, transaction: &TransitionTransaction) { + self.cleanup_api = api; + self.cleanup_transaction = Some(transaction.clone()); + } + + fn update_cleanup_transaction(&mut self, transaction: &TransitionTransaction) { + self.cleanup_transaction = Some(transaction.clone()); + } + fn cleanup_candidate(&self) -> std::io::Result<&TransitionUploadCandidate> { self.candidate .as_ref() @@ -5559,11 +5570,13 @@ impl TransitionUploadCleanup { api: Option>, transaction: &mut TransitionTransaction, ) -> std::io::Result<()> { + let api = api.or_else(|| self.cleanup_api.clone()); self.cleanup_api = api.clone(); let candidate = self.cleanup_candidate()?.clone(); let owner_error = persist_rejected_transition_cleanup_owner(api.as_ref(), transaction, &candidate) .await .err(); + self.update_cleanup_transaction(transaction); let cleanup = cleanup_rejected_transition_upload_durably( &self.lease, &self.object, @@ -5615,15 +5628,37 @@ impl Drop for TransitionUploadCleanup { } }; let object = self.object.clone(); - let cleanup_version = candidate.cleanup_version().to_string(); - let version_id_exact = candidate.cleanup_version_is_exact(); + let candidate = candidate.clone(); let cleanup_api = self.cleanup_api.clone(); + let mut cleanup_transaction = self.cleanup_transaction.clone(); if let Ok(handle) = tokio::runtime::Handle::try_current() { handle.spawn(async move { - if let Err(err) = - cleanup_rejected_transition_upload_durably(&lease, &object, &cleanup_version, version_id_exact, cleanup_api) - .await - { + let owner_error = match (cleanup_api.as_ref(), cleanup_transaction.as_mut()) { + (Some(api), Some(transaction)) => { + persist_rejected_transition_cleanup_owner(Some(api), transaction, &candidate) + .await + .err() + } + _ => None, + }; + let cleanup_version = candidate.cleanup_version().to_string(); + let cleanup = cleanup_rejected_transition_upload_durably( + &lease, + &object, + &cleanup_version, + candidate.cleanup_version_is_exact(), + cleanup_api, + ) + .await; + let result = match (owner_error, cleanup) { + (_, Ok(())) => Ok(()), + (None, Err(cleanup_error)) => Err(cleanup_error), + (Some(owner_error), Err(cleanup_error)) => Err(crate::error::stable_io_error( + "cancelled transition upload cleanup error followed a transaction owner update failure", + format!("owner error: {owner_error}; cleanup error: {cleanup_error}"), + )), + }; + if let Err(err) = result { warn!( event = EVENT_LIFECYCLE_TRANSITION_CLEANUP, component = LOG_COMPONENT_ECSTORE, @@ -5646,25 +5681,36 @@ async fn persist_rejected_transition_cleanup_owner( transaction: &mut TransitionTransaction, candidate: &TransitionUploadCandidate, ) -> Result<()> { - match transaction.state { - TransitionTransactionState::UploadOutcomeUnknown => { - transaction - .advance( - transaction.fence(), - TransitionTransactionState::Uploaded, - Some(TransitionRemoteVersion::known_from_put_response(candidate.remote_version().to_string())), - ) - .map_err(Error::other)?; - } - TransitionTransactionState::Uploaded => {} + let expected = transaction.clone(); + let remote_version = TransitionRemoteVersion::known_from_put_response(candidate.remote_version().to_string()); + let decision = match transaction.state { + TransitionTransactionState::UploadOutcomeUnknown => TransitionCleanupDecision::RemoteVersionRecoveredAfterCancellation, + TransitionTransactionState::Uploaded => TransitionCleanupDecision::UploadAbortedBeforeLocalCommit, + TransitionTransactionState::CleanupPending if transaction.remote_version == remote_version => return Ok(()), state => { return Err(Error::other_with_context( "transition transaction state cannot own a rejected upload", format!("state {state:?}"), )); } - } - save_transition_transaction_if_available(api, transaction).await + }; + let mut cleanup = expected.clone(); + cleanup + .mark_cleanup_pending( + expected.fence(), + crate::bucket::lifecycle::transition_transaction::TransitionCleanupProof { + transaction_id: expected.transaction_id, + write_id: expected.write_id, + remote_object: expected.remote_object.clone(), + remote_version, + backend_fingerprint: expected.backend_fingerprint, + decision, + }, + ) + .map_err(Error::other)?; + compare_and_save_transition_transaction_if_available(api, &expected, &cleanup).await?; + *transaction = cleanup; + Ok(()) } pub(crate) async fn cleanup_rejected_transition_upload_durably( @@ -5779,6 +5825,26 @@ async fn save_transition_transaction_if_available(api: Option<&Arc>, tr } } +async fn compare_and_save_transition_transaction_if_available( + api: Option<&Arc>, + expected: &TransitionTransaction, + next: &TransitionTransaction, +) -> Result<()> { + if let Some(api) = api { + // The transition worker already has a deep poll chain. Keep the CAS + // read/write/receipt future off Tokio's default worker stack. + return Box::pin(save_transition_transaction_record_if_current(api.clone(), expected, next)).await; + } + #[cfg(test)] + { + Ok(()) + } + #[cfg(not(test))] + { + Err(Error::other("transition transaction store is unavailable")) + } +} + async fn advance_and_save_transition_transaction( api: Option<&Arc>, transaction: &mut TransitionTransaction, @@ -5787,10 +5853,14 @@ async fn advance_and_save_transition_transaction( ) -> Result<()> { #[cfg(test)] record_transition_uploaded_save_attempt(transaction, next); - transaction - .advance(transaction.fence(), next, remote_version) + let expected = transaction.clone(); + let mut advanced = expected.clone(); + advanced + .advance(expected.fence(), next, remote_version) .map_err(Error::other)?; - save_transition_transaction_if_available(api, transaction).await + compare_and_save_transition_transaction_if_available(api, &expected, &advanced).await?; + *transaction = advanced; + Ok(()) } #[cfg(test)] @@ -5874,29 +5944,29 @@ fn record_transition_uploaded_save_attempt(transaction: &TransitionTransaction, } } -async fn delete_transition_transaction_if_available(api: Option<&Arc>, transaction_id: Uuid) -> Result<()> { +async fn delete_transition_transaction_if_available( + api: Option<&Arc>, + transaction: &TransitionTransaction, +) -> Result<()> { if let Some(api) = api { - let transaction = match load_transition_transaction_record(api.clone(), transaction_id).await { - Ok(transaction) => transaction, - Err(Error::ConfigNotFound) => return Ok(()), - Err(err) => return Err(err), - }; - return delete_transition_transaction_record(api.clone(), &transaction).await; + // Conditional delete now includes a read and terminal receipt; box it + // for the same transition-worker stack bound as the CAS path above. + return Box::pin(delete_transition_transaction_record(api.clone(), transaction)).await; } Ok(()) } async fn delete_transition_transaction_after_remote_cleanup( api: Option<&Arc>, - transaction_id: Uuid, + transaction: &TransitionTransaction, bucket: &str, object: &str, ) { - if let Err(err) = delete_transition_transaction_if_available(api, transaction_id).await { + if let Err(err) = delete_transition_transaction_if_available(api, transaction).await { warn!( bucket = bucket, object = object, - transaction_id = %transaction_id, + transaction_id = %transaction.transaction_id, error = ?err, "transition remote candidate was cleaned but transaction record cleanup failed" ); @@ -6031,6 +6101,86 @@ async fn pause_after_transition_upload_candidate_recorded() { } } +#[cfg(test)] +struct TransitionUploadedCommitBarrierState { + bucket: String, + object: String, + arrived: tokio::sync::Notify, + release: tokio::sync::Notify, +} + +#[cfg(test)] +pub(crate) struct TransitionUploadedCommitBarrier { + state: Arc, +} + +#[cfg(test)] +static TRANSITION_UPLOADED_COMMIT_BARRIER: std::sync::OnceLock< + std::sync::Mutex>>, +> = std::sync::OnceLock::new(); + +#[cfg(test)] +impl TransitionUploadedCommitBarrier { + pub(crate) fn install(bucket: &str, object: &str) -> Self { + let state = Arc::new(TransitionUploadedCommitBarrierState { + bucket: bucket.to_string(), + object: object.to_string(), + arrived: tokio::sync::Notify::new(), + release: tokio::sync::Notify::new(), + }); + let mut slot = TRANSITION_UPLOADED_COMMIT_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("transition uploaded commit barrier mutex should not poison"); + assert!( + slot.is_none(), + "transition uploaded commit barrier must be installed by one test at a time" + ); + *slot = Some(Arc::clone(&state)); + drop(slot); + Self { state } + } + + pub(crate) async fn wait_until_paused(&self) { + tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified()) + .await + .expect("transition should persist Uploaded before acquiring its commit lock"); + } + + pub(crate) fn release(&self) { + self.state.release.notify_one(); + } +} + +#[cfg(test)] +impl Drop for TransitionUploadedCommitBarrier { + fn drop(&mut self) { + self.state.release.notify_one(); + let mut slot = TRANSITION_UPLOADED_COMMIT_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("transition uploaded commit barrier mutex should not poison"); + if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) { + *slot = None; + } + } +} + +#[cfg(test)] +async fn pause_after_transition_uploaded_persisted(bucket: &str, object: &str) { + let barrier = TRANSITION_UPLOADED_COMMIT_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("transition uploaded commit barrier mutex should not poison") + .as_ref() + .filter(|barrier| barrier.bucket == bucket && barrier.object == object) + .cloned(); + if let Some(barrier) = barrier { + barrier.arrived.notify_one(); + barrier.release.notified().await; + } +} + #[cfg(test)] #[derive(Clone, Copy, PartialEq, Eq)] enum TransitionCommitPause { @@ -8841,6 +8991,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { }; let mut upload_cleanup = TransitionUploadCleanup::new(tgt_client, &dest_obj); + upload_cleanup.set_cleanup_owner(transaction_api.clone(), &transaction); advance_and_save_transition_transaction( transaction_api.as_ref(), &mut transaction, @@ -8848,6 +8999,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { None, ) .await?; + upload_cleanup.update_cleanup_transaction(&transaction); let remote_upload = { let lease = &upload_cleanup.lease; let recorded_candidate = &mut upload_cleanup.candidate; @@ -8872,7 +9024,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { failure.error )))); } - delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) + delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object) .await; } return Err(failure.error); @@ -8886,7 +9038,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { "{err}; rejected remote upload cleanup failed: {cleanup_err}" )))); } - delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object).await; + delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object).await; return Err(err.into()); } let fleet_proof = remote_version_state_writer_fleet_proof(); @@ -8902,7 +9054,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { "{err}; rejected remote upload cleanup failed: {cleanup_err}" )))); } - delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) + delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object) .await; return Err(err.into()); } @@ -8921,9 +9073,13 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { "{err}; uploaded transition transaction persist failed and cleanup failed: {cleanup_err}" )))); } - delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object).await; + delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object).await; return Err(err); } + upload_cleanup.update_cleanup_transaction(&transaction); + + #[cfg(test)] + pause_after_transition_uploaded_persisted(bucket, object).await; let commit_opts = opts.as_commit_opts(); // Note: Using clone() here is necessary because ObjectOptions has 124 fields. @@ -8937,7 +9093,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { if upload_cleanup.cleanup().await.is_ok() { delete_transition_transaction_after_remote_cleanup( transaction_api.as_ref(), - transaction_id, + &transaction, bucket, object, ) @@ -8954,7 +9110,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { Err(err) => { drop(transition_lock_guard); if upload_cleanup.cleanup().await.is_ok() { - delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) + delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object) .await; } return Err(err); @@ -8970,8 +9126,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let already_transitioned = current_fi.transition_status == TRANSITION_COMPLETE; drop(transition_lock_guard); if upload_cleanup.cleanup().await.is_ok() { - delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) - .await; + delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object).await; } if already_transitioned { return Ok(()); @@ -9000,8 +9155,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { if transition_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) { drop(transition_lock_guard); if upload_cleanup.cleanup().await.is_ok() { - delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) - .await; + delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object).await; } return Err(StorageError::NamespaceLockQuorumUnavailable { mode: "transition_object_commit", @@ -9016,8 +9170,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { if !upload_cleanup.lease.is_current_generation() { drop(transition_lock_guard); if upload_cleanup.cleanup().await.is_ok() { - delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) - .await; + delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object).await; } return Err(Error::other("remote tier configuration changed during transition")); } @@ -9033,8 +9186,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { { drop(transition_lock_guard); if upload_cleanup.cleanup().await.is_ok() { - delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) - .await; + delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object).await; } return Err(Error::other("remote version state fleet capability changed during transition")); } @@ -9048,8 +9200,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { { drop(transition_lock_guard); if upload_cleanup.cleanup().await.is_ok() { - delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) - .await; + delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), &transaction, bucket, object).await; } return Err(err); } @@ -9065,19 +9216,16 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { drop(transition_lock_guard); return Err(err); } - match transaction.advance(transaction.fence(), TransitionTransactionState::Committed, None) { - Ok(_) => { - if let Err(err) = save_transition_transaction_if_available(transaction_api.as_ref(), &transaction).await { - warn!( - bucket = bucket, - object = object, - transaction_id = %transaction_id, - error = ?err, - "transition committed locally but transaction committed-state persist failed" - ); - } else if let Err(err) = - delete_transition_transaction_if_available(transaction_api.as_ref(), transaction_id).await - { + match advance_and_save_transition_transaction( + transaction_api.as_ref(), + &mut transaction, + TransitionTransactionState::Committed, + None, + ) + .await + { + Ok(()) => { + if let Err(err) = delete_transition_transaction_if_available(transaction_api.as_ref(), &transaction).await { warn!( bucket = bucket, object = object, diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 5aaf3e5e8..7a169fe33 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -832,11 +832,12 @@ mod tests { }, transition_transaction::{ TRANSITION_TRANSACTION_RECORD_PREFIX, TransitionCleanupDecision, TransitionCleanupProof, TransitionOperatorError, - TransitionOperatorProbe, TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, - TransitionTransaction, TransitionTransactionInit, TransitionTransactionState, + TransitionOperatorProbe, TransitionRecoveryClaimBarrier, TransitionRemoteVersion, TransitionSourceIdentity, + TransitionSourceVersionMode, TransitionTransaction, TransitionTransactionInit, TransitionTransactionState, delete_transition_candidate_for_operator, finalize_missing_transition_transaction_for_operator, inspect_transition_transaction_for_operator, load_transition_transaction_record, - recover_transition_transaction_records, save_transition_transaction_record, + recover_transition_transaction_records, recover_transition_transaction_records_at, + save_transition_transaction_record, save_transition_transaction_record_if_current, transition_transaction_record_object_name, }, validate_durable_ilm_record, @@ -864,6 +865,7 @@ mod tests { tier_mutation_peer::{TierMutationPeerError, TierMutationPeerState, handle_tier_mutation_peer_request}, warm_backend::{TransitionCandidateProbe, WarmBackend}, }, + set_disk::SetDiskTransitionUploadedCommitBarrier as TransitionUploadedCommitBarrier, storage_api_contracts::list::ListOperations as _, }; #[cfg(feature = "test-util")] @@ -17469,7 +17471,82 @@ mod tests { #[cfg(feature = "test-util")] #[tokio::test] #[serial_test::serial(storage_class_env)] - async fn transition_transaction_recovery_deletes_uploaded_remote_candidate() { + async fn transition_transaction_recovery_retains_active_uploaded_candidate_until_commit() { + let temp_dir = tempfile::tempdir().expect("create temp store dir"); + let (ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "transition-transaction-active-uploaded", &[4])) + .await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + + let tier_name = "TXACTIVEUPLOADED"; + let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await; + let bucket = "transition-transaction-active-uploaded-bucket"; + let object = "object.bin"; + let payload = b"active Uploaded ownership must survive recovery until local commit".repeat(1024); + store + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("source bucket should be created"); + let mut reader = PutObjReader::from_vec(payload.clone()); + let source = store + .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .await + .expect("source object should be written"); + let barrier = TransitionUploadedCommitBarrier::install(bucket, object); + let transition_store = store.clone(); + let transition = tokio::spawn(async move { + transition_store + .transition_object( + bucket, + object, + &ObjectOptions { + transition: TransitionOptions { + status: TRANSITION_PENDING.to_string(), + tier: tier_name.to_string(), + etag: source.etag.clone().expect("source object should have an ETag"), + ..Default::default() + }, + version_id: source.version_id.map(|version_id| version_id.to_string()), + mod_time: source.mod_time, + ..Default::default() + }, + ) + .await + }); + barrier.wait_until_paused().await; + + let stats = recover_transition_transaction_records(store.clone(), 100, None) + .await + .expect("recovery should inspect the active Uploaded transaction"); + assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (1, 0, 1, 0)); + assert_eq!(backend.object_count().await, 1, "active ownership must retain the remote candidate"); + assert_eq!(backend.remove_count().await, 0, "active ownership must fence remote DELETE"); + assert_eq!(transition_transaction_record_count(store.clone()).await, 1); + + barrier.release(); + transition + .await + .expect("transition task should join") + .expect("transition should commit after the barrier is released"); + assert_eq!(transition_transaction_record_count(store.clone()).await, 0); + assert_eq!(backend.remove_count().await, 0); + + let mut body = Vec::new(); + store + .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("committed tier object should remain readable") + .stream + .read_to_end(&mut body) + .await + .expect("tier object body should drain"); + assert_eq!(body, payload); + } + + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn transition_transaction_recovery_deletes_expired_uploaded_remote_candidate() { let temp_dir = tempfile::tempdir().expect("create temp store dir"); let (ctx, store, _shutdown) = without_storage_class_env(build_isolated_test_store(temp_dir.path(), "transition-transaction-recovery", &[4])).await; @@ -17477,12 +17554,114 @@ mod tests { let tier_name = "TXRECOVERY"; let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await; + let backend_identity = TierConfigMgr::acquire_operation_lease(&ctx.tier_config_mgr(), tier_name) + .await + .expect("tier lease should resolve") + .backend_identity(); + let exact_version = uuid::Uuid::new_v4().to_string(); + let cases = [ + ( + "exact", + exact_version.clone(), + TransitionRemoteVersion::versioned(exact_version), + TransitionSourceVersionMode::Versioned, + ), + ( + "suspended-null", + "null".to_string(), + TransitionRemoteVersion::versioned("null"), + TransitionSourceVersionMode::VersionSuspended, + ), + ( + "known-unversioned", + String::new(), + TransitionRemoteVersion::unversioned(), + TransitionSourceVersionMode::Unversioned, + ), + ]; + let mut expected_removes = Vec::new(); + for (case, put_version, remote_version, source_mode) in cases { + let mut transaction = TransitionTransaction::new(TransitionTransactionInit { + deployment_id: ctx.deployment_id().expect("test store should initialize deployment id"), + transaction_id: uuid::Uuid::new_v4(), + owner_epoch: uuid::Uuid::new_v4(), + write_id: uuid::Uuid::new_v4(), + source: TransitionSourceIdentity { + bucket: "source-bucket".to_string(), + object: format!("source-{case}"), + version_id: (source_mode == TransitionSourceVersionMode::Versioned).then(uuid::Uuid::new_v4), + data_dir: uuid::Uuid::new_v4(), + mod_time_unix_nanos: 1_770_000_000_000_000_000, + size: 42, + etag: "source-etag".to_string(), + version_mode: source_mode, + }, + tier_name: tier_name.to_string(), + backend_fingerprint: backend_identity, + not_after_unix_nanos: 1, + }) + .expect("transaction should build"); + transaction + .advance(transaction.fence(), TransitionTransactionState::Uploaded, Some(remote_version)) + .expect("transaction should enter uploaded state"); + backend.set_put_remote_version(Some(put_version.clone())).await; + let candidate = bytes::Bytes::from_static(b"orphan candidate"); + backend + .put( + &transaction.remote_object, + ReaderImpl::Body(candidate.clone()), + i64::try_from(candidate.len()).expect("test candidate length should fit i64"), + ) + .await + .expect("mock backend should accept candidate"); + save_transition_transaction_record(store.clone(), &transaction) + .await + .expect("transaction record should persist"); + expected_removes.push((transaction.remote_object, put_version)); + } + + let stats = recover_transition_transaction_records(store.clone(), 100, None) + .await + .expect("transition transaction recovery should run"); + + assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (3, 3, 0, 0)); + assert_eq!(transition_transaction_record_count(store.clone()).await, 0); + let mut actual_removes = backend.remove_versions().await; + actual_removes.sort(); + expected_removes.sort(); + assert_eq!(actual_removes, expected_removes, "recovery must preserve each remote version shape"); + assert_eq!(backend.exact_remove_count(), 2); + assert_eq!(backend.object_count().await, 0); + + let replay = recover_transition_transaction_records(store, 100, None) + .await + .expect("replayed recovery should remain idempotent"); + assert_eq!((replay.scanned, replay.recovered, replay.retained, replay.failed), (0, 0, 0, 0)); + assert_eq!( + backend.remove_versions().await.len(), + 3, + "each expired candidate must be deleted only once" + ); + } + + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn transition_transaction_recovery_cas_loses_to_active_state_advance() { + let temp_dir = tempfile::tempdir().expect("create temp store dir"); + let (ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "transition-transaction-recovery-cas", &[4])) + .await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + + let tier_name = "TXRECOVERYCAS"; + let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await; let backend_identity = TierConfigMgr::acquire_operation_lease(&ctx.tier_config_mgr(), tier_name) .await .expect("tier lease should resolve") .backend_identity(); let remote_version = uuid::Uuid::new_v4().to_string(); - let mut transaction = TransitionTransaction::new(TransitionTransactionInit { + let mut uploaded = TransitionTransaction::new(TransitionTransactionInit { deployment_id: ctx.deployment_id().expect("test store should initialize deployment id"), transaction_id: uuid::Uuid::new_v4(), owner_epoch: uuid::Uuid::new_v4(), @@ -17499,7 +17678,97 @@ mod tests { }, tier_name: tier_name.to_string(), backend_fingerprint: backend_identity, - not_after_unix_nanos: 1_780_000_000_000_000_000, + not_after_unix_nanos: 1, + }) + .expect("transaction should build"); + uploaded + .advance( + uploaded.fence(), + TransitionTransactionState::Uploaded, + Some(TransitionRemoteVersion::versioned(remote_version.clone())), + ) + .expect("transaction should enter uploaded state"); + backend.set_put_remote_version(Some(remote_version)).await; + let candidate = bytes::Bytes::from_static(b"CAS-owned transition remote candidate"); + backend + .put( + &uploaded.remote_object, + ReaderImpl::Body(candidate.clone()), + i64::try_from(candidate.len()).expect("test candidate length should fit i64"), + ) + .await + .expect("mock backend should accept candidate"); + save_transition_transaction_record(store.clone(), &uploaded) + .await + .expect("transaction record should persist"); + + let barrier = TransitionRecoveryClaimBarrier::install(uploaded.transaction_id); + let recovery_store = store.clone(); + let recovery = tokio::spawn(async move { recover_transition_transaction_records(recovery_store, 100, None).await }); + barrier.wait_until_paused().await; + + let mut active = uploaded.clone(); + active + .advance(active.fence(), TransitionTransactionState::LocalCommitStarted, None) + .expect("active owner should advance to local commit"); + save_transition_transaction_record_if_current(store.clone(), &uploaded, &active) + .await + .expect("active owner should win the persisted CAS"); + barrier.release(); + + let stats = recovery + .await + .expect("recovery task should join") + .expect("recovery should treat the lost CAS as a retained transaction"); + assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (1, 0, 1, 0)); + assert_eq!( + load_transition_transaction_record(store, uploaded.transaction_id) + .await + .expect("newer transaction revision must remain"), + active + ); + assert_eq!(backend.object_count().await, 1, "a stale recovery must not delete the candidate"); + assert_eq!(backend.remove_count().await, 0); + } + + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn transition_transaction_recovery_quarantines_missing_required_fields() { + let temp_dir = tempfile::tempdir().expect("create temp store dir"); + let (ctx, store, _shutdown) = without_storage_class_env(build_isolated_test_store( + temp_dir.path(), + "transition-transaction-recovery-corrupt", + &[4], + )) + .await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + + let tier_name = "TXRECOVERYCORRUPT"; + let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await; + let backend_identity = TierConfigMgr::acquire_operation_lease(&ctx.tier_config_mgr(), tier_name) + .await + .expect("tier lease should resolve") + .backend_identity(); + let remote_version = uuid::Uuid::new_v4().to_string(); + let mut transaction = TransitionTransaction::new(TransitionTransactionInit { + deployment_id: ctx.deployment_id().expect("test store should initialize deployment id"), + transaction_id: uuid::Uuid::new_v4(), + owner_epoch: uuid::Uuid::new_v4(), + write_id: uuid::Uuid::new_v4(), + source: TransitionSourceIdentity { + bucket: "source-bucket".to_string(), + object: "source-object".to_string(), + version_id: None, + data_dir: uuid::Uuid::new_v4(), + mod_time_unix_nanos: 1_770_000_000_000_000_000, + size: 42, + etag: "source-etag".to_string(), + version_mode: TransitionSourceVersionMode::Unversioned, + }, + tier_name: tier_name.to_string(), + backend_fingerprint: backend_identity, + not_after_unix_nanos: 1, }) .expect("transaction should build"); transaction @@ -17509,8 +17778,8 @@ mod tests { Some(TransitionRemoteVersion::versioned(remote_version.clone())), ) .expect("transaction should enter uploaded state"); - backend.set_put_remote_version(Some(remote_version.clone())).await; - let candidate = bytes::Bytes::from_static(b"orphan candidate"); + backend.set_put_remote_version(Some(remote_version)).await; + let candidate = bytes::Bytes::from_static(b"corrupt journal candidate"); backend .put( &transaction.remote_object, @@ -17519,23 +17788,34 @@ mod tests { ) .await .expect("mock backend should accept candidate"); - save_transition_transaction_record(store.clone(), &transaction) - .await - .expect("transaction record should persist"); + + let mut persisted: serde_json::Value = serde_json::from_slice(&transaction.encode().expect("transaction should encode")) + .expect("transaction should be JSON"); + persisted["transaction"] + .as_object_mut() + .expect("transaction payload should be an object") + .remove("not_after_unix_nanos"); + let path = + transition_transaction_record_object_name(transaction.transaction_id).expect("transaction path should be canonical"); + com::save_config( + store.clone(), + &path, + serde_json::to_vec(&persisted).expect("corrupt fixture should encode"), + ) + .await + .expect("corrupt transaction fixture should persist"); let stats = recover_transition_transaction_records(store.clone(), 100, None) .await - .expect("transition transaction recovery should run"); - - assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (1, 1, 0, 0)); - assert_eq!(transition_transaction_record_count(store.clone()).await, 0); + .expect("recovery scan should isolate a corrupt record"); + assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (1, 0, 0, 1)); assert_eq!( - backend.remove_versions().await, - vec![(transaction.remote_object.clone(), remote_version)], - "recovery must delete the exact uploaded candidate" + transition_transaction_record_count(store).await, + 1, + "corrupt evidence must remain quarantined" ); - assert_eq!(backend.exact_remove_count(), 1); - assert_eq!(backend.object_count().await, 0); + assert_eq!(backend.object_count().await, 1, "corrupt evidence must never authorize remote DELETE"); + assert_eq!(backend.remove_count().await, 0); } #[cfg(feature = "test-util")] @@ -18130,14 +18410,14 @@ mod tests { #[cfg(feature = "test-util")] #[tokio::test] #[serial_test::serial(storage_class_env)] - async fn transition_transaction_recovery_deletes_provider_recovered_unknown_upload() { + async fn transition_transaction_recovery_handles_provider_recovered_unknown_upload() { let versioned_remote = uuid::Uuid::new_v4().to_string(); let nil_remote = uuid::Uuid::nil().to_string(); - for (case, tier_name, remote_version) in [ - ("missing", "TXPROBEMISSING", None), - ("unversioned", "TXPROBEUNVERSIONED", Some(String::new())), - ("versioned", "TXPROBEVERSIONED", Some(versioned_remote)), - ("nil-version", "TXPROBENILVERSION", Some(nil_remote)), + for (case, tier_name, remote_version, should_recover) in [ + ("missing", "TXPROBEMISSING", None, true), + ("unversioned", "TXPROBEUNVERSIONED", Some(String::new()), true), + ("versioned", "TXPROBEVERSIONED", Some(versioned_remote), true), + ("nil-version", "TXPROBENILVERSION", Some(nil_remote), false), ] { let temp_dir = tempfile::tempdir().expect("create temp store dir"); let (ctx, store, _shutdown) = without_storage_class_env(build_isolated_test_store( @@ -18170,7 +18450,7 @@ mod tests { }, tier_name: tier_name.to_string(), backend_fingerprint: backend_identity, - not_after_unix_nanos: 1_780_000_000_000_000_000, + not_after_unix_nanos: 1, }) .expect("transaction should build"); transaction @@ -18197,14 +18477,18 @@ mod tests { .await .expect("transition transaction recovery should run"); - assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (1, 1, 0, 0)); - assert_eq!(transition_transaction_record_count(store.clone()).await, 0); + assert_eq!( + (stats.scanned, stats.recovered, stats.retained, stats.failed), + if should_recover { (1, 1, 0, 0) } else { (1, 0, 1, 0) } + ); + assert_eq!(transition_transaction_record_count(store.clone()).await, usize::from(!should_recover)); assert_eq!( backend.object_count().await, - 0, - "case {case}: recovered unknown upload candidate must be absent" + usize::from(!should_recover), + "case {case}: only a valid provider version state may be recovered destructively" ); let removed = remote_version + .filter(|_| should_recover) .map(|version| vec![(transaction.remote_object.clone(), version)]) .unwrap_or_default(); assert_eq!( @@ -18478,9 +18762,10 @@ mod tests { assert_eq!(backend.remove_count().await, 0, "unsupported recovery must not attempt cleanup"); backend.set_transition_candidate_probe_override(None).await; - let stats = recover_transition_transaction_records(store.clone(), 100, None) - .await - .expect("provider-authoritative recovery should run"); + let stats = + recover_transition_transaction_records_at(store.clone(), 100, None, i128::from(transaction.not_after_unix_nanos) + 1) + .await + .expect("provider-authoritative recovery should run"); assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (1, 1, 0, 0)); assert_eq!(transition_transaction_record_count(store.clone()).await, 0); assert_eq!(backend.object_count().await, 0, "recovery must delete the provider-confirmed candidate"); diff --git a/crates/scanner/tests/lifecycle_integration_test.rs b/crates/scanner/tests/lifecycle_integration_test.rs index fd81ab3e7..26ef8496c 100644 --- a/crates/scanner/tests/lifecycle_integration_test.rs +++ b/crates/scanner/tests/lifecycle_integration_test.rs @@ -47,8 +47,8 @@ use storage_api::lifecycle::{ TransitionCleanupStoreBarrier, TransitionOptions, assert_transition_meta_consistent, enqueue_transition_for_existing_objects, expire_transitioned_object, free_version_count, get_bucket_metadata, get_global_tier_config_mgr, init_background_expiry, init_bucket_metadata_sys, init_local_disks, is_err_object_not_found, is_err_version_not_found, new_disk, - path2_bucket_object_with_base_path, recover_transition_transaction_records, register_mock_tier_util, update_bucket_metadata, - wait_for_free_version_absence, + path2_bucket_object_with_base_path, recover_transition_transaction_records, recover_transition_transaction_records_at, + register_mock_tier_util, update_bucket_metadata, wait_for_free_version_absence, }; static GLOBAL_ENV: OnceLock<(Vec, Arc)> = OnceLock::new(); @@ -944,6 +944,14 @@ mod serial_tests { .is_cancelled() ); + tokio::time::timeout(Duration::from_secs(5), async { + while backend.exact_remove_count() < 1 { + tokio::task::yield_now().await; + } + }) + .await + .expect("Drop must persist cleanup ownership before attempting the exact remote delete"); + let retained = tokio::time::timeout(Duration::from_secs(30), async { loop { let recovery = recover_transition_transaction_records(ecstore.clone(), 100, None) @@ -1088,7 +1096,7 @@ mod serial_tests { .expect_err("a versioned candidate must not commit to an unversioned tier"); match case { - CleanupCase::Persisted | CleanupCase::DeleteFallback => { + CleanupCase::Persisted => { assert_eq!(backend.remove_versions().await, backend.put_versions().await); assert_eq!(backend.object_count().await, 0, "cleanup must remove the exact candidate"); let recovery = recover_transition_transaction_records(ecstore.clone(), 100, None) @@ -1102,6 +1110,25 @@ mod serial_tests { .expect("successful reconciliation must remove every transition transaction"); assert_eq!(empty.scanned, 0); } + CleanupCase::DeleteFallback => { + assert_eq!(backend.remove_versions().await, backend.put_versions().await); + assert_eq!(backend.object_count().await, 0, "cleanup must remove the exact candidate"); + let retained = recover_transition_transaction_records(ecstore.clone(), 100, None) + .await + .expect("active unknown ownership must remain fenced after the transaction store was offline"); + assert_eq!((retained.scanned, retained.recovered, retained.retained, retained.failed), (1, 0, 1, 0)); + let recovered = recover_transition_transaction_records_at(ecstore.clone(), 100, None, i128::MAX) + .await + .expect("expired unknown ownership may use the provider's missing proof"); + assert_eq!( + (recovered.scanned, recovered.recovered, recovered.retained, recovered.failed), + (1, 1, 0, 0) + ); + let empty = recover_transition_transaction_records(ecstore.clone(), 100, None) + .await + .expect("expired missing-candidate reconciliation must remove the transaction"); + assert_eq!(empty.scanned, 0); + } CleanupCase::RetryPersisted => { assert!( !err.to_string().contains("journal retry error"), @@ -1139,9 +1166,9 @@ mod serial_tests { assert_eq!(retained.recovered, 0); assert_eq!(retained.retained + retained.failed, 1); backend.set_remove_failure(false); - let recovered = recover_transition_transaction_records(ecstore.clone(), 100, None) + let recovered = recover_transition_transaction_records_at(ecstore.clone(), 100, None, i128::MAX) .await - .expect("recovery should delete the candidate after the backend becomes available"); + .expect("expired recovery should delete the candidate after the backend becomes available"); assert_eq!( (recovered.scanned, recovered.recovered, recovered.retained, recovered.failed), (1, 1, 0, 0) diff --git a/crates/scanner/tests/storage_api/mod.rs b/crates/scanner/tests/storage_api/mod.rs index 186983c29..c71522dc8 100644 --- a/crates/scanner/tests/storage_api/mod.rs +++ b/crates/scanner/tests/storage_api/mod.rs @@ -12,7 +12,9 @@ // See the License for the specific language governing permissions and // limitations under the License. -pub(crate) use rustfs_ecstore::api::bucket::lifecycle::transition_transaction::recover_transition_transaction_records; +pub(crate) use rustfs_ecstore::api::bucket::lifecycle::transition_transaction::{ + recover_transition_transaction_records, recover_transition_transaction_records_at, +}; pub(crate) use rustfs_ecstore::api::bucket::lifecycle::{ bucket_lifecycle_audit::LcEventSrc, bucket_lifecycle_ops::{ @@ -54,6 +56,7 @@ pub(crate) mod lifecycle { assert_transition_meta_consistent, enqueue_transition_for_existing_objects, expire_transitioned_object, free_version_count, get_bucket_metadata, get_global_tier_config_mgr, init_background_expiry, init_bucket_metadata_sys, init_local_disks, is_err_object_not_found, is_err_version_not_found, new_disk, path2_bucket_object_with_base_path, - recover_transition_transaction_records, register_mock_tier_util, update_bucket_metadata, wait_for_free_version_absence, + recover_transition_transaction_records, recover_transition_transaction_records_at, register_mock_tier_util, + update_bucket_metadata, wait_for_free_version_absence, }; }