feat(ilm): execute legacy recovery dispositions

crates/ecstore/src/api/mod.rs

@@ -78,10 +78,8 @@ pub mod bucket {
 
         pub mod recovery_disposition {
             pub use crate::bucket::lifecycle::recovery_disposition::{
-                CreatedIlmRecoveryDisposition, IlmRecoveryDisposition, IlmRecoveryDispositionAction, IlmRecoveryDispositionError,
-                IlmRecoveryDispositionIdentity, IlmRecoveryDispositionOwnerLease, IlmRecoveryDispositionReasonCode,
-                IlmRecoveryDispositionState, ObservedIlmRecoveryDisposition, create_recovery_disposition_if_absent,
-                load_recovery_disposition, recovery_disposition_id, save_recovery_disposition_if_current,
+                IlmRecoveryDispositionExecutionOutcome, IlmRecoveryDispositionReasonCode, IlmRecoveryDispositionState,
+                dry_run_recovery_disposition, execute_recovery_disposition,
             };
         }
 

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -32,6 +32,7 @@ use crate::bucket::lifecycle::manual_transition_job::{
     record_manual_transition_worker_result_with_reason, renew_manual_transition_job_lease_if_owned,
     save_manual_transition_job_record_if_current, save_manual_transition_task_if_absent, update_manual_transition_job_record,
 };
+use crate::bucket::lifecycle::recovery_disposition_runtime::run_recovery_disposition_maintenance_loop;
 use crate::bucket::lifecycle::replication_sink;
 use crate::bucket::lifecycle::replication_sink::{
     DeleteReplicationConfigSnapshot, ReplicationObjectBridge, ReplicationStatusType, replication_state_to_filemeta,
@@ -149,6 +150,7 @@ pub type ExpiryOpType = Box<dyn ExpiryOp + Send + Sync + 'static>;
 static XXHASH_SEED: u64 = 0;
 static TIER_FREE_VERSION_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
 static MANUAL_TRANSITION_JOB_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
+static RECOVERY_DISPOSITION_MAINTENANCE_STARTED: OnceLock<()> = OnceLock::new();
 
 #[cfg(test)]
 #[derive(Default)]
@@ -2398,9 +2400,20 @@ pub async fn init_background_expiry(api: Arc<ECStore>) {
     let _ = spawn_tier_free_version_recovery_once(api.clone(), &TIER_FREE_VERSION_RECOVERY_STARTED);
     spawn_tier_delete_journal_recovery_once(api.clone());
     spawn_transition_transaction_recovery_once(api.clone());
+    spawn_recovery_disposition_maintenance_once(api.clone());
     spawn_manual_transition_job_recovery_once(api);
 }
 
+fn spawn_recovery_disposition_maintenance_once(api: Arc<ECStore>) -> Option<JoinHandle<()>> {
+    let cancel_token = api.ctx.background_cancel_token()?;
+    if RECOVERY_DISPOSITION_MAINTENANCE_STARTED.set(()).is_err() {
+        return None;
+    }
+    Some(tokio::spawn(async move {
+        run_recovery_disposition_maintenance_loop(api, cancel_token).await;
+    }))
+}
+
 fn spawn_manual_transition_job_recovery_once(api: Arc<ECStore>) -> Option<JoinHandle<()>> {
     if MANUAL_TRANSITION_JOB_RECOVERY_STARTED.set(()).is_err() {
         return None;

crates/ecstore/src/bucket/lifecycle/config_boundary.rs

@@ -161,20 +161,30 @@ where
             DeletedObject = DeletedObject,
         >,
 {
-    match api
-        .delete_object(
-            RUSTFS_META_BUCKET,
-            file,
-            ObjectOptions {
-                http_preconditions: Some(HTTPPreconditions {
-                    if_match: Some(etag.to_string()),
-                    ..Default::default()
-                }),
-                ..Default::default()
-            },
-        )
-        .await
-    {
+    delete_config_if_match_with_opts(api, file, etag, ObjectOptions::default()).await
+}
+
+pub(crate) async fn delete_config_if_match_with_opts<S>(
+    api: Arc<S>,
+    file: &str,
+    etag: &str,
+    mut options: ObjectOptions,
+) -> Result<()>
+where
+    S: ObjectOperations<
+            Error = Error,
+            ObjectInfo = ObjectInfo,
+            ObjectOptions = ObjectOptions,
+            FileInfo = FileInfo,
+            ObjectToDelete = ObjectToDelete,
+            DeletedObject = DeletedObject,
+        >,
+{
+    options.http_preconditions = Some(HTTPPreconditions {
+        if_match: Some(etag.to_string()),
+        ..Default::default()
+    });
+    match api.delete_object(RUSTFS_META_BUCKET, file, options).await {
         Ok(_) => Ok(()),
         Err(err) => {
             if err == Error::FileNotFound || matches!(err, Error::ObjectNotFound(_, _)) {

crates/ecstore/src/bucket/lifecycle/mod.rs

@@ -26,6 +26,7 @@ mod object_lock_boundary;
 pub use self::core as lifecycle;
 pub mod recovery_control;
 pub mod recovery_disposition;
+pub(crate) mod recovery_disposition_runtime;
 pub mod recovery_export;
 mod replication_sink;
 pub mod rule;

crates/ecstore/src/bucket/lifecycle/recovery_control.rs

@@ -485,6 +485,20 @@ impl IlmRecoveryControl {
         self.validate()
     }
 
+    pub fn abandon_for_operator(&mut self, expected_source_generation: &IlmRecoverySourceGeneration) -> Result<()> {
+        if self.owner.is_some()
+            || self.classification != IlmRecoveryClassification::RetainedAmbiguous
+            || &self.observed_source_generation != expected_source_generation
+        {
+            return Err(IlmRecoveryControlError::InvalidSuccessor(
+                "operator abandonment requires the exact ownerless retained source generation",
+            ));
+        }
+        self.bump_revision()?;
+        self.classification = IlmRecoveryClassification::Abandoned;
+        self.validate()
+    }
+
     pub fn validate_successor(&self, next: &Self) -> Result<()> {
         self.validate()?;
         next.validate()?;
@@ -508,12 +522,34 @@ impl IlmRecoveryControl {
                 self.validate_failure_successor(next)
             }
             (Some(_), None) => self.validate_finish_successor(next),
+            (None, None)
+                if self.classification == IlmRecoveryClassification::RetainedAmbiguous
+                    && next.classification == IlmRecoveryClassification::Abandoned =>
+            {
+                self.validate_operator_abandon_successor(next)
+            }
             (None, None) => Err(IlmRecoveryControlError::InvalidSuccessor(
                 "ownerless control cannot advance without a claim",
             )),
         }
     }
 
+    fn validate_operator_abandon_successor(&self, next: &Self) -> Result<()> {
+        if next.observed_source_generation != self.observed_source_generation
+            || next.attempt_count != self.attempt_count
+            || next.consecutive_failure_count != self.consecutive_failure_count
+            || next.first_failure_at_unix_nanos != self.first_failure_at_unix_nanos
+            || next.last_failure_at_unix_nanos != self.last_failure_at_unix_nanos
+            || next.next_attempt_at_unix_nanos != self.next_attempt_at_unix_nanos
+            || next.last_error_code != self.last_error_code
+        {
+            return Err(IlmRecoveryControlError::InvalidSuccessor(
+                "operator abandonment changed recovery history or source generation",
+            ));
+        }
+        Ok(())
+    }
+
     fn validate_claim_successor(&self, next: &Self) -> Result<()> {
         if self.classification != IlmRecoveryClassification::Retrying
             || next.classification != IlmRecoveryClassification::Retrying
@@ -827,6 +863,23 @@ pub async fn observe_recovery_source(
     api: Arc<ECStore>,
     canonical_path: &str,
     source_schema: &str,
+) -> EcstoreResult<ObservedIlmRecoverySource> {
+    observe_recovery_source_with_options(api, canonical_path, source_schema, false).await
+}
+
+pub(crate) async fn observe_recovery_source_no_lock(
+    api: Arc<ECStore>,
+    canonical_path: &str,
+    source_schema: &str,
+) -> EcstoreResult<ObservedIlmRecoverySource> {
+    observe_recovery_source_with_options(api, canonical_path, source_schema, true).await
+}
+
+async fn observe_recovery_source_with_options(
+    api: Arc<ECStore>,
+    canonical_path: &str,
+    source_schema: &str,
+    no_lock: bool,
 ) -> EcstoreResult<ObservedIlmRecoverySource> {
     validate_canonical_source_path(canonical_path).map_err(recovery_control_store_error)?;
     if source_schema.trim().is_empty() {
@@ -837,7 +890,16 @@ pub async fn observe_recovery_source(
     let mut observations = Vec::new();
     for set in api.all_set_disks() {
         let authority = format!("pool-{}/set-{}", set.pool_index, set.set_index);
-        match config_boundary::read_config_with_metadata(set, canonical_path, &ObjectOptions::default()).await {
+        match config_boundary::read_config_with_metadata(
+            set,
+            canonical_path,
+            &ObjectOptions {
+                no_lock,
+                ..Default::default()
+            },
+        )
+        .await
+        {
             Ok((data, metadata)) => {
                 let etag = metadata
                     .etag
@@ -1255,6 +1317,36 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn operator_abandonment_is_an_exact_ownerless_retained_successor() {
+        let mut retained = IlmRecoveryControl::new(
+            control().identity,
+            generation(),
+            IlmRecoveryClassification::RetainedAmbiguous,
+            1_000_000_000,
+            IlmRecoveryErrorCode::OperatorDispositionRequired,
+        )
+        .expect("retained control should build");
+        let previous = retained.clone();
+        retained
+            .abandon_for_operator(&previous.observed_source_generation)
+            .expect("exact retained generation should be abandonable");
+        previous
+            .validate_successor(&retained)
+            .expect("operator abandonment should be a valid successor");
+        assert_eq!(retained.classification, IlmRecoveryClassification::Abandoned);
+        assert_eq!(retained.revision, previous.revision + 1);
+
+        let mut wrong_generation = previous.clone();
+        let mut generation = previous.observed_source_generation.clone();
+        generation.source_etag = "different".to_string();
+        assert!(wrong_generation.abandon_for_operator(&generation).is_err());
+
+        let mut mutated_history = retained.clone();
+        mutated_history.attempt_count += 1;
+        assert!(previous.validate_successor(&mutated_history).is_err());
+    }
+
     #[test]
     fn recovery_control_view_redacts_source_and_owner_details() {
         let mut control = control();

crates/ecstore/src/store/init.rs

@@ -828,8 +828,14 @@ mod tests {
             recovery_control::{
                 IlmRecoveryClassification, IlmRecoveryControl, IlmRecoveryControlIdentity, IlmRecoveryErrorCode,
                 IlmRecoveryProtocol, MAX_RECOVERY_ATTEMPTS, list_recovery_controls, load_recovery_control,
-                observe_recovery_source, save_recovery_control_if_absent,
+                observe_recovery_source, recovery_control_record_object_name, save_recovery_control_if_absent,
             },
+            recovery_disposition::{
+                IlmRecoveryDispositionExecutionOutcome, IlmRecoveryDispositionState, RecoveryDispositionCrashStage,
+                dry_run_recovery_disposition, execute_recovery_disposition, inject_recovery_disposition_crash_once,
+                load_recovery_disposition,
+            },
+            recovery_disposition_runtime::garbage_collect_completed_recovery_disposition,
             recovery_export::{
                 create_recovery_export, inspect_recovery_export_observation, load_recovery_export,
                 recovery_export_record_object_name,
@@ -16936,6 +16942,301 @@ mod tests {
         );
     }
 
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn legacy_recovery_disposition_removes_only_local_journals_and_replays() {
+        Box::pin(legacy_recovery_disposition_removes_only_local_journals_and_replays_case()).await;
+    }
+
+    #[cfg(feature = "test-util")]
+    async fn legacy_recovery_disposition_removes_only_local_journals_and_replays_case() {
+        let temp_dir = tempfile::tempdir().expect("create legacy disposition store dir");
+        let (ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "legacy-recovery-disposition", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+
+        let tier_name = "LEGACY-DISPOSITION";
+        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 disposition tier lease should resolve")
+            .backend_identity();
+        let fixtures = [
+            serde_json::json!({
+                "version": 1,
+                "obj_name": "legacy/disposition-v1",
+                "version_id": "opaque-disposition-v1",
+                "tier_name": tier_name,
+            }),
+            serde_json::json!({
+                "version": 2,
+                "obj_name": "legacy/disposition-v2",
+                "version_id": "opaque-disposition-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 disposition fixture should encode");
+            let entry = crate::bucket::lifecycle::tier_delete_journal::decode_tier_delete_journal_entry(&data)
+                .expect("legacy disposition fixture should decode");
+            let path = tier_delete_journal_object_name(&entry);
+            com::save_config(store.clone(), &path, data)
+                .await
+                .expect("legacy disposition fixture should persist");
+            journal_paths.push(path);
+        }
+
+        let recovered = recover_tier_delete_journal_entries(store.clone(), 100, None)
+            .await
+            .expect("legacy disposition recovery scan should finish");
+        assert_eq!((recovered.scanned, recovered.deleted, recovered.failed), (2, 0, 0));
+        assert_eq!(tier_delete_journal_count(store.clone()).await, 2);
+
+        let mut controls = list_recovery_controls(
+            store.clone(),
+            IlmRecoveryProtocol::TierDeleteJournal,
+            Some(IlmRecoveryClassification::RetainedAmbiguous),
+            100,
+            None,
+        )
+        .await
+        .expect("legacy disposition controls should be listable")
+        .records;
+        controls.sort_by(|left, right| left.control_id.cmp(&right.control_id));
+        assert_eq!(controls.len(), 2, "both legacy schemas must support disposition");
+
+        let actor_sha256 = rustfs_utils::crypto::hex_sha256(b"legacy-disposition-actor", ToOwned::to_owned);
+        let wrong_actor_sha256 = rustfs_utils::crypto::hex_sha256(b"different-disposition-actor", ToOwned::to_owned);
+        let wrong_export_sha256 = "ff".repeat(32);
+        for (index, control) in controls.iter().enumerate() {
+            let observation = inspect_recovery_export_observation(store.clone(), &control.control_id)
+                .await
+                .expect("legacy disposition source should be observable");
+            let export = create_recovery_export(store.clone(), &observation, &actor_sha256)
+                .await
+                .expect("legacy disposition export should persist");
+            let confirmed_at_unix_nanos = i64::try_from(OffsetDateTime::now_utc().unix_timestamp_nanos())
+                .expect("legacy disposition timestamp should fit i64");
+
+            if index == 0 {
+                let wrong_hash = Box::pin(execute_recovery_disposition(
+                    store.clone(),
+                    &observation,
+                    &export.export_id,
+                    &wrong_export_sha256,
+                    &actor_sha256,
+                    confirmed_at_unix_nanos,
+                ))
+                .await
+                .expect_err("a mismatched export checksum must fail before local deletion");
+                assert_eq!(wrong_hash, Error::PreconditionFailed);
+                assert_eq!(tier_delete_journal_count(store.clone()).await, 2);
+            }
+
+            let dry_run = dry_run_recovery_disposition(
+                store.clone(),
+                &observation,
+                &export.export_id,
+                &export.content_sha256,
+                &actor_sha256,
+                confirmed_at_unix_nanos,
+            )
+            .await
+            .expect("legacy disposition dry-run should validate exact local state");
+            assert_eq!(dry_run.source_copy_count, observation.source_generation.copies.len());
+            assert_eq!(
+                tier_delete_journal_count(store.clone()).await,
+                fixtures.len() - index,
+                "dry-run must not delete a legacy journal"
+            );
+            assert!(
+                matches!(
+                    load_recovery_disposition(store.clone(), IlmRecoveryProtocol::TierDeleteJournal, &dry_run.disposition_id,)
+                        .await,
+                    Err(Error::ConfigNotFound)
+                ),
+                "dry-run must not persist a disposition record"
+            );
+
+            if index == 0 {
+                inject_recovery_disposition_crash_once(RecoveryDispositionCrashStage::AfterLocalDelete);
+                Box::pin(execute_recovery_disposition(
+                    store.clone(),
+                    &observation,
+                    &export.export_id,
+                    &export.content_sha256,
+                    &actor_sha256,
+                    confirmed_at_unix_nanos,
+                ))
+                .await
+                .expect_err("the injected crash must stop after local delete commits");
+                assert_eq!(tier_delete_journal_count(store.clone()).await, 1);
+                let interrupted =
+                    load_recovery_disposition(store.clone(), IlmRecoveryProtocol::TierDeleteJournal, &dry_run.disposition_id)
+                        .await
+                        .expect("the applying disposition must survive the post-delete crash");
+                assert_eq!(interrupted.disposition.state, IlmRecoveryDispositionState::Applying);
+                assert!(
+                    interrupted.disposition.confirmed_absent.is_empty(),
+                    "the crash must occur before absence progress is persisted"
+                );
+            } else {
+                inject_recovery_disposition_crash_once(RecoveryDispositionCrashStage::AfterControlAbandon);
+                Box::pin(execute_recovery_disposition(
+                    store.clone(),
+                    &observation,
+                    &export.export_id,
+                    &export.content_sha256,
+                    &actor_sha256,
+                    confirmed_at_unix_nanos,
+                ))
+                .await
+                .expect_err("the injected crash must stop after control abandonment commits");
+                let interrupted =
+                    load_recovery_disposition(store.clone(), IlmRecoveryProtocol::TierDeleteJournal, &dry_run.disposition_id)
+                        .await
+                        .expect("the applying disposition must survive the post-control crash");
+                assert_eq!(interrupted.disposition.state, IlmRecoveryDispositionState::Applying);
+                assert_eq!(
+                    interrupted.disposition.confirmed_absent.len(),
+                    interrupted.disposition.identity.source_generation.copies.len()
+                );
+
+                let abandoned =
+                    load_recovery_control(store.clone(), IlmRecoveryProtocol::TierDeleteJournal, &observation.control_id)
+                        .await
+                        .expect("the abandoned control must survive the injected crash");
+                let exact_abandoned = abandoned.control.encode().expect("the exact abandoned control should encode");
+                let mut wrong_history = abandoned.control;
+                wrong_history.last_error_code = IlmRecoveryErrorCode::CleanupFailed;
+                let control_path = recovery_control_record_object_name(observation.protocol, &observation.control_id)
+                    .expect("control path should remain canonical");
+                com::save_config(
+                    store.clone(),
+                    &control_path,
+                    wrong_history
+                        .encode()
+                        .expect("the alternate valid control history should encode"),
+                )
+                .await
+                .expect("the alternate control history fixture should persist");
+                let wrong_history_err = Box::pin(execute_recovery_disposition(
+                    store.clone(),
+                    &observation,
+                    &export.export_id,
+                    &export.content_sha256,
+                    &actor_sha256,
+                    confirmed_at_unix_nanos + 1,
+                ))
+                .await
+                .expect_err("a different abandoned control history must not bridge to completion");
+                assert_eq!(wrong_history_err, Error::PreconditionFailed);
+                com::save_config(store.clone(), &control_path, exact_abandoned)
+                    .await
+                    .expect("the exact abandoned control fixture should be restored");
+            }
+
+            let replay_confirmed_at_unix_nanos = confirmed_at_unix_nanos + 2;
+            let executed = Box::pin(execute_recovery_disposition(
+                store.clone(),
+                &observation,
+                &export.export_id,
+                &export.content_sha256,
+                &actor_sha256,
+                replay_confirmed_at_unix_nanos,
+            ))
+            .await
+            .expect("a later request must resume and complete the interrupted disposition");
+            assert_eq!(executed.state, IlmRecoveryDispositionState::Completed);
+            assert_eq!(executed.outcome, IlmRecoveryDispositionExecutionOutcome::Completed);
+            assert_eq!(executed.confirmed_absent_copy_count, executed.source_copy_count);
+            assert_eq!(tier_delete_journal_count(store.clone()).await, fixtures.len() - index - 1);
+            assert!(matches!(
+                com::read_config(store.clone(), &observation.canonical_source_path).await,
+                Err(Error::ConfigNotFound)
+            ));
+            if index == 0 {
+                let untouched = journal_paths
+                    .iter()
+                    .find(|path| *path != &observation.canonical_source_path)
+                    .expect("the other legacy journal should remain");
+                com::read_config(store.clone(), untouched)
+                    .await
+                    .expect("disposition must not remove a different legacy journal");
+            }
+
+            let abandoned = load_recovery_control(store.clone(), IlmRecoveryProtocol::TierDeleteJournal, &observation.control_id)
+                .await
+                .expect("abandoned recovery control should remain inspectable");
+            assert_eq!(abandoned.control.classification, IlmRecoveryClassification::Abandoned);
+            assert_eq!(abandoned.control.revision, observation.control_revision + 1);
+            assert_eq!(abandoned.control.observed_source_generation, observation.source_generation);
+
+            let persisted =
+                load_recovery_disposition(store.clone(), IlmRecoveryProtocol::TierDeleteJournal, &executed.disposition_id)
+                    .await
+                    .expect("completed disposition should remain durable");
+            assert_eq!(persisted.disposition.state, IlmRecoveryDispositionState::Completed);
+
+            let replayed = Box::pin(execute_recovery_disposition(
+                store.clone(),
+                &observation,
+                &export.export_id,
+                &export.content_sha256,
+                &actor_sha256,
+                replay_confirmed_at_unix_nanos + 1,
+            ))
+            .await
+            .expect("same actor should replay the completed disposition");
+            assert_eq!(replayed.state, IlmRecoveryDispositionState::Completed);
+            assert_eq!(replayed.outcome, IlmRecoveryDispositionExecutionOutcome::Replayed);
+
+            let wrong_actor = Box::pin(execute_recovery_disposition(
+                store.clone(),
+                &observation,
+                &export.export_id,
+                &export.content_sha256,
+                &wrong_actor_sha256,
+                replay_confirmed_at_unix_nanos + 2,
+            ))
+            .await
+            .expect_err("a different actor must not replay a completed disposition");
+            assert_eq!(wrong_actor, Error::PreconditionFailed);
+
+            assert!(
+                !Box::pin(garbage_collect_completed_recovery_disposition(
+                    store.clone(),
+                    &persisted,
+                    persisted.disposition.retain_until_unix_nanos - 1,
+                ))
+                .await
+                .expect("completed disposition should remain before retention expires")
+            );
+            assert!(
+                Box::pin(garbage_collect_completed_recovery_disposition(
+                    store.clone(),
+                    &persisted,
+                    persisted.disposition.retain_until_unix_nanos,
+                ))
+                .await
+                .expect("expired completed disposition should be garbage collected")
+            );
+            assert!(matches!(
+                load_recovery_disposition(store.clone(), IlmRecoveryProtocol::TierDeleteJournal, &executed.disposition_id).await,
+                Err(Error::ConfigNotFound)
+            ));
+            assert_eq!(backend.remove_count().await, 0, "legacy disposition must not call the remote tier");
+            assert_eq!(backend.exact_remove_count(), 0, "legacy disposition must not issue exact remote DELETE");
+            assert!(
+                backend.op_log().await.is_empty(),
+                "legacy disposition must not invoke any backend operation"
+            );
+        }
+    }
+
     #[cfg(feature = "test-util")]
     #[tokio::test]
     #[serial_test::serial(storage_class_env)]