From f32597bdb0104ad753949b985770010e5a6a446a Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 21 Jul 2026 11:17:24 +0800 Subject: [PATCH] fix(ilm): recover uploaded transition transactions (#5068) * fix(ilm): recover uploaded transition transactions Persist transition transaction records through production transition uploads, use transaction-scoped remote object names, and start a conservative recovery loop for uploaded candidates. Co-Authored-By: heihutu * fix(ilm): reconcile committed transition records Drop LocalCommitStarted transaction records only after object metadata confirms that the local transition commit already points at the same tier object and version. Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 23 +- .../lifecycle/transition_transaction.rs | 373 +++++++++++++++++- crates/ecstore/src/runtime/instance.rs | 19 +- crates/ecstore/src/set_disk/ops/object.rs | 291 +++++++++++++- crates/ecstore/src/store/init.rs | 212 ++++++++++ 5 files changed, 890 insertions(+), 28 deletions(-) diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 1196c2a24..6978aff14 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -32,6 +32,7 @@ use crate::bucket::lifecycle::tier_free_version_recovery::{ }; use crate::bucket::lifecycle::tier_last_day_stats::{DailyAllTierStats, LastDayTierStats}; use crate::bucket::lifecycle::tier_sweeper::{Jentry, delete_object_from_remote_tier_idempotent_with_manager_and_identity}; +use crate::bucket::lifecycle::transition_transaction::run_transition_transaction_recovery_loop; use crate::bucket::versioning_sys::BucketVersioningSys; use crate::client::object_api_utils::new_getobjectreader; use crate::disk::error::DiskError; @@ -1535,7 +1536,8 @@ pub async fn init_background_expiry(api: Arc) { ExpiryState::resize_workers(workers, api.clone()).await; let _ = spawn_tier_free_version_recovery_once(api.clone(), &TIER_FREE_VERSION_RECOVERY_STARTED); - spawn_tier_delete_journal_recovery_once(api); + spawn_tier_delete_journal_recovery_once(api.clone()); + spawn_transition_transaction_recovery_once(api); } fn spawn_tier_free_version_recovery_once(api: Arc, started: &OnceLock<()>) -> Option> { @@ -1802,6 +1804,25 @@ fn spawn_tier_delete_journal_recovery_once(api: Arc) { }); } +fn spawn_transition_transaction_recovery_once(api: Arc) { + let Some(cancel_token) = api.ctx.background_cancel_token() else { + error!( + event = EVENT_LIFECYCLE_WORKER_STATE, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + store_id = %api.id, + "Transition transaction recovery was not started because the store shutdown token is unavailable" + ); + return; + }; + if !api.ctx.mark_transition_transaction_recovery_started(api.id) { + return; + } + tokio::spawn(async move { + run_transition_transaction_recovery_loop(api, cancel_token).await; + }); +} + #[derive(Debug, Clone)] struct StaleMultipartUploadCandidate { path: String, diff --git a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs index 00c684363..56d776738 100644 --- a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs +++ b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs @@ -12,12 +12,32 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::{future::Future, sync::Arc, time::Duration}; + use rustfs_utils::crypto::{hex_sha256, is_sha256_checksum}; use serde::{Deserialize, Serialize}; +use tokio_util::sync::CancellationToken; +use tracing::{debug, warn}; use uuid::Uuid; +use crate::bucket::lifecycle::config_boundary; +use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE; +use crate::bucket::lifecycle::tier_sweeper::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::storage_api_contracts::{list::ListOperations as _, object::ObjectOperations as _}; +use crate::store::ECStore; + +const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle"; +const EVENT_LIFECYCLE_TRANSITION_TRANSACTION_RECOVERY: &str = "lifecycle_transition_transaction_recovery"; +pub const DEFAULT_TRANSITION_TRANSACTION_RECOVERY_LIMIT: usize = 1_000; +const TRANSITION_TRANSACTION_RECOVERY_INTERVAL: Duration = Duration::from_secs(60); +const TRANSITION_TRANSACTION_RECOVERY_TIMEOUT: Duration = Duration::from_secs(300); pub const TRANSITION_TRANSACTION_SCHEMA: &str = "rustfs-transition-transaction-v1"; pub const TRANSITION_TRANSACTION_PREFIX: &str = "ilm/transition-transactions"; +pub const TRANSITION_TRANSACTION_RECORD_PREFIX: &str = "ilm/transition-transactions/records"; pub const MAX_TRANSITION_TRANSACTION_SIZE: usize = 64 * 1024; pub type Result = std::result::Result; @@ -209,7 +229,7 @@ pub struct TransitionTransaction { pub write_id: Uuid, pub source: TransitionSourceIdentity, pub tier_name: String, - pub backend_fingerprint: String, + pub backend_fingerprint: [u8; 32], pub remote_object: String, pub remote_version: TransitionRemoteVersion, pub state: TransitionTransactionState, @@ -224,7 +244,7 @@ pub struct TransitionTransactionInit { pub write_id: Uuid, pub source: TransitionSourceIdentity, pub tier_name: String, - pub backend_fingerprint: String, + pub backend_fingerprint: [u8; 32], pub not_after_unix_nanos: i64, } @@ -277,9 +297,6 @@ impl TransitionTransaction { if self.tier_name.is_empty() { return Err(TransitionTransactionError::Corrupt("tier name is empty")); } - if !is_sha256_checksum(&self.backend_fingerprint) { - return Err(TransitionTransactionError::Corrupt("backend fingerprint is not a sha256 checksum")); - } if self.remote_object != canonical_transition_remote_object(self.deployment_id, &self.source.bucket, self.transaction_id, self.write_id)? { @@ -479,7 +496,7 @@ pub struct TransitionCleanupProof { pub write_id: Uuid, pub remote_object: String, pub remote_version: TransitionRemoteVersion, - pub backend_fingerprint: String, + pub backend_fingerprint: [u8; 32], pub decision: TransitionCleanupDecision, } @@ -533,6 +550,321 @@ pub fn canonical_transition_remote_object( )) } +pub fn transition_transaction_record_object_name(transaction_id: Uuid) -> Result { + if transaction_id.is_nil() { + return Err(TransitionTransactionError::Corrupt("transaction_id is nil")); + } + let transaction_key = transaction_id.simple().to_string(); + Ok(format!( + "{}/{}/{}/{}.json", + TRANSITION_TRANSACTION_RECORD_PREFIX, + &transaction_key[..2], + &transaction_key[2..4], + transaction_key + )) +} + +pub(crate) async fn save_transition_transaction_record( + api: Arc, + transaction: &TransitionTransaction, +) -> 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)?; + config_boundary::save_config(api, &object, data).await +} + +pub(crate) async fn load_transition_transaction_record( + api: Arc, + transaction_id: Uuid, +) -> EcstoreResult { + 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) +} + +pub(crate) async fn delete_transition_transaction_record(api: Arc, transaction_id: Uuid) -> EcstoreResult<()> { + let object = transition_transaction_record_object_name(transaction_id).map_err(transition_transaction_store_error)?; + match config_boundary::delete_config(api, &object).await { + Ok(()) | Err(Error::ConfigNotFound) => Ok(()), + Err(err) => Err(err), + } +} + +fn transition_transaction_store_error(err: TransitionTransactionError) -> Error { + Error::other(err) +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TransitionTransactionRecoveryStats { + pub scanned: usize, + pub recovered: usize, + pub retained: usize, + pub failed: usize, + pub next_marker: Option, + pub truncated: bool, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TransitionTransactionRecoveryOutcome { + RemoteCandidateDeleted, + RecordDeleted, + Retained, +} + +fn transition_transaction_id_from_record_object_name(object: &str) -> Result { + let prefix = format!("{TRANSITION_TRANSACTION_RECORD_PREFIX}/"); + let suffix = object + .strip_prefix(&prefix) + .ok_or(TransitionTransactionError::Corrupt("transaction record path has wrong prefix"))?; + let file_name = suffix + .rsplit('/') + .next() + .ok_or(TransitionTransactionError::Corrupt("transaction record path is incomplete"))?; + let transaction_key = file_name + .strip_suffix(".json") + .ok_or(TransitionTransactionError::Corrupt("transaction record path has wrong suffix"))?; + if transaction_key.len() != 32 || !transaction_key.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return Err(TransitionTransactionError::Corrupt("transaction record path has invalid transaction id")); + } + Uuid::parse_str(transaction_key).map_err(|_| TransitionTransactionError::Corrupt("transaction record path has invalid uuid")) +} + +pub async fn process_transition_transaction_record( + api: Arc, + transaction: &TransitionTransaction, +) -> EcstoreResult { + transaction.validate().map_err(transition_transaction_store_error)?; + match transaction.state { + TransitionTransactionState::Uploaded => { + delete_transition_remote_candidate(api.clone(), transaction).await?; + delete_transition_transaction_record(api, transaction.transaction_id).await?; + Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted) + } + TransitionTransactionState::LocalCommitStarted if local_commit_matches_transaction(api.clone(), transaction).await? => { + delete_transition_transaction_record(api, transaction.transaction_id).await?; + Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) + } + TransitionTransactionState::AbortedNoRemote | TransitionTransactionState::Committed => { + delete_transition_transaction_record(api, transaction.transaction_id).await?; + Ok(TransitionTransactionRecoveryOutcome::RecordDeleted) + } + TransitionTransactionState::UploadStarted + | TransitionTransactionState::UploadOutcomeUnknown + | TransitionTransactionState::LocalCommitStarted + | TransitionTransactionState::CleanupPending => Ok(TransitionTransactionRecoveryOutcome::Retained), + } +} + +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 object = api + .get_object_info(&transaction.source.bucket, &transaction.source.object, &opts) + .await?; + let transitioned = &object.transitioned_object; + Ok(transitioned.status == TRANSITION_COMPLETE + && transitioned.name == transaction.remote_object + && transitioned.tier == transaction.tier_name + && transitioned.version_id == transaction.remote_version.tier_delete_version_id().unwrap_or_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; + delete_object_from_remote_tier_idempotent_with_manager_and_identity( + &transaction.remote_object, + version_id, + &transaction.tier_name, + transaction.backend_fingerprint, + &api.tier_config_mgr(), + version_id_exact, + ) + .await + .map(|_| ()) + .map_err(Error::other) +} + +pub async fn recover_transition_transaction_records( + api: Arc, + limit: usize, + marker: Option, +) -> EcstoreResult { + if limit == 0 { + return Err(Error::other("transition transaction recovery limit must be greater than zero")); + } + + let list_limit = i32::try_from(limit).map_or(i32::MAX, |value| value); + let list = api + .clone() + .list_objects_v2( + RUSTFS_META_BUCKET, + TRANSITION_TRANSACTION_RECORD_PREFIX, + marker.clone(), + None, + list_limit, + false, + None, + false, + ) + .await?; + + let mut stats = TransitionTransactionRecoveryStats { + scanned: 0, + recovered: 0, + retained: 0, + failed: 0, + next_marker: list.next_continuation_token, + truncated: list.is_truncated, + }; + + for object in list.objects { + stats.scanned += 1; + let transaction_id = match transition_transaction_id_from_record_object_name(&object.name) { + Ok(transaction_id) => transaction_id, + Err(err) => { + stats.failed += 1; + warn!( + event = EVENT_LIFECYCLE_TRANSITION_TRANSACTION_RECOVERY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + transaction_record = %object.name, + error = ?err, + "Failed to derive transition transaction id from record path" + ); + continue; + } + }; + let transaction = match load_transition_transaction_record(api.clone(), transaction_id).await { + Ok(transaction) => transaction, + Err(Error::ConfigNotFound) => continue, + Err(err) => { + stats.failed += 1; + warn!( + event = EVENT_LIFECYCLE_TRANSITION_TRANSACTION_RECOVERY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + transaction_record = %object.name, + transaction_id = %transaction_id, + error = ?err, + "Failed to load transition transaction record" + ); + continue; + } + }; + + match process_transition_transaction_record(api.clone(), &transaction).await { + Ok( + TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted + | TransitionTransactionRecoveryOutcome::RecordDeleted, + ) => { + stats.recovered += 1; + } + Ok(TransitionTransactionRecoveryOutcome::Retained) => { + stats.retained += 1; + debug!( + event = EVENT_LIFECYCLE_TRANSITION_TRANSACTION_RECOVERY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + transaction_record = %object.name, + transaction_id = %transaction.transaction_id, + state = ?transaction.state, + "Transition transaction recovery retained record for a later reconcile pass" + ); + } + Err(err) => { + stats.failed += 1; + debug!( + event = EVENT_LIFECYCLE_TRANSITION_TRANSACTION_RECOVERY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + transaction_record = %object.name, + transaction_id = %transaction.transaction_id, + state = ?transaction.state, + error = ?err, + "Transition transaction recovery will retry later" + ); + } + } + } + + Ok(stats) +} + +pub async fn run_transition_transaction_recovery_loop(api: Arc, cancel_token: CancellationToken) { + let mut interval = tokio::time::interval(TRANSITION_TRANSACTION_RECOVERY_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + let mut marker: Option = None; + + loop { + tokio::select! { + biased; + _ = cancel_token.cancelled() => return, + _ = interval.tick() => {}, + } + + let recovery = + recover_transition_transaction_records(api.clone(), DEFAULT_TRANSITION_TRANSACTION_RECOVERY_LIMIT, marker.clone()); + let Some(result) = + await_transition_transaction_recovery(&cancel_token, TRANSITION_TRANSACTION_RECOVERY_TIMEOUT, recovery).await + else { + return; + }; + match result { + Ok(stats) => { + marker = stats.next_marker; + debug!( + event = EVENT_LIFECYCLE_TRANSITION_TRANSACTION_RECOVERY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + scanned = stats.scanned, + recovered = stats.recovered, + retained = stats.retained, + failed = stats.failed, + truncated = stats.truncated, + next_marker = ?marker, + "Recovered transition transaction records" + ); + } + Err(err) => { + warn!( + event = EVENT_LIFECYCLE_TRANSITION_TRANSACTION_RECOVERY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + next_marker = ?marker, + error = ?err, + "Failed to recover transition transaction records" + ); + } + } + } +} + +async fn await_transition_transaction_recovery( + cancel_token: &CancellationToken, + timeout: Duration, + recovery: F, +) -> Option> +where + F: Future>, +{ + tokio::select! { + _ = cancel_token.cancelled() => None, + result = tokio::time::timeout(timeout, recovery) => Some(match result { + Ok(result) => result, + Err(_) => Err(Error::other(format!( + "transition transaction recovery timed out after {} seconds", + timeout.as_secs() + ))), + }), + } +} + fn state_change_allowed(from: TransitionTransactionState, to: TransitionTransactionState) -> bool { matches!( (from, to), @@ -564,7 +896,7 @@ mod tests { use super::*; - const BACKEND_FINGERPRINT: &str = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + const BACKEND_FINGERPRINT: [u8; 32] = [7; 32]; #[derive(Default)] struct MemoryTransactionStore { @@ -633,7 +965,7 @@ mod tests { write_id: Uuid::new_v4(), source: source_identity(TransitionSourceVersionMode::Versioned), tier_name: "warm-tier".to_string(), - backend_fingerprint: BACKEND_FINGERPRINT.to_string(), + backend_fingerprint: BACKEND_FINGERPRINT, not_after_unix_nanos: 1_780_000_000_000_000_000, }) .expect("valid transaction should be created") @@ -656,7 +988,7 @@ mod tests { write_id: transaction.write_id, remote_object: transaction.remote_object.clone(), remote_version: transaction.remote_version.clone(), - backend_fingerprint: transaction.backend_fingerprint.clone(), + backend_fingerprint: transaction.backend_fingerprint, decision, } } @@ -735,7 +1067,7 @@ mod tests { write_id: Uuid::new_v4(), source: source_identity(TransitionSourceVersionMode::Versioned), tier_name: "warm-tier".to_string(), - backend_fingerprint: BACKEND_FINGERPRINT.to_string(), + backend_fingerprint: BACKEND_FINGERPRINT, not_after_unix_nanos: 1, }), Err(TransitionTransactionError::Corrupt("deployment_id is nil")) @@ -751,7 +1083,7 @@ mod tests { write_id: Uuid::new_v4(), source, tier_name: "warm-tier".to_string(), - backend_fingerprint: BACKEND_FINGERPRINT.to_string(), + backend_fingerprint: BACKEND_FINGERPRINT, not_after_unix_nanos: 1, }), Err(TransitionTransactionError::Corrupt("source version_id is nil")) @@ -767,7 +1099,7 @@ mod tests { write_id: Uuid::new_v4(), source, tier_name: "warm-tier".to_string(), - backend_fingerprint: BACKEND_FINGERPRINT.to_string(), + backend_fingerprint: BACKEND_FINGERPRINT, not_after_unix_nanos: 1, }), Err(TransitionTransactionError::Corrupt("non-versioned source mode must not carry version_id")) @@ -781,7 +1113,7 @@ mod tests { write_id: Uuid::new_v4(), source: source_identity(TransitionSourceVersionMode::VersionSuspended), tier_name: "warm-tier".to_string(), - backend_fingerprint: BACKEND_FINGERPRINT.to_string(), + backend_fingerprint: BACKEND_FINGERPRINT, not_after_unix_nanos: 1, }) .is_ok() @@ -922,4 +1254,19 @@ mod tests { Err(TransitionTransactionError::Corrupt("encoded transaction exceeds maximum size")) )); } + + #[test] + fn record_object_name_is_stable_and_sanitized() { + let transaction_id = Uuid::parse_str("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee").expect("test uuid should parse"); + + let object = transition_transaction_record_object_name(transaction_id).expect("record object should build"); + + assert_eq!(object, "ilm/transition-transactions/records/aa/aa/aaaaaaaabbbbccccddddeeeeeeeeeeee.json"); + let file_name = object.rsplit('/').next().expect("record object should have file name"); + assert!(!file_name.contains('-')); + assert!(matches!( + transition_transaction_record_object_name(Uuid::nil()), + Err(TransitionTransactionError::Corrupt("transaction_id is nil")) + )); + } } diff --git a/crates/ecstore/src/runtime/instance.rs b/crates/ecstore/src/runtime/instance.rs index a695cea6f..2c8ab72c6 100644 --- a/crates/ecstore/src/runtime/instance.rs +++ b/crates/ecstore/src/runtime/instance.rs @@ -160,6 +160,7 @@ pub struct InstanceContext { /// Replaces the process-global cancel-token static. background_cancel_token: OnceLock, tier_delete_journal_recovery_stores: std::sync::Mutex>, + transition_transaction_recovery_stores: std::sync::Mutex>, #[cfg(test)] tier_delete_journal_recovery_wakeup: tokio::sync::Notify, } @@ -197,6 +198,7 @@ impl InstanceContext { bucket_metadata_sys: std::sync::Mutex::new(None), background_cancel_token: OnceLock::new(), tier_delete_journal_recovery_stores: std::sync::Mutex::new(HashSet::new()), + transition_transaction_recovery_stores: std::sync::Mutex::new(HashSet::new()), #[cfg(test)] tier_delete_journal_recovery_wakeup: tokio::sync::Notify::new(), } @@ -370,6 +372,13 @@ impl InstanceContext { .insert(store_id) } + pub(crate) fn mark_transition_transaction_recovery_started(&self, store_id: Uuid) -> bool { + self.transition_transaction_recovery_stores + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(store_id) + } + #[cfg(test)] pub(crate) fn wake_tier_delete_journal_recovery(&self) { self.tier_delete_journal_recovery_wakeup.notify_one(); @@ -438,8 +447,14 @@ impl std::fmt::Debug for InstanceContext { &self .tier_delete_journal_recovery_stores .lock() - .map(|stores| stores.len()) - .unwrap_or_default(), + .map_or(0, |stores| stores.len()), + ) + .field( + "transition_transaction_recovery_store_count", + &self + .transition_transaction_recovery_stores + .lock() + .map_or(0, |stores| stores.len()), ) .finish_non_exhaustive() } diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index ce66badd6..d23ee4d7d 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -24,6 +24,11 @@ use super::super::*; use crate::bucket::lifecycle::{ tier_delete_journal::{persist_tier_delete_journal_entry, remove_tier_delete_journal_entry}, tier_sweeper::{Jentry, RemoteTierDeleteOutcome, delete_object_from_remote_tier_with_lease_idempotent}, + transition_transaction::{ + TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction, + TransitionTransactionInit, TransitionTransactionState, delete_transition_transaction_record, + save_transition_transaction_record, + }, }; use crate::disk::OldCurrentSize; use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppressed}; @@ -1662,6 +1667,10 @@ async fn transition_cleanup_store(ctx: &Arc) -> Option> { if let Some(api) = runtime_sources::object_store_handle().filter(|api| Arc::ptr_eq(&api.ctx, ctx)) { return Some(api); } @@ -1670,6 +1679,117 @@ async fn transition_cleanup_store(ctx: &Arc Result { + if let Some(deployment_id) = ctx.deployment_id() { + return Ok(deployment_id); + } + #[cfg(test)] + { + Ok(Uuid::new_v4()) + } + #[cfg(not(test))] + { + Err(Error::other("transition transaction requires initialized deployment id")) + } +} + +fn transition_transaction_not_after_unix_nanos() -> Result { + let not_after = (OffsetDateTime::now_utc() + time::Duration::days(7)).unix_timestamp_nanos(); + i64::try_from(not_after).map_err(|_| Error::other("transition transaction deadline timestamp overflow")) +} + +fn transition_source_version_mode(opts: &ObjectOptions) -> TransitionSourceVersionMode { + if opts.versioned { + TransitionSourceVersionMode::Versioned + } else if opts.version_suspended { + TransitionSourceVersionMode::VersionSuspended + } else { + TransitionSourceVersionMode::Unversioned + } +} + +fn transition_source_identity( + bucket: &str, + object: &str, + fi: &FileInfo, + opts: &ObjectOptions, + stored_etag: &str, +) -> Result { + let version_mode = transition_source_version_mode(opts); + let mod_time = fi + .mod_time + .ok_or_else(|| Error::other("transition source identity requires mod_time"))? + .unix_timestamp_nanos(); + let mod_time_unix_nanos = + i64::try_from(mod_time).map_err(|_| Error::other("transition source mod_time timestamp overflow"))?; + let version_id = (version_mode == TransitionSourceVersionMode::Versioned) + .then_some(fi.version_id) + .flatten(); + let data_dir = fi + .data_dir + .ok_or_else(|| Error::other("transition source identity requires data_dir"))?; + Ok(TransitionSourceIdentity { + bucket: bucket.to_string(), + object: object.to_string(), + version_id, + data_dir, + mod_time_unix_nanos, + size: fi.size, + etag: stored_etag.to_string(), + version_mode, + }) +} + +async fn save_transition_transaction_if_available(api: Option<&Arc>, transaction: &TransitionTransaction) -> Result<()> { + if let Some(api) = api { + return save_transition_transaction_record(api.clone(), transaction).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, + next: TransitionTransactionState, + remote_version: Option, +) -> Result<()> { + transaction + .advance(transaction.fence(), next, remote_version) + .map_err(Error::other)?; + save_transition_transaction_if_available(api, transaction).await +} + +async fn delete_transition_transaction_if_available(api: Option<&Arc>, transaction_id: Uuid) -> Result<()> { + if let Some(api) = api { + return delete_transition_transaction_record(api.clone(), transaction_id).await; + } + Ok(()) +} + +async fn delete_transition_transaction_after_remote_cleanup( + api: Option<&Arc>, + transaction_id: Uuid, + bucket: &str, + object: &str, +) { + if let Err(err) = delete_transition_transaction_if_available(api, transaction_id).await { + warn!( + bucket = bucket, + object = object, + transaction_id = %transaction_id, + error = ?err, + "transition remote candidate was cleaned but transaction record cleanup failed" + ); + } +} + #[cfg(feature = "test-util")] #[derive(Default)] struct TransitionCleanupStoreBarrierState { @@ -3138,13 +3258,22 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } }*/ - let dest_obj = gen_transition_objname(bucket); - if let Err(err) = dest_obj { - return Err(to_object_err(err, vec![])); - } - let dest_obj = dest_obj?; - let oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended); + let transaction_api = transition_object_store(&self.ctx).await; + let mut transaction = TransitionTransaction::new(TransitionTransactionInit { + deployment_id: transition_deployment_id(&self.ctx)?, + transaction_id: Uuid::new_v4(), + owner_epoch: Uuid::new_v4(), + write_id: Uuid::new_v4(), + source: transition_source_identity(bucket, object, &fi, opts, &stored_etag)?, + tier_name: opts.transition.tier.clone(), + backend_fingerprint: tgt_client.backend_identity(), + not_after_unix_nanos: transition_transaction_not_after_unix_nanos()?, + }) + .map_err(Error::other)?; + save_transition_transaction_if_available(transaction_api.as_ref(), &transaction).await?; + let transaction_id = transaction.transaction_id; + let dest_obj = transaction.remote_object.clone(); let mut transition_meta = (*oi.user_defined).clone(); transition_meta.insert("name".to_string(), object.to_string()); @@ -3233,6 +3362,8 @@ 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) + .await; } return Err(failure.error); } @@ -3245,12 +3376,33 @@ 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; return Err(err.into()); } + if let Err(err) = advance_and_save_transition_transaction( + transaction_api.as_ref(), + &mut transaction, + TransitionTransactionState::Uploaded, + Some(TransitionRemoteVersion::known_from_put_response(candidate.remote_version().to_string())), + ) + .await + { + let cleanup_api = transition_cleanup_store(&self.ctx).await; + if let Err(cleanup_err) = upload_cleanup.cleanup_rejected_upload(cleanup_api).await { + return Err(StorageError::Io(std::io::Error::other(format!( + "{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; + return Err(err); + } let transition_version_id = match parse_transition_version_id(candidate.remote_version()) { Ok(version_id) => version_id, Err(err) => { - let _cleanup_result = upload_cleanup.cleanup().await; + if upload_cleanup.cleanup().await.is_ok() { + delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) + .await; + } return Err(err.into()); } }; @@ -3264,7 +3416,15 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { match self.acquire_write_lock_diag("transition_object_commit", bucket, object).await { Ok(guard) => Some(guard), Err(err) => { - let _cleanup_result = upload_cleanup.cleanup().await; + if upload_cleanup.cleanup().await.is_ok() { + delete_transition_transaction_after_remote_cleanup( + transaction_api.as_ref(), + transaction_id, + bucket, + object, + ) + .await; + } return Err(err); } } @@ -3275,7 +3435,10 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { Ok(current) => current, Err(err) => { drop(transition_lock_guard); - let _cleanup_result = upload_cleanup.cleanup().await; + if upload_cleanup.cleanup().await.is_ok() { + delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) + .await; + } return Err(err); } }; @@ -3287,7 +3450,10 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { if current_fi.transition_status == TRANSITION_COMPLETE || !source_matches { let already_transitioned = current_fi.transition_status == TRANSITION_COMPLETE; drop(transition_lock_guard); - let _cleanup_result = upload_cleanup.cleanup().await; + if upload_cleanup.cleanup().await.is_ok() { + delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) + .await; + } if already_transitioned { return Ok(()); } @@ -3310,7 +3476,10 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { pause_transition_commit(bucket, object, TransitionCommitPause::BeforeLockLost).await; if transition_lock_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) { drop(transition_lock_guard); - let _cleanup_result = upload_cleanup.cleanup().await; + if upload_cleanup.cleanup().await.is_ok() { + delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) + .await; + } return Err(StorageError::NamespaceLockQuorumUnavailable { mode: "transition_object_commit", bucket: bucket.to_string(), @@ -3323,11 +3492,29 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { pause_transition_commit(bucket, object, TransitionCommitPause::BeforeLeaseValidation).await; if !upload_cleanup.lease.is_current_generation() { drop(transition_lock_guard); - let _cleanup_result = upload_cleanup.cleanup().await; + if upload_cleanup.cleanup().await.is_ok() { + delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object) + .await; + } return Err(Error::other("remote tier configuration changed during transition")); } #[cfg(test)] pause_transition_commit(bucket, object, TransitionCommitPause::AfterLeaseValidation).await; + if let Err(err) = advance_and_save_transition_transaction( + transaction_api.as_ref(), + &mut transaction, + TransitionTransactionState::LocalCommitStarted, + None, + ) + .await + { + 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; + } + return Err(err); + } upload_cleanup.disarm(); if let Err(err) = self.delete_object_version(bucket, object, &fi, false).await { warn!( @@ -3340,6 +3527,38 @@ 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 + { + warn!( + bucket = bucket, + object = object, + transaction_id = %transaction_id, + error = ?err, + "transition committed locally but transaction cleanup failed" + ); + } + } + Err(err) => { + warn!( + bucket = bucket, + object = object, + transaction_id = %transaction_id, + error = ?err, + "transition committed locally but transaction committed-state advance failed" + ); + } + } // delete_object_version persisted transition_status=complete and freed the // local data, but does not touch the GET metadata cache. Drop any cached @@ -5349,6 +5568,54 @@ mod transition_upload_integrity_tests { } } + #[tokio::test] + #[serial_test::serial] + async fn transition_uses_transaction_canonical_remote_object_name() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "transition-transaction-remote-object-bucket"; + let object = "object.bin"; + let payload = b"transition remote object must be bound to its transaction id".repeat(1024); + let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await; + let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); + let remote_version = Uuid::new_v4().to_string(); + let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; + backend.set_put_remote_version(Some(remote_version.clone())).await; + + set_disks + .transition_object(bucket, object, &transition_options(&original, tier_name)) + .await + .expect("transition should commit"); + + let put_versions = backend.put_versions().await; + assert_eq!(put_versions.len(), 1, "transition should upload one remote candidate"); + let remote_object = &put_versions[0].0; + assert!( + remote_object.starts_with(crate::bucket::lifecycle::transition_transaction::TRANSITION_TRANSACTION_PREFIX), + "remote object should be transaction-scoped: {remote_object}" + ); + let (fi, _, _) = set_disks + .get_object_fileinfo( + bucket, + object, + &ObjectOptions { + no_lock: true, + metadata_cache_safe: false, + ..Default::default() + }, + true, + false, + ) + .await + .expect("committed transition metadata should be readable"); + assert_eq!(fi.transition_status, TRANSITION_COMPLETE); + assert_eq!(fi.transitioned_objname, *remote_object); + assert_eq!( + fi.transition_version_id, + Some(Uuid::parse_str(&remote_version).expect("test version id should parse")) + ); + assert!(backend.contains(remote_object).await, "committed remote object should remain available"); + } + async fn corrupt_beyond_read_quorum( temp_dirs: &[tempfile::TempDir], bucket: &str, diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 2c6869eb5..051dea66e 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -534,12 +534,19 @@ mod tests { TIER_DELETE_JOURNAL_PREFIX, persist_tier_delete_journal_entry, recover_tier_delete_journal_entries, }, tier_sweeper::Jentry, + transition_transaction::{ + TRANSITION_TRANSACTION_RECORD_PREFIX, TransitionRemoteVersion, TransitionSourceIdentity, + TransitionSourceVersionMode, TransitionTransaction, TransitionTransactionInit, TransitionTransactionState, + recover_transition_transaction_records, save_transition_transaction_record, + }, }, + client::transition_api::ReaderImpl, disk::RUSTFS_META_BUCKET, runtime::{global::set_object_store_resolver, sources as runtime_sources}, services::tier::{ test_util::{MockWarmBackend, TransitionCleanupStoreBarrier, register_mock_tier}, tier::TierConfigMgr, + warm_backend::WarmBackend, }, storage_api_contracts::{ bucket::{BucketOperations as _, MakeBucketOptions}, @@ -989,6 +996,25 @@ mod tests { .len() } + #[cfg(feature = "test-util")] + async fn transition_transaction_record_count(store: Arc) -> usize { + store + .list_objects_v2( + RUSTFS_META_BUCKET, + TRANSITION_TRANSACTION_RECORD_PREFIX, + None, + None, + 100, + false, + None, + false, + ) + .await + .expect("transition transaction records should be listable") + .objects + .len() + } + #[cfg(feature = "test-util")] async fn wait_for_tier_delete_journal_recovery( store: Arc, @@ -1326,4 +1352,190 @@ mod tests { ); assert!(!Arc::ptr_eq(&ctx_a, &ctx_b), "the regression requires two distinct instance contexts"); } + + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn transition_transaction_recovery_deletes_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; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + + 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 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: Some(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: TransitionSourceVersionMode::Versioned, + }, + 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::Uploaded, + 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 + .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.remove_versions().await, + vec![(transaction.remote_object.clone(), remote_version)], + "recovery must delete the exact uploaded candidate" + ); + assert_eq!(backend.exact_remove_count(), 1); + assert_eq!(backend.object_count().await, 0); + } + + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn transition_transaction_recovery_drops_record_after_confirmed_local_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-committed", &[4])).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + + let tier_name = "TXCOMMITTED"; + 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 bucket = "transition-transaction-committed-bucket"; + let object = "object.bin"; + store + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("bucket should be created"); + let mut reader = PutObjReader::from_vec(b"confirmed local commit record cleanup".repeat(1024)); + let original = store + .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .await + .expect("source object should be written"); + let opts = ObjectOptions { + no_lock: true, + transition: TransitionOptions { + status: TRANSITION_PENDING.to_string(), + tier: tier_name.to_string(), + etag: original.etag.clone().expect("source object should have etag"), + ..Default::default() + }, + mod_time: original.mod_time, + ..Default::default() + }; + store + .transition_object(bucket, object, &opts) + .await + .expect("transition should commit"); + let committed = store + .get_object_info( + bucket, + object, + &ObjectOptions { + no_lock: true, + metadata_cache_safe: false, + ..Default::default() + }, + ) + .await + .expect("committed object info should be readable"); + let mut remote_parts = committed.transitioned_object.name.rsplit('/'); + let write_id = uuid::Uuid::parse_str(remote_parts.next().expect("remote object should contain write id")) + .expect("write id should parse"); + let transaction_id = uuid::Uuid::parse_str(remote_parts.next().expect("remote object should contain transaction id")) + .expect("transaction id should parse"); + let mut transaction = TransitionTransaction::new(TransitionTransactionInit { + deployment_id: ctx.deployment_id().expect("test store should initialize deployment id"), + transaction_id, + owner_epoch: uuid::Uuid::new_v4(), + write_id, + source: TransitionSourceIdentity { + bucket: bucket.to_string(), + object: object.to_string(), + version_id: None, + data_dir: uuid::Uuid::new_v4(), + mod_time_unix_nanos: original + .mod_time + .expect("source object should have mod_time") + .unix_timestamp_nanos() + .try_into() + .expect("test timestamp should fit i64"), + size: original.size, + etag: original.etag.expect("source object should have etag"), + 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::Uploaded, + Some(TransitionRemoteVersion::known_from_put_response( + committed.transitioned_object.version_id.clone(), + )), + ) + .expect("transaction should enter uploaded state"); + transaction + .advance(transaction.fence(), TransitionTransactionState::LocalCommitStarted, None) + .expect("transaction should enter local commit state"); + save_transition_transaction_record(store.clone(), &transaction) + .await + .expect("transaction record should persist"); + assert_eq!(transition_transaction_record_count(store.clone()).await, 1); + + 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, + 1, + "confirmed local commit recovery must not delete the committed remote body" + ); + assert_eq!(backend.remove_count().await, 0); + } }