fix(ecstore): repair durable ILM receipt recovery

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

@@ -169,7 +169,7 @@ pub(crate) enum DurableIlmRecordCheckpoint {
     ManualTransitionJob {
         content_sha256: String,
         identity_sha256: String,
-        updated_at_unix_nanos: i128,
+        updated_at_unix_nanos: i64,
         state: manual_transition_job::ManualTransitionJobState,
         scan_completed: bool,
         cancel_requested: bool,
@@ -181,7 +181,7 @@ pub(crate) enum DurableIlmRecordCheckpoint {
     ManualTransitionScope {
         content_sha256: String,
         identity_sha256: String,
-        updated_at_unix_nanos: i128,
+        updated_at_unix_nanos: i64,
     },
     ManualTransitionTask {
         content_sha256: String,
@@ -817,13 +817,15 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
                 job.created_at_unix_nanos,
             ))?;
             let progress_proof = ManualTransitionJobProgressProof::new(&job.report, &job.queue_snapshot, job.cursor_revision)?;
+            let updated_at_unix_nanos = i64::try_from(job.updated_at_unix_nanos)
+                .map_err(|_| Error::other("manual transition job updated_at exceeds durable ILM checkpoint range"))?;
             (
                 "job_id",
                 job_id.to_string(),
                 DurableIlmRecordCheckpoint::ManualTransitionJob {
                     content_sha256,
                     identity_sha256,
-                    updated_at_unix_nanos: job.updated_at_unix_nanos,
+                    updated_at_unix_nanos,
                     state: job.state,
                     scan_completed: job.scan_completed,
                     cancel_requested: job.cancel_requested,
@@ -850,13 +852,15 @@ pub(crate) fn validate_durable_ilm_record(path: &str, data: &[u8]) -> Result<Val
                 &admission.tier,
                 admission.dry_run,
             ))?;
+            let updated_at_unix_nanos = i64::try_from(admission.updated_at_unix_nanos)
+                .map_err(|_| Error::other("manual transition scope updated_at exceeds durable ILM checkpoint range"))?;
             (
                 "job_id",
                 admission.job_id.to_string(),
                 DurableIlmRecordCheckpoint::ManualTransitionScope {
                     content_sha256,
                     identity_sha256,
-                    updated_at_unix_nanos: admission.updated_at_unix_nanos,
+                    updated_at_unix_nanos,
                 },
             )
         }
@@ -980,6 +984,22 @@ mod tests {
         assert_eq!(legacy.compacted().expect("legacy checkpoint should compact"), compact);
     }
 
+    #[test]
+    fn manual_transition_job_checkpoint_rejects_timestamp_outside_wire_range() {
+        let options = super::super::bucket_lifecycle_ops::ManualTransitionRunOptions::default();
+        let mut job = manual_transition_job::ManualTransitionJobRecord::new(
+            Uuid::new_v4(),
+            "checkpoint-timestamp-bucket",
+            &options,
+            "owner",
+        );
+        job.updated_at_unix_nanos = i128::from(i64::MAX) + 1;
+
+        let err = try_manual_job_checkpoint(&job).expect_err("out-of-range checkpoint timestamp must fail closed");
+
+        assert!(err.to_string().contains("updated_at exceeds durable ILM checkpoint range"));
+    }
+
     #[test]
     fn manual_transition_job_checkpoint_rejects_progress_poison() {
         let options = super::super::bucket_lifecycle_ops::ManualTransitionRunOptions::default();

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

@@ -433,16 +433,22 @@ 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)
+    let target_pool_indices = api
+        .record_durable_ilm_decommission_terminal_target_pools(&path, &data)
         .await
-        .map_err(std::io::Error::other)
+        .map_err(std::io::Error::other)?;
+    if !target_pool_indices.is_empty() {
+        for target_pool_idx in target_pool_indices {
+            match config_boundary::delete_config(api.pools[target_pool_idx].clone(), &path).await {
+                Ok(()) | Err(Error::ConfigNotFound) => {}
+                Err(err) => return Err(std::io::Error::other(err)),
+            }
+        }
+        return Ok(());
+    }
+    remove_tier_delete_journal_entry(api, je).await
 }
 
 fn object_info_references_tier_delete(info: &ObjectInfo, je: &Jentry) -> std::io::Result<bool> {

crates/ecstore/src/core/pools.rs

@@ -5478,7 +5478,7 @@ impl ECStore {
         }
     }
 
-    async fn advance_durable_ilm_decommission_receipts(&self, path: &str, data: &[u8], terminal: bool) -> Result<()> {
+    async fn advance_durable_ilm_decommission_receipts(&self, path: &str, data: &[u8], terminal: bool) -> Result<Vec<usize>> {
         let active_runs = {
             let pool_meta = self.pool_meta.read().await;
             pool_meta
@@ -5495,20 +5495,25 @@ impl ECStore {
                 .collect::<Vec<_>>()
         };
         if active_runs.is_empty() {
-            return Ok(());
+            return Ok(Vec::new());
         }
 
         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}")))?;
+        let mut terminal_target_pool_indices = Vec::new();
         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
+                    let found = self
                         .advance_durable_ilm_decommission_receipt(pool_idx, &receipt_path, &record, terminal)
                         .await?;
+                    receipt_found |= found;
+                    if terminal && found && !terminal_target_pool_indices.contains(&pool_idx) {
+                        terminal_target_pool_indices.push(pool_idx);
+                    }
                 }
             }
             if terminal && !receipt_found {
@@ -5518,14 +5523,27 @@ impl ECStore {
                 )));
             }
         }
-        Ok(())
+        Ok(terminal_target_pool_indices)
     }
 
     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
+        self.advance_durable_ilm_decommission_receipts(path, data, false)
+            .await
+            .map(|_| ())
     }
 
     pub(crate) async fn record_durable_ilm_decommission_terminal(&self, path: &str, data: &[u8]) -> Result<()> {
+        self.record_durable_ilm_decommission_terminal_target_pools(path, data)
+            .await
+            .map(|_| ())
+    }
+
+    /// Record terminal proof and return its non-source receipt pools for targeted cleanup.
+    pub(crate) async fn record_durable_ilm_decommission_terminal_target_pools(
+        &self,
+        path: &str,
+        data: &[u8],
+    ) -> Result<Vec<usize>> {
         self.advance_durable_ilm_decommission_receipts(path, data, true).await
     }
 
@@ -7021,6 +7039,7 @@ mod pools_tests {
         let job_bytes = job.encode().expect("large manual job should remain within its record limit");
         assert!(job_bytes.len() > DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE);
         let record = validate_durable_ilm_record(&path, &job_bytes).expect("large manual job should validate");
+        let expected_checkpoint = record.checkpoint.clone();
         let mut receipt = DecommissionDurableIlmReceipt::new(&path, &record);
         receipt.terminal_checkpoint = Some(record.checkpoint);
 
@@ -7029,7 +7048,8 @@ mod pools_tests {
 
         assert!(encoded.len() <= DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE);
         assert_eq!(decoded.source_path, path);
-        assert!(decoded.terminal_checkpoint.is_some());
+        assert_eq!(decoded.checkpoint, expected_checkpoint);
+        assert_eq!(decoded.terminal_checkpoint, Some(expected_checkpoint));
     }
 
     #[test]