fix(ecstore): track ILM recovery across decommission

2026-08-22 12:26:37 +00:00 · 2026-08-22 02:23:58 +08:00
parent 50c208f716
commit 812cf9662a
8 changed files with 973 additions and 183 deletions
@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use rustfs_utils::crypto::hex_sha256;
+use serde::{Deserialize, Serialize};
 use uuid::Uuid;

 use super::{manual_transition_job, tier_delete_journal, transition_transaction};
@@ -89,6 +91,7 @@ pub(crate) struct ValidatedDurableIlmRecord {
    pub(crate) namespace: &'static str,
    pub(crate) id_kind: &'static str,
    pub(crate) id: String,
+    pub(crate) checkpoint: DurableIlmRecordCheckpoint,
 }

 impl ValidatedDurableIlmRecord {
@@ -97,6 +100,180 @@ impl ValidatedDurableIlmRecord {
    }
 }

+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
+pub(crate) enum DurableIlmRecordCheckpoint {
+    TierDeleteJournal {
+        content_sha256: String,
+        identity_sha256: String,
+        committed: bool,
+    },
+    TransitionTransaction {
+        content_sha256: String,
+        identity_sha256: String,
+        remote_version_sha256: String,
+        remote_version_known: bool,
+        revision: u64,
+        state: transition_transaction::TransitionTransactionState,
+    },
+    ManualTransitionJob {
+        content_sha256: String,
+        identity_sha256: String,
+        updated_at_unix_nanos: i128,
+        state: manual_transition_job::ManualTransitionJobState,
+        scan_completed: bool,
+        cancel_requested: bool,
+    },
+    ManualTransitionScope {
+        content_sha256: String,
+        identity_sha256: String,
+        updated_at_unix_nanos: i128,
+    },
+    ManualTransitionTask {
+        content_sha256: String,
+    },
+    ManualTransitionWorkerResult {
+        content_sha256: String,
+    },
+}
+
+impl DurableIlmRecordCheckpoint {
+    pub(crate) fn content_sha256(&self) -> &str {
+        match self {
+            Self::TierDeleteJournal { content_sha256, .. }
+            | Self::TransitionTransaction { content_sha256, .. }
+            | Self::ManualTransitionJob { content_sha256, .. }
+            | Self::ManualTransitionScope { content_sha256, .. }
+            | Self::ManualTransitionTask { content_sha256 }
+            | Self::ManualTransitionWorkerResult { content_sha256 } => content_sha256,
+        }
+    }
+
+    pub(crate) fn validate_successor(&self, next: &Self) -> Result<()> {
+        if self == next {
+            return Ok(());
+        }
+
+        let valid = match (self, next) {
+            (
+                Self::TierDeleteJournal {
+                    identity_sha256: previous_identity,
+                    committed: previous_committed,
+                    ..
+                },
+                Self::TierDeleteJournal {
+                    identity_sha256: next_identity,
+                    committed: next_committed,
+                    ..
+                },
+            ) => {
+                previous_identity == next_identity
+                    && (previous_committed == next_committed || (!previous_committed && *next_committed))
+            }
+            (
+                Self::TransitionTransaction {
+                    identity_sha256: previous_identity,
+                    remote_version_sha256: previous_remote_version,
+                    remote_version_known: previous_remote_version_known,
+                    revision: previous_revision,
+                    state: previous_state,
+                    ..
+                },
+                Self::TransitionTransaction {
+                    identity_sha256: next_identity,
+                    remote_version_sha256: next_remote_version,
+                    revision: next_revision,
+                    state: next_state,
+                    ..
+                },
+            ) => {
+                previous_identity == next_identity
+                    && transition_state_distance(*previous_state, *next_state)
+                        .and_then(|distance| previous_revision.checked_add(distance))
+                        .is_some_and(|expected_revision| *next_revision == expected_revision)
+                    && (!previous_remote_version_known || previous_remote_version == next_remote_version)
+            }
+            (
+                Self::ManualTransitionJob {
+                    identity_sha256: previous_identity,
+                    updated_at_unix_nanos: previous_updated_at,
+                    state: previous_state,
+                    scan_completed: previous_scan_completed,
+                    cancel_requested: previous_cancel_requested,
+                    ..
+                },
+                Self::ManualTransitionJob {
+                    identity_sha256: next_identity,
+                    updated_at_unix_nanos: next_updated_at,
+                    state: next_state,
+                    scan_completed: next_scan_completed,
+                    cancel_requested: next_cancel_requested,
+                    ..
+                },
+            ) => {
+                previous_identity == next_identity
+                    && next_updated_at > previous_updated_at
+                    && manual_job_state_reaches(*previous_state, *next_state)
+                    && (!previous_scan_completed || *next_scan_completed)
+                    && (!previous_cancel_requested || *next_cancel_requested)
+            }
+            (
+                Self::ManualTransitionScope {
+                    identity_sha256: previous_identity,
+                    updated_at_unix_nanos: previous_updated_at,
+                    ..
+                },
+                Self::ManualTransitionScope {
+                    identity_sha256: next_identity,
+                    updated_at_unix_nanos: next_updated_at,
+                    ..
+                },
+            ) => previous_identity == next_identity && next_updated_at > previous_updated_at,
+            _ => false,
+        };
+
+        if valid {
+            Ok(())
+        } else {
+            Err(Error::other("durable ILM record generation is not a monotonic successor"))
+        }
+    }
+}
+
+fn transition_state_distance(
+    from: transition_transaction::TransitionTransactionState,
+    to: transition_transaction::TransitionTransactionState,
+) -> Option<u64> {
+    use transition_transaction::TransitionTransactionState::{
+        AbortedNoRemote, CleanupPending, Committed, LocalCommitStarted, UploadOutcomeUnknown, UploadStarted, Uploaded,
+    };
+
+    match (from, to) {
+        (UploadStarted, UploadOutcomeUnknown | AbortedNoRemote | Uploaded) => Some(1),
+        (UploadStarted, LocalCommitStarted | CleanupPending) => Some(2),
+        (UploadStarted, Committed) => Some(3),
+        (UploadOutcomeUnknown, Uploaded | CleanupPending) => Some(1),
+        (UploadOutcomeUnknown, LocalCommitStarted) => Some(2),
+        (UploadOutcomeUnknown, Committed) => Some(3),
+        (Uploaded, LocalCommitStarted | CleanupPending) => Some(1),
+        (Uploaded, Committed) => Some(2),
+        (LocalCommitStarted, Committed | CleanupPending) => Some(1),
+        _ => None,
+    }
+}
+
+fn manual_job_state_reaches(
+    from: manual_transition_job::ManualTransitionJobState,
+    to: manual_transition_job::ManualTransitionJobState,
+) -> bool {
+    from == to || from == manual_transition_job::ManualTransitionJobState::Running
+}
+
+fn checkpoint_hash<T: Serialize>(value: &T) -> Result<String> {
+    let encoded = serde_json::to_vec(value).map_err(Error::other)?;
+    Ok(hex_sha256(&encoded, ToOwned::to_owned))
+}
+
 fn path_is_in_namespace(path: &str, namespace: &DurableIlmNamespace) -> bool {
    let Some(suffix) = path.strip_prefix(namespace.prefix) else {
        return false;
@@ -163,7 +340,8 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
        )));
    }

-    let (id_kind, id) = match namespace.kind {
+    let content_sha256 = hex_sha256(data, ToOwned::to_owned);
+    let (id_kind, id, checkpoint) = match namespace.kind {
        DurableIlmRecordKind::TierDeleteJournal => {
            let entry = tier_delete_journal::decode_tier_delete_journal_entry(data)?;
            if tier_delete_journal::tier_delete_journal_object_name(&entry) != path {
@@ -173,12 +351,52 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
                .strip_prefix(namespace.prefix)
                .and_then(|suffix| suffix.strip_suffix(".json"))
                .ok_or_else(|| Error::other("tier delete journal path is invalid"))?;
-            ("operation_id", operation_id.to_string())
+            let identity_sha256 = checkpoint_hash(&(
+                &entry.obj_name,
+                &entry.version_id,
+                &entry.tier_name,
+                entry.backend_identity,
+                entry.version_id_exact,
+                entry.version_state,
+                &entry.source,
+            ))?;
+            (
+                "operation_id",
+                operation_id.to_string(),
+                DurableIlmRecordCheckpoint::TierDeleteJournal {
+                    content_sha256,
+                    identity_sha256,
+                    committed: entry.state == super::tier_sweeper::TierDeleteJournalState::Committed,
+                },
+            )
        }
        DurableIlmRecordKind::TransitionTransaction => {
            let transaction = transition_transaction::decode_transition_transaction_record(path, data)
                .map_err(|err| Error::other(err.to_string()))?;
-            ("transaction_id", transaction.transaction_id.to_string())
+            let identity_sha256 = checkpoint_hash(&(
+                transaction.deployment_id,
+                transaction.transaction_id,
+                transaction.owner_epoch,
+                transaction.write_id,
+                &transaction.source,
+                &transaction.tier_name,
+                transaction.backend_fingerprint,
+                &transaction.remote_object,
+                transaction.not_after_unix_nanos,
+            ))?;
+            let remote_version_sha256 = checkpoint_hash(&transaction.remote_version)?;
+            (
+                "transaction_id",
+                transaction.transaction_id.to_string(),
+                DurableIlmRecordCheckpoint::TransitionTransaction {
+                    content_sha256,
+                    identity_sha256,
+                    remote_version_sha256,
+                    remote_version_known: !transaction.remote_version.is_unknown(),
+                    revision: transaction.revision,
+                    state: transaction.state,
+                },
+            )
        }
        DurableIlmRecordKind::ManualTransitionJob => {
            let job_id = manual_transition_job::manual_transition_job_id_from_record_object_name(path)
@@ -188,9 +406,31 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
            if canonical != path {
                return Err(Error::other("manual transition job path is not canonical"));
            }
-            manual_transition_job::ManualTransitionJobRecord::decode(job_id, data)
+            let job = manual_transition_job::ManualTransitionJobRecord::decode(job_id, data)
                .map_err(|err| Error::other(err.to_string()))?;
-            ("job_id", job_id.to_string())
+            let identity_sha256 = checkpoint_hash(&(
+                job.job_id,
+                &job.scope_key,
+                &job.bucket,
+                &job.prefix,
+                &job.tier,
+                job.dry_run,
+                job.max_objects,
+                job.max_duration,
+                job.created_at_unix_nanos,
+            ))?;
+            (
+                "job_id",
+                job_id.to_string(),
+                DurableIlmRecordCheckpoint::ManualTransitionJob {
+                    content_sha256,
+                    identity_sha256,
+                    updated_at_unix_nanos: job.updated_at_unix_nanos,
+                    state: job.state,
+                    scan_completed: job.scan_completed,
+                    cancel_requested: job.cancel_requested,
+                },
+            )
        }
        DurableIlmRecordKind::ManualTransitionScope => {
            let admission: manual_transition_job::ManualTransitionScopeAdmission =
@@ -201,7 +441,24 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
            if canonical != path {
                return Err(Error::other("manual transition scope content does not match its path"));
            }
-            ("job_id", admission.job_id.to_string())
+            let identity_sha256 = checkpoint_hash(&(
+                &admission.schema,
+                &admission.scope_key,
+                admission.job_id,
+                &admission.bucket,
+                &admission.prefix,
+                &admission.tier,
+                admission.dry_run,
+            ))?;
+            (
+                "job_id",
+                admission.job_id.to_string(),
+                DurableIlmRecordCheckpoint::ManualTransitionScope {
+                    content_sha256,
+                    identity_sha256,
+                    updated_at_unix_nanos: admission.updated_at_unix_nanos,
+                },
+            )
        }
        DurableIlmRecordKind::ManualTransitionTask => {
            let (job_id, task_key) = parse_manual_sharded_record(path, namespace.prefix)?;
@@ -212,7 +469,11 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
            }
            manual_transition_job::ManualTransitionTaskRecord::decode(job_id, &task_key, data)
                .map_err(|err| Error::other(err.to_string()))?;
-            ("job_id", job_id.to_string())
+            (
+                "job_id",
+                job_id.to_string(),
+                DurableIlmRecordCheckpoint::ManualTransitionTask { content_sha256 },
+            )
        }
        DurableIlmRecordKind::ManualTransitionWorkerResult => {
            let (job_id, task_key) = parse_manual_sharded_record(path, namespace.prefix)?;
@@ -223,7 +484,11 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
            }
            manual_transition_job::ManualTransitionWorkerResultRecord::decode(job_id, &task_key, data)
                .map_err(|err| Error::other(err.to_string()))?;
-            ("job_id", job_id.to_string())
+            (
+                "job_id",
+                job_id.to_string(),
+                DurableIlmRecordCheckpoint::ManualTransitionWorkerResult { content_sha256 },
+            )
        }
    };

@@ -231,6 +496,7 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
        namespace: namespace.name,
        id_kind,
        id,
+        checkpoint,
    })
 }

@@ -320,7 +320,7 @@ impl ManualTransitionJobRecord {
            }
        }
        self.queue_snapshot = queue_snapshot;
-        self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
+        self.advance_updated_at();
        self.mark_terminal_if_worker_drained();
    }

@@ -363,14 +363,14 @@ impl ManualTransitionJobRecord {
        self.report.tier_failure = scan_tier_failure.saturating_add(transition_failed);
        self.report.tier_failure_by_reason = scan_tier_failure_by_reason;
        self.queue_snapshot = queue_snapshot;
-        self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
+        self.advance_updated_at();
        self.mark_terminal_if_worker_drained();
        true
    }

    pub fn mark_cancel_requested(&mut self) {
        self.cancel_requested = true;
-        self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
+        self.advance_updated_at();
    }

    pub fn claim_recovery_lease(&mut self, owner_id: impl Into<String>, queue_snapshot: ManualTransitionQueueSnapshot) {
@@ -384,7 +384,7 @@ impl ManualTransitionJobRecord {
    pub fn abandon_recovery_lease(&mut self, lease_id: Uuid) {
        if self.state == ManualTransitionJobState::Running && self.lease_id == lease_id {
            self.lease_expires_at_unix_nanos = 0;
-            self.updated_at_unix_nanos = OffsetDateTime::now_utc().unix_timestamp_nanos();
+            self.advance_updated_at();
        }
    }

@@ -402,7 +402,7 @@ impl ManualTransitionJobRecord {

    pub fn renew_lease(&mut self, queue_snapshot: ManualTransitionQueueSnapshot) {
        let now = OffsetDateTime::now_utc().unix_timestamp_nanos();
-        self.updated_at_unix_nanos = now;
+        self.updated_at_unix_nanos = self.updated_at_unix_nanos.saturating_add(1).max(now);
        self.lease_expires_at_unix_nanos = manual_transition_job_lease_expires_at(now);
        self.queue_snapshot = queue_snapshot;
    }
@@ -467,9 +467,13 @@ impl ManualTransitionJobRecord {
    }

    fn mark_updated_terminal(&mut self) {
+        self.advance_updated_at();
+        self.completed_at_unix_nanos = Some(self.updated_at_unix_nanos);
+    }
+
+    fn advance_updated_at(&mut self) {
        let now = OffsetDateTime::now_utc().unix_timestamp_nanos();
-        self.updated_at_unix_nanos = now;
-        self.completed_at_unix_nanos = Some(now);
+        self.updated_at_unix_nanos = self.updated_at_unix_nanos.saturating_add(1).max(now);
    }

    fn mark_terminal_if_worker_drained(&mut self) {
@@ -1113,7 +1117,8 @@ pub fn manual_transition_scope_record_object_name(scope_key: &str) -> Result<Str
 pub async fn save_manual_transition_job_record(api: Arc<ECStore>, job: &ManualTransitionJobRecord) -> EcstoreResult<()> {
    let object = manual_transition_job_record_object_name(job.job_id).map_err(manual_transition_job_store_error)?;
    let data = job.encode().map_err(manual_transition_job_store_error)?;
-    config_boundary::save_config(api, &object, data).await
+    config_boundary::save_config(api.clone(), &object, data.clone()).await?;
+    api.record_durable_ilm_decommission_progress(&object, &data).await
 }

 pub async fn load_manual_transition_job_record(api: Arc<ECStore>, job_id: Uuid) -> EcstoreResult<ManualTransitionJobRecord> {
@@ -1146,9 +1151,9 @@ pub async fn save_manual_transition_job_record_if_current(
    let object = manual_transition_job_record_object_name(job.job_id).map_err(manual_transition_job_store_error)?;
    let data = job.encode().map_err(manual_transition_job_store_error)?;
    config_boundary::save_config_with_opts_quiet(
-        api,
+        api.clone(),
        &object,
-        data,
+        data.clone(),
        &ObjectOptions {
            max_parity: true,
            http_preconditions: Some(HTTPPreconditions {
@@ -1158,7 +1163,8 @@ pub async fn save_manual_transition_job_record_if_current(
            ..Default::default()
        },
    )
-    .await
+    .await?;
+    api.record_durable_ilm_decommission_progress(&object, &data).await
 }

 /// Applies a job-record mutation with optimistic concurrency control.
@@ -1596,9 +1602,9 @@ pub async fn save_manual_transition_scope_admission_if_absent(
    let object = manual_transition_scope_record_object_name(&admission.scope_key).map_err(manual_transition_job_store_error)?;
    let data = serde_json::to_vec(admission).map_err(Error::other)?;
    config_boundary::save_config_with_opts(
-        api,
+        api.clone(),
        &object,
-        data,
+        data.clone(),
        &ObjectOptions {
            max_parity: true,
            http_preconditions: Some(HTTPPreconditions {
@@ -1608,7 +1614,8 @@ pub async fn save_manual_transition_scope_admission_if_absent(
            ..Default::default()
        },
    )
-    .await
+    .await?;
+    api.record_durable_ilm_decommission_progress(&object, &data).await
 }

 pub async fn load_manual_transition_scope_admission(
@@ -1646,9 +1653,9 @@ pub async fn save_manual_transition_scope_admission_if_current(
    let object = manual_transition_scope_record_object_name(&admission.scope_key).map_err(manual_transition_job_store_error)?;
    let data = serde_json::to_vec(admission).map_err(Error::other)?;
    match config_boundary::save_config_with_opts(
-        api,
+        api.clone(),
        &object,
-        data,
+        data.clone(),
        &ObjectOptions {
            max_parity: true,
            http_preconditions: Some(HTTPPreconditions {
@@ -1664,7 +1671,8 @@ pub async fn save_manual_transition_scope_admission_if_current(
            Err(Error::PreconditionFailed)
        }
        result => result,
-    }
+    }?;
+    api.record_durable_ilm_decommission_progress(&object, &data).await
 }

 pub async fn claim_manual_transition_scope_admission(
@@ -1957,13 +1965,15 @@ pub async fn delete_manual_transition_scope_admission_if_current(
    job_id: Uuid,
    lease_id: Uuid,
 ) -> EcstoreResult<bool> {
-    let etag = match load_manual_transition_scope_admission_with_etag(api.clone(), scope_key).await {
-        Ok((admission, etag)) if admission.job_id == job_id && admission.lease_id == lease_id => etag,
+    let (admission, etag) = match load_manual_transition_scope_admission_with_etag(api.clone(), scope_key).await {
+        Ok((admission, etag)) if admission.job_id == job_id && admission.lease_id == lease_id => (admission, etag),
        Ok(_) => return Ok(false),
        Err(Error::ConfigNotFound) => return Ok(true),
        Err(err) => return Err(err),
    };
    let object = manual_transition_scope_record_object_name(scope_key).map_err(manual_transition_job_store_error)?;
+    let data = serde_json::to_vec(&admission).map_err(Error::other)?;
+    api.record_durable_ilm_decommission_terminal(&object, &data).await?;
    match config_boundary::delete_config_if_match(api, &object, &etag).await {
        Ok(()) | Err(Error::ConfigNotFound) => Ok(true),
        Err(Error::PreconditionFailed) => Ok(false),
@@ -433,9 +433,18 @@ async fn process_committed_tier_delete_journal_entry(api: Arc<ECStore>, je: &Jen
        )
        .await?;
    }
+    record_tier_delete_journal_decommission_terminal(&api, je).await?;
    remove_tier_delete_journal_entry(api, je).await
 }

+async fn record_tier_delete_journal_decommission_terminal(api: &Arc<ECStore>, je: &Jentry) -> std::io::Result<()> {
+    let path = tier_delete_journal_object_name(je);
+    let data = encode_tier_delete_journal_entry(je).map_err(std::io::Error::other)?;
+    api.record_durable_ilm_decommission_terminal(&path, &data)
+        .await
+        .map_err(std::io::Error::other)
+}
+
 fn object_info_references_tier_delete(info: &ObjectInfo, je: &Jentry) -> std::io::Result<bool> {
    if info.transitioned_object.status != rustfs_filemeta::TRANSITION_COMPLETE
        || info.transitioned_object.name != je.obj_name
@@ -585,7 +585,8 @@ pub(crate) async fn save_transition_transaction_record(
    let object =
        transition_transaction_record_object_name(transaction.transaction_id).map_err(transition_transaction_store_error)?;
    let data = transaction.encode().map_err(transition_transaction_store_error)?;
-    config_boundary::save_config(api, &object, data).await
+    config_boundary::save_config(api.clone(), &object, data.clone()).await?;
+    api.record_durable_ilm_decommission_progress(&object, &data).await
 }

 pub(crate) async fn load_transition_transaction_record(
@@ -597,8 +598,14 @@ pub(crate) async fn load_transition_transaction_record(
    TransitionTransaction::decode(transaction_id, &data).map_err(transition_transaction_store_error)
 }

-pub(crate) async fn delete_transition_transaction_record(api: Arc<ECStore>, transaction_id: Uuid) -> EcstoreResult<()> {
-    let object = transition_transaction_record_object_name(transaction_id).map_err(transition_transaction_store_error)?;
+pub(crate) async fn delete_transition_transaction_record(
+    api: Arc<ECStore>,
+    transaction: &TransitionTransaction,
+) -> EcstoreResult<()> {
+    let object =
+        transition_transaction_record_object_name(transaction.transaction_id).map_err(transition_transaction_store_error)?;
+    let data = transaction.encode().map_err(transition_transaction_store_error)?;
+    api.record_durable_ilm_decommission_terminal(&object, &data).await?;
    match config_boundary::delete_config(api, &object).await {
        Ok(()) | Err(Error::ConfigNotFound) => Ok(()),
        Err(err) => Err(err),
@@ -814,7 +821,7 @@ pub async fn finalize_missing_transition_transaction_for_operator(
    if probe != TransitionOperatorProbe::Missing {
        return Err(TransitionOperatorError::CandidateNotMissing(probe));
    }
-    delete_transition_transaction_record(api, transaction_id)
+    delete_transition_transaction_record(api, &transaction)
        .await
        .map_err(TransitionOperatorError::Store)
 }
@@ -850,22 +857,22 @@ pub async fn process_transition_transaction_record(
    match transaction.state {
        TransitionTransactionState::Uploaded => {
            delete_transition_remote_candidate(api.clone(), transaction).await?;
-            delete_transition_transaction_record(api, transaction.transaction_id).await?;
+            delete_transition_transaction_record(api, transaction).await?;
            Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
        }
        TransitionTransactionState::CleanupPending => match local_commit_matches_transaction(api.clone(), transaction).await {
            Ok(true) => {
-                delete_transition_transaction_record(api, transaction.transaction_id).await?;
+                delete_transition_transaction_record(api, transaction).await?;
                Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
            }
            Ok(false) => {
                delete_transition_remote_candidate(api.clone(), transaction).await?;
-                delete_transition_transaction_record(api, transaction.transaction_id).await?;
+                delete_transition_transaction_record(api, transaction).await?;
                Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
            }
            Err(err) if transition_source_is_missing(&err) => {
                delete_transition_remote_candidate(api.clone(), transaction).await?;
-                delete_transition_transaction_record(api, transaction.transaction_id).await?;
+                delete_transition_transaction_record(api, transaction).await?;
                Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
            }
            Err(err) => Err(err),
@@ -873,7 +880,7 @@ pub async fn process_transition_transaction_record(
        TransitionTransactionState::LocalCommitStarted => {
            match local_commit_matches_transaction(api.clone(), transaction).await {
                Ok(true) => {
-                    delete_transition_transaction_record(api, transaction.transaction_id).await?;
+                    delete_transition_transaction_record(api, transaction).await?;
                    Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
                }
                Ok(false) => Ok(TransitionTransactionRecoveryOutcome::Retained),
@@ -882,7 +889,7 @@ pub async fn process_transition_transaction_record(
            }
        }
        TransitionTransactionState::AbortedNoRemote | TransitionTransactionState::Committed => {
-            delete_transition_transaction_record(api, transaction.transaction_id).await?;
+            delete_transition_transaction_record(api, transaction).await?;
            Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
        }
        TransitionTransactionState::UploadOutcomeUnknown => recover_unknown_upload_outcome(api, transaction).await,
@@ -908,7 +915,7 @@ async fn recover_unknown_upload_outcome(
        .map_err(Error::other)?
    {
        TransitionCandidateProbe::Missing => {
-            delete_transition_transaction_record(api, transaction.transaction_id).await?;
+            delete_transition_transaction_record(api, transaction).await?;
            Ok(TransitionTransactionRecoveryOutcome::RecordDeleted)
        }
        TransitionCandidateProbe::UnversionedPresent => {
@@ -926,7 +933,7 @@ async fn recover_unknown_upload_outcome(
            )
            .await
            .map_err(Error::other)?;
-            delete_transition_transaction_record(api, transaction.transaction_id).await?;
+            delete_transition_transaction_record(api, transaction).await?;
            Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
        }
        TransitionCandidateProbe::VersionedPresent(version_id) => {
@@ -959,7 +966,7 @@ async fn cleanup_recovered_unknown_upload_candidate(
        .map_err(transition_transaction_store_error)?;
    save_transition_transaction_record(api.clone(), &cleanup).await?;
    delete_transition_remote_candidate(api.clone(), &cleanup).await?;
-    delete_transition_transaction_record(api, cleanup.transaction_id).await?;
+    delete_transition_transaction_record(api, &cleanup).await?;
    Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
 }

@@ -410,10 +410,21 @@ pub(crate) async fn read_config_limited_preserve_empty<S>(api: Arc<S>, file: &st
 where
    S: EcstoreObjectIO,
 {
-    let (data, _obj) = read_config_with_metadata_inner(api, file, &ObjectOptions::default(), true, Some(max_bytes)).await?;
+    let (data, _obj) = read_config_limited_preserve_empty_with_metadata(api, file, max_bytes).await?;
    Ok(data)
 }

+pub(crate) async fn read_config_limited_preserve_empty_with_metadata<S>(
+    api: Arc<S>,
+    file: &str,
+    max_bytes: usize,
+) -> Result<(Vec<u8>, ObjectInfo)>
+where
+    S: EcstoreObjectIO,
+{
+    read_config_with_metadata_inner(api, file, &ObjectOptions::default(), true, Some(max_bytes)).await
+}
+
 /// Read an existing config object without treating an empty payload as absent.
 /// Callers that validate their own payload format need to distinguish corruption
 /// from `ConfigNotFound`.
@@ -16,7 +16,7 @@ use crate::bucket::replication::replication_state_from_filemeta;
 use crate::bucket::versioning_sys::BucketVersioningSys;
 use crate::bucket::{
    lifecycle::{
-        ILM_META_PREFIX, LifecycleExpiryConfigs, ValidatedDurableIlmRecord,
+        DurableIlmRecordCheckpoint, ILM_META_PREFIX, LifecycleExpiryConfigs, ValidatedDurableIlmRecord,
        bucket_lifecycle_audit::LcEventSrc,
        bucket_lifecycle_ops::{
            LifecycleOps, apply_expiry_on_transitioned_object, apply_expiry_rule_in, eval_action_from_lifecycle,
@@ -30,8 +30,8 @@ use crate::bucket::{
 };
 use crate::cache_value::metacache_set::{ListPathRawOptions, list_path_raw};
 use crate::config::com::{
-    CONFIG_PREFIX, delete_config, read_config, read_config_limited_preserve_empty, read_config_no_lock, save_config,
-    save_config_with_opts,
+    CONFIG_PREFIX, delete_config, read_config, read_config_limited_preserve_empty,
+    read_config_limited_preserve_empty_with_metadata, read_config_no_lock, save_config, save_config_with_opts,
 };
 use crate::data_movement;
 use crate::data_movement::backpressure::{self, DataMovementOperation};
@@ -53,7 +53,7 @@ use crate::storage_api_contracts::{
    heal::HealOperations as _,
    list::ListOperations as _,
    namespace::NamespaceLocking as _,
-    object::{EcstoreObjectIO, ObjectIO as _, ObjectOperations as _},
+    object::{EcstoreObjectIO, HTTPPreconditions, ObjectIO as _, ObjectOperations as _},
 };
 use crate::{core::sets::Sets, store::ECStore};
 use byteorder::{ByteOrder, LittleEndian, WriteBytesExt};
@@ -102,8 +102,9 @@ const DECOMMISSION_TARGET_CAPACITY_OVERHEAD_PERCENT: usize = 30;
 const DECOMMISSION_LISTING_MAX_ATTEMPTS: usize = 3;
 const DECOMMISSION_LISTING_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(5);
 const DECOMMISSION_DURABLE_ILM_RECEIPT_ROOT: &str = "decommission/ilm-receipts";
-const DECOMMISSION_DURABLE_ILM_RECEIPT_SCHEMA: &str = "v1";
+const DECOMMISSION_DURABLE_ILM_RECEIPT_SCHEMA: &str = "v2";
 const DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE: usize = 16 * 1024;
+const DECOMMISSION_DURABLE_ILM_RECEIPT_CAS_ATTEMPTS: usize = 3;
 /// Background decommission walks must tolerate slow object migrations; the
 /// stall timeout is the drive-health bound, not the total listing duration.
 const DECOMMISSION_BACKGROUND_WALKDIR_STALL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
@@ -802,22 +803,24 @@ fn resolve_decommission_check_after_list_result(list_result: Result<()>, entry_e
 fn validate_decommission_durable_ilm_copy(
    path: &str,
    source_record: &ValidatedDurableIlmRecord,
-    source: &[u8],
    target: &[u8],
-) -> Result<()> {
-    validate_durable_ilm_record(path, target).map_err(|err| {
+) -> Result<ValidatedDurableIlmRecord> {
+    let target_record = validate_durable_ilm_record(path, target).map_err(|err| {
        Error::other(format!(
            "target durable ILM record is invalid at path `{path}` {}: {err}",
            source_record.context()
        ))
    })?;
-    if source != target {
-        return Err(Error::other(format!(
-            "target durable ILM record content mismatch at path `{path}` {}",
-            source_record.context()
-        )));
-    }
-    Ok(())
+    source_record
+        .checkpoint
+        .validate_successor(&target_record.checkpoint)
+        .map_err(|err| {
+            Error::other(format!(
+                "target durable ILM record generation mismatch at path `{path}` {}: {err}",
+                source_record.context()
+            ))
+        })?;
+    Ok(target_record)
 }

 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -827,17 +830,19 @@ struct DecommissionDurableIlmReceipt {
    namespace: String,
    id_kind: String,
    id: String,
-    target_content_sha256: String,
+    checkpoint: DurableIlmRecordCheckpoint,
+    terminal_checkpoint: Option<DurableIlmRecordCheckpoint>,
 }

 impl DecommissionDurableIlmReceipt {
-    fn new(path: &str, record: &ValidatedDurableIlmRecord, target: &[u8]) -> Self {
+    fn new(path: &str, record: &ValidatedDurableIlmRecord) -> Self {
        Self {
            source_path: path.to_string(),
            namespace: record.namespace.to_string(),
            id_kind: record.id_kind.to_string(),
            id: record.id.clone(),
-            target_content_sha256: hex_sha256(target, ToOwned::to_owned),
+            checkpoint: record.checkpoint.clone(),
+            terminal_checkpoint: None,
        }
    }

@@ -860,13 +865,22 @@ impl DecommissionDurableIlmReceipt {
                self.source_path
            )));
        }
-        if !is_sha256_checksum(&self.target_content_sha256) {
+        if !is_sha256_checksum(self.checkpoint.content_sha256()) {
            return Err(Error::other(format!(
                "receipt target checksum is invalid for source path `{}` {}",
                self.source_path,
                self.context()
            )));
        }
+        if let Some(terminal_checkpoint) = &self.terminal_checkpoint {
+            self.checkpoint.validate_successor(terminal_checkpoint).map_err(|err| {
+                Error::other(format!(
+                    "receipt terminal checkpoint is invalid for source path `{}` {}: {err}",
+                    self.source_path,
+                    self.context()
+                ))
+            })?;
+        }
        Ok(())
    }

@@ -918,14 +932,80 @@ struct PersistedDecommissionDurableIlmReceipt {
    receipt: DecommissionDurableIlmReceipt,
 }

-fn decommission_durable_ilm_receipt_prefix(cmd_line: &str) -> String {
-    let pool_key = hex_sha256(cmd_line.as_bytes(), ToOwned::to_owned);
-    format!("{DECOMMISSION_DURABLE_ILM_RECEIPT_ROOT}/{pool_key}/")
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct DecommissionDurableIlmReceiptLocator {
+    run_token: String,
+    source_path: String,
+    id_kind: String,
+    id: String,
 }

-fn decommission_durable_ilm_receipt_path(prefix: &str, source_path: &str) -> String {
-    let source_key = hex_sha256(source_path.as_bytes(), ToOwned::to_owned);
-    format!("{prefix}{source_key}.json")
+impl DecommissionDurableIlmReceiptLocator {
+    fn context(&self) -> String {
+        format!("source path `{}` {} `{}`", self.source_path, self.id_kind, self.id)
+    }
+}
+
+fn decommission_durable_ilm_receipt_run_token(cmd_line: &str) -> String {
+    hex_sha256(cmd_line.as_bytes(), ToOwned::to_owned)
+}
+
+fn decommission_durable_ilm_receipt_run_prefix(run_token: &str) -> String {
+    format!("{DECOMMISSION_DURABLE_ILM_RECEIPT_ROOT}/{run_token}/")
+}
+
+fn decommission_durable_ilm_receipt_path(run_token: &str, source_path: &str, id_kind: &str, id: &str) -> String {
+    format!(
+        "{}{}/{}/{}.json",
+        decommission_durable_ilm_receipt_run_prefix(run_token),
+        source_path,
+        id_kind,
+        id
+    )
+}
+
+fn parse_decommission_durable_ilm_receipt_path(path: &str) -> Result<DecommissionDurableIlmReceiptLocator> {
+    let prefix = format!("{DECOMMISSION_DURABLE_ILM_RECEIPT_ROOT}/");
+    let suffix = path
+        .strip_prefix(&prefix)
+        .ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` has the wrong root")))?;
+    let (run_token, record_path) = suffix
+        .split_once('/')
+        .ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` is missing its record path")))?;
+    let mut parts = record_path.rsplitn(3, '/');
+    let id = parts
+        .next()
+        .and_then(|file| file.strip_suffix(".json"))
+        .filter(|id| !id.is_empty())
+        .ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` is missing its record id")))?;
+    let id_kind = parts
+        .next()
+        .filter(|id_kind| matches!(*id_kind, "operation_id" | "transaction_id" | "job_id"))
+        .ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` has an invalid id kind")))?;
+    let source_path = parts
+        .next()
+        .filter(|source_path| !source_path.is_empty())
+        .ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` is missing its source path")))?;
+    if !is_sha256_checksum(run_token) {
+        return Err(Error::other(format!("durable ILM receipt path `{path}` has an invalid run token")));
+    }
+    match id_kind {
+        "operation_id" if !is_sha256_checksum(id) => {
+            return Err(Error::other(format!("durable ILM receipt path `{path}` has an invalid operation id")));
+        }
+        "transaction_id" | "job_id" if uuid::Uuid::parse_str(id).is_err() => {
+            return Err(Error::other(format!("durable ILM receipt path `{path}` has an invalid UUID")));
+        }
+        _ => {}
+    }
+    classify_durable_ilm_record(source_path)?
+        .ok_or_else(|| Error::other(format!("durable ILM receipt path `{path}` does not identify a durable ILM source path")))?;
+    Ok(DecommissionDurableIlmReceiptLocator {
+        run_token: run_token.to_string(),
+        source_path: source_path.to_string(),
+        id_kind: id_kind.to_string(),
+        id: id.to_string(),
+    })
 }

 fn resolve_decommission_pool_meta_reload_result(result: Result<()>, stage: &str) -> Result<()> {
@@ -4393,7 +4473,7 @@ impl ECStore {
        Ok(ret)
    }

-    async fn durable_ilm_receipt_prefix(&self, source_pool_idx: usize) -> Result<String> {
+    async fn durable_ilm_receipt_run_token(&self, source_pool_idx: usize) -> Result<String> {
        let pool_meta = self.pool_meta.read().await;
        let cmd_line = pool_meta
            .pools
@@ -4401,7 +4481,7 @@ impl ECStore {
            .ok_or_else(|| invalid_decommission_pool_index_error(pool_meta.pools.len(), source_pool_idx))?
            .cmd_line
            .clone();
-        Ok(decommission_durable_ilm_receipt_prefix(&cmd_line))
+        Ok(decommission_durable_ilm_receipt_run_token(&cmd_line))
    }

    async fn load_decommissioned_durable_ilm_target(
@@ -4410,7 +4490,7 @@ impl ECStore {
        path: &str,
        max_record_size: usize,
        record_context: &str,
-    ) -> Result<(usize, Vec<u8>)> {
+    ) -> Result<Option<(usize, Vec<u8>)>> {
        let mut target = None::<(usize, Vec<u8>)>;
        let mut first_read_error = None;
        for (target_pool_idx, pool) in self.pools.iter().enumerate() {
@@ -4442,40 +4522,62 @@ impl ECStore {
            }
        }

-        target.ok_or_else(|| {
-            first_read_error.unwrap_or_else(|| {
-                Error::other(format!("target durable ILM record is missing at path `{path}` {record_context}"))
-            })
-        })
+        match (target, first_read_error) {
+            (Some(target), _) => Ok(Some(target)),
+            (None, Some(err)) => Err(err),
+            (None, None) => Ok(None),
+        }
+    }
+
+    async fn list_decommission_durable_ilm_receipt_paths_in_pool(&self, pool_idx: usize, prefix: &str) -> Result<Vec<String>> {
+        let pool = self
+            .pools
+            .get(pool_idx)
+            .ok_or_else(|| invalid_decommission_pool_index_error(self.pools.len(), pool_idx))?;
+        let mut receipts = Vec::new();
+        let mut continuation = None;
+        loop {
+            let page = pool
+                .clone()
+                .list_objects_v2(RUSTFS_META_BUCKET, prefix, continuation, None, 1000, false, None, false)
+                .await
+                .map_err(|err| {
+                    Error::other(format!(
+                        "failed to list durable ILM decommission receipts under `{prefix}` in pool {pool_idx}: {err}"
+                    ))
+                })?;
+            receipts.extend(page.objects.into_iter().map(|object| object.name));
+            if !page.is_truncated {
+                break;
+            }
+            continuation = Some(page.next_continuation_token.ok_or_else(|| {
+                Error::other(format!(
+                    "durable ILM decommission receipt listing under `{prefix}` in pool {pool_idx} was truncated without a continuation token"
+                ))
+            })?);
+        }
+        Ok(receipts)
    }

    async fn list_decommission_durable_ilm_receipts(&self, source_pool_idx: usize) -> Result<Vec<(usize, String)>> {
-        let prefix = self.durable_ilm_receipt_prefix(source_pool_idx).await?;
+        let run_token = self.durable_ilm_receipt_run_token(source_pool_idx).await?;
+        let prefix = decommission_durable_ilm_receipt_run_prefix(&run_token);
        let mut receipts = Vec::new();
-        for (pool_idx, pool) in self.pools.iter().enumerate() {
+        for pool_idx in 0..self.pools.len() {
            if pool_idx == source_pool_idx {
                continue;
            }
-            let mut continuation = None;
-            loop {
-                let page = pool
-                    .clone()
-                    .list_objects_v2(RUSTFS_META_BUCKET, &prefix, continuation, None, 1000, false, None, false)
-                    .await
-                    .map_err(|err| {
-                        Error::other(format!(
-                            "failed to list durable ILM decommission receipts under `{prefix}` in pool {pool_idx}: {err}"
-                        ))
-                    })?;
-                receipts.extend(page.objects.into_iter().map(|object| (pool_idx, object.name)));
-                if !page.is_truncated {
-                    break;
+            for receipt_path in self
+                .list_decommission_durable_ilm_receipt_paths_in_pool(pool_idx, &prefix)
+                .await?
+            {
+                let locator = parse_decommission_durable_ilm_receipt_path(&receipt_path)?;
+                if locator.run_token != run_token {
+                    return Err(Error::other(format!(
+                        "durable ILM receipt path `{receipt_path}` has an unexpected run token"
+                    )));
                }
-                continuation = Some(page.next_continuation_token.ok_or_else(|| {
-                    Error::other(format!(
-                        "durable ILM decommission receipt listing under `{prefix}` in pool {pool_idx} was truncated without a continuation token"
-                    ))
-                })?);
+                receipts.push((pool_idx, receipt_path));
            }
        }
        Ok(receipts)
@@ -4487,8 +4589,8 @@ impl ECStore {
        target_pool_idx: usize,
        receipt: &DecommissionDurableIlmReceipt,
    ) -> Result<()> {
-        let prefix = self.durable_ilm_receipt_prefix(source_pool_idx).await?;
-        let receipt_path = decommission_durable_ilm_receipt_path(&prefix, &receipt.source_path);
+        let run_token = self.durable_ilm_receipt_run_token(source_pool_idx).await?;
+        let receipt_path = decommission_durable_ilm_receipt_path(&run_token, &receipt.source_path, &receipt.id_kind, &receipt.id);
        let encoded = receipt.encode().map_err(|err| {
            Error::other(format!(
                "failed to encode durable ILM decommission receipt `{receipt_path}` for source path `{}` {}: {err}",
@@ -4507,36 +4609,57 @@ impl ECStore {
            })
    }

+    fn validate_decommission_durable_ilm_receipt_locator(
+        receipt_path: &str,
+        locator: &DecommissionDurableIlmReceiptLocator,
+        receipt: &DecommissionDurableIlmReceipt,
+    ) -> Result<()> {
+        if locator.source_path != receipt.source_path || locator.id_kind != receipt.id_kind || locator.id != receipt.id {
+            return Err(Error::other(format!(
+                "durable ILM decommission receipt path `{receipt_path}` identity {} does not match receipt {}",
+                locator.context(),
+                receipt.context()
+            )));
+        }
+        Ok(())
+    }
+
+    async fn read_decommission_durable_ilm_receipt(
+        &self,
+        receipt_pool_idx: usize,
+        receipt_path: &str,
+    ) -> Result<DecommissionDurableIlmReceipt> {
+        let locator = parse_decommission_durable_ilm_receipt_path(receipt_path)?;
+        let data = read_config_limited_preserve_empty(
+            self.pools[receipt_pool_idx].clone(),
+            receipt_path,
+            DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE,
+        )
+        .await
+        .map_err(|err| {
+            Error::other(format!(
+                "failed to read durable ILM decommission receipt `{receipt_path}` from pool {receipt_pool_idx} for {}: {err}",
+                locator.context()
+            ))
+        })?;
+        let receipt = DecommissionDurableIlmReceipt::decode(&data).map_err(|err| {
+            Error::other(format!(
+                "durable ILM decommission receipt `{receipt_path}` in pool {receipt_pool_idx} for {} is invalid: {err}",
+                locator.context()
+            ))
+        })?;
+        Self::validate_decommission_durable_ilm_receipt_locator(receipt_path, &locator, &receipt)?;
+        Ok(receipt)
+    }
+
    async fn verify_decommission_durable_ilm_receipts(&self, source_pool_idx: usize) -> Result<()> {
-        let prefix = self.durable_ilm_receipt_prefix(source_pool_idx).await?;
        for (receipt_pool_idx, receipt_path) in self.list_decommission_durable_ilm_receipts(source_pool_idx).await? {
-            let data = read_config_limited_preserve_empty(
-                self.pools[receipt_pool_idx].clone(),
-                &receipt_path,
-                DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE,
-            )
-            .await
-            .map_err(|err| {
-                Error::other(format!(
-                    "failed to read durable ILM decommission receipt `{receipt_path}` from pool {receipt_pool_idx}: {err}"
-                ))
-            })?;
-            let receipt = DecommissionDurableIlmReceipt::decode(&data).map_err(|err| {
-                Error::other(format!(
-                    "durable ILM decommission receipt `{receipt_path}` in pool {receipt_pool_idx} is invalid: {err}"
-                ))
-            })?;
-            let expected_receipt_path = decommission_durable_ilm_receipt_path(&prefix, &receipt.source_path);
-            if receipt_path != expected_receipt_path {
-                return Err(Error::other(format!(
-                    "durable ILM decommission receipt path `{receipt_path}` does not match source path `{}` {}",
-                    receipt.source_path,
-                    receipt.context()
-                )));
-            }
+            let receipt = self
+                .read_decommission_durable_ilm_receipt(receipt_pool_idx, &receipt_path)
+                .await?;
            let namespace = classify_durable_ilm_record(&receipt.source_path)?
                .ok_or_else(|| Error::other(format!("path `{}` is not a durable ILM record", receipt.source_path)))?;
-            let (_, target) = self
+            let target = self
                .load_decommissioned_durable_ilm_target(
                    source_pool_idx,
                    &receipt.source_path,
@@ -4544,28 +4667,50 @@ impl ECStore {
                    &receipt.context(),
                )
                .await?;
-            let target_record = validate_durable_ilm_record(&receipt.source_path, &target).map_err(|err| {
-                Error::other(format!(
-                    "target durable ILM record is invalid at path `{}` {}: {err}",
-                    receipt.source_path,
-                    receipt.context()
-                ))
-            })?;
-            if target_record.namespace != receipt.namespace
-                || target_record.id_kind != receipt.id_kind
-                || target_record.id != receipt.id
-            {
+            if let Some((_, target)) = target {
+                let target_record = validate_durable_ilm_record(&receipt.source_path, &target).map_err(|err| {
+                    Error::other(format!(
+                        "target durable ILM record is invalid at path `{}` {}: {err}",
+                        receipt.source_path,
+                        receipt.context()
+                    ))
+                })?;
+                let identity_matches = target_record.namespace == receipt.namespace
+                    && target_record.id_kind == receipt.id_kind
+                    && target_record.id == receipt.id;
+                let reused_manual_scope = receipt.terminal_checkpoint.is_some()
+                    && matches!(
+                        (&receipt.checkpoint, &target_record.checkpoint),
+                        (
+                            DurableIlmRecordCheckpoint::ManualTransitionScope { .. },
+                            DurableIlmRecordCheckpoint::ManualTransitionScope { .. }
+                        )
+                    );
+                if !identity_matches && !reused_manual_scope {
+                    return Err(Error::other(format!(
+                        "target durable ILM record identity mismatch at path `{}` {}; decoded {}",
+                        receipt.source_path,
+                        receipt.context(),
+                        target_record.context()
+                    )));
+                }
+                if identity_matches {
+                    receipt
+                        .terminal_checkpoint
+                        .as_ref()
+                        .unwrap_or(&receipt.checkpoint)
+                        .validate_successor(&target_record.checkpoint)
+                        .map_err(|err| {
+                            Error::other(format!(
+                                "target durable ILM record generation mismatch at path `{}` {}: {err}",
+                                receipt.source_path,
+                                receipt.context()
+                            ))
+                        })?;
+                }
+            } else if receipt.terminal_checkpoint.is_none() {
                return Err(Error::other(format!(
-                    "target durable ILM record identity mismatch at path `{}` {}; decoded {}",
-                    receipt.source_path,
-                    receipt.context(),
-                    target_record.context()
-                )));
-            }
-            let target_content_sha256 = hex_sha256(&target, ToOwned::to_owned);
-            if target_content_sha256 != receipt.target_content_sha256 {
-                return Err(Error::other(format!(
-                    "target durable ILM record content mismatch at path `{}` {}",
+                    "target durable ILM record is missing at path `{}` {} without a recovery terminal checkpoint",
                    receipt.source_path,
                    receipt.context()
                )));
@@ -4574,6 +4719,180 @@ impl ECStore {
        Ok(())
    }

+    async fn advance_durable_ilm_decommission_receipt(
+        &self,
+        pool_idx: usize,
+        receipt_path: &str,
+        record: &ValidatedDurableIlmRecord,
+        terminal: bool,
+    ) -> Result<bool> {
+        let stage = if terminal { "terminal" } else { "progress" };
+        let locator = parse_decommission_durable_ilm_receipt_path(receipt_path)?;
+        let mut attempt = 1;
+        loop {
+            let (receipt_data, metadata) = match read_config_limited_preserve_empty_with_metadata(
+                self.pools[pool_idx].clone(),
+                receipt_path,
+                DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE,
+            )
+            .await
+            {
+                Ok(receipt) => receipt,
+                Err(err)
+                    if matches!(&err, Error::ConfigNotFound | Error::FileNotFound | Error::FileVersionNotFound)
+                        || is_err_object_not_found(&err)
+                        || is_err_version_not_found(&err) =>
+                {
+                    return Ok(false);
+                }
+                Err(err) => {
+                    return Err(Error::other(format!(
+                        "failed to read durable ILM decommission receipt `{receipt_path}` from pool {pool_idx} for {}: {err}",
+                        locator.context()
+                    )));
+                }
+            };
+            let mut receipt = DecommissionDurableIlmReceipt::decode(&receipt_data).map_err(|err| {
+                Error::other(format!(
+                    "durable ILM decommission receipt `{receipt_path}` in pool {pool_idx} for {} is invalid: {err}",
+                    locator.context()
+                ))
+            })?;
+            Self::validate_decommission_durable_ilm_receipt_locator(receipt_path, &locator, &receipt)?;
+            receipt.checkpoint.validate_successor(&record.checkpoint).map_err(|err| {
+                Error::other(format!(
+                    "{stage} durable ILM record generation mismatch at path `{}` {}: {err}",
+                    receipt.source_path,
+                    receipt.context()
+                ))
+            })?;
+
+            if terminal {
+                if let Some(existing) = &receipt.terminal_checkpoint {
+                    if existing == &record.checkpoint || record.checkpoint.validate_successor(existing).is_ok() {
+                        return Ok(true);
+                    }
+                    existing.validate_successor(&record.checkpoint).map_err(|err| {
+                        Error::other(format!(
+                            "terminal durable ILM record checkpoint conflicts at path `{}` {}: {err}",
+                            receipt.source_path,
+                            receipt.context()
+                        ))
+                    })?;
+                }
+                receipt.terminal_checkpoint = Some(record.checkpoint.clone());
+            } else {
+                if let Some(existing) = &receipt.terminal_checkpoint {
+                    if existing == &record.checkpoint {
+                        return Ok(true);
+                    }
+                    existing.validate_successor(&record.checkpoint).map_err(|err| {
+                        Error::other(format!(
+                            "progress durable ILM record conflicts with terminal checkpoint at path `{}` {}: {err}",
+                            receipt.source_path,
+                            receipt.context()
+                        ))
+                    })?;
+                    receipt.terminal_checkpoint = None;
+                }
+                if receipt.checkpoint == record.checkpoint {
+                    return Ok(true);
+                }
+                receipt.checkpoint = record.checkpoint.clone();
+            }
+
+            let etag = metadata.etag.filter(|etag| !etag.trim().is_empty()).ok_or_else(|| {
+                Error::other(format!(
+                    "durable ILM decommission receipt `{receipt_path}` in pool {pool_idx} is missing an ETag"
+                ))
+            })?;
+            let encoded = receipt.encode()?;
+            match save_config_with_opts(
+                self.pools[pool_idx].clone(),
+                receipt_path,
+                encoded,
+                &ObjectOptions {
+                    max_parity: true,
+                    http_preconditions: Some(HTTPPreconditions {
+                        if_match: Some(etag),
+                        ..Default::default()
+                    }),
+                    ..Default::default()
+                },
+            )
+            .await
+            {
+                Ok(()) => return Ok(true),
+                Err(Error::PreconditionFailed) if attempt < DECOMMISSION_DURABLE_ILM_RECEIPT_CAS_ATTEMPTS => {
+                    attempt += 1;
+                    continue;
+                }
+                Err(Error::PreconditionFailed) => {
+                    return Err(Error::other(format!(
+                        "failed to persist {stage} durable ILM decommission receipt `{receipt_path}` for {} after concurrent updates",
+                        locator.context()
+                    )));
+                }
+                Err(err) => {
+                    return Err(Error::other(format!(
+                        "failed to persist {stage} durable ILM decommission receipt `{receipt_path}` for {}: {err}",
+                        locator.context()
+                    )));
+                }
+            }
+        }
+    }
+
+    async fn advance_durable_ilm_decommission_receipts(&self, path: &str, data: &[u8], terminal: bool) -> Result<()> {
+        let active_runs = {
+            let pool_meta = self.pool_meta.read().await;
+            pool_meta
+                .pools
+                .iter()
+                .enumerate()
+                .filter(|(_, pool)| {
+                    pool.decommission
+                        .as_ref()
+                        .is_some_and(|info| info.has_decommission_state() && !info.complete)
+                })
+                .map(|(pool_idx, pool)| (pool_idx, decommission_durable_ilm_receipt_run_token(&pool.cmd_line)))
+                .collect::<Vec<_>>()
+        };
+        if active_runs.is_empty() {
+            return Ok(());
+        }
+
+        let stage = if terminal { "terminal" } else { "progress" };
+        let record = validate_durable_ilm_record(path, data)
+            .map_err(|err| Error::other(format!("{stage} durable ILM record is invalid at path `{path}`: {err}")))?;
+        for (source_pool_idx, run_token) in active_runs {
+            let receipt_path = decommission_durable_ilm_receipt_path(&run_token, path, record.id_kind, &record.id);
+            let mut receipt_found = false;
+            for pool_idx in 0..self.pools.len() {
+                if pool_idx != source_pool_idx {
+                    receipt_found |= self
+                        .advance_durable_ilm_decommission_receipt(pool_idx, &receipt_path, &record, terminal)
+                        .await?;
+                }
+            }
+            if terminal && !receipt_found {
+                return Err(Error::other(format!(
+                    "terminal durable ILM record at path `{path}` {} is retained until its decommission receipt is committed",
+                    record.context()
+                )));
+            }
+        }
+        Ok(())
+    }
+
+    pub(crate) async fn record_durable_ilm_decommission_progress(&self, path: &str, data: &[u8]) -> Result<()> {
+        self.advance_durable_ilm_decommission_receipts(path, data, false).await
+    }
+
+    pub(crate) async fn record_durable_ilm_decommission_terminal(&self, path: &str, data: &[u8]) -> Result<()> {
+        self.advance_durable_ilm_decommission_receipts(path, data, true).await
+    }
+
    async fn cleanup_decommission_durable_ilm_receipts(&self, source_pool_idx: usize) -> Result<()> {
        for (pool_idx, receipt_path) in self.list_decommission_durable_ilm_receipts(source_pool_idx).await? {
            match delete_config(self.pools[pool_idx].clone(), &receipt_path).await {
@@ -4609,9 +4928,15 @@ impl ECStore {
            .map_err(|err| Error::other(format!("source durable ILM record is invalid at path `{path}`: {err}")))?;
        let (target_pool_idx, target) = self
            .load_decommissioned_durable_ilm_target(source_pool_idx, path, namespace.max_record_size, &source_record.context())
-            .await?;
-        validate_decommission_durable_ilm_copy(path, &source_record, &source, &target)?;
-        let receipt = DecommissionDurableIlmReceipt::new(path, &source_record, &target);
+            .await?
+            .ok_or_else(|| {
+                Error::other(format!(
+                    "target durable ILM record is missing at path `{path}` {}",
+                    source_record.context()
+                ))
+            })?;
+        let target_record = validate_decommission_durable_ilm_copy(path, &source_record, &target)?;
+        let receipt = DecommissionDurableIlmReceipt::new(path, &target_record);
        self.persist_decommission_durable_ilm_receipt(source_pool_idx, target_pool_idx, &receipt)
            .await?;

@@ -4650,6 +4975,14 @@ impl ECStore {
        Ok(self.list_decommission_durable_ilm_receipts(source_pool_idx).await?.len())
    }

+    #[cfg(test)]
+    pub(crate) async fn decommission_durable_ilm_receipt_paths_for_test(
+        &self,
+        source_pool_idx: usize,
+    ) -> Result<Vec<(usize, String)>> {
+        self.list_decommission_durable_ilm_receipts(source_pool_idx).await
+    }
+
    #[cfg(test)]
    pub(crate) async fn cleanup_decommission_durable_ilm_receipts_for_test(&self, source_pool_idx: usize) -> Result<()> {
        self.cleanup_decommission_durable_ilm_receipts(source_pool_idx).await
@@ -37,7 +37,7 @@ use crate::bucket::lifecycle::{
    transition_transaction::{
        TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction,
        TransitionTransactionInit, TransitionTransactionState, delete_transition_transaction_record,
-        save_transition_transaction_record,
+        load_transition_transaction_record, save_transition_transaction_record,
    },
 };
 use crate::bucket::quota::reservation;
@@ -4246,7 +4246,12 @@ fn record_transition_uploaded_save_attempt(transaction: &TransitionTransaction,

 async fn delete_transition_transaction_if_available(api: Option<&Arc<ECStore>>, transaction_id: Uuid) -> Result<()> {
    if let Some(api) = api {
-        return delete_transition_transaction_record(api.clone(), transaction_id).await;
+        let transaction = match load_transition_transaction_record(api.clone(), transaction_id).await {
+            Ok(transaction) => transaction,
+            Err(Error::ConfigNotFound) => return Ok(()),
+            Err(err) => return Err(err),
+        };
+        return delete_transition_transaction_record(api.clone(), &transaction).await;
    }
    Ok(())
 }
@@ -3055,7 +3055,7 @@ mod tests {
            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "durable-ilm-decommission", &[4, 4])).await;

        let tier_name = "DECOMMISSION-ILM";
-        let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await;
+        let backend = register_transition_reconcile_test_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")
@@ -3073,7 +3073,7 @@ mod tests {
        let tier_path = tier_delete_journal_object_name(&tier_entry);
        let tier_bytes = encode_tier_delete_journal_entry(&tier_entry).expect("tier journal should encode");

-        let transaction = TransitionTransaction::new(TransitionTransactionInit {
+        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(),
@@ -3093,6 +3093,10 @@ mod tests {
            not_after_unix_nanos: 1_780_000_000_000_000_000,
        })
        .expect("transition transaction should build");
+        let regressed_transaction = transaction.clone();
+        transaction
+            .advance(transaction.fence(), TransitionTransactionState::UploadOutcomeUnknown, None)
+            .expect("transition transaction should advance before migration");
        let transaction_path = transition_transaction_record_object_name(transaction.transaction_id)
            .expect("transition transaction path should build");
        let transaction_bytes = transaction.encode().expect("transition transaction should encode");
@@ -3212,6 +3216,25 @@ mod tests {
                *expected
            );
        }
+        assert_eq!(
+            store
+                .decommission_durable_ilm_receipt_count_for_test(0)
+                .await
+                .expect("no receipt should exist before the final sweep"),
+            0
+        );
+        let isolated_tier_stats = recover_tier_delete_journal_entries(store.clone(), 100, None)
+            .await
+            .expect("tier recovery should retain a terminal record until its receipt is committed");
+        assert!(isolated_tier_stats.scanned >= 1);
+        assert_eq!(isolated_tier_stats.deleted, 0);
+        assert!(isolated_tier_stats.failed >= 1);
+        assert_eq!(
+            com::read_config(store.pools[1].clone(), &tier_path)
+                .await
+                .expect("receipt isolation must retain the target tier journal"),
+            tier_bytes
+        );

        com::delete_config(store.pools[1].clone(), &manual_job_path)
            .await
@@ -3404,6 +3427,161 @@ mod tests {
            .await
            .expect("post-crash target transaction should restore");

+        let mut wrong_manual_job = manual_job.clone();
+        wrong_manual_job.job_id = uuid::Uuid::new_v4();
+        com::save_config(
+            store.pools[1].clone(),
+            &manual_job_path,
+            wrong_manual_job.encode().expect("wrong-id job should encode"),
+        )
+        .await
+        .expect("post-crash target manual job should accept the wrong-id fixture");
+        let wrong_id_after_crash = store
+            .complete_decommission(0)
+            .await
+            .expect_err("completion must reject a target record with the wrong id")
+            .to_string();
+        assert!(wrong_id_after_crash.contains(&manual_job_path));
+        assert!(wrong_id_after_crash.contains(&manual_job_id.to_string()));
+        com::save_config(store.pools[1].clone(), &manual_job_path, manual_job_bytes.clone())
+            .await
+            .expect("post-crash target manual job should restore after the wrong-id check");
+
+        com::save_config(
+            store.pools[1].clone(),
+            &transaction_path,
+            regressed_transaction
+                .encode()
+                .expect("regressed transition transaction should encode"),
+        )
+        .await
+        .expect("post-crash target transaction should accept the regression fixture");
+        let regression_after_crash = store
+            .complete_decommission(0)
+            .await
+            .expect_err("completion must reject a lower transition transaction revision")
+            .to_string();
+        assert!(regression_after_crash.contains("generation mismatch"));
+        assert!(regression_after_crash.contains(&transaction_path));
+        assert!(regression_after_crash.contains(&transaction.transaction_id.to_string()));
+        com::save_config(store.pools[1].clone(), &transaction_path, transaction_bytes.clone())
+            .await
+            .expect("post-crash target transaction should restore after the regression check");
+
+        let (manual_task_receipt_pool, manual_task_receipt_path) = store
+            .decommission_durable_ilm_receipt_paths_for_test(0)
+            .await
+            .expect("durable ILM receipt paths should be listable")
+            .into_iter()
+            .find(|(_, path)| path.contains(&manual_task_path))
+            .expect("manual task receipt should retain its reversible source path");
+        let manual_task_receipt_bytes =
+            com::read_config(store.pools[manual_task_receipt_pool].clone(), &manual_task_receipt_path)
+                .await
+                .expect("manual task receipt should be readable before corruption");
+        com::save_config(
+            store.pools[manual_task_receipt_pool].clone(),
+            &manual_task_receipt_path,
+            b"{corrupt".to_vec(),
+        )
+        .await
+        .expect("manual task receipt should corrupt deterministically");
+        let corrupt_receipt = store
+            .complete_decommission(0)
+            .await
+            .expect_err("completion must fail closed on a corrupt receipt")
+            .to_string();
+        assert!(corrupt_receipt.contains(&manual_task_path));
+        assert!(corrupt_receipt.contains(&manual_job_id.to_string()));
+        com::save_config(
+            store.pools[manual_task_receipt_pool].clone(),
+            &manual_task_receipt_path,
+            manual_task_receipt_bytes,
+        )
+        .await
+        .expect("manual task receipt should restore after the corruption check");
+
+        let tier_stats = recover_tier_delete_journal_entries(store.clone(), 100, None)
+            .await
+            .expect("tier journal recovery should consume the migrated record before completion");
+        assert_eq!((tier_stats.scanned, tier_stats.deleted, tier_stats.failed), (1, 1, 0));
+        assert!(matches!(com::read_config(store.clone(), &tier_path).await, Err(Error::ConfigNotFound)));
+
+        let recovered_transition_version = "recovered-transition-version".to_string();
+        backend
+            .set_transition_candidate_probe_override(Some(TransitionCandidateProbe::VersionedPresent(
+                recovered_transition_version.clone(),
+            )))
+            .await;
+        let transaction_stats = recover_transition_transaction_records(store.clone(), 100, None)
+            .await
+            .expect("transition recovery should advance and consume the migrated transaction before completion");
+        backend.set_transition_candidate_probe_override(None).await;
+        assert_eq!(
+            (
+                transaction_stats.scanned,
+                transaction_stats.recovered,
+                transaction_stats.retained,
+                transaction_stats.failed,
+            ),
+            (1, 1, 0, 0)
+        );
+        assert!(matches!(
+            com::read_config(store.clone(), &transaction_path).await,
+            Err(Error::ConfigNotFound)
+        ));
+        com::save_config(
+            store.pools[1].clone(),
+            &transaction_path,
+            regressed_transaction
+                .encode()
+                .expect("post-terminal transition rollback should encode"),
+        )
+        .await
+        .expect("target should accept the post-terminal rollback fixture");
+        let post_terminal_regression = store
+            .complete_decommission(0)
+            .await
+            .expect_err("terminal proof must not mask a lower transition revision")
+            .to_string();
+        assert!(post_terminal_regression.contains("generation mismatch"));
+        assert!(post_terminal_regression.contains(&transaction_path));
+        assert!(post_terminal_regression.contains(&transaction.transaction_id.to_string()));
+        com::delete_config(store.pools[1].clone(), &transaction_path)
+            .await
+            .expect("post-terminal rollback fixture should be removed");
+
+        let manual_stats = recover_manual_transition_jobs_once(store.clone(), 100, None)
+            .await
+            .expect("manual recovery should advance the migrated job and consume its scope before completion");
+        assert_eq!(
+            (manual_stats.scanned, manual_stats.resumed, manual_stats.skipped, manual_stats.failed,),
+            (1, 1, 0, 0)
+        );
+        assert!(matches!(
+            com::read_config(store.clone(), &manual_scope_path).await,
+            Err(Error::ConfigNotFound)
+        ));
+        let recovered_manual_job_bytes = com::read_config(store.pools[1].clone(), &manual_job_path)
+            .await
+            .expect("manual recovery should retain the advanced job record");
+        assert_ne!(recovered_manual_job_bytes, manual_job_bytes);
+
+        com::save_config(store.pools[1].clone(), &manual_job_path, manual_job_bytes.clone())
+            .await
+            .expect("target manual job should accept the rollback fixture");
+        let manual_regression = store
+            .complete_decommission(0)
+            .await
+            .expect_err("completion must reject a manual job generation rollback")
+            .to_string();
+        assert!(manual_regression.contains("generation mismatch"));
+        assert!(manual_regression.contains(&manual_job_path));
+        assert!(manual_regression.contains(&manual_job_id.to_string()));
+        com::save_config(store.pools[1].clone(), &manual_job_path, recovered_manual_job_bytes)
+            .await
+            .expect("target manual job should restore its recovered generation");
+
        store
            .complete_decommission(0)
            .await
@@ -3426,38 +3604,9 @@ mod tests {
            .cleanup_decommission_durable_ilm_receipts_for_test(0)
            .await
            .expect("receipt cleanup should be idempotent");
-
-        let tier_stats = recover_tier_delete_journal_entries(store.clone(), 100, None)
-            .await
-            .expect("tier journal recovery should consume the migrated record");
-        assert_eq!((tier_stats.scanned, tier_stats.deleted, tier_stats.failed), (1, 1, 0));
-        assert_eq!(
-            backend.remove_versions().await,
-            vec![(tier_entry.obj_name.clone(), tier_entry.version_id.clone())]
-        );
-        let transaction_stats = recover_transition_transaction_records(store.clone(), 100, None)
-            .await
-            .expect("transition transaction recovery should read the migrated record");
-        assert_eq!(
-            (
-                transaction_stats.scanned,
-                transaction_stats.recovered,
-                transaction_stats.retained,
-                transaction_stats.failed,
-            ),
-            (1, 0, 1, 0)
-        );
-        let manual_stats = recover_manual_transition_jobs_once(store.clone(), 100, None)
-            .await
-            .expect("manual recovery should reconcile migrated job, scope, task, and result records");
-        assert_eq!(
-            (manual_stats.scanned, manual_stats.resumed, manual_stats.skipped, manual_stats.failed,),
-            (1, 1, 0, 0)
-        );
-        assert!(matches!(
-            com::read_config(store.clone(), &manual_scope_path).await,
-            Err(Error::ConfigNotFound)
-        ));
+        let removed_versions = backend.remove_versions().await;
+        assert!(removed_versions.contains(&(tier_entry.obj_name.clone(), tier_entry.version_id.clone())));
+        assert!(removed_versions.contains(&(transaction.remote_object.clone(), recovered_transition_version)));
    }

    #[cfg(feature = "test-util")]