Merge branch 'main' into fix/bounded-transition-recovery

crates/ecstore/src/api/mod.rs

@@ -69,6 +69,13 @@ pub mod bucket {
             };
         }
 
+        pub mod recovery_control {
+            pub use crate::bucket::lifecycle::recovery_control::{
+                IlmRecoveryClassification, IlmRecoveryControlPage, IlmRecoveryControlView, IlmRecoveryProtocol,
+                inspect_recovery_control, list_recovery_controls,
+            };
+        }
+
         pub mod transition_transaction {
             pub use crate::bucket::lifecycle::transition_transaction::{
                 TransitionOperatorDeleteResult, TransitionOperatorError, TransitionOperatorProbe, TransitionOperatorStatus,

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

@@ -150,18 +150,6 @@ static XXHASH_SEED: u64 = 0;
 static TIER_FREE_VERSION_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
 static MANUAL_TRANSITION_JOB_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
 
-#[cfg(test)]
-#[derive(Default)]
-struct FreeVersionPostRemoteDeleteTestBarrier {
-    arrived: Notify,
-    release: Notify,
-}
-
-#[cfg(test)]
-tokio::task_local! {
-    static FREE_VERSION_POST_REMOTE_DELETE_TEST_BARRIER: Arc<FreeVersionPostRemoteDeleteTestBarrier>;
-}
-
 pub const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging";
 #[allow(
     dead_code,
@@ -922,11 +910,6 @@ async fn cleanup_free_version_exact(api: Arc<ECStore>, oi: &ObjectInfo, cancel:
             })??;
         }
     }
-    #[cfg(test)]
-    if let Ok(barrier) = FREE_VERSION_POST_REMOTE_DELETE_TEST_BARRIER.try_with(Arc::clone) {
-        barrier.arrived.notify_one();
-        barrier.release.notified().await;
-    }
     if !free_version_cleanup_fences_current(&topology_generation, &api, &bucket_guard, &object_guards, &lease, cancel, deadline) {
         // Remote DELETE is idempotent, but a changed fence makes the local
         // outcome ambiguous. Keep every marker for a fully fenced retry.
@@ -5848,7 +5831,7 @@ mod tests {
     #[cfg(feature = "test-util")]
     use crate::services::tier::test_util::register_mock_tier;
     #[cfg(feature = "test-util")]
-    use crate::services::tier::tier::{TIER_DRIVER_TEST_FACTORY, TierConfigMgr, TierDriverTestFactory};
+    use crate::services::tier::tier::TierConfigMgr;
     #[cfg(feature = "test-util")]
     use crate::services::tier::warm_backend::{TransitionCandidateProbe, WarmBackend as _};
     use crate::set_disk::{MultipartCommitBarrier, MultipartCommitPause};
@@ -7847,119 +7830,6 @@ mod tests {
         }
     }
 
-    #[cfg(feature = "test-util")]
-    #[tokio::test]
-    #[serial]
-    async fn tier_remove_waits_for_inflight_free_version_local_commit() {
-        let (disk_paths, ecstore) = setup_test_env().await;
-        let bucket = format!("tier-remove-free-version-{}", Uuid::new_v4());
-        let object = "free-version";
-        create_test_bucket(&ecstore, &bucket).await;
-        let (backend, identity_hex) = register_recovery_mock_tier(&ecstore).await;
-        let tier_manager = ecstore.tier_config_mgr();
-        {
-            let manager = tier_manager.read().await;
-            manager
-                .save_tiering_config(Arc::clone(&ecstore))
-                .await
-                .expect("mock tier configuration should persist before removal");
-        }
-        seed_recoverable_free_version(&disk_paths, &bucket, object, None, Some(identity_hex)).await;
-        let page = list_tier_free_versions(Arc::clone(&ecstore), 1, None, None, CancellationToken::new())
-            .await
-            .expect("seeded free version should be listed");
-        let oi = page
-            .items
-            .into_iter()
-            .next()
-            .expect("seeded free version should be recoverable");
-
-        backend
-            .set_put_remote_version(Some(oi.transitioned_object.version_id.clone()))
-            .await;
-        let seed_lease = TierConfigMgr::acquire_operation_lease(&tier_manager, "WARM")
-            .await
-            .expect("mock tier lease should be available");
-        seed_lease
-            .put(&oi.transitioned_object.name, ReaderImpl::Body(Bytes::from_static(b"body")), 4)
-            .await
-            .expect("remote free-version tuple should be seeded");
-        drop(seed_lease);
-
-        let barrier = Arc::new(super::FreeVersionPostRemoteDeleteTestBarrier::default());
-        let cleanup_barrier = Arc::clone(&barrier);
-        let cleanup_store = Arc::clone(&ecstore);
-        let cleanup_oi = oi.clone();
-        let cleanup = tokio::spawn(async move {
-            super::FREE_VERSION_POST_REMOTE_DELETE_TEST_BARRIER
-                .scope(cleanup_barrier, async move {
-                    super::cleanup_free_version_exact(cleanup_store, &cleanup_oi, &CancellationToken::new()).await
-                })
-                .await
-        });
-        tokio::time::timeout(StdDuration::from_secs(30), barrier.arrived.notified())
-            .await
-            .expect("free-version cleanup should pause after the remote delete");
-        assert!(!backend.contains(&oi.transitioned_object.name).await);
-
-        let remove_manager = Arc::clone(&tier_manager);
-        let remove_store = Arc::clone(&ecstore);
-        let remove_backend = backend.clone();
-        let remove_driver_factory: TierDriverTestFactory = Arc::new(move |_| Ok(Box::new(remove_backend.clone())));
-        let mut remove = tokio::spawn(async move {
-            TIER_DRIVER_TEST_FACTORY
-                .scope(
-                    remove_driver_factory,
-                    TierConfigMgr::remove_and_save(&remove_manager, remove_store, "WARM", true),
-                )
-                .await
-        });
-        let prepared = tokio::time::timeout(StdDuration::from_secs(30), async {
-            loop {
-                match TierConfigMgr::acquire_operation_lease(&tier_manager, "WARM").await {
-                    Ok(lease) => drop(lease),
-                    Err(err) if TierConfigMgr::operation_lease_blocked_by_mutation(&err) => break,
-                    Err(err) => panic!("tier remove should only block new operations while cleanup is paused: {err}"),
-                }
-                tokio::task::yield_now().await;
-            }
-        });
-        tokio::select! {
-            prepared = prepared => {
-                prepared.expect("tier remove should install its prepared admission fence");
-            }
-            result = &mut remove => {
-                panic!("tier remove finished before installing its prepared admission fence: {result:?}");
-            }
-        }
-        assert!(!remove.is_finished(), "tier remove must wait for the leased local cleanup commit");
-
-        barrier.release.notify_one();
-        tokio::time::timeout(StdDuration::from_secs(30), cleanup)
-            .await
-            .expect("free-version cleanup should finish after release")
-            .expect("free-version cleanup task should join")
-            .expect("free-version cleanup should keep its generation current");
-        tokio::time::timeout(StdDuration::from_secs(30), remove)
-            .await
-            .expect("tier remove should finish after local cleanup")
-            .expect("tier remove task should join")
-            .expect("tier remove should pass its fresh authoritative proof");
-
-        assert!(!tier_manager.read().await.is_tier_valid("WARM"));
-        for disk_path in &disk_paths {
-            assert!(
-                !fs::try_exists(disk_path.join(&bucket).join(object))
-                    .await
-                    .expect("post-removal free-version path check should succeed")
-            );
-        }
-        ecstore
-            .delete_bucket(&bucket, &DeleteBucketOptions::default())
-            .await
-            .expect("empty free-version test bucket should be removed");
-    }
-
     #[cfg(feature = "test-util")]
     #[tokio::test]
     #[serial]

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

@@ -22,7 +22,7 @@ use super::{
     bucket_lifecycle_ops::{
         ManualTransitionQueueSnapshot, ManualTransitionRunReport, decode_manual_transition_continuation_token,
     },
-    manual_transition_job, tier_delete_journal, transition_transaction,
+    manual_transition_job, recovery_control, tier_delete_journal, transition_transaction,
 };
 use crate::error::{Error, Result};
 use crate::services::tier::tier_probe_intent;
@@ -41,6 +41,7 @@ pub(crate) enum DurableIlmRecordKind {
     ManualTransitionScope,
     ManualTransitionTask,
     ManualTransitionWorkerResult,
+    RecoveryControl,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -105,8 +106,14 @@ pub(crate) const MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE: DurableIlmNamespace
     max_record_size: manual_transition_job::MAX_MANUAL_TRANSITION_WORKER_RESULT_RECORD_SIZE,
     kind: DurableIlmRecordKind::ManualTransitionWorkerResult,
 };
+pub(crate) const RECOVERY_CONTROL_NAMESPACE: DurableIlmNamespace = DurableIlmNamespace {
+    name: "recovery-control",
+    prefix: recovery_control::ILM_RECOVERY_CONTROL_PREFIX,
+    max_record_size: recovery_control::MAX_ILM_RECOVERY_CONTROL_SIZE,
+    kind: DurableIlmRecordKind::RecoveryControl,
+};
 
-pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 9] = [
+pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 10] = [
     TIER_DELETE_JOURNAL_NAMESPACE,
     TIER_DELETE_JOURNAL_V6_NAMESPACE,
     TIER_DELETE_DISPATCH_MANIFEST_NAMESPACE,
@@ -116,6 +123,7 @@ pub(crate) const DURABLE_ILM_NAMESPACES: [DurableIlmNamespace; 9] = [
     MANUAL_TRANSITION_SCOPE_NAMESPACE,
     MANUAL_TRANSITION_TASK_NAMESPACE,
     MANUAL_TRANSITION_WORKER_RESULT_NAMESPACE,
+    RECOVERY_CONTROL_NAMESPACE,
 ];
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -241,6 +249,18 @@ pub(crate) enum DurableIlmRecordCheckpoint {
     ManualTransitionWorkerResult {
         content_sha256: String,
     },
+    RecoveryControl {
+        content_sha256: String,
+        identity_sha256: String,
+        source_generation_sha256: String,
+        first_seen_at_unix_nanos: i64,
+        revision: u64,
+        classification: recovery_control::IlmRecoveryClassification,
+        attempt_count: u64,
+        consecutive_failure_count: u32,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        owner_fence_sha256: Option<String>,
+    },
 }
 
 impl DurableIlmRecordCheckpoint {
@@ -254,7 +274,8 @@ impl DurableIlmRecordCheckpoint {
             | Self::ManualTransitionJob { content_sha256, .. }
             | Self::ManualTransitionScope { content_sha256, .. }
             | Self::ManualTransitionTask { content_sha256 }
-            | Self::ManualTransitionWorkerResult { content_sha256 } => content_sha256,
+            | Self::ManualTransitionWorkerResult { content_sha256 }
+            | Self::RecoveryControl { content_sha256, .. } => content_sha256,
         }
     }
 
@@ -528,6 +549,51 @@ impl DurableIlmRecordCheckpoint {
                     ..
                 },
             ) => previous_identity == next_identity && next_updated_at > previous_updated_at,
+            (
+                Self::RecoveryControl {
+                    identity_sha256: previous_identity,
+                    source_generation_sha256: previous_generation,
+                    first_seen_at_unix_nanos: previous_first_seen,
+                    revision: previous_revision,
+                    classification: previous_classification,
+                    attempt_count: previous_attempts,
+                    consecutive_failure_count: previous_failures,
+                    owner_fence_sha256: previous_owner,
+                    ..
+                },
+                Self::RecoveryControl {
+                    identity_sha256: next_identity,
+                    source_generation_sha256: next_generation,
+                    first_seen_at_unix_nanos: next_first_seen,
+                    revision: next_revision,
+                    classification: next_classification,
+                    attempt_count: next_attempts,
+                    consecutive_failure_count: next_failures,
+                    owner_fence_sha256: next_owner,
+                    ..
+                },
+            ) => {
+                let adjacent = previous_identity == next_identity
+                    && previous_first_seen == next_first_seen
+                    && previous_revision.checked_add(1) == Some(*next_revision);
+                let claim = next_owner.is_some()
+                    && *previous_classification == recovery_control::IlmRecoveryClassification::Retrying
+                    && *next_classification == recovery_control::IlmRecoveryClassification::Retrying
+                    && previous_attempts.checked_add(1) == Some(*next_attempts)
+                    && previous_failures == next_failures;
+                let source_refresh = previous_owner.is_some()
+                    && previous_owner == next_owner
+                    && *previous_classification == recovery_control::IlmRecoveryClassification::Retrying
+                    && *next_classification == recovery_control::IlmRecoveryClassification::Retrying
+                    && previous_attempts == next_attempts
+                    && previous_failures == next_failures
+                    && previous_generation != next_generation;
+                let completion = previous_owner.is_some()
+                    && next_owner.is_none()
+                    && previous_generation == next_generation
+                    && previous_attempts == next_attempts;
+                adjacent && (claim || source_refresh || completion)
+            }
             _ => false,
         };
 
@@ -553,6 +619,14 @@ impl DurableIlmRecordCheckpoint {
         {
             return false;
         }
+        if let Self::RecoveryControl { classification, .. } = terminal
+            && !matches!(
+                classification,
+                recovery_control::IlmRecoveryClassification::Terminal | recovery_control::IlmRecoveryClassification::Abandoned
+            )
+        {
+            return false;
+        }
         if self == terminal || self.validate_successor(terminal).is_ok() {
             return true;
         }
@@ -652,6 +726,32 @@ impl DurableIlmRecordCheckpoint {
                         .is_some_and(|distance| tier_probe_state_reaches(*previous_state, *terminal_state, distance))
                     && (!previous_remote_version_known || previous_remote_version == terminal_remote_version)
             }
+            (
+                Self::RecoveryControl {
+                    identity_sha256: previous_identity,
+                    source_generation_sha256: previous_generation,
+                    first_seen_at_unix_nanos: previous_first_seen,
+                    revision: previous_revision,
+                    attempt_count: previous_attempts,
+                    ..
+                },
+                Self::RecoveryControl {
+                    identity_sha256: terminal_identity,
+                    source_generation_sha256: terminal_generation,
+                    first_seen_at_unix_nanos: terminal_first_seen,
+                    revision: terminal_revision,
+                    attempt_count: terminal_attempts,
+                    classification:
+                        recovery_control::IlmRecoveryClassification::Terminal | recovery_control::IlmRecoveryClassification::Abandoned,
+                    ..
+                },
+            ) => {
+                previous_identity == terminal_identity
+                    && (previous_generation == terminal_generation || terminal_attempts > previous_attempts)
+                    && previous_first_seen == terminal_first_seen
+                    && terminal_revision > previous_revision
+                    && terminal_attempts >= previous_attempts
+            }
             _ => false,
         }
     }
@@ -1219,6 +1319,35 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
                 },
             )
         }
+        DurableIlmRecordKind::RecoveryControl => {
+            let (protocol, control_id) = recovery_control::recovery_control_id_from_record_object_name(path)
+                .map_err(|err| Error::other(err.to_string()))?;
+            let control =
+                recovery_control::IlmRecoveryControl::decode(&control_id, data).map_err(|err| Error::other(err.to_string()))?;
+            let canonical = recovery_control::recovery_control_record_object_name(protocol, &control_id)
+                .map_err(|err| Error::other(err.to_string()))?;
+            if canonical != path || control.identity.protocol != protocol {
+                return Err(Error::other("ILM recovery control path is not canonical"));
+            }
+            let identity_sha256 = checkpoint_hash(&control.identity)?;
+            let source_generation_sha256 = checkpoint_hash(&control.observed_source_generation)?;
+            let owner_fence_sha256 = control.owner.as_ref().map(checkpoint_hash).transpose()?;
+            (
+                "control_id",
+                control_id,
+                DurableIlmRecordCheckpoint::RecoveryControl {
+                    content_sha256,
+                    identity_sha256,
+                    source_generation_sha256,
+                    first_seen_at_unix_nanos: control.first_seen_at_unix_nanos,
+                    revision: control.revision,
+                    classification: control.classification,
+                    attempt_count: control.attempt_count,
+                    consecutive_failure_count: control.consecutive_failure_count,
+                    owner_fence_sha256,
+                },
+            )
+        }
         DurableIlmRecordKind::ManualTransitionJob => {
             let job_id = manual_transition_job::manual_transition_job_id_from_record_object_name(path)
                 .map_err(|err| Error::other(err.to_string()))?;
@@ -1412,6 +1541,94 @@ mod tests {
             .checkpoint
     }
 
+    fn recovery_control_fixture() -> recovery_control::IlmRecoveryControl {
+        let source_path = "ilm/transition-transactions/records/12/34/1234567890abcdef1234567890abcdef.json";
+        let generation = recovery_control::IlmRecoverySourceGeneration::new(
+            transition_transaction::TRANSITION_TRANSACTION_SCHEMA,
+            "source-etag",
+            "a".repeat(64),
+            vec![recovery_control::IlmRecoverySourceCopy {
+                authority: "pool-0/set-0".to_string(),
+                canonical_path: source_path.to_string(),
+                etag: "source-etag".to_string(),
+                encoded_len: 128,
+                content_sha256: "a".repeat(64),
+            }],
+        )
+        .expect("source generation should build");
+        recovery_control::IlmRecoveryControl::new(
+            recovery_control::IlmRecoveryControlIdentity {
+                protocol: recovery_control::IlmRecoveryProtocol::TransitionTransaction,
+                canonical_source_path: source_path.to_string(),
+                stable_operation_identity: "12345678-90ab-cdef-1234-567890abcdef".to_string(),
+                record_class: "transition_transaction_v1".to_string(),
+            },
+            generation,
+            recovery_control::IlmRecoveryClassification::Retrying,
+            1_000_000_000,
+            recovery_control::IlmRecoveryErrorCode::None,
+        )
+        .expect("recovery control should build")
+    }
+
+    fn recovery_control_checkpoint(control: &recovery_control::IlmRecoveryControl) -> DurableIlmRecordCheckpoint {
+        let control_id = control.identity.source_operation_digest().expect("control id should derive");
+        let path = recovery_control::recovery_control_record_object_name(control.identity.protocol, &control_id)
+            .expect("control path should build");
+        let encoded = control.encode().expect("control should encode");
+        let namespace = classify_durable_ilm_record(&path)
+            .expect("recovery control namespace should classify")
+            .expect("recovery control should be durable");
+        assert_eq!(namespace, &RECOVERY_CONTROL_NAMESPACE);
+        validate_durable_ilm_record(&path, &encoded)
+            .expect("recovery control should validate")
+            .checkpoint
+    }
+
+    #[test]
+    fn recovery_control_checkpoint_tracks_claim_retry_and_terminal_generations() {
+        let initial_control = recovery_control_fixture();
+        let initial = recovery_control_checkpoint(&initial_control);
+
+        let mut claimed_control = initial_control;
+        let mut advanced_generation = claimed_control.observed_source_generation.clone();
+        advanced_generation.source_schema = "rustfs-transition-transaction-v2".to_string();
+        claimed_control
+            .claim_for_source_generation("node-a", Uuid::new_v4(), 2_000_000_000, 300_000_000_000, advanced_generation)
+            .expect("control should claim");
+        let claimed = recovery_control_checkpoint(&claimed_control);
+        initial.validate_successor(&claimed).expect("claim should advance receipt");
+
+        let mut retry_control = claimed_control;
+        retry_control
+            .record_retryable_failure(3_000_000_000, recovery_control::IlmRecoveryErrorCode::BackendTimeout)
+            .expect("retry should persist");
+        let retry = recovery_control_checkpoint(&retry_control);
+        claimed.validate_successor(&retry).expect("retry should advance receipt");
+
+        let ready_at = retry_control
+            .next_attempt_at_unix_nanos
+            .expect("retry deadline should persist");
+        let mut terminal_control = retry_control;
+        terminal_control
+            .claim("node-b", Uuid::new_v4(), ready_at, 300_000_000_000)
+            .expect("retry should claim");
+        let reclaimed = recovery_control_checkpoint(&terminal_control);
+        retry.validate_successor(&reclaimed).expect("reclaim should advance receipt");
+        terminal_control
+            .finish_attempt(
+                recovery_control::IlmRecoveryClassification::Terminal,
+                recovery_control::IlmRecoveryErrorCode::None,
+            )
+            .expect("control should terminate");
+        let terminal = recovery_control_checkpoint(&terminal_control);
+        reclaimed
+            .validate_successor(&terminal)
+            .expect("terminal state should advance receipt");
+        assert!(initial.is_predecessor_of_terminal(&terminal));
+        assert!(!initial.is_predecessor_of_terminal(&retry));
+    }
+
     #[test]
     fn tier_probe_intent_checkpoint_tracks_exact_monotonic_generations() {
         let initial_intent = tier_probe_intent_fixture();

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

@@ -24,6 +24,7 @@ pub(crate) use metadata_boundary::{LifecycleExpiryConfigs, get_expiry_configs, g
 mod object_handlers_common;
 mod object_lock_boundary;
 pub use self::core as lifecycle;
+pub mod recovery_control;
 mod replication_sink;
 pub mod rule;
 mod runtime_boundary;

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

@@ -23,6 +23,11 @@ use uuid::Uuid;
 use crate::bucket::lifecycle::config_boundary;
 use crate::bucket::lifecycle::durable_namespace::TRANSITION_TRANSACTION_NAMESPACE;
 use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
+use crate::bucket::lifecycle::recovery_control::{
+    IlmRecoveryClassification, IlmRecoveryControl, IlmRecoveryControlIdentity, IlmRecoveryErrorCode, IlmRecoveryProtocol,
+    ObservedIlmRecoveryControl, load_recovery_control, observe_recovery_source, recovery_control_record_object_name,
+    save_recovery_control_if_absent, save_recovery_control_if_current,
+};
 use crate::bucket::lifecycle::tier_sweeper::{
     delete_confirmed_transition_candidate_exact_with_lease_idempotent,
     delete_object_from_remote_tier_idempotent_with_manager_and_identity,
@@ -44,6 +49,7 @@ const EVENT_LIFECYCLE_TRANSITION_TRANSACTION_RECOVERY: &str = "lifecycle_transit
 pub const DEFAULT_TRANSITION_TRANSACTION_RECOVERY_LIMIT: usize = 1_000;
 const TRANSITION_TRANSACTION_RECOVERY_INTERVAL: Duration = Duration::from_secs(60);
 const TRANSITION_TRANSACTION_RECOVERY_TIMEOUT: Duration = Duration::from_secs(300);
+const TRANSITION_RECOVERY_CONTROL_LEASE_NANOS: i64 = 15 * 60 * 1_000_000_000;
 pub const TRANSITION_TRANSACTION_SCHEMA: &str = "rustfs-transition-transaction-v1";
 pub const TRANSITION_TRANSACTION_PREFIX: &str = "ilm/transition-transactions";
 pub const TRANSITION_TRANSACTION_RECORD_PREFIX: &str = TRANSITION_TRANSACTION_NAMESPACE.prefix;
@@ -737,6 +743,8 @@ pub enum TransitionTransactionRecoveryOutcome {
     RemoteCandidateDeleted,
     RecordDeleted,
     Retained,
+    RetainedAmbiguous(IlmRecoveryErrorCode),
+    OperatorRequired(IlmRecoveryErrorCode),
 }
 
 #[cfg(test)]
@@ -817,6 +825,80 @@ async fn pause_before_transition_recovery_claim(transaction_id: Uuid) {
     }
 }
 
+#[cfg(test)]
+#[derive(Default)]
+struct TransitionRecoveryTerminalBarrierState {
+    transaction_id: Uuid,
+    arrived: tokio::sync::Notify,
+    release: tokio::sync::Notify,
+}
+
+#[cfg(test)]
+pub(crate) struct TransitionRecoveryTerminalBarrier {
+    state: Arc<TransitionRecoveryTerminalBarrierState>,
+}
+
+#[cfg(test)]
+static TRANSITION_RECOVERY_TERMINAL_BARRIER: std::sync::OnceLock<
+    std::sync::Mutex<Option<Arc<TransitionRecoveryTerminalBarrierState>>>,
+> = std::sync::OnceLock::new();
+
+#[cfg(test)]
+impl TransitionRecoveryTerminalBarrier {
+    pub(crate) fn install(transaction_id: Uuid) -> Self {
+        let state = Arc::new(TransitionRecoveryTerminalBarrierState {
+            transaction_id,
+            ..Default::default()
+        });
+        let mut slot = TRANSITION_RECOVERY_TERMINAL_BARRIER
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("transition recovery terminal barrier mutex should not poison");
+        assert!(
+            slot.is_none(),
+            "transition recovery terminal barrier must be installed by one test at a time"
+        );
+        *slot = Some(Arc::clone(&state));
+        drop(slot);
+        Self { state }
+    }
+
+    pub(crate) async fn wait_until_paused(&self) {
+        tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
+            .await
+            .expect("transition recovery should persist terminal control before source cleanup");
+    }
+}
+
+#[cfg(test)]
+impl Drop for TransitionRecoveryTerminalBarrier {
+    fn drop(&mut self) {
+        self.state.release.notify_one();
+        let mut slot = TRANSITION_RECOVERY_TERMINAL_BARRIER
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("transition recovery terminal barrier mutex should not poison");
+        if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) {
+            *slot = None;
+        }
+    }
+}
+
+#[cfg(test)]
+async fn pause_after_transition_recovery_terminal(transaction_id: Uuid) {
+    let barrier = TRANSITION_RECOVERY_TERMINAL_BARRIER
+        .get_or_init(|| std::sync::Mutex::new(None))
+        .lock()
+        .expect("transition recovery terminal barrier mutex should not poison")
+        .as_ref()
+        .filter(|barrier| barrier.transaction_id == transaction_id)
+        .cloned();
+    if let Some(barrier) = barrier {
+        barrier.arrived.notify_one();
+        barrier.release.notified().await;
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
 #[serde(rename_all = "snake_case")]
 pub enum TransitionOperatorProbe {
@@ -1020,17 +1102,35 @@ fn transition_transaction_id_from_record_object_name(object: &str) -> Result<Uui
     let suffix = object
         .strip_prefix(&prefix)
         .ok_or(TransitionTransactionError::Corrupt("transaction record path has wrong prefix"))?;
-    let file_name = suffix
-        .rsplit('/')
+    let mut parts = suffix.split('/');
+    let shard_a = parts
         .next()
         .ok_or(TransitionTransactionError::Corrupt("transaction record path is incomplete"))?;
+    let shard_b = parts
+        .next()
+        .ok_or(TransitionTransactionError::Corrupt("transaction record path is incomplete"))?;
+    let file_name = parts
+        .next()
+        .ok_or(TransitionTransactionError::Corrupt("transaction record path is incomplete"))?;
+    if parts.next().is_some() {
+        return Err(TransitionTransactionError::Corrupt("transaction record path is not canonical"));
+    }
     let transaction_key = file_name
         .strip_suffix(".json")
         .ok_or(TransitionTransactionError::Corrupt("transaction record path has wrong suffix"))?;
-    if transaction_key.len() != 32 || !transaction_key.bytes().all(|byte| byte.is_ascii_hexdigit()) {
+    if transaction_key.len() != 32
+        || !transaction_key
+            .bytes()
+            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
+        || shard_a != &transaction_key[..2]
+        || shard_b != &transaction_key[2..4]
+    {
         return Err(TransitionTransactionError::Corrupt("transaction record path has invalid transaction id"));
     }
-    Uuid::parse_str(transaction_key).map_err(|_| TransitionTransactionError::Corrupt("transaction record path has invalid uuid"))
+    Uuid::parse_str(transaction_key)
+        .ok()
+        .filter(|transaction_id| !transaction_id.is_nil())
+        .ok_or(TransitionTransactionError::Corrupt("transaction record path has invalid uuid"))
 }
 
 pub async fn process_transition_transaction_record(
@@ -1055,6 +1155,27 @@ async fn process_transition_transaction_record_at(
 ) -> EcstoreResult<TransitionTransactionRecoveryOutcome> {
     let record_name =
         transition_transaction_record_object_name(observed.transaction_id).map_err(transition_transaction_store_error)?;
+    let now_unix_nanos =
+        i64::try_from(now_unix_nanos).map_err(|_| Error::other("transition transaction recovery timestamp does not fit i64"))?;
+    let recovery_control_identity = transition_recovery_control_identity(observed, &record_name);
+    let recovery_control_id = recovery_control_identity
+        .source_operation_digest()
+        .map_err(|err| Error::other(err.to_string()))?;
+    let control_record_name =
+        recovery_control_record_object_name(IlmRecoveryProtocol::TransitionTransaction, &recovery_control_id)
+            .map_err(|err| Error::other(err.to_string()))?;
+    let control_lock = if transition_state_needs_recovery_control(observed, now_unix_nanos) {
+        Some(
+            api.new_ns_lock(RUSTFS_META_BUCKET, &format!("{control_record_name}.recovery-lock"))
+                .await?,
+        )
+    } else {
+        None
+    };
+    let _control_guard = match &control_lock {
+        Some(lock) => Some(lock.get_write_lock(crate::set_disk::get_lock_acquire_timeout()).await?),
+        None => None,
+    };
     // The synthetic key avoids nesting the recovery lock with the config
     // object's own I/O lock. Holding it across the bounded source proof and
     // remote DELETE elects one destructive recovery worker across nodes.
@@ -1073,55 +1194,400 @@ async fn process_transition_transaction_record_at(
         return Ok(TransitionTransactionRecoveryOutcome::Retained);
     }
 
-    match current.state {
+    let mut recovery_control = if transition_state_needs_recovery_control(&current, now_unix_nanos) {
+        if cleanup_terminal_transition_recovery_control(
+            api.clone(),
+            &current,
+            &record_name,
+            &recovery_control_identity,
+            &recovery_control_id,
+        )
+        .await?
+        {
+            return Ok(TransitionTransactionRecoveryOutcome::RecordDeleted);
+        }
+        match claim_transition_recovery_control(
+            api.clone(),
+            &current,
+            &record_name,
+            recovery_control_identity,
+            &recovery_control_id,
+            now_unix_nanos,
+        )
+        .await?
+        {
+            Some(control) => Some(control),
+            None => return Ok(TransitionTransactionRecoveryOutcome::Retained),
+        }
+    } else {
+        None
+    };
+
+    let recovery = match current.state {
         TransitionTransactionState::Uploaded => {
-            if transition_transaction_ownership_is_active(&current, now_unix_nanos) {
-                return Ok(TransitionTransactionRecoveryOutcome::Retained);
-            }
-            let mut cleanup = current.clone();
-            cleanup
-                .mark_cleanup_pending(
-                    current.fence(),
-                    TransitionCleanupProof {
-                        transaction_id: current.transaction_id,
-                        write_id: current.write_id,
-                        remote_object: current.remote_object.clone(),
-                        remote_version: current.remote_version.clone(),
-                        backend_fingerprint: current.backend_fingerprint,
-                        decision: TransitionCleanupDecision::UploadAbortedBeforeLocalCommit,
-                    },
-                )
-                .map_err(transition_transaction_store_error)?;
-            #[cfg(test)]
-            pause_before_transition_recovery_claim(current.transaction_id).await;
-            match save_transition_transaction_record_if_current(api.clone(), &current, &cleanup).await {
-                Ok(()) => recover_cleanup_pending(api, &cleanup).await,
-                Err(Error::PreconditionFailed) | Err(Error::ConfigNotFound) => Ok(TransitionTransactionRecoveryOutcome::Retained),
-                Err(err) => Err(err),
+            if transition_transaction_ownership_is_active(&current, i128::from(now_unix_nanos)) {
+                Ok(TransitionTransactionRecoveryOutcome::Retained)
+            } else {
+                let mut cleanup = current.clone();
+                cleanup
+                    .mark_cleanup_pending(
+                        current.fence(),
+                        TransitionCleanupProof {
+                            transaction_id: current.transaction_id,
+                            write_id: current.write_id,
+                            remote_object: current.remote_object.clone(),
+                            remote_version: current.remote_version.clone(),
+                            backend_fingerprint: current.backend_fingerprint,
+                            decision: TransitionCleanupDecision::UploadAbortedBeforeLocalCommit,
+                        },
+                    )
+                    .map_err(transition_transaction_store_error)?;
+                #[cfg(test)]
+                pause_before_transition_recovery_claim(current.transaction_id).await;
+                match save_transition_transaction_record_if_current(api.clone(), &current, &cleanup).await {
+                    Ok(()) => recover_cleanup_pending(api.clone(), &cleanup).await,
+                    Err(Error::PreconditionFailed) | Err(Error::ConfigNotFound) => {
+                        Ok(TransitionTransactionRecoveryOutcome::Retained)
+                    }
+                    Err(err) => Err(err),
+                }
             }
         }
-        TransitionTransactionState::CleanupPending => recover_cleanup_pending(api, &current).await,
+        TransitionTransactionState::CleanupPending => recover_cleanup_pending(api.clone(), &current).await,
         TransitionTransactionState::LocalCommitStarted => match local_commit_matches_transaction(api.clone(), &current).await {
-            Ok(true) => {
-                delete_transition_transaction_record(api, &current).await?;
-                Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
-            }
-            Ok(false) => Ok(TransitionTransactionRecoveryOutcome::Retained),
-            Err(err) if transition_source_is_missing(&err) => Ok(TransitionTransactionRecoveryOutcome::Retained),
+            Ok(true) => Ok(TransitionTransactionRecoveryOutcome::RecordDeleted),
+            Ok(false) => Ok(TransitionTransactionRecoveryOutcome::OperatorRequired(
+                IlmRecoveryErrorCode::LocalCommitAmbiguous,
+            )),
+            Err(err) if transition_source_is_missing(&err) => Ok(TransitionTransactionRecoveryOutcome::OperatorRequired(
+                IlmRecoveryErrorCode::LocalCommitAmbiguous,
+            )),
             Err(err) => Err(err),
         },
         TransitionTransactionState::AbortedNoRemote | TransitionTransactionState::Committed => {
-            delete_transition_transaction_record(api, &current).await?;
             Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
         }
         TransitionTransactionState::UploadOutcomeUnknown => {
-            if transition_transaction_ownership_is_active(&current, now_unix_nanos) {
+            if transition_transaction_ownership_is_active(&current, i128::from(now_unix_nanos)) {
                 Ok(TransitionTransactionRecoveryOutcome::Retained)
             } else {
-                recover_unknown_upload_outcome(api, &current).await
+                recover_unknown_upload_outcome(api.clone(), &current).await
             }
         }
-        TransitionTransactionState::UploadStarted => Ok(TransitionTransactionRecoveryOutcome::Retained),
+        TransitionTransactionState::UploadStarted => {
+            if transition_transaction_ownership_is_active(&current, i128::from(now_unix_nanos)) {
+                Ok(TransitionTransactionRecoveryOutcome::Retained)
+            } else {
+                Ok(TransitionTransactionRecoveryOutcome::RetainedAmbiguous(
+                    IlmRecoveryErrorCode::RemoteVersionUnknown,
+                ))
+            }
+        }
+    };
+
+    if let Some(mut control) = recovery_control.take() {
+        let source_to_delete = if matches!(
+            recovery,
+            Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted
+                | TransitionTransactionRecoveryOutcome::RecordDeleted)
+        ) {
+            let refreshed =
+                refresh_transition_recovery_control_source(api.clone(), control, &record_name, current.transaction_id).await?;
+            control = refreshed.0;
+            refreshed.1
+        } else {
+            None
+        };
+        persist_transition_recovery_result(api.clone(), control, &recovery, now_unix_nanos).await?;
+        if let Some(source) = source_to_delete {
+            #[cfg(test)]
+            pause_after_transition_recovery_terminal(source.transaction_id).await;
+            delete_transition_transaction_record(api, &source).await?;
+        }
+    } else if matches!(
+        recovery,
+        Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted | TransitionTransactionRecoveryOutcome::RecordDeleted)
+    ) {
+        delete_transition_transaction_record(api, &current).await?;
+    }
+    recovery
+}
+
+fn transition_recovery_control_identity(transaction: &TransitionTransaction, record_name: &str) -> IlmRecoveryControlIdentity {
+    IlmRecoveryControlIdentity {
+        protocol: IlmRecoveryProtocol::TransitionTransaction,
+        canonical_source_path: record_name.to_string(),
+        stable_operation_identity: transaction.transaction_id.to_string(),
+        record_class: "transition_transaction_v1".to_string(),
+    }
+}
+
+#[cfg(test)]
+pub(crate) fn transition_recovery_control_id(transaction: &TransitionTransaction) -> Result<String> {
+    let record_name = transition_transaction_record_object_name(transaction.transaction_id)?;
+    transition_recovery_control_identity(transaction, &record_name)
+        .source_operation_digest()
+        .map_err(|_| TransitionTransactionError::Corrupt("transition recovery control identity is invalid"))
+}
+
+fn transition_state_needs_recovery_control(transaction: &TransitionTransaction, now_unix_nanos: i64) -> bool {
+    now_unix_nanos >= transaction.not_after_unix_nanos
+        && !matches!(
+            transaction.state,
+            TransitionTransactionState::AbortedNoRemote | TransitionTransactionState::Committed
+        )
+}
+
+async fn cleanup_terminal_transition_recovery_control(
+    api: Arc<ECStore>,
+    transaction: &TransitionTransaction,
+    record_name: &str,
+    identity: &IlmRecoveryControlIdentity,
+    control_id: &str,
+) -> EcstoreResult<bool> {
+    let observed = match load_recovery_control(api.clone(), IlmRecoveryProtocol::TransitionTransaction, control_id).await {
+        Ok(observed) => observed,
+        Err(Error::ConfigNotFound) => return Ok(false),
+        Err(err) => return Err(err),
+    };
+    if observed.control.classification != IlmRecoveryClassification::Terminal {
+        return Ok(false);
+    }
+    let source = observe_recovery_source(api.clone(), record_name, TRANSITION_TRANSACTION_SCHEMA).await?;
+    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.transaction_id, data).is_ok_and(|decoded| decoded == *transaction)
+        });
+    if observed.control.identity != *identity || !exact_source {
+        return Ok(false);
+    }
+    delete_transition_transaction_record(api, transaction).await?;
+    Ok(true)
+}
+
+async fn claim_transition_recovery_control(
+    api: Arc<ECStore>,
+    transaction: &TransitionTransaction,
+    record_name: &str,
+    identity: IlmRecoveryControlIdentity,
+    control_id: &str,
+    now_unix_nanos: i64,
+) -> EcstoreResult<Option<ObservedIlmRecoveryControl>> {
+    let existing = match load_recovery_control(api.clone(), IlmRecoveryProtocol::TransitionTransaction, control_id).await {
+        Ok(control) => Some(control),
+        Err(Error::ConfigNotFound) => None,
+        Err(err) => return Err(err),
+    };
+    if let Some(observed) = existing.as_ref() {
+        if observed.control.identity != identity {
+            return Ok(None);
+        }
+        if observed
+            .control
+            .owner
+            .as_ref()
+            .is_some_and(|owner| owner.lease_expires_at_unix_nanos <= now_unix_nanos)
+        {
+            let mut expired = observed.control.clone();
+            expired
+                .record_expired_attempt(now_unix_nanos)
+                .map_err(|err| Error::other(err.to_string()))?;
+            save_recovery_control_if_current(api, observed, &expired).await?;
+            return Ok(None);
+        }
+        if !observed.control.should_attempt_at(now_unix_nanos) {
+            return Ok(None);
+        }
+    }
+
+    let source = match observe_recovery_source(api.clone(), record_name, TRANSITION_TRANSACTION_SCHEMA).await {
+        Ok(source) => source,
+        Err(err) => {
+            if let Some(observed) = existing {
+                persist_transition_recovery_source_failure(api, observed, now_unix_nanos).await?;
+                return Ok(None);
+            }
+            return Err(err);
+        }
+    };
+    let source_matches = source.is_consistent()
+        && source.canonical_data.as_deref().is_some_and(|data| {
+            TransitionTransaction::decode(transaction.transaction_id, data).is_ok_and(|observed| observed == *transaction)
+        });
+    let source_error = if source_matches {
+        IlmRecoveryErrorCode::None
+    } else if source.canonical_data.is_some() {
+        IlmRecoveryErrorCode::SourceGenerationChanged
+    } else {
+        IlmRecoveryErrorCode::SourceDivergent
+    };
+
+    let mut observed = match existing {
+        Some(control) => control,
+        None => {
+            let candidate = IlmRecoveryControl::new(
+                identity.clone(),
+                source.generation.clone(),
+                if source_matches {
+                    IlmRecoveryClassification::Retrying
+                } else {
+                    IlmRecoveryClassification::Corrupt
+                },
+                now_unix_nanos,
+                source_error,
+            )
+            .map_err(|err| Error::other(err.to_string()))?;
+            match save_recovery_control_if_absent(api.clone(), &candidate).await {
+                Ok(()) | Err(Error::PreconditionFailed) => {}
+                Err(err) => return Err(err),
+            }
+            load_recovery_control(api.clone(), IlmRecoveryProtocol::TransitionTransaction, control_id).await?
+        }
+    };
+    if observed.control.identity != identity || !observed.control.should_attempt_at(now_unix_nanos) {
+        return Ok(None);
+    }
+
+    let mut claimed = observed.control.clone();
+    claimed
+        .claim_for_source_generation(
+            api.id.to_string(),
+            Uuid::new_v4(),
+            now_unix_nanos,
+            TRANSITION_RECOVERY_CONTROL_LEASE_NANOS,
+            source.generation,
+        )
+        .map_err(|err| Error::other(err.to_string()))?;
+    save_recovery_control_if_current(api.clone(), &observed, &claimed).await?;
+    observed = load_recovery_control(api.clone(), IlmRecoveryProtocol::TransitionTransaction, control_id).await?;
+    if observed.control != claimed {
+        return Err(Error::PreconditionFailed);
+    }
+    if !source_matches {
+        let mut corrupt = observed.control.clone();
+        corrupt
+            .finish_attempt(IlmRecoveryClassification::Corrupt, source_error)
+            .map_err(|err| Error::other(err.to_string()))?;
+        save_recovery_control_if_current(api, &observed, &corrupt).await?;
+        return Ok(None);
+    }
+    Ok(Some(observed))
+}
+
+async fn persist_transition_recovery_source_failure(
+    api: Arc<ECStore>,
+    observed: ObservedIlmRecoveryControl,
+    now_unix_nanos: i64,
+) -> EcstoreResult<()> {
+    let mut claimed = observed.control.clone();
+    claimed
+        .claim(
+            api.id.to_string(),
+            Uuid::new_v4(),
+            now_unix_nanos,
+            TRANSITION_RECOVERY_CONTROL_LEASE_NANOS,
+        )
+        .map_err(|err| Error::other(err.to_string()))?;
+    save_recovery_control_if_current(api.clone(), &observed, &claimed).await?;
+    let claimed = load_recovery_control(
+        api.clone(),
+        IlmRecoveryProtocol::TransitionTransaction,
+        &claimed
+            .identity
+            .source_operation_digest()
+            .map_err(|err| Error::other(err.to_string()))?,
+    )
+    .await?;
+    let mut failed = claimed.control.clone();
+    failed
+        .record_retryable_failure(now_unix_nanos, IlmRecoveryErrorCode::SourceUnavailable)
+        .map_err(|err| Error::other(err.to_string()))?;
+    save_recovery_control_if_current(api, &claimed, &failed).await
+}
+
+async fn refresh_transition_recovery_control_source(
+    api: Arc<ECStore>,
+    mut observed: ObservedIlmRecoveryControl,
+    record_name: &str,
+    transaction_id: Uuid,
+) -> EcstoreResult<(ObservedIlmRecoveryControl, Option<TransitionTransaction>)> {
+    let transaction = match load_transition_transaction_record(api.clone(), transaction_id).await {
+        Ok(transaction) => transaction,
+        Err(Error::ConfigNotFound) => return Ok((observed, None)),
+        Err(err) => return Err(err),
+    };
+    let source = observe_recovery_source(api.clone(), record_name, TRANSITION_TRANSACTION_SCHEMA).await?;
+    let exact_source = source.is_consistent()
+        && 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(Error::PreconditionFailed);
+    }
+    if observed.control.observed_source_generation != source.generation {
+        let mut refreshed = observed.control.clone();
+        refreshed
+            .refresh_owned_source_generation(source.generation)
+            .map_err(|err| Error::other(err.to_string()))?;
+        save_recovery_control_if_current(api.clone(), &observed, &refreshed).await?;
+        observed = load_recovery_control(
+            api,
+            IlmRecoveryProtocol::TransitionTransaction,
+            &refreshed
+                .identity
+                .source_operation_digest()
+                .map_err(|err| Error::other(err.to_string()))?,
+        )
+        .await?;
+        if observed.control != refreshed {
+            return Err(Error::PreconditionFailed);
+        }
+    }
+    Ok((observed, Some(transaction)))
+}
+
+async fn persist_transition_recovery_result(
+    api: Arc<ECStore>,
+    observed: ObservedIlmRecoveryControl,
+    recovery: &EcstoreResult<TransitionTransactionRecoveryOutcome>,
+    now_unix_nanos: i64,
+) -> EcstoreResult<()> {
+    let mut next = observed.control.clone();
+    match recovery {
+        Ok(
+            TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted | TransitionTransactionRecoveryOutcome::RecordDeleted,
+        ) => next
+            .finish_attempt(IlmRecoveryClassification::Terminal, IlmRecoveryErrorCode::None)
+            .map_err(|err| Error::other(err.to_string()))?,
+        Ok(TransitionTransactionRecoveryOutcome::Retained) => next
+            .record_retryable_failure(now_unix_nanos, IlmRecoveryErrorCode::SourceGenerationChanged)
+            .map_err(|err| Error::other(err.to_string()))?,
+        Ok(TransitionTransactionRecoveryOutcome::RetainedAmbiguous(code)) => next
+            .finish_attempt(IlmRecoveryClassification::RetainedAmbiguous, *code)
+            .map_err(|err| Error::other(err.to_string()))?,
+        Ok(TransitionTransactionRecoveryOutcome::OperatorRequired(code)) => next
+            .finish_attempt(IlmRecoveryClassification::OperatorRequired, *code)
+            .map_err(|err| Error::other(err.to_string()))?,
+        Err(err) => next
+            .record_retryable_failure(now_unix_nanos, transition_recovery_error_code(err))
+            .map_err(|err| Error::other(err.to_string()))?,
+    }
+    save_recovery_control_if_current(api, &observed, &next).await
+}
+
+fn transition_recovery_error_code(err: &Error) -> IlmRecoveryErrorCode {
+    match err {
+        Error::PreconditionFailed => IlmRecoveryErrorCode::CasConflict,
+        Error::ConfigNotFound
+        | Error::FileNotFound
+        | Error::FileVersionNotFound
+        | Error::ObjectNotFound(_, _)
+        | Error::VersionNotFound(_, _, _)
+        | Error::BucketNotFound(_) => IlmRecoveryErrorCode::SourceUnavailable,
+        Error::SlowDown => IlmRecoveryErrorCode::BackendThrottled,
+        _ => IlmRecoveryErrorCode::Unknown,
     }
 }
 
@@ -1134,10 +1600,7 @@ async fn recover_cleanup_pending(
     transaction: &TransitionTransaction,
 ) -> EcstoreResult<TransitionTransactionRecoveryOutcome> {
     match local_commit_matches_transaction(api.clone(), transaction).await {
-        Ok(true) => {
-            delete_transition_transaction_record(api, transaction).await?;
-            Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
-        }
+        Ok(true) => Ok(TransitionTransactionRecoveryOutcome::RecordDeleted),
         Ok(false) => delete_unreferenced_transition_candidate(api, transaction).await,
         Err(err) if transition_source_is_missing(&err) => delete_unreferenced_transition_candidate(api, transaction).await,
         Err(err) => Err(err),
@@ -1157,7 +1620,6 @@ async fn delete_unreferenced_transition_candidate(
         return Ok(TransitionTransactionRecoveryOutcome::Retained);
     }
     delete_transition_remote_candidate(api.clone(), &current).await?;
-    delete_transition_transaction_record(api, &current).await?;
     Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
 }
 
@@ -1178,24 +1640,26 @@ async fn recover_unknown_upload_outcome(
         .await
         .map_err(Error::other)?
     {
-        TransitionCandidateProbe::Missing => {
-            delete_transition_transaction_record(api, transaction).await?;
-            Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
-        }
+        TransitionCandidateProbe::Missing => Ok(TransitionTransactionRecoveryOutcome::RecordDeleted),
         TransitionCandidateProbe::UnversionedPresent => {
             cleanup_recovered_unknown_upload_candidate(api, transaction, TransitionRemoteVersion::unversioned()).await
         }
         TransitionCandidateProbe::VersionedPresent(version_id)
             if Uuid::parse_str(&version_id).is_ok_and(|version_id| version_id.is_nil()) =>
         {
-            Ok(TransitionTransactionRecoveryOutcome::Retained)
+            Ok(TransitionTransactionRecoveryOutcome::RetainedAmbiguous(
+                IlmRecoveryErrorCode::RemoteVersionUnknown,
+            ))
         }
         TransitionCandidateProbe::VersionedPresent(version_id) => {
             cleanup_recovered_unknown_upload_candidate(api, transaction, TransitionRemoteVersion::versioned(version_id)).await
         }
-        TransitionCandidateProbe::Ambiguous | TransitionCandidateProbe::Unsupported => {
-            Ok(TransitionTransactionRecoveryOutcome::Retained)
-        }
+        TransitionCandidateProbe::Ambiguous => Ok(TransitionTransactionRecoveryOutcome::RetainedAmbiguous(
+            IlmRecoveryErrorCode::RemoteProbeAmbiguous,
+        )),
+        TransitionCandidateProbe::Unsupported => Ok(TransitionTransactionRecoveryOutcome::RetainedAmbiguous(
+            IlmRecoveryErrorCode::RemoteProbeUnsupported,
+        )),
     }
 }
 
@@ -1323,6 +1787,11 @@ async fn recover_transition_transaction_records_with_now(
             false,
         )
         .await?;
+    if list.is_truncated && list.next_continuation_token.is_none() {
+        return Err(Error::other(
+            "transition transaction recovery returned a truncated page without a continuation marker",
+        ));
+    }
 
     let mut stats = TransitionTransactionRecoveryStats {
         scanned: 0,
@@ -1381,7 +1850,11 @@ async fn recover_transition_transaction_records_with_now(
             ) => {
                 stats.recovered += 1;
             }
-            Ok(TransitionTransactionRecoveryOutcome::Retained) => {
+            Ok(
+                TransitionTransactionRecoveryOutcome::Retained
+                | TransitionTransactionRecoveryOutcome::RetainedAmbiguous(_)
+                | TransitionTransactionRecoveryOutcome::OperatorRequired(_),
+            ) => {
                 stats.retained += 1;
                 debug!(
                     event = EVENT_LIFECYCLE_TRANSITION_TRANSACTION_RECOVERY,
@@ -1509,11 +1982,74 @@ fn state_requires_known_remote_version(state: TransitionTransactionState) -> boo
 #[cfg(test)]
 mod tests {
     use std::collections::HashMap;
+    use std::sync::atomic::{AtomicBool, Ordering};
 
     use super::*;
 
     const BACKEND_FINGERPRINT: [u8; 32] = [7; 32];
 
+    struct RecoveryAttemptDropGuard(Arc<AtomicBool>);
+
+    impl Drop for RecoveryAttemptDropGuard {
+        fn drop(&mut self) {
+            self.0.store(true, Ordering::SeqCst);
+        }
+    }
+
+    async fn pending_recovery_attempt(started: Arc<tokio::sync::Notify>, dropped: Arc<AtomicBool>) -> EcstoreResult<()> {
+        let _drop_guard = RecoveryAttemptDropGuard(dropped);
+        started.notify_one();
+        std::future::pending().await
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn transition_recovery_timeout_and_cancellation_drop_inflight_attempts() {
+        let timeout_started = Arc::new(tokio::sync::Notify::new());
+        let timeout_dropped = Arc::new(AtomicBool::new(false));
+        let timeout_task = tokio::spawn({
+            let started = Arc::clone(&timeout_started);
+            let dropped = Arc::clone(&timeout_dropped);
+            async move {
+                await_transition_transaction_recovery(
+                    &CancellationToken::new(),
+                    TRANSITION_TRANSACTION_RECOVERY_TIMEOUT,
+                    pending_recovery_attempt(started, dropped),
+                )
+                .await
+            }
+        });
+        timeout_started.notified().await;
+        tokio::time::advance(TRANSITION_TRANSACTION_RECOVERY_TIMEOUT).await;
+        let timed_out = timeout_task.await.expect("timeout wrapper task should join");
+        assert!(matches!(timed_out, Some(Err(_))), "outer timeout should fail the recovery pass");
+        assert!(timeout_dropped.load(Ordering::SeqCst), "outer timeout must drop its in-flight attempt");
+
+        let cancel_token = CancellationToken::new();
+        let cancel_started = Arc::new(tokio::sync::Notify::new());
+        let cancel_dropped = Arc::new(AtomicBool::new(false));
+        let cancel_task = tokio::spawn({
+            let cancel_token = cancel_token.clone();
+            let started = Arc::clone(&cancel_started);
+            let dropped = Arc::clone(&cancel_dropped);
+            async move {
+                await_transition_transaction_recovery(
+                    &cancel_token,
+                    TRANSITION_TRANSACTION_RECOVERY_TIMEOUT,
+                    pending_recovery_attempt(started, dropped),
+                )
+                .await
+            }
+        });
+        cancel_started.notified().await;
+        cancel_token.cancel();
+        let cancelled = cancel_task.await.expect("cancellation wrapper task should join");
+        assert!(cancelled.is_none(), "outer cancellation should stop the recovery loop");
+        assert!(
+            cancel_dropped.load(Ordering::SeqCst),
+            "outer cancellation must drop its in-flight attempt"
+        );
+    }
+
     #[derive(Default)]
     struct MemoryTransactionStore {
         records: HashMap<Uuid, Vec<u8>>,
@@ -1968,5 +2504,19 @@ mod tests {
             transition_transaction_record_object_name(Uuid::nil()),
             Err(TransitionTransactionError::Corrupt("transaction_id is nil"))
         ));
+        assert_eq!(
+            transition_transaction_id_from_record_object_name(&object).expect("canonical record path should parse"),
+            transaction_id
+        );
+        for malformed in [
+            object.to_ascii_uppercase(),
+            object.replace("/aa/aa/", "/ff/aa/"),
+            object.replace("/aa/aa/", "/aa/aa/extra/"),
+        ] {
+            assert!(matches!(
+                transition_transaction_id_from_record_object_name(&malformed),
+                Err(TransitionTransactionError::Corrupt(_))
+            ));
+        }
     }
 }

crates/ecstore/src/services/tier/tier.rs

@@ -143,12 +143,11 @@ struct TierDriverBuildBarrier {
 static TIER_DRIVER_BUILD_BARRIER: LazyLock<Mutex<Option<Arc<TierDriverBuildBarrier>>>> = LazyLock::new(|| Mutex::new(None));
 
 #[cfg(test)]
-pub(crate) type TierDriverTestFactory =
-    Arc<dyn Fn(&TierConfig) -> std::result::Result<WarmBackendImpl, AdminError> + Send + Sync + 'static>;
+type TierDriverTestFactory = Arc<dyn Fn(&TierConfig) -> std::result::Result<WarmBackendImpl, AdminError> + Send + Sync + 'static>;
 
 #[cfg(test)]
 tokio::task_local! {
-    pub(crate) static TIER_DRIVER_TEST_FACTORY: TierDriverTestFactory;
+    static TIER_DRIVER_TEST_FACTORY: TierDriverTestFactory;
 }
 
 #[cfg(test)]
@@ -1378,42 +1377,27 @@ async fn ensure_no_authoritative_persisted_references<S>(
 where
     S: TierReferenceProofStore,
 {
-    ensure_no_authoritative_persisted_references_with(
-        api.clone(),
-        TIER_DELETE_JOURNAL_PREFIX,
-        "tier-delete journal",
-        |_object, data| {
-            let journal = decode_tier_delete_journal_entry(data).map_err(io::Error::other)?;
-            Ok((
-                journal.tier_name.clone(),
-                tier_persisted_reference_blocks_any_target(&journal.tier_name, journal.backend_identity, targets),
-            ))
-        },
-    )
+    ensure_no_authoritative_persisted_references_with(api.clone(), TIER_DELETE_JOURNAL_PREFIX, |_object, data| {
+        let journal = decode_tier_delete_journal_entry(data).map_err(io::Error::other)?;
+        Ok((
+            journal.tier_name.clone(),
+            tier_persisted_reference_blocks_any_target(&journal.tier_name, journal.backend_identity, targets),
+        ))
+    })
     .await?;
-    ensure_no_authoritative_persisted_references_with(
-        api,
-        TRANSITION_TRANSACTION_RECORD_PREFIX,
-        "transition transaction",
-        |object, data| {
-            let transaction = decode_transition_transaction_record(object, data).map_err(io::Error::other)?;
-            Ok((
-                transaction.tier_name.clone(),
-                tier_persisted_reference_blocks_any_target(
-                    &transaction.tier_name,
-                    Some(transaction.backend_fingerprint),
-                    targets,
-                ),
-            ))
-        },
-    )
+    ensure_no_authoritative_persisted_references_with(api, TRANSITION_TRANSACTION_RECORD_PREFIX, |object, data| {
+        let transaction = decode_transition_transaction_record(object, data).map_err(io::Error::other)?;
+        Ok((
+            transaction.tier_name.clone(),
+            tier_persisted_reference_blocks_any_target(&transaction.tier_name, Some(transaction.backend_fingerprint), targets),
+        ))
+    })
     .await
 }
 
 async fn ensure_no_authoritative_persisted_references_with<S, F>(
     api: Arc<S>,
     prefix: &str,
-    reference_kind: &str,
     blocks_target: F,
 ) -> std::result::Result<(), AdminError>
 where
@@ -1442,7 +1426,7 @@ where
                 .map_err(tier_reference_proof_admin_error)?;
             let (tier_name, blocks) = blocks_target(&object.name, &data).map_err(tier_reference_proof_admin_error)?;
             if blocks {
-                return Err(tier_reference_proof_persisted_in_use_error(&tier_name, reference_kind, &object.name));
+                return Err(tier_reference_proof_persisted_in_use_error(&tier_name, &object.name));
             }
         }
         if !page.is_truncated {
@@ -1504,21 +1488,16 @@ fn tier_persisted_reference_blocks_target(
 
 fn tier_reference_proof_in_use_error(tier_name: &str, object: &ObjectInfo) -> AdminError {
     let mut err = ERR_TIER_BACKEND_IN_USE.clone();
-    let reference_kind = if object.transitioned_object.free_version {
-        "free-version ownership"
-    } else {
-        "transitioned-object"
-    };
     err.message = format!(
-        "Remote tier {tier_name} still has a {reference_kind} reference, for example {}/{}",
+        "Remote tier {tier_name} still has object references, for example {}/{}",
         object.bucket, object.name
     );
     err
 }
 
-fn tier_reference_proof_persisted_in_use_error(tier_name: &str, reference_kind: &str, object: &str) -> AdminError {
+fn tier_reference_proof_persisted_in_use_error(tier_name: &str, object: &str) -> AdminError {
     let mut err = ERR_TIER_BACKEND_IN_USE.clone();
-    err.message = format!("Remote tier {tier_name} still has a {reference_kind} reference, for example {object}");
+    err.message = format!("Remote tier {tier_name} still has a persisted reference, for example {object}");
     err
 }
 
@@ -3706,17 +3685,13 @@ impl TierConfigMgr {
         let manager = handle.read().await;
         let runtime = tier_driver_runtime(handle, &manager);
         let runtime = lock_unpoisoned(&runtime);
-        let prepared = runtime
-            .prepared_mutation_blocks
-            .values()
-            .any(|blocked_mutation_id| *blocked_mutation_id == mutation_id);
-        let committed = runtime
+        if !runtime
             .committed_mutation_blocks
             .values()
-            .any(|mutation_ids| mutation_ids.contains(&mutation_id));
-        if !prepared && !committed {
+            .any(|mutation_ids| mutation_ids.contains(&mutation_id))
+        {
             let mut err = ERR_TIER_INVALID_CONFIG.clone();
-            err.message = "Remote tier mutation fence was not installed".to_string();
+            err.message = "Remote tier committed mutation fence was not installed".to_string();
             return Err(err);
         }
         Ok(MutationBlockAllowance {
@@ -3923,6 +3898,14 @@ impl TierConfigMgr {
         Self::begin_tier_transition_with_destinations(handle, manager, changed, replaced_destinations, mutation_block_allowance)
     }
 
+    fn begin_tier_transition(
+        handle: &Arc<RwLock<Self>>,
+        manager: &mut Self,
+        changed: HashSet<String>,
+    ) -> std::result::Result<TierPublishTransition, AdminError> {
+        Self::begin_tier_transition_with_destinations(handle, manager, changed, HashMap::new(), None)
+    }
+
     fn begin_tier_transition_with_destinations(
         handle: &Arc<RwLock<Self>>,
         manager: &mut Self,
@@ -4193,7 +4176,7 @@ impl TierConfigMgr {
                     let mut config_lock = config_lock;
                     let coordinated_config_update = config_lock.is_some();
                     let mut update = Some(update);
-                    let (mutation_kind, explicit_tier_name, mutation_force, current_for_targets, driver_tier, target_tiers) =
+                    let (mutation_kind, explicit_tier_name, mutation_force, current_for_targets, driver_tier, mut transition) =
                         match mutation {
                             TierCandidateMutation::Prevalidated(prepared) => {
                                 if version != prepared.version {
@@ -4209,8 +4192,9 @@ impl TierConfigMgr {
                                     )));
                                 }
                                 candidate = prepared.candidate;
-                                let target_tiers = {
-                                    let manager = handle.read().await;
+                                let validation_deadline = Instant::now() + TIER_REMOTE_VALIDATION_TIMEOUT;
+                                let mut transition = {
+                                    let mut manager = handle.write().await;
                                     let mut target_tiers = changed_tier_names(&manager, &candidate);
                                     if let Some(tier_name) = prepared.explicit_tier_name.as_ref()
                                         && (manager.tiers.contains_key(tier_name)
@@ -4219,7 +4203,8 @@ impl TierConfigMgr {
                                     {
                                         target_tiers.insert(tier_name.clone());
                                     }
-                                    target_tiers
+                                    Self::begin_tier_transition(&handle, &mut manager, target_tiers)
+                                        .map_err(TierConfigUpdateError::Publish)?
                                 };
                                 (
                                     prepared.kind,
@@ -4227,7 +4212,7 @@ impl TierConfigMgr {
                                     prepared.force,
                                     prepared.current,
                                     prepared.driver_tier,
-                                    target_tiers,
+                                    transition,
                                 )
                             }
                             mutation => {
@@ -4252,9 +4237,11 @@ impl TierConfigMgr {
                                     last_refreshed_at: candidate.last_refreshed_at,
                                 };
                                 let validation_deadline = Instant::now() + TIER_REMOTE_VALIDATION_TIMEOUT;
-                                let target_tiers = {
-                                    let manager = handle.read().await;
-                                    mutation.target_tiers(&manager, &candidate)
+                                let mut transition = {
+                                    let mut manager = handle.write().await;
+                                    let target_tiers = mutation.target_tiers(&manager, &candidate);
+                                    Self::begin_tier_transition(&handle, &mut manager, target_tiers)
+                                        .map_err(TierConfigUpdateError::Publish)?
                                 };
                                 let driver_tier = apply_tier_candidate_mutation(mutation, &mut candidate, validation_deadline)
                                     .await
@@ -4265,7 +4252,7 @@ impl TierConfigMgr {
                                     mutation_force,
                                     current_for_targets,
                                     driver_tier,
-                                    target_tiers,
+                                    transition,
                                 )
                             }
                         };
@@ -4288,83 +4275,28 @@ impl TierConfigMgr {
                     save_coordinator_tier_mutation_intent(api.clone(), coordinator_intent.as_ref())
                         .await
                         .map_err(TierConfigUpdateError::Save)?;
-                    let mut blocked_target_tiers = target_tiers.clone();
                     if let Some(intent) = coordinator_intent.as_ref() {
-                        blocked_target_tiers.extend(intent.affected_targets.iter().map(|target| target.tier_name.clone()));
-                    }
-                    if let Some(intent) = coordinator_intent.as_ref() {
-                        // `target_tiers` may include a stale local-only manager
-                        // entry that is absent from the persisted proof
-                        // snapshot. Fence that local transition under the same
-                        // mutation ID as well; recovery may discard this
-                        // process-local superset, which advances the revision
-                        // and makes the deferred transition fail closed.
-                        TierConfigMgr::apply_prepared_mutation_intent_block_for_tiers(&handle, intent, &blocked_target_tiers)
+                        TierConfigMgr::apply_prepared_mutation_intent_block(&handle, intent)
                             .await
                             .map_err(TierConfigUpdateError::Publish)?;
                     }
-                    let prepared_mutation_block_allowance = match coordinator_intent.as_ref() {
-                        Some(intent) => Some(
-                            TierConfigMgr::mutation_block_allowance_for(&handle, intent.mutation_id)
-                                .await
-                                .map_err(TierConfigUpdateError::Publish)?,
-                        ),
-                        None => None,
-                    };
-                    // A durable coordinator intent supplies the admission
-                    // fence that lets us defer generation revocation. Keep the
-                    // original early transition for no-intent paths (for
-                    // example, reconciling a stale local manager to an
-                    // idempotently removed persisted tier), where there is no
-                    // Prepared record capable of blocking a new lease.
-                    let (mut transition, deferred_target_tiers) = if coordinator_intent.is_some() {
-                        (None, Some(target_tiers))
-                    } else {
-                        let transition = {
-                            let mut manager = handle.write().await;
-                            Self::begin_tier_transition_with_destinations(
-                                &handle,
-                                &mut manager,
-                                target_tiers,
-                                HashMap::new(),
-                                None,
-                            )
-                            .map_err(TierConfigUpdateError::Publish)?
-                        };
-                        (Some(transition), None)
-                    };
                     if coordinated_config_update {
                         drop(update.take());
                         drop(config_lock.take());
                     }
-                    // The durable Prepared block closes admission before the
-                    // zero-reference proof, but deliberately leaves already
-                    // issued generations current. In particular, an exact
-                    // free-version cleanup that has completed remote DELETE
-                    // must still be able to remove its local ownership marker;
-                    // revoking its generation here would strand that marker
-                    // and make this mutation reject its own interrupted work.
-                    if let Some(intent) = coordinator_intent.as_ref()
-                        && let Err(drain_error) =
-                            TierConfigMgr::wait_for_blocked_tier_operation_leases_for_tiers(&handle, &blocked_target_tiers).await
-                    {
-                        if !abort_prepared_tier_mutation(&handle, api.clone(), Some(intent), Vec::new()).await {
+                    let drain_deadline = Instant::now() + TIER_REMOTE_VALIDATION_TIMEOUT;
+                    if let Err(drain_error) = transition.wait_for_active_leases_until(drain_deadline).await {
+                        if !abort_prepared_tier_mutation(&handle, api.clone(), coordinator_intent.as_ref(), Vec::new()).await {
                             warn!(
                                 event = "tier_mutation_abort",
                                 component = LOG_COMPONENT_ECSTORE,
                                 subsystem = LOG_SUBSYSTEM_TIER,
                                 result = "prepared_intent_retained",
-                                mutation_id = %intent.mutation_id,
+                                coordinator_intent = coordinator_intent.is_some(),
                                 "tier mutation lease drain failed and abort was incomplete"
                             );
                         }
                         return Err(TierConfigUpdateError::Publish(drain_error));
-                    } else if let Some(transition) = transition.as_ref() {
-                        let drain_deadline = Instant::now() + TIER_REMOTE_VALIDATION_TIMEOUT;
-                        transition
-                            .wait_for_active_leases_until(drain_deadline)
-                            .await
-                            .map_err(TierConfigUpdateError::Publish)?;
                     }
                     let prepared_peers = if let Some(intent) = coordinator_intent.as_ref() {
                         let peers = match remote_tier_mutation_peers().await {
@@ -4431,71 +4363,6 @@ impl TierConfigMgr {
                         }
                         return Err(TierConfigUpdateError::Publish(proof_error));
                     }
-                    // No affected-tier lease can start after Prepared, and the
-                    // existing set was drained above. It is now safe to revoke
-                    // the generation for publication without invalidating a
-                    // cleanup between its remote and local commit boundaries.
-                    if transition.is_none() {
-                        let target_tiers = deferred_target_tiers.ok_or_else(|| {
-                            let mut err = ERR_TIER_INVALID_CONFIG.clone();
-                            err.message = "Remote tier mutation lost its deferred transition targets".to_string();
-                            TierConfigUpdateError::Publish(err)
-                        })?;
-                        let mut manager = handle.write().await;
-                        transition = Some(
-                            match Self::begin_tier_transition_with_destinations(
-                                &handle,
-                                &mut manager,
-                                target_tiers,
-                                HashMap::new(),
-                                prepared_mutation_block_allowance.as_ref(),
-                            ) {
-                                Ok(transition) => transition,
-                                Err(transition_error) => {
-                                    drop(manager);
-                                    if !abort_prepared_tier_mutation(
-                                        &handle,
-                                        api.clone(),
-                                        coordinator_intent.as_ref(),
-                                        prepared_peers,
-                                    )
-                                    .await
-                                    {
-                                        warn!(
-                                            event = "tier_mutation_abort",
-                                            component = LOG_COMPONENT_ECSTORE,
-                                            subsystem = LOG_SUBSYSTEM_TIER,
-                                            result = "prepared_intent_retained",
-                                            coordinator_intent = coordinator_intent.is_some(),
-                                            "tier mutation publish transition failed and abort was incomplete"
-                                        );
-                                    }
-                                    return Err(TierConfigUpdateError::Publish(transition_error));
-                                }
-                            },
-                        );
-                    }
-                    let mut transition = transition.ok_or_else(|| {
-                        let mut err = ERR_TIER_INVALID_CONFIG.clone();
-                        err.message = "Remote tier mutation lost its publish transition".to_string();
-                        TierConfigUpdateError::Publish(err)
-                    })?;
-                    let drain_deadline = Instant::now() + TIER_REMOTE_VALIDATION_TIMEOUT;
-                    if let Err(drain_error) = transition.wait_for_active_leases_until(drain_deadline).await {
-                        drop(transition);
-                        if !abort_prepared_tier_mutation(&handle, api.clone(), coordinator_intent.as_ref(), prepared_peers).await
-                        {
-                            warn!(
-                                event = "tier_mutation_abort",
-                                component = LOG_COMPONENT_ECSTORE,
-                                subsystem = LOG_SUBSYSTEM_TIER,
-                                result = "prepared_intent_retained",
-                                coordinator_intent = coordinator_intent.is_some(),
-                                "tier mutation publish drain failed and abort was incomplete"
-                            );
-                        }
-                        return Err(TierConfigUpdateError::Publish(drain_error));
-                    }
                     let candidate_digest = tier_config_candidate_digest(&candidate).map_err(TierConfigUpdateError::Save)?;
                     if coordinated_config_update {
                         config_lock = match Self::acquire_tier_config_write_lock(api.clone()).await {
@@ -5547,40 +5414,11 @@ impl TierConfigMgr {
         handle: &Arc<RwLock<Self>>,
         intent: &TierMutationIntent,
     ) -> std::result::Result<(), AdminError> {
-        let target_tiers = intent
-            .affected_targets
-            .iter()
-            .map(|target| target.tier_name.clone())
-            .collect();
-        Self::apply_prepared_mutation_intent_block_for_tiers(handle, intent, &target_tiers).await
-    }
-
-    async fn apply_prepared_mutation_intent_block_for_tiers(
-        handle: &Arc<RwLock<Self>>,
-        intent: &TierMutationIntent,
-        target_tiers: &HashSet<String>,
-    ) -> std::result::Result<(), AdminError> {
-        if intent.state != TierMutationIntentState::Prepared {
-            return Ok(());
-        }
         let manager = handle.read().await;
         let runtime = tier_driver_runtime(handle, &manager);
         let mut runtime = lock_unpoisoned(&runtime);
         let mut prepared_mutation_blocks = runtime.prepared_mutation_blocks.clone();
         Self::collect_prepared_mutation_intent_block(&mut prepared_mutation_blocks, intent)?;
-        for tier_name in target_tiers {
-            match prepared_mutation_blocks.entry(tier_name.clone()) {
-                Entry::Vacant(entry) => {
-                    entry.insert(intent.mutation_id);
-                }
-                Entry::Occupied(entry) if *entry.get() == intent.mutation_id => {}
-                Entry::Occupied(_) => {
-                    let mut err = ERR_TIER_BACKEND_IN_USE.clone();
-                    err.message = format!("Remote tier {tier_name} already has another prepared mutation");
-                    return Err(err);
-                }
-            }
-        }
         if prepared_mutation_blocks == runtime.prepared_mutation_blocks {
             return Ok(());
         }
@@ -5596,18 +5434,6 @@ impl TierConfigMgr {
     pub(crate) async fn wait_for_blocked_tier_operation_leases(
         handle: &Arc<RwLock<Self>>,
         intent: &TierMutationIntent,
-    ) -> std::result::Result<(), AdminError> {
-        let target_tiers = intent
-            .affected_targets
-            .iter()
-            .map(|target| target.tier_name.clone())
-            .collect();
-        Self::wait_for_blocked_tier_operation_leases_for_tiers(handle, &target_tiers).await
-    }
-
-    async fn wait_for_blocked_tier_operation_leases_for_tiers(
-        handle: &Arc<RwLock<Self>>,
-        target_tiers: &HashSet<String>,
     ) -> std::result::Result<(), AdminError> {
         let generations = {
             let manager = handle.read().await;
@@ -5615,9 +5441,10 @@ impl TierConfigMgr {
                 return Ok(());
             };
             let runtime = lock_unpoisoned(&runtime);
-            target_tiers
+            intent
+                .affected_targets
                 .iter()
-                .filter_map(|tier_name| runtime.generations.get(tier_name).cloned())
+                .filter_map(|target| runtime.generations.get(&target.tier_name).cloned())
                 .collect::<Vec<_>>()
         };
         let drain = async {
@@ -14587,13 +14414,6 @@ mod tests {
                 .push(object);
         }
 
-        fn remove_listed_version(&self, bucket: &str, object: &str) {
-            self.listed_versions
-                .lock()
-                .expect("tier reference fixture should not poison")
-                .retain(|version| version.bucket != bucket || version.name != object);
-        }
-
         fn add_lifecycle_config(&self, bucket: &str, config: BucketLifecycleConfiguration) {
             self.lifecycle_configs
                 .lock()
@@ -15862,152 +15682,6 @@ mod tests {
         );
     }
 
-    #[tokio::test]
-    async fn tier_remove_prepared_fence_allows_inflight_free_version_cleanup_to_finish() {
-        let store = Arc::new(CasConfigStore::default());
-        let tier = build_rustfs_tier("COLD-A");
-        let identity = tier_backend_identity(&tier).expect("test tier identity should encode");
-        let mut persisted = empty_mgr();
-        persisted.tiers.insert("COLD-A".to_string(), tier.clone_with_credentials());
-        persisted
-            .save_tiering_config_if_current(store.clone(), None)
-            .await
-            .expect("free-version drain fixture should persist");
-
-        let manager = TierConfigMgr::new();
-        {
-            let mut guard = manager.write().await;
-            guard.tiers.insert("COLD-A".to_string(), tier);
-            guard.tiers.insert("COLD-B".to_string(), build_rustfs_tier("COLD-B"));
-            guard
-                .replace_driver("COLD-A", Box::new(LeaseTestBackend::ready("cleanup")))
-                .expect("cleanup driver generation should install");
-            guard
-                .replace_driver("COLD-B", Box::new(LeaseTestBackend::ready("stale-local")))
-                .expect("stale local driver generation should install");
-        }
-        let cleanup_lease = TierConfigMgr::acquire_operation_lease(&manager, "COLD-A")
-            .await
-            .expect("in-flight cleanup lease should be available");
-        let stale_local_lease = TierConfigMgr::acquire_operation_lease(&manager, "COLD-B")
-            .await
-            .expect("stale local tier lease should be available");
-        let mut free_version = transitioned_tier_object("photos", "2026/free-version.jpg", "COLD-A", Some(identity));
-        free_version.transitioned_object.status = "pending".to_string();
-        free_version.transitioned_object.free_version = true;
-        store.add_listed_version(free_version);
-
-        let remove_manager = manager.clone();
-        let remove_store = store.clone();
-        let remove = tokio::spawn(async move {
-            TIER_MUTATION_TEST_PEERS
-                .scope(
-                    Vec::new(),
-                    TierConfigMgr::remove_and_save_with(&remove_manager, remove_store, "COLD-A", true),
-                )
-                .await
-        });
-
-        tokio::time::timeout(Duration::from_secs(1), async {
-            loop {
-                let guard = manager.read().await;
-                let runtime = registered_tier_driver_runtime(&guard).expect("runtime should remain registered");
-                let prepared = {
-                    let runtime = lock_unpoisoned(&runtime);
-                    runtime.prepared_mutation_blocks.contains_key("COLD-A")
-                        && runtime.prepared_mutation_blocks.contains_key("COLD-B")
-                };
-                if prepared {
-                    break;
-                }
-                drop(guard);
-                tokio::task::yield_now().await;
-            }
-        })
-        .await
-        .expect("tier remove should install its durable prepared fence");
-
-        assert!(
-            cleanup_lease.is_current(&manager).await,
-            "the prepared fence must let the already leased cleanup finish its exact local marker deletion"
-        );
-        assert!(
-            stale_local_lease.is_current(&manager).await,
-            "the local superset fence must also let an already leased stale-manager operation finish"
-        );
-        let blocked = match TierConfigMgr::acquire_operation_lease(&manager, "COLD-A").await {
-            Ok(_) => panic!("the prepared fence must reject new tier operations"),
-            Err(err) => err,
-        };
-        assert!(TierConfigMgr::operation_lease_blocked_by_mutation(&blocked));
-        let stale_blocked = match TierConfigMgr::acquire_operation_lease(&manager, "COLD-B").await {
-            Ok(_) => panic!("the local superset fence must reject new stale-manager operations"),
-            Err(err) => err,
-        };
-        assert!(TierConfigMgr::operation_lease_blocked_by_mutation(&stale_blocked));
-
-        store.remove_listed_version("photos", "2026/free-version.jpg");
-        drop(cleanup_lease);
-        drop(stale_local_lease);
-
-        tokio::time::timeout(Duration::from_secs(5), remove)
-            .await
-            .expect("tier remove should finish after both in-flight operations release their leases")
-            .expect("tier remove task should join")
-            .expect("tier remove should pass once the in-flight cleanup removes its marker");
-        assert!(!manager.read().await.tiers.contains_key("COLD-A"));
-        assert!(!manager.read().await.tiers.contains_key("COLD-B"));
-        assert!(
-            !load_tier_config_for_update(store)
-                .await
-                .expect("removed tier config should reload")
-                .0
-                .tiers
-                .contains_key("COLD-A")
-        );
-    }
-
-    #[tokio::test]
-    async fn no_intent_stale_manager_removal_keeps_early_generation_drain() {
-        let store = Arc::new(CasConfigStore::default());
-        empty_mgr()
-            .save_tiering_config_if_current(store.clone(), None)
-            .await
-            .expect("empty persisted tier config should exist");
-        let manager = TierConfigMgr::new();
-        {
-            let mut guard = manager.write().await;
-            install_lease_backend(&mut guard, "COLD-A", LeaseTestBackend::ready("stale"));
-        }
-        let old = TierConfigMgr::acquire_operation_lease(&manager, "COLD-A")
-            .await
-            .expect("stale manager lease should be available");
-
-        let remove_manager = manager.clone();
-        let remove_store = store.clone();
-        let remove =
-            tokio::spawn(async move { TierConfigMgr::remove_and_save_with(&remove_manager, remove_store, "COLD-A", true).await });
-        tokio::time::timeout(Duration::from_secs(1), async {
-            while old.is_current(&manager).await {
-                tokio::task::yield_now().await;
-            }
-        })
-        .await
-        .expect("no-intent stale-manager reconciliation should revoke before its proof");
-        let blocked = match TierConfigMgr::acquire_operation_lease(&manager, "COLD-A").await {
-            Ok(_) => panic!("stale-manager reconciliation must not admit a new operation"),
-            Err(err) => err,
-        };
-        assert!(TierConfigMgr::operation_lease_blocked_by_mutation(&blocked));
-
-        drop(old);
-        remove
-            .await
-            .expect("stale-manager removal task should join")
-            .expect("stale-manager removal should converge to the persisted empty config");
-        assert!(!manager.read().await.tiers.contains_key("COLD-A"));
-    }
-
     async fn assert_lifecycle_only_reference_obeys_force(clear: bool, force: bool) {
         let store = Arc::new(CasConfigStore::default());
         let tier = build_rustfs_tier("COLD-A");
@@ -16911,23 +16585,12 @@ mod tests {
             .await
         });
         tokio::time::timeout(Duration::from_secs(1), async {
-            loop {
-                let guard = manager.read().await;
-                let runtime = registered_tier_driver_runtime(&guard).expect("runtime should remain registered");
-                let prepared = lock_unpoisoned(&runtime).prepared_mutation_blocks.contains_key("COLD-A");
-                if prepared {
-                    break;
-                }
-                drop(guard);
+            while old.is_current(&manager).await {
                 tokio::task::yield_now().await;
             }
         })
         .await
-        .expect("owned update should install its prepared fence before caller cancellation");
-        assert!(
-            old.is_current(&manager).await,
-            "an already leased operation must remain current until it can finish"
-        );
+        .expect("owned update should revoke before caller cancellation");
         caller.abort();
         drop(old);
 
@@ -16972,23 +16635,12 @@ mod tests {
             .await
         });
         tokio::time::timeout(Duration::from_secs(1), async {
-            loop {
-                let guard = manager.read().await;
-                let runtime = registered_tier_driver_runtime(&guard).expect("runtime should remain registered");
-                let prepared = lock_unpoisoned(&runtime).prepared_mutation_blocks.contains_key("COLD-A");
-                if prepared {
-                    break;
-                }
-                drop(guard);
+            while old.is_current(&manager).await {
                 tokio::task::yield_now().await;
             }
         })
         .await
-        .expect("owned update should install its prepared fence before caller cancellation");
-        assert!(
-            old.is_current(&manager).await,
-            "the prepared fence must not invalidate an already leased operation"
-        );
+        .expect("owned update should revoke before caller cancellation");
         caller.abort();
 
         let config_file = tier_config_lock_path();
@@ -17374,71 +17026,6 @@ mod tests {
         assert!(current.tiers.contains_key("COLD-B"));
     }
 
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn reference_proof_rejects_a_changed_prepared_fence_revision_before_publish() {
-        let manager = TierConfigMgr::new();
-        let store = Arc::new(CasConfigStore::default());
-        let mut persisted = empty_mgr();
-        persisted.tiers.insert("COLD-A".to_string(), build_rustfs_tier("COLD-A"));
-        persisted
-            .save_tiering_config_if_current(store.clone(), None)
-            .await
-            .expect("prepared-fence revision fixture should persist");
-        {
-            let mut guard = manager.write().await;
-            install_lease_backend(&mut guard, "COLD-A", LeaseTestBackend::ready("old"));
-        }
-
-        let barrier = tier_reference_proof_test_barrier();
-        let scoped_barrier = barrier.clone();
-        let update_manager = manager.clone();
-        let update_store = store.clone();
-        let update = tokio::spawn(async move {
-            TIER_REFERENCE_PROOF_TEST_BARRIER
-                .scope(
-                    scoped_barrier,
-                    TIER_MUTATION_TEST_PEERS.scope(
-                        Vec::new(),
-                        TierConfigMgr::update_candidate_with_config_lock(
-                            &update_manager,
-                            update_store,
-                            TierCandidateMutation::Remove("COLD-A".to_string(), true),
-                        ),
-                    ),
-                )
-                .await
-        });
-        barrier.arrived.notified().await;
-
-        let unrelated = prepared_remove_intent("COLD-B", uuid::Uuid::from_u128(0x2237));
-        TierConfigMgr::apply_prepared_mutation_intent_block(&manager, &unrelated)
-            .await
-            .expect("an unrelated prepared fence should advance the runtime revision");
-        barrier.release.add_permits(1);
-
-        let err = update
-            .await
-            .expect("tier update task should join")
-            .expect_err("a reference proof cannot authorize publication across a fence revision change");
-        let TierConfigUpdateError::Publish(err) = err else {
-            panic!("the stale prepared-fence allowance should fail publication: {err:?}");
-        };
-        assert!(err.message.contains("changed before replacement"), "{err}");
-        assert!(manager.read().await.tiers.contains_key("COLD-A"));
-        assert!(
-            load_tier_config_for_update(store)
-                .await
-                .expect("rejected tier config should remain readable")
-                .0
-                .tiers
-                .contains_key("COLD-A")
-        );
-        TierConfigMgr::clear_prepared_mutation_intent_block(&manager, unrelated.mutation_id)
-            .await
-            .expect("unrelated test fence should clear");
-    }
-
     #[tokio::test]
     #[serial_test::serial]
     async fn caller_cancellation_after_durable_prepare_does_not_hide_the_mutation() {

crates/ecstore/src/store/init.rs

@@ -825,6 +825,11 @@ mod tests {
                 manual_transition_scope_record_object_name, manual_transition_task_object_name,
                 manual_transition_worker_result_object_name, manual_transition_worker_result_task_key,
             },
+            recovery_control::{
+                IlmRecoveryClassification, IlmRecoveryControl, IlmRecoveryControlIdentity, IlmRecoveryErrorCode,
+                IlmRecoveryProtocol, MAX_RECOVERY_ATTEMPTS, load_recovery_control, observe_recovery_source,
+                save_recovery_control_if_absent,
+            },
             tier_delete_journal::{
                 DecommissionCheckpointTargetFailureHook, TIER_DELETE_DISPATCH_MANIFEST_PREFIX, TIER_DELETE_JOURNAL_PREFIX,
                 TierDeleteChunkTestBarrier, TierDeleteChunkTestStage, TierDeleteDispatchBatchLimitGuard,
@@ -844,12 +849,13 @@ mod tests {
             },
             transition_transaction::{
                 TRANSITION_TRANSACTION_RECORD_PREFIX, TransitionCleanupDecision, TransitionCleanupProof, TransitionOperatorError,
-                TransitionOperatorProbe, TransitionRecoveryClaimBarrier, 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, save_transition_transaction_record_if_current,
+                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,
+                save_transition_transaction_record_if_current, transition_recovery_control_id,
                 transition_transaction_record_object_name,
             },
             validate_durable_ilm_record,
@@ -19239,6 +19245,105 @@ mod tests {
         assert!(!Arc::ptr_eq(&ctx_a, &ctx_b), "the regression requires two distinct instance contexts");
     }
 
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn transition_transaction_recovery_expires_abandoned_attempt_at_budget_bound() {
+        let temp_dir = tempfile::tempdir().expect("create temp store dir");
+        let (ctx, store, _shutdown) = without_storage_class_env(build_isolated_test_store(
+            temp_dir.path(),
+            "transition-transaction-expired-attempt-budget",
+            &[4],
+        ))
+        .await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+
+        let transaction = TransitionTransaction::new(TransitionTransactionInit {
+            deployment_id: ctx.deployment_id().expect("test store should initialize deployment id"),
+            transaction_id: uuid::Uuid::new_v4(),
+            owner_epoch: uuid::Uuid::new_v4(),
+            write_id: uuid::Uuid::new_v4(),
+            source: TransitionSourceIdentity {
+                bucket: "source-bucket".to_string(),
+                object: "source-object".to_string(),
+                version_id: Some(uuid::Uuid::new_v4()),
+                data_dir: uuid::Uuid::new_v4(),
+                mod_time_unix_nanos: 1_770_000_000_000_000_000,
+                size: 42,
+                etag: "source-etag".to_string(),
+                version_mode: TransitionSourceVersionMode::Versioned,
+            },
+            tier_name: "UNUSEDABANDONEDTIER".to_string(),
+            backend_fingerprint: [7; 32],
+            not_after_unix_nanos: 1,
+        })
+        .expect("transaction should build");
+        save_transition_transaction_record(store.clone(), &transaction)
+            .await
+            .expect("transaction record should persist");
+
+        let record_name =
+            transition_transaction_record_object_name(transaction.transaction_id).expect("transaction record name should derive");
+        let source = observe_recovery_source(
+            store.clone(),
+            &record_name,
+            crate::bucket::lifecycle::transition_transaction::TRANSITION_TRANSACTION_SCHEMA,
+        )
+        .await
+        .expect("transaction source generation should be observable");
+        let mut control = IlmRecoveryControl::new(
+            IlmRecoveryControlIdentity {
+                protocol: IlmRecoveryProtocol::TransitionTransaction,
+                canonical_source_path: record_name,
+                stable_operation_identity: transaction.transaction_id.to_string(),
+                record_class: "transition_transaction_v1".to_string(),
+            },
+            source.generation,
+            IlmRecoveryClassification::Retrying,
+            2_000_000_000,
+            IlmRecoveryErrorCode::None,
+        )
+        .expect("recovery control should build");
+        let mut now = 3_000_000_000;
+        for _ in 1..MAX_RECOVERY_ATTEMPTS {
+            now = now.max(control.next_attempt_at_unix_nanos.unwrap_or(now));
+            control
+                .claim("cancelled-or-timed-out-owner", uuid::Uuid::new_v4(), now, 1)
+                .expect("abandoned attempt should claim");
+            control
+                .record_expired_attempt(now + 1)
+                .expect("expired attempt should consume retry budget");
+            now += 2;
+        }
+        now = now.max(control.next_attempt_at_unix_nanos.expect("last retry should have a backoff"));
+        control
+            .claim("cancelled-or-timed-out-owner", uuid::Uuid::new_v4(), now, 1)
+            .expect("final abandoned attempt should claim");
+        save_recovery_control_if_absent(store.clone(), &control)
+            .await
+            .expect("claimed recovery control should persist");
+
+        let stats = recover_transition_transaction_records_at(store.clone(), 100, None, i128::from(now + 1))
+            .await
+            .expect("recovery should account for the expired attempt");
+        assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (1, 0, 1, 0));
+
+        let control_id = transition_recovery_control_id(&transaction).expect("control id should derive");
+        let persisted = load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &control_id)
+            .await
+            .expect("expired recovery control should remain inspectable");
+        assert_eq!(persisted.control.classification, IlmRecoveryClassification::OperatorRequired);
+        assert_eq!(persisted.control.attempt_count, u64::from(MAX_RECOVERY_ATTEMPTS));
+        assert_eq!(persisted.control.consecutive_failure_count, MAX_RECOVERY_ATTEMPTS);
+        assert_eq!(persisted.control.last_error_code, IlmRecoveryErrorCode::AttemptLeaseExpired);
+        assert!(persisted.control.owner.is_none());
+        assert_eq!(
+            transition_transaction_record_count(store).await,
+            1,
+            "budget exhaustion must retain the source record"
+        );
+    }
+
     #[cfg(feature = "test-util")]
     #[tokio::test]
     #[serial_test::serial(storage_class_env)]
@@ -19351,6 +19456,7 @@ mod tests {
             ),
         ];
         let mut expected_removes = Vec::new();
+        let mut recovery_control_ids = Vec::new();
         for (case, put_version, remote_version, source_mode) in cases {
             let mut transaction = TransitionTransaction::new(TransitionTransactionInit {
                 deployment_id: ctx.deployment_id().expect("test store should initialize deployment id"),
@@ -19388,6 +19494,8 @@ mod tests {
             save_transition_transaction_record(store.clone(), &transaction)
                 .await
                 .expect("transaction record should persist");
+            recovery_control_ids
+                .push(transition_recovery_control_id(&transaction).expect("transition recovery control id should derive"));
             expected_removes.push((transaction.remote_object, put_version));
         }
 
@@ -19403,6 +19511,14 @@ mod tests {
         assert_eq!(actual_removes, expected_removes, "recovery must preserve each remote version shape");
         assert_eq!(backend.exact_remove_count(), 2);
         assert_eq!(backend.object_count().await, 0);
+        for control_id in recovery_control_ids {
+            let control = load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &control_id)
+                .await
+                .expect("completed recovery control should remain inspectable");
+            assert_eq!(control.control.classification, IlmRecoveryClassification::Terminal);
+            assert_eq!(control.control.attempt_count, 1);
+            assert!(control.control.owner.is_none());
+        }
 
         let replay = recover_transition_transaction_records(store, 100, None)
             .await
@@ -19415,6 +19531,94 @@ mod tests {
         );
     }
 
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn transition_transaction_recovery_resumes_source_cleanup_after_terminal_crash() {
+        let temp_dir = tempfile::tempdir().expect("create temp store dir");
+        let (ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "transition-transaction-terminal-crash", &[4]))
+                .await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+
+        let tier_name = "TXTERMINALCRASH";
+        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("tier lease should resolve")
+            .backend_identity();
+        let remote_version = uuid::Uuid::new_v4().to_string();
+        let mut transaction = TransitionTransaction::new(TransitionTransactionInit {
+            deployment_id: ctx.deployment_id().expect("test store should initialize deployment id"),
+            transaction_id: uuid::Uuid::new_v4(),
+            owner_epoch: uuid::Uuid::new_v4(),
+            write_id: uuid::Uuid::new_v4(),
+            source: TransitionSourceIdentity {
+                bucket: "source-bucket".to_string(),
+                object: "source-object".to_string(),
+                version_id: Some(uuid::Uuid::new_v4()),
+                data_dir: uuid::Uuid::new_v4(),
+                mod_time_unix_nanos: 1_770_000_000_000_000_000,
+                size: 42,
+                etag: "source-etag".to_string(),
+                version_mode: TransitionSourceVersionMode::Versioned,
+            },
+            tier_name: tier_name.to_string(),
+            backend_fingerprint: backend_identity,
+            not_after_unix_nanos: 1,
+        })
+        .expect("transaction should build");
+        transaction
+            .advance(
+                transaction.fence(),
+                TransitionTransactionState::Uploaded,
+                Some(TransitionRemoteVersion::versioned(remote_version.clone())),
+            )
+            .expect("transaction should enter uploaded state");
+        backend.set_put_remote_version(Some(remote_version)).await;
+        let candidate = bytes::Bytes::from_static(b"terminal crash candidate");
+        backend
+            .put(
+                &transaction.remote_object,
+                ReaderImpl::Body(candidate.clone()),
+                i64::try_from(candidate.len()).expect("test candidate length should fit i64"),
+            )
+            .await
+            .expect("mock backend should accept candidate");
+        save_transition_transaction_record(store.clone(), &transaction)
+            .await
+            .expect("transaction record should persist");
+        let control_id = transition_recovery_control_id(&transaction).expect("control id should derive");
+
+        let barrier = TransitionRecoveryTerminalBarrier::install(transaction.transaction_id);
+        let recovery_store = store.clone();
+        let recovery = tokio::spawn(async move { recover_transition_transaction_records(recovery_store, 100, None).await });
+        barrier.wait_until_paused().await;
+        let terminal = load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &control_id)
+            .await
+            .expect("terminal control should persist before source cleanup");
+        assert_eq!(terminal.control.classification, IlmRecoveryClassification::Terminal);
+        assert_eq!(transition_transaction_record_count(store.clone()).await, 1);
+        assert_eq!(backend.object_count().await, 0);
+        assert_eq!(backend.exact_remove_count(), 1);
+
+        recovery.abort();
+        assert!(
+            recovery
+                .await
+                .expect_err("recovery should be cancelled at the crash boundary")
+                .is_cancelled()
+        );
+        drop(barrier);
+
+        let replay = recover_transition_transaction_records(store.clone(), 100, None)
+            .await
+            .expect("terminal control should resume source cleanup without another remote delete");
+        assert_eq!((replay.scanned, replay.recovered, replay.retained, replay.failed), (1, 1, 0, 0));
+        assert_eq!(transition_transaction_record_count(store).await, 0);
+        assert_eq!(backend.exact_remove_count(), 1, "terminal replay must not repeat the remote delete");
+    }
+
     #[cfg(feature = "test-util")]
     #[tokio::test]
     #[serial_test::serial(storage_class_env)]
@@ -19472,6 +19676,8 @@ mod tests {
         save_transition_transaction_record(store.clone(), &uploaded)
             .await
             .expect("transaction record should persist");
+        let recovery_control_id =
+            transition_recovery_control_id(&uploaded).expect("transition recovery control id should derive");
 
         let barrier = TransitionRecoveryClaimBarrier::install(uploaded.transaction_id);
         let recovery_store = store.clone();
@@ -19493,11 +19699,17 @@ mod tests {
             .expect("recovery should treat the lost CAS as a retained transaction");
         assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (1, 0, 1, 0));
         assert_eq!(
-            load_transition_transaction_record(store, uploaded.transaction_id)
+            load_transition_transaction_record(store.clone(), uploaded.transaction_id)
                 .await
                 .expect("newer transaction revision must remain"),
             active
         );
+        let control = load_recovery_control(store, IlmRecoveryProtocol::TransitionTransaction, &recovery_control_id)
+            .await
+            .expect("lost source CAS should retain a retryable recovery control");
+        assert_eq!(control.control.classification, IlmRecoveryClassification::Retrying);
+        assert_eq!(control.control.consecutive_failure_count, 1);
+        assert_eq!(control.control.last_error_code, IlmRecoveryErrorCode::SourceGenerationChanged);
         assert_eq!(backend.object_count().await, 1, "a stale recovery must not delete the candidate");
         assert_eq!(backend.remove_count().await, 0);
     }
@@ -19799,27 +20011,15 @@ mod tests {
             not_after_unix_nanos: 1_780_000_000_000_000_000,
         })
         .expect("transaction should build");
-        let uploaded_fence = transaction
+        transaction
             .advance(
                 transaction.fence(),
                 TransitionTransactionState::Uploaded,
-                Some(TransitionRemoteVersion::versioned(remote_version)),
+                Some(TransitionRemoteVersion::versioned(remote_version.clone())),
             )
             .expect("transaction should enter uploaded state");
-        transaction
-            .mark_cleanup_pending(
-                uploaded_fence,
-                TransitionCleanupProof {
-                    transaction_id: transaction.transaction_id,
-                    write_id: transaction.write_id,
-                    remote_object: transaction.remote_object.clone(),
-                    remote_version: transaction.remote_version.clone(),
-                    backend_fingerprint: transaction.backend_fingerprint,
-                    decision: TransitionCleanupDecision::UploadAbortedBeforeLocalCommit,
-                },
-            )
-            .expect("transaction should enter cleanup pending state");
         let candidate = bytes::Bytes::from_static(b"cleanup pending candidate retained after failure");
+        backend.set_put_remote_version(Some(remote_version)).await;
         backend
             .put(
                 &transaction.remote_object,
@@ -19831,6 +20031,8 @@ mod tests {
         save_transition_transaction_record(store.clone(), &transaction)
             .await
             .expect("transaction record should persist");
+        let recovery_control_id =
+            transition_recovery_control_id(&transaction).expect("transition recovery control id should derive");
 
         backend.set_remove_failure(true);
         let stats = recover_transition_transaction_records(store.clone(), 100, None)
@@ -19846,6 +20048,42 @@ mod tests {
         assert_eq!(backend.remove_versions().await, Vec::<(String, String)>::new());
         assert_eq!(backend.exact_remove_count(), 1);
         assert_eq!(backend.object_count().await, 1);
+        let control = load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &recovery_control_id)
+            .await
+            .expect("failed recovery control should persist");
+        assert_eq!(control.control.classification, IlmRecoveryClassification::Retrying);
+        assert_eq!(control.control.attempt_count, 1);
+        assert_eq!(control.control.consecutive_failure_count, 1);
+        assert!(
+            control
+                .control
+                .next_attempt_at_unix_nanos
+                .is_some_and(|next| next > OffsetDateTime::now_utc().unix_timestamp_nanos() as i64)
+        );
+
+        backend.set_remove_failure(false);
+        let replay = recover_transition_transaction_records(store.clone(), 100, None)
+            .await
+            .expect("recovery before the persisted deadline should be skipped");
+        assert_eq!((replay.scanned, replay.recovered, replay.retained, replay.failed), (1, 0, 1, 0));
+        assert_eq!(backend.exact_remove_count(), 1, "persisted backoff must prevent an immediate retry");
+
+        let retry_at = control
+            .control
+            .next_attempt_at_unix_nanos
+            .expect("retry deadline should persist");
+        let retried = recover_transition_transaction_records_at(store.clone(), 100, None, i128::from(retry_at) + 1)
+            .await
+            .expect("recovery at the persisted deadline should retry the advanced source generation");
+        assert_eq!((retried.scanned, retried.recovered, retried.retained, retried.failed), (1, 1, 0, 0));
+        assert_eq!(backend.exact_remove_count(), 2);
+        assert_eq!(backend.object_count().await, 0);
+        assert_eq!(transition_transaction_record_count(store.clone()).await, 0);
+        let terminal = load_recovery_control(store, IlmRecoveryProtocol::TransitionTransaction, &recovery_control_id)
+            .await
+            .expect("completed retry control should remain inspectable");
+        assert_eq!(terminal.control.classification, IlmRecoveryClassification::Terminal);
+        assert_eq!(terminal.control.attempt_count, 2);
     }
 
     #[cfg(feature = "test-util")]
@@ -20146,6 +20384,10 @@ mod tests {
         local_commit_started
             .advance(local_commit_started.fence(), TransitionTransactionState::LocalCommitStarted, None)
             .expect("transaction should enter local commit state");
+        let upload_started_control_id =
+            transition_recovery_control_id(&upload_started).expect("upload-started control id should derive");
+        let local_commit_control_id =
+            transition_recovery_control_id(&local_commit_started).expect("local-commit control id should derive");
 
         backend.set_put_remote_version(Some(remote_version)).await;
         for transaction in [&upload_started, &local_commit_started] {
@@ -20176,6 +20418,19 @@ mod tests {
         assert_eq!(backend.object_count().await, 2, "recovery must not delete an unproven remote candidate");
         assert_eq!(backend.remove_count().await, 0);
         assert_eq!(backend.exact_remove_count(), 0);
+        let upload_started_control =
+            load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &upload_started_control_id)
+                .await
+                .expect("upload-started control should persist");
+        assert_eq!(
+            upload_started_control.control.classification,
+            IlmRecoveryClassification::RetainedAmbiguous
+        );
+        let local_commit_control =
+            load_recovery_control(store, IlmRecoveryProtocol::TransitionTransaction, &local_commit_control_id)
+                .await
+                .expect("local-commit control should persist");
+        assert_eq!(local_commit_control.control.classification, IlmRecoveryClassification::OperatorRequired);
     }
 
     #[cfg(feature = "test-util")]
@@ -20526,6 +20781,12 @@ mod tests {
             "an unsupported provider probe must retain the unknown upload"
         );
         assert_eq!(transition_transaction_record_count(store.clone()).await, 1);
+        let recovery_control_id =
+            transition_recovery_control_id(&transaction).expect("transition recovery control id should derive");
+        let control = load_recovery_control(store.clone(), IlmRecoveryProtocol::TransitionTransaction, &recovery_control_id)
+            .await
+            .expect("unsupported probe control should persist");
+        assert_eq!(control.control.classification, IlmRecoveryClassification::RetainedAmbiguous);
         assert!(
             backend.contains(&transaction.remote_object).await,
             "unsupported recovery must not delete the candidate"

crates/scanner/tests/lifecycle_integration_test.rs

@@ -1020,6 +1020,7 @@ mod serial_tests {
         }
 
         let (_disk_paths, ecstore) = setup_isolated_test_env(false).await;
+        let expired_recovery_time = i128::from(i64::MAX / 2);
 
         for case in [
             CleanupCase::Persisted,
@@ -1117,7 +1118,7 @@ mod serial_tests {
                         .await
                         .expect("active unknown ownership must remain fenced after the transaction store was offline");
                     assert_eq!((retained.scanned, retained.recovered, retained.retained, retained.failed), (1, 0, 1, 0));
-                    let recovered = recover_transition_transaction_records_at(ecstore.clone(), 100, None, i128::MAX)
+                    let recovered = recover_transition_transaction_records_at(ecstore.clone(), 100, None, expired_recovery_time)
                         .await
                         .expect("expired unknown ownership may use the provider's missing proof");
                     assert_eq!(
@@ -1166,7 +1167,7 @@ mod serial_tests {
                     assert_eq!(retained.recovered, 0);
                     assert_eq!(retained.retained + retained.failed, 1);
                     backend.set_remove_failure(false);
-                    let recovered = recover_transition_transaction_records_at(ecstore.clone(), 100, None, i128::MAX)
+                    let recovered = recover_transition_transaction_records_at(ecstore.clone(), 100, None, expired_recovery_time)
                         .await
                         .expect("expired recovery should delete the candidate after the backend becomes available");
                     assert_eq!(

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

@@ -91,7 +91,7 @@ Lock ordering is part of the recovery contract. Callers acquire only the locks n
 
 | Path | Current acquisition order | Operations allowed while held | Operations forbidden while held |
 |---|---|---|---|
-| Tier add/edit/remove/clear | A short tier-config namespace WRITE lock captures the persisted config ETag, then releases before backend validation. After validation, namespace WRITE then `admin_updates` protect the ETag check and durable coordinator Prepared write. The Prepared runtime block rejects new tier-operation leases while already-issued leases remain current and drain through their complete local cleanup. Both guards are released for that drain, peer Prepare, and reference proof. Only after the proof succeeds does the coordinator revoke the blocked generation, then it reacquires both guards in the same order for final identity checks and config CAS. Both are released again after the coordinator becomes durably Committed | Backend validation, existing leased cleanup, peer fanout, and reference proof run without either exclusive guard. The exact Prepared mutation ID is the only allowance for the late publish transition; its generation drain is expected to be empty because the Prepared block admitted no new lease. Immediately before config CAS the coordinator revalidates the ETag, candidate digest, exact Prepared intent identity, and intent expiry. The durable Committed intent is recovery authority while peer Commit and local publication finish without the exclusive guards | Revoking a generation before already-leased cleanup has removed its exact local ownership marker; admitting a new lease after Prepared; ordinary manager `RwLock` or runtime-state `Mutex` across awaited network I/O. Recovery must retain an unexpired Prepared coordinator whose old ETag is still current; the old ETag alone is not abandonment proof. Remote object DELETE is never part of mutation |
+| Tier add/edit/remove/clear | A short tier-config namespace WRITE lock captures the persisted config ETag, then releases before backend validation. After validation, namespace WRITE then `admin_updates` protect the ETag check and durable coordinator Prepared write. Both guards are released for lease drain, peer Prepare, and reference proof, then reacquired in the same order for final identity checks and config CAS. Both are released again after the coordinator becomes durably Committed | Backend validation, peer fanout, and reference proof run without either exclusive guard. Immediately before config CAS the coordinator revalidates the ETag, candidate digest, exact Prepared intent identity, and intent expiry. The durable Committed intent is recovery authority while peer Commit and local publication finish without the exclusive guards | Ordinary manager `RwLock` and runtime-state `Mutex` guards must not cross awaited network I/O. Recovery must retain an unexpired Prepared coordinator whose old ETag is still current; the old ETag alone is not abandonment proof. Remote object DELETE is never part of mutation |
 | v6 manifest prepare | Caller already holds the bucket-lifecycle WRITE fence; caller acquires a bucket-metadata transaction READ guard covering the Object Lock and bucket-incarnation snapshot and keeps it through local mutation; exact tier-generation leases; fleet/topology proof; for a single dispatch, synthetic manifest-operation WRITE; for a child, parent-operation WRITE then child-operation WRITE | Build and write one immutable bounded journal set and manifest, validate exact set/digest, then authorize local dispatch while both caller-held bucket guards and all leases remain current. A parent binding is durable before child authorization | Remote tier DELETE; per-object worker cleanup; releasing the metadata guard or a required lease before the authorized local mutation completes; child-to-parent nested lock acquisition |
 | v6 manifest/parent recovery | Fleet/topology proof; bucket-lifecycle WRITE lock; then exactly one synthetic manifest- or parent-operation WRITE lock | Read/write manifest, parent, and journal metadata; verify exact set/digest/binding; converge or roll back child records; advance a parent only after child completion | Remote tier DELETE; per-object worker cleanup; rollback after authorization; taking a child lock while holding a parent lock in background recovery |
 | v5 journal destructive recovery | Synthetic per-journal recovery lock; bucket-lifecycle READ lock; exact tier-generation lease; all physical object READ locks in stable pool/set order | Authoritative source/free-version scan; fenced state CAS; for an eligible terminal state, one bounded remote DELETE; conditional record cleanup | Any delete when a lock or lease is lost; publishing local metadata; selecting an arbitrary backend/version |

rustfs/src/admin/handlers/ilm_transition.rs

@@ -18,18 +18,20 @@ use crate::admin::runtime_sources::object_store_from_extensions;
 use crate::admin::storage_api::bucket::is_reserved_or_invalid_bucket;
 use crate::admin::storage_api::error::StorageError;
 use crate::admin::storage_api::lifecycle::{
-    ManualTransitionCancelCheck, ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionProgressSink,
-    ManualTransitionQueueSnapshot, ManualTransitionRunOptions, ManualTransitionRunReport, ManualTransitionScopeAdmission,
-    ManualTransitionScopeAdmissionClaim, TransitionOperatorDeleteResult, TransitionOperatorError,
-    claim_manual_transition_scope_admission, 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_transition_transaction_for_operator,
+    IlmRecoveryClassification, IlmRecoveryProtocol, ManualTransitionCancelCheck, ManualTransitionJobRecord,
+    ManualTransitionJobState, ManualTransitionProgressSink, ManualTransitionQueueSnapshot, ManualTransitionRunOptions,
+    ManualTransitionRunReport, ManualTransitionScopeAdmission, ManualTransitionScopeAdmissionClaim,
+    TransitionOperatorDeleteResult, TransitionOperatorError, claim_manual_transition_scope_admission,
+    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_transition_transaction_for_operator, list_recovery_controls,
     load_manual_transition_job_record, load_manual_transition_scope_admission, 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,
 };
 use crate::admin::storage_api::runtime::ECStore;
+use crate::admin::storage_api::s3::{S3ErrorCode as AdminS3ErrorCode, error as admin_s3_error};
 use crate::admin::utils::json_response;
 use crate::server::{ADMIN_PREFIX, RemoteAddr};
 use http::HeaderMap;
@@ -230,9 +232,48 @@ pub fn register_ilm_transition_route(r: &mut S3Router<AdminOperation>) -> std::i
         format!("{ADMIN_PREFIX}/v3/ilm/transition/reconcile/{{transaction_id}}").as_str(),
         AdminOperation(&TransitionReconcileApplyHandler {}),
     )?;
+    r.insert(
+        Method::GET,
+        format!("{ADMIN_PREFIX}/v3/ilm/recovery/records").as_str(),
+        AdminOperation(&IlmRecoveryControlListHandler {}),
+    )?;
+    r.insert(
+        Method::GET,
+        format!("{ADMIN_PREFIX}/v3/ilm/recovery/records/{{control_id}}").as_str(),
+        AdminOperation(&IlmRecoveryControlInspectHandler {}),
+    )?;
     Ok(())
 }
 
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct IlmRecoveryControlListQuery {
+    protocol: IlmRecoveryProtocol,
+    #[serde(default)]
+    classification: Option<IlmRecoveryClassification>,
+    #[serde(default = "default_recovery_control_list_limit")]
+    limit: usize,
+    #[serde(default)]
+    marker: Option<String>,
+}
+
+const fn default_recovery_control_list_limit() -> usize {
+    100
+}
+
+fn parse_recovery_control_list_query(query: Option<&str>) -> S3Result<IlmRecoveryControlListQuery> {
+    let query = query.ok_or_else(|| admin_s3_error(AdminS3ErrorCode::InvalidRequest, "protocol is required"))?;
+    let parsed: IlmRecoveryControlListQuery = serde_urlencoded::from_bytes(query.as_bytes())
+        .map_err(|_| admin_s3_error(AdminS3ErrorCode::InvalidArgument, "invalid ILM recovery control query"))?;
+    if !(1..=1_000).contains(&parsed.limit) {
+        return Err(admin_s3_error(AdminS3ErrorCode::InvalidArgument, "limit must be between 1 and 1000"));
+    }
+    if parsed.marker.as_ref().is_some_and(|marker| marker.is_empty()) {
+        return Err(admin_s3_error(AdminS3ErrorCode::InvalidArgument, "marker must not be empty"));
+    }
+    Ok(parsed)
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum ManualTransitionRunMode {
     EnqueueOnly,
@@ -423,6 +464,26 @@ fn transition_transaction_id_from_params(params: &Params<'_, '_>) -> S3Result<Uu
         .map_err(|_| s3_error!(InvalidArgument, "invalid transition transaction id"))
 }
 
+fn recovery_control_id_from_params(params: &Params<'_, '_>) -> S3Result<String> {
+    let control_id = params.get("control_id").unwrap_or("");
+    if control_id.len() != 64
+        || !control_id
+            .bytes()
+            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
+    {
+        return Err(admin_s3_error(AdminS3ErrorCode::InvalidArgument, "invalid ILM recovery control id"));
+    }
+    Ok(control_id.to_string())
+}
+
+fn map_recovery_control_error(err: StorageError) -> S3Error {
+    if err == StorageError::ConfigNotFound {
+        admin_s3_error(AdminS3ErrorCode::NoSuchKey, "ILM recovery control not found")
+    } else {
+        admin_s3_error(AdminS3ErrorCode::InternalError, "ILM recovery control request failed")
+    }
+}
+
 fn map_transition_operator_error(err: TransitionOperatorError) -> S3Error {
     match err {
         TransitionOperatorError::NotFound => s3_error!(NoSuchKey, "transition transaction not found"),
@@ -1031,6 +1092,40 @@ impl Operation for TransitionReconcileInspectHandler {
     }
 }
 
+pub struct IlmRecoveryControlListHandler {}
+
+#[async_trait::async_trait]
+impl Operation for IlmRecoveryControlListHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        authorize_transition_admin_request(&req, AdminAction::ListTierAction).await?;
+        let query = parse_recovery_control_list_query(req.uri.query())?;
+        let Some(store) = object_store_from_extensions(&req.extensions) else {
+            return Err(admin_s3_error(AdminS3ErrorCode::InternalError, "object store is not initialized"));
+        };
+        let page = list_recovery_controls(store, query.protocol, query.classification, query.limit, query.marker)
+            .await
+            .map_err(map_recovery_control_error)?;
+        json_response(StatusCode::OK, &page)
+    }
+}
+
+pub struct IlmRecoveryControlInspectHandler {}
+
+#[async_trait::async_trait]
+impl Operation for IlmRecoveryControlInspectHandler {
+    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        authorize_transition_admin_request(&req, AdminAction::ListTierAction).await?;
+        let control_id = recovery_control_id_from_params(&params)?;
+        let Some(store) = object_store_from_extensions(&req.extensions) else {
+            return Err(admin_s3_error(AdminS3ErrorCode::InternalError, "object store is not initialized"));
+        };
+        let control = inspect_recovery_control(store, &control_id)
+            .await
+            .map_err(map_recovery_control_error)?;
+        json_response(StatusCode::OK, &control)
+    }
+}
+
 pub struct TransitionReconcileApplyHandler {}
 
 #[async_trait::async_trait]
@@ -1104,6 +1199,49 @@ mod tests {
         f(&matched.params)
     }
 
+    fn with_recovery_control_params<T>(path: &str, f: impl FnOnce(&Params<'_, '_>) -> T) -> T {
+        let mut router = Router::new();
+        router
+            .insert("/rustfs/admin/v3/ilm/recovery/records/{control_id}", ())
+            .expect("route should insert");
+        let matched = router.at(path).expect("route should match");
+        f(&matched.params)
+    }
+
+    #[test]
+    fn recovery_control_query_is_bounded_and_strict() {
+        let query = parse_recovery_control_list_query(Some("protocol=transition_transaction"))
+            .expect("minimal recovery query should parse");
+        assert_eq!(query.protocol, IlmRecoveryProtocol::TransitionTransaction);
+        assert_eq!(query.classification, None);
+        assert_eq!(query.limit, 100);
+
+        let filtered = parse_recovery_control_list_query(Some(
+            "protocol=tier_delete_journal&classification=retained_ambiguous&limit=1000&marker=opaque",
+        ))
+        .expect("bounded filtered query should parse");
+        assert_eq!(filtered.protocol, IlmRecoveryProtocol::TierDeleteJournal);
+        assert_eq!(filtered.classification, Some(IlmRecoveryClassification::RetainedAmbiguous));
+        assert_eq!(filtered.limit, 1000);
+        assert!(parse_recovery_control_list_query(None).is_err());
+        assert!(parse_recovery_control_list_query(Some("protocol=transition_transaction&limit=0")).is_err());
+        assert!(parse_recovery_control_list_query(Some("protocol=transition_transaction&limit=1001")).is_err());
+        assert!(parse_recovery_control_list_query(Some("protocol=unknown")).is_err());
+        assert!(parse_recovery_control_list_query(Some("protocol=transition_transaction&extra=true")).is_err());
+    }
+
+    #[test]
+    fn recovery_control_id_is_canonical_lowercase_sha256() {
+        let id = "ab".repeat(32);
+        with_recovery_control_params(&format!("/rustfs/admin/v3/ilm/recovery/records/{id}"), |params| {
+            assert_eq!(recovery_control_id_from_params(params).expect("control id should parse"), id);
+        });
+        let uppercase = "AB".repeat(32);
+        with_recovery_control_params(&format!("/rustfs/admin/v3/ilm/recovery/records/{uppercase}"), |params| {
+            assert!(recovery_control_id_from_params(params).is_err())
+        });
+    }
+
     fn manual_transition_job_request(method: Method, path: &'static str) -> S3Request<Body> {
         S3Request {
             input: Body::empty(),

rustfs/src/admin/handlers/tier.rs

@@ -613,16 +613,6 @@ impl Operation for RemoveTier {
             return if err.code == ERR_TIER_NOT_FOUND.code {
                 Err(S3Error::with_message(S3ErrorCode::Custom("TierNotFound".into()), "tier not found"))
             } else if let Some(response) = tier_backend_error_response(&err) {
-                warn!(
-                    event = EVENT_ADMIN_TIER_STATE,
-                    component = LOG_COMPONENT_ADMIN,
-                    subsystem = LOG_SUBSYSTEM_TIER,
-                    action = "remove_tier",
-                    tier_name = %tier_name,
-                    result = "remove_blocked",
-                    error = ?err,
-                    "admin tier state"
-                );
                 Err(response)
             } else {
                 warn!(

rustfs/src/admin/storage_api.rs

@@ -232,6 +232,9 @@ pub(crate) mod lifecycle {
     pub(crate) type ManualTransitionRunOptions =
         super::ecstore_bucket::lifecycle::bucket_lifecycle_ops::ManualTransitionRunOptions;
     pub(crate) type ManualTransitionRunReport = super::ecstore_bucket::lifecycle::bucket_lifecycle_ops::ManualTransitionRunReport;
+    pub(crate) use super::ecstore_bucket::lifecycle::recovery_control::{
+        IlmRecoveryClassification, IlmRecoveryProtocol, inspect_recovery_control, list_recovery_controls,
+    };
     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,