From 28fdcc87be976382e28c23f2044650e3b9a010e9 Mon Sep 17 00:00:00 2001 From: cxymds Date: Mon, 20 Jul 2026 20:54:32 +0800 Subject: [PATCH] fix(tiering): make rejected upload cleanup durable (#5059) * fix(tiering): make rejected upload cleanup durable * fix(tiering): close transition upload cancellation gap * test(tiering): cover failed upload without candidate * test(tiering): synchronize cancelled cleanup recovery * test(tiering): stabilize cancelled cleanup recovery Prefer cancellation when the tier delete journal recovery worker is racing an immediate tick, and build the cancelled-cleanup regression store with an already-cancelled token so production recovery cannot consume the test journal. Co-Authored-By: heihutu --------- Co-authored-by: houseme Co-authored-by: heihutu --- crates/ecstore/src/api/mod.rs | 8 +- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 42 +- .../bucket/lifecycle/tier_delete_journal.rs | 202 ++++++- .../src/bucket/lifecycle/tier_sweeper.rs | 46 +- crates/ecstore/src/bucket/metadata_sys.rs | 4 + crates/ecstore/src/runtime/instance.rs | 67 ++- crates/ecstore/src/services/tier/test_util.rs | 124 ++++- crates/ecstore/src/services/tier/tier.rs | 61 ++ .../ecstore/src/services/tier/warm_backend.rs | 272 ++++++++- crates/ecstore/src/set_disk/mod.rs | 2 + crates/ecstore/src/set_disk/ops/object.rs | 524 ++++++++++++++++-- crates/ecstore/src/store/init.rs | 315 ++++++++++- .../tests/lifecycle_integration_test.rs | 335 ++++++++++- crates/scanner/tests/storage_api/mod.rs | 16 +- 14 files changed, 1914 insertions(+), 104 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 4f85333ff..0b902719d 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -67,6 +67,8 @@ pub mod bucket { } pub mod tier_delete_journal { + #[cfg(feature = "test-util")] + pub use crate::bucket::lifecycle::tier_delete_journal::recover_tier_delete_journal_entries; pub use crate::bucket::lifecycle::tier_delete_journal::{ persist_tier_delete_journal_entry, record_tier_delete_journal_backend_identity, }; @@ -444,9 +446,9 @@ pub mod tier { #[cfg(feature = "test-util")] pub mod test_util { pub use crate::services::tier::test_util::{ - FaultConfig, MockStoredObject, MockWarmBackend, MockWarmOp, TransitionMeta, assert_transition_meta_consistent, - free_version_count, read_transition_meta, register_mock_tier, register_mock_tier_backend, - wait_for_free_version_absence, + FaultConfig, MockStoredObject, MockWarmBackend, MockWarmOp, TransitionCleanupStoreBarrier, TransitionMeta, + assert_transition_meta_consistent, free_version_count, read_transition_meta, register_mock_tier, + register_mock_tier_backend, wait_for_free_version_absence, }; } } diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 3a7916504..566d07c76 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -107,7 +107,6 @@ pub type ExpiryOpType = Box; static XXHASH_SEED: u64 = 0; static TIER_FREE_VERSION_RECOVERY_STARTED: OnceLock<()> = OnceLock::new(); -static TIER_DELETE_JOURNAL_RECOVERY_STARTED: OnceLock<()> = OnceLock::new(); pub const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging"; pub const AMZ_TAG_COUNT: &str = "x-amz-tagging-count"; @@ -419,6 +418,7 @@ async fn delete_free_version_remote_object( &oi.transitioned_object.tier, identity, tier_config_mgr, + false, ) .await?; Ok(()) @@ -1504,12 +1504,20 @@ fn spawn_tier_free_version_recovery_once(api: Arc) { } fn spawn_tier_delete_journal_recovery_once(api: Arc) { - if TIER_DELETE_JOURNAL_RECOVERY_STARTED.set(()).is_err() { + 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, + "Tier delete journal recovery was not started because the store shutdown token is unavailable" + ); + return; + }; + if !api.ctx.mark_tier_delete_journal_recovery_started(api.id) { return; } - tokio::spawn(async move { - let cancel_token = runtime_sources::background_services_cancel_token().unwrap_or_default(); run_tier_delete_journal_recovery_loop(api, cancel_token).await; }); } @@ -1989,7 +1997,7 @@ fn transitioned_cleanup_tuple(oi: &ObjectInfo) -> Result<(&str, &str, &str), std if transitioned.status != lifecycle::TRANSITION_COMPLETE { return Err(std::io::Error::other("transitioned object cleanup tuple is not complete")); } - if transitioned.name.is_empty() || transitioned.version_id.is_empty() || transitioned.tier.is_empty() { + if transitioned.name.is_empty() || transitioned.tier.is_empty() { return Err(std::io::Error::other("transitioned object cleanup tuple is incomplete")); } Ok((&transitioned.name, &transitioned.version_id, &transitioned.tier)) @@ -3576,6 +3584,7 @@ mod tests { version_id: "remote-version".to_string(), tier_name: "WARM".to_string(), backend_identity: Some([1; 32]), + version_id_exact: false, }; let err = state @@ -3664,6 +3673,7 @@ mod tests { version_id: "remote-version".to_string(), tier_name: "WARM".to_string(), backend_identity: Some([1; 32]), + version_id_exact: false, }; state @@ -3814,7 +3824,7 @@ mod tests { } #[test] - fn transitioned_cleanup_tuple_requires_remote_name_version_and_tier() { + fn transitioned_cleanup_tuple_preserves_versioned_remote() { let mut oi = ObjectInfo::default(); oi.transitioned_object.status = crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE.to_string(); oi.transitioned_object.name = "remote/object".to_string(); @@ -3827,15 +3837,29 @@ mod tests { } #[test] - fn transitioned_cleanup_tuple_rejects_missing_remote_version() { + fn transitioned_cleanup_tuple_accepts_unversioned_remote() { let mut oi = ObjectInfo::default(); oi.transitioned_object.status = crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE.to_string(); oi.transitioned_object.name = "remote/object".to_string(); oi.transitioned_object.tier = "WARM".to_string(); - let err = transitioned_cleanup_tuple(&oi).expect_err("missing version must be rejected"); + let tuple = transitioned_cleanup_tuple(&oi).expect("an empty remote version identifies an unversioned tier bucket"); - assert!(err.to_string().contains("cleanup tuple is incomplete")); + assert_eq!(tuple, ("remote/object", "", "WARM")); + } + + #[test] + fn transitioned_cleanup_tuple_rejects_missing_remote_name_or_tier() { + for (name, tier) in [("", "WARM"), ("remote/object", "")] { + let mut oi = ObjectInfo::default(); + oi.transitioned_object.status = crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE.to_string(); + oi.transitioned_object.name = name.to_string(); + oi.transitioned_object.tier = tier.to_string(); + + let err = transitioned_cleanup_tuple(&oi).expect_err("remote name and tier must remain required"); + + assert!(err.to_string().contains("cleanup tuple is incomplete")); + } } #[test] diff --git a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs index 1c6778880..474036ef8 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::sync::Arc; +use std::{future::Future, sync::Arc, time::Duration}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; @@ -38,8 +38,11 @@ const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle"; const EVENT_LIFECYCLE_TIER_DELETE_JOURNAL: &str = "lifecycle_tier_delete_journal"; pub const DEFAULT_TIER_DELETE_JOURNAL_RECOVERY_LIMIT: usize = 1_000; +const TIER_DELETE_JOURNAL_RECOVERY_INTERVAL: Duration = Duration::from_secs(60); +const TIER_DELETE_JOURNAL_RECOVERY_TIMEOUT: Duration = Duration::from_secs(300); const TIER_DELETE_JOURNAL_VERSION: u8 = 2; -const TIER_DELETE_JOURNAL_PREFIX: &str = "ilm/tier-delete-journal/"; +const TIER_DELETE_JOURNAL_EXACT_VERSION: u8 = 3; +pub(crate) const TIER_DELETE_JOURNAL_PREFIX: &str = "ilm/tier-delete-journal/"; #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] @@ -50,12 +53,16 @@ struct PersistedTierDeleteJournalEntry { tier_name: String, #[serde(default)] backend_identity: Option<[u8; 32]>, + #[serde(default, skip_serializing_if = "Option::is_none")] + version_id_exact: Option, } impl PersistedTierDeleteJournalEntry { fn from_jentry(je: &Jentry) -> Self { Self { - version: if je.backend_identity.is_some() { + version: if je.version_id_exact { + TIER_DELETE_JOURNAL_EXACT_VERSION + } else if je.backend_identity.is_some() { TIER_DELETE_JOURNAL_VERSION } else { 1 @@ -64,6 +71,7 @@ impl PersistedTierDeleteJournalEntry { version_id: je.version_id.clone(), tier_name: je.tier_name.clone(), backend_identity: je.backend_identity, + version_id_exact: je.version_id_exact.then_some(true), } } @@ -76,12 +84,32 @@ impl PersistedTierDeleteJournalEntry { if self.obj_name.is_empty() || self.tier_name.is_empty() { return Err(Error::other("tier delete journal entry is incomplete")); } - let backend_identity = match self.version { - 1 => None, - TIER_DELETE_JOURNAL_VERSION => Some( - self.backend_identity - .ok_or_else(|| Error::other("tier delete journal v2 entry is missing its backend identity"))?, + if self.version != TIER_DELETE_JOURNAL_EXACT_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) = match self.version { + 1 => (None, false), + TIER_DELETE_JOURNAL_VERSION => ( + Some( + self.backend_identity + .ok_or_else(|| Error::other("tier delete journal v2 entry is missing its backend identity"))?, + ), + false, ), + TIER_DELETE_JOURNAL_EXACT_VERSION => { + if self.version_id.is_empty() || self.version_id_exact != Some(true) { + return Err(Error::other("tier delete journal v3 entry is missing its exact version constraint")); + } + ( + Some( + self.backend_identity + .ok_or_else(|| Error::other("tier delete journal v3 entry is missing its backend identity"))?, + ), + true, + ) + } version => return Err(Error::other(format!("unsupported tier delete journal version {version}"))), }; Ok(Jentry { @@ -89,6 +117,7 @@ impl PersistedTierDeleteJournalEntry { version_id: self.version_id, tier_name: self.tier_name, backend_identity, + version_id_exact, }) } } @@ -113,6 +142,10 @@ pub(crate) fn tier_delete_journal_object_name(je: &Jentry) -> String { hasher.update([0]); hasher.update(backend_identity); } + if je.version_id_exact { + hasher.update([0]); + hasher.update(b"exact-version-id"); + } format!( "{TIER_DELETE_JOURNAL_PREFIX}{}.json", rustfs_utils::crypto::hex(hasher.finalize().as_slice()) @@ -185,6 +218,7 @@ pub async fn process_tier_delete_journal_entry(api: Arc, je: &Jentry) - &je.tier_name, backend_identity, &api.tier_config_mgr(), + je.version_id_exact, ) .await?; remove_tier_delete_journal_entry(api, je).await @@ -294,16 +328,33 @@ pub async fn recover_tier_delete_journal_entries( } pub async fn run_tier_delete_journal_recovery_loop(api: Arc, cancel_token: CancellationToken) { - let mut interval = tokio::time::interval(std::time::Duration::from_secs(60)); + let mut interval = tokio::time::interval(TIER_DELETE_JOURNAL_RECOVERY_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); let mut marker: Option = None; loop { + #[cfg(test)] tokio::select! { + biased; _ = cancel_token.cancelled() => return, - _ = interval.tick() => {} + _ = interval.tick() => {}, + _ = api.ctx.wait_for_tier_delete_journal_recovery() => {}, + } + #[cfg(not(test))] + tokio::select! { + biased; + _ = cancel_token.cancelled() => return, + _ = interval.tick() => {}, } - match recover_tier_delete_journal_entries(api.clone(), DEFAULT_TIER_DELETE_JOURNAL_RECOVERY_LIMIT, marker.clone()).await { + let recovery = + recover_tier_delete_journal_entries(api.clone(), DEFAULT_TIER_DELETE_JOURNAL_RECOVERY_LIMIT, marker.clone()); + let Some(result) = + await_tier_delete_journal_recovery(&cancel_token, TIER_DELETE_JOURNAL_RECOVERY_TIMEOUT, recovery).await + else { + return; + }; + match result { Ok(stats) => { marker = stats.next_marker; debug!( @@ -332,13 +383,36 @@ pub async fn run_tier_delete_journal_recovery_loop(api: Arc, cancel_tok } } +async fn await_tier_delete_journal_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!( + "tier delete journal recovery timed out after {} seconds", + timeout.as_secs() + ))), + }), + } +} + #[cfg(test)] mod tests { use super::{ - decode_tier_delete_journal_entry, encode_tier_delete_journal_entry, record_tier_delete_journal_backend_identity, - tier_delete_journal_object_name, + TIER_DELETE_JOURNAL_EXACT_VERSION, await_tier_delete_journal_recovery, 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::error::Result; + use std::time::Duration; + use tokio_util::sync::CancellationToken; fn journal_entry() -> Jentry { Jentry { @@ -346,6 +420,7 @@ mod tests { version_id: "remote-version".to_string(), tier_name: "WARM".to_string(), backend_identity: Some([7; 32]), + version_id_exact: false, } } @@ -360,6 +435,82 @@ mod tests { assert_eq!(decoded.version_id, je.version_id); assert_eq!(decoded.tier_name, je.tier_name); assert_eq!(decoded.backend_identity, je.backend_identity); + assert_eq!(decoded.version_id_exact, je.version_id_exact); + } + + #[test] + fn tier_delete_journal_roundtrips_exact_put_response_constraint() { + let mut exact = journal_entry(); + exact.version_id = uuid::Uuid::nil().to_string(); + exact.version_id_exact = true; + let mut normalized = exact.clone(); + normalized.version_id_exact = false; + + let encoded = encode_tier_delete_journal_entry(&exact).expect("exact journal entry should encode"); + let persisted: serde_json::Value = serde_json::from_slice(&encoded).expect("exact journal JSON should decode"); + let decoded = decode_tier_delete_journal_entry(&encoded).expect("exact journal entry should decode"); + + assert_eq!(persisted["version"], TIER_DELETE_JOURNAL_EXACT_VERSION); + assert_eq!(persisted["version_id_exact"], true); + assert!(decoded.version_id_exact); + assert_ne!(tier_delete_journal_object_name(&exact), tier_delete_journal_object_name(&normalized)); + } + + #[test] + fn tier_delete_journal_rejects_invalid_exact_version_constraints() { + let identity = vec![7_u8; 32]; + let invalid = [ + serde_json::json!({ + "version": 1, + "obj_name": "remote/object", + "version_id": "exact-version", + "tier_name": "WARM", + "version_id_exact": true, + }), + serde_json::json!({ + "version": 2, + "obj_name": "remote/object", + "version_id": "exact-version", + "tier_name": "WARM", + "backend_identity": identity, + "version_id_exact": true, + }), + serde_json::json!({ + "version": TIER_DELETE_JOURNAL_EXACT_VERSION, + "obj_name": "remote/object", + "version_id": "", + "tier_name": "WARM", + "backend_identity": identity, + "version_id_exact": true, + }), + serde_json::json!({ + "version": TIER_DELETE_JOURNAL_EXACT_VERSION, + "obj_name": "remote/object", + "version_id": "exact-version", + "tier_name": "WARM", + "backend_identity": identity, + }), + serde_json::json!({ + "version": TIER_DELETE_JOURNAL_EXACT_VERSION, + "obj_name": "remote/object", + "version_id": "exact-version", + "tier_name": "WARM", + "backend_identity": identity, + "version_id_exact": false, + }), + serde_json::json!({ + "version": TIER_DELETE_JOURNAL_EXACT_VERSION, + "obj_name": "remote/object", + "version_id": "exact-version", + "tier_name": "WARM", + "version_id_exact": true, + }), + ]; + + for persisted in invalid { + let encoded = serde_json::to_vec(&persisted).expect("invalid journal fixture should encode"); + decode_tier_delete_journal_entry(&encoded).expect_err("invalid exact journal constraint must fail closed"); + } } #[test] @@ -476,4 +627,29 @@ mod tests { assert!(err.to_string().contains("decode tier delete journal failed")); } + + #[tokio::test] + async fn tier_delete_journal_recovery_has_a_hard_outer_timeout() { + let result = await_tier_delete_journal_recovery( + &CancellationToken::new(), + Duration::from_millis(10), + std::future::pending::>(), + ) + .await + .expect("an elapsed timeout should return a recovery error") + .expect_err("a permanently pending recovery must time out"); + + assert!(result.to_string().contains("recovery timed out"), "{result}"); + } + + #[tokio::test] + async fn tier_delete_journal_recovery_drops_in_flight_work_on_shutdown() { + let cancel = CancellationToken::new(); + cancel.cancel(); + + let result = + await_tier_delete_journal_recovery(&cancel, Duration::from_secs(30), std::future::pending::>()).await; + + assert!(result.is_none(), "shutdown must cancel the in-flight recovery future"); + } } diff --git a/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs b/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs index 3fe5acc48..284e5a21e 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs @@ -249,6 +249,7 @@ impl ObjSweeper { version_id: self.transition_version_id.clone(), tier_name: self.transition_tier.clone(), backend_identity: None, + version_id_exact: false, }); } None @@ -284,6 +285,7 @@ pub struct Jentry { pub(crate) version_id: String, pub(crate) tier_name: String, pub(crate) backend_identity: Option, + pub(crate) version_id_exact: bool, } impl ExpiryOp for Jentry { @@ -328,13 +330,14 @@ async fn delete_object_from_remote_tier_raw_with_manager( let lease = TierConfigMgr::acquire_operation_lease(&tier_config_mgr, tier_name) .await .map_err(std::io::Error::other)?; - delete_object_from_remote_tier_raw_with_lease(obj_name, rv_id, &lease).await + delete_object_from_remote_tier_raw_with_lease(obj_name, rv_id, &lease, false).await } async fn delete_object_from_remote_tier_raw_with_lease( obj_name: &str, rv_id: &str, lease: &TierOperationLease, + version_id_exact: bool, ) -> Result<(), std::io::Error> { if remote_delete_breaker_is_open(Instant::now()).await { metrics::counter!(METRIC_DELETE_REMOTE_BREAKER_TOTAL).increment(1); @@ -347,7 +350,11 @@ async fn delete_object_from_remote_tier_raw_with_lease( .map_err(|_| std::io::Error::other(ERR_REMOTE_DELETE_LIMITER_CLOSED))?; let _inflight = RemoteDeleteInflightGuard::new(); - lease.remove(obj_name, rv_id).await + if version_id_exact { + lease.remove_exact(obj_name, rv_id).await + } else { + lease.remove(obj_name, rv_id).await + } } #[cfg(test)] @@ -388,19 +395,21 @@ pub(crate) async fn delete_object_from_remote_tier_idempotent_with_manager_and_i tier_name: &str, backend_identity: TierDestinationId, tier_config_mgr: &Arc>, + version_id_exact: bool, ) -> Result { let lease = TierConfigMgr::acquire_operation_lease_for_backend_identity(tier_config_mgr, tier_name, backend_identity) .await .map_err(std::io::Error::other)?; - delete_object_from_remote_tier_with_lease_idempotent(obj_name, rv_id, &lease).await + delete_object_from_remote_tier_with_lease_idempotent(obj_name, rv_id, &lease, version_id_exact).await } pub(crate) async fn delete_object_from_remote_tier_with_lease_idempotent( obj_name: &str, rv_id: &str, lease: &TierOperationLease, + version_id_exact: bool, ) -> Result { - match delete_object_from_remote_tier_raw_with_lease(obj_name, rv_id, lease).await { + match delete_object_from_remote_tier_raw_with_lease(obj_name, rv_id, lease, version_id_exact).await { Ok(()) => Ok(RemoteTierDeleteOutcome::Deleted), Err(err) if is_remote_tier_not_found_error(&err) => Ok(RemoteTierDeleteOutcome::AlreadyRemoved), Err(err) => { @@ -450,6 +459,7 @@ pub fn transitioned_force_delete_journal_entry(transitioned: &TransitionedObject version_id: transitioned.version_id.clone(), tier_name: transitioned.tier.clone(), backend_identity: None, + version_id_exact: false, }) } @@ -574,6 +584,7 @@ mod test { "WARM", mismatched, &manager, + false, ) .await .expect_err("journal recovery must fail closed when the tier name was rebound"); @@ -581,6 +592,33 @@ mod test { assert!(err.to_string().contains("identity no longer matches")); } + #[cfg(feature = "test-util")] + #[tokio::test] + async fn journal_delete_dispatches_an_exact_version_constraint() { + let manager = crate::services::tier::tier::TierConfigMgr::new(); + let backend = crate::services::tier::test_util::register_mock_tier(&manager, "WARM").await; + let lease = crate::services::tier::tier::TierConfigMgr::acquire_operation_lease(&manager, "WARM") + .await + .expect("test tier lease should be available"); + let identity = lease.backend_identity(); + drop(lease); + + let outcome = delete_object_from_remote_tier_idempotent_with_manager_and_identity( + "remote/object", + "exact-version", + "WARM", + identity, + &manager, + true, + ) + .await + .expect("an exact journal delete should reach the backend"); + + assert_eq!(outcome, RemoteTierDeleteOutcome::Deleted); + assert_eq!(backend.exact_remove_count(), 1); + assert_eq!(backend.remove_count().await, 1); + } + #[test] fn breaker_opens_at_threshold_and_recovers_after_window() { let mut breaker = RemoteDeleteBreaker::new(3, Duration::from_secs(30)); diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index 09358fb71..96b984824 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -412,6 +412,10 @@ impl BucketMetadataSys { } } + pub(crate) fn object_store(&self) -> Arc { + self.api.clone() + } + pub async fn init(&mut self, buckets: Vec) { let _ = self.init_internal(buckets).await; } diff --git a/crates/ecstore/src/runtime/instance.rs b/crates/ecstore/src/runtime/instance.rs index beb7aba25..a695cea6f 100644 --- a/crates/ecstore/src/runtime/instance.rs +++ b/crates/ecstore/src/runtime/instance.rs @@ -50,7 +50,7 @@ use crate::services::event_notification::EventNotifier; use crate::services::tier::tier::TierConfigMgr; use rustfs_lock::{GlobalLockManager, get_global_lock_manager}; use s3s::region::Region; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::sync::{Arc, OnceLock}; use tokio::sync::{OnceCell, RwLock}; use tokio_util::sync::CancellationToken; @@ -159,6 +159,9 @@ pub struct InstanceContext { /// workers (scanner/heal/tier/lifecycle) without touching another instance. /// Replaces the process-global cancel-token static. background_cancel_token: OnceLock, + tier_delete_journal_recovery_stores: std::sync::Mutex>, + #[cfg(test)] + tier_delete_journal_recovery_wakeup: tokio::sync::Notify, } impl InstanceContext { @@ -193,6 +196,9 @@ impl InstanceContext { local_disk_set_drives: Arc::new(RwLock::new(Vec::new())), bucket_metadata_sys: std::sync::Mutex::new(None), background_cancel_token: OnceLock::new(), + tier_delete_journal_recovery_stores: std::sync::Mutex::new(HashSet::new()), + #[cfg(test)] + tier_delete_journal_recovery_wakeup: tokio::sync::Notify::new(), } } @@ -353,6 +359,27 @@ impl InstanceContext { self.background_cancel_token.get().cloned() } + pub(crate) fn bind_background_cancel_token(&self, token: CancellationToken) -> CancellationToken { + self.background_cancel_token.get_or_init(|| token).clone() + } + + pub(crate) fn mark_tier_delete_journal_recovery_started(&self, store_id: Uuid) -> bool { + self.tier_delete_journal_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(); + } + + #[cfg(test)] + pub(crate) async fn wait_for_tier_delete_journal_recovery(&self) { + self.tier_delete_journal_recovery_wakeup.notified().await; + } + /// Update this instance's erasure setup type. pub async fn update_erasure_type(&self, setup_type: SetupType) { *self.erasure_kind.write().await = setup_type; @@ -406,6 +433,14 @@ impl std::fmt::Debug for InstanceContext { .field("replication_stats_set", &self.replication_stats.get().is_some()) .field("replication_pool_set", &self.replication_pool.get().is_some()) .field("background_cancel_token_set", &self.background_cancel_token.get().is_some()) + .field( + "tier_delete_journal_recovery_store_count", + &self + .tier_delete_journal_recovery_stores + .lock() + .map(|stores| stores.len()) + .unwrap_or_default(), + ) .finish_non_exhaustive() } } @@ -740,6 +775,36 @@ mod tests { assert!(ctx_a.background_cancel_token().unwrap().is_cancelled()); } + #[test] + fn tier_delete_journal_recovery_is_deduplicated_per_store_and_instance() { + let ctx_a = InstanceContext::new(); + let ctx_b = InstanceContext::new(); + let store_a = Uuid::new_v4(); + let store_b = Uuid::new_v4(); + + assert!(ctx_a.mark_tier_delete_journal_recovery_started(store_a)); + assert!(!ctx_a.mark_tier_delete_journal_recovery_started(store_a)); + assert!(ctx_a.mark_tier_delete_journal_recovery_started(store_b)); + assert!(ctx_b.mark_tier_delete_journal_recovery_started(store_a)); + } + + #[test] + fn background_cancel_token_binds_the_provided_shutdown_token() { + let ctx = InstanceContext::new(); + let shutdown = CancellationToken::new(); + let first = ctx.bind_background_cancel_token(shutdown.clone()); + let second = ctx.bind_background_cancel_token(CancellationToken::new()); + + shutdown.cancel(); + assert!(first.is_cancelled()); + assert!(second.is_cancelled()); + assert!( + ctx.background_cancel_token() + .expect("shutdown token should be published") + .is_cancelled() + ); + } + // Phase 5 acceptance (backlog#939): two independent instance contexts share // NONE of the runtime state that used to live in process globals. This is // the end-to-end proof that the object-graph isolation carrier works — every diff --git a/crates/ecstore/src/services/tier/test_util.rs b/crates/ecstore/src/services/tier/test_util.rs index 405cd4349..c69758fe0 100644 --- a/crates/ecstore/src/services/tier/test_util.rs +++ b/crates/ecstore/src/services/tier/test_util.rs @@ -57,7 +57,10 @@ use std::collections::HashMap; use std::io::Cursor; use std::path::Path; -use std::sync::Arc; +use std::sync::{ + Arc, + atomic::{AtomicBool, AtomicUsize, Ordering}, +}; use std::time::Duration; use async_trait::async_trait; @@ -75,6 +78,21 @@ use crate::services::tier::warm_backend::{WarmBackend, WarmBackendGetOpts, build use rustfs_filemeta::FileMeta; use rustfs_utils::path::path_join_buf; +/// One-shot barrier before rejected transition cleanup resolves its ECStore. +pub struct TransitionCleanupStoreBarrier(crate::set_disk::SetDiskTransitionCleanupStoreBarrier); + +impl TransitionCleanupStoreBarrier { + /// Install the barrier for the next rejected transition cleanup. + pub fn install() -> Self { + Self(crate::set_disk::SetDiskTransitionCleanupStoreBarrier::install()) + } + + /// Wait until the rejected transition reaches cleanup-store resolution. + pub async fn wait_until_paused(&self) { + self.0.wait_until_paused().await; + } +} + /// Default polling cadence used by the `wait_for_*` helpers. const POLL_INTERVAL: Duration = Duration::from_millis(50); @@ -142,11 +160,15 @@ struct MockWarmBackendInner { faults: Mutex, put_read_limit: Mutex>, put_remote_version: Mutex>, + reject_non_empty_remote_versions: AtomicBool, + fail_remove: AtomicBool, + exact_remove_count: AtomicUsize, op_log: Mutex>, put_versions: Mutex>, remove_versions: Mutex>, put_barrier: Mutex>>, get_barrier: Mutex>>, + remove_barrier: Mutex>>, } #[derive(Default)] @@ -162,6 +184,23 @@ struct MockGetBarrierState { fail_after_release: bool, } +#[derive(Default)] +struct MockRemoveBarrierState { + arrived: Notify, + release: Notify, + operation_dropped: Notify, +} + +struct MockRemoveOperationGuard { + state: Arc, +} + +impl Drop for MockRemoveOperationGuard { + fn drop(&mut self) { + self.state.operation_dropped.notify_one(); + } +} + /// One-shot barrier that pauses a mock tier PUT after storing its remote body. pub struct MockPutBarrier { state: Arc, @@ -212,6 +251,38 @@ impl Drop for MockGetBarrier { } } +/// One-shot barrier that pauses and then fails a mock tier DELETE. +pub struct MockRemoveBarrier { + state: Arc, +} + +impl MockRemoveBarrier { + /// Wait until DELETE reaches the deterministic failure point. + pub async fn wait_until_paused(&self) { + tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified()) + .await + .expect("mock tier DELETE should reach the deterministic barrier"); + } + + /// Release the paused DELETE, which then returns an injected error. + pub fn release(&self) { + self.state.release.notify_one(); + } + + /// Wait until the paused DELETE future completes or is cancelled. + pub async fn wait_until_operation_dropped(&self) { + tokio::time::timeout(Duration::from_secs(30), self.state.operation_dropped.notified()) + .await + .expect("mock tier DELETE operation should be dropped"); + } +} + +impl Drop for MockRemoveBarrier { + fn drop(&mut self) { + self.state.release.notify_one(); + } +} + /// In-memory [`WarmBackend`] for lifecycle / tiering integration tests. /// /// Cloning shares the same underlying storage, fault configuration, and @@ -235,6 +306,15 @@ impl MockWarmBackend { MockPutBarrier { state } } + /// Pause and then fail the next DELETE after it reaches the backend. + pub async fn arm_failing_remove_barrier(&self) -> MockRemoveBarrier { + let state = Arc::new(MockRemoveBarrierState::default()); + let mut barrier = self.inner.remove_barrier.lock().await; + assert!(barrier.is_none(), "mock tier DELETE barrier is already armed"); + *barrier = Some(state.clone()); + MockRemoveBarrier { state } + } + /// Arm a one-shot pause before the next tier GET, then return an error /// after the test releases it. pub async fn arm_failing_get_barrier(&self) -> MockGetBarrier { @@ -298,6 +378,16 @@ impl MockWarmBackend { *self.inner.put_remote_version.lock().await = remote_version; } + /// Reject non-empty remote versions before transition metadata is committed. + pub fn set_reject_non_empty_remote_versions(&self, reject: bool) { + self.inner.reject_non_empty_remote_versions.store(reject, Ordering::Release); + } + + /// Enable or disable a persistent remove failure for durability tests. + pub fn set_remove_failure(&self, fail: bool) { + self.inner.fail_remove.store(fail, Ordering::Release); + } + async fn precondition(&self) -> Result<(), std::io::Error> { let (latency, error) = { let faults = self.inner.faults.lock().await; @@ -339,6 +429,11 @@ impl MockWarmBackend { .count() } + /// Number of exact-version trait remove calls, including failed attempts. + pub fn exact_remove_count(&self) -> usize { + self.inner.exact_remove_count.load(Ordering::Acquire) + } + /// Return the exact object/version pairs produced by successful tier PUTs. pub async fn put_versions(&self) -> Vec<(String, String)> { self.inner.put_versions.lock().await.clone() @@ -484,6 +579,13 @@ impl MockWarmBackend { #[async_trait] impl WarmBackend for MockWarmBackend { + fn validate_remote_version_id(&self, remote_version_id: &str) -> Result<(), std::io::Error> { + if self.inner.reject_non_empty_remote_versions.load(Ordering::Acquire) && !remote_version_id.is_empty() { + return Err(std::io::Error::other("mock warm backend requires an unversioned remote object")); + } + Ok(()) + } + async fn put(&self, object: &str, r: ReaderImpl, _length: i64) -> Result { self.precondition().await?; let bytes = self.read_bytes(r).await?; @@ -582,6 +684,15 @@ impl WarmBackend for MockWarmBackend { async fn remove(&self, object: &str, rv: &str) -> Result<(), std::io::Error> { self.precondition().await?; + if let Some(barrier) = self.inner.remove_barrier.lock().await.take() { + let _operation = MockRemoveOperationGuard { state: barrier.clone() }; + barrier.arrived.notify_one(); + barrier.release.notified().await; + return Err(std::io::Error::other("mock warm backend remove failure after barrier")); + } + if self.inner.fail_remove.load(Ordering::Acquire) { + return Err(std::io::Error::other("mock warm backend remove failure")); + } let mut objects = self.inner.objects.lock().await; if let Some(stored) = objects.get(object) && !rv.is_empty() @@ -603,6 +714,17 @@ impl WarmBackend for MockWarmBackend { Ok(()) } + async fn remove_exact(&self, object: &str, rv: &str) -> Result<(), std::io::Error> { + if rv.is_empty() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "an exact mock tier delete requires a remote version ID", + )); + } + self.inner.exact_remove_count.fetch_add(1, Ordering::AcqRel); + self.remove(object, rv).await + } + async fn in_use(&self) -> Result { self.precondition().await?; self.record(MockWarmOp::InUse).await; diff --git a/crates/ecstore/src/services/tier/tier.rs b/crates/ecstore/src/services/tier/tier.rs index 946734c2b..134e5b1d1 100644 --- a/crates/ecstore/src/services/tier/tier.rs +++ b/crates/ecstore/src/services/tier/tier.rs @@ -672,6 +672,14 @@ struct SharedWarmBackendProxy(SharedWarmBackend); #[async_trait::async_trait] impl WarmBackend for SharedWarmBackendProxy { + async fn validate(&self) -> io::Result<()> { + self.0.validate().await + } + + fn validate_remote_version_id(&self, remote_version_id: &str) -> io::Result<()> { + self.0.validate_remote_version_id(remote_version_id) + } + async fn put(&self, object: &str, r: crate::client::transition_api::ReaderImpl, length: i64) -> io::Result { self.0.put(object, r, length).await } @@ -699,6 +707,10 @@ impl WarmBackend for SharedWarmBackendProxy { self.0.remove(object, rv).await } + async fn remove_exact(&self, object: &str, rv: &str) -> io::Result<()> { + self.0.remove_exact(object, rv).await + } + async fn in_use(&self) -> io::Result { self.0.in_use().await } @@ -3046,6 +3058,22 @@ mod tests { #[async_trait::async_trait] impl WarmBackend for MockWarmBackend { + async fn validate(&self) -> std::result::Result<(), std::io::Error> { + if self.healthy { + Ok(()) + } else { + Err(std::io::Error::other("mock validation failed")) + } + } + + fn validate_remote_version_id(&self, remote_version_id: &str) -> std::result::Result<(), std::io::Error> { + if remote_version_id == "unsupported-version" { + Err(std::io::Error::other("mock remote version rejected")) + } else { + Ok(()) + } + } + async fn put(&self, _object: &str, _r: ReaderImpl, _length: i64) -> std::result::Result { if self.healthy { Ok("mock-version".to_string()) @@ -3085,6 +3113,13 @@ mod tests { } } + async fn remove_exact(&self, object: &str, rv: &str) -> std::result::Result<(), std::io::Error> { + if rv == "exact-only" { + return Err(std::io::Error::other("mock exact remove forwarded")); + } + self.remove(object, rv).await + } + async fn in_use(&self) -> std::result::Result { match self.in_use_value { Some(b) => Ok(b), @@ -3442,6 +3477,32 @@ mod tests { .expect_err("an unhealthy backend must fail verification"); } + #[tokio::test] + async fn shared_backend_proxy_forwards_validation_hooks() { + let unhealthy: SharedWarmBackend = Arc::new(MockWarmBackend { + in_use_value: Some(false), + healthy: false, + }); + let proxy = SharedWarmBackendProxy(unhealthy); + let err = proxy.validate().await.expect_err("proxy must forward backend validation"); + assert_eq!(err.to_string(), "mock validation failed"); + + let healthy: SharedWarmBackend = Arc::new(MockWarmBackend { + in_use_value: Some(false), + healthy: true, + }); + let proxy = SharedWarmBackendProxy(healthy); + let err = proxy + .validate_remote_version_id("unsupported-version") + .expect_err("proxy must forward remote version validation"); + assert_eq!(err.to_string(), "mock remote version rejected"); + let err = proxy + .remove_exact("remote-object", "exact-only") + .await + .expect_err("proxy must forward exact-version cleanup"); + assert_eq!(err.to_string(), "mock exact remove forwarded"); + } + // ---- pure query helpers -------------------------------------------- #[test] diff --git a/crates/ecstore/src/services/tier/warm_backend.rs b/crates/ecstore/src/services/tier/warm_backend.rs index 51a2c7c77..9bf70acbd 100644 --- a/crates/ecstore/src/services/tier/warm_backend.rs +++ b/crates/ecstore/src/services/tier/warm_backend.rs @@ -25,7 +25,7 @@ use crate::client::{ }; use crate::error::is_err_bucket_not_found; use crate::services::tier::{ - tier::ERR_TIER_TYPE_UNSUPPORTED, + tier::{ERR_TIER_INVALID_CONFIG, ERR_TIER_TYPE_UNSUPPORTED}, tier_config::{TierConfig, TierType}, tier_handlers::{ERR_TIER_BUCKET_NOT_FOUND, ERR_TIER_NOT_FOUND, ERR_TIER_PERM_ERR}, warm_backend_aliyun::WarmBackendAliyun, @@ -65,6 +65,14 @@ pub struct WarmBackendGetOpts { #[async_trait::async_trait] pub trait WarmBackend { + async fn validate(&self) -> Result<(), std::io::Error> { + Ok(()) + } + + fn validate_remote_version_id(&self, _remote_version_id: &str) -> Result<(), std::io::Error> { + Ok(()) + } + /// Return `Ok` only after the backend has consumed the complete declared /// body and its storage service has acknowledged the PUT. The built-in S3 /// family uses the transition client's declared-length request plus @@ -83,6 +91,15 @@ pub trait WarmBackend { ) -> Result; async fn get(&self, object: &str, rv: &str, opts: WarmBackendGetOpts) -> Result; async fn remove(&self, object: &str, rv: &str) -> Result<(), std::io::Error>; + async fn remove_exact(&self, object: &str, rv: &str) -> Result<(), std::io::Error> { + if rv.is_empty() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "an exact tier delete requires a remote version ID", + )); + } + self.remove(object, rv).await + } async fn in_use(&self) -> Result; } @@ -165,16 +182,23 @@ pub fn build_transition_put_options(storage_class: String, mut metadata: HashMap pub async fn check_warm_backend(w: Option<&WarmBackendImpl>) -> Result<(), AdminError> { let w = w.ok_or_else(|| ERR_TIER_NOT_FOUND.clone())?; + w.validate().await.map_err(|_| ERR_TIER_INVALID_CONFIG.clone())?; let remote_version_id = w .put(PROBE_OBJECT, ReaderImpl::Body(Bytes::from("RustFS".as_bytes().to_vec())), 5) - .await; - if let Err(err) = remote_version_id { - return Err(ERR_TIER_PERM_ERR.clone()); + .await + .map_err(|_| ERR_TIER_PERM_ERR.clone())?; + + if w.validate_remote_version_id(&remote_version_id).is_err() { + w.remove_exact(PROBE_OBJECT, &remote_version_id) + .await + .map_err(|_| ERR_TIER_PERM_ERR.clone())?; + return Err(ERR_TIER_INVALID_CONFIG.clone()); } - let r = w.get(PROBE_OBJECT, "", WarmBackendGetOpts::default()).await; + let read_result = w.get(PROBE_OBJECT, &remote_version_id, WarmBackendGetOpts::default()).await; + let remove_result = w.remove(PROBE_OBJECT, &remote_version_id).await; //xhttp.DrainBody(r); - if let Err(err) = r { + if read_result.is_err() || remove_result.is_err() { //if is_err_bucket_not_found(&err) { // return Err(ERR_TIER_BUCKET_NOT_FOUND); //} @@ -185,11 +209,6 @@ pub async fn check_warm_backend(w: Option<&WarmBackendImpl>) -> Result<(), Admin return Err(ERR_TIER_PERM_ERR.clone()); //} } - if let Ok(version_id) = remote_version_id { - if let Err(err) = w.remove(PROBE_OBJECT, &version_id).await { - return Err(ERR_TIER_PERM_ERR.clone()); - }; - } Ok(()) } @@ -391,6 +410,237 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result, + puts: Arc, + removes: Arc, + } + + struct RejectingProbeVersionBackend { + gets: Arc, + removed_versions: Arc>>, + } + + struct RecordingProbeBackend { + get_versions: Arc>>, + removed_versions: Arc>>, + fail_get: bool, + } + + #[async_trait::async_trait] + impl WarmBackend for RejectingValidationBackend { + async fn validate(&self) -> Result<(), std::io::Error> { + self.validations.fetch_add(1, Ordering::SeqCst); + Err(std::io::Error::other("invalid backend configuration")) + } + + async fn put(&self, _object: &str, _r: ReaderImpl, _length: i64) -> Result { + self.puts.fetch_add(1, Ordering::SeqCst); + Ok(String::new()) + } + + async fn put_with_meta( + &self, + object: &str, + r: ReaderImpl, + length: i64, + _meta: HashMap, + ) -> Result { + self.put(object, r, length).await + } + + async fn get(&self, _object: &str, _rv: &str, _opts: WarmBackendGetOpts) -> Result { + Err(std::io::Error::other("get must not run after validation failure")) + } + + async fn remove(&self, _object: &str, _rv: &str) -> Result<(), std::io::Error> { + self.removes.fetch_add(1, Ordering::SeqCst); + Err(std::io::Error::other("remove must not run after validation failure")) + } + + async fn in_use(&self) -> Result { + Err(std::io::Error::other("in_use must not run after validation failure")) + } + } + + #[async_trait::async_trait] + impl WarmBackend for RejectingProbeVersionBackend { + fn validate_remote_version_id(&self, remote_version_id: &str) -> Result<(), std::io::Error> { + if remote_version_id.is_empty() { + Ok(()) + } else { + Err(std::io::Error::other("probe returned a version ID")) + } + } + + async fn put(&self, _object: &str, _r: ReaderImpl, _length: i64) -> Result { + Ok(uuid::Uuid::nil().to_string()) + } + + async fn put_with_meta( + &self, + object: &str, + r: ReaderImpl, + length: i64, + _meta: HashMap, + ) -> Result { + self.put(object, r, length).await + } + + async fn get(&self, _object: &str, _rv: &str, _opts: WarmBackendGetOpts) -> Result { + self.gets.fetch_add(1, Ordering::SeqCst); + Err(std::io::Error::other("GET must not run for a rejected probe version")) + } + + async fn remove(&self, _object: &str, _rv: &str) -> Result<(), std::io::Error> { + Err(std::io::Error::other("generic remove must not run for a rejected fresh PUT response")) + } + + async fn remove_exact(&self, _object: &str, rv: &str) -> Result<(), std::io::Error> { + self.removed_versions.lock().await.push(rv.to_string()); + Ok(()) + } + + async fn in_use(&self) -> Result { + Ok(false) + } + } + + #[async_trait::async_trait] + impl WarmBackend for RecordingProbeBackend { + async fn put(&self, _object: &str, _r: ReaderImpl, _length: i64) -> Result { + Ok(PROBE_VERSION.to_string()) + } + + async fn put_with_meta( + &self, + object: &str, + r: ReaderImpl, + length: i64, + _meta: HashMap, + ) -> Result { + self.put(object, r, length).await + } + + async fn get(&self, _object: &str, rv: &str, _opts: WarmBackendGetOpts) -> Result { + self.get_versions.lock().await.push(rv.to_string()); + if self.fail_get { + Err(std::io::Error::other("probe GET failed")) + } else { + Ok(ReadCloser::new(std::io::Cursor::new(Vec::new()))) + } + } + + async fn remove(&self, _object: &str, rv: &str) -> Result<(), std::io::Error> { + self.removed_versions.lock().await.push(rv.to_string()); + Ok(()) + } + + async fn in_use(&self) -> Result { + Ok(false) + } + } + + #[tokio::test] + async fn check_warm_backend_validates_before_probe_io() { + let validations = Arc::new(AtomicUsize::new(0)); + let puts = Arc::new(AtomicUsize::new(0)); + let removes = Arc::new(AtomicUsize::new(0)); + let backend: WarmBackendImpl = Box::new(RejectingValidationBackend { + validations: validations.clone(), + puts: puts.clone(), + removes: removes.clone(), + }); + + let err = check_warm_backend(Some(&backend)) + .await + .expect_err("invalid backend configuration should fail before probe I/O"); + + assert_eq!(err.code, ERR_TIER_INVALID_CONFIG.code); + assert_eq!(validations.load(Ordering::SeqCst), 1); + assert_eq!(puts.load(Ordering::SeqCst), 0); + assert_eq!(removes.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn default_exact_remove_rejects_an_empty_version() { + let removes = Arc::new(AtomicUsize::new(0)); + let backend = RejectingValidationBackend { + validations: Arc::new(AtomicUsize::new(0)), + puts: Arc::new(AtomicUsize::new(0)), + removes: removes.clone(), + }; + + let err = backend + .remove_exact("remote-object", "") + .await + .expect_err("an empty exact constraint must fail closed"); + + assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput); + assert_eq!(removes.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn check_warm_backend_removes_exact_probe_when_versioning_drifts() { + let gets = Arc::new(AtomicUsize::new(0)); + let removed_versions = Arc::new(tokio::sync::Mutex::new(Vec::new())); + let backend: WarmBackendImpl = Box::new(RejectingProbeVersionBackend { + gets: gets.clone(), + removed_versions: removed_versions.clone(), + }); + + let err = check_warm_backend(Some(&backend)) + .await + .expect_err("a probe version ID must fail an unversioned backend check"); + + assert_eq!(err.code, ERR_TIER_INVALID_CONFIG.code); + assert_eq!(gets.load(Ordering::SeqCst), 0); + assert_eq!(removed_versions.lock().await.as_slice(), [uuid::Uuid::nil().to_string()]); + } + + #[tokio::test] + async fn check_warm_backend_forwards_probe_version_to_get_and_remove() { + let get_versions = Arc::new(tokio::sync::Mutex::new(Vec::new())); + let removed_versions = Arc::new(tokio::sync::Mutex::new(Vec::new())); + let backend: WarmBackendImpl = Box::new(RecordingProbeBackend { + get_versions: get_versions.clone(), + removed_versions: removed_versions.clone(), + fail_get: false, + }); + + check_warm_backend(Some(&backend)) + .await + .expect("a successful probe should validate, read, and remove its object"); + + assert_eq!(get_versions.lock().await.as_slice(), [PROBE_VERSION]); + assert_eq!(removed_versions.lock().await.as_slice(), [PROBE_VERSION]); + } + + #[tokio::test] + async fn check_warm_backend_removes_probe_after_get_failure() { + let get_versions = Arc::new(tokio::sync::Mutex::new(Vec::new())); + let removed_versions = Arc::new(tokio::sync::Mutex::new(Vec::new())); + let backend: WarmBackendImpl = Box::new(RecordingProbeBackend { + get_versions: get_versions.clone(), + removed_versions: removed_versions.clone(), + fail_get: true, + }); + + let err = check_warm_backend(Some(&backend)) + .await + .expect_err("a failed probe GET should return a permission error after cleanup"); + + assert_eq!(err.code, ERR_TIER_PERM_ERR.code); + assert_eq!(get_versions.lock().await.as_slice(), [PROBE_VERSION]); + assert_eq!(removed_versions.lock().await.as_slice(), [PROBE_VERSION]); + } #[test] fn build_transition_put_options_preserves_content_headers() { diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 69e608e9f..405c7230c 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -687,6 +687,8 @@ mod core; mod ctx; mod metadata; mod ops; +#[cfg(feature = "test-util")] +pub(crate) use ops::object::TransitionCleanupStoreBarrier as SetDiskTransitionCleanupStoreBarrier; pub(crate) use ops::object::body_cache_plaintext_len; mod read; mod replication; diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index ffea1d72c..ce66badd6 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -21,10 +21,14 @@ use super::super::*; -use crate::bucket::lifecycle::tier_sweeper::{RemoteTierDeleteOutcome, delete_object_from_remote_tier_with_lease_idempotent}; +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}, +}; use crate::disk::OldCurrentSize; use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppressed}; use crate::services::tier::tier::{TierConfigMgr, TierOperationLease}; +use crate::store::ECStore; use futures::FutureExt as _; use std::future::Future; @@ -1356,12 +1360,11 @@ enum TransitionUploadRemoteVersion { impl TransitionUploadCandidate { pub(crate) fn from_put_response(remote_version: String) -> Self { - let remote_version = - if remote_version.is_empty() || Uuid::parse_str(&remote_version).is_ok_and(|version_id| version_id.is_nil()) { - TransitionUploadRemoteVersion::KnownUnversioned(remote_version) - } else { - TransitionUploadRemoteVersion::KnownExact(remote_version) - }; + let remote_version = if remote_version.is_empty() { + TransitionUploadRemoteVersion::KnownUnversioned(remote_version) + } else { + TransitionUploadRemoteVersion::KnownExact(remote_version) + }; Self { remote_version } } @@ -1378,6 +1381,10 @@ impl TransitionUploadCandidate { TransitionUploadRemoteVersion::KnownUnversioned(_) => "", } } + + fn cleanup_version_is_exact(&self) -> bool { + matches!(&self.remote_version, TransitionUploadRemoteVersion::KnownExact(_)) + } } pub(crate) async fn complete_transition_upload( @@ -1444,22 +1451,18 @@ where pub(crate) async fn cleanup_uncommitted_transition_upload( lease: &TierOperationLease, object: &str, - candidate: &TransitionUploadCandidate, + cleanup_version: &str, + version_id_exact: bool, ) -> std::io::Result { - delete_object_from_remote_tier_with_lease_idempotent(object, candidate.cleanup_version(), lease).await + delete_object_from_remote_tier_with_lease_idempotent(object, cleanup_version, lease, version_id_exact).await } -fn log_transition_upload_cleanup_failure( - lease: &TierOperationLease, - object: &str, - candidate: &TransitionUploadCandidate, - err: &std::io::Error, -) { +fn log_transition_upload_cleanup_failure(lease: &TierOperationLease, object: &str, cleanup_version: &str, err: &std::io::Error) { warn!( tier = lease.tier_name(), tier_generation = lease.generation(), object, - remote_version = candidate.cleanup_version(), + remote_version = cleanup_version, error = ?err, "failed to clean uncommitted transition upload" ); @@ -1468,33 +1471,72 @@ fn log_transition_upload_cleanup_failure( pub(crate) struct TransitionUploadCleanup { lease: TierOperationLease, object: String, - candidate: TransitionUploadCandidate, + candidate: Option, + cleanup_ctx: Arc, + cleanup_api: Option>, armed: bool, } impl TransitionUploadCleanup { - pub(crate) fn new(lease: TierOperationLease, object: &str, candidate: TransitionUploadCandidate) -> Self { + pub(crate) fn new( + lease: TierOperationLease, + object: &str, + cleanup_ctx: Arc, + ) -> Self { Self { lease, object: object.to_string(), - candidate, + candidate: None, + cleanup_ctx, + cleanup_api: None, armed: true, } } + fn cleanup_candidate(&self) -> std::io::Result<&TransitionUploadCandidate> { + self.candidate + .as_ref() + .ok_or_else(|| std::io::Error::other("transition upload cleanup has no confirmed remote candidate")) + } + pub(crate) async fn cleanup(&mut self) -> std::io::Result { - match cleanup_uncommitted_transition_upload(&self.lease, &self.object, &self.candidate).await { + let candidate = self.cleanup_candidate()?; + let result = cleanup_uncommitted_transition_upload( + &self.lease, + &self.object, + candidate.cleanup_version(), + candidate.cleanup_version_is_exact(), + ) + .await; + match result { Ok(outcome) => { self.armed = false; Ok(outcome) } Err(err) => { - log_transition_upload_cleanup_failure(&self.lease, &self.object, &self.candidate, &err); + log_transition_upload_cleanup_failure(&self.lease, &self.object, candidate.cleanup_version(), &err); Err(err) } } } + async fn cleanup_rejected_upload(&mut self, api: Option>) -> std::io::Result<()> { + self.cleanup_api = api.clone(); + let candidate = self.cleanup_candidate()?; + let result = cleanup_rejected_transition_upload_durably( + &self.lease, + &self.object, + candidate.cleanup_version(), + candidate.cleanup_version_is_exact(), + api, + ) + .await; + if result.is_ok() { + self.armed = false; + } + result + } + pub(crate) fn disarm(&mut self) { self.armed = false; } @@ -1505,6 +1547,9 @@ impl Drop for TransitionUploadCleanup { if !self.armed { return; } + let Some(candidate) = self.candidate.as_ref() else { + return; + }; let lease = match self.lease.try_clone() { Ok(lease) => lease, Err(err) => { @@ -1519,17 +1564,240 @@ impl Drop for TransitionUploadCleanup { } }; let object = self.object.clone(); - let candidate = self.candidate.clone(); + let cleanup_version = candidate.cleanup_version().to_string(); + let version_id_exact = candidate.cleanup_version_is_exact(); + let cleanup_api = self.cleanup_api.clone(); + let cleanup_ctx = self.cleanup_ctx.clone(); if let Ok(handle) = tokio::runtime::Handle::try_current() { handle.spawn(async move { - if let Err(err) = cleanup_uncommitted_transition_upload(&lease, &object, &candidate).await { - log_transition_upload_cleanup_failure(&lease, &object, &candidate, &err); + let api = match cleanup_api { + Some(api) => Some(api), + None => transition_cleanup_store(&cleanup_ctx).await, + }; + if let Err(err) = + cleanup_rejected_transition_upload_durably(&lease, &object, &cleanup_version, version_id_exact, api).await + { + warn!( + tier = lease.tier_name(), + tier_generation = lease.generation(), + object, + remote_version = cleanup_version, + error = ?err, + "cancelled transition upload was neither deleted nor journaled" + ); } }); } } } +async fn cleanup_rejected_transition_upload_durably( + lease: &TierOperationLease, + object: &str, + cleanup_version: &str, + version_id_exact: bool, + api: Option>, +) -> std::io::Result<()> { + let journal_entry = Jentry { + obj_name: object.to_string(), + version_id: cleanup_version.to_string(), + tier_name: lease.tier_name().to_string(), + backend_identity: Some(lease.backend_identity()), + version_id_exact, + }; + + let journal_error = if let Some(api) = api.as_ref() { + match persist_tier_delete_journal_entry(api.clone(), &journal_entry).await { + Ok(()) => { + match cleanup_uncommitted_transition_upload(lease, object, cleanup_version, version_id_exact).await { + Ok(_) => { + if let Err(err) = remove_tier_delete_journal_entry(api.clone(), &journal_entry).await { + warn!( + tier = lease.tier_name(), + object, + error = ?err, + "rejected transition upload was deleted but its cleanup journal was retained" + ); + } + } + Err(err) => log_transition_upload_cleanup_failure(lease, object, cleanup_version, &err), + } + return Ok(()); + } + Err(err) => err, + } + } else { + std::io::Error::other("object store unavailable for rejected transition cleanup journal") + }; + warn!( + tier = lease.tier_name(), + object, + error = ?journal_error, + "failed to persist rejected transition upload cleanup journal" + ); + + let cleanup_error = match cleanup_uncommitted_transition_upload(lease, object, cleanup_version, version_id_exact).await { + Ok(_) => return Ok(()), + Err(err) => { + log_transition_upload_cleanup_failure(lease, object, cleanup_version, &err); + err + } + }; + if let Some(api) = api { + match persist_tier_delete_journal_entry(api, &journal_entry).await { + Ok(()) => return Ok(()), + Err(retry_error) => { + return Err(std::io::Error::other(format!( + "rejected transition upload was neither deleted nor journaled: initial journal error: {journal_error}; cleanup error: {cleanup_error}; journal retry error: {retry_error}" + ))); + } + } + } + Err(std::io::Error::other(format!( + "rejected transition upload was neither deleted nor journaled: journal error: {journal_error}; cleanup error: {cleanup_error}" + ))) +} + +async fn transition_cleanup_store(ctx: &Arc) -> Option> { + #[cfg(feature = "test-util")] + pause_transition_cleanup_store().await; + + if let Some(api) = runtime_sources::object_store_handle().filter(|api| Arc::ptr_eq(&api.ctx, ctx)) { + return Some(api); + } + let metadata_sys = ctx.bucket_metadata_sys()?; + let api = metadata_sys.read().await.object_store(); + Arc::ptr_eq(&api.ctx, ctx).then_some(api) +} + +#[cfg(feature = "test-util")] +#[derive(Default)] +struct TransitionCleanupStoreBarrierState { + arrived: tokio::sync::Notify, + release: tokio::sync::Notify, +} + +#[cfg(feature = "test-util")] +/// One-shot test barrier placed before transition cleanup resolves its ECStore. +pub(crate) struct TransitionCleanupStoreBarrier { + state: Arc, +} + +#[cfg(feature = "test-util")] +static TRANSITION_CLEANUP_STORE_BARRIER: std::sync::OnceLock>>> = + std::sync::OnceLock::new(); + +#[cfg(feature = "test-util")] +impl TransitionCleanupStoreBarrier { + /// Install the process-local barrier for the next cleanup-store resolution. + pub(crate) fn install() -> Self { + let state = Arc::new(TransitionCleanupStoreBarrierState::default()); + let mut slot = TRANSITION_CLEANUP_STORE_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("transition cleanup store barrier mutex should not poison"); + assert!(slot.is_none(), "transition cleanup store barrier must be installed by one test at a time"); + *slot = Some(state.clone()); + drop(slot); + Self { state } + } + + /// Wait until a transition reaches the cleanup-store resolution boundary. + pub(crate) async fn wait_until_paused(&self) { + tokio::time::timeout(std::time::Duration::from_secs(30), self.state.arrived.notified()) + .await + .expect("transition should reach the cleanup store barrier"); + } +} + +#[cfg(feature = "test-util")] +impl Drop for TransitionCleanupStoreBarrier { + fn drop(&mut self) { + self.state.release.notify_one(); + let mut slot = TRANSITION_CLEANUP_STORE_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("transition cleanup store barrier mutex should not poison"); + if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) { + *slot = None; + } + } +} + +#[cfg(feature = "test-util")] +async fn pause_transition_cleanup_store() { + let barrier = TRANSITION_CLEANUP_STORE_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("transition cleanup store barrier mutex should not poison") + .take(); + if let Some(barrier) = barrier { + barrier.arrived.notify_one(); + barrier.release.notified().await; + } +} + +#[cfg(all(test, feature = "test-util"))] +struct TransitionUploadCandidateBarrier { + state: Arc, +} + +#[cfg(all(test, feature = "test-util"))] +static TRANSITION_UPLOAD_CANDIDATE_BARRIER: std::sync::OnceLock< + std::sync::Mutex>>, +> = std::sync::OnceLock::new(); + +#[cfg(all(test, feature = "test-util"))] +impl TransitionUploadCandidateBarrier { + fn install() -> Self { + let state = Arc::new(TransitionCleanupStoreBarrierState::default()); + let mut slot = TRANSITION_UPLOAD_CANDIDATE_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("transition upload candidate barrier mutex should not poison"); + assert!( + slot.is_none(), + "transition upload candidate barrier must be installed by one test at a time" + ); + *slot = Some(state.clone()); + drop(slot); + Self { state } + } + + async fn wait_until_paused(&self) { + tokio::time::timeout(std::time::Duration::from_secs(30), self.state.arrived.notified()) + .await + .expect("transition should record its remote upload candidate"); + } +} + +#[cfg(all(test, feature = "test-util"))] +impl Drop for TransitionUploadCandidateBarrier { + fn drop(&mut self) { + self.state.release.notify_one(); + let mut slot = TRANSITION_UPLOAD_CANDIDATE_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("transition upload candidate barrier mutex should not poison"); + if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) { + *slot = None; + } + } +} + +#[cfg(all(test, feature = "test-util"))] +async fn pause_after_transition_upload_candidate_recorded() { + let barrier = TRANSITION_UPLOAD_CANDIDATE_BARRIER + .get_or_init(|| std::sync::Mutex::new(None)) + .lock() + .expect("transition upload candidate barrier mutex should not poison") + .take(); + if let Some(barrier) = barrier { + barrier.arrived.notify_one(); + barrier.release.notified().await; + } +} + #[cfg(test)] #[derive(Clone, Copy, PartialEq, Eq)] enum TransitionCommitPause { @@ -1798,17 +2066,20 @@ mod transition_version_id_tests { use uuid::Uuid; #[test] - fn normalizes_unversioned_remote_ids() { + fn normalizes_persisted_unversioned_ids_and_preserves_put_constraints() { assert_eq!(parse_transition_version_id("").expect("empty remote version should be valid"), None); assert_eq!( parse_transition_version_id(&Uuid::nil().to_string()).expect("nil remote version should be valid"), None ); - assert_eq!( - TransitionUploadCandidate::from_put_response(Uuid::nil().to_string()).cleanup_version(), - "" - ); - assert_eq!(TransitionUploadCandidate::from_put_response(String::new()).cleanup_version(), ""); + let nil_put_response = Uuid::nil().to_string(); + let nil_candidate = TransitionUploadCandidate::from_put_response(nil_put_response.clone()); + assert_eq!(nil_candidate.cleanup_version(), nil_put_response); + assert!(nil_candidate.cleanup_version_is_exact()); + + let empty_candidate = TransitionUploadCandidate::from_put_response(String::new()); + assert_eq!(empty_candidate.cleanup_version(), ""); + assert!(!empty_candidate.cleanup_version_is_exact()); } #[test] @@ -2937,33 +3208,52 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { Ok(writer.produced()) }; - let rv = complete_transition_upload( - tgt_client.put_with_meta(&dest_obj, reader, fi.size, transition_meta), - producer, - expected_size, - consumed, - ) - .await; + let mut upload_cleanup = TransitionUploadCleanup::new(tgt_client, &dest_obj, self.ctx.clone()); + let remote_upload = { + let lease = &upload_cleanup.lease; + let recorded_candidate = &mut upload_cleanup.candidate; + let remote_object = &dest_obj; + async move { + let remote_version = lease.put_with_meta(remote_object, reader, fi.size, transition_meta).await?; + *recorded_candidate = Some(TransitionUploadCandidate::from_put_response(remote_version.clone())); + #[cfg(all(test, feature = "test-util"))] + pause_after_transition_upload_candidate_recorded().await; + Ok(remote_version) + } + }; + let rv = complete_transition_upload(remote_upload, producer, expected_size, consumed).await; let candidate = match rv { Ok(completion) => completion.candidate, Err(failure) => { - if let Some(candidate) = failure.candidate { - let mut upload_cleanup = TransitionUploadCleanup::new(tgt_client, &dest_obj, candidate); - let _cleanup_result = upload_cleanup.cleanup().await; + if failure.candidate.is_some() { + 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!( + "{}; rejected remote upload cleanup failed: {cleanup_err}", + failure.error + )))); + } } return Err(failure.error); } }; + if let Err(err) = upload_cleanup.lease.validate_remote_version_id(candidate.remote_version()) { + 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}; rejected remote upload cleanup failed: {cleanup_err}" + )))); + } + return Err(err.into()); + } let transition_version_id = match parse_transition_version_id(candidate.remote_version()) { Ok(version_id) => version_id, Err(err) => { - let mut upload_cleanup = TransitionUploadCleanup::new(tgt_client, &dest_obj, candidate); let _cleanup_result = upload_cleanup.cleanup().await; return Err(err.into()); } }; - let mut upload_cleanup = TransitionUploadCleanup::new(tgt_client, &dest_obj, candidate); let mut commit_opts = opts.clone(); commit_opts.no_lock = true; @@ -4744,6 +5034,26 @@ mod transition_commit_failure_tests { .await .expect("transitioned metadata should resolve"); let expected_identity = rustfs_utils::crypto::hex(old_identity); + let rustfs_identity_key = format!( + "{}{}", + rustfs_utils::http::metadata_compat::RUSTFS_INTERNAL_PREFIX, + rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID + ); + let minio_identity_key = format!( + "{}{}", + rustfs_utils::http::metadata_compat::MINIO_INTERNAL_PREFIX, + rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID + ); + assert_eq!( + transitioned.user_defined.get(&rustfs_identity_key), + Some(&expected_identity), + "transition commits must write the RustFS destination identity key" + ); + assert_eq!( + transitioned.user_defined.get(&minio_identity_key), + Some(&expected_identity), + "transition commits must write the MinIO-compatible destination identity key" + ); assert_eq!( rustfs_utils::http::metadata_compat::get_str( &transitioned.user_defined, @@ -5058,6 +5368,32 @@ mod transition_upload_integrity_tests { } } + #[tokio::test] + #[serial_test::serial] + async fn remote_put_failure_preserves_error_without_cleanup_candidate() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "transition-remote-put-failure-bucket"; + let object = "object.bin"; + let payload = b"a failed remote PUT must not manufacture a cleanup candidate".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 backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; + backend.set_unreachable(true).await; + + let error = set_disks + .transition_object(bucket, object, &transition_options(&original, tier_name)) + .await + .expect_err("an unreachable tier must fail the remote PUT"); + assert!( + matches!(error, StorageError::Io(ref err) if err.kind() == std::io::ErrorKind::ConnectionRefused), + "the original remote PUT error must be preserved: {error:?}" + ); + assert_eq!(backend.remove_count().await, 0, "an unconfirmed candidate must not be cleaned up"); + assert_eq!(backend.exact_remove_count(), 0, "an unconfirmed candidate must not reach exact cleanup"); + assert_eq!(backend.object_count().await, 0); + assert_local_source_intact(&set_disks, bucket, object, &payload).await; + } + #[tokio::test] #[serial_test::serial] async fn partial_remote_acceptance_cleans_exact_candidate_and_preserves_source() { @@ -5080,8 +5416,8 @@ mod transition_upload_integrity_tests { let removed_versions = backend.remove_versions().await; assert_eq!(removed_versions.len(), 1); assert_eq!( - removed_versions[0].1, "", - "the nil UUID response is the backend's unversioned sentinel and must not become an S3 versionId" + removed_versions[0].1, remote_version, + "a non-empty fresh PUT response must be used as the exact cleanup constraint" ); assert_eq!(backend.object_count().await, 0); assert_local_source_intact(&set_disks, bucket, object, &payload).await; @@ -5247,12 +5583,55 @@ mod transition_upload_integrity_tests { assert_local_source_intact(&set_disks, bucket, object, &payload).await; } + #[tokio::test] + #[serial_test::serial] + async fn cancelled_after_put_response_before_upload_join_cleans_candidate() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "transition-cancel-before-upload-join-bucket"; + let object = "object.bin"; + let payload = b"a confirmed remote upload must survive cancellation until cleanup owns it".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 backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; + let barrier = TransitionUploadCandidateBarrier::install(); + + let transition_set = Arc::clone(&set_disks); + let transition = tokio::spawn(async move { + transition_set + .transition_object(bucket, object, &transition_options(&original, tier_name)) + .await + }); + barrier.wait_until_paused().await; + let put_versions = backend.put_versions().await; + assert_eq!(put_versions.len(), 1, "the remote PUT response must identify one cleanup candidate"); + assert_eq!(backend.object_count().await, 1, "the candidate must exist at the cancellation point"); + + transition.abort(); + assert!( + transition + .await + .expect_err("aborted transition task should report cancellation") + .is_cancelled() + ); + drop(barrier); + + assert!( + backend + .wait_for_remote_absence(&put_versions[0].0, Duration::from_secs(5)) + .await, + "the pre-created cleanup guard must remove a candidate recorded before upload finalization completes" + ); + assert_eq!(backend.remove_versions().await, put_versions); + assert_local_source_intact(&set_disks, bucket, object, &payload).await; + } + #[tokio::test] #[serial_test::serial] async fn real_bitrot_producer_failures_do_not_commit_transition() { let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await; + backend.set_reject_non_empty_remote_versions(true); for position in [ ShardCorruptionPosition::First, @@ -5412,6 +5791,63 @@ mod transition_upload_integrity_tests { assert_local_source_intact(&set_disks, bucket, object, &payload).await; } + #[tokio::test] + #[serial_test::serial] + async fn backend_version_constraint_rejects_uuid_candidate_before_commit() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "transition-version-constraint-bucket"; + let object = "object.bin"; + let payload = b"version-constrained backend candidate must retain local data".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; + backend.set_reject_non_empty_remote_versions(true); + + set_disks + .transition_object(bucket, object, &transition_options(&original, tier_name)) + .await + .expect_err("a backend requiring unversioned objects must reject a UUID-shaped version"); + + let put_versions = backend.put_versions().await; + let removed_versions = backend.remove_versions().await; + assert_eq!(removed_versions, put_versions); + assert_eq!(removed_versions.len(), 1); + assert_eq!(removed_versions[0].1, remote_version); + assert_eq!(backend.object_count().await, 0); + assert_local_source_intact(&set_disks, bucket, object, &payload).await; + } + + #[tokio::test] + #[serial_test::serial] + async fn backend_version_constraint_cleans_nil_uuid_with_exact_version() { + let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; + let bucket = "transition-nil-version-constraint-bucket"; + let object = "object.bin"; + let payload = b"a fresh nil UUID response remains an exact cleanup constraint".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::nil().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; + backend.set_reject_non_empty_remote_versions(true); + + set_disks + .transition_object(bucket, object, &transition_options(&original, tier_name)) + .await + .expect_err("a backend requiring an empty version must reject a nil UUID string"); + + let put_versions = backend.put_versions().await; + let removed_versions = backend.remove_versions().await; + assert_eq!(removed_versions.len(), 1); + assert_eq!(removed_versions[0].0, put_versions[0].0); + assert_eq!(removed_versions[0].1, remote_version); + assert_eq!(backend.exact_remove_count(), 1); + assert_eq!(backend.object_count().await, 0); + assert_local_source_intact(&set_disks, bucket, object, &payload).await; + } + #[tokio::test] #[serial_test::serial] async fn remote_cleanup_failure_after_version_rejection_preserves_source_and_candidate() { diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index e6e090364..2c6869eb5 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -208,6 +208,8 @@ impl ECStore { ctx: CancellationToken, instance_ctx: Arc, ) -> Result> { + instance_ctx.bind_background_cancel_token(ctx.clone()); + // let layouts = DisksLayout::from_volumes(endpoints.as_slice())?; // Validate topology and environment overrides before opening any disk. @@ -524,6 +526,27 @@ mod tests { should_auto_start_rebalance_after_init, should_auto_start_rebalance_after_recovered_meta, should_resume_local_decommission, should_retry_local_decommission_resume, wait_for_local_decommission_resume_delay, }; + #[cfg(feature = "test-util")] + use crate::{ + bucket::lifecycle::{ + lifecycle::{TRANSITION_PENDING, TransitionOptions}, + tier_delete_journal::{ + TIER_DELETE_JOURNAL_PREFIX, persist_tier_delete_journal_entry, recover_tier_delete_journal_entries, + }, + tier_sweeper::Jentry, + }, + 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, + }, + storage_api_contracts::{ + bucket::{BucketOperations as _, MakeBucketOptions}, + list::ListOperations as _, + object::ObjectOperations as _, + }, + }; use crate::{ core::pools::{POOL_META_VERSION, PoolDecommissionInfo, PoolMeta, PoolStatus}, disk::endpoint::Endpoint, @@ -898,7 +921,24 @@ mod tests { temp_dir: &std::path::Path, cmd_line: &str, pool_drive_counts: &[usize], - ) -> (Arc, Arc) { + ) -> ( + Arc, + Arc, + CancellationToken, + ) { + build_isolated_test_store_with_shutdown(temp_dir, cmd_line, pool_drive_counts, CancellationToken::new()).await + } + + async fn build_isolated_test_store_with_shutdown( + temp_dir: &std::path::Path, + cmd_line: &str, + pool_drive_counts: &[usize], + shutdown: CancellationToken, + ) -> ( + Arc, + Arc, + CancellationToken, + ) { let mut pools = Vec::with_capacity(pool_drive_counts.len()); for (pool_index, &drives_per_set) in pool_drive_counts.iter().enumerate() { let mut endpoints = Vec::with_capacity(drives_per_set); @@ -930,13 +970,43 @@ mod tests { let store = crate::store::ECStore::new_with_instance_ctx( "127.0.0.1:0".parse().expect("test address"), endpoint_pools, - CancellationToken::new(), + shutdown.clone(), instance_ctx.clone(), ) .await .expect("store should build around the fresh context"); - (instance_ctx, store) + (instance_ctx, store, shutdown) + } + + #[cfg(feature = "test-util")] + async fn tier_delete_journal_count(store: Arc) -> usize { + store + .list_objects_v2(RUSTFS_META_BUCKET, TIER_DELETE_JOURNAL_PREFIX, None, None, 100, false, None, false) + .await + .expect("tier delete journal should be listable") + .objects + .len() + } + + #[cfg(feature = "test-util")] + async fn wait_for_tier_delete_journal_recovery( + store: Arc, + backend: &MockWarmBackend, + expected_removes: usize, + ) { + tokio::time::timeout(Duration::from_secs(30), async { + loop { + if backend.remove_versions().await.len() >= expected_removes + && tier_delete_journal_count(store.clone()).await == 0 + { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("tier delete journal recovery should complete"); } // Phase 5 follow-up (backlog#1052): building a real store through the @@ -948,7 +1018,7 @@ mod tests { #[serial_test::serial(storage_class_env)] async fn new_with_instance_ctx_threads_context_through_store_graph() { let temp_dir = tempfile::tempdir().expect("create temp store dir"); - let (instance_ctx, store) = + let (instance_ctx, store, _shutdown) = without_storage_class_env(build_isolated_test_store(temp_dir.path(), "instance-ctx-store-graph-test", &[4])).await; assert!( @@ -989,7 +1059,7 @@ mod tests { #[serial_test::serial(storage_class_env)] async fn new_with_instance_ctx_applies_default_parity_to_each_real_pool() { let temp_dir = tempfile::tempdir().expect("create multi-pool store dir"); - let (_, store) = + let (_, store, _shutdown) = without_storage_class_env(build_isolated_test_store(temp_dir.path(), "pool-parity-regression", &[4, 2])).await; assert_eq!(store.pools.len(), 2); @@ -1008,9 +1078,9 @@ mod tests { async fn two_stores_initialize_their_own_bucket_metadata_sys() { let temp_a = tempfile::tempdir().expect("create temp store dir a"); let temp_b = tempfile::tempdir().expect("create temp store dir b"); - let (ctx_a, store_a) = + let (ctx_a, store_a, _shutdown_a) = without_storage_class_env(build_isolated_test_store(temp_a.path(), "bucket-metadata-isolation-a", &[4])).await; - let (ctx_b, store_b) = + let (ctx_b, store_b, _shutdown_b) = without_storage_class_env(build_isolated_test_store(temp_b.path(), "bucket-metadata-isolation-b", &[4])).await; crate::bucket::metadata_sys::init_bucket_metadata_sys(store_a.clone(), Vec::new()).await; @@ -1025,4 +1095,235 @@ mod tests { .expect("store B's context must hold its metadata system"); assert!(!Arc::ptr_eq(&sys_a, &sys_b), "each store must own a distinct bucket metadata system"); } + + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn tier_delete_journal_recovery_spawns_for_each_store() { + let temp_a = tempfile::tempdir().expect("create temp store dir a"); + let temp_b = tempfile::tempdir().expect("create temp store dir b"); + let (ctx_a, store_a, shutdown_a) = + without_storage_class_env(build_isolated_test_store(temp_a.path(), "tier-journal-recovery-a", &[4])).await; + let (ctx_b, store_b, shutdown_b) = + without_storage_class_env(build_isolated_test_store(temp_b.path(), "tier-journal-recovery-b", &[4])).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store_a.clone(), Vec::new()).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store_b.clone(), Vec::new()).await; + + assert!( + !ctx_a.mark_tier_delete_journal_recovery_started(store_a.id), + "store A should have claimed its production recovery worker" + ); + assert!( + !ctx_b.mark_tier_delete_journal_recovery_started(store_b.id), + "store B should have claimed its production recovery worker" + ); + assert!(!shutdown_a.is_cancelled()); + assert!(!shutdown_b.is_cancelled()); + + let tier_a = "JOURNAL-A"; + let tier_b = "JOURNAL-B"; + let backend_a = register_mock_tier(&ctx_a.tier_config_mgr(), tier_a).await; + let backend_b = register_mock_tier(&ctx_b.tier_config_mgr(), tier_b).await; + let identity_a = TierConfigMgr::acquire_operation_lease(&ctx_a.tier_config_mgr(), tier_a) + .await + .expect("store A tier lease should resolve") + .backend_identity(); + let identity_b = TierConfigMgr::acquire_operation_lease(&ctx_b.tier_config_mgr(), tier_b) + .await + .expect("store B tier lease should resolve") + .backend_identity(); + let entry_a = Jentry { + obj_name: "remote-a".to_string(), + version_id: "version-a".to_string(), + tier_name: tier_a.to_string(), + backend_identity: Some(identity_a), + version_id_exact: false, + }; + let entry_b = Jentry { + obj_name: "remote-b".to_string(), + version_id: "version-b".to_string(), + tier_name: tier_b.to_string(), + backend_identity: Some(identity_b), + version_id_exact: false, + }; + let remove_a = backend_a.arm_failing_remove_barrier().await; + persist_tier_delete_journal_entry(store_a.clone(), &entry_a) + .await + .expect("store A journal should persist"); + persist_tier_delete_journal_entry(store_b.clone(), &entry_b) + .await + .expect("store B journal should persist"); + + ctx_a.wake_tier_delete_journal_recovery(); + ctx_b.wake_tier_delete_journal_recovery(); + remove_a.wait_until_paused().await; + wait_for_tier_delete_journal_recovery(store_b.clone(), &backend_b, 1).await; + + shutdown_a.cancel(); + remove_a.wait_until_operation_dropped().await; + assert!( + ctx_a + .background_cancel_token() + .expect("store A shutdown token should be bound") + .is_cancelled() + ); + assert!( + !ctx_b + .background_cancel_token() + .expect("store B shutdown token should be bound") + .is_cancelled(), + "cancelling store A must not stop store B" + ); + assert_eq!(tier_delete_journal_count(store_a.clone()).await, 1); + + let recovered_a = recover_tier_delete_journal_entries(store_a.clone(), 100, None) + .await + .expect("the cancelled store A worker must leave its journal recoverable"); + assert_eq!((recovered_a.scanned, recovered_a.deleted, recovered_a.failed), (1, 1, 0)); + assert_eq!(backend_a.remove_versions().await, vec![("remote-a".to_string(), "version-a".to_string())]); + + let second_entry_b = Jentry { + obj_name: "remote-b-2".to_string(), + version_id: "version-b-2".to_string(), + ..entry_b + }; + persist_tier_delete_journal_entry(store_b.clone(), &second_entry_b) + .await + .expect("store B second journal should persist"); + ctx_b.wake_tier_delete_journal_recovery(); + wait_for_tier_delete_journal_recovery(store_b.clone(), &backend_b, 2).await; + assert_eq!( + backend_b.remove_versions().await, + vec![ + ("remote-b".to_string(), "version-b".to_string()), + ("remote-b-2".to_string(), "version-b-2".to_string()), + ] + ); + + shutdown_b.cancel(); + } + + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn cancelled_transition_cleanup_journals_to_its_own_instance_store() { + struct ResolverReset(Arc>>>); + + impl Drop for ResolverReset { + fn drop(&mut self) { + *self.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner) = None; + } + } + + let temp_a = tempfile::tempdir().expect("create transition store dir a"); + let temp_b = tempfile::tempdir().expect("create transition store dir b"); + let shutdown_a = CancellationToken::new(); + let shutdown_b = CancellationToken::new(); + shutdown_a.cancel(); + shutdown_b.cancel(); + let (ctx_a, store_a, shutdown_a) = without_storage_class_env(build_isolated_test_store_with_shutdown( + temp_a.path(), + "transition-cleanup-context-a", + &[4], + shutdown_a, + )) + .await; + let (ctx_b, store_b, shutdown_b) = without_storage_class_env(build_isolated_test_store_with_shutdown( + temp_b.path(), + "transition-cleanup-context-b", + &[4], + shutdown_b, + )) + .await; + assert!(shutdown_a.is_cancelled()); + assert!(shutdown_b.is_cancelled()); + crate::bucket::metadata_sys::init_bucket_metadata_sys(store_a.clone(), Vec::new()).await; + + let resolver_target = Arc::new(std::sync::Mutex::new(Some(Arc::downgrade(&store_b)))); + let resolver_store = resolver_target.clone(); + assert!( + set_object_store_resolver(Arc::new(move || { + resolver_store + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .as_ref() + .and_then(std::sync::Weak::upgrade) + })), + "the cross-context regression test must install the only process object-store resolver" + ); + let _resolver_reset = ResolverReset(resolver_target); + assert!( + runtime_sources::object_store_handle().is_some_and(|store| Arc::ptr_eq(&store, &store_b)), + "the process resolver must deliberately point at store B" + ); + + let tier_name = "CROSSCTXA"; + let backend = register_mock_tier(&ctx_a.tier_config_mgr(), tier_name).await; + backend.set_put_remote_version(Some(uuid::Uuid::new_v4().to_string())).await; + backend.set_reject_non_empty_remote_versions(true); + let remove_barrier = backend.arm_failing_remove_barrier().await; + + let bucket = "transition-cleanup-context-a"; + let object = "rejected-candidate.bin"; + store_a + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("store A bucket should be created"); + let mut reader = PutObjReader::from_vec(b"cross-context rejected transition cleanup".repeat(1024)); + let original = store_a + .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .await + .expect("store A 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("the source object should have an ETag"), + ..Default::default() + }, + version_id: original.version_id.map(|version| version.to_string()), + mod_time: original.mod_time, + ..Default::default() + }; + + let cleanup_store_barrier = TransitionCleanupStoreBarrier::install(); + let transition_store = store_a.clone(); + let transition = tokio::spawn(async move { transition_store.transition_object(bucket, object, &opts).await }); + cleanup_store_barrier.wait_until_paused().await; + transition.abort(); + assert!( + transition + .await + .expect_err("the transition task should observe cancellation") + .is_cancelled() + ); + + remove_barrier.wait_until_paused().await; + let journal_counts = ( + tier_delete_journal_count(store_a.clone()).await, + tier_delete_journal_count(store_b.clone()).await, + ); + assert_eq!( + journal_counts, + (1, 0), + "the journal must land only on store A even while the process resolver points at store B" + ); + assert_eq!(backend.object_count().await, 1, "failed cleanup should retain the remote candidate"); + remove_barrier.release(); + remove_barrier.wait_until_operation_dropped().await; + + let recovered = recover_tier_delete_journal_entries(store_a.clone(), 100, None) + .await + .expect("store A should recover its own cancelled-transition journal"); + assert_eq!((recovered.scanned, recovered.deleted, recovered.failed), (1, 1, 0)); + assert_eq!(tier_delete_journal_count(store_a.clone()).await, 0); + assert_eq!(tier_delete_journal_count(store_b.clone()).await, 0); + assert_eq!( + backend.object_count().await, + 0, + "store A recovery should delete the exact remote candidate" + ); + assert!(!Arc::ptr_eq(&ctx_a, &ctx_b), "the regression requires two distinct instance contexts"); + } } diff --git a/crates/scanner/tests/lifecycle_integration_test.rs b/crates/scanner/tests/lifecycle_integration_test.rs index 1039b09a1..abec22392 100644 --- a/crates/scanner/tests/lifecycle_integration_test.rs +++ b/crates/scanner/tests/lifecycle_integration_test.rs @@ -41,10 +41,11 @@ use storage_api::lifecycle::{ BUCKET_LIFECYCLE_CONFIG, BucketOperations, BucketOptions, BucketVersioningSys, CompletePart, DiskOption, ECStore, EcstoreError, Endpoint, EndpointServerPools, Endpoints, IlmAction, LcEvent, LcEventSrc, ListOperations as _, MakeBucketOptions, MockWarmBackend, MultipartOperations as _, ObjectIO as _, ObjectOperations as _, PoolEndpoints, - STORAGE_FORMAT_FILE, TransitionOptions, assert_transition_meta_consistent, enqueue_transition_for_existing_objects, - expire_transitioned_object, free_version_count, get_bucket_metadata, get_global_tier_config_mgr, init_background_expiry, - init_bucket_metadata_sys, init_local_disks, is_err_object_not_found, is_err_version_not_found, new_disk, - path2_bucket_object_with_base_path, register_mock_tier_util, update_bucket_metadata, wait_for_free_version_absence, + STORAGE_FORMAT_FILE, TRANSITION_PENDING, TransitionCleanupStoreBarrier, TransitionOptions, assert_transition_meta_consistent, + enqueue_transition_for_existing_objects, expire_transitioned_object, free_version_count, get_bucket_metadata, + get_global_tier_config_mgr, init_background_expiry, init_bucket_metadata_sys, init_local_disks, is_err_object_not_found, + is_err_version_not_found, new_disk, path2_bucket_object_with_base_path, recover_tier_delete_journal_entries, + register_mock_tier_util, update_bucket_metadata, wait_for_free_version_absence, }; static GLOBAL_ENV: OnceLock<(Vec, Arc)> = OnceLock::new(); @@ -603,6 +604,7 @@ mod serial_tests { let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); let backend = register_mock_tier(&tier_name).await; + backend.set_put_remote_version(Some(String::new())).await; let bucket_name = format!("test-expire-get-race-{}", &Uuid::new_v4().simple().to_string()[..8]); let object_name = "test/race-object.bin"; @@ -639,6 +641,10 @@ mod serial_tests { .get_object_info(bucket_name.as_str(), object_name, &ObjectOptions::default()) .await .expect("Failed to load transitioned object info"); + assert!( + oi.transitioned_object.version_id.is_empty(), + "the regression must exercise an unversioned remote tier" + ); // Concurrent GET loop: hammer GET while the expiry runs. Every outcome // must be a full correct body or a clean not-found -- never a tier-fetch @@ -732,6 +738,327 @@ mod serial_tests { ); } + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + #[serial] + #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial"] + async fn rejected_transition_candidate_is_recovered_from_persisted_delete_journal() { + let (_disk_paths, ecstore) = setup_isolated_test_env(false).await; + + let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); + let backend = register_mock_tier(&tier_name).await; + backend.set_put_read_limit(Some(4096)).await; + backend.set_put_remote_version(Some(String::new())).await; + backend.set_remove_failure(true); + + let bucket_name = format!("test-transition-cleanup-journal-{}", &Uuid::new_v4().simple().to_string()[..8]); + let object_name = "test/rejected-candidate.bin"; + let payload = b"rejected remote candidate must not replace the local source".repeat(1024); + + create_test_bucket(&ecstore, bucket_name.as_str()).await; + upload_test_object(&ecstore, bucket_name.as_str(), object_name, &payload).await; + let original = ecstore + .get_object_info(bucket_name.as_str(), object_name, &ObjectOptions::default()) + .await + .expect("source object metadata should resolve before transition"); + let opts = ObjectOptions { + no_lock: true, + transition: TransitionOptions { + status: TRANSITION_PENDING.to_string(), + tier: tier_name, + etag: original.etag.clone().expect("uploaded source object should have an ETag"), + ..Default::default() + }, + version_id: original.version_id.map(|version| version.to_string()), + mod_time: original.mod_time, + ..Default::default() + }; + + ecstore + .transition_object(bucket_name.as_str(), object_name, &opts) + .await + .expect_err("a remotely accepted partial upload must not commit transition metadata"); + + let put_versions = backend.put_versions().await; + assert_eq!(put_versions.len(), 1, "transition should create exactly one remote candidate"); + assert!(put_versions[0].1.is_empty()); + assert_eq!( + backend.object_count().await, + 1, + "failed cleanup must retain the remote candidate for recovery" + ); + assert!( + backend.remove_versions().await.is_empty(), + "no cleanup path may delete the candidate while remove failures are enabled" + ); + + let retained = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) + .await + .expect("tier delete journal recovery should scan the persisted candidate"); + assert_eq!(retained.scanned, 1); + assert_eq!(retained.deleted, 0); + assert_eq!(retained.failed, 1); + assert_eq!(backend.object_count().await, 1, "failed recovery must retain the remote candidate"); + + backend.set_remove_failure(false); + let recovered = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) + .await + .expect("tier delete journal recovery should delete the retained candidate"); + assert_eq!(recovered.scanned, 1); + assert_eq!(recovered.deleted, 1); + assert_eq!(recovered.failed, 0); + let removed_versions = backend.remove_versions().await; + assert!(!removed_versions.is_empty(), "recovery must issue at least one successful delete"); + assert!( + removed_versions.iter().all(|removed| removed == &put_versions[0]), + "every idempotent cleanup must delete the exact PUT object and version" + ); + assert_eq!(backend.object_count().await, 0, "recovery should remove the rejected remote candidate"); + + let empty = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) + .await + .expect("a removed tier delete journal entry should no longer be listed"); + assert_eq!(empty.scanned, 0, "successful recovery must remove the persisted journal entry"); + assert_eq!( + read_object_fully(&ecstore, bucket_name.as_str(), object_name).await, + payload, + "rejected transition cleanup must leave the source object readable" + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + #[serial] + #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial"] + async fn cancelled_before_cleanup_store_resolution_persists_journal() { + let (_disk_paths, ecstore) = setup_isolated_test_env(false).await; + let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); + let backend = register_mock_tier(&tier_name).await; + backend.set_put_remote_version(Some(Uuid::new_v4().to_string())).await; + backend.set_reject_non_empty_remote_versions(true); + backend.set_remove_failure(true); + let cleanup_store_barrier = TransitionCleanupStoreBarrier::install(); + + let bucket_name = format!("test-transition-cancel-cleanup-{}", &Uuid::new_v4().simple().to_string()[..8]); + let object_name = "test/rejected-candidate.bin"; + let payload = b"cancelled rejected cleanup must retain durable recovery evidence".repeat(1024); + create_test_bucket(&ecstore, bucket_name.as_str()).await; + upload_test_object(&ecstore, bucket_name.as_str(), object_name, &payload).await; + let original = ecstore + .get_object_info(bucket_name.as_str(), object_name, &ObjectOptions::default()) + .await + .expect("source object metadata should resolve before transition"); + let opts = ObjectOptions { + no_lock: true, + transition: TransitionOptions { + status: TRANSITION_PENDING.to_string(), + tier: tier_name, + etag: original.etag.clone().expect("uploaded source object should have an ETag"), + ..Default::default() + }, + version_id: original.version_id.map(|version| version.to_string()), + mod_time: original.mod_time, + ..Default::default() + }; + + let transition_store = ecstore.clone(); + let transition_bucket = bucket_name.clone(); + let transition = tokio::spawn(async move { + transition_store + .transition_object(transition_bucket.as_str(), object_name, &opts) + .await + }); + cleanup_store_barrier.wait_until_paused().await; + transition.abort(); + assert!( + transition + .await + .expect_err("aborted transition task should be cancelled") + .is_cancelled() + ); + + let retained = tokio::time::timeout(Duration::from_secs(30), async { + loop { + let recovery = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) + .await + .expect("the cancelled transition journal should be readable"); + if recovery.scanned > 0 { + break recovery; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("Drop should persist the rejected candidate through the saved instance context"); + assert_eq!((retained.scanned, retained.deleted, retained.failed), (1, 0, 1)); + tokio::time::timeout(Duration::from_secs(5), async { + while backend.exact_remove_count() < 2 { + tokio::task::yield_now().await; + } + }) + .await + .expect("Drop cleanup and failed journal recovery must both preserve the exact version constraint"); + let failed_exact_attempts = backend.exact_remove_count(); + assert_eq!(backend.object_count().await, 1); + assert!(backend.remove_versions().await.is_empty()); + + backend.set_remove_failure(false); + let recovered = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) + .await + .expect("recovery should delete the candidate retained by the cancelled transition"); + assert_eq!((recovered.scanned, recovered.deleted, recovered.failed), (1, 1, 0)); + assert_eq!(backend.remove_versions().await, backend.put_versions().await); + assert_eq!(backend.exact_remove_count(), failed_exact_attempts + 1); + assert_eq!(backend.object_count().await, 0); + let empty = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) + .await + .expect("successful recovery should remove the cancellation journal"); + assert_eq!(empty.scanned, 0); + assert_eq!( + read_object_fully(&ecstore, bucket_name.as_str(), object_name).await, + payload, + "cancelled transition cleanup must preserve the local source" + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] + #[serial] + #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial"] + async fn rejected_transition_cleanup_durability_matrix() { + #[derive(Clone, Copy)] + enum CleanupCase { + Persisted, + DeleteFallback, + RetryPersisted, + FullyFailed, + } + + let (_disk_paths, ecstore) = setup_isolated_test_env(false).await; + + for case in [ + CleanupCase::Persisted, + CleanupCase::DeleteFallback, + CleanupCase::RetryPersisted, + CleanupCase::FullyFailed, + ] { + let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase(); + let backend = register_mock_tier(&tier_name).await; + let remote_version = Uuid::new_v4().to_string(); + backend.set_put_remote_version(Some(remote_version.clone())).await; + backend.set_reject_non_empty_remote_versions(true); + backend.set_remove_failure(matches!(case, CleanupCase::FullyFailed)); + let put_barrier = backend.arm_put_barrier().await; + let remove_barrier = if matches!(case, CleanupCase::RetryPersisted) { + Some(backend.arm_failing_remove_barrier().await) + } else { + None + }; + + let bucket_name = format!("test-transition-journal-failure-{}", &Uuid::new_v4().simple().to_string()[..8]); + let object_name = "test/rejected-candidate.bin"; + let payload = b"journal failure must either delete the exact candidate or report both failures".repeat(1024); + create_test_bucket(&ecstore, bucket_name.as_str()).await; + upload_test_object(&ecstore, bucket_name.as_str(), object_name, &payload).await; + let original = ecstore + .get_object_info(bucket_name.as_str(), object_name, &ObjectOptions::default()) + .await + .expect("source object metadata should resolve before transition"); + let opts = ObjectOptions { + no_lock: true, + transition: TransitionOptions { + status: TRANSITION_PENDING.to_string(), + tier: tier_name, + etag: original.etag.clone().expect("uploaded source object should have an ETag"), + ..Default::default() + }, + version_id: original.version_id.map(|version| version.to_string()), + mod_time: original.mod_time, + ..Default::default() + }; + + let transition_store = ecstore.clone(); + let transition_bucket = bucket_name.clone(); + let transition = tokio::spawn(async move { + transition_store + .transition_object(transition_bucket.as_str(), object_name, &opts) + .await + }); + put_barrier.wait_until_paused().await; + let set = ecstore.pools[0].get_disks(0); + let mut saved_disks = if matches!(case, CleanupCase::Persisted) { + None + } else { + let mut disks = set.disks.write().await; + let saved = std::mem::take(&mut *disks); + *disks = vec![None; saved.len()]; + Some(saved) + }; + put_barrier.release(); + if let Some(remove_barrier) = remove_barrier.as_ref() { + remove_barrier.wait_until_paused().await; + *set.disks.write().await = saved_disks.take().expect("offline disks should be restorable"); + backend.set_remove_failure(true); + remove_barrier.release(); + } + let transition_result = tokio::time::timeout(Duration::from_secs(30), transition).await; + if let Some(saved_disks) = saved_disks { + *set.disks.write().await = saved_disks; + } + let err = transition_result + .expect("transition should finish while validating cleanup durability") + .expect("transition task should not panic") + .expect_err("a versioned candidate must not commit to an unversioned tier"); + + match case { + CleanupCase::Persisted | CleanupCase::DeleteFallback => { + assert_eq!(backend.remove_versions().await, backend.put_versions().await); + assert_eq!(backend.object_count().await, 0, "cleanup must remove the exact candidate"); + let recovery = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) + .await + .expect("successful cleanup must not retain a journal entry"); + assert_eq!(recovery.scanned, 0); + } + CleanupCase::RetryPersisted => { + assert!( + !err.to_string().contains("journal retry error"), + "a successful journal retry must preserve the original version-constraint error" + ); + assert_eq!(backend.object_count().await, 1); + let retained = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) + .await + .expect("the retried journal should be recoverable"); + assert_eq!((retained.scanned, retained.deleted, retained.failed), (1, 0, 1)); + backend.set_remove_failure(false); + let recovered = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) + .await + .expect("recovery should delete the exact retried candidate"); + assert_eq!((recovered.scanned, recovered.deleted, recovered.failed), (1, 1, 0)); + assert_eq!(backend.remove_versions().await, backend.put_versions().await); + assert_eq!(backend.object_count().await, 0); + let empty = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) + .await + .expect("successful recovery must remove the retried journal"); + assert_eq!(empty.scanned, 0); + } + CleanupCase::FullyFailed => { + let message = err.to_string(); + assert!(message.contains("initial journal error"), "{message}"); + assert!(message.contains("cleanup error"), "{message}"); + assert!(message.contains("journal retry error"), "{message}"); + assert_eq!(backend.object_count().await, 1, "both failed safeguards must leave the candidate visible"); + assert!(backend.remove_versions().await.is_empty()); + let recovery = recover_tier_delete_journal_entries(ecstore.clone(), 100, None) + .await + .expect("failed journal writes must not create partial recovery entries"); + assert_eq!(recovery.scanned, 0); + } + } + assert_eq!( + read_object_fully(&ecstore, bucket_name.as_str(), object_name).await, + payload, + "rejected transition cleanup must preserve the local source" + ); + } + } + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[serial] #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"] diff --git a/crates/scanner/tests/storage_api/mod.rs b/crates/scanner/tests/storage_api/mod.rs index cc653eeae..aa066fcfb 100644 --- a/crates/scanner/tests/storage_api/mod.rs +++ b/crates/scanner/tests/storage_api/mod.rs @@ -12,10 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. +pub(crate) use rustfs_ecstore::api::bucket::lifecycle::tier_delete_journal::recover_tier_delete_journal_entries; pub(crate) use rustfs_ecstore::api::bucket::lifecycle::{ bucket_lifecycle_audit::LcEventSrc, bucket_lifecycle_ops::{enqueue_transition_for_existing_objects, expire_transitioned_object, init_background_expiry}, - lifecycle::{Event as LcEvent, IlmAction, TransitionOptions}, + lifecycle::{Event as LcEvent, IlmAction, TRANSITION_PENDING, TransitionOptions}, }; pub(crate) use rustfs_ecstore::api::bucket::metadata::BUCKET_LIFECYCLE_CONFIG; pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::{ @@ -32,8 +33,8 @@ pub(crate) use rustfs_ecstore::api::storage::{ECStore, init_local_disks}; // backend and xl.meta assertion helpers now live in ecstore behind the // `test-util` feature instead of being copied into this crate. pub(crate) use rustfs_ecstore::api::tier::test_util::{ - MockWarmBackend, assert_transition_meta_consistent, free_version_count, register_mock_tier as register_mock_tier_util, - wait_for_free_version_absence, + MockWarmBackend, TransitionCleanupStoreBarrier, assert_transition_meta_consistent, free_version_count, + register_mock_tier as register_mock_tier_util, wait_for_free_version_absence, }; use rustfs_storage_api as storage_contracts; @@ -45,10 +46,11 @@ pub(crate) mod lifecycle { pub(crate) use super::{ BUCKET_LIFECYCLE_CONFIG, BucketVersioningSys, DiskOption, ECStore, EcstoreError, Endpoint, EndpointServerPools, - Endpoints, IlmAction, LcEvent, LcEventSrc, MockWarmBackend, PoolEndpoints, STORAGE_FORMAT_FILE, TransitionOptions, - assert_transition_meta_consistent, enqueue_transition_for_existing_objects, expire_transitioned_object, - free_version_count, get_bucket_metadata, get_global_tier_config_mgr, init_background_expiry, init_bucket_metadata_sys, - init_local_disks, is_err_object_not_found, is_err_version_not_found, new_disk, path2_bucket_object_with_base_path, + Endpoints, IlmAction, LcEvent, LcEventSrc, MockWarmBackend, PoolEndpoints, STORAGE_FORMAT_FILE, TRANSITION_PENDING, + TransitionCleanupStoreBarrier, TransitionOptions, assert_transition_meta_consistent, + enqueue_transition_for_existing_objects, expire_transitioned_object, free_version_count, get_bucket_metadata, + get_global_tier_config_mgr, init_background_expiry, init_bucket_metadata_sys, init_local_disks, is_err_object_not_found, + is_err_version_not_found, new_disk, path2_bucket_object_with_base_path, recover_tier_delete_journal_entries, register_mock_tier_util, update_bucket_metadata, wait_for_free_version_absence, }; }