From 3e7987088dd1d713cb60bef07bf1e5c6d368fdec Mon Sep 17 00:00:00 2001 From: cxymds Date: Sun, 6 Sep 2026 11:12:37 +0800 Subject: [PATCH] feat(ilm): persist legacy tier recovery controls --- .../bucket/lifecycle/tier_delete_journal.rs | 195 ++++++++++++++++++ crates/ecstore/src/store/init.rs | 139 ++++++++++++- 2 files changed, 332 insertions(+), 2 deletions(-) diff --git a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs index 2270bce3b..ee503fe4f 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs @@ -35,6 +35,10 @@ use crate::bucket::lifecycle::config_boundary; use crate::bucket::lifecycle::durable_namespace::{ TIER_DELETE_JOURNAL_NAMESPACE, TIER_DELETE_JOURNAL_V6_NAMESPACE, validate_durable_ilm_record, }; +use crate::bucket::lifecycle::recovery_control::{ + IlmRecoveryClassification, IlmRecoveryControl, IlmRecoveryControlIdentity, IlmRecoveryErrorCode, IlmRecoveryProtocol, + load_recovery_control, observe_recovery_source, save_recovery_control_if_absent, +}; use crate::bucket::lifecycle::runtime_boundary; use crate::bucket::lifecycle::tier_sweeper::{ Jentry, TierDeleteDispatchBinding, TierDeleteJournalState, TierDeleteSourceIdentity, @@ -78,6 +82,13 @@ const TIER_DELETE_DISPATCH_MEMBER_DELETE_CONCURRENCY: usize = 32; const TIER_DELETE_DISPATCH_PREPARE_CONCURRENCY: usize = 16; const TIER_DELETE_DISPATCH_CAS_CONCURRENCY: usize = 32; const TIER_DELETE_JOURNAL_VERSION: u8 = 2; +const TIER_DELETE_JOURNAL_V1_RECOVERY_SCHEMA: &str = "rustfs-tier-delete-journal-v1"; +const TIER_DELETE_JOURNAL_V2_RECOVERY_SCHEMA: &str = "rustfs-tier-delete-journal-v2"; +const TIER_DELETE_JOURNAL_UNKNOWN_RECOVERY_SCHEMA: &str = "rustfs-tier-delete-journal-unknown"; +const TIER_DELETE_JOURNAL_V1_RECOVERY_CLASS: &str = "tier_delete_journal_v1"; +const TIER_DELETE_JOURNAL_V2_RECOVERY_CLASS: &str = "tier_delete_journal_v2"; +const TIER_DELETE_JOURNAL_CORRUPT_RECOVERY_CLASS: &str = "tier_delete_journal_corrupt"; +const CORRUPT_TIER_DELETE_JOURNAL_IDENTITY: &str = "corrupt"; const TIER_DELETE_JOURNAL_EXACT_VERSION: u8 = 3; const TIER_DELETE_JOURNAL_STATE_VERSION: u8 = 4; const TIER_DELETE_JOURNAL_TRANSACTION_VERSION: u8 = 5; @@ -5509,6 +5520,127 @@ enum TierDeleteJournalEntryRecoveryOutcome { Failed, } +fn canonical_legacy_tier_delete_journal_identity(object_name: &str) -> Option<&str> { + let identity = object_name + .strip_prefix(TIER_DELETE_JOURNAL_LEGACY_PREFIX)? + .strip_suffix(".json")?; + (rustfs_utils::crypto::is_sha256_checksum(identity) + && !identity + .bytes() + .any(|byte| byte.is_ascii_hexdigit() && byte.is_ascii_uppercase())) + .then_some(identity) +} + +fn legacy_tier_delete_recovery_descriptor(entry: &Jentry) -> Option<(&'static str, &'static str)> { + match entry.persisted_version { + 1 => Some((TIER_DELETE_JOURNAL_V1_RECOVERY_SCHEMA, TIER_DELETE_JOURNAL_V1_RECOVERY_CLASS)), + TIER_DELETE_JOURNAL_VERSION => Some((TIER_DELETE_JOURNAL_V2_RECOVERY_SCHEMA, TIER_DELETE_JOURNAL_V2_RECOVERY_CLASS)), + _ => None, + } +} + +fn legacy_tier_delete_control_matches( + control: &IlmRecoveryControl, + identity: &IlmRecoveryControlIdentity, + generation: &crate::bucket::lifecycle::recovery_control::IlmRecoverySourceGeneration, + classification: IlmRecoveryClassification, + error_code: IlmRecoveryErrorCode, +) -> bool { + control.identity == *identity + && control.observed_source_generation == *generation + && control.classification == classification + && control.last_error_code == error_code + && control.owner.is_none() + && control.attempt_count == 0 + && control.consecutive_failure_count == 0 +} + +fn legacy_tier_delete_control_is_scheduler_fence(control: &IlmRecoveryControl, identity: &IlmRecoveryControlIdentity) -> bool { + control.identity == *identity && control.owner.is_none() && !control.classification.permits_automatic_attempt() +} + +async fn persist_legacy_tier_delete_recovery_control( + api: Arc, + object_name: &str, + observed_data: &[u8], + stable_operation_identity: String, + source_schema: &'static str, + record_class: &'static str, + intended_classification: IlmRecoveryClassification, + intended_error_code: IlmRecoveryErrorCode, +) -> Result<()> { + let identity = IlmRecoveryControlIdentity { + protocol: IlmRecoveryProtocol::TierDeleteJournal, + canonical_source_path: object_name.to_string(), + stable_operation_identity, + record_class: record_class.to_string(), + }; + let control_id = identity.source_operation_digest().map_err(Error::other)?; + match load_recovery_control(api.clone(), IlmRecoveryProtocol::TierDeleteJournal, &control_id).await { + Ok(observed) if legacy_tier_delete_control_is_scheduler_fence(&observed.control, &identity) => return Ok(()), + Ok(_) => return Err(Error::PreconditionFailed), + Err(Error::ConfigNotFound) => {} + Err(err) => return Err(err), + } + + let source = observe_recovery_source(api.clone(), object_name, source_schema).await?; + let exact_source = source.is_consistent() && source.canonical_data.as_deref() == Some(observed_data); + let (classification, error_code) = if exact_source { + (intended_classification, intended_error_code) + } else { + (IlmRecoveryClassification::Corrupt, IlmRecoveryErrorCode::SourceDivergent) + }; + let candidate = IlmRecoveryControl::new( + identity.clone(), + source.generation.clone(), + classification, + i64::try_from(time::OffsetDateTime::now_utc().unix_timestamp_nanos()) + .map_err(|_| Error::other("tier delete journal recovery timestamp does not fit i64"))?, + error_code, + ) + .map_err(Error::other)?; + + match save_recovery_control_if_absent(api.clone(), &candidate).await { + Ok(()) | Err(Error::PreconditionFailed) => {} + Err(save_error) => match load_recovery_control(api.clone(), IlmRecoveryProtocol::TierDeleteJournal, &control_id).await { + Ok(observed) + if legacy_tier_delete_control_matches( + &observed.control, + &identity, + &source.generation, + classification, + error_code, + ) => + { + return Ok(()); + } + Ok(_) | Err(_) => return Err(save_error), + }, + } + + let observed = load_recovery_control(api, IlmRecoveryProtocol::TierDeleteJournal, &control_id).await?; + if !legacy_tier_delete_control_matches(&observed.control, &identity, &source.generation, classification, error_code) { + return Err(Error::PreconditionFailed); + } + Ok(()) +} + +async fn retain_corrupt_legacy_tier_delete_journal(api: Arc, object_name: &str, data: &[u8]) -> Result<()> { + canonical_legacy_tier_delete_journal_identity(object_name) + .ok_or_else(|| Error::other("tier delete journal path is not canonical"))?; + persist_legacy_tier_delete_recovery_control( + api, + object_name, + data, + CORRUPT_TIER_DELETE_JOURNAL_IDENTITY.to_string(), + TIER_DELETE_JOURNAL_UNKNOWN_RECOVERY_SCHEMA, + TIER_DELETE_JOURNAL_CORRUPT_RECOVERY_CLASS, + IlmRecoveryClassification::Corrupt, + IlmRecoveryErrorCode::SourceCorrupt, + ) + .await +} + async fn recover_tier_delete_journal_entry(api: Arc, object_name: String) -> TierDeleteJournalEntryRecoveryOutcome { let data = match config_boundary::read_config(api.clone(), &object_name).await { Ok(data) => data, @@ -5529,6 +5661,22 @@ async fn recover_tier_delete_journal_entry(api: Arc, object_name: Strin let je = match decode_tier_delete_journal_entry(&data) { Ok(je) => je, Err(err) => { + if canonical_legacy_tier_delete_journal_identity(&object_name).is_some() { + return match retain_corrupt_legacy_tier_delete_journal(api, &object_name, &data).await { + Ok(()) => TierDeleteJournalEntryRecoveryOutcome::Retained, + Err(control_error) => { + warn!( + event = EVENT_LIFECYCLE_TIER_DELETE_JOURNAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + journal_object = %object_name, + error = ?control_error, + "Failed to retain corrupt tier delete journal recovery control" + ); + TierDeleteJournalEntryRecoveryOutcome::Failed + } + }; + } warn!( event = EVENT_LIFECYCLE_TIER_DELETE_JOURNAL, component = LOG_COMPONENT_ECSTORE, @@ -5542,6 +5690,22 @@ async fn recover_tier_delete_journal_entry(api: Arc, object_name: Strin }; if tier_delete_journal_object_name(&je) != object_name { + if canonical_legacy_tier_delete_journal_identity(&object_name).is_some() { + return match retain_corrupt_legacy_tier_delete_journal(api, &object_name, &data).await { + Ok(()) => TierDeleteJournalEntryRecoveryOutcome::Retained, + Err(err) => { + warn!( + event = EVENT_LIFECYCLE_TIER_DELETE_JOURNAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + journal_object = %object_name, + error = ?err, + "Failed to retain mismatched tier delete journal recovery control" + ); + TierDeleteJournalEntryRecoveryOutcome::Failed + } + }; + } warn!( event = EVENT_LIFECYCLE_TIER_DELETE_JOURNAL, component = LOG_COMPONENT_ECSTORE, @@ -5552,6 +5716,37 @@ async fn recover_tier_delete_journal_entry(api: Arc, object_name: Strin return TierDeleteJournalEntryRecoveryOutcome::Failed; } + if let Some((source_schema, record_class)) = legacy_tier_delete_recovery_descriptor(&je) { + let stable_operation_identity = canonical_legacy_tier_delete_journal_identity(&object_name) + .expect("decoded legacy journal path was validated against its canonical object name") + .to_string(); + return match persist_legacy_tier_delete_recovery_control( + api, + &object_name, + &data, + stable_operation_identity, + source_schema, + record_class, + IlmRecoveryClassification::RetainedAmbiguous, + IlmRecoveryErrorCode::RemoteVersionUnknown, + ) + .await + { + Ok(()) => TierDeleteJournalEntryRecoveryOutcome::Retained, + Err(err) => { + warn!( + event = EVENT_LIFECYCLE_TIER_DELETE_JOURNAL, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + journal_object = %object_name, + error = ?err, + "Failed to retain legacy tier delete journal recovery control" + ); + TierDeleteJournalEntryRecoveryOutcome::Failed + } + }; + } + match api .durable_ilm_terminal_receipt_covers_active_source(&object_name, &data) .await diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 239cb20cc..86662f182 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -827,8 +827,8 @@ mod tests { }, recovery_control::{ IlmRecoveryClassification, IlmRecoveryControl, IlmRecoveryControlIdentity, IlmRecoveryErrorCode, - IlmRecoveryProtocol, MAX_RECOVERY_ATTEMPTS, load_recovery_control, observe_recovery_source, - save_recovery_control_if_absent, + IlmRecoveryProtocol, MAX_RECOVERY_ATTEMPTS, list_recovery_controls, load_recovery_control, + observe_recovery_source, save_recovery_control_if_absent, }, tier_delete_journal::{ DecommissionCheckpointTargetFailureHook, TIER_DELETE_DISPATCH_MANIFEST_PREFIX, TIER_DELETE_JOURNAL_PREFIX, @@ -16760,6 +16760,141 @@ mod tests { } } + #[cfg(feature = "test-util")] + #[tokio::test] + #[serial_test::serial(storage_class_env)] + async fn legacy_tier_delete_journals_create_redacted_recovery_controls_without_remote_calls() { + let temp_dir = tempfile::tempdir().expect("create legacy journal recovery store dir"); + let (ctx, store, _shutdown) = + without_storage_class_env(build_isolated_test_store(temp_dir.path(), "legacy-tier-journal-recovery", &[4])).await; + crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + + let tier_name = "LEGACY-RECOVERY"; + let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await; + let backend_identity = TierConfigMgr::acquire_operation_lease(&ctx.tier_config_mgr(), tier_name) + .await + .expect("legacy recovery tier lease should resolve") + .backend_identity(); + let fixtures = [ + serde_json::json!({ + "version": 1, + "obj_name": "legacy/remote-v1", + "version_id": "opaque-v1", + "tier_name": tier_name, + }), + serde_json::json!({ + "version": 2, + "obj_name": "legacy/remote-v2", + "version_id": "opaque-v2", + "tier_name": tier_name, + "backend_identity": backend_identity, + }), + ]; + let mut journal_paths = Vec::new(); + for fixture in &fixtures { + let data = serde_json::to_vec(&fixture).expect("legacy journal fixture should encode"); + let entry = crate::bucket::lifecycle::tier_delete_journal::decode_tier_delete_journal_entry(&data) + .expect("legacy journal fixture should decode"); + let path = tier_delete_journal_object_name(&entry); + com::save_config(store.clone(), &path, data) + .await + .expect("legacy journal fixture should persist"); + journal_paths.push(path); + } + let corrupt_path = format!( + "{TIER_DELETE_JOURNAL_PREFIX}/{}.json", + rustfs_utils::crypto::hex_sha256(b"corrupt legacy tier journal", ToOwned::to_owned) + ); + com::save_config(store.clone(), &corrupt_path, b"{corrupt".to_vec()) + .await + .expect("corrupt legacy journal fixture should persist"); + + let (first, concurrent) = tokio::join!( + recover_tier_delete_journal_entries(store.clone(), 100, None), + recover_tier_delete_journal_entries(store.clone(), 100, None), + ); + for stats in [first, concurrent] { + let stats = stats.expect("concurrent legacy journal recovery scan should finish"); + assert_eq!((stats.scanned, stats.deleted, stats.failed), (3, 0, 0)); + } + assert_eq!(tier_delete_journal_count(store.clone()).await, 3); + assert_eq!(backend.remove_count().await, 0, "legacy recovery must not call the remote tier"); + assert_eq!(backend.exact_remove_count(), 0, "legacy recovery must not issue exact remote DELETE"); + assert!(backend.op_log().await.is_empty(), "legacy recovery must not invoke any backend operation"); + + let mut first_controls = list_recovery_controls(store.clone(), IlmRecoveryProtocol::TierDeleteJournal, None, 100, None) + .await + .expect("legacy recovery controls should be listable") + .records; + first_controls.sort_by(|left, right| left.control_id.cmp(&right.control_id)); + assert_eq!(first_controls.len(), 3); + assert_eq!( + first_controls + .iter() + .filter(|control| control.classification == IlmRecoveryClassification::RetainedAmbiguous) + .count(), + 2 + ); + assert_eq!( + first_controls + .iter() + .filter(|control| control.classification == IlmRecoveryClassification::Corrupt) + .count(), + 1 + ); + for view in &first_controls { + assert_eq!(view.protocol, IlmRecoveryProtocol::TierDeleteJournal); + assert_eq!(view.revision, 1); + assert_eq!(view.attempt_count, 0); + let encoded = serde_json::to_string(view).expect("recovery control view should encode"); + for secret in ["legacy/remote-v1", "legacy/remote-v2", "opaque-v1", "opaque-v2", tier_name] { + assert!(!encoded.contains(secret), "recovery control view must redact `{secret}`"); + } + let persisted = load_recovery_control(store.clone(), IlmRecoveryProtocol::TierDeleteJournal, &view.control_id) + .await + .expect("legacy recovery control should load"); + match view.source_schema.as_str() { + "rustfs-tier-delete-journal-v1" => { + assert_eq!(persisted.control.identity.record_class, "tier_delete_journal_v1"); + assert_eq!(view.last_error_code, IlmRecoveryErrorCode::RemoteVersionUnknown); + } + "rustfs-tier-delete-journal-v2" => { + assert_eq!(persisted.control.identity.record_class, "tier_delete_journal_v2"); + assert_eq!(view.last_error_code, IlmRecoveryErrorCode::RemoteVersionUnknown); + } + "rustfs-tier-delete-journal-unknown" => { + assert_eq!(persisted.control.identity.record_class, "tier_delete_journal_corrupt"); + assert_eq!(view.last_error_code, IlmRecoveryErrorCode::SourceCorrupt); + } + schema => panic!("unexpected legacy recovery source schema: {schema}"), + } + } + + com::save_config( + store.clone(), + &journal_paths[0], + serde_json::to_vec_pretty(&fixtures[0]).expect("rewritten legacy journal fixture should encode"), + ) + .await + .expect("equivalent legacy journal rewrite should persist"); + let second = recover_tier_delete_journal_entries(store.clone(), 100, None) + .await + .expect("repeated legacy journal recovery scan should finish"); + assert_eq!((second.scanned, second.deleted, second.failed), (3, 0, 0)); + let mut second_controls = list_recovery_controls(store, IlmRecoveryProtocol::TierDeleteJournal, None, 100, None) + .await + .expect("repeated legacy recovery controls should remain listable") + .records; + second_controls.sort_by(|left, right| left.control_id.cmp(&right.control_id)); + assert_eq!(second_controls, first_controls, "repeated scans must not reset durable controls"); + assert_eq!(backend.remove_count().await, 0, "repeated recovery must remain remote-call free"); + assert_eq!(backend.exact_remove_count(), 0); + assert!( + backend.op_log().await.is_empty(), + "repeated recovery must not invoke any backend operation" + ); + } + #[cfg(feature = "test-util")] #[tokio::test] #[serial_test::serial(storage_class_env)]