Merge branch 'main' into feat/ilm-legacy-recovery-execution

crates/e2e_test/src/replication_extension_test.rs

@@ -3837,77 +3837,6 @@ async fn test_bucket_replication_converges_delete_marker_and_version_purge() ->
     Ok(())
 }
 
-/// Regression for rustfs/backlog#2340 (not Wasabi specific): a directory
-/// marker (`prefix/` with a body) in a versioned bucket is stored as the null
-/// version, like MinIO (`putOpts`: "for directory objects skip creating new
-/// versions"), and must still replicate to completion instead of staying
-/// `PENDING`.
-#[tokio::test]
-async fn test_bucket_replication_replicates_directory_marker_in_versioned_bucket() -> TestResult {
-    init_logging();
-
-    let mut source_env = RustFSTestEnvironment::new().await?;
-    let mut source_env_vars = replication_fast_env();
-    source_env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
-    source_env.start_rustfs_server_with_env(vec![], &source_env_vars).await?;
-
-    let mut target_env = RustFSTestEnvironment::new().await?;
-    target_env.start_rustfs_server_without_cleanup(vec![]).await?;
-
-    let source_bucket = "replication-dir-marker-src";
-    let target_bucket = "replication-dir-marker-dst";
-    let source_client = source_env.create_s3_client();
-    let target_client = target_env.create_s3_client();
-
-    source_client.create_bucket().bucket(source_bucket).send().await?;
-    target_client.create_bucket().bucket(target_bucket).send().await?;
-    enable_bucket_versioning(&source_env, source_bucket).await?;
-    enable_bucket_versioning(&target_env, target_bucket).await?;
-    let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?;
-    put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
-
-    let marker_key = "dir/trailing/";
-    let body = b"directory marker body";
-    let put = source_client
-        .put_object()
-        .bucket(source_bucket)
-        .key(marker_key)
-        .body(ByteStream::from_static(body))
-        .send()
-        .await?;
-    assert!(
-        put.version_id()
-            .is_none_or(|id| id == "null" || id == uuid::Uuid::nil().to_string()),
-        "a directory marker is the null version even in a versioned bucket: {:?}",
-        put.version_id()
-    );
-
-    wait_for_source_replication_status(&source_client, source_bucket, marker_key, "COMPLETED", false).await?;
-
-    let replica = target_client
-        .get_object()
-        .bucket(target_bucket)
-        .key(marker_key)
-        .send()
-        .await?;
-    assert_eq!(replica.body.collect().await?.into_bytes().as_ref(), body);
-    let listed = target_client
-        .list_object_versions()
-        .bucket(target_bucket)
-        .prefix(marker_key)
-        .send()
-        .await?;
-    let marker_versions: Vec<_> = listed.versions().iter().filter(|v| v.key() == Some(marker_key)).collect();
-    assert_eq!(marker_versions.len(), 1, "the marker must land exactly once: {marker_versions:?}");
-    assert_eq!(
-        marker_versions[0].version_id(),
-        Some("null"),
-        "the replica keeps the null version identity"
-    );
-
-    Ok(())
-}
-
 #[tokio::test]
 async fn test_bucket_replication_disabled_delete_marker_does_not_propagate() -> TestResult {
     init_logging();

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

docs/architecture/ilm-tiering-persistence-contracts.md

@@ -28,7 +28,7 @@ These are approved-target invariants. A protocol's explicitly labeled current ex
 | Phase | Authoritative owner | May issue remote DELETE? | Ownership transfer evidence |
 |---|---|---:|---|
 | Remote PUT is in flight or its response is unknown | Transition transaction | Only cleanup of its own canonical candidate, subject to the transaction recovery predicate | Durable transaction identity plus a known remote-version state; the approved target also requires expiry and durable takeover of the creator fence |
-| Local transition commit is complete | Exact transitioned version in `xl.meta` | No | Current recovery finds the transaction's logical bucket/object/version and checks `TRANSITION_COMPLETE` plus the same remote object, tier, and remote version. It does not compare the recorded data directory, modification time, size, or ETag; the approved target adds that full source comparison |
+| Local transition commit is complete | Exact transitioned version in `xl.meta` | No | Recovery finds the transaction's logical bucket/object/version and requires the complete recorded source identity (version ID, data directory, modification time, size, and ETag), `TRANSITION_COMPLETE`, and the same remote object, tier, and remote version before removing only the transaction record |
 | An ordinary delete removes that transitioned version | Hidden `xl.meta` free-version | Yes | Metadata quorum atomically removes the visible version and preserves its exact tier tuple in the free-version |
 | A recursive prefix/delete-all operation cannot preserve per-object markers | v6 journal bound to an immutable single dispatch manifest or a chunk-parent-bound child manifest | Yes, but only after child/manifest completion and all-pool absence proof | `DispatchAuthorized`, exact local destructive mutation, every journal `Committed`, then child/manifest `Completed`; a chunk parent advances only after that child completion |
 | Tier configuration mutation, manual job, or decommission receipt | Intent/admission/copy proof only | No | These records gate configuration, scheduling, or migration; they never become remote-object cleanup owners |
@@ -134,7 +134,7 @@ The creator owns the canonical remote candidate until local metadata commits the
 | `UploadOutcomeUnknown`; probe returns `VersionedPresent` whose identifier is a nil UUID | Transaction recovery retains ownership evidence | Retain | A nil identifier is invalid exact-version evidence and never becomes unversioned or remote-delete authority |
 | `UploadOutcomeUnknown`; probe ambiguous, unsupported, or errors | Transaction recovery retains ownership evidence | Retain | No destructive action; operator reconcile may inspect after expiry |
 | `Uploaded` | Originating transition attempt until expiry; after expiry, the worker that wins `Uploaded -> CleanupPending` by exact ETag CAS | Retain while active; after expiry, persist `CleanupPending`, then recheck and delete the unreferenced candidate or record | Current CAS fences the predecessor, but approved v2 also requires a durable recovery lease, full all-pool source/free-version proof, and before/after fence checks |
-| `LocalCommitStarted`; logical source lookup returns `TRANSITION_COMPLETE` with the same remote object, tier, and remote version | Transition committer until ownership transfers to `xl.meta` | Delete transaction record | Current recovery treats this tuple as ownership transfer. The approved target additionally compares recorded source version ID, data directory, modification time, size, and ETag before conditional terminal cleanup |
+| `LocalCommitStarted`; logical source lookup returns the complete recorded source identity and `TRANSITION_COMPLETE` with the same remote object, tier, and remote version | Transition committer until ownership transfers to `xl.meta` | Delete transaction record | Recovery treats only the full source and remote tuple as ownership transfer; a mismatch remains operator-required and cannot authorize remote deletion |
 | `LocalCommitStarted`; logical source is missing, its transition tuple differs, or the read is uncertain | Transaction record/recovery | Retain | No remote delete without a separate durable cleanup proof |
 | `CleanupPending`; logical source lookup returns the same current transition predicate | `xl.meta` is remote reachability owner; recovery owns only record cleanup | Delete transaction record | `xl.meta` is owner; do not delete remote. The approved target adds the full recorded source comparison |
 | `CleanupPending`; logical source is absent or its transition tuple differs | Transaction recovery | Delete exact candidate, then record | Cleanup proof, known version state, exact backend lease, durable owner fence, and before/after identity checks |

docs/architecture/s3-compatibility-matrix.md

@@ -62,7 +62,6 @@ Object keys are stored as file-system paths under each drive (`{drive}/{bucket}/
 | Behavior | RustFS | AWS S3 | Why |
 |---|---|---|---|
 | Object key with a `.` or `..` path segment, or an empty segment (`//`), such as `a//b/./c/../d` | `400 InvalidArgument` (`check_object_args` in `crates/ecstore/src/bucket/utils.rs`, mirroring MinIO `IsValidObjectPrefix`) | Accepted as an opaque key | A `..` segment would resolve to a parent directory and `.`/`//` segments would alias other keys on disk; encoding them would change the MinIO-compatible on-disk format. |
-| Directory marker (key ending in `/`, with or without a body) in a versioned bucket | Stored as the null version: `PutObject`/`HeadObject` report version id `00000000-0000-0000-0000-000000000000`, `ListObjectVersions` reports `null`, and a later PUT of the same key overwrites in place (`put_opts` in `rustfs/src/storage/options.rs`, mirroring MinIO `putOpts`: "for directory objects skip creating new versions") | A real version id per PUT, with a version history | The marker only exists to make an empty prefix listable; keeping a history for it would leave hidden versions behind every prefix delete. Replication still copies the marker as its null version (`test_bucket_replication_replicates_directory_marker_in_versioned_bucket` in `crates/e2e_test/src/replication_extension_test.rs`). |
 
 ## Update Rule
 

docs/operations/tier-ilm-debugging.md

@@ -173,7 +173,7 @@ Historical transition transactions in `upload_outcome_unknown` state can use an
 
 ## Inspect and disposition retained recovery records
 
-This section describes an **approved target that is not implemented yet**. Current servers do not expose the routes below and continue to quarantine tier-delete journal v1/v2 records. Do not remove internal metadata objects by hand: that loses ETag, all-pool, decommission, export, and audit guarantees.
+Current servers expose the routes below for retained recovery controls. Do not remove internal metadata objects by hand: that loses ETag, all-pool, decommission, export, and audit guarantees.
 
 The approved read-only inventory is bounded and paginated:
 
@@ -223,6 +223,39 @@ Malformed/unsupported records and journal v3-v6 cannot use abandon. Known-versio
 
 Automatic retry state survives restart. Retryable transport/quorum failures use a 60-second exponential base capped at one hour and a deterministic 80-to-100-percent multiplier, so jitter never increases the capped delay. After 32 consecutive failures or seven days from the first persisted failure, automatic work stops at `operator_required`. Unsupported or ambiguous evidence goes directly to `retained_ambiguous`/`operator_required`; age alone never deletes it. Resolved controls, immutable exports, and completed disposition receipts have minimum 30-day, 90-day, and 365-day retention respectively, and are collected only after exact source absence, decommission, successor, and audit checks.
 
+### Retry a retained transition transaction
+
+For a `transition_transaction` control, inspect returns an additional `transition_retry` object when the exact transaction source and recovery-control generation are still consistent. It contains `retry_ready`, `control_revision`, `source_generation_sha256`, the current classification and counters, and a bounded refusal reason. A missing `transition_retry` with `transition_retry_not_ready_reason=source_or_control_not_ready` means the server could not reconstruct exact live evidence; do not retry from an older response.
+
+First perform a dry-run with the exact revision and source-generation digest returned by the latest inspect:
+
+```json
+POST /rustfs/admin/v3/ilm/recovery/records/<control-id>
+{
+  "action": "retry_transition_recovery",
+  "mode": "dry_run",
+  "expected_control_revision": 7,
+  "expected_source_generation_sha256": "<sha256>"
+}
+```
+
+After repairing the reported storage, tier, or capability problem, repeat inspect and dry-run, then execute with the newly observed values:
+
+```json
+POST /rustfs/admin/v3/ilm/recovery/records/<control-id>
+{
+  "action": "retry_transition_recovery",
+  "mode": "execute",
+  "expected_control_revision": 7,
+  "expected_source_generation_sha256": "<sha256>",
+  "confirm": true
+}
+```
+
+Execution performs one ETag-CAS update of the exact ownerless `retained_ambiguous` or `operator_required` control to `retrying`. It preserves the lifetime attempt count and failure history, clears only the consecutive-failure backoff, and does not mutate the transaction source or issue a tier PUT, GET, probe, or DELETE. The normal recovery worker then acquires a fresh bounded owner lease and repeats every source and remote proof before any side effect.
+
+A historical v1 `UploadStarted` record can return to `retained_ambiguous` because its bytes do not prove whether PUT reached the provider. `LocalCommitStarted` becomes terminal only when the local object still matches the recorded version ID, data directory, modification time, size, ETag, and exact transitioned remote tuple; otherwise it returns to `operator_required`. Retrying is therefore a bounded re-evaluation after an underlying repair, not an override of missing evidence.
+
 The full schema, lease, mixed-version, retry, privacy, and metric requirements are in [../architecture/ilm-tiering-persistence-contracts.md](../architecture/ilm-tiering-persistence-contracts.md#bounded-recovery-control-and-operator-disposition).
 
 ## Reconcile legacy transition-version metadata

rustfs/src/admin/handlers/ilm_transition.rs

@@ -18,18 +18,21 @@ use crate::admin::runtime_sources::{current_action_credentials, object_store_fro
 use crate::admin::storage_api::bucket::is_reserved_or_invalid_bucket;
 use crate::admin::storage_api::error::StorageError;
 use crate::admin::storage_api::lifecycle::{
-    IlmRecoveryClassification, IlmRecoveryControlView, IlmRecoveryExportObservation, IlmRecoveryProtocol,
-    ManualTransitionCancelCheck, ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionProgressSink,
-    ManualTransitionQueueSnapshot, ManualTransitionRunOptions, ManualTransitionRunReport, ManualTransitionScopeAdmission,
-    ManualTransitionScopeAdmissionClaim, TransitionOperatorDeleteResult, TransitionOperatorError,
+    IlmRecoveryClassification, IlmRecoveryControlView, IlmRecoveryDispositionExecutionOutcome, IlmRecoveryDispositionReasonCode,
+    IlmRecoveryDispositionState, IlmRecoveryExportObservation, IlmRecoveryProtocol, ManualTransitionCancelCheck,
+    ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionProgressSink, ManualTransitionQueueSnapshot,
+    ManualTransitionRunOptions, ManualTransitionRunReport, ManualTransitionScopeAdmission, ManualTransitionScopeAdmissionClaim,
+    TransitionOperatorDeleteResult, TransitionOperatorError, TransitionRecoveryRetryResult, TransitionRecoveryRetryStatus,
     claim_manual_transition_scope_admission, create_recovery_export, delete_manual_transition_scope_admission_if_current,
-    delete_transition_candidate_for_operator, enqueue_transition_for_existing_objects_scoped,
-    finalize_missing_transition_transaction_for_operator, inspect_recovery_control, inspect_recovery_export_observation,
+    delete_transition_candidate_for_operator, dry_run_recovery_disposition, enqueue_transition_for_existing_objects_scoped,
+    execute_recovery_disposition, finalize_missing_transition_transaction_for_operator, inspect_recovery_control,
+    inspect_recovery_export_observation, inspect_transition_recovery_retry_for_operator,
     inspect_transition_transaction_for_operator, list_recovery_controls, load_manual_transition_job_record,
     load_manual_transition_scope_admission, load_recovery_export, manual_transition_job_lease_expired,
     manual_transition_queue_snapshot, manual_transition_scope_admission_lease_expired,
     persist_manual_transition_job_progress_if_owned, renew_manual_transition_job_lease_if_owned,
-    request_manual_transition_job_cancel, save_manual_transition_job_record, update_manual_transition_job_record,
+    request_manual_transition_job_cancel, retry_transition_recovery_for_operator, save_manual_transition_job_record,
+    update_manual_transition_job_record,
 };
 use crate::admin::storage_api::runtime::ECStore;
 use crate::admin::storage_api::s3::{S3ErrorCode as AdminS3ErrorCode, error as admin_s3_error};
@@ -257,7 +260,7 @@ pub fn register_ilm_transition_route(r: &mut S3Router<AdminOperation>) -> std::i
     r.insert(
         Method::POST,
         format!("{ADMIN_PREFIX}/v3/ilm/recovery/records/{{control_id}}").as_str(),
-        AdminOperation(&IlmRecoveryExportCreateHandler {}),
+        AdminOperation(&IlmRecoveryRecordMutationHandler {}),
     )?;
     r.insert(
         Method::GET,
@@ -537,6 +540,16 @@ fn map_recovery_export_error(err: StorageError) -> S3Error {
     }
 }
 
+fn map_recovery_disposition_error(err: StorageError) -> S3Error {
+    if err == StorageError::ConfigNotFound {
+        admin_s3_error(AdminS3ErrorCode::NoSuchKey, "ILM recovery export or disposition not found")
+    } else if err == StorageError::SlowDown {
+        admin_s3_error(AdminS3ErrorCode::SlowDown, "ILM recovery disposition admission capacity is exhausted")
+    } else {
+        admin_s3_error(AdminS3ErrorCode::OperationAborted, "ILM recovery disposition request cannot proceed")
+    }
+}
+
 fn recovery_export_download_headers(export_id: &str, encoded_len: usize) -> S3Result<HeaderMap> {
     let mut headers = HeaderMap::new();
     headers.insert(header::CONTENT_TYPE, HeaderValue::from_static("application/json"));
@@ -600,12 +613,14 @@ struct IlmRecoveryControlInspectResponse {
     observation_receipt: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     observation_receipt_expires_at_unix_nanos: Option<i64>,
-}
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-enum IlmRecoveryDispositionReasonCode {
-    LegacyRemoteCleanupAbandoned,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    disposition_dry_run_receipt: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    disposition_dry_run_receipt_expires_at_unix_nanos: Option<i64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    transition_retry: Option<TransitionRecoveryRetryStatus>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    transition_retry_not_ready_reason: Option<&'static str>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -625,6 +640,13 @@ enum IlmRecoveryRecordMutationRequest {
         #[serde(default)]
         acknowledge_remote_cleanup_abandoned: Option<bool>,
     },
+    RetryTransitionRecovery {
+        mode: IlmRecoveryReceiptMode,
+        expected_control_revision: u64,
+        expected_source_generation_sha256: String,
+        #[serde(default)]
+        confirm: Option<bool>,
+    },
 }
 
 fn parse_recovery_record_mutation_request(body: &[u8]) -> S3Result<IlmRecoveryRecordMutationRequest> {
@@ -647,10 +669,22 @@ fn parse_recovery_record_mutation_request(body: &[u8]) -> S3Result<IlmRecoveryRe
             "ILM recovery dry-run must not include terminal confirmation fields",
         ));
     }
+    if matches!(
+        &request,
+        IlmRecoveryRecordMutationRequest::RetryTransitionRecovery {
+            mode: IlmRecoveryReceiptMode::DryRun,
+            ..
+        }
+    ) && value.as_object().is_some_and(|object| object.contains_key("confirm"))
+    {
+        return Err(admin_s3_error(
+            AdminS3ErrorCode::InvalidArgument,
+            "ILM recovery retry dry-run must not include confirm",
+        ));
+    }
     Ok(request)
 }
 
-#[allow(dead_code)]
 enum ValidatedIlmRecoveryRecordMutation<'a> {
     Export {
         observation_receipt: &'a str,
@@ -667,6 +701,14 @@ enum ValidatedIlmRecoveryRecordMutation<'a> {
         export_sha256: &'a str,
         reason_code: IlmRecoveryDispositionReasonCode,
     },
+    RetryTransitionDryRun {
+        expected_control_revision: u64,
+        expected_source_generation_sha256: &'a str,
+    },
+    RetryTransitionExecute {
+        expected_control_revision: u64,
+        expected_source_generation_sha256: &'a str,
+    },
 }
 
 fn validate_recovery_record_mutation_request(
@@ -722,9 +764,62 @@ fn validate_recovery_record_mutation_request(
                 )),
             }
         }
+        IlmRecoveryRecordMutationRequest::RetryTransitionRecovery {
+            mode,
+            expected_control_revision,
+            expected_source_generation_sha256,
+            confirm,
+        } => {
+            if *expected_control_revision == 0 {
+                return Err(admin_s3_error(
+                    AdminS3ErrorCode::InvalidArgument,
+                    "transition recovery retry requires a nonzero expected control revision",
+                ));
+            }
+            validate_recovery_sha256(
+                expected_source_generation_sha256,
+                "invalid transition recovery source generation checksum",
+            )?;
+            match mode {
+                IlmRecoveryReceiptMode::DryRun if confirm.is_none() => {
+                    Ok(ValidatedIlmRecoveryRecordMutation::RetryTransitionDryRun {
+                        expected_control_revision: *expected_control_revision,
+                        expected_source_generation_sha256,
+                    })
+                }
+                IlmRecoveryReceiptMode::DryRun => Err(admin_s3_error(
+                    AdminS3ErrorCode::InvalidArgument,
+                    "ILM recovery retry dry-run must not include confirm",
+                )),
+                IlmRecoveryReceiptMode::Execute if *confirm == Some(true) => {
+                    Ok(ValidatedIlmRecoveryRecordMutation::RetryTransitionExecute {
+                        expected_control_revision: *expected_control_revision,
+                        expected_source_generation_sha256,
+                    })
+                }
+                IlmRecoveryReceiptMode::Execute => Err(admin_s3_error(
+                    AdminS3ErrorCode::InvalidRequest,
+                    "transition recovery retry requires confirm=true",
+                )),
+            }
+        }
     }
 }
 
+#[derive(Debug, Serialize)]
+struct IlmTransitionRecoveryRetryDryRunResponse {
+    action: &'static str,
+    mode: IlmRecoveryReceiptMode,
+    status: TransitionRecoveryRetryStatus,
+}
+
+#[derive(Debug, Serialize)]
+struct IlmTransitionRecoveryRetryExecuteResponse {
+    action: &'static str,
+    mode: IlmRecoveryReceiptMode,
+    result: TransitionRecoveryRetryResult,
+}
+
 #[derive(Debug, Serialize)]
 struct IlmRecoveryExportCreateResponse {
     export_id: String,
@@ -733,33 +828,30 @@ struct IlmRecoveryExportCreateResponse {
     outcome: &'static str,
 }
 
-// These response envelopes pin the future disposition wire contract before
-// its storage state machine is connected to this handler.
-#[allow(dead_code)]
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 enum IlmRecoveryDispositionDryRunStatus {
     Ready,
 }
 
-#[allow(dead_code)]
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
-enum IlmRecoveryDispositionState {
+enum IlmRecoveryDispositionResponseState {
     Applying,
     Completed,
 }
 
-#[allow(dead_code)]
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
-#[serde(rename_all = "snake_case")]
-enum IlmRecoveryDispositionOutcome {
-    AcceptedForRecovery,
-    Completed,
-    Replayed,
+fn recovery_disposition_response_state(state: IlmRecoveryDispositionState) -> S3Result<IlmRecoveryDispositionResponseState> {
+    match state {
+        IlmRecoveryDispositionState::Applying => Ok(IlmRecoveryDispositionResponseState::Applying),
+        IlmRecoveryDispositionState::Completed => Ok(IlmRecoveryDispositionResponseState::Completed),
+        IlmRecoveryDispositionState::Prepared => Err(admin_s3_error(
+            AdminS3ErrorCode::OperationAborted,
+            "ILM recovery disposition is accepted but not yet applying",
+        )),
+    }
 }
 
-#[allow(dead_code)]
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 struct IlmRecoveryDispositionDryRunResponse {
@@ -776,15 +868,14 @@ struct IlmRecoveryDispositionDryRunResponse {
     observation_receipt_expires_at_unix_nanos: i64,
 }
 
-#[allow(dead_code)]
 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 struct IlmRecoveryDispositionExecuteResponse {
     action: IlmRecoveryReceiptAction,
     mode: IlmRecoveryReceiptMode,
     disposition_id: String,
-    state: IlmRecoveryDispositionState,
-    outcome: IlmRecoveryDispositionOutcome,
+    state: IlmRecoveryDispositionResponseState,
+    outcome: IlmRecoveryDispositionExecutionOutcome,
     confirmed_absent_copy_count: usize,
     source_copy_count: usize,
 }
@@ -925,6 +1016,12 @@ fn map_transition_operator_error(err: TransitionOperatorError) -> S3Error {
         TransitionOperatorError::CandidateVersionMismatch { .. } => {
             s3_error!(OperationAborted, "remote candidate version does not match requested exact version")
         }
+        TransitionOperatorError::StaleRecoveryControl => {
+            s3_error!(OperationAborted, "transition recovery control or source generation changed")
+        }
+        TransitionOperatorError::RetryNotAllowed => {
+            s3_error!(OperationAborted, "transition recovery control is not eligible for operator retry")
+        }
         TransitionOperatorError::Store(_) | TransitionOperatorError::Remote(_) => {
             s3_error!(InternalError, "transition reconciliation failed")
         }
@@ -1528,21 +1625,58 @@ impl Operation for IlmRecoveryControlInspectHandler {
         let control = inspect_recovery_control(store.clone(), &control_id)
             .await
             .map_err(map_recovery_control_error)?;
+        let (transition_retry, transition_retry_not_ready_reason) =
+            if control.protocol == IlmRecoveryProtocol::TransitionTransaction {
+                match inspect_transition_recovery_retry_for_operator(store.clone(), &control_id).await {
+                    Ok(status) => (Some(status), None),
+                    Err(_) => (None, Some("source_or_control_not_ready")),
+                }
+            } else {
+                (None, None)
+            };
         let now = OffsetDateTime::now_utc();
-        let (export_ready, export_not_ready_reason, observation_receipt, expires_at) =
-            match inspect_recovery_export_observation(store, &control_id).await {
-                Ok(observation) => match issue_recovery_observation_receipt(
-                    observation,
-                    actor_sha256,
+        let (
+            export_ready,
+            export_not_ready_reason,
+            observation_receipt,
+            observation_receipt_expires_at_unix_nanos,
+            disposition_dry_run_receipt,
+            disposition_dry_run_receipt_expires_at_unix_nanos,
+        ) = match inspect_recovery_export_observation(store, &control_id).await {
+            Ok(observation) => {
+                let export_receipt = issue_recovery_observation_receipt(
+                    observation.clone(),
+                    actor_sha256.clone(),
                     IlmRecoveryReceiptAction::Export,
                     IlmRecoveryReceiptMode::Execute,
                     now,
-                ) {
+                );
+                let disposition_receipt = issue_recovery_observation_receipt(
+                    observation,
+                    actor_sha256,
+                    IlmRecoveryReceiptAction::AbandonRemoteCleanup,
+                    IlmRecoveryReceiptMode::DryRun,
+                    now,
+                );
+                let (export_ready, export_not_ready_reason, observation_receipt, export_expires_at) = match export_receipt {
                     Ok((token, expires_at)) => (true, None, Some(token), Some(expires_at)),
                     Err(_) => (false, Some("receipt_key_unavailable"), None, None),
-                },
-                Err(_) => (false, Some("fleet_or_source_not_ready"), None, None),
-            };
+                };
+                let (disposition_receipt, disposition_expires_at) = match disposition_receipt {
+                    Ok((token, expires_at)) => (Some(token), Some(expires_at)),
+                    Err(_) => (None, None),
+                };
+                (
+                    export_ready,
+                    export_not_ready_reason,
+                    observation_receipt,
+                    export_expires_at,
+                    disposition_receipt,
+                    disposition_expires_at,
+                )
+            }
+            Err(_) => (false, Some("fleet_or_source_not_ready"), None, None, None, None),
+        };
         json_response(
             StatusCode::OK,
             &IlmRecoveryControlInspectResponse {
@@ -1550,16 +1684,20 @@ impl Operation for IlmRecoveryControlInspectHandler {
                 export_ready,
                 export_not_ready_reason,
                 observation_receipt,
-                observation_receipt_expires_at_unix_nanos: expires_at,
+                observation_receipt_expires_at_unix_nanos,
+                disposition_dry_run_receipt,
+                disposition_dry_run_receipt_expires_at_unix_nanos,
+                transition_retry,
+                transition_retry_not_ready_reason,
             },
         )
     }
 }
 
-pub struct IlmRecoveryExportCreateHandler {}
+pub struct IlmRecoveryRecordMutationHandler {}
 
 #[async_trait::async_trait]
-impl Operation for IlmRecoveryExportCreateHandler {
+impl Operation for IlmRecoveryRecordMutationHandler {
     async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         let actor_sha256 = authorize_recovery_admin_request(&req, AdminAction::SetTierAction).await?;
         let control_id = recovery_control_id_from_params(&params)?;
@@ -1567,35 +1705,161 @@ impl Operation for IlmRecoveryExportCreateHandler {
             return Err(admin_s3_error(AdminS3ErrorCode::InternalError, "object store is not initialized"));
         };
         let body = req.input.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE).await.map_err(|_| {
-            admin_s3_error(AdminS3ErrorCode::InvalidRequest, "ILM recovery export body is too large or unreadable")
+            admin_s3_error(AdminS3ErrorCode::InvalidRequest, "ILM recovery request body is too large or unreadable")
         })?;
         let request = parse_recovery_record_mutation_request(&body)?;
-        let ValidatedIlmRecoveryRecordMutation::Export { observation_receipt } =
-            validate_recovery_record_mutation_request(&request)?
-        else {
-            return Err(admin_s3_error(AdminS3ErrorCode::InvalidArgument, "unsupported ILM recovery action"));
-        };
-        let receipt = decode_recovery_receipt(observation_receipt, &recovery_receipt_credentials()?)?;
-        let now = i64::try_from(OffsetDateTime::now_utc().unix_timestamp_nanos())
+        let mutation = validate_recovery_record_mutation_request(&request)?;
+        let now = OffsetDateTime::now_utc();
+        let now_unix_nanos = i64::try_from(now.unix_timestamp_nanos())
             .map_err(|_| admin_s3_error(AdminS3ErrorCode::InternalError, "ILM recovery receipt timestamp is invalid"))?;
-        let observation = validate_recovery_observation_receipt(
-            receipt,
-            &actor_sha256,
-            &control_id,
-            IlmRecoveryReceiptAction::Export,
-            IlmRecoveryReceiptMode::Execute,
-            now,
-        )?;
-        let created = create_recovery_export(store, &observation, &actor_sha256)
-            .await
-            .map_err(map_recovery_export_error)?;
-        let response = IlmRecoveryExportCreateResponse {
-            download_url: format!("{ADMIN_PREFIX}/v3/ilm/recovery/exports/{}", created.export_id),
-            outcome: if created.replayed { "replayed" } else { "created" },
-            export_id: created.export_id,
-            export_sha256: created.content_sha256,
-        };
-        json_response(StatusCode::OK, &response)
+        match mutation {
+            ValidatedIlmRecoveryRecordMutation::Export { observation_receipt } => {
+                let receipt_credentials = recovery_receipt_credentials()?;
+                let receipt = decode_recovery_receipt(observation_receipt, &receipt_credentials)?;
+                let observation = validate_recovery_observation_receipt(
+                    receipt,
+                    &actor_sha256,
+                    &control_id,
+                    IlmRecoveryReceiptAction::Export,
+                    IlmRecoveryReceiptMode::Execute,
+                    now_unix_nanos,
+                )?;
+                let created = create_recovery_export(store, &observation, &actor_sha256)
+                    .await
+                    .map_err(map_recovery_export_error)?;
+                let response = IlmRecoveryExportCreateResponse {
+                    download_url: format!("{ADMIN_PREFIX}/v3/ilm/recovery/exports/{}", created.export_id),
+                    outcome: if created.replayed { "replayed" } else { "created" },
+                    export_id: created.export_id,
+                    export_sha256: created.content_sha256,
+                };
+                json_response(StatusCode::OK, &response)
+            }
+            ValidatedIlmRecoveryRecordMutation::AbandonDryRun {
+                observation_receipt,
+                export_id,
+                export_sha256,
+                reason_code: IlmRecoveryDispositionReasonCode::LegacyRemoteCleanupAbandoned,
+            } => {
+                let receipt_credentials = recovery_receipt_credentials()?;
+                let receipt = decode_recovery_receipt(observation_receipt, &receipt_credentials)?;
+                let observation = validate_recovery_observation_receipt(
+                    receipt,
+                    &actor_sha256,
+                    &control_id,
+                    IlmRecoveryReceiptAction::AbandonRemoteCleanup,
+                    IlmRecoveryReceiptMode::DryRun,
+                    now_unix_nanos,
+                )?;
+                let dry_run =
+                    dry_run_recovery_disposition(store, &observation, export_id, export_sha256, &actor_sha256, now_unix_nanos)
+                        .await
+                        .map_err(map_recovery_disposition_error)?;
+                let execute_receipt_now = OffsetDateTime::now_utc();
+                let (execute_receipt, execute_receipt_expires_at_unix_nanos) = issue_recovery_observation_receipt(
+                    observation,
+                    actor_sha256,
+                    IlmRecoveryReceiptAction::AbandonRemoteCleanup,
+                    IlmRecoveryReceiptMode::Execute,
+                    execute_receipt_now,
+                )?;
+                json_response(
+                    StatusCode::OK,
+                    &IlmRecoveryDispositionDryRunResponse {
+                        action: IlmRecoveryReceiptAction::AbandonRemoteCleanup,
+                        mode: IlmRecoveryReceiptMode::DryRun,
+                        status: IlmRecoveryDispositionDryRunStatus::Ready,
+                        disposition_id: dry_run.disposition_id,
+                        export_id: dry_run.export_id,
+                        export_sha256: dry_run.export_content_sha256,
+                        source_generation_sha256: dry_run.source_generation_sha256,
+                        copy_set_sha256: dry_run.copy_set_sha256,
+                        source_copy_count: dry_run.source_copy_count,
+                        observation_receipt: execute_receipt,
+                        observation_receipt_expires_at_unix_nanos: execute_receipt_expires_at_unix_nanos,
+                    },
+                )
+            }
+            ValidatedIlmRecoveryRecordMutation::AbandonExecute {
+                observation_receipt,
+                export_id,
+                export_sha256,
+                reason_code: IlmRecoveryDispositionReasonCode::LegacyRemoteCleanupAbandoned,
+            } => {
+                let receipt_credentials = recovery_receipt_credentials()?;
+                let receipt = decode_recovery_receipt(observation_receipt, &receipt_credentials)?;
+                let observation = validate_recovery_observation_receipt(
+                    receipt,
+                    &actor_sha256,
+                    &control_id,
+                    IlmRecoveryReceiptAction::AbandonRemoteCleanup,
+                    IlmRecoveryReceiptMode::Execute,
+                    now_unix_nanos,
+                )?;
+                let execution =
+                    execute_recovery_disposition(store, &observation, export_id, export_sha256, &actor_sha256, now_unix_nanos)
+                        .await
+                        .map_err(map_recovery_disposition_error)?;
+                let state = recovery_disposition_response_state(execution.state)?;
+                json_response(
+                    StatusCode::OK,
+                    &IlmRecoveryDispositionExecuteResponse {
+                        action: IlmRecoveryReceiptAction::AbandonRemoteCleanup,
+                        mode: IlmRecoveryReceiptMode::Execute,
+                        disposition_id: execution.disposition_id,
+                        state,
+                        outcome: execution.outcome,
+                        confirmed_absent_copy_count: execution.confirmed_absent_copy_count,
+                        source_copy_count: execution.source_copy_count,
+                    },
+                )
+            }
+            ValidatedIlmRecoveryRecordMutation::RetryTransitionDryRun {
+                expected_control_revision,
+                expected_source_generation_sha256,
+            } => {
+                let status = inspect_transition_recovery_retry_for_operator(store, &control_id)
+                    .await
+                    .map_err(map_transition_operator_error)?;
+                if !status.retry_ready {
+                    return Err(map_transition_operator_error(TransitionOperatorError::RetryNotAllowed));
+                }
+                if status.control_revision != expected_control_revision
+                    || status.source_generation_sha256 != expected_source_generation_sha256
+                {
+                    return Err(map_transition_operator_error(TransitionOperatorError::StaleRecoveryControl));
+                }
+                json_response(
+                    StatusCode::OK,
+                    &IlmTransitionRecoveryRetryDryRunResponse {
+                        action: "retry_transition_recovery",
+                        mode: IlmRecoveryReceiptMode::DryRun,
+                        status,
+                    },
+                )
+            }
+            ValidatedIlmRecoveryRecordMutation::RetryTransitionExecute {
+                expected_control_revision,
+                expected_source_generation_sha256,
+            } => {
+                let result = retry_transition_recovery_for_operator(
+                    store,
+                    &control_id,
+                    expected_control_revision,
+                    expected_source_generation_sha256,
+                )
+                .await
+                .map_err(map_transition_operator_error)?;
+                json_response(
+                    StatusCode::OK,
+                    &IlmTransitionRecoveryRetryExecuteResponse {
+                        action: "retry_transition_recovery",
+                        mode: IlmRecoveryReceiptMode::Execute,
+                        result,
+                    },
+                )
+            }
+        }
     }
 }
 
@@ -1809,17 +2073,74 @@ mod tests {
         assert!(!token.contains("actor-a"));
         assert!(!token.contains("ilm/tier-delete-journal"));
         assert_eq!(decode_recovery_receipt(&token, &credentials).unwrap(), payload);
-        assert!(
-            validate_recovery_observation_receipt(
-                payload.clone(),
-                &payload.actor_sha256,
-                &payload.observation.control_id,
-                IlmRecoveryReceiptAction::Export,
+        let receipt_classes = [
+            (payload.clone(), IlmRecoveryReceiptAction::Export, IlmRecoveryReceiptMode::Execute),
+            (
+                IlmRecoveryObservationReceipt {
+                    action: IlmRecoveryReceiptAction::AbandonRemoteCleanup,
+                    mode: IlmRecoveryReceiptMode::DryRun,
+                    ..payload.clone()
+                },
+                IlmRecoveryReceiptAction::AbandonRemoteCleanup,
+                IlmRecoveryReceiptMode::DryRun,
+            ),
+            (
+                IlmRecoveryObservationReceipt {
+                    action: IlmRecoveryReceiptAction::AbandonRemoteCleanup,
+                    mode: IlmRecoveryReceiptMode::Execute,
+                    ..payload.clone()
+                },
+                IlmRecoveryReceiptAction::AbandonRemoteCleanup,
                 IlmRecoveryReceiptMode::Execute,
-                payload.issued_at_unix_nanos,
+            ),
+        ];
+        let expected_classes = [
+            (IlmRecoveryReceiptAction::Export, IlmRecoveryReceiptMode::Execute),
+            (IlmRecoveryReceiptAction::AbandonRemoteCleanup, IlmRecoveryReceiptMode::DryRun),
+            (IlmRecoveryReceiptAction::AbandonRemoteCleanup, IlmRecoveryReceiptMode::Execute),
+        ];
+        for (receipt, actual_action, actual_mode) in &receipt_classes {
+            for (expected_action, expected_mode) in expected_classes {
+                let result = validate_recovery_observation_receipt(
+                    receipt.clone(),
+                    &receipt.actor_sha256,
+                    &receipt.observation.control_id,
+                    expected_action,
+                    expected_mode,
+                    receipt.issued_at_unix_nanos,
+                );
+                if (*actual_action, *actual_mode) == (expected_action, expected_mode) {
+                    assert!(result.is_ok(), "the matching receipt class must validate");
+                } else {
+                    assert_eq!(
+                        result.expect_err("receipts must not cross action or mode boundaries").code(),
+                        &S3ErrorCode::AccessDenied
+                    );
+                }
+            }
+
+            let actor_mismatch = validate_recovery_observation_receipt(
+                receipt.clone(),
+                &hex_sha256(b"actor-b", ToOwned::to_owned),
+                &receipt.observation.control_id,
+                *actual_action,
+                *actual_mode,
+                receipt.issued_at_unix_nanos,
             )
-            .is_ok()
-        );
+            .expect_err("receipts must remain bound to the authenticated actor");
+            assert_eq!(actor_mismatch.code(), &S3ErrorCode::AccessDenied);
+
+            let expired = validate_recovery_observation_receipt(
+                receipt.clone(),
+                &receipt.actor_sha256,
+                &receipt.observation.control_id,
+                *actual_action,
+                *actual_mode,
+                receipt.expires_at_unix_nanos,
+            )
+            .expect_err("expired receipts must fail closed");
+            assert_eq!(expired.code(), &S3ErrorCode::AccessDenied);
+        }
         let assert_denied = |receipt: IlmRecoveryObservationReceipt, actor: &str, control: &str, now: i64| {
             let err = validate_recovery_observation_receipt(
                 receipt,
@@ -1832,19 +2153,7 @@ mod tests {
             .expect_err("invalid observation receipt must be denied");
             assert_eq!(err.code(), &S3ErrorCode::AccessDenied);
         };
-        assert_denied(
-            payload.clone(),
-            &hex_sha256(b"actor-b", ToOwned::to_owned),
-            &payload.observation.control_id,
-            payload.issued_at_unix_nanos,
-        );
         assert_denied(payload.clone(), &payload.actor_sha256, &"cd".repeat(32), payload.issued_at_unix_nanos);
-        assert_denied(
-            payload.clone(),
-            &payload.actor_sha256,
-            &payload.observation.control_id,
-            payload.expires_at_unix_nanos,
-        );
 
         let mut invalid = payload.clone();
         invalid.schema = "rustfs-ilm-recovery-observation-receipt-v2".to_string();
@@ -1986,6 +2295,48 @@ mod tests {
             }) if observed_export_id == export_id && observed_export_sha256 == export_sha256
         ));
 
+        let source_generation_sha256 = "ef".repeat(32);
+        let retry_dry_run_json = format!(
+            r#"{{"action":"retry_transition_recovery","mode":"dry_run","expected_control_revision":7,"expected_source_generation_sha256":"{source_generation_sha256}"}}"#
+        );
+        let retry_dry_run = parse_recovery_record_mutation_request(retry_dry_run_json.as_bytes()).unwrap();
+        assert!(matches!(
+            validate_recovery_record_mutation_request(&retry_dry_run),
+            Ok(ValidatedIlmRecoveryRecordMutation::RetryTransitionDryRun {
+                expected_control_revision: 7,
+                expected_source_generation_sha256: observed,
+            }) if observed == source_generation_sha256
+        ));
+        let retry_execute_json = retry_dry_run_json.replace(r#""mode":"dry_run""#, r#""mode":"execute","confirm":true"#);
+        let retry_execute = parse_recovery_record_mutation_request(retry_execute_json.as_bytes()).unwrap();
+        assert!(matches!(
+            validate_recovery_record_mutation_request(&retry_execute),
+            Ok(ValidatedIlmRecoveryRecordMutation::RetryTransitionExecute {
+                expected_control_revision: 7,
+                expected_source_generation_sha256: observed,
+            }) if observed == source_generation_sha256
+        ));
+        assert!(
+            parse_recovery_record_mutation_request(
+                retry_dry_run_json
+                    .replace(r#""mode":"dry_run""#, r#""mode":"dry_run","confirm":false"#)
+                    .as_bytes()
+            )
+            .is_err()
+        );
+        let retry_without_confirmation = retry_execute_json.replace(r#""confirm":true,"#, "");
+        if let Ok(request) = parse_recovery_record_mutation_request(retry_without_confirmation.as_bytes()) {
+            assert!(validate_recovery_record_mutation_request(&request).is_err());
+        }
+        for invalid in [
+            retry_execute_json.replace(r#""expected_control_revision":7"#, r#""expected_control_revision":0"#),
+            retry_execute_json.replace(source_generation_sha256.as_str(), "EF".repeat(32).as_str()),
+            retry_execute_json.replace(source_generation_sha256.as_str(), "too-short"),
+        ] {
+            let request = parse_recovery_record_mutation_request(invalid.as_bytes()).unwrap();
+            assert!(validate_recovery_record_mutation_request(&request).is_err());
+        }
+
         let dry_run_with_confirmation = dry_run_json.replace(r#""mode":"dry_run""#, r#""mode":"dry_run","confirm":false"#);
         assert!(parse_recovery_record_mutation_request(dry_run_with_confirmation.as_bytes()).is_err());
         assert!(parse_recovery_record_mutation_request(dry_run_json.replace('}', r#","confirm":null}"#).as_bytes()).is_err());
@@ -1994,10 +2345,15 @@ mod tests {
         for invalid in [
             execute_json.replace(r#""confirm":true,"#, ""),
             execute_json.replace(r#""confirm":true"#, r#""confirm":false"#),
+            execute_json.replace(r#""confirm":true"#, r#""confirm":null"#),
             execute_json.replace(
                 r#""acknowledge_remote_cleanup_abandoned":true"#,
                 r#""acknowledge_remote_cleanup_abandoned":false"#,
             ),
+            execute_json.replace(
+                r#""acknowledge_remote_cleanup_abandoned":true"#,
+                r#""acknowledge_remote_cleanup_abandoned":null"#,
+            ),
             execute_json.replace(export_id.as_str(), uppercase_export_id.as_str()),
             execute_json.replace(export_sha256.as_str(), "too-short"),
             execute_json.replace("opaque-execute", ""),
@@ -2010,8 +2366,19 @@ mod tests {
             }
         }
 
+        assert!(parse_recovery_record_mutation_request(execute_json.replace(r#""mode":"execute","#, "").as_bytes()).is_err());
+        assert!(
+            parse_recovery_record_mutation_request(
+                dry_run_json
+                    .replace("legacy_remote_cleanup_abandoned", "operator_override")
+                    .as_bytes()
+            )
+            .is_err()
+        );
+
         for invalid in [
             br#"{"action":"export","observation_receipt":"opaque","extra":true}"#.as_slice(),
+            br#"{"action":"abandon_remote_cleanup","mode":null}"#.as_slice(),
             br#"{"action":"abandon_remote_cleanup","mode":"preview"}"#.as_slice(),
             br#"{"action":"unknown","observation_receipt":"opaque"}"#.as_slice(),
         ] {
@@ -2051,11 +2418,24 @@ mod tests {
             observation_receipt_expires_at_unix_nanos: 900_000_000_001,
         };
         let dry_run_json = serde_json::to_value(&dry_run).unwrap();
-        assert_eq!(dry_run_json["action"], "abandon_remote_cleanup");
-        assert_eq!(dry_run_json["mode"], "dry_run");
-        assert_eq!(dry_run_json["status"], "ready");
         assert_eq!(
-            serde_json::from_value::<IlmRecoveryDispositionDryRunResponse>(dry_run_json).unwrap(),
+            dry_run_json,
+            serde_json::json!({
+                "action": "abandon_remote_cleanup",
+                "mode": "dry_run",
+                "status": "ready",
+                "disposition_id": "ab".repeat(32),
+                "export_id": "cd".repeat(32),
+                "export_sha256": "ef".repeat(32),
+                "source_generation_sha256": "12".repeat(32),
+                "copy_set_sha256": "34".repeat(32),
+                "source_copy_count": 2,
+                "observation_receipt": "opaque-execute",
+                "observation_receipt_expires_at_unix_nanos": 900_000_000_001_i64,
+            })
+        );
+        assert_eq!(
+            serde_json::from_value::<IlmRecoveryDispositionDryRunResponse>(dry_run_json.clone()).unwrap(),
             dry_run
         );
 
@@ -2063,22 +2443,68 @@ mod tests {
             action: IlmRecoveryReceiptAction::AbandonRemoteCleanup,
             mode: IlmRecoveryReceiptMode::Execute,
             disposition_id: "ab".repeat(32),
-            state: IlmRecoveryDispositionState::Applying,
-            outcome: IlmRecoveryDispositionOutcome::AcceptedForRecovery,
+            state: IlmRecoveryDispositionResponseState::Applying,
+            outcome: IlmRecoveryDispositionExecutionOutcome::AcceptedForRecovery,
             confirmed_absent_copy_count: 1,
             source_copy_count: 2,
         };
         let execute_json = serde_json::to_value(&execute).unwrap();
-        assert_eq!(execute_json["state"], "applying");
-        assert_eq!(execute_json["outcome"], "accepted_for_recovery");
         assert_eq!(
-            serde_json::from_value::<IlmRecoveryDispositionExecuteResponse>(execute_json).unwrap(),
+            execute_json,
+            serde_json::json!({
+                "action": "abandon_remote_cleanup",
+                "mode": "execute",
+                "disposition_id": "ab".repeat(32),
+                "state": "applying",
+                "outcome": "accepted_for_recovery",
+                "confirmed_absent_copy_count": 1,
+                "source_copy_count": 2,
+            })
+        );
+        assert_eq!(
+            serde_json::from_value::<IlmRecoveryDispositionExecuteResponse>(execute_json.clone()).unwrap(),
             execute
         );
 
-        let mut unknown = serde_json::to_value(&dry_run).unwrap();
-        unknown["unexpected"] = serde_json::json!(true);
-        assert!(serde_json::from_value::<IlmRecoveryDispositionDryRunResponse>(unknown).is_err());
+        let mut unknown_dry_run = dry_run_json;
+        unknown_dry_run["unexpected"] = serde_json::json!(true);
+        assert!(serde_json::from_value::<IlmRecoveryDispositionDryRunResponse>(unknown_dry_run).is_err());
+
+        let mut unknown_execute = execute_json;
+        unknown_execute["unexpected"] = serde_json::json!(true);
+        assert!(serde_json::from_value::<IlmRecoveryDispositionExecuteResponse>(unknown_execute).is_err());
+    }
+
+    #[test]
+    fn prepared_recovery_disposition_is_operation_aborted_and_not_a_wire_state() {
+        let err = recovery_disposition_response_state(IlmRecoveryDispositionState::Prepared)
+            .expect_err("Prepared must not escape through the closed execute response");
+        assert_eq!(err.code(), &S3ErrorCode::OperationAborted);
+        assert_eq!(err.message(), Some("ILM recovery disposition is accepted but not yet applying"));
+        assert!(serde_json::from_str::<IlmRecoveryDispositionResponseState>(r#""prepared""#).is_err());
+        assert_eq!(
+            serde_json::to_string(&IlmRecoveryDispositionResponseState::Applying).unwrap(),
+            r#""applying""#
+        );
+        assert_eq!(
+            serde_json::to_string(&IlmRecoveryDispositionResponseState::Completed).unwrap(),
+            r#""completed""#
+        );
+    }
+
+    #[test]
+    fn recovery_disposition_error_mapping_is_stable_and_fail_closed() {
+        let not_found = map_recovery_disposition_error(StorageError::ConfigNotFound);
+        assert_eq!(not_found.code(), &S3ErrorCode::NoSuchKey);
+        assert_eq!(not_found.message(), Some("ILM recovery export or disposition not found"));
+
+        let overloaded = map_recovery_disposition_error(StorageError::SlowDown);
+        assert_eq!(overloaded.code(), &S3ErrorCode::SlowDown);
+        assert_eq!(overloaded.message(), Some("ILM recovery disposition admission capacity is exhausted"));
+
+        let stale = map_recovery_disposition_error(StorageError::PreconditionFailed);
+        assert_eq!(stale.code(), &S3ErrorCode::OperationAborted);
+        assert_eq!(stale.message(), Some("ILM recovery disposition request cannot proceed"));
     }
 
     #[test]
@@ -2094,7 +2520,7 @@ mod tests {
         );
     }
 
-    fn manual_transition_job_request(method: Method, path: &'static str) -> S3Request<Body> {
+    fn credential_less_admin_request(method: Method, path: &'static str) -> S3Request<Body> {
         S3Request {
             input: Body::empty(),
             method,
@@ -2470,7 +2896,7 @@ mod tests {
     #[tokio::test]
     async fn transition_admin_gate_keeps_its_missing_credentials_response() {
         let err = authorize_transition_admin_request(
-            &manual_transition_job_request(Method::GET, "/rustfs/admin/v3/ilm/transition/jobs/job-123"),
+            &credential_less_admin_request(Method::GET, "/rustfs/admin/v3/ilm/transition/jobs/job-123"),
             AdminAction::ListTierAction,
         )
         .await
@@ -2483,8 +2909,8 @@ mod tests {
     #[tokio::test]
     async fn recovery_admin_gate_keeps_its_missing_credentials_response() {
         let err = authorize_recovery_admin_request(
-            &manual_transition_job_request(Method::GET, "/rustfs/admin/v3/ilm/recovery/controls/control-123"),
-            AdminAction::ListTierAction,
+            &credential_less_admin_request(Method::POST, "/rustfs/admin/v3/ilm/recovery/records/control-id"),
+            AdminAction::SetTierAction,
         )
         .await
         .expect_err("a recovery admin request without credentials must fail");
@@ -2616,7 +3042,7 @@ mod tests {
     async fn manual_transition_job_handlers_reject_missing_credentials_before_status_contract() {
         let status_err = ManualTransitionJobStatusHandler {}
             .call(
-                manual_transition_job_request(Method::GET, "/rustfs/admin/v3/ilm/transition/jobs/job-123"),
+                credential_less_admin_request(Method::GET, "/rustfs/admin/v3/ilm/transition/jobs/job-123"),
                 Params::new(),
             )
             .await
@@ -2626,7 +3052,7 @@ mod tests {
 
         let cancel_err = ManualTransitionJobCancelHandler {}
             .call(
-                manual_transition_job_request(Method::DELETE, "/rustfs/admin/v3/ilm/transition/jobs/job-123"),
+                credential_less_admin_request(Method::DELETE, "/rustfs/admin/v3/ilm/transition/jobs/job-123"),
                 Params::new(),
             )
             .await

rustfs/src/admin/storage_api.rs

@@ -235,12 +235,18 @@ pub(crate) mod lifecycle {
     pub(crate) use super::ecstore_bucket::lifecycle::recovery_control::{
         IlmRecoveryClassification, IlmRecoveryControlView, IlmRecoveryProtocol, inspect_recovery_control, list_recovery_controls,
     };
+    pub(crate) use super::ecstore_bucket::lifecycle::recovery_disposition::{
+        IlmRecoveryDispositionExecutionOutcome, IlmRecoveryDispositionReasonCode, IlmRecoveryDispositionState,
+        dry_run_recovery_disposition, execute_recovery_disposition,
+    };
     pub(crate) use super::ecstore_bucket::lifecycle::recovery_export::{
         IlmRecoveryExportObservation, create_recovery_export, inspect_recovery_export_observation, load_recovery_export,
     };
     pub(crate) use super::ecstore_bucket::lifecycle::transition_transaction::{
-        TransitionOperatorDeleteResult, TransitionOperatorError, delete_transition_candidate_for_operator,
-        finalize_missing_transition_transaction_for_operator, inspect_transition_transaction_for_operator,
+        TransitionOperatorDeleteResult, TransitionOperatorError, 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,
     };
 
     pub(crate) async fn enqueue_transition_for_existing_objects_scoped(