diff --git a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs index de6ebe8f6..282a090c2 100644 --- a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs +++ b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs @@ -26,6 +26,7 @@ use crate::bucket::lifecycle::tier_sweeper::delete_object_from_remote_tier_idemp 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::store::ECStore; @@ -673,12 +674,69 @@ pub async fn process_transition_transaction_record( delete_transition_transaction_record(api, transaction.transaction_id).await?; Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) } - TransitionTransactionState::UploadStarted | TransitionTransactionState::UploadOutcomeUnknown => { + TransitionTransactionState::UploadOutcomeUnknown => recover_unknown_upload_outcome(api, transaction).await, + TransitionTransactionState::UploadStarted => Ok(TransitionTransactionRecoveryOutcome::Retained), + } +} + +async fn recover_unknown_upload_outcome( + api: Arc, + transaction: &TransitionTransaction, +) -> EcstoreResult { + let lease = TierConfigMgr::acquire_operation_lease_for_backend_identity( + &api.tier_config_mgr(), + &transaction.tier_name, + transaction.backend_fingerprint, + ) + .await + .map_err(Error::other)?; + + match lease + .probe_transition_candidate(&transaction.remote_object) + .await + .map_err(Error::other)? + { + TransitionCandidateProbe::Missing => { + delete_transition_transaction_record(api, transaction.transaction_id).await?; + Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) + } + TransitionCandidateProbe::UnversionedPresent => { + cleanup_recovered_unknown_upload_candidate(api, transaction, TransitionRemoteVersion::unversioned()).await + } + TransitionCandidateProbe::VersionedPresent(version_id) => { + cleanup_recovered_unknown_upload_candidate(api, transaction, TransitionRemoteVersion::versioned(version_id)).await + } + TransitionCandidateProbe::Ambiguous | TransitionCandidateProbe::Unsupported => { Ok(TransitionTransactionRecoveryOutcome::Retained) } } } +async fn cleanup_recovered_unknown_upload_candidate( + api: Arc, + transaction: &TransitionTransaction, + remote_version: TransitionRemoteVersion, +) -> EcstoreResult { + let mut cleanup = transaction.clone(); + cleanup + .mark_cleanup_pending( + transaction.fence(), + TransitionCleanupProof { + transaction_id: transaction.transaction_id, + write_id: transaction.write_id, + remote_object: transaction.remote_object.clone(), + remote_version, + backend_fingerprint: transaction.backend_fingerprint, + decision: TransitionCleanupDecision::RemoteVersionRecoveredAfterCancellation, + }, + ) + .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.transaction_id).await?; + Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted) +} + fn transition_source_is_missing(err: &Error) -> bool { matches!( err, diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 482104ae3..ba8c400c5 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -2534,10 +2534,6 @@ mod tests { }; let upload_started = new_transaction(); - let mut upload_outcome_unknown = new_transaction(); - upload_outcome_unknown - .advance(upload_outcome_unknown.fence(), TransitionTransactionState::UploadOutcomeUnknown, None) - .expect("transaction should enter unknown upload outcome state"); let mut local_commit_started = new_transaction(); local_commit_started .advance( @@ -2551,7 +2547,7 @@ mod tests { .expect("transaction should enter local commit state"); backend.set_put_remote_version(Some(remote_version)).await; - for transaction in [&upload_started, &upload_outcome_unknown, &local_commit_started] { + for transaction in [&upload_started, &local_commit_started] { let candidate = bytes::Bytes::from_static(b"unproven transition remote candidate"); backend .put( @@ -2570,14 +2566,104 @@ mod tests { .await .expect("transition transaction recovery should run"); - assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (3, 0, 3, 0)); + assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (2, 0, 2, 0)); assert_eq!( transition_transaction_record_count(store.clone()).await, - 3, - "an unknown upload outcome or unproven local commit must remain for authoritative reconcile" + 2, + "an upload without completion proof or unproven local commit must remain for authoritative reconcile" ); - assert_eq!(backend.object_count().await, 3, "recovery must not delete an unproven remote candidate"); + assert_eq!(backend.object_count().await, 2, "recovery must not delete an unproven remote candidate"); assert_eq!(backend.remove_count().await, 0); assert_eq!(backend.exact_remove_count(), 0); } + + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn transition_transaction_recovery_deletes_provider_recovered_unknown_upload() { + let versioned_remote = uuid::Uuid::new_v4().to_string(); + for (case, tier_name, remote_version) in [ + ("missing", "TXPROBEMISSING", None), + ("unversioned", "TXPROBEUNVERSIONED", Some(String::new())), + ("versioned", "TXPROBEVERSIONED", Some(versioned_remote)), + ] { + 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(), + &format!("transition-transaction-probe-{case}"), + &[4], + )) + .await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + + 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 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_780_000_000_000_000_000, + }) + .expect("transaction should build"); + transaction + .advance(transaction.fence(), TransitionTransactionState::UploadOutcomeUnknown, None) + .expect("transaction should enter unknown upload outcome state"); + + if let Some(version) = &remote_version { + backend.set_put_remote_version(Some(version.clone())).await; + let candidate = bytes::Bytes::from_static(b"provider-recovered transition remote 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"); + + 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); + assert_eq!( + backend.object_count().await, + 0, + "case {case}: recovered unknown upload candidate must be absent" + ); + let removed = remote_version + .map(|version| vec![(transaction.remote_object.clone(), version)]) + .unwrap_or_default(); + assert_eq!( + backend.remove_versions().await, + removed, + "case {case}: recovery must delete only provider-recovered candidates" + ); + assert_eq!( + backend.exact_remove_count(), + usize::from(removed.first().is_some_and(|(_, version)| !version.is_empty())) + ); + } + } }