fix(ecstore): anchor decommission ILM receipts

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

@@ -387,17 +387,13 @@ fn manual_job_cursor_reaches(
         (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))) = (
+            let (Ok((Some(previous_marker), _)), Ok((Some(next_marker), _))) = (
                 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,
     }
@@ -665,12 +661,16 @@ mod tests {
             .checkpoint
     }
 
-    fn continuation_token(marker: &str) -> String {
-        let encoded = serde_json::to_vec(&serde_json::json!({ "marker": marker, "version_marker": null }))
+    fn continuation_token_with_version(marker: &str, version_marker: Option<&str>) -> String {
+        let encoded = serde_json::to_vec(&serde_json::json!({ "marker": marker, "version_marker": version_marker }))
             .expect("continuation token should encode");
         base64_simd::URL_SAFE_NO_PAD.encode_to_string(&encoded)
     }
 
+    fn continuation_token(marker: &str) -> String {
+        continuation_token_with_version(marker, None)
+    }
+
     #[test]
     fn unknown_ilm_record_requires_namespace_registration() {
         let err = classify_durable_ilm_record("ilm/future-durable/jobs/one.json")
@@ -767,6 +767,22 @@ mod tests {
                 .is_err()
         );
 
+        let mut same_marker_version_previous = previous.clone();
+        same_marker_version_previous.report.continuation_token =
+            Some(continuation_token_with_version("logs/b", Some("opaque-newer-version")));
+        let same_marker_version_previous_checkpoint = manual_job_checkpoint(&same_marker_version_previous);
+        let mut same_marker_version_rollback = same_marker_version_previous.clone();
+        same_marker_version_rollback.updated_at_unix_nanos += 1;
+        same_marker_version_rollback.report.scanned += 1;
+        same_marker_version_rollback.report.continuation_token =
+            Some(continuation_token_with_version("logs/b", Some("opaque-stale-version")));
+        assert!(
+            same_marker_version_previous_checkpoint
+                .validate_successor(&manual_job_checkpoint(&same_marker_version_rollback))
+                .is_err(),
+            "opaque version markers must not be treated as ordered progress"
+        );
+
         let mut worker_result_rollback = next.clone();
         worker_result_rollback.updated_at_unix_nanos += 1;
         worker_result_rollback.report.transition_completed = 0;

crates/ecstore/src/core/pools.rs

@@ -104,8 +104,11 @@ 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_MANIFEST_ROOT: &str = "decommission/ilm-manifests";
 const DECOMMISSION_DURABLE_ILM_RECEIPT_SCHEMA: &str = "v2";
+const DECOMMISSION_DURABLE_ILM_MANIFEST_SCHEMA: &str = "v1";
 const DECOMMISSION_DURABLE_ILM_RECEIPT_MAX_SIZE: usize = 16 * 1024;
+const DECOMMISSION_DURABLE_ILM_MANIFEST_MAX_SIZE: usize = 4 * 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.
@@ -1046,6 +1049,72 @@ struct PersistedDecommissionDurableIlmReceipt {
     receipt: DecommissionDurableIlmReceipt,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct DecommissionDurableIlmManifest {
+    schema: String,
+    run_token: String,
+    receipt_count: u64,
+    receipt_paths_sha256: String,
+}
+
+impl DecommissionDurableIlmManifest {
+    fn new(run_token: &str, receipt_paths: &[String]) -> Result<Self> {
+        let manifest = Self {
+            schema: DECOMMISSION_DURABLE_ILM_MANIFEST_SCHEMA.to_string(),
+            run_token: run_token.to_string(),
+            receipt_count: u64::try_from(receipt_paths.len())
+                .map_err(|_| Error::other("durable ILM expected manifest receipt count exceeds u64"))?,
+            receipt_paths_sha256: decommission_durable_ilm_manifest_paths_sha256(receipt_paths)?,
+        };
+        manifest.validate(run_token, receipt_paths)?;
+        Ok(manifest)
+    }
+
+    fn validate(&self, run_token: &str, receipt_paths: &[String]) -> Result<()> {
+        if self.schema != DECOMMISSION_DURABLE_ILM_MANIFEST_SCHEMA {
+            return Err(Error::other(format!(
+                "unsupported durable ILM expected manifest schema `{}`",
+                self.schema
+            )));
+        }
+        if self.run_token != run_token || !is_sha256_checksum(&self.run_token) {
+            return Err(Error::other("durable ILM expected manifest run token is invalid"));
+        }
+        let receipt_count = u64::try_from(receipt_paths.len())
+            .map_err(|_| Error::other("durable ILM expected manifest receipt count exceeds u64"))?;
+        if self.receipt_count != receipt_count {
+            return Err(Error::other(format!(
+                "durable ILM expected manifest receipt count mismatch: expected {}, found {receipt_count}",
+                self.receipt_count
+            )));
+        }
+        if !is_sha256_checksum(&self.receipt_paths_sha256)
+            || self.receipt_paths_sha256 != decommission_durable_ilm_manifest_paths_sha256(receipt_paths)?
+        {
+            return Err(Error::other("durable ILM expected manifest receipt paths checksum mismatch"));
+        }
+        Ok(())
+    }
+
+    fn encode(&self) -> Result<Vec<u8>> {
+        let encoded = serde_json::to_vec(self)?;
+        if encoded.len() > DECOMMISSION_DURABLE_ILM_MANIFEST_MAX_SIZE {
+            return Err(Error::other("durable ILM expected manifest exceeds maximum size"));
+        }
+        Ok(encoded)
+    }
+
+    fn decode(data: &[u8], run_token: &str, receipt_paths: &[String]) -> Result<Self> {
+        if data.len() > DECOMMISSION_DURABLE_ILM_MANIFEST_MAX_SIZE {
+            return Err(Error::other("durable ILM expected manifest exceeds maximum size"));
+        }
+        let manifest: Self = serde_json::from_slice(data)?;
+        manifest.validate(run_token, receipt_paths)?;
+        Ok(manifest)
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 struct DecommissionDurableIlmReceiptLocator {
     run_token: String,
@@ -1079,6 +1148,17 @@ fn decommission_durable_ilm_receipt_path(run_token: &str, source_path: &str, id_
     )
 }
 
+fn decommission_durable_ilm_manifest_path(run_token: &str) -> String {
+    format!("{DECOMMISSION_DURABLE_ILM_MANIFEST_ROOT}/{run_token}.json")
+}
+
+fn decommission_durable_ilm_manifest_paths_sha256(receipt_paths: &[String]) -> Result<String> {
+    let mut sorted_paths = receipt_paths.iter().map(String::as_str).collect::<Vec<_>>();
+    sorted_paths.sort_unstable();
+    let encoded = serde_json::to_vec(&sorted_paths)?;
+    Ok(hex_sha256(&encoded, ToOwned::to_owned))
+}
+
 fn parse_decommission_durable_ilm_receipt_path(path: &str) -> Result<DecommissionDurableIlmReceiptLocator> {
     let prefix = format!("{DECOMMISSION_DURABLE_ILM_RECEIPT_ROOT}/");
     let suffix = path
@@ -4795,6 +4875,131 @@ impl ECStore {
         Ok(receipts)
     }
 
+    async fn list_decommission_durable_ilm_manifest_receipts(&self, source_pool_idx: usize) -> Result<Vec<String>> {
+        let run_token = self.durable_ilm_receipt_run_token(source_pool_idx).await?;
+        let prefix = decommission_durable_ilm_receipt_run_prefix(&run_token);
+        let receipt_paths = self
+            .list_decommission_durable_ilm_receipt_paths_in_pool(source_pool_idx, &prefix)
+            .await?;
+        for receipt_path in &receipt_paths {
+            let locator = parse_decommission_durable_ilm_receipt_path(receipt_path)?;
+            if locator.run_token != run_token {
+                return Err(Error::other(format!(
+                    "durable ILM expected manifest receipt path `{receipt_path}` has an unexpected run token"
+                )));
+            }
+        }
+        Ok(receipt_paths)
+    }
+
+    async fn persist_decommission_durable_ilm_manifest(&self, source_pool_idx: usize) -> Result<()> {
+        let run_token = self.durable_ilm_receipt_run_token(source_pool_idx).await?;
+        let receipt_paths = self.list_decommission_durable_ilm_manifest_receipts(source_pool_idx).await?;
+        for receipt_path in &receipt_paths {
+            self.read_decommission_durable_ilm_receipt(source_pool_idx, receipt_path)
+                .await?;
+        }
+        let manifest = DecommissionDurableIlmManifest::new(&run_token, &receipt_paths)?;
+        let manifest_path = decommission_durable_ilm_manifest_path(&run_token);
+        let encoded = manifest.encode()?;
+        let mut attempt = 1;
+        loop {
+            match read_config_limited_preserve_empty(
+                self.pools[source_pool_idx].clone(),
+                &manifest_path,
+                DECOMMISSION_DURABLE_ILM_MANIFEST_MAX_SIZE,
+            )
+            .await
+            {
+                Ok(existing) => {
+                    DecommissionDurableIlmManifest::decode(&existing, &run_token, &receipt_paths).map_err(|err| {
+                        Error::other(format!(
+                            "durable ILM expected manifest `{manifest_path}` in source pool {source_pool_idx} is invalid: {err}"
+                        ))
+                    })?;
+                    return Ok(());
+                }
+                Err(err)
+                    if matches!(&err, Error::ConfigNotFound | Error::FileNotFound | Error::FileVersionNotFound)
+                        || is_err_object_not_found(&err)
+                        || is_err_version_not_found(&err) => {}
+                Err(err) => {
+                    return Err(Error::other(format!(
+                        "failed to read durable ILM expected manifest `{manifest_path}` from source pool {source_pool_idx}: {err}"
+                    )));
+                }
+            }
+            match save_config_with_opts(
+                self.pools[source_pool_idx].clone(),
+                &manifest_path,
+                encoded.clone(),
+                &ObjectOptions {
+                    max_parity: true,
+                    http_preconditions: Some(HTTPPreconditions {
+                        if_none_match: Some("*".to_string()),
+                        ..Default::default()
+                    }),
+                    ..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 expected manifest `{manifest_path}` after concurrent updates"
+                    )));
+                }
+                Err(err) => {
+                    return Err(Error::other(format!(
+                        "failed to persist durable ILM expected manifest `{manifest_path}` in source pool {source_pool_idx}: {err}"
+                    )));
+                }
+            }
+        }
+    }
+
+    async fn load_decommission_durable_ilm_manifest(
+        &self,
+        source_pool_idx: usize,
+    ) -> Result<HashMap<String, DecommissionDurableIlmReceipt>> {
+        let run_token = self.durable_ilm_receipt_run_token(source_pool_idx).await?;
+        let receipt_paths = self.list_decommission_durable_ilm_manifest_receipts(source_pool_idx).await?;
+        let manifest_path = decommission_durable_ilm_manifest_path(&run_token);
+        let data = read_config_limited_preserve_empty(
+            self.pools[source_pool_idx].clone(),
+            &manifest_path,
+            DECOMMISSION_DURABLE_ILM_MANIFEST_MAX_SIZE,
+        )
+        .await
+        .map_err(|err| {
+            Error::other(format!(
+                "failed to read durable ILM expected manifest `{manifest_path}` from source pool {source_pool_idx}: {err}"
+            ))
+        })?;
+        DecommissionDurableIlmManifest::decode(&data, &run_token, &receipt_paths).map_err(|err| {
+            Error::other(format!(
+                "durable ILM expected manifest `{manifest_path}` in source pool {source_pool_idx} is invalid: {err}"
+            ))
+        })?;
+
+        let mut receipts = HashMap::with_capacity(receipt_paths.len());
+        for receipt_path in receipt_paths {
+            let receipt = self
+                .read_decommission_durable_ilm_receipt(source_pool_idx, &receipt_path)
+                .await?;
+            if receipts.insert(receipt_path.clone(), receipt).is_some() {
+                return Err(Error::other(format!(
+                    "durable ILM expected manifest contains duplicate receipt path `{receipt_path}`"
+                )));
+            }
+        }
+        Ok(receipts)
+    }
+
     async fn persist_decommission_durable_ilm_receipt(
         &self,
         source_pool_idx: usize,
@@ -5013,10 +5218,69 @@ impl ECStore {
     }
 
     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 expected_receipts = self.load_decommission_durable_ilm_manifest(source_pool_idx).await?;
+        let receipt_paths = self.list_decommission_durable_ilm_receipts(source_pool_idx).await?;
+        let present_receipt_paths = receipt_paths
+            .iter()
+            .map(|(_, receipt_path)| receipt_path.as_str())
+            .collect::<HashSet<_>>();
+        for (expected_path, expected) in &expected_receipts {
+            if !present_receipt_paths.contains(expected_path.as_str()) {
+                return Err(Error::other(format!(
+                    "durable ILM decommission receipt is missing at `{expected_path}` for source path `{}` {}",
+                    expected.source_path,
+                    expected.context()
+                )));
+            }
+        }
+
+        for (receipt_pool_idx, receipt_path) in receipt_paths {
+            let expected = expected_receipts.get(&receipt_path).ok_or_else(|| {
+                Error::other(format!(
+                    "durable ILM decommission receipt `{receipt_path}` in pool {receipt_pool_idx} is absent from the expected manifest"
+                ))
+            })?;
             let receipt = self
                 .read_decommission_durable_ilm_receipt(receipt_pool_idx, &receipt_path)
                 .await?;
+            if receipt.source_path != expected.source_path
+                || receipt.namespace != expected.namespace
+                || receipt.id_kind != expected.id_kind
+                || receipt.id != expected.id
+            {
+                return Err(Error::other(format!(
+                    "durable ILM decommission receipt identity mismatch at `{receipt_path}` for source path `{}` {}; decoded {}",
+                    expected.source_path,
+                    expected.context(),
+                    receipt.context()
+                )));
+            }
+            expected.checkpoint.validate_successor(&receipt.checkpoint).map_err(|err| {
+                Error::other(format!(
+                    "durable ILM decommission receipt generation mismatch at `{receipt_path}` for source path `{}` {}: {err}",
+                    expected.source_path,
+                    expected.context()
+                ))
+            })?;
+            match (&expected.terminal_checkpoint, &receipt.terminal_checkpoint) {
+                (Some(expected_terminal), Some(receipt_terminal)) => {
+                    expected_terminal.validate_successor(receipt_terminal).map_err(|err| {
+                        Error::other(format!(
+                            "durable ILM decommission terminal receipt generation mismatch at `{receipt_path}` for source path `{}` {}: {err}",
+                            expected.source_path,
+                            expected.context()
+                        ))
+                    })?;
+                }
+                (Some(_), None) => {
+                    return Err(Error::other(format!(
+                        "durable ILM decommission terminal receipt is missing at `{receipt_path}` for source path `{}` {}",
+                        expected.source_path,
+                        expected.context()
+                    )));
+                }
+                (None, _) => {}
+            }
             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
@@ -5266,6 +5530,28 @@ impl ECStore {
                 }
             }
         }
+        for receipt_path in self.list_decommission_durable_ilm_manifest_receipts(source_pool_idx).await? {
+            match delete_config(self.pools[source_pool_idx].clone(), &receipt_path).await {
+                Ok(()) | Err(Error::ConfigNotFound | Error::FileNotFound | Error::FileVersionNotFound) => {}
+                Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => {}
+                Err(err) => {
+                    return Err(Error::other(format!(
+                        "failed to clean durable ILM expected manifest receipt `{receipt_path}` from source pool {source_pool_idx}: {err}"
+                    )));
+                }
+            }
+        }
+        let run_token = self.durable_ilm_receipt_run_token(source_pool_idx).await?;
+        let manifest_path = decommission_durable_ilm_manifest_path(&run_token);
+        match delete_config(self.pools[source_pool_idx].clone(), &manifest_path).await {
+            Ok(()) | Err(Error::ConfigNotFound | Error::FileNotFound | Error::FileVersionNotFound) => {}
+            Err(err) if is_err_object_not_found(&err) || is_err_version_not_found(&err) => {}
+            Err(err) => {
+                return Err(Error::other(format!(
+                    "failed to clean durable ILM expected manifest `{manifest_path}` from source pool {source_pool_idx}: {err}"
+                )));
+            }
+        }
         Ok(())
     }
 
@@ -5290,11 +5576,12 @@ impl ECStore {
         let target = self
             .load_decommissioned_durable_ilm_target(source_pool_idx, path, namespace.max_record_size, &source_record.context())
             .await?;
-        if let Some((target_pool_idx, target)) = target {
+        let manifest_receipt = 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?;
+            receipt
         } else {
             self.load_decommission_durable_ilm_terminal_receipt(source_pool_idx, path, &source_record)
                 .await?
@@ -5303,8 +5590,10 @@ impl ECStore {
                         "target durable ILM record is missing at path `{path}` {} without a matching terminal receipt",
                         source_record.context()
                     ))
-                })?;
-        }
+                })?
+        };
+        self.persist_decommission_durable_ilm_receipt(source_pool_idx, source_pool_idx, &manifest_receipt)
+            .await?;
 
         let cleanup_result = data_movement::cleanup_source_entry_if_unchanged(
             source_set,
@@ -5520,6 +5809,7 @@ impl ECStore {
             }
         }
 
+        self.persist_decommission_durable_ilm_manifest(idx).await?;
         self.verify_decommission_durable_ilm_receipts(idx).await?;
 
         Ok(())

crates/ecstore/src/store/init.rs

@@ -3568,6 +3568,19 @@ mod tests {
             .expect("pool metadata should reload after the simulated pre-complete crash");
         *store.pool_meta.write().await = crash_restarted_pool_meta;
 
+        let (manual_job_receipt_pool, manual_job_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_job_path))
+            .expect("manual job should have one target receipt");
+        let manual_job_receipt_bytes = com::read_config(store.pools[manual_job_receipt_pool].clone(), &manual_job_receipt_path)
+            .await
+            .expect("manual job receipt should be readable before deletion");
+        com::delete_config(store.pools[manual_job_receipt_pool].clone(), &manual_job_receipt_path)
+            .await
+            .expect("manual job receipt should delete after source cleanup");
         com::delete_config(store.pools[1].clone(), &manual_job_path)
             .await
             .expect("post-crash target manual job should delete");
@@ -3576,6 +3589,7 @@ mod tests {
             .await
             .expect_err("completion must reject a missing target after source cleanup and restart")
             .to_string();
+        assert!(missing_after_crash.contains("receipt"));
         assert!(missing_after_crash.contains(&manual_job_path));
         assert!(missing_after_crash.contains(&manual_job_id.to_string()));
         assert!(
@@ -3588,6 +3602,13 @@ mod tests {
         com::save_config(store.pools[1].clone(), &manual_job_path, manual_job_bytes.clone())
             .await
             .expect("post-crash target manual job should restore");
+        com::save_config(
+            store.pools[manual_job_receipt_pool].clone(),
+            &manual_job_receipt_path,
+            manual_job_receipt_bytes,
+        )
+        .await
+        .expect("manual job receipt should restore after the missing-receipt check");
 
         com::save_config(store.pools[1].clone(), &transaction_path, b"{corrupt".to_vec())
             .await