diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 75890bb15..bd48d3ff1 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -5928,6 +5928,8 @@ mod tests { backend_identity: Some([1; 32]), version_id_exact: true, version_state: rustfs_filemeta::TransitionVersionState::Exact, + state: crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::Committed, + source: None, }; let err = state @@ -6040,6 +6042,8 @@ mod tests { backend_identity: Some([1; 32]), version_id_exact: true, version_state: rustfs_filemeta::TransitionVersionState::Exact, + state: crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::Committed, + source: None, }; state @@ -10146,6 +10150,8 @@ mod tests { backend_identity: Some(identity), version_id_exact: false, version_state: rustfs_filemeta::TransitionVersionState::Unknown, + state: crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::Committed, + source: None, }; let err = crate::bucket::lifecycle::tier_delete_journal::process_tier_delete_journal_entry(ecstore, &je) @@ -10185,6 +10191,8 @@ mod tests { backend_identity: Some(identity), version_id_exact: true, version_state: rustfs_filemeta::TransitionVersionState::Exact, + state: crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::Committed, + source: None, }; crate::set_disk::cleanup_rejected_transition_upload_durably( diff --git a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs index a217d4978..6542e88b8 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs @@ -20,8 +20,10 @@ use tokio_util::sync::CancellationToken; use tracing::{debug, warn}; use crate::bucket::lifecycle::config_boundary; +use crate::bucket::lifecycle::runtime_boundary; use crate::bucket::lifecycle::tier_sweeper::{ - Jentry, delete_confirmed_transition_candidate_exact_with_manager_and_identity, + Jentry, TierDeleteJournalState, TierDeleteSourceIdentity, + 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; @@ -30,7 +32,7 @@ use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader use crate::services::tier::tier::tier_destination_id_from_metadata; use crate::storage_api_contracts::{ list::ListOperations as _, - object::{DeletedObject, ObjectIO, ObjectOperations, ObjectToDelete}, + object::{DeletedObject, HTTPPreconditions, ObjectIO, ObjectOperations, ObjectToDelete}, range::HTTPRangeSpec, }; use crate::store::ECStore; @@ -46,6 +48,7 @@ const TIER_DELETE_JOURNAL_RECOVERY_TIMEOUT: Duration = Duration::from_secs(300); const TIER_DELETE_JOURNAL_VERSION: u8 = 2; const TIER_DELETE_JOURNAL_EXACT_VERSION: u8 = 3; const TIER_DELETE_JOURNAL_STATE_VERSION: u8 = 4; +const TIER_DELETE_JOURNAL_TRANSACTION_VERSION: u8 = 5; pub(crate) const TIER_DELETE_JOURNAL_PREFIX: &str = "ilm/tier-delete-journal/"; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -61,13 +64,22 @@ struct PersistedTierDeleteJournalEntry { version_id_exact: Option, #[serde(default, skip_serializing_if = "Option::is_none")] version_state: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + state: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + source: Option, } impl PersistedTierDeleteJournalEntry { fn from_jentry(je: &Jentry) -> Result { validate_version_state(je.version_state, &je.version_id, je.version_id_exact)?; let legacy_unknown = je.version_state == rustfs_filemeta::TransitionVersionState::Unknown; - let version = if legacy_unknown { + let version = if je.source.is_some() || je.state == TierDeleteJournalState::Prepared { + if je.backend_identity.is_none() { + return Err(Error::other("tier delete transaction is missing its backend identity")); + } + TIER_DELETE_JOURNAL_TRANSACTION_VERSION + } else if legacy_unknown { if je.backend_identity.is_some() { TIER_DELETE_JOURNAL_VERSION } else { @@ -87,6 +99,10 @@ impl PersistedTierDeleteJournalEntry { backend_identity: je.backend_identity, version_id_exact: je.version_id_exact.then_some(true), version_state: (!legacy_unknown).then_some(je.version_state), + state: (version == TIER_DELETE_JOURNAL_TRANSACTION_VERSION).then_some(je.state), + source: (version == TIER_DELETE_JOURNAL_TRANSACTION_VERSION) + .then(|| je.source.clone()) + .flatten(), }) } @@ -101,14 +117,21 @@ impl PersistedTierDeleteJournalEntry { } if self.version != TIER_DELETE_JOURNAL_EXACT_VERSION && self.version != TIER_DELETE_JOURNAL_STATE_VERSION + && self.version != TIER_DELETE_JOURNAL_TRANSACTION_VERSION && self.version_id_exact.unwrap_or(false) { return Err(Error::other( "legacy tier delete journal entry has an unsupported exact version constraint", )); } - let (backend_identity, version_id_exact, version_state) = match self.version { - 1 => (None, false, rustfs_filemeta::TransitionVersionState::Unknown), + let (backend_identity, version_id_exact, version_state, state, source) = match self.version { + 1 => ( + None, + false, + rustfs_filemeta::TransitionVersionState::Unknown, + TierDeleteJournalState::Committed, + None, + ), TIER_DELETE_JOURNAL_VERSION => ( Some( self.backend_identity @@ -116,6 +139,8 @@ impl PersistedTierDeleteJournalEntry { ), false, rustfs_filemeta::TransitionVersionState::Unknown, + TierDeleteJournalState::Committed, + None, ), TIER_DELETE_JOURNAL_EXACT_VERSION => { if self.version_id.is_empty() || self.version_id_exact != Some(true) { @@ -128,6 +153,8 @@ impl PersistedTierDeleteJournalEntry { ), true, rustfs_filemeta::TransitionVersionState::Exact, + TierDeleteJournalState::Committed, + None, ) } TIER_DELETE_JOURNAL_STATE_VERSION => { @@ -143,6 +170,31 @@ impl PersistedTierDeleteJournalEntry { ), exact, state, + TierDeleteJournalState::Committed, + None, + ) + } + TIER_DELETE_JOURNAL_TRANSACTION_VERSION => { + let state = self + .state + .ok_or_else(|| Error::other("tier delete journal v5 entry is missing its state"))?; + let source = self + .source + .ok_or_else(|| Error::other("tier delete journal v5 entry is missing its source identity"))?; + let exact = self.version_id_exact.unwrap_or(false); + let version_state = self + .version_state + .ok_or_else(|| Error::other("tier delete journal v5 entry is missing its version state"))?; + validate_version_state(version_state, &self.version_id, exact)?; + ( + Some( + self.backend_identity + .ok_or_else(|| Error::other("tier delete journal v5 entry is missing its backend identity"))?, + ), + exact, + version_state, + state, + Some(source), ) } version => return Err(Error::other(format!("unsupported tier delete journal version {version}"))), @@ -154,6 +206,8 @@ impl PersistedTierDeleteJournalEntry { backend_identity, version_id_exact, version_state, + state, + source, }) } } @@ -201,6 +255,20 @@ pub(crate) fn tier_delete_journal_object_name(je: &Jentry) -> String { hasher.update([0]); hasher.update(b"exact-version-id"); } + if let Some(source) = &je.source { + hasher.update([0]); + hasher.update(source.bucket.as_bytes()); + hasher.update([0]); + hasher.update(source.object.as_bytes()); + hasher.update([0]); + hasher.update(source.version_id.as_deref().unwrap_or_default().as_bytes()); + hasher.update([0]); + hasher.update(source.data_dir.as_deref().unwrap_or_default().as_bytes()); + hasher.update([0]); + hasher.update(source.etag.as_deref().unwrap_or_default().as_bytes()); + hasher.update([0]); + hasher.update(source.mod_time.as_deref().unwrap_or_default().as_bytes()); + } format!( "{TIER_DELETE_JOURNAL_PREFIX}{}.json", rustfs_utils::crypto::hex(hasher.finalize().as_slice()) @@ -246,6 +314,66 @@ where .map_err(std::io::Error::other) } +pub async fn commit_tier_delete_journal_entry(api: Arc, je: &Jentry) -> std::io::Result<()> +where + S: ObjectIO< + Error = Error, + RangeSpec = HTTPRangeSpec, + HeaderMap = http::HeaderMap, + ObjectOptions = ObjectOptions, + ObjectInfo = ObjectInfo, + GetObjectReader = GetObjectReader, + PutObjectReader = PutObjReader, + >, +{ + let mut committed = je.clone(); + committed.state = TierDeleteJournalState::Committed; + persist_tier_delete_journal_entry(api, &committed).await +} + +pub async fn abort_tier_delete_journal_entry(api: Arc, je: &Jentry) -> std::io::Result<()> +where + S: ObjectOperations< + Error = Error, + ObjectInfo = ObjectInfo, + ObjectOptions = ObjectOptions, + FileInfo = FileInfo, + ObjectToDelete = ObjectToDelete, + DeletedObject = DeletedObject, + >, +{ + remove_tier_delete_journal_entry(api, je).await +} + +pub async fn abort_prepared_tier_delete_journal_entry(api: Arc, je: &Jentry) -> std::io::Result<()> { + let name = tier_delete_journal_object_name(je); + let (data, metadata) = match config_boundary::read_config_with_metadata(api.clone(), &name, &ObjectOptions::default()).await { + Ok(result) => result, + Err(Error::ConfigNotFound) | Err(Error::FileNotFound) => return Ok(()), + Err(err) => return Err(std::io::Error::other(err)), + }; + let current = decode_tier_delete_journal_entry(&data).map_err(std::io::Error::other)?; + if current.state != TierDeleteJournalState::Prepared { + return Ok(()); + } + let etag = metadata + .etag + .ok_or_else(|| std::io::Error::other("prepared tier delete journal has no entity tag"))?; + match config_boundary::delete_config_if_match(api, &name, &etag).await { + Ok(()) | Err(Error::ConfigNotFound) => Ok(()), + Err(Error::PreconditionFailed) => Err(std::io::Error::new( + std::io::ErrorKind::WouldBlock, + "prepared tier delete journal changed before abort", + )), + Err(err) => Err(std::io::Error::other(err)), + } +} + +pub(crate) async fn enqueue_committed_tier_delete_journal_entry(je: &Jentry) -> std::io::Result<()> { + let expiry_state = runtime_boundary::expiry_state_handle(); + expiry_state.write().await.enqueue_tier_journal_entry(je) +} + pub async fn remove_tier_delete_journal_entry(api: Arc, je: &Jentry) -> std::io::Result<()> where S: ObjectOperations< @@ -264,6 +392,13 @@ where } pub async fn process_tier_delete_journal_entry(api: Arc, je: &Jentry) -> std::io::Result<()> { + if je.state == TierDeleteJournalState::Prepared { + return reconcile_prepared_tier_delete_journal_entry(api, je).await; + } + process_committed_tier_delete_journal_entry(api, je).await +} + +async fn process_committed_tier_delete_journal_entry(api: Arc, je: &Jentry) -> std::io::Result<()> { if je.version_state == rustfs_filemeta::TransitionVersionState::Unknown { return Err(std::io::Error::new( std::io::ErrorKind::InvalidData, @@ -296,6 +431,87 @@ pub async fn process_tier_delete_journal_entry(api: Arc, je: &Jentry) - remove_tier_delete_journal_entry(api, je).await } +async fn reconcile_prepared_tier_delete_journal_entry(api: Arc, je: &Jentry) -> std::io::Result<()> { + let (data, metadata) = + config_boundary::read_config_with_metadata(api.clone(), &tier_delete_journal_object_name(je), &ObjectOptions::default()) + .await + .map_err(std::io::Error::other)?; + let current = decode_tier_delete_journal_entry(&data).map_err(std::io::Error::other)?; + if current.state != TierDeleteJournalState::Prepared { + return Err(std::io::Error::new( + std::io::ErrorKind::WouldBlock, + "prepared tier delete journal changed before reconciliation", + )); + } + let Some(etag) = metadata.etag else { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "prepared tier delete journal has no entity tag", + )); + }; + let source = je + .source + .as_ref() + .ok_or_else(|| std::io::Error::new(std::io::ErrorKind::InvalidData, "prepared tier delete journal has no source"))?; + match api + .get_object_info(&source.bucket, &source.object, &source.lookup_options()) + .await + { + Ok(info) if source.matches(&info) => { + match config_boundary::delete_config_if_match(api, &tier_delete_journal_object_name(¤t), &etag).await { + Ok(()) => Ok(()), + Err(Error::PreconditionFailed) => Err(std::io::Error::new( + std::io::ErrorKind::WouldBlock, + "prepared tier delete journal changed before abort", + )), + Err(err) => Err(std::io::Error::other(err)), + } + } + Ok(_info) if source.has_stable_identity() => { + commit_prepared_tier_delete_journal_entry_if_current(api, current, etag).await + } + Ok(_) => Err(std::io::Error::new( + std::io::ErrorKind::WouldBlock, + "prepared tier delete journal source identity is not sufficient to confirm deletion", + )), + Err(Error::ObjectNotFound(_, _)) | Err(Error::FileNotFound) | Err(Error::FileVersionNotFound) => { + commit_prepared_tier_delete_journal_entry_if_current(api, current, etag).await + } + Err(err) => Err(std::io::Error::other(err)), + } +} + +async fn commit_prepared_tier_delete_journal_entry_if_current( + api: Arc, + mut committed: Jentry, + etag: String, +) -> std::io::Result<()> { + committed.state = TierDeleteJournalState::Committed; + let data = encode_tier_delete_journal_entry(&committed).map_err(std::io::Error::other)?; + match config_boundary::save_config_with_opts( + api.clone(), + &tier_delete_journal_object_name(&committed), + data, + &ObjectOptions { + max_parity: true, + http_preconditions: Some(HTTPPreconditions { + if_match: Some(etag), + ..Default::default() + }), + ..Default::default() + }, + ) + .await + { + Ok(()) => process_committed_tier_delete_journal_entry(api, &committed).await, + Err(Error::PreconditionFailed) => Err(std::io::Error::new( + std::io::ErrorKind::WouldBlock, + "prepared tier delete journal changed before commit", + )), + Err(err) => Err(std::io::Error::other(err)), + } +} + pub async fn recover_tier_delete_journal_entries( api: Arc, limit: usize, @@ -482,10 +698,13 @@ mod tests { decode_tier_delete_journal_entry, encode_tier_delete_journal_entry, record_tier_delete_journal_backend_identity, tier_delete_journal_object_name, }; - use crate::bucket::lifecycle::tier_sweeper::Jentry; + use crate::bucket::lifecycle::tier_sweeper::{Jentry, TierDeleteJournalState, TierDeleteSourceIdentity}; use crate::error::Result; + use crate::object_api::ObjectInfo; use std::time::Duration; + use time::OffsetDateTime; use tokio_util::sync::CancellationToken; + use uuid::Uuid; fn journal_entry() -> Jentry { Jentry { @@ -495,6 +714,8 @@ mod tests { backend_identity: Some([7; 32]), version_id_exact: true, version_state: rustfs_filemeta::TransitionVersionState::Exact, + state: TierDeleteJournalState::Committed, + source: None, } } @@ -513,6 +734,55 @@ mod tests { assert_eq!(decoded.version_state, je.version_state); } + #[test] + fn tier_delete_transaction_roundtrips_prepared_source_identity() { + let mut je = journal_entry(); + je.state = TierDeleteJournalState::Prepared; + je.source = Some(TierDeleteSourceIdentity { + bucket: "bucket".to_string(), + object: "object".to_string(), + version_id: Some("version".to_string()), + versioned: true, + version_suspended: false, + data_dir: Some("data-dir".to_string()), + etag: Some("etag".to_string()), + mod_time: Some("mod-time".to_string()), + }); + + let encoded = encode_tier_delete_journal_entry(&je).expect("prepared transaction should encode"); + let value: serde_json::Value = serde_json::from_slice(&encoded).expect("transaction should be JSON"); + assert_eq!(value["version"], serde_json::json!(5)); + assert_eq!(value["state"], serde_json::json!("Prepared")); + assert!(value["source"].is_object()); + + let decoded = decode_tier_delete_journal_entry(&encoded).expect("prepared transaction should decode"); + assert_eq!(decoded.state, TierDeleteJournalState::Prepared); + assert_eq!(decoded.source, je.source); + } + + #[test] + fn tier_delete_source_identity_rejects_recreated_object() { + let version_id = Uuid::from_u128(1); + let data_dir = Uuid::from_u128(2); + let mod_time = OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(1); + let info = ObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + version_id: Some(version_id), + data_dir: Some(data_dir), + mod_time: Some(mod_time), + ..Default::default() + }; + let source = TierDeleteSourceIdentity::from_object_info("bucket", "object", &info, true, false); + assert!(source.matches(&info)); + + let recreated = ObjectInfo { + data_dir: Some(Uuid::from_u128(3)), + ..info + }; + assert!(!source.matches(&recreated)); + } + #[test] fn tier_delete_journal_roundtrips_exact_put_response_constraint() { let mut exact = journal_entry(); diff --git a/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs b/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs index c4ca3b809..a934cdb11 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs @@ -23,10 +23,12 @@ use crate::bucket::lifecycle::bucket_lifecycle_ops::ExpiryOp; use crate::bucket::lifecycle::lifecycle::{self, ObjectOpts}; use crate::bucket::lifecycle::tier_delete_journal::persist_tier_delete_journal_entry; use crate::client::signer_error::error_chain_contains_signer_header_marker; +use crate::object_api::ObjectInfo; use crate::services::tier::tier::{TierConfigMgr, TierDestinationId, TierOperationLease}; use crate::storage_api_contracts::lifecycle::TransitionedObject; use crate::store::ECStore; use rustfs_utils::get_env_usize; +use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use std::any::Any; use std::collections::VecDeque; @@ -257,6 +259,8 @@ impl ObjSweeper { rustfs_filemeta::TransitionVersionState::SuspendedNull | rustfs_filemeta::TransitionVersionState::Exact ), version_state: self.transition_version_state, + state: TierDeleteJournalState::Committed, + source: None, }); } None @@ -285,6 +289,76 @@ impl ObjSweeper { } } +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) enum TierDeleteJournalState { + Prepared, + Committed, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub(crate) struct TierDeleteSourceIdentity { + pub(crate) bucket: String, + pub(crate) object: String, + pub(crate) version_id: Option, + pub(crate) versioned: bool, + pub(crate) version_suspended: bool, + pub(crate) data_dir: Option, + pub(crate) etag: Option, + pub(crate) mod_time: Option, +} + +impl TierDeleteSourceIdentity { + pub(crate) fn from_object_info( + bucket: &str, + object: &str, + info: &ObjectInfo, + versioned: bool, + version_suspended: bool, + ) -> Self { + Self { + bucket: bucket.to_string(), + object: object.to_string(), + version_id: info.version_id.map(|id| id.to_string()), + versioned, + version_suspended, + data_dir: info.data_dir.map(|id| id.to_string()), + etag: info.etag.clone(), + mod_time: info.mod_time.map(|time| time.to_string()), + } + } + + pub(crate) fn lookup_options(&self) -> crate::object_api::ObjectOptions { + crate::object_api::ObjectOptions { + version_id: self.version_id.clone(), + versioned: self.versioned, + version_suspended: self.version_suspended, + ..Default::default() + } + } + + pub(crate) fn matches(&self, info: &ObjectInfo) -> bool { + if self.bucket != info.bucket { + return false; + } + if let Some(version_id) = &self.version_id { + return info.version_id.map(|id| id.to_string()).as_deref() == Some(version_id.as_str()) + && self.data_dir == info.data_dir.map(|id| id.to_string()); + } + if self.data_dir.is_some() { + return self.data_dir == info.data_dir.map(|id| id.to_string()); + } + self.etag.is_some() + && self.etag == info.etag + && self.mod_time.is_some() + && self.mod_time == info.mod_time.map(|time| time.to_string()) + } + + pub(crate) fn has_stable_identity(&self) -> bool { + self.version_id.is_some() || self.data_dir.is_some() || (self.etag.is_some() && self.mod_time.is_some()) + } +} + #[derive(Debug, Clone)] #[allow(unused_assignments)] pub struct Jentry { @@ -294,6 +368,8 @@ pub struct Jentry { pub(crate) backend_identity: Option, pub(crate) version_id_exact: bool, pub(crate) version_state: rustfs_filemeta::TransitionVersionState, + pub(crate) state: TierDeleteJournalState, + pub(crate) source: Option, } impl ExpiryOp for Jentry { @@ -554,9 +630,48 @@ pub fn transitioned_force_delete_journal_entry( rustfs_filemeta::TransitionVersionState::SuspendedNull | rustfs_filemeta::TransitionVersionState::Exact ), version_state: transition_version_state, + state: TierDeleteJournalState::Committed, + source: None, }) } +pub(crate) fn attach_tier_delete_source( + je: &mut Jentry, + bucket: &str, + object: &str, + info: &ObjectInfo, + versioned: bool, + version_suspended: bool, +) { + je.state = TierDeleteJournalState::Prepared; + je.source = Some(TierDeleteSourceIdentity::from_object_info( + bucket, + object, + info, + versioned, + version_suspended, + )); +} + +pub(crate) fn transitioned_delete_journal_entry_for_source( + version_id: Option, + versioned: bool, + suspended: bool, + bucket: &str, + object: &str, + source: &ObjectInfo, +) -> Option { + let mut je = transitioned_delete_journal_entry( + version_id, + versioned, + suspended, + &source.transitioned_object, + source.transition_version_state, + )?; + attach_tier_delete_source(&mut je, bucket, object, source, versioned, suspended); + Some(je) +} + #[cfg(test)] mod test { use crate::client::signer_error::invalid_utf8_header_error; diff --git a/crates/ecstore/src/core/sets.rs b/crates/ecstore/src/core/sets.rs index 1f580b3fb..c2f0f7749 100644 --- a/crates/ecstore/src/core/sets.rs +++ b/crates/ecstore/src/core/sets.rs @@ -339,7 +339,9 @@ impl Sets { futures.push(set.delete_object(bucket, object, opt.clone())); } - let _results = join_all(futures).await; + if let Some(err) = join_all(futures).await.into_iter().find_map(Result::err) { + return Err(err); + } Ok(()) } diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index 89f2f3204..b2fcc0469 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -112,6 +112,9 @@ pub struct ObjectOptions { pub want_checksum: Option, pub skip_verify_bitrot: bool, pub capacity_scope_token: Option, + /// Storage-owned journal writer used by the atomic delete path. This is + /// populated only by the `ECStore` wrapper that holds the namespace locks. + pub tier_delete_journal_api: Option>, } impl ObjectOptions { diff --git a/crates/ecstore/src/services/tier/tier.rs b/crates/ecstore/src/services/tier/tier.rs index f60999573..f6c9114f2 100644 --- a/crates/ecstore/src/services/tier/tier.rs +++ b/crates/ecstore/src/services/tier/tier.rs @@ -9383,6 +9383,8 @@ mod tests { backend_identity: Some(current_identity), version_id_exact: true, version_state: rustfs_filemeta::TransitionVersionState::Exact, + state: crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::Committed, + source: None, }; journal_store .insert_config_object( diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 133a0bc60..be4a27264 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -24,10 +24,14 @@ use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks use crate::set_disk::read::GetObjectDownstreamWriter; use crate::bucket::lifecycle::{ - tier_delete_journal::{persist_tier_delete_journal_entry, remove_tier_delete_journal_entry}, + tier_delete_journal::{ + enqueue_committed_tier_delete_journal_entry, persist_tier_delete_journal_entry, + record_tier_delete_journal_backend_identity, remove_tier_delete_journal_entry, + }, tier_sweeper::{ - Jentry, RemoteTierDeleteOutcome, delete_confirmed_transition_candidate_exact_with_lease_idempotent, - delete_object_from_remote_tier_with_lease_idempotent, + Jentry, RemoteTierDeleteOutcome, TierDeleteJournalState, + delete_confirmed_transition_candidate_exact_with_lease_idempotent, delete_object_from_remote_tier_with_lease_idempotent, + transitioned_delete_journal_entry_for_source, }, transition_transaction::{ TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction, @@ -1851,6 +1855,8 @@ pub(crate) async fn cleanup_rejected_transition_upload_durably( } else { rustfs_filemeta::TransitionVersionState::Exact }, + state: TierDeleteJournalState::Committed, + source: None, }; let journal_error = if let Some(api) = api.as_ref() { @@ -3208,6 +3214,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { object_lock_delete_check_required(metadata_sys::get_in(&self.ctx, bucket).await.ok().as_deref()); let mut vers_map: HashMap<&String, FileInfoVersions> = HashMap::new(); + let mut journal_entries: Vec<(usize, Jentry)> = Vec::new(); for (i, dobj) in objects.iter().enumerate() { if del_errs[i].is_some() { @@ -3232,7 +3239,8 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let marker_delete = dobj.version_id.is_none() || dobj.synthetic_version_id; let replication_needs_source = replicate_delete && (!marker_delete || delete_config_snapshot.active_delete_marker_rules_require_tags(&replication_object_name)); - let (goi, gerr) = if object_lock_checks_required || replication_needs_source { + let (goi, gerr) = if object_lock_checks_required || replication_needs_source || opts.tier_delete_journal_api.is_some() + { let (goi, _write_quorum, gerr) = self.get_object_info_and_quorum(bucket, &dobj.object_name, &check_opts).await; (goi, gerr) } else { @@ -3241,8 +3249,14 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { let source_missing = gerr .as_ref() .is_some_and(|err| is_err_object_not_found(err) || is_err_version_not_found(err)); + let explicit_delete_marker = opts.tier_delete_journal_api.is_some() + && dobj.version_id.is_some() + && goi.delete_marker + && goi.version_id == version_id + && matches!(gerr.as_ref(), Some(StorageError::MethodNotAllowed)); if let Some(err) = gerr.as_ref() && !source_missing + && !explicit_delete_marker { del_errs[i] = Some(err.clone()); continue; @@ -3255,6 +3269,23 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { continue; } + if opts.tier_delete_journal_api.is_some() + && let Some(mut je) = transitioned_delete_journal_entry_for_source( + version_id, + versioned, + version_suspended, + bucket, + &replication_object_name, + &goi, + ) + { + if let Err(err) = record_tier_delete_journal_backend_identity(&mut je, &goi.user_defined) { + del_errs[i] = Some(Error::other(err)); + continue; + } + journal_entries.push((i, je)); + } + let mut admitted = dobj.clone(); admitted.object_name = replication_object_name; if admitted.synthetic_version_id { @@ -3300,6 +3331,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { } } + if goi.delete_marker && dobj.version_id.is_some() && goi.version_id == version_id { + vr.deleted = true; + vr.mod_time = goi.mod_time; + } + let v = { if vers_map.contains_key(&dobj.object_name) { let val = vers_map.get_mut(&dobj.object_name).unwrap(); @@ -3372,6 +3408,23 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { return (del_objects, del_errs); } + let mut persisted_journal_entries = Vec::with_capacity(journal_entries.len()); + if let Some(api) = opts.tier_delete_journal_api.as_ref() { + for (idx, mut je) in journal_entries { + if let Err(err) = persist_tier_delete_journal_entry(Arc::clone(api), &je).await { + del_errs[idx] = Some(Error::other(err)); + continue; + } + je.state = TierDeleteJournalState::Prepared; + persisted_journal_entries.push((idx, je)); + } + } + + for fi_vers in &mut vers { + fi_vers.versions.retain(|fi| del_errs[fi.idx].is_none()); + } + vers.retain(|fi_vers| !fi_vers.versions.is_empty()); + let rollback_dir = Uuid::new_v4(); let disks = self.disks.read().await; @@ -3535,6 +3588,37 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { // TODO: add_partial + if let Some(api) = opts.tier_delete_journal_api.as_ref() { + for (idx, je) in persisted_journal_entries { + if del_errs[idx].is_none() { + let mut committed = je; + committed.state = TierDeleteJournalState::Committed; + if let Err(err) = persist_tier_delete_journal_entry(Arc::clone(api), &committed).await { + warn!( + object = %committed.obj_name, + tier = %committed.tier_name, + error = ?err, + "batch tier delete committed locally but journal commit failed; recovery will retry" + ); + } else if let Err(err) = enqueue_committed_tier_delete_journal_entry(&committed).await { + warn!( + object = %committed.obj_name, + tier = %committed.tier_name, + error = ?err, + "batch tier delete journal committed but could not be queued; recovery will retry" + ); + } + } else if let Err(err) = remove_tier_delete_journal_entry(Arc::clone(api), &je).await { + warn!( + object = %je.obj_name, + tier = %je.tier_name, + error = ?err, + "failed to remove aborted batch tier delete journal" + ); + } + } + } + if dist_erasure { self.release_dist_delete_object_locks_batch(dist_batch_lock_ids).await; } diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index c89320908..f8b1e2c26 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -1379,6 +1379,8 @@ mod tests { backend_identity: Some(identity_a), version_id_exact: true, version_state: rustfs_filemeta::TransitionVersionState::Exact, + state: crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::Committed, + source: None, }; let entry_b = Jentry { obj_name: "remote-b".to_string(), @@ -1387,6 +1389,8 @@ mod tests { backend_identity: Some(identity_b), version_id_exact: true, version_state: rustfs_filemeta::TransitionVersionState::Exact, + state: crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::Committed, + source: None, }; let remove_a = backend_a.arm_failing_remove_barrier().await; persist_tier_delete_journal_entry(store_a.clone(), &entry_a) diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 6ffed22b7..fb74c86db 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -13,6 +13,16 @@ // limitations under the License. use super::*; +use crate::bucket::lifecycle::{ + tier_delete_journal::{ + abort_prepared_tier_delete_journal_entry as abort_prepared_journal_entry_if_current, commit_tier_delete_journal_entry, + enqueue_committed_tier_delete_journal_entry, persist_tier_delete_journal_entry, + record_tier_delete_journal_backend_identity, + }, + tier_sweeper::{ + Jentry, attach_tier_delete_source, transitioned_delete_journal_entry_for_source, transitioned_force_delete_journal_entry, + }, +}; use crate::bucket::replication::ReplicationObjectBridge; use crate::disk::OldCurrentSize; use crate::object_api::DeleteLockFence; @@ -33,6 +43,181 @@ use std::{ }; use tokio::io::{AsyncRead, ReadBuf}; +const FORCE_DELETE_LIST_PAGE_SIZE: i32 = 1_000; + +fn build_tier_delete_journal_entry( + bucket: &str, + object: &str, + opts: &ObjectOptions, + source: &ObjectInfo, +) -> Result> { + let version_id = opts.version_id.as_deref().map(Uuid::parse_str).transpose()?; + let source_object = decode_dir_object(object); + let Some(mut je) = (if opts.delete_prefix { + transitioned_force_delete_journal_entry(&source.transitioned_object, source.transition_version_state).map(|mut je| { + attach_tier_delete_source(&mut je, bucket, source_object.as_str(), source, opts.versioned, opts.version_suspended); + je + }) + } else { + transitioned_delete_journal_entry_for_source( + version_id, + opts.versioned, + opts.version_suspended, + bucket, + source_object.as_str(), + source, + ) + }) else { + return Ok(None); + }; + record_tier_delete_journal_backend_identity(&mut je, &source.user_defined).map_err(Error::other)?; + Ok(Some(je)) +} + +async fn prepare_tier_delete_journal_entry( + api: &Arc, + bucket: &str, + object: &str, + opts: &ObjectOptions, + source: &ObjectInfo, +) -> Result> { + let Some(je) = build_tier_delete_journal_entry(bucket, object, opts, source)? else { + return Ok(None); + }; + persist_tier_delete_journal_entry(Arc::clone(api), &je) + .await + .map_err(Error::other)?; + Ok(Some(je)) +} + +async fn abort_prepared_tier_delete_journal_entry(api: &Arc, je: &Jentry) { + if let Err(err) = abort_prepared_journal_entry_if_current(Arc::clone(api), je).await { + warn!( + object = %je.obj_name, + tier = %je.tier_name, + error = ?err, + "failed to remove aborted tier delete journal" + ); + } +} + +async fn abort_prepared_tier_delete_journal_entries(api: &Arc, entries: &[Jentry]) { + for entry in entries { + abort_prepared_tier_delete_journal_entry(api, entry).await; + } +} + +async fn commit_prepared_tier_delete_journal_entry(api: &Arc, je: &Jentry) { + if let Err(err) = commit_tier_delete_journal_entry(Arc::clone(api), je).await { + warn!( + object = %je.obj_name, + tier = %je.tier_name, + error = ?err, + "tier delete committed locally but journal commit failed; recovery will retry" + ); + return; + } + let mut committed = je.clone(); + committed.state = crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState::Committed; + if let Err(err) = enqueue_committed_tier_delete_journal_entry(&committed).await { + warn!( + object = %je.obj_name, + tier = %je.tier_name, + error = ?err, + "tier delete journal committed but could not be queued; recovery will retry" + ); + } +} + +async fn commit_prepared_tier_delete_journal_entries(api: &Arc, entries: &[Jentry]) { + for entry in entries { + commit_prepared_tier_delete_journal_entry(api, entry).await; + } +} + +async fn prepare_prefix_tier_delete_journal_entries( + api: &Arc, + bucket: &str, + prefix: &str, + opts: &ObjectOptions, +) -> Result> { + let mut marker = None; + let mut version_marker = None; + let mut entries = Vec::new(); + + loop { + let page = Arc::clone(api) + .list_object_versions_for_lifecycle( + bucket, + prefix, + marker.clone(), + version_marker.clone(), + None, + FORCE_DELETE_LIST_PAGE_SIZE, + ) + .await?; + + for source in page.objects { + if let Some(entry) = build_tier_delete_journal_entry(bucket, &source.name, opts, &source)? { + entries.push(entry); + } + } + + if !page.is_truncated { + break; + } + + let next_marker = page + .next_marker + .ok_or_else(|| Error::other("truncated force delete listing has no next marker"))?; + let next_version_marker = page.next_version_idmarker; + if marker.as_deref() == Some(next_marker.as_str()) && version_marker == next_version_marker { + return Err(Error::other("force delete listing marker did not advance")); + } + marker = Some(next_marker); + version_marker = next_version_marker; + } + + let mut persisted = Vec::with_capacity(entries.len()); + for entry in entries { + if let Err(err) = persist_tier_delete_journal_entry(Arc::clone(api), &entry).await { + abort_prepared_tier_delete_journal_entries(api, &persisted).await; + return Err(Error::other(err)); + } + persisted.push(entry); + } + Ok(persisted) +} +async fn delete_prefix_with_tier_delete_journal( + store: &ECStore, + bucket: &str, + object: &str, + opts: &ObjectOptions, + tier_journal_api: Option<&Arc>, +) -> Result<()> { + let journal_entry = if let Some(api) = tier_journal_api { + Some(prepare_prefix_tier_delete_journal_entries(api, bucket, object, opts).await?) + } else { + None + }; + + let result = store.delete_prefix(bucket, object, opts).await; + match result { + Ok(()) => { + if let (Some(api), Some(entries)) = (tier_journal_api, journal_entry.as_ref()) { + commit_prepared_tier_delete_journal_entries(api, entries).await; + } + Ok(()) + } + Err(err) => { + if let (Some(api), Some(entries)) = (tier_journal_api, journal_entry.as_ref()) { + abort_prepared_tier_delete_journal_entries(api, entries).await; + } + Err(err) + } + } +} + /// A GET whose object identity has been resolved while its namespace read lock /// remains held, but whose body reader has not been constructed yet. /// @@ -1092,8 +1277,49 @@ impl ECStore { purged } + pub async fn delete_object_with_tier_delete_journal( + self: &Arc, + bucket: &str, + object: &str, + opts: ObjectOptions, + ) -> Result { + let result = self + .handle_delete_object_with_journal(bucket, object, opts, Some(Arc::clone(self))) + .await; + if result.is_ok() { + list_objects::observe_list_objects_mutation(self, bucket).await; + } + result + } + + pub async fn delete_objects_with_tier_delete_journal( + self: &Arc, + bucket: &str, + objects: Vec, + opts: ObjectOptions, + ) -> (Vec, Vec>) { + let result = self + .handle_delete_objects_with_journal(bucket, objects, opts, Some(Arc::clone(self))) + .await; + let success_count = result.1.iter().filter(|err| err.is_none()).count(); + if success_count > 0 { + list_objects::observe_list_objects_mutations(self, bucket, success_count).await; + } + result + } + #[instrument(skip(self))] pub(super) async fn handle_delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result { + self.handle_delete_object_with_journal(bucket, object, opts, None).await + } + + pub(super) async fn handle_delete_object_with_journal( + &self, + bucket: &str, + object: &str, + opts: ObjectOptions, + tier_journal_api: Option>, + ) -> Result { check_del_obj_args(bucket, object)?; let object = if opts.delete_prefix && !opts.delete_prefix_object { @@ -1103,11 +1329,12 @@ impl ECStore { }; let object = object.as_str(); let mut opts = opts; + opts.tier_delete_journal_api = tier_journal_api.clone(); if opts.delete_prefix && !opts.delete_prefix_object { // Prefix deletes cover multiple object keys; an exact lock on the prefix string // would not protect child objects. - self.delete_prefix(bucket, object, &opts).await?; + delete_prefix_with_tier_delete_journal(self, bucket, object, &opts, tier_journal_api.as_ref()).await?; return Ok(ObjectInfo::default()); } @@ -1119,7 +1346,7 @@ impl ECStore { }; if opts.delete_prefix { - self.delete_prefix(bucket, object, &opts).await?; + delete_prefix_with_tier_delete_journal(self, bucket, object, &opts, tier_journal_api.as_ref()).await?; return Ok(ObjectInfo::default()); } @@ -1215,8 +1442,25 @@ impl ECStore { )); } + let journal_entry = if let Some(api) = tier_journal_api.as_ref() { + prepare_tier_delete_journal_entry(api, bucket, object, &opts, &pinfo.object_info).await? + } else { + None + }; + if !errs.is_empty() && !opts.versioned && !opts.version_suspended { - let mut obj = self.delete_object_from_all_pools(bucket, object, &opts, errs).await?; + let mut obj = match self.delete_object_from_all_pools(bucket, object, &opts, errs).await { + Ok(obj) => obj, + Err(err) => { + if let (Some(api), Some(je)) = (tier_journal_api.as_ref(), journal_entry.as_ref()) { + abort_prepared_tier_delete_journal_entry(api, je).await; + } + return Err(err); + } + }; + if let (Some(api), Some(je)) = (tier_journal_api.as_ref(), journal_entry.as_ref()) { + commit_prepared_tier_delete_journal_entry(api, je).await; + } obj.name = decode_dir_object(object); return Ok(obj); } @@ -1224,6 +1468,9 @@ impl ECStore { for pool in self.pools.iter() { match pool.delete_object(bucket, object, opts.clone()).await { Ok(res) => { + if let (Some(api), Some(je)) = (tier_journal_api.as_ref(), journal_entry.as_ref()) { + commit_prepared_tier_delete_journal_entry(api, je).await; + } let mut obj = res; obj.name = decode_dir_object(object); return Ok(obj); @@ -1236,6 +1483,10 @@ impl ECStore { } } + if let (Some(api), Some(je)) = (tier_journal_api.as_ref(), journal_entry.as_ref()) { + abort_prepared_tier_delete_journal_entry(api, je).await; + } + if let Some(ver) = opts.version_id { return Err(StorageError::VersionNotFound(bucket.to_owned(), object.to_owned(), ver)); } @@ -1249,6 +1500,16 @@ impl ECStore { bucket: &str, objects: Vec, opts: ObjectOptions, + ) -> (Vec, Vec>) { + self.handle_delete_objects_with_journal(bucket, objects, opts, None).await + } + + pub(super) async fn handle_delete_objects_with_journal( + &self, + bucket: &str, + objects: Vec, + opts: ObjectOptions, + tier_journal_api: Option>, ) -> (Vec, Vec>) { // encode object name let objects: Vec = objects @@ -1269,6 +1530,7 @@ impl ECStore { } let mut opts = opts; + opts.tier_delete_journal_api = tier_journal_api; if opts.delete_replication_config_snapshot.is_none() { match ReplicationObjectBridge::delete_request_config_in(&self.ctx, bucket).await { Ok(snapshot) => opts.delete_replication_config_snapshot = Some(Arc::new(snapshot)), @@ -1625,6 +1887,7 @@ impl ECStore { mod tests { use super::*; use crate::bucket::lifecycle::core::TRANSITION_COMPLETE; + use crate::bucket::lifecycle::tier_sweeper::TierDeleteJournalState; use crate::bucket::replication::{ ReplicationState, ReplicationStatusType, VersionPurgeStatusType, replication_state_to_filemeta, replication_statuses_map, version_purge_statuses_map, @@ -1640,6 +1903,7 @@ mod tests { }; use crate::set_disk::SetDisks; use crate::storage_api_contracts::bucket::MakeBucketOptions; + use crate::storage_api_contracts::lifecycle::TransitionedObject; use bytes::Bytes; use std::io::Cursor; use std::sync::Arc; @@ -1660,6 +1924,55 @@ mod tests { struct BodyCacheHookGuard; + #[test] + fn tier_delete_entry_is_prepared_and_bound_to_source_generation() { + let identity = [9_u8; 32]; + let mut metadata = HashMap::new(); + rustfs_utils::http::metadata_compat::insert_str( + &mut metadata, + rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID, + rustfs_utils::crypto::hex(identity), + ); + let version_id = Uuid::from_u128(1); + let data_dir = Uuid::from_u128(2); + let source = ObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + version_id: Some(version_id), + data_dir: Some(data_dir), + user_defined: Arc::new(metadata), + transitioned_object: TransitionedObject { + name: "remote/object".to_string(), + version_id: "remote-version".to_string(), + tier: "WARM".to_string(), + status: TRANSITION_COMPLETE.to_string(), + ..Default::default() + }, + transition_version_state: rustfs_filemeta::TransitionVersionState::Exact, + ..Default::default() + }; + let entry = build_tier_delete_journal_entry( + "bucket", + "object", + &ObjectOptions { + version_id: Some(version_id.to_string()), + versioned: true, + ..Default::default() + }, + &source, + ) + .expect("transition source should produce a journal entry") + .expect("completed transition should be journaled"); + + assert_eq!(entry.state, TierDeleteJournalState::Prepared); + assert_eq!(entry.backend_identity, Some(identity)); + let data_dir_string = data_dir.to_string(); + assert_eq!( + entry.source.as_ref().and_then(|source| source.data_dir.as_deref()), + Some(data_dir_string.as_str()) + ); + } + impl Drop for BodyCacheHookGuard { fn drop(&mut self) { clear_get_object_body_cache_hook(); diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 9f9f88481..7e6d3c99e 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -32,7 +32,6 @@ use super::storage_api::object_usecase::bucket::{ bucket_lifecycle_audit::LcEventSrc, bucket_lifecycle_ops::{enqueue_transition_immediate, post_restore_opts}, lifecycle::{self, TransitionOptions}, - tier_delete_journal, tier_sweeper, }, metadata_sys, object_lock::{ @@ -108,8 +107,8 @@ use super::storage_api::object_usecase::{ validate_ssec_for_read, wrap_response_with_cors, }; use crate::app::runtime_sources::{ - AppContext, current_app_context, current_expiry_state_handle, current_notify_interface_for_context, - current_object_data_cache_for_context, current_object_store_handle_for_context, + AppContext, current_app_context, current_notify_interface_for_context, current_object_data_cache_for_context, + current_object_store_handle_for_context, }; use crate::config::RustFSBufferConfig; use crate::delete_tail_activity::{DeleteTailActivityGuard, DeleteTailStage}; @@ -751,53 +750,6 @@ fn tune_reader_stream_buffer_size( selected_size } -async fn enqueue_transitioned_delete_cleanup( - store: Arc, - bucket: &str, - object: &str, - opts: &ObjectOptions, - existing: Option<&ObjectInfo>, -) -> std::io::Result<()> { - let Some(existing) = existing else { - return Ok(()); - }; - let _activity_guard = DeleteTailActivityGuard::new(DeleteTailStage::Cleanup); - - let je = if opts.delete_prefix { - tier_sweeper::transitioned_force_delete_journal_entry(&existing.transitioned_object, existing.transition_version_state) - } else { - let version_id = opts.version_id.as_ref().and_then(|v| Uuid::parse_str(v).ok()); - tier_sweeper::transitioned_delete_journal_entry( - version_id, - opts.versioned, - opts.version_suspended, - &existing.transitioned_object, - existing.transition_version_state, - ) - }; - let Some(mut je) = je else { - return Ok(()); - }; - - tier_delete_journal::record_tier_delete_journal_backend_identity(&mut je, &existing.user_defined)?; - tier_delete_journal::persist_tier_delete_journal_entry(store, &je).await?; - - let expiry_state = current_expiry_state_handle(); - let mut expiry_state = expiry_state.write().await; - if let Err(err) = expiry_state.enqueue_tier_journal_entry(&je) { - warn!( - bucket, - object, - remote_object = %existing.transitioned_object.name, - remote_version_id = %existing.transitioned_object.version_id, - tier = %existing.transitioned_object.tier, - error = ?err, - "transitioned object cleanup journal persisted but was not queued" - ); - } - Ok(()) -} - pin_project! { struct ExtractArchiveEtagReader { #[pin] @@ -6932,7 +6884,7 @@ impl DefaultObjectUsecase { invalidate_object_data_cache_objects_before_mutation(&cache_adapter, &bucket, cache_keys_before_delete.iter()).await; let (dobjs, errs) = store - .delete_objects( + .delete_objects_with_tier_delete_journal( &bucket, object_to_delete.clone(), ObjectOptions { @@ -6969,27 +6921,6 @@ impl DefaultObjectUsecase { { delete_results[didx].delete_object = Some(dobjs[i].clone()); let (versioned, version_suspended) = object_versioning[i]; - if let Err(err) = enqueue_transitioned_delete_cleanup( - store.clone(), - &bucket, - &object_to_delete[i].object_name, - &ObjectOptions { - version_id: object_to_delete[i].version_id.map(|v| v.to_string()), - versioned, - version_suspended, - ..Default::default() - }, - existing_object_infos[i].as_ref(), - ) - .await - { - warn!( - bucket = %bucket, - object = %object_to_delete[i].object_name, - error = ?err, - "failed to persist transitioned object cleanup journal" - ); - } let creates_delete_marker = object_to_delete[i].version_id.is_none() && versioned && !version_suspended; if creates_delete_marker { record_bucket_delete_marker_memory(&bucket).await; @@ -7246,7 +7177,10 @@ impl DefaultObjectUsecase { } let obj_info = { - match store.delete_object(&bucket, &key, opts.clone()).await { + match store + .delete_object_with_tier_delete_journal(&bucket, &key, opts.clone()) + .await + { Ok(obj) => obj, Err(err) => { if is_err_bucket_not_found(&err) { @@ -7263,16 +7197,6 @@ impl DefaultObjectUsecase { } }; - if let Err(err) = - enqueue_transitioned_delete_cleanup(store.clone(), &bucket, &key, &opts, existing_object_info.as_ref()).await - { - warn!( - bucket = %bucket, - object = %key, - error = ?err, - "failed to persist transitioned object cleanup journal" - ); - } if force_delete { let _ = invalidate_object_data_cache_prefix_after_delete(&cache_adapter, &bucket, &key).await; } else { @@ -9864,115 +9788,6 @@ mod tests { (store, context) } - #[tokio::test] - #[serial_test::serial] - async fn transitioned_delete_cleanup_persists_known_state_and_rejects_unknown_state() { - let store = crate::app::gating_test_env::shared_gating_ecstore().await; - if current_app_context().is_none() { - crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await; - } - let identity = [11_u8; 32]; - let mut metadata = HashMap::new(); - rustfs_utils::http::metadata_compat::insert_str( - &mut metadata, - rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID, - rustfs_utils::crypto::hex(identity), - ); - let mut current = ObjectInfo { - user_defined: Arc::new(metadata), - ..Default::default() - }; - current.transitioned_object.status = lifecycle::TRANSITION_COMPLETE.to_string(); - current.transitioned_object.tier = "WARM".to_string(); - current.transitioned_object.name = "remote/identity-bound".to_string(); - current.transitioned_object.version_id = "remote-version".to_string(); - current.transition_version_state = rustfs_filemeta::TransitionVersionState::Exact; - - let journal_name = |remote_object: &str, backend_identity: Option<[u8; 32]>, version_id_exact: bool| { - use sha2::{Digest, Sha256}; - - let mut hasher = Sha256::new(); - hasher.update(b"WARM"); - hasher.update([0]); - hasher.update(remote_object.as_bytes()); - hasher.update([0]); - hasher.update(b"remote-version"); - if let Some(backend_identity) = backend_identity { - hasher.update([0]); - hasher.update(backend_identity); - } - if version_id_exact { - hasher.update([0]); - hasher.update(b"exact-version-id"); - } - format!("ilm/tier-delete-journal/{}.json", rustfs_utils::crypto::hex(hasher.finalize().as_slice())) - }; - - enqueue_transitioned_delete_cleanup(store.clone(), "bucket", "identity-bound", &ObjectOptions::default(), Some(¤t)) - .await - .expect("normal transitioned delete should persist an identity-bound journal"); - let mut identity_bound = store - .get_object_reader( - ".rustfs.sys", - &journal_name("remote/identity-bound", Some(identity), true), - None, - http::HeaderMap::new(), - &ObjectOptions::default(), - ) - .await - .expect("identity-bound journal should be readable"); - let mut identity_bound_data = Vec::new(); - tokio::io::AsyncReadExt::read_to_end(&mut identity_bound.stream, &mut identity_bound_data) - .await - .expect("identity-bound journal body should be readable"); - let identity_bound: serde_json::Value = - serde_json::from_slice(&identity_bound_data).expect("identity-bound journal should decode as JSON"); - assert_eq!(identity_bound["version"], serde_json::json!(4)); - assert_eq!(identity_bound["backend_identity"], serde_json::json!(identity)); - assert_eq!(identity_bound["version_id_exact"], serde_json::json!(true)); - assert_eq!(identity_bound["version_state"], serde_json::json!("exact")); - - current.user_defined = Arc::new(HashMap::new()); - current.transitioned_object.name = "remote/legacy".to_string(); - current.transition_version_state = rustfs_filemeta::TransitionVersionState::Unknown; - enqueue_transitioned_delete_cleanup( - store.clone(), - "bucket", - "legacy", - &ObjectOptions { - delete_prefix: true, - ..Default::default() - }, - Some(¤t), - ) - .await - .expect("unknown force-delete cleanup should fail closed without a journal"); - let legacy_err = match store - .get_object_reader( - ".rustfs.sys", - &journal_name("remote/legacy", None, false), - None, - http::HeaderMap::new(), - &ObjectOptions::default(), - ) - .await - { - Ok(_) => panic!("unknown remote version state must not persist a delete journal"), - Err(err) => err, - }; - assert!( - matches!( - &legacy_err, - StorageError::FileNotFound - | StorageError::ObjectNotFound(_, _) - | StorageError::FileVersionNotFound - | StorageError::VersionNotFound(_, _, _) - | StorageError::VolumeNotFound - ), - "unknown remote version state must leave no journal, got {legacy_err:?}" - ); - } - async fn put_real_cold_fill_object(store: &Arc, bucket: &str, object: &str, body: &[u8]) -> ObjectInfo { let mut reader = PutObjReader::from_vec(body.to_vec()); store diff --git a/rustfs/src/app/runtime_sources.rs b/rustfs/src/app/runtime_sources.rs index 07adaf539..3b2475986 100644 --- a/rustfs/src/app/runtime_sources.rs +++ b/rustfs/src/app/runtime_sources.rs @@ -13,7 +13,6 @@ // limitations under the License. use crate::app::object_data_cache::ObjectDataCacheAdapter; -use crate::app::storage_api::runtime_sources::ExpiryState; #[cfg(test)] use crate::app::storage_api::runtime_sources::TierConfigMgr; use crate::runtime_sources as root_runtime_sources; @@ -24,6 +23,7 @@ pub(crate) use crate::runtime_sources::{ use rustfs_s3select_api::{QueryResult, server::dbms::DatabaseManagerSystem}; use s3s::dto::SelectObjectContentInput; use std::sync::Arc; +#[cfg(test)] use tokio::sync::RwLock; pub(crate) fn current_app_context() -> Option> { @@ -55,10 +55,6 @@ pub(crate) async fn current_s3select_db( .await } -pub(crate) fn current_expiry_state_handle() -> Arc> { - root_runtime_sources::current_expiry_state_handle().unwrap_or_else(ExpiryState::new) -} - #[cfg(test)] pub(crate) fn current_tier_config_handle() -> Arc> { root_runtime_sources::current_tier_config_handle().unwrap_or_else(TierConfigMgr::new) diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 54a5341bf..d487d2701 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -210,7 +210,6 @@ pub(crate) mod runtime { } pub(crate) mod runtime_sources { - pub(crate) type ExpiryState = super::runtime::ExpiryState; #[cfg(test)] pub(crate) type TierConfigMgr = super::runtime::TierConfigMgr; } @@ -373,57 +372,6 @@ pub(crate) mod bucket { } } pub(crate) use lifecycle_contract as lifecycle; - - pub(crate) mod tier_delete_journal { - use std::sync::Arc; - - pub(crate) fn record_tier_delete_journal_backend_identity( - je: &mut super::tier_sweeper::Jentry, - metadata: &std::collections::HashMap, - ) -> std::io::Result<()> { - crate::storage::storage_api::ecstore_bucket::lifecycle::tier_delete_journal::record_tier_delete_journal_backend_identity(je, metadata) - } - - pub(crate) async fn persist_tier_delete_journal_entry( - api: Arc, - je: &super::tier_sweeper::Jentry, - ) -> std::io::Result<()> { - crate::storage::storage_api::ecstore_bucket::lifecycle::tier_delete_journal::persist_tier_delete_journal_entry( - api, je, - ) - .await - } - } - - pub(crate) mod tier_sweeper { - pub(crate) type Jentry = crate::storage::storage_api::ecstore_bucket::lifecycle::tier_sweeper::Jentry; - - pub(crate) fn transitioned_delete_journal_entry( - version_id: Option, - versioned: bool, - suspended: bool, - transitioned: &super::super::super::storage_contracts::TransitionedObject, - transition_version_state: rustfs_filemeta::TransitionVersionState, - ) -> Option { - crate::storage::storage_api::ecstore_bucket::lifecycle::tier_sweeper::transitioned_delete_journal_entry( - version_id, - versioned, - suspended, - transitioned, - transition_version_state, - ) - } - - pub(crate) fn transitioned_force_delete_journal_entry( - transitioned: &super::super::super::storage_contracts::TransitionedObject, - transition_version_state: rustfs_filemeta::TransitionVersionState, - ) -> Option { - crate::storage::storage_api::ecstore_bucket::lifecycle::tier_sweeper::transitioned_force_delete_journal_entry( - transitioned, - transition_version_state, - ) - } - } } pub(crate) mod metadata { diff --git a/rustfs/src/runtime_sources.rs b/rustfs/src/runtime_sources.rs index 90877f48a..830a38e4e 100644 --- a/rustfs/src/runtime_sources.rs +++ b/rustfs/src/runtime_sources.rs @@ -35,7 +35,6 @@ pub(crate) use context::{ resolve_bucket_monitor_handle as current_bucket_monitor_handle, resolve_buffer_config as current_buffer_config, resolve_daily_tier_stats as current_daily_tier_stats, resolve_deployment_id as current_deployment_id, resolve_encryption_service as current_encryption_service, resolve_endpoints_handle as current_endpoints_handle, - resolve_expiry_state_handle as current_expiry_state_handle, resolve_federated_identity_service as current_federated_identity_service, resolve_iam_handle as current_iam_handle, resolve_iam_ready as current_iam_ready, resolve_internode_metrics as current_internode_metrics, resolve_kms_runtime_service_manager as current_kms_runtime_service_manager,