fix(ecstore): close ILM receipt recovery gaps

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

@@ -16,7 +16,12 @@ use rustfs_utils::crypto::hex_sha256;
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
-use super::{manual_transition_job, tier_delete_journal, transition_transaction};
+use super::{
+    bucket_lifecycle_ops::{
+        ManualTransitionQueueSnapshot, ManualTransitionRunReport, decode_manual_transition_continuation_token,
+    },
+    manual_transition_job, tier_delete_journal, transition_transaction,
+};
 use crate::error::{Error, Result};
 
 pub(crate) const ILM_META_PREFIX: &str = "ilm";
@@ -94,6 +99,13 @@ pub(crate) struct ValidatedDurableIlmRecord {
     pub(crate) checkpoint: DurableIlmRecordCheckpoint,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct ManualTransitionJobProgressCheckpoint {
+    report: ManualTransitionRunReport,
+    queue_snapshot: ManualTransitionQueueSnapshot,
+}
+
 impl ValidatedDurableIlmRecord {
     pub(crate) fn context(&self) -> String {
         format!("namespace `{}` {} `{}`", self.namespace, self.id_kind, self.id)
@@ -123,6 +135,8 @@ pub(crate) enum DurableIlmRecordCheckpoint {
         state: manual_transition_job::ManualTransitionJobState,
         scan_completed: bool,
         cancel_requested: bool,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        progress: Option<Box<ManualTransitionJobProgressCheckpoint>>,
     },
     ManualTransitionScope {
         content_sha256: String,
@@ -151,6 +165,14 @@ impl DurableIlmRecordCheckpoint {
 
     pub(crate) fn validate_successor(&self, next: &Self) -> Result<()> {
         if self == next {
+            if let Self::ManualTransitionJob {
+                progress: Some(progress),
+                ..
+            } = self
+                && !manual_job_progress_is_valid(progress)
+            {
+                return Err(Error::other("durable ILM manual transition checkpoint is invalid"));
+            }
             return Ok(());
         }
 
@@ -200,6 +222,7 @@ impl DurableIlmRecordCheckpoint {
                     state: previous_state,
                     scan_completed: previous_scan_completed,
                     cancel_requested: previous_cancel_requested,
+                    progress: previous_progress,
                     ..
                 },
                 Self::ManualTransitionJob {
@@ -208,6 +231,7 @@ impl DurableIlmRecordCheckpoint {
                     state: next_state,
                     scan_completed: next_scan_completed,
                     cancel_requested: next_cancel_requested,
+                    progress: next_progress,
                     ..
                 },
             ) => {
@@ -216,6 +240,7 @@ impl DurableIlmRecordCheckpoint {
                     && manual_job_state_reaches(*previous_state, *next_state)
                     && (!previous_scan_completed || *next_scan_completed)
                     && (!previous_cancel_requested || *next_cancel_requested)
+                    && manual_job_progress_reaches(previous_progress.as_deref(), next_progress.as_deref(), *next_scan_completed)
             }
             (
                 Self::ManualTransitionScope {
@@ -269,6 +294,129 @@ fn manual_job_state_reaches(
     from == to || from == manual_transition_job::ManualTransitionJobState::Running
 }
 
+fn manual_job_progress_reaches(
+    previous: Option<&ManualTransitionJobProgressCheckpoint>,
+    next: Option<&ManualTransitionJobProgressCheckpoint>,
+    next_scan_completed: bool,
+) -> bool {
+    let (previous, next) = match (previous, next) {
+        (None, Some(next)) => return manual_job_progress_is_valid(next),
+        (Some(previous), Some(next)) => (previous, next),
+        _ => return false,
+    };
+    let previous_report = &previous.report;
+    let next_report = &next.report;
+
+    macro_rules! counters_do_not_regress {
+        ($($field:ident),+ $(,)?) => {
+            $(previous_report.$field <= next_report.$field)&&+
+        };
+    }
+
+    let counters_monotonic = counters_do_not_regress!(
+        scanned,
+        eligible,
+        enqueued,
+        dry_run_eligible,
+        skipped_not_transition,
+        skipped_tier,
+        skipped_delete_marker,
+        skipped_directory,
+        skipped_replication,
+        skipped_already_transitioned,
+        skipped_already_in_flight,
+        skipped_queue_full,
+        skipped_queue_closed,
+        skipped_queue_timeout,
+        transition_completed,
+        transition_failed,
+        tier_failure,
+    );
+    let failure_reasons_monotonic = previous_report.tier_failure_by_reason.iter().all(|(reason, previous_count)| {
+        next_report
+            .tier_failure_by_reason
+            .get(reason)
+            .is_some_and(|next_count| next_count >= previous_count)
+    });
+    let flags_monotonic = (!previous_report.lifecycle_config_found || next_report.lifecycle_config_found)
+        && (!previous_report.truncated_by_limit || next_report.truncated_by_limit)
+        && (!previous_report.truncated_by_duration || next_report.truncated_by_duration)
+        && (!previous_report.cancelled || next_report.cancelled);
+    let cursor_monotonic = manual_job_cursor_reaches(previous_report, next_report, next_scan_completed);
+    let progress_valid = manual_job_progress_is_valid(previous) && manual_job_progress_is_valid(next);
+
+    previous_report.bucket == next_report.bucket
+        && previous_report.prefix == next_report.prefix
+        && previous_report.tier == next_report.tier
+        && previous_report.dry_run == next_report.dry_run
+        && counters_monotonic
+        && failure_reasons_monotonic
+        && flags_monotonic
+        && cursor_monotonic
+        && progress_valid
+}
+
+fn manual_job_progress_is_valid(progress: &ManualTransitionJobProgressCheckpoint) -> bool {
+    manual_job_worker_results_are_valid(&progress.report)
+        && manual_job_queue_snapshot_is_valid(&progress.queue_snapshot)
+        && manual_job_cursor_is_valid(progress.report.continuation_token.as_deref())
+}
+
+fn manual_job_worker_results_are_valid(report: &ManualTransitionRunReport) -> bool {
+    let reason_total = report
+        .tier_failure_by_reason
+        .values()
+        .try_fold(0u64, |total, count| total.checked_add(*count));
+    report
+        .transition_completed
+        .checked_add(report.transition_failed)
+        .is_some_and(|total| total <= report.enqueued)
+        && report.transition_failed <= report.tier_failure
+        && reason_total.is_some_and(|total| total <= report.tier_failure)
+}
+
+fn manual_job_cursor_reaches(
+    previous: &ManualTransitionRunReport,
+    next: &ManualTransitionRunReport,
+    next_scan_completed: bool,
+) -> bool {
+    if previous.continuation_token == next.continuation_token {
+        return manual_job_cursor_is_valid(previous.continuation_token.as_deref());
+    }
+    match (&previous.continuation_token, &next.continuation_token) {
+        (None, Some(next_token)) => next.scanned > previous.scanned && manual_job_cursor_is_valid(Some(next_token)),
+        (Some(_), None) => next_scan_completed,
+        (Some(previous_token), Some(next_token)) if next.scanned > previous.scanned => {
+            let (Ok((Some(previous_marker), previous_version)), Ok((Some(next_marker), next_version))) = (
+                decode_manual_transition_continuation_token(previous_token),
+                decode_manual_transition_continuation_token(next_token),
+            ) else {
+                return false;
+            };
+            next_marker > previous_marker
+                || (next_marker == previous_marker
+                    && previous_version.is_some()
+                    && next_version.is_some()
+                    && previous_version != next_version)
+        }
+        _ => false,
+    }
+}
+
+fn manual_job_cursor_is_valid(token: Option<&str>) -> bool {
+    let Some(token) = token else {
+        return true;
+    };
+    matches!(decode_manual_transition_continuation_token(token), Ok((Some(_), _)))
+}
+
+fn manual_job_queue_snapshot_is_valid(snapshot: &ManualTransitionQueueSnapshot) -> bool {
+    (snapshot.queue_capacity > 0 || snapshot.queued == 0)
+        && (snapshot.queue_capacity == 0 || snapshot.queued <= snapshot.queue_capacity)
+        && (snapshot.workers > 0 || snapshot.active == 0)
+        && (snapshot.workers == 0 || snapshot.active <= snapshot.workers)
+}
+
 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))
@@ -429,6 +577,10 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
                     state: job.state,
                     scan_completed: job.scan_completed,
                     cancel_requested: job.cancel_requested,
+                    progress: Some(Box::new(ManualTransitionJobProgressCheckpoint {
+                        report: job.report,
+                        queue_snapshot: job.queue_snapshot,
+                    })),
                 },
             )
         }
@@ -504,6 +656,21 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
 mod tests {
     use super::*;
 
+    fn manual_job_checkpoint(job: &manual_transition_job::ManualTransitionJobRecord) -> DurableIlmRecordCheckpoint {
+        let path =
+            manual_transition_job::manual_transition_job_record_object_name(job.job_id).expect("manual job path should build");
+        let encoded = job.encode().expect("manual job should encode");
+        validate_durable_ilm_record(&path, &encoded)
+            .expect("manual job checkpoint should validate")
+            .checkpoint
+    }
+
+    fn continuation_token(marker: &str) -> String {
+        let encoded = serde_json::to_vec(&serde_json::json!({ "marker": marker, "version_marker": null }))
+            .expect("continuation token should encode");
+        base64_simd::URL_SAFE_NO_PAD.encode_to_string(&encoded)
+    }
+
     #[test]
     fn unknown_ilm_record_requires_namespace_registration() {
         let err = classify_durable_ilm_record("ilm/future-durable/jobs/one.json")
@@ -524,4 +691,119 @@ mod tests {
             }
         }
     }
+
+    #[test]
+    fn manual_transition_job_checkpoint_rejects_progress_poison() {
+        let options = super::super::bucket_lifecycle_ops::ManualTransitionRunOptions::default();
+        let mut initial =
+            manual_transition_job::ManualTransitionJobRecord::new(Uuid::new_v4(), "manual-checkpoint-bucket", &options, "owner");
+        let initial_checkpoint = manual_job_checkpoint(&initial);
+        initial.updated_at_unix_nanos += 1;
+        initial.report.scanned = 1;
+        initial.report.continuation_token = Some(continuation_token("logs/a"));
+        let first_page_checkpoint = manual_job_checkpoint(&initial);
+        initial_checkpoint
+            .validate_successor(&first_page_checkpoint)
+            .expect("the first durable cursor should advance from no cursor");
+
+        let mut legacy_checkpoint = initial_checkpoint;
+        let DurableIlmRecordCheckpoint::ManualTransitionJob { progress, .. } = &mut legacy_checkpoint else {
+            panic!("manual job should produce a manual checkpoint");
+        };
+        *progress = None;
+        legacy_checkpoint
+            .validate_successor(&first_page_checkpoint)
+            .expect("legacy checkpoints should upgrade to validated progress");
+
+        let mut previous = initial;
+        previous.updated_at_unix_nanos += 1;
+        previous.report.scanned = 10;
+        previous.report.eligible = 8;
+        previous.report.enqueued = 2;
+        previous.report.transition_completed = 1;
+        previous.report.continuation_token = Some(continuation_token("logs/b"));
+        previous.queue_snapshot = ManualTransitionQueueSnapshot {
+            queue_capacity: 10,
+            queued: 1,
+            active: 1,
+            workers: 2,
+            queue_full: 2,
+            queue_send_timeout: 1,
+            ..Default::default()
+        };
+        let previous_checkpoint = manual_job_checkpoint(&previous);
+
+        let mut next = previous.clone();
+        next.updated_at_unix_nanos += 1;
+        next.report.scanned = 11;
+        next.report.eligible = 9;
+        next.report.transition_completed = 2;
+        next.report.continuation_token = Some(continuation_token("logs/c"));
+        next.queue_snapshot.queued = 0;
+        next.queue_snapshot.active = 0;
+        next.queue_snapshot.queue_full = 3;
+        let next_checkpoint = manual_job_checkpoint(&next);
+        previous_checkpoint
+            .validate_successor(&next_checkpoint)
+            .expect("forward job progress should validate");
+
+        let mut counter_rollback = next.clone();
+        counter_rollback.updated_at_unix_nanos += 1;
+        counter_rollback.report.scanned = 9;
+        assert!(
+            previous_checkpoint
+                .validate_successor(&manual_job_checkpoint(&counter_rollback))
+                .is_err()
+        );
+
+        let mut cursor_rollback = next.clone();
+        cursor_rollback.updated_at_unix_nanos += 1;
+        cursor_rollback.report.scanned = previous.report.scanned;
+        cursor_rollback.report.scanned += 1;
+        cursor_rollback.report.continuation_token = Some(continuation_token("logs/a"));
+        assert!(
+            previous_checkpoint
+                .validate_successor(&manual_job_checkpoint(&cursor_rollback))
+                .is_err()
+        );
+
+        let mut worker_result_rollback = next.clone();
+        worker_result_rollback.updated_at_unix_nanos += 1;
+        worker_result_rollback.report.transition_completed = 0;
+        assert!(
+            previous_checkpoint
+                .validate_successor(&manual_job_checkpoint(&worker_result_rollback))
+                .is_err()
+        );
+
+        let mut worker_result_overflow = next.clone();
+        worker_result_overflow.updated_at_unix_nanos += 1;
+        worker_result_overflow.report.enqueued = u64::MAX;
+        worker_result_overflow.report.transition_completed = u64::MAX;
+        worker_result_overflow.report.transition_failed = 1;
+        worker_result_overflow.report.tier_failure = 1;
+        assert!(
+            previous_checkpoint
+                .validate_successor(&manual_job_checkpoint(&worker_result_overflow))
+                .is_err()
+        );
+
+        let mut invalid_cursor = next.clone();
+        invalid_cursor.updated_at_unix_nanos += 1;
+        invalid_cursor.report.continuation_token = Some("not-base64".to_string());
+        assert!(
+            previous_checkpoint
+                .validate_successor(&manual_job_checkpoint(&invalid_cursor))
+                .is_err()
+        );
+
+        let mut queue_state_poison = next;
+        queue_state_poison.updated_at_unix_nanos += 1;
+        queue_state_poison.queue_snapshot.queued = queue_state_poison.queue_snapshot.queue_capacity + 1;
+        assert!(
+            previous_checkpoint
+                .validate_successor(&manual_job_checkpoint(&queue_state_poison))
+                .is_err()
+        );
+    }
 }

crates/ecstore/src/core/pools.rs

@@ -924,6 +924,65 @@ impl DecommissionDurableIlmReceipt {
     }
 }
 
+fn merge_decommission_durable_ilm_receipts(
+    existing: &DecommissionDurableIlmReceipt,
+    incoming: &DecommissionDurableIlmReceipt,
+) -> Result<DecommissionDurableIlmReceipt> {
+    if existing.source_path != incoming.source_path
+        || existing.namespace != incoming.namespace
+        || existing.id_kind != incoming.id_kind
+        || existing.id != incoming.id
+    {
+        return Err(Error::other(format!(
+            "durable ILM receipt identity conflict for source path `{}` {}; incoming {}",
+            existing.source_path,
+            existing.context(),
+            incoming.context()
+        )));
+    }
+
+    let checkpoint =
+        if existing.checkpoint == incoming.checkpoint || incoming.checkpoint.validate_successor(&existing.checkpoint).is_ok() {
+            existing.checkpoint.clone()
+        } else {
+            existing.checkpoint.validate_successor(&incoming.checkpoint).map_err(|err| {
+                Error::other(format!(
+                    "durable ILM receipt checkpoint conflict for source path `{}` {}: {err}",
+                    existing.source_path,
+                    existing.context()
+                ))
+            })?;
+            incoming.checkpoint.clone()
+        };
+    let terminal_checkpoint = match (&existing.terminal_checkpoint, &incoming.terminal_checkpoint) {
+        (Some(existing), Some(incoming)) if existing == incoming => Some(existing.clone()),
+        (Some(existing), Some(incoming)) if incoming.validate_successor(existing).is_ok() => Some(existing.clone()),
+        (Some(existing), Some(incoming)) => {
+            existing.validate_successor(incoming).map_err(|err| {
+                Error::other(format!(
+                    "durable ILM receipt terminal checkpoint conflict for source path `{}` {}: {err}",
+                    existing.source_path,
+                    existing.context()
+                ))
+            })?;
+            Some(incoming.clone())
+        }
+        (Some(existing), None) => Some(existing.clone()),
+        (None, Some(incoming)) => Some(incoming.clone()),
+        (None, None) => None,
+    };
+    let merged = DecommissionDurableIlmReceipt {
+        source_path: existing.source_path.clone(),
+        namespace: existing.namespace.clone(),
+        id_kind: existing.id_kind.clone(),
+        id: existing.id.clone(),
+        checkpoint,
+        terminal_checkpoint,
+    };
+    merged.validate()?;
+    Ok(merged)
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 struct PersistedDecommissionDurableIlmReceipt {
@@ -946,8 +1005,9 @@ impl DecommissionDurableIlmReceiptLocator {
     }
 }
 
-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_token(cmd_line: &str, start_time: OffsetDateTime) -> String {
+    let identity = format!("{cmd_line}\0{}", start_time.unix_timestamp_nanos());
+    hex_sha256(identity.as_bytes(), ToOwned::to_owned)
 }
 
 fn decommission_durable_ilm_receipt_run_prefix(run_token: &str) -> String {
@@ -4475,13 +4535,16 @@ impl ECStore {
 
     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
+        let pool = pool_meta
             .pools
             .get(source_pool_idx)
-            .ok_or_else(|| invalid_decommission_pool_index_error(pool_meta.pools.len(), source_pool_idx))?
-            .cmd_line
-            .clone();
-        Ok(decommission_durable_ilm_receipt_run_token(&cmd_line))
+            .ok_or_else(|| invalid_decommission_pool_index_error(pool_meta.pools.len(), source_pool_idx))?;
+        let start_time = pool
+            .decommission
+            .as_ref()
+            .and_then(|info| info.start_time)
+            .ok_or_else(|| Error::other(format!("decommission run identity is missing for pool {source_pool_idx}")))?;
+        Ok(decommission_durable_ilm_receipt_run_token(&pool.cmd_line, start_time))
     }
 
     async fn load_decommissioned_durable_ilm_target(
@@ -4522,11 +4585,10 @@ impl ECStore {
             }
         }
 
-        match (target, first_read_error) {
-            (Some(target), _) => Ok(Some(target)),
-            (None, Some(err)) => Err(err),
-            (None, None) => Ok(None),
+        if let Some(err) = first_read_error {
+            return Err(err);
         }
+        Ok(target)
     }
 
     async fn list_decommission_durable_ilm_receipt_paths_in_pool(&self, pool_idx: usize, prefix: &str) -> Result<Vec<String>> {
@@ -4591,22 +4653,98 @@ impl ECStore {
     ) -> Result<()> {
         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}",
-                receipt.source_path,
-                receipt.context()
-            ))
-        })?;
-        save_config(self.pools[target_pool_idx].clone(), &receipt_path, encoded)
+        let locator = parse_decommission_durable_ilm_receipt_path(&receipt_path)?;
+        let mut attempt = 1;
+        loop {
+            let (merged, http_preconditions) = match read_config_limited_preserve_empty_with_metadata(
+                self.pools[target_pool_idx].clone(),
+                &receipt_path,
+                DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE,
+            )
             .await
-            .map_err(|err| {
+            {
+                Ok((existing_data, metadata)) => {
+                    let existing = DecommissionDurableIlmReceipt::decode(&existing_data).map_err(|err| {
+                        Error::other(format!(
+                            "durable ILM decommission receipt `{receipt_path}` in pool {target_pool_idx} for {} is invalid: {err}",
+                            locator.context()
+                        ))
+                    })?;
+                    Self::validate_decommission_durable_ilm_receipt_locator(&receipt_path, &locator, &existing)?;
+                    let merged = merge_decommission_durable_ilm_receipts(&existing, receipt)?;
+                    if merged == existing {
+                        return Ok(());
+                    }
+                    let etag = metadata.etag.filter(|etag| !etag.trim().is_empty()).ok_or_else(|| {
+                        Error::other(format!(
+                            "durable ILM decommission receipt `{receipt_path}` in pool {target_pool_idx} is missing an ETag"
+                        ))
+                    })?;
+                    (
+                        merged,
+                        HTTPPreconditions {
+                            if_match: Some(etag),
+                            ..Default::default()
+                        },
+                    )
+                }
+                Err(err)
+                    if matches!(&err, Error::ConfigNotFound | Error::FileNotFound | Error::FileVersionNotFound)
+                        || is_err_object_not_found(&err)
+                        || is_err_version_not_found(&err) =>
+                {
+                    (
+                        receipt.clone(),
+                        HTTPPreconditions {
+                            if_none_match: Some("*".to_string()),
+                            ..Default::default()
+                        },
+                    )
+                }
+                Err(err) => {
+                    return Err(Error::other(format!(
+                        "failed to read durable ILM decommission receipt `{receipt_path}` from pool {target_pool_idx} for {}: {err}",
+                        locator.context()
+                    )));
+                }
+            };
+            let encoded = merged.encode().map_err(|err| {
                 Error::other(format!(
-                    "failed to persist durable ILM decommission receipt `{receipt_path}` for source path `{}` {}: {err}",
+                    "failed to encode durable ILM decommission receipt `{receipt_path}` for source path `{}` {}: {err}",
                     receipt.source_path,
                     receipt.context()
                 ))
-            })
+            })?;
+            match save_config_with_opts(
+                self.pools[target_pool_idx].clone(),
+                &receipt_path,
+                encoded,
+                &ObjectOptions {
+                    max_parity: true,
+                    http_preconditions: Some(http_preconditions),
+                    ..Default::default()
+                },
+            )
+            .await
+            {
+                Ok(()) => return Ok(()),
+                Err(Error::PreconditionFailed) if attempt < DECOMMISSION_DURABLE_ILM_RECEIPT_CAS_ATTEMPTS => {
+                    attempt += 1;
+                }
+                Err(Error::PreconditionFailed) => {
+                    return Err(Error::other(format!(
+                        "failed to persist durable ILM decommission receipt `{receipt_path}` for {} after concurrent updates",
+                        locator.context()
+                    )));
+                }
+                Err(err) => {
+                    return Err(Error::other(format!(
+                        "failed to persist durable ILM decommission receipt `{receipt_path}` for {}: {err}",
+                        locator.context()
+                    )));
+                }
+            }
+        }
     }
 
     fn validate_decommission_durable_ilm_receipt_locator(
@@ -4652,6 +4790,78 @@ impl ECStore {
         Ok(receipt)
     }
 
+    async fn load_decommission_durable_ilm_terminal_receipt(
+        &self,
+        source_pool_idx: usize,
+        path: &str,
+        source_record: &ValidatedDurableIlmRecord,
+    ) -> Result<Option<DecommissionDurableIlmReceipt>> {
+        let run_token = self.durable_ilm_receipt_run_token(source_pool_idx).await?;
+        let receipt_path = decommission_durable_ilm_receipt_path(&run_token, path, source_record.id_kind, &source_record.id);
+        let locator = parse_decommission_durable_ilm_receipt_path(&receipt_path)?;
+        let mut proof = None::<DecommissionDurableIlmReceipt>;
+        for pool_idx in 0..self.pools.len() {
+            if pool_idx == source_pool_idx {
+                continue;
+            }
+            let data = match read_config_limited_preserve_empty(
+                self.pools[pool_idx].clone(),
+                &receipt_path,
+                DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE,
+            )
+            .await
+            {
+                Ok(data) => data,
+                Err(err)
+                    if matches!(&err, Error::ConfigNotFound | Error::FileNotFound | Error::FileVersionNotFound)
+                        || is_err_object_not_found(&err)
+                        || is_err_version_not_found(&err) =>
+                {
+                    continue;
+                }
+                Err(err) => {
+                    return Err(Error::other(format!(
+                        "failed to read terminal durable ILM decommission receipt `{receipt_path}` from pool {pool_idx} for {}: {err}",
+                        source_record.context()
+                    )));
+                }
+            };
+            let receipt = DecommissionDurableIlmReceipt::decode(&data).map_err(|err| {
+                Error::other(format!(
+                    "terminal durable ILM decommission receipt `{receipt_path}` in pool {pool_idx} for {} is invalid: {err}",
+                    source_record.context()
+                ))
+            })?;
+            Self::validate_decommission_durable_ilm_receipt_locator(&receipt_path, &locator, &receipt)?;
+            if receipt.namespace != source_record.namespace
+                || receipt.id_kind != source_record.id_kind
+                || receipt.id != source_record.id
+            {
+                return Err(Error::other(format!(
+                    "terminal durable ILM decommission receipt identity mismatch at path `{path}` {}; receipt {}",
+                    source_record.context(),
+                    receipt.context()
+                )));
+            }
+            source_record
+                .checkpoint
+                .validate_successor(&receipt.checkpoint)
+                .map_err(|err| {
+                    Error::other(format!(
+                        "terminal durable ILM decommission receipt does not cover source at path `{path}` {}: {err}",
+                        source_record.context()
+                    ))
+                })?;
+            if receipt.terminal_checkpoint.is_some() {
+                proof = Some(match proof {
+                    Some(existing) => merge_decommission_durable_ilm_receipts(&existing, &receipt)?,
+                    None => receipt,
+                });
+            }
+        }
+        Ok(proof)
+    }
+
     async fn verify_decommission_durable_ilm_receipts(&self, source_pool_idx: usize) -> Result<()> {
         for (receipt_pool_idx, receipt_path) in self.list_decommission_durable_ilm_receipts(source_pool_idx).await? {
             let receipt = self
@@ -4850,12 +5060,13 @@ impl ECStore {
                 .pools
                 .iter()
                 .enumerate()
-                .filter(|(_, pool)| {
+                .filter_map(|(pool_idx, pool)| {
                     pool.decommission
                         .as_ref()
-                        .is_some_and(|info| info.has_decommission_state() && !info.complete)
+                        .filter(|info| info.has_decommission_state() && !info.complete)
+                        .and_then(|info| info.start_time)
+                        .map(|start_time| (pool_idx, decommission_durable_ilm_receipt_run_token(&pool.cmd_line, start_time)))
                 })
-                .map(|(pool_idx, pool)| (pool_idx, decommission_durable_ilm_receipt_run_token(&pool.cmd_line)))
                 .collect::<Vec<_>>()
         };
         if active_runs.is_empty() {
@@ -4926,19 +5137,24 @@ impl ECStore {
             .map_err(|err| Error::other(format!("failed to read source durable ILM record at path `{path}`: {err}")))?;
         let source_record = validate_durable_ilm_record(path, &source)
             .map_err(|err| Error::other(format!("source durable ILM record is invalid at path `{path}`: {err}")))?;
-        let (target_pool_idx, target) = self
+        let target = self
             .load_decommissioned_durable_ilm_target(source_pool_idx, path, namespace.max_record_size, &source_record.context())
-            .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?;
+        if let Some((target_pool_idx, target)) = target {
+            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?;
+        } else {
+            self.load_decommission_durable_ilm_terminal_receipt(source_pool_idx, path, &source_record)
+                .await?
+                .ok_or_else(|| {
+                    Error::other(format!(
+                        "target durable ILM record is missing at path `{path}` {} without a matching terminal receipt",
+                        source_record.context()
+                    ))
+                })?;
+        }
 
         let cleanup_result = data_movement::cleanup_source_entry_if_unchanged(
             source_set,
@@ -6125,10 +6341,12 @@ pub(crate) fn fallback_free_capacity_dedup(disks: &[rustfs_madmin::Disk]) -> usi
 mod pools_tests {
     use super::{
         DECOMMISSION_META_PREFIXES, DECOMMISSION_PROGRESS_SAVE_INTERVAL, DECOMMISSION_PROGRESS_SAVE_ITEM_THRESHOLD,
-        DecomBucketInfo, DecommissionStartPoolState, DecommissionTerminalState, ListCallback, PoolDecommissionInfo, PoolMeta,
-        PoolSpaceInfo, PoolStatus, apply_decommission_status_space_info, bind_decommission_cancelers,
-        bind_missing_decommission_cancelers, cancel_decommission_canceler, classify_decommission_terminal_state,
-        count_decommission_item, decommission_cancel_signal_result, decommission_item_size, decommission_meta_bucket_options,
+        DecomBucketInfo, DecommissionDurableIlmReceipt, DecommissionStartPoolState, DecommissionTerminalState, ListCallback,
+        PoolDecommissionInfo, PoolMeta, PoolSpaceInfo, PoolStatus, apply_decommission_status_space_info,
+        bind_decommission_cancelers, bind_missing_decommission_cancelers, cancel_decommission_canceler,
+        classify_decommission_terminal_state, count_decommission_item, decommission_cancel_signal_result,
+        decommission_durable_ilm_receipt_path, decommission_durable_ilm_receipt_run_prefix,
+        decommission_durable_ilm_receipt_run_token, decommission_item_size, decommission_meta_bucket_options,
         decommission_start_pool_state, dedup_indices, default_decommission_bucket_concurrency,
         ensure_decommission_cancel_allowed, ensure_decommission_clear_allowed, ensure_decommission_listing_disks_available,
         ensure_decommission_not_rebalancing, ensure_decommission_start_allowed, ensure_decommission_start_keeps_active_pool,
@@ -6138,9 +6356,9 @@ mod pools_tests {
         ensure_valid_decommission_pool_index, first_resumable_decommission_queue_indices, get_by_index,
         has_active_decommission_canceler, is_decommission_active, is_decommission_cancel_requested,
         load_decommission_entry_versions, local_decommission_queue_prefix, mark_decommission_bucket_done,
-        merge_pool_status_refresh, missing_decommission_worker_prefix, observe_decommission_terminal_reload_result,
-        pool_meta_has_active_decommission, reconcile_decommission_meta_buckets, require_decommission_store,
-        resolve_decommission_bucket_done_save_result, resolve_decommission_bucket_state,
+        merge_decommission_durable_ilm_receipts, merge_pool_status_refresh, missing_decommission_worker_prefix,
+        observe_decommission_terminal_reload_result, pool_meta_has_active_decommission, reconcile_decommission_meta_buckets,
+        require_decommission_store, resolve_decommission_bucket_done_save_result, resolve_decommission_bucket_state,
         resolve_decommission_check_after_list_result, resolve_decommission_entry_cleanup_delete_result,
         resolve_decommission_entry_exact_versions, resolve_decommission_entry_reload_result,
         resolve_decommission_listing_worker_result, resolve_decommission_optional_bucket_config_result,
@@ -6157,6 +6375,7 @@ mod pools_tests {
         validate_start_decommission_request, wait_decommission_listing_retry, wait_decommission_worker_drain,
         with_decommission_entry_context,
     };
+    use crate::bucket::lifecycle::DurableIlmRecordCheckpoint;
     use crate::data_movement;
     use crate::disk::endpoint::Endpoint;
     use crate::error::{Error, StorageError};
@@ -6203,6 +6422,59 @@ mod pools_tests {
         }
     }
 
+    #[test]
+    fn decommission_receipt_run_token_changes_with_persisted_start_time() {
+        let first = OffsetDateTime::from_unix_timestamp(1_000).expect("first run timestamp should be valid");
+        let second = OffsetDateTime::from_unix_timestamp(2_000).expect("second run timestamp should be valid");
+        let first_token = decommission_durable_ilm_receipt_run_token("pool-0", first);
+        let second_token = decommission_durable_ilm_receipt_run_token("pool-0", second);
+
+        assert_ne!(first_token, second_token);
+        assert_eq!(first_token, decommission_durable_ilm_receipt_run_token("pool-0", first));
+        let operation_id = "a".repeat(64);
+        let old_receipt = decommission_durable_ilm_receipt_path(
+            &first_token,
+            &format!("ilm/tier-delete-journal/{operation_id}.json"),
+            "operation_id",
+            &operation_id,
+        );
+        assert!(!old_receipt.starts_with(&decommission_durable_ilm_receipt_run_prefix(&second_token)));
+    }
+
+    #[test]
+    fn decommission_receipt_merge_preserves_terminal_proof() {
+        let operation_id = "a".repeat(64);
+        let source_path = format!("ilm/tier-delete-journal/{operation_id}.json");
+        let checkpoint = DurableIlmRecordCheckpoint::TierDeleteJournal {
+            content_sha256: "b".repeat(64),
+            identity_sha256: "c".repeat(64),
+            committed: false,
+        };
+        let terminal_checkpoint = DurableIlmRecordCheckpoint::TierDeleteJournal {
+            content_sha256: "d".repeat(64),
+            identity_sha256: "c".repeat(64),
+            committed: true,
+        };
+        let incoming = DecommissionDurableIlmReceipt {
+            source_path,
+            namespace: "tier-delete-journal".to_string(),
+            id_kind: "operation_id".to_string(),
+            id: operation_id,
+            checkpoint: checkpoint.clone(),
+            terminal_checkpoint: None,
+        };
+        let existing = DecommissionDurableIlmReceipt {
+            terminal_checkpoint: Some(terminal_checkpoint.clone()),
+            ..incoming.clone()
+        };
+
+        let merged = merge_decommission_durable_ilm_receipts(&existing, &incoming)
+            .expect("retry receipt must merge with a terminal receipt");
+
+        assert_eq!(merged.checkpoint, checkpoint);
+        assert_eq!(merged.terminal_checkpoint, Some(terminal_checkpoint));
+    }
+
     #[test]
     fn test_apply_decommission_status_space_info_adds_idle_pool_usage() {
         let status = apply_decommission_status_space_info(

crates/ecstore/src/store/init.rs

@@ -585,6 +585,7 @@ mod tests {
         client::transition_api::ReaderImpl,
         config::com,
         core::pools::DecomBucketInfo,
+        data_movement::SourceCleanupDeleteBarrier,
         disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET, STORAGE_FORMAT_FILE},
         runtime::{global::set_object_store_resolver, sources as runtime_sources},
         services::tier::{
@@ -3046,6 +3047,181 @@ mod tests {
         ));
     }
 
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn decommission_durable_ilm_target_read_error_is_not_masked_by_peer_success() {
+        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(), "durable-ilm-target-read-error", &[4, 4, 4]))
+                .await;
+        let job_id = uuid::Uuid::new_v4();
+        let job =
+            ManualTransitionJobRecord::new(job_id, "manual-target-read-error", &ManualTransitionRunOptions::default(), "owner");
+        let path = manual_transition_job_record_object_name(job_id).expect("manual job path should build");
+        let data = job.encode().expect("manual job should encode");
+        for pool in &store.pools {
+            com::save_config(pool.clone(), &path, data.clone())
+                .await
+                .expect("manual job fixture should persist in every pool");
+        }
+        store.pool_meta.write().await.pools[0].decommission = Some(PoolDecommissionInfo {
+            start_time: Some(OffsetDateTime::now_utc()),
+            ..Default::default()
+        });
+
+        let failing_target = store.pools[2].get_disks_by_key(&path);
+        let original_disks = {
+            let mut disks = failing_target.disks.write().await;
+            let original = disks.clone();
+            for disk in disks.iter_mut().take(3) {
+                *disk = None;
+            }
+            original
+        };
+        let error = store
+            .verify_and_cleanup_decommissioned_durable_ilm_record_for_test(0, store.pools[0].get_disks_by_key(&path), &path)
+            .await
+            .expect_err("one target read-quorum error must fail closed despite another target success")
+            .to_string();
+        *failing_target.disks.write().await = original_disks;
+
+        assert!(error.contains(&path));
+        assert!(error.contains("pool 2"));
+        assert_eq!(
+            com::read_config(store.pools[0].clone(), &path)
+                .await
+                .expect("target read error must retain the source"),
+            data
+        );
+        assert_eq!(
+            store
+                .decommission_durable_ilm_receipt_count_for_test(0)
+                .await
+                .expect("failed target verification should not create a receipt"),
+            0
+        );
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn decommission_durable_ilm_terminal_receipt_recovers_failed_source_cleanup() {
+        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(), "durable-ilm-terminal-receipt", &[4, 4])).await;
+        let tier_name = "DECOMMISSION-RECEIPT";
+        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")
+            .backend_identity();
+        let entry = Jentry {
+            obj_name: "receipt-recovery-object".to_string(),
+            version_id: "receipt-recovery-version".to_string(),
+            tier_name: tier_name.to_string(),
+            backend_identity: Some(backend_identity),
+            version_id_exact: true,
+            version_state: rustfs_filemeta::TransitionVersionState::Exact,
+            state: TierDeleteJournalState::Committed,
+            source: None,
+        };
+        let path = tier_delete_journal_object_name(&entry);
+        let data = encode_tier_delete_journal_entry(&entry).expect("tier journal should encode");
+        com::save_config(store.pools[0].clone(), &path, data.clone())
+            .await
+            .expect("source tier journal should persist");
+        com::save_config(store.pools[1].clone(), &path, data.clone())
+            .await
+            .expect("target tier journal should persist");
+        let active_pool_meta = {
+            let mut pool_meta = store.pool_meta.write().await;
+            pool_meta.pools[0].decommission = Some(PoolDecommissionInfo {
+                start_time: Some(OffsetDateTime::now_utc()),
+                ..Default::default()
+            });
+            pool_meta.clone()
+        };
+        active_pool_meta
+            .save(store.pools.clone())
+            .await
+            .expect("active decommission run identity should persist");
+
+        let source_set = store.pools[0].get_disks_by_key(&path);
+        let barrier = SourceCleanupDeleteBarrier::install(RUSTFS_META_BUCKET, &path);
+        let cleanup_store = store.clone();
+        let cleanup_set = source_set.clone();
+        let cleanup_path = path.clone();
+        let cleanup = tokio::spawn(async move {
+            cleanup_store
+                .verify_and_cleanup_decommissioned_durable_ilm_record_for_test(0, cleanup_set, &cleanup_path)
+                .await
+        });
+        barrier.wait_until_paused().await;
+        let original_source_disks = {
+            let mut disks = source_set.disks.write().await;
+            let original = disks.clone();
+            for disk in disks.iter_mut().take(3) {
+                *disk = None;
+            }
+            original
+        };
+        barrier.release();
+        let cleanup_error = cleanup
+            .await
+            .expect("source cleanup task should not panic")
+            .expect_err("injected source delete quorum failure must fail cleanup")
+            .to_string();
+        *source_set.disks.write().await = original_source_disks;
+        drop(barrier);
+
+        assert!(cleanup_error.contains("source durable ILM cleanup failed"));
+        assert_eq!(
+            store
+                .decommission_durable_ilm_receipt_count_for_test(0)
+                .await
+                .expect("receipt should persist before source cleanup"),
+            1
+        );
+        assert_eq!(
+            com::read_config(store.pools[0].clone(), &path)
+                .await
+                .expect("failed cleanup must retain the source"),
+            data
+        );
+
+        let mut restarted_pool_meta = PoolMeta::default();
+        restarted_pool_meta
+            .load(store.pools[0].clone(), store.pools.clone())
+            .await
+            .expect("decommission run identity should reload after restart");
+        *store.pool_meta.write().await = restarted_pool_meta;
+        let stats = recover_tier_delete_journal_entries(store.clone(), 100, None)
+            .await
+            .expect("target recovery should commit terminal proof and delete the target");
+        assert_eq!((stats.scanned, stats.deleted, stats.failed), (1, 1, 0));
+        assert!(matches!(
+            com::read_config(store.pools[1].clone(), &path).await,
+            Err(Error::ConfigNotFound)
+        ));
+        assert_eq!(
+            com::read_config(store.pools[0].clone(), &path)
+                .await
+                .expect("target recovery must not delete the decommission source"),
+            data
+        );
+
+        store
+            .verify_and_cleanup_decommissioned_durable_ilm_record_for_test(0, source_set, &path)
+            .await
+            .expect("terminal receipt should authorize cleanup after target deletion");
+        assert!(matches!(
+            com::read_config(store.pools[0].clone(), &path).await,
+            Err(Error::ConfigNotFound)
+        ));
+        assert!(backend.remove_versions().await.contains(&(entry.obj_name, entry.version_id)));
+    }
+
     #[cfg(feature = "test-util")]
     #[tokio::test]
     #[serial_test::serial(storage_class_env)]