feat(ilm): execute legacy recovery dispositions (#7304)

* feat(ilm): execute legacy recovery dispositions

* feat(ilm): retry retained transition recovery (#7308)

* fix(admin): use gateway errors for recovery retries

---------

Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: Zhengchao An <anzhengchao@gmail.com>

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,
             };
         }
 
@@ -95,8 +93,9 @@ pub mod bucket {
         pub mod transition_transaction {
             pub use crate::bucket::lifecycle::transition_transaction::{
                 TransitionOperatorDeleteResult, TransitionOperatorError, TransitionOperatorProbe, TransitionOperatorStatus,
-                delete_transition_candidate_for_operator, finalize_missing_transition_transaction_for_operator,
-                inspect_transition_transaction_for_operator,
+                TransitionRecoveryRetryResult, TransitionRecoveryRetryStatus, delete_transition_candidate_for_operator,
+                finalize_missing_transition_transaction_for_operator, inspect_transition_recovery_retry_for_operator,
+                inspect_transition_transaction_for_operator, retry_transition_recovery_for_operator,
             };
             #[cfg(feature = "test-util")]
             pub use crate::bucket::lifecycle::transition_transaction::{

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,40 @@ 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 retry_for_operator(&mut self, expected_source_generation: &IlmRecoverySourceGeneration) -> Result<()> {
+        if self.owner.is_some()
+            || !matches!(
+                self.classification,
+                IlmRecoveryClassification::RetainedAmbiguous | IlmRecoveryClassification::OperatorRequired
+            )
+            || self.attempt_count == u64::MAX
+            || &self.observed_source_generation != expected_source_generation
+        {
+            return Err(IlmRecoveryControlError::InvalidSuccessor(
+                "operator retry requires the exact ownerless retained source generation",
+            ));
+        }
+        self.bump_revision()?;
+        self.classification = IlmRecoveryClassification::Retrying;
+        self.consecutive_failure_count = 0;
+        self.next_attempt_at_unix_nanos = None;
+        self.validate()
+    }
+
     pub fn validate_successor(&self, next: &Self) -> Result<()> {
         self.validate()?;
         next.validate()?;
@@ -508,12 +542,58 @@ 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)
+                if matches!(
+                    self.classification,
+                    IlmRecoveryClassification::RetainedAmbiguous | IlmRecoveryClassification::OperatorRequired
+                ) && next.classification == IlmRecoveryClassification::Retrying =>
+            {
+                self.validate_operator_retry_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_operator_retry_successor(&self, next: &Self) -> Result<()> {
+        if next.observed_source_generation != self.observed_source_generation
+            || next.attempt_count != self.attempt_count
+            || next.consecutive_failure_count != 0
+            || 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.is_some()
+            || next.last_error_code != self.last_error_code
+        {
+            return Err(IlmRecoveryControlError::InvalidSuccessor(
+                "operator retry 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 +907,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 +934,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 +1361,99 @@ 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 operator_retry_rearms_exact_retained_generation_without_resetting_history() {
+        for classification in [
+            IlmRecoveryClassification::RetainedAmbiguous,
+            IlmRecoveryClassification::OperatorRequired,
+        ] {
+            let mut retained = control();
+            retained
+                .claim("node-a", Uuid::new_v4(), 2_000_000_000, 1)
+                .expect("attempt should claim");
+            retained
+                .record_retryable_failure(2_000_000_001, IlmRecoveryErrorCode::BackendTimeout)
+                .expect("failure should persist");
+            retained.classification = classification;
+            retained.next_attempt_at_unix_nanos = None;
+            if classification == IlmRecoveryClassification::OperatorRequired {
+                retained.attempt_count = u64::from(MAX_RECOVERY_ATTEMPTS);
+                retained.consecutive_failure_count = MAX_RECOVERY_ATTEMPTS;
+            }
+            retained.validate().expect("retained control should remain valid");
+
+            let previous = retained.clone();
+            retained
+                .retry_for_operator(&previous.observed_source_generation)
+                .expect("exact retained generation should be retryable");
+            previous
+                .validate_successor(&retained)
+                .expect("operator retry should be a valid successor");
+            assert_eq!(retained.classification, IlmRecoveryClassification::Retrying);
+            assert_eq!(retained.revision, previous.revision + 1);
+            assert_eq!(retained.attempt_count, previous.attempt_count);
+            assert_eq!(retained.first_failure_at_unix_nanos, previous.first_failure_at_unix_nanos);
+            assert_eq!(retained.last_failure_at_unix_nanos, previous.last_failure_at_unix_nanos);
+            assert_eq!(retained.last_error_code, previous.last_error_code);
+            assert_eq!(retained.consecutive_failure_count, 0);
+            assert_eq!(retained.next_attempt_at_unix_nanos, None);
+            assert!(retained.should_attempt_at(2_000_000_002));
+
+            if classification == IlmRecoveryClassification::OperatorRequired {
+                retained
+                    .claim("node-b", Uuid::new_v4(), 2_000_000_002, 1)
+                    .expect("operator retry should authorize one new bounded attempt");
+                retained
+                    .record_retryable_failure(2_000_000_003, IlmRecoveryErrorCode::BackendTimeout)
+                    .expect("the bounded attempt failure should persist");
+                assert_eq!(retained.classification, IlmRecoveryClassification::OperatorRequired);
+                assert_eq!(retained.attempt_count, u64::from(MAX_RECOVERY_ATTEMPTS) + 1);
+                assert_eq!(retained.consecutive_failure_count, 1);
+            }
+
+            let mut wrong_generation = previous;
+            let mut changed_generation = wrong_generation.observed_source_generation.clone();
+            changed_generation.source_etag = "changed".to_string();
+            assert!(wrong_generation.retry_for_operator(&changed_generation).is_err());
+        }
+
+        let mut exhausted = control();
+        exhausted.classification = IlmRecoveryClassification::OperatorRequired;
+        exhausted.attempt_count = u64::MAX;
+        let generation = exhausted.observed_source_generation.clone();
+        assert!(exhausted.retry_for_operator(&generation).is_err());
+    }
+
     #[test]
     fn recovery_control_view_redacts_source_and_owner_details() {
         let mut control = control();

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

@@ -34,7 +34,7 @@ use crate::bucket::lifecycle::tier_sweeper::{
 };
 use crate::disk::RUSTFS_META_BUCKET;
 use crate::error::{Error, Result as EcstoreResult};
-use crate::object_api::ObjectOptions;
+use crate::object_api::{ObjectInfo, ObjectOptions};
 use crate::services::tier::{tier::TierConfigMgr, warm_backend::TransitionCandidateProbe};
 use crate::storage_api_contracts::{
     list::ListOperations as _,
@@ -954,6 +954,10 @@ pub enum TransitionOperatorError {
         expected: String,
         actual: TransitionOperatorProbe,
     },
+    #[error("transition recovery control is stale")]
+    StaleRecoveryControl,
+    #[error("transition recovery control is not eligible for operator retry")]
+    RetryNotAllowed,
     #[error("transition transaction store failed: {0}")]
     Store(#[source] Error),
     #[error("remote tier reconciliation failed: {0}")]
@@ -962,6 +966,179 @@ pub enum TransitionOperatorError {
 
 type TransitionOperatorResult<T> = std::result::Result<T, TransitionOperatorError>;
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct TransitionRecoveryRetryStatus {
+    pub control_id: String,
+    pub transaction_id: Uuid,
+    pub state: TransitionTransactionState,
+    pub classification: IlmRecoveryClassification,
+    pub control_revision: u64,
+    pub attempt_count: u64,
+    pub consecutive_failure_count: u32,
+    pub last_error_code: IlmRecoveryErrorCode,
+    pub source_generation_sha256: String,
+    pub copy_set_sha256: String,
+    pub retry_ready: bool,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub retry_not_ready_reason: Option<&'static str>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct TransitionRecoveryRetryResult {
+    pub control_id: String,
+    pub transaction_id: Uuid,
+    pub previous_revision: u64,
+    pub revision: u64,
+    pub classification: IlmRecoveryClassification,
+    pub attempt_count: u64,
+    pub source_generation_sha256: String,
+}
+
+struct TransitionRecoveryRetryContext {
+    observed: ObservedIlmRecoveryControl,
+    transaction: TransitionTransaction,
+    source_generation_sha256: String,
+}
+
+fn transition_recovery_retry_readiness(control: &IlmRecoveryControl) -> (bool, Option<&'static str>) {
+    if control.owner.is_some() {
+        return (false, Some("attempt_owned"));
+    }
+    match control.classification {
+        IlmRecoveryClassification::RetainedAmbiguous | IlmRecoveryClassification::OperatorRequired => (true, None),
+        IlmRecoveryClassification::Retrying => (false, Some("already_retrying")),
+        IlmRecoveryClassification::Corrupt => (false, Some("source_corrupt")),
+        IlmRecoveryClassification::Abandoned => (false, Some("source_abandoned")),
+        IlmRecoveryClassification::Terminal => (false, Some("source_terminal")),
+    }
+}
+
+async fn load_transition_recovery_retry_context(
+    api: Arc<ECStore>,
+    control_id: &str,
+) -> TransitionOperatorResult<TransitionRecoveryRetryContext> {
+    let observed = match load_recovery_control(api.clone(), IlmRecoveryProtocol::TransitionTransaction, control_id).await {
+        Ok(observed) => observed,
+        Err(Error::ConfigNotFound) => return Err(TransitionOperatorError::NotFound),
+        Err(err) => return Err(TransitionOperatorError::Store(err)),
+    };
+    let transaction_id = Uuid::parse_str(&observed.control.identity.stable_operation_identity)
+        .ok()
+        .filter(|transaction_id| !transaction_id.is_nil())
+        .ok_or(TransitionOperatorError::StaleRecoveryControl)?;
+    let canonical_path = transition_transaction_record_object_name(transaction_id)
+        .map_err(|err| TransitionOperatorError::Store(Error::other(err)))?;
+    if observed.control.identity.canonical_source_path != canonical_path
+        || observed.control.identity.record_class != "transition_transaction_v1"
+    {
+        return Err(TransitionOperatorError::StaleRecoveryControl);
+    }
+    let transaction = match load_transition_transaction_record(api.clone(), transaction_id).await {
+        Ok(transaction) => transaction,
+        Err(Error::ConfigNotFound) => return Err(TransitionOperatorError::NotFound),
+        Err(err) => return Err(TransitionOperatorError::Store(err)),
+    };
+    let source = observe_recovery_source(api, &canonical_path, TRANSITION_TRANSACTION_SCHEMA)
+        .await
+        .map_err(TransitionOperatorError::Store)?;
+    let exact_source = source.is_consistent()
+        && source.generation == observed.control.observed_source_generation
+        && source
+            .canonical_data
+            .as_deref()
+            .is_some_and(|data| TransitionTransaction::decode(transaction_id, data).is_ok_and(|decoded| decoded == transaction));
+    if !exact_source {
+        return Err(TransitionOperatorError::StaleRecoveryControl);
+    }
+    let generation = serde_json::to_vec(&observed.control.observed_source_generation)
+        .map_err(|err| TransitionOperatorError::Store(Error::other(err)))?;
+    Ok(TransitionRecoveryRetryContext {
+        observed,
+        transaction,
+        source_generation_sha256: hex_sha256(&generation, ToOwned::to_owned),
+    })
+}
+
+pub async fn inspect_transition_recovery_retry_for_operator(
+    api: Arc<ECStore>,
+    control_id: &str,
+) -> TransitionOperatorResult<TransitionRecoveryRetryStatus> {
+    let context = load_transition_recovery_retry_context(api, control_id).await?;
+    let (retry_ready, retry_not_ready_reason) = transition_recovery_retry_readiness(&context.observed.control);
+    Ok(TransitionRecoveryRetryStatus {
+        control_id: control_id.to_string(),
+        transaction_id: context.transaction.transaction_id,
+        state: context.transaction.state,
+        classification: context.observed.control.classification,
+        control_revision: context.observed.control.revision,
+        attempt_count: context.observed.control.attempt_count,
+        consecutive_failure_count: context.observed.control.consecutive_failure_count,
+        last_error_code: context.observed.control.last_error_code,
+        source_generation_sha256: context.source_generation_sha256,
+        copy_set_sha256: context.observed.control.observed_source_generation.copy_set_sha256.clone(),
+        retry_ready,
+        retry_not_ready_reason,
+    })
+}
+
+pub async fn retry_transition_recovery_for_operator(
+    api: Arc<ECStore>,
+    control_id: &str,
+    expected_control_revision: u64,
+    expected_source_generation_sha256: &str,
+) -> TransitionOperatorResult<TransitionRecoveryRetryResult> {
+    let control_object = recovery_control_record_object_name(IlmRecoveryProtocol::TransitionTransaction, control_id)
+        .map_err(|err| TransitionOperatorError::Store(Error::other(err)))?;
+    let retry_lock = api
+        .new_ns_lock(RUSTFS_META_BUCKET, &format!("{control_object}.recovery-lock"))
+        .await
+        .map_err(TransitionOperatorError::Store)?;
+    let retry_guard = retry_lock
+        .get_write_lock(crate::set_disk::get_lock_acquire_timeout())
+        .await
+        .map_err(|err| TransitionOperatorError::Store(Error::other(err)))?;
+    let context = load_transition_recovery_retry_context(api.clone(), control_id).await?;
+    let (retry_ready, _) = transition_recovery_retry_readiness(&context.observed.control);
+    if !retry_ready {
+        return Err(TransitionOperatorError::RetryNotAllowed);
+    }
+    if retry_guard.is_lock_lost()
+        || expected_control_revision == 0
+        || context.observed.control.revision != expected_control_revision
+        || context.source_generation_sha256 != expected_source_generation_sha256
+    {
+        return Err(TransitionOperatorError::StaleRecoveryControl);
+    }
+    let previous_revision = context.observed.control.revision;
+    let mut next = context.observed.control.clone();
+    next.retry_for_operator(&context.observed.control.observed_source_generation)
+        .map_err(|_| TransitionOperatorError::RetryNotAllowed)?;
+    if retry_guard.is_lock_lost() {
+        return Err(TransitionOperatorError::StaleRecoveryControl);
+    }
+    save_recovery_control_if_current(api.clone(), &context.observed, &next)
+        .await
+        .map_err(|err| match err {
+            Error::PreconditionFailed => TransitionOperatorError::StaleRecoveryControl,
+            err => TransitionOperatorError::Store(err),
+        })?;
+    let persisted = load_recovery_control(api, IlmRecoveryProtocol::TransitionTransaction, control_id)
+        .await
+        .map_err(TransitionOperatorError::Store)?;
+    if retry_guard.is_lock_lost() || persisted.control != next {
+        return Err(TransitionOperatorError::StaleRecoveryControl);
+    }
+    Ok(TransitionRecoveryRetryResult {
+        control_id: control_id.to_string(),
+        transaction_id: context.transaction.transaction_id,
+        previous_revision,
+        revision: persisted.control.revision,
+        classification: persisted.control.classification,
+        attempt_count: persisted.control.attempt_count,
+        source_generation_sha256: context.source_generation_sha256,
+    })
+}
+
 fn validate_operator_reconcile_transaction(
     transaction: &TransitionTransaction,
     now_unix_nanos: i128,
@@ -1706,12 +1883,27 @@ async fn local_commit_matches_transaction(api: Arc<ECStore>, transaction: &Trans
         .get_object_info(&transaction.source.bucket, &transaction.source.object, &opts)
         .await?;
     let transitioned = &object.transitioned_object;
-    Ok(transitioned.status == TRANSITION_COMPLETE
+    Ok(local_object_matches_transition_source(&object, &transaction.source)
+        && transitioned.status == TRANSITION_COMPLETE
         && transitioned.name == transaction.remote_object
         && transitioned.tier == transaction.tier_name
         && transitioned.version_id == transaction.remote_version.tier_delete_version_id().unwrap_or_default())
 }
 
+fn local_object_matches_transition_source(object: &ObjectInfo, source: &TransitionSourceIdentity) -> bool {
+    let observed_version_id = object.version_id.filter(|version_id| !version_id.is_nil());
+    let observed_mod_time = object
+        .mod_time
+        .and_then(|mod_time| i64::try_from(mod_time.unix_timestamp_nanos()).ok());
+    object.bucket == source.bucket
+        && object.name == source.object
+        && observed_version_id == source.version_id
+        && object.data_dir == Some(source.data_dir)
+        && observed_mod_time == Some(source.mod_time_unix_nanos)
+        && object.size == source.size
+        && object.etag.as_deref() == Some(source.etag.as_str())
+}
+
 fn transition_source_lookup_options(transaction: &TransitionTransaction) -> ObjectOptions {
     ObjectOptions {
         version_id: match transaction.source.version_mode {
@@ -2183,6 +2375,41 @@ mod tests {
         }
     }
 
+    #[test]
+    fn local_commit_proof_requires_the_complete_source_identity() {
+        let source = source_identity(TransitionSourceVersionMode::Versioned);
+        let exact = ObjectInfo {
+            bucket: source.bucket.clone(),
+            name: source.object.clone(),
+            version_id: source.version_id,
+            data_dir: Some(source.data_dir),
+            mod_time: Some(
+                time::OffsetDateTime::from_unix_timestamp_nanos(i128::from(source.mod_time_unix_nanos))
+                    .expect("source timestamp should be valid"),
+            ),
+            size: source.size,
+            etag: Some(source.etag.clone()),
+            ..Default::default()
+        };
+        assert!(local_object_matches_transition_source(&exact, &source));
+
+        let mut changed = exact.clone();
+        changed.version_id = Some(Uuid::new_v4());
+        assert!(!local_object_matches_transition_source(&changed, &source));
+        changed = exact.clone();
+        changed.data_dir = Some(Uuid::new_v4());
+        assert!(!local_object_matches_transition_source(&changed, &source));
+        changed = exact.clone();
+        changed.mod_time = changed.mod_time.map(|value| value + Duration::from_nanos(1));
+        assert!(!local_object_matches_transition_source(&changed, &source));
+        changed = exact.clone();
+        changed.size += 1;
+        assert!(!local_object_matches_transition_source(&changed, &source));
+        changed = exact;
+        changed.etag = Some("different-etag".to_string());
+        assert!(!local_object_matches_transition_source(&changed, &source));
+    }
+
     fn cleanup_proof(transaction: &TransitionTransaction, decision: TransitionCleanupDecision) -> TransitionCleanupProof {
         TransitionCleanupProof {
             transaction_id: transaction.transaction_id,

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,
@@ -856,9 +862,10 @@ mod tests {
                 TransitionOperatorProbe, TransitionRecoveryClaimBarrier, TransitionRecoveryTerminalBarrier,
                 TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction,
                 TransitionTransactionInit, TransitionTransactionState, delete_transition_candidate_for_operator,
-                finalize_missing_transition_transaction_for_operator, inspect_transition_transaction_for_operator,
-                load_transition_transaction_record, recover_transition_transaction_records,
-                recover_transition_transaction_records_at, save_transition_transaction_record,
+                finalize_missing_transition_transaction_for_operator, inspect_transition_recovery_retry_for_operator,
+                inspect_transition_transaction_for_operator, load_transition_transaction_record,
+                recover_transition_transaction_records, recover_transition_transaction_records_at,
+                retry_transition_recovery_for_operator, save_transition_transaction_record,
                 save_transition_transaction_record_if_current, transition_recovery_control_id,
                 transition_transaction_record_object_name,
             },
@@ -16936,6 +16943,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)]
@@ -20328,7 +20630,7 @@ mod tests {
             bucket: bucket.to_string(),
             object: object.to_string(),
             version_id: None,
-            data_dir: uuid::Uuid::new_v4(),
+            data_dir: original.data_dir.expect("source object should have data_dir"),
             mod_time_unix_nanos: original
                 .mod_time
                 .expect("source object should have mod_time")
@@ -20462,7 +20764,7 @@ mod tests {
                 bucket: bucket.to_string(),
                 object: object.to_string(),
                 version_id: None,
-                data_dir: uuid::Uuid::new_v4(),
+                data_dir: original.data_dir.expect("source object should have data_dir"),
                 mod_time_unix_nanos: original
                     .mod_time
                     .expect("source object should have mod_time")
@@ -20603,10 +20905,77 @@ mod tests {
             IlmRecoveryClassification::RetainedAmbiguous
         );
         let local_commit_control =
-            load_recovery_control(store, IlmRecoveryProtocol::TransitionTransaction, &local_commit_control_id)
+            load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &local_commit_control_id)
                 .await
                 .expect("local-commit control should persist");
         assert_eq!(local_commit_control.control.classification, IlmRecoveryClassification::OperatorRequired);
+
+        let upload_status = inspect_transition_recovery_retry_for_operator(store.clone(), &upload_started_control_id)
+            .await
+            .expect("retained upload should be inspectable for a bounded retry");
+        let local_status = inspect_transition_recovery_retry_for_operator(store.clone(), &local_commit_control_id)
+            .await
+            .expect("operator-required local commit should be inspectable for a bounded retry");
+        assert!(upload_status.retry_ready);
+        assert!(local_status.retry_ready);
+        assert!(matches!(
+            retry_transition_recovery_for_operator(
+                store.clone(),
+                &upload_started_control_id,
+                upload_status.control_revision + 1,
+                &upload_status.source_generation_sha256,
+            )
+            .await,
+            Err(TransitionOperatorError::StaleRecoveryControl)
+        ));
+
+        let put_count_before_retry = backend.put_count().await;
+        let get_count_before_retry = backend.get_count().await;
+        let remove_count_before_retry = backend.remove_count().await;
+        let upload_retry = retry_transition_recovery_for_operator(
+            store.clone(),
+            &upload_started_control_id,
+            upload_status.control_revision,
+            &upload_status.source_generation_sha256,
+        )
+        .await
+        .expect("exact retained upload generation should be rearmed");
+        let local_retry = retry_transition_recovery_for_operator(
+            store.clone(),
+            &local_commit_control_id,
+            local_status.control_revision,
+            &local_status.source_generation_sha256,
+        )
+        .await
+        .expect("exact operator-required local commit generation should be rearmed");
+        assert_eq!(upload_retry.classification, IlmRecoveryClassification::Retrying);
+        assert_eq!(local_retry.classification, IlmRecoveryClassification::Retrying);
+        assert_eq!(upload_retry.attempt_count, upload_status.attempt_count);
+        assert_eq!(local_retry.attempt_count, local_status.attempt_count);
+        assert_eq!(backend.put_count().await, put_count_before_retry);
+        assert_eq!(backend.get_count().await, get_count_before_retry);
+        assert_eq!(backend.remove_count().await, remove_count_before_retry);
+        assert_eq!(backend.exact_remove_count(), 0, "operator retry must not directly issue remote DELETE");
+
+        let retried = recover_transition_transaction_records(store.clone(), 100, None)
+            .await
+            .expect("rearmed records should be re-evaluated through normal recovery");
+        assert_eq!((retried.scanned, retried.recovered, retried.retained, retried.failed), (2, 0, 2, 0));
+        let upload_retained =
+            load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &upload_started_control_id)
+                .await
+                .expect("upload retry result should persist");
+        let local_retained = load_recovery_control(store, IlmRecoveryProtocol::TransitionTransaction, &local_commit_control_id)
+            .await
+            .expect("local commit retry result should persist");
+        assert_eq!(upload_retained.control.classification, IlmRecoveryClassification::RetainedAmbiguous);
+        assert_eq!(local_retained.control.classification, IlmRecoveryClassification::OperatorRequired);
+        assert_eq!(upload_retained.control.attempt_count, upload_status.attempt_count + 1);
+        assert_eq!(local_retained.control.attempt_count, local_status.attempt_count + 1);
+        assert_eq!(backend.put_count().await, put_count_before_retry);
+        assert_eq!(backend.get_count().await, get_count_before_retry);
+        assert_eq!(backend.remove_count().await, remove_count_before_retry);
+        assert_eq!(backend.exact_remove_count(), 0);
     }
 
     #[cfg(feature = "test-util")]