feat(ilm): persist legacy tier recovery controls

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<ECStore>,
+    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<ECStore>, 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<ECStore>, 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<ECStore>, 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<ECStore>, 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<ECStore>, 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

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)]