test(scanner): preserve checkpoint debt across process restarts

crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs

@@ -44,27 +44,37 @@ fn scan_options(scan_mode: HealScanMode) -> ScannerDiskScanOptions {
 /// Real cache persistence codec and CAS calls, backed by two bounded local files.
 #[derive(Debug)]
 struct FixtureStore {
-    root: tempfile::TempDir,
+    root: std::path::PathBuf,
+    _owned_root: Option<tempfile::TempDir>,
     reject_save: AtomicBool,
 }
 
 impl FixtureStore {
     fn new() -> Arc<Self> {
+        let root = tempfile::tempdir().expect("checkpoint fixture storage directory");
         Arc::new(Self {
-            root: tempfile::tempdir().expect("checkpoint fixture storage directory"),
+            root: root.path().to_path_buf(),
+            _owned_root: Some(root),
+            reject_save: AtomicBool::new(false),
+        })
+    }
+
+    fn reopen(root: std::path::PathBuf) -> Arc<Self> {
+        assert!(root.is_dir(), "reopen only an existing checkpoint fixture directory");
+        Arc::new(Self {
+            root,
+            _owned_root: None,
             reject_save: AtomicBool::new(false),
         })
     }
 
     fn path(&self, object: &str) -> std::path::PathBuf {
         assert!(object.ends_with(CACHE_NAME) || object.ends_with(&format!("{CACHE_NAME}.bkp")));
-        self.root
-            .path()
-            .join(if object.ends_with(".bkp") { "backup" } else { "main" })
+        self.root.join(if object.ends_with(".bkp") { "backup" } else { "main" })
     }
 
     async fn strict_load(&self) -> DataUsageCache {
-        let bytes = tokio::fs::read(self.root.path().join("main"))
+        let bytes = tokio::fs::read(self.root.join("main"))
             .await
             .expect("saved checkpoint fixture must exist");
         decode_fixture(&bytes).expect("saved checkpoint fixture must contain a valid bucket root")
@@ -1320,7 +1330,7 @@ where
         .load_with_revisions(store.clone(), CACHE_NAME)
         .await
         .expect("load final checkpoint");
-    let before = tokio::fs::read(store.root.path().join("main"))
+    let before = tokio::fs::read(store.root.join("main"))
         .await
         .expect("read durable checkpoint bytes");
     let epoch_error = loaded
@@ -1335,7 +1345,7 @@ where
         .await
         .expect_err("injected save failure must not report durable progress");
     assert_eq!(
-        tokio::fs::read(store.root.path().join("main"))
+        tokio::fs::read(store.root.join("main"))
             .await
             .expect("read unchanged checkpoint bytes"),
         before

crates/scanner/src/scanner_folder/tests/checkpoint_fixture/periodic_snapshot.rs

@@ -14,7 +14,10 @@
 
 //! Periodic snapshots captured while nested traversal frames are still active.
 
+use super::super::enumeration_restart::{REQUEST_ENV, read_bounded};
 use super::*;
+use sha2::{Digest, Sha256};
+use std::path::{Path, PathBuf};
 use tokio::sync::Notify;
 
 const OBJECTS: u64 = 8;
@@ -176,7 +179,10 @@ async fn nested_periodic_snapshot(
     (loaded, identity, scanner, guard)
 }
 
-async fn finish_resumed_walk(cache: DataUsageCache, original: &FolderScanner) -> (DataUsageCache, u64, TestGuard) {
+async fn finish_resumed_walk_scanner(
+    cache: DataUsageCache,
+    original: &FolderScanner,
+) -> (FolderScanner, DataUsageEntry, TestGuard) {
     let (mut scanner, unused_root) = build_test_scanner().await;
     let guard = TestGuard {
         temp_dir: Some(unused_root),
@@ -201,6 +207,11 @@ async fn finish_resumed_walk(cache: DataUsageCache, original: &FolderScanner) ->
         )
         .await
         .expect("finish resumed nested periodic walk");
+    (scanner, entry, guard)
+}
+
+async fn finish_resumed_walk(cache: DataUsageCache, original: &FolderScanner) -> (DataUsageCache, u64, TestGuard) {
+    let (scanner, _entry, guard) = finish_resumed_walk_scanner(cache, original).await;
     let visits = scanner.budget.progress().0;
     (scanner.new_cache, visits, guard)
 }
@@ -972,27 +983,306 @@ async fn periodic_failed_subtree_resumes_healthy_descendants_and_compacted_debt(
     }
 }
 
+const DEBT_FRONTIER: &str = "bucket/prefix/a中";
+const DEBT_COMPLETED_FRONTIER: &str = "bucket/prefix";
+
+#[derive(serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+struct CheckpointDebtRequest {
+    workspace: PathBuf,
+    round: u32,
+}
+
+#[derive(serde::Serialize, serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+struct CheckpointDebtState {
+    writer_pid: u32,
+    root_id: Uuid,
+    disk_root: PathBuf,
+    cache_root: PathBuf,
+    identity: crate::DataUsageScanIdentity,
+}
+
+fn checkpoint_debt_observation(
+    cache: &DataUsageCache,
+    identity: crate::DataUsageScanIdentity,
+    frontier: &str,
+    failed_objects: usize,
+) -> serde_json::Value {
+    assert_eq!(cache.validated_scan_frontier(), Some(frontier));
+    assert_eq!(cache.info.scan_identity, Some(identity));
+    assert_eq!(cache.info.source, Some(SOURCE));
+    assert!(cache.info.failed_objects.is_empty(), "entry debt must not depend on diagnostic records");
+    assert!(!cache.info.snapshot_complete, "this fixture does not certify a complete sweep");
+    let totals = cache
+        .checked_flatten_complete_scope("bucket")
+        .expect("connected checkpoint accounting");
+    let healthy = usize::try_from(OBJECTS).expect("healthy originals plus Unicode sibling") + usize::from(failed_objects == 0);
+    assert_eq!((totals.objects, totals.size, totals.failed_objects), (healthy, healthy, failed_objects));
+    serde_json::json!({
+        "frontier": cache.validated_scan_frontier(),
+        "receipt": cache.info.scan_coverage_receipt,
+        "leader_epoch": cache.info.leader_epoch,
+        "objects": totals.objects, "bytes": totals.size, "failed_objects": totals.failed_objects,
+        "diagnostics": cache.info.failed_objects.len(), "snapshot_complete": cache.info.snapshot_complete,
+    })
+}
+
+/// The existing restart driver runs this exact test in four sequential processes:
+/// writer, two leader handoffs, and a final handoff that repairs the original file.
+async fn checkpoint_debt_restart_round(request: &CheckpointDebtRequest) -> serde_json::Value {
+    assert!(request.round < 4);
+    let workspace = tokio::fs::canonicalize(&request.workspace)
+        .await
+        .expect("existing restart workspace");
+    assert_eq!(workspace, request.workspace);
+    assert_eq!(
+        tokio::fs::canonicalize(std::env::temp_dir())
+            .await
+            .expect("worker temporary root"),
+        workspace,
+        "the driver must own every fixture directory"
+    );
+    let state_path = workspace.join("checkpoint-state.json");
+    let (state, mut original, mut root_guard, prepared, mut visits) = if request.round == 0 {
+        assert!(!state_path.exists(), "the writer requires a fresh workspace");
+        let (mut loaded, identity, mut original, guard) = nested_periodic_snapshot(true, (0, 0)).await;
+        loaded.info.failed_objects.clear();
+        original.failed_object_ttl_secs = 0;
+        original.failed_objects_max = 0;
+        write_checkpoint_object(Path::new(&original.root), "prefix/a中", &[(None, 1)]).await;
+        let (mut cache, visits, _walk_guard) = finish_resumed_walk(loaded, &original).await;
+        assert_eq!(visits, 7, "the writer must finish the original partial walk and Unicode sibling");
+        cache
+            .seal_scan_frontier(Some(DEBT_FRONTIER))
+            .expect("seal completed Unicode sibling");
+        let state = CheckpointDebtState {
+            writer_pid: std::process::id(),
+            root_id: Uuid::new_v4(),
+            disk_root: tokio::fs::canonicalize(&original.root).await.expect("original disk root"),
+            cache_root: workspace.join("cache"),
+            identity,
+        };
+        assert!(state.disk_root.starts_with(&workspace) && state.disk_root != workspace);
+        tokio::fs::write(state.disk_root.join("checkpoint-root-id"), state.root_id.to_string())
+            .await
+            .expect("persist original fixture identity");
+        (state, original, guard, Some(cache), visits)
+    } else {
+        let state: CheckpointDebtState =
+            serde_json::from_slice(&read_bounded(&state_path).await).expect("original fixture state");
+        assert_ne!(state.writer_pid, std::process::id(), "handoff must run after the writer process exits");
+        let (mut original, unused_root) = build_test_scanner().await;
+        let guard = TestGuard {
+            temp_dir: Some(unused_root),
+        };
+        assert_eq!(
+            tokio::fs::canonicalize(&state.disk_root).await.expect("reopen original disk"),
+            state.disk_root
+        );
+        let endpoint = Endpoint::try_from(state.disk_root.to_string_lossy().as_ref()).expect("original fixture endpoint");
+        original.root = state.disk_root.to_string_lossy().to_string();
+        original.local_disk = new_disk(
+            &endpoint,
+            &DiskOption {
+                cleanup: false,
+                health_check: false,
+            },
+        )
+        .await
+        .expect("reopen original disk in a fresh process");
+        (state, original, guard, None, 0)
+    };
+    original.failed_object_ttl_secs = 0;
+    original.failed_objects_max = 0;
+    assert!(!state.root_id.is_nil());
+    assert!(state.disk_root.starts_with(&workspace) && state.disk_root != workspace);
+    assert_eq!(state.cache_root, workspace.join("cache"));
+    assert_eq!(
+        tokio::fs::canonicalize(&state.cache_root).await.expect("original cache root"),
+        state.cache_root
+    );
+    assert_eq!(
+        read_bounded(&state.disk_root.join("checkpoint-root-id")).await,
+        state.root_id.to_string().as_bytes(),
+        "each process must reopen the original fixture"
+    );
+    let store = FixtureStore::reopen(state.cache_root.clone());
+    let main_path = store.path(CACHE_NAME);
+    if request.round == 0 {
+        assert!(!main_path.exists(), "writer must not overwrite a previous checkpoint");
+    }
+    let mut cache = DataUsageCache::default();
+    let revisions = cache
+        .load_with_revisions(store.clone(), CACHE_NAME)
+        .await
+        .expect("reload actual checkpoint revisions");
+    let mut before = serde_json::Value::Null;
+    if let Some(prepared) = prepared {
+        cache = prepared;
+    } else {
+        let frontier = if request.round == 1 {
+            DEBT_FRONTIER
+        } else {
+            DEBT_COMPLETED_FRONTIER
+        };
+        before = checkpoint_debt_observation(&cache, state.identity, frontier, 1);
+        before["cache_sha256"] = serde_json::json!(Sha256::digest(read_bounded(&main_path).await).to_vec());
+        let previous: serde_json::Value =
+            serde_json::from_slice(&read_bounded(&workspace.join(format!("round-{}.json", request.round - 1))).await)
+                .expect("previous successful process report");
+        assert_eq!(previous["round"], request.round - 1);
+        assert_eq!(previous["root_id"], serde_json::json!(state.root_id));
+        assert_eq!(
+            previous["after"], before,
+            "serialized frontier, receipt, and entry debt must survive process exit"
+        );
+        assert_eq!(cache.info.leader_epoch, 7 + u64::from(request.round));
+        let receipt = cache.info.scan_coverage_receipt.clone();
+        assert_eq!(
+            cache.prepare_bucket_checkpoint("bucket", 11, 8 + u64::from(request.round), SOURCE, PLAN, state.identity),
+            crate::DataUsageCachePrepareOutcome::Reused,
+        );
+        assert_eq!(
+            cache.info.scan_coverage_receipt, receipt,
+            "leader adoption must retain the original receipt"
+        );
+        assert_eq!(cache.validated_scan_frontier(), Some(frontier));
+    }
+    let bad_metadata_path = state.disk_root.join("bucket/prefix/a/0000/xl.meta");
+    let metadata_before = read_bounded(&bad_metadata_path).await;
+    assert_eq!(
+        metadata_before, b"not-valid-filemeta",
+        "failure debt must refer to the original unrepaired file"
+    );
+    if request.round == 3 {
+        write_checkpoint_object(&state.disk_root, "prefix/a/0000", &[(None, 1)]).await;
+    }
+    if request.round > 0 {
+        let receipt = cache.info.scan_coverage_receipt.clone();
+        let (scanner, root, _walk_guard) = finish_resumed_walk_scanner(cache, &original).await;
+        visits = scanner.budget.progress().0;
+        assert_eq!(
+            visits,
+            match request.round {
+                1 => 4,
+                2 => 0,
+                3 => 1,
+                _ => unreachable!("bounded resumed round"),
+            },
+            "covered healthy descendants must skip reads"
+        );
+        assert!(!scanner.coverage_gap, "only proven completed coverage may be checkpointed");
+        assert_eq!(scanner.checkpoint_depth, 0, "the completed root has no suspended ancestors");
+        assert_eq!(scanner.coverage_frontier.as_deref(), Some(DEBT_COMPLETED_FRONTIER));
+        let totals = scanner
+            .new_cache
+            .checked_flatten_complete_scope("bucket")
+            .expect("resumed accounting before checkpoint sealing");
+        let healthy = usize::try_from(OBJECTS).expect("healthy originals plus Unicode sibling") + usize::from(request.round == 3);
+        assert_eq!(
+            (totals.objects, totals.size, totals.failed_objects),
+            (healthy, healthy, usize::from(request.round != 3))
+        );
+        // A low-level walk advances resume hints without sealing its mutable
+        // cache. Persist the same snapshot primitive used by periodic emission.
+        cache = scanner
+            .checkpoint_snapshot(
+                &CachedFolder {
+                    name: "bucket".to_string(),
+                    parent: None,
+                    object_heal_prob_div: 1,
+                },
+                &hash_path("bucket"),
+                &root,
+            )
+            .expect("seal the real completed root checkpoint");
+        if request.round == 2 {
+            assert_eq!(
+                cache.info.scan_coverage_receipt, receipt,
+                "an unchanged completed range must retain the same receipt"
+            );
+        } else {
+            assert_ne!(
+                cache.info.scan_coverage_receipt, receipt,
+                "new completed coverage or repaired entry accounting must change the receipt"
+            );
+        }
+    }
+    let failed_objects = usize::from(request.round != 3);
+    let frontier = if request.round == 0 {
+        DEBT_FRONTIER
+    } else {
+        DEBT_COMPLETED_FRONTIER
+    };
+    let expected = checkpoint_debt_observation(&cache, state.identity, frontier, failed_objects);
+    assert_eq!(cache.info.leader_epoch, 8 + u64::from(request.round));
+    cache
+        .save_with_revisions_for_epoch(store.clone(), CACHE_NAME, &revisions, 0)
+        .await
+        .expect("persist combined healthy coverage and failure debt with CAS");
+    let saved = store.strict_load().await;
+    let mut after = checkpoint_debt_observation(&saved, state.identity, frontier, failed_objects);
+    assert_eq!(after, expected, "the saved codec must retain the complete checkpoint observation");
+    after["cache_sha256"] = serde_json::json!(Sha256::digest(read_bounded(&main_path).await).to_vec());
+    let metadata_after = read_bounded(&bad_metadata_path).await;
+    if request.round == 3 {
+        assert_ne!(metadata_after, metadata_before, "repair must replace the actual bad metadata");
+    } else {
+        assert_eq!(metadata_after, metadata_before, "handoffs must leave the failed source unrepaired");
+    }
+    if request.round == 0 {
+        tokio::fs::write(&state_path, serde_json::to_vec(&state).expect("fixture state JSON"))
+            .await
+            .expect("persist original disk/cache identity");
+        root_guard.temp_dir = None; // The driver-owned workspace outlives this writer process.
+    }
+    serde_json::json!({
+        "schema": 1, "mode": "checkpoint_debt", "pid": std::process::id(), "round": request.round,
+        "writer_pid": state.writer_pid, "root_id": state.root_id,
+        "disk_root": state.disk_root, "cache_root": state.cache_root, "identity": state.identity,
+        "source": SOURCE, "before": before, "after": after, "objects_processed": visits,
+        "diagnostic_ttl_secs": original.failed_object_ttl_secs,
+        "metadata_repaired": request.round == 3,
+        "metadata_sha256_before": Sha256::digest(metadata_before).to_vec(),
+        "metadata_sha256_after": Sha256::digest(metadata_after).to_vec(),
+    })
+}
+
 #[tokio::test]
 #[serial]
 async fn periodic_resume_uses_traversal_order_for_punctuation_siblings() {
-    let (loaded, _, original, _guard) = nested_periodic_snapshot(true, (0, 0)).await;
-    write_checkpoint_object(std::path::Path::new(&original.root), "prefix/a-", &[(None, 1)]).await;
-    let (mut cache, _, _walk_guard) = finish_resumed_walk(loaded, &original).await;
-    cache.info.failed_objects.clear();
-    cache
-        .seal_scan_frontier(Some("bucket/prefix/a-"))
-        .expect("seal completed punctuation sibling");
-    let (finished, visits, _resumed_guard) = finish_resumed_walk(cache, &original).await;
-    assert_eq!(visits, 4, "healthy children of the earlier failed sibling must remain covered");
-    let totals = finished
-        .checked_flatten_complete_scope("bucket")
-        .expect("punctuation sibling accounting");
-    assert_eq!(
-        totals.objects,
-        usize::try_from(OBJECTS).expect("healthy originals plus punctuation sibling")
-    );
-    assert_eq!(totals.size, totals.objects);
-    assert_eq!(totals.failed_objects, 1);
+    if let Some(path) = std::env::var_os(REQUEST_ENV) {
+        let request: CheckpointDebtRequest =
+            serde_json::from_slice(&read_bounded(Path::new(&path)).await).expect("bounded checkpoint-debt restart request");
+        let report = checkpoint_debt_restart_round(&request).await;
+        tokio::fs::write(
+            request.workspace.join(format!("round-{}.json", request.round)),
+            serde_json::to_vec(&report).expect("checkpoint-debt report JSON"),
+        )
+        .await
+        .expect("write checkpoint-debt restart report");
+        return;
+    }
+    for sibling in ["prefix/a-", "prefix/a中"] {
+        let (loaded, _, original, _guard) = nested_periodic_snapshot(true, (0, 0)).await;
+        write_checkpoint_object(std::path::Path::new(&original.root), sibling, &[(None, 1)]).await;
+        let (mut cache, _, _walk_guard) = finish_resumed_walk(loaded, &original).await;
+        cache.info.failed_objects.clear();
+        let frontier = format!("bucket/{sibling}");
+        cache
+            .seal_scan_frontier(Some(&frontier))
+            .expect("seal completed punctuation or Unicode sibling");
+        assert_eq!(cache.validated_scan_frontier(), Some(frontier.as_str()));
+        let (finished, visits, _resumed_guard) = finish_resumed_walk(cache, &original).await;
+        assert_eq!(visits, 4, "healthy children of the earlier failed sibling must remain covered");
+        let totals = finished
+            .checked_flatten_complete_scope("bucket")
+            .expect("punctuation or Unicode sibling accounting");
+        assert_eq!(totals.objects, usize::try_from(OBJECTS).expect("healthy originals plus sibling"));
+        assert_eq!(totals.size, totals.objects);
+        assert_eq!(totals.failed_objects, 1);
+    }
 }
 
 #[tokio::test]

crates/scanner/src/scanner_folder/tests/enumeration_restart.rs

@@ -6,7 +6,7 @@ use std::path::{Path, PathBuf};
 use tokio::io::AsyncReadExt;
 
 const MAX_CACHE_BYTES: u64 = 1024 * 1024;
-const REQUEST_ENV: &str = "RUSTFS_ENUMERATION_REQUEST";
+pub(super) const REQUEST_ENV: &str = "RUSTFS_ENUMERATION_REQUEST";
 
 struct Observation {
     root: PathBuf,
@@ -72,7 +72,7 @@ struct Request {
     round: u32,
 }
 
-async fn read_bounded(path: &Path) -> Vec<u8> {
+pub(super) async fn read_bounded(path: &Path) -> Vec<u8> {
     let file = tokio::fs::File::open(path).await.expect("open fixture artifact");
     let mut bytes = Vec::new();
     file.take(MAX_CACHE_BYTES + 1)

scripts/diagnose_scanner_enumeration_restart.py

@@ -2,13 +2,17 @@
 """Strict restart diagnostic using the real scanner libtest worker, not a walker model."""
 
 import argparse
+import hashlib
 import json
 import os
 from pathlib import Path
 import subprocess
 import sys
+import uuid
 
 WORKER = "scanner_folder::tests::enumeration_restart::enumeration_restart_worker"
+DEBT_WORKER = ("scanner_folder::tests::checkpoint_fixture::periodic_snapshot::"
+               "periodic_resume_uses_traversal_order_for_punctuation_siblings")
 MAX_REPORT_BYTES = 16384
 
 
@@ -141,27 +145,110 @@ def validate_recoverable_quantum(reports, *, objects, budget, require_converged)
         raise ValueError("fixed-budget restart convergence was not established")
 
 
+def validate_checkpoint_debt_report(report, *, round_number, pid, workspace, previous):
+    if not isinstance(report, dict):
+        raise ValueError("checkpoint-debt report must be an object")
+    repaired = round_number == 3
+    for key, expected in {"schema": 1, "round": round_number, "pid": pid,
+                          "diagnostic_ttl_secs": 0,
+                          "objects_processed": (7, 4, 0, 1)[round_number]}.items():
+        if type(report.get(key)) is not int or report[key] != expected:
+            raise ValueError(f"checkpoint-debt report mismatch: {key}")
+    if report.get("mode") != "checkpoint_debt" or report.get("metadata_repaired") is not repaired:
+        raise ValueError("worker did not execute the requested checkpoint-debt phase")
+    if any(item["pid"] == pid for item in previous):
+        raise ValueError("checkpoint-debt handoffs require distinct OS processes")
+    writer_pid = previous[0]["pid"] if previous else pid
+    if type(report.get("writer_pid")) is not int or report["writer_pid"] != writer_pid:
+        raise ValueError("writer identity changed across checkpoint handoff")
+    if type(report.get("root_id")) is not str or uuid.UUID(report["root_id"]).int == 0:
+        raise ValueError("missing original fixture identity")
+    if type(report.get("disk_root")) is not str or type(report.get("cache_root")) is not str:
+        raise ValueError("missing original disk/cache paths")
+    disk_root = Path(report["disk_root"])
+    cache_root = Path(report["cache_root"])
+    if (disk_root.resolve(strict=True) != disk_root or disk_root.parent != workspace
+            or cache_root != workspace / "cache" or cache_root.resolve(strict=True) != cache_root):
+        raise ValueError("worker did not reopen the driver-owned original roots")
+    if (disk_root / "checkpoint-root-id").read_text(encoding="utf-8") != report["root_id"]:
+        raise ValueError("original disk identity receipt does not match")
+    if not isinstance(report.get("identity"), dict) or report.get("source") != {"pool_index": 0, "set_index": 0}:
+        raise ValueError("missing checkpoint scan identity/source")
+    if previous:
+        for key in ("root_id", "disk_root", "cache_root", "identity", "source"):
+            if report[key] != previous[0][key]:
+                raise ValueError(f"original checkpoint fixture changed: {key}")
+        if report.get("before") != previous[-1]["after"]:
+            raise ValueError("frontier, receipt, debt or cache bytes did not survive process exit")
+    elif report.get("before") is not None:
+        raise ValueError("writer unexpectedly reused a previous checkpoint")
+    after = report.get("after")
+    if not isinstance(after, dict):
+        raise ValueError("missing persisted checkpoint observation")
+    for key, expected in {"objects": 9 if repaired else 8, "bytes": 9 if repaired else 8,
+                          "failed_objects": 0 if repaired else 1, "diagnostics": 0,
+                          "leader_epoch": 8 + round_number}.items():
+        if type(after.get(key)) is not int or after[key] != expected:
+            raise ValueError(f"persisted checkpoint mismatch: {key}")
+    frontier = "bucket/prefix/a中" if round_number == 0 else "bucket/prefix"
+    receipt = after.get("receipt")
+    if (after.get("frontier") != frontier or after.get("snapshot_complete") is not False
+            or not isinstance(receipt, dict) or receipt.get("through") != frontier):
+        raise ValueError("missing exact completed frontier/coverage receipt")
+    for digest in (receipt.get("digest"), after.get("cache_sha256"),
+                   report.get("metadata_sha256_before"), report.get("metadata_sha256_after")):
+        if (not isinstance(digest, list) or len(digest) != 32 or not any(digest)
+                or any(type(byte) is not int or not 0 <= byte <= 255 for byte in digest)):
+            raise ValueError("invalid checkpoint or metadata digest")
+    with (cache_root / "main").open("rb") as handle:
+        cache_bytes = handle.read(1048577)
+    if len(cache_bytes) > 1048576 or list(hashlib.sha256(cache_bytes).digest()) != after["cache_sha256"]:
+        raise ValueError("report does not describe the persisted main cache bytes")
+    if previous:
+        unchanged_receipt = receipt == report["before"]["receipt"]
+        if round_number == 2 and not unchanged_receipt:
+            raise ValueError("unchanged completed coverage replaced its receipt")
+        if round_number in (1, 3) and unchanged_receipt:
+            raise ValueError("expanded coverage or repaired entry accounting did not change the receipt")
+    bad_digest = list(hashlib.sha256(b"not-valid-filemeta").digest())
+    if report["metadata_sha256_before"] != bad_digest:
+        raise ValueError("phase did not start from the original corrupt metadata")
+    with (disk_root / "bucket/prefix/a/0000/xl.meta").open("rb") as handle:
+        metadata_bytes = handle.read(1048577)
+    if (len(metadata_bytes) > 1048576
+            or list(hashlib.sha256(metadata_bytes).digest()) != report["metadata_sha256_after"]):
+        raise ValueError("report does not describe the original metadata file")
+    if (report["metadata_sha256_after"] != bad_digest) is not repaired:
+        raise ValueError("failure debt cleared without the requested actual metadata repair")
+
+
 def run(args):
     binary = args.test_binary.resolve(strict=True)
-    listed = subprocess.run([str(binary), WORKER, "--exact", "--list"],
+    selected_worker = DEBT_WORKER if args.checkpoint_debt else WORKER
+    listed = subprocess.run([str(binary), selected_worker, "--exact", "--list"],
                             check=True, capture_output=True, text=True, timeout=30)
-    if f"{WORKER}: test" not in listed.stdout.splitlines():
+    if f"{selected_worker}: test" not in listed.stdout.splitlines():
         raise ValueError("binary does not contain the exact scanner worker test")
     workspace = args.output.resolve()
     workspace.mkdir()  # Refuse reuse/overwrite of previous evidence or customer data.
+    if args.checkpoint_debt:
+        (workspace / "cache").mkdir()
     reports = []
     replayed_raw_window = False
-    for round_number in range(args.rounds):
-        request = {"workspace": str(workspace), "objects": args.objects,
-                   "raw_entry_budget": args.raw_entry_budget, "round": round_number}
+    for round_number in range(4 if args.checkpoint_debt else args.rounds):
+        request = {"workspace": str(workspace), "round": round_number}
+        if not args.checkpoint_debt:
+            request.update(objects=args.objects, raw_entry_budget=args.raw_entry_budget)
         request_path = workspace / "request.json"
         request_path.write_text(json.dumps(request), encoding="utf-8")
         env = dict(os.environ, RUSTFS_ENUMERATION_REQUEST=str(request_path),
                    RUST_MIN_STACK="4194304", NO_PROXY="localhost,127.0.0.1,::1",
                    no_proxy="localhost,127.0.0.1,::1")
-        with subprocess.Popen([str(binary), WORKER, "--exact", "--test-threads=1"],
-                              env=env, stdout=subprocess.DEVNULL,
-                              stderr=subprocess.DEVNULL) as worker:
+        if args.checkpoint_debt:
+            env.update(TMPDIR=str(workspace), TMP=str(workspace), TEMP=str(workspace))
+        with (workspace / f"round-{round_number}.log").open("wb") as log, subprocess.Popen(
+                [str(binary), selected_worker, "--exact", "--test-threads=1"],
+                env=env, stdout=log, stderr=subprocess.STDOUT) as worker:
             try:
                 status = worker.wait(timeout=args.timeout)
             except subprocess.TimeoutExpired:
@@ -176,8 +263,20 @@ def run(args):
             if len(raw) > MAX_REPORT_BYTES:
                 raise ValueError("oversized worker report")
             report = json.loads(raw)
-            validate_report(report, round_number=round_number, pid=worker.pid,
-                            objects=args.objects, budget=args.raw_entry_budget)
+            if args.checkpoint_debt:
+                validate_checkpoint_debt_report(report, round_number=round_number, pid=worker.pid,
+                                                workspace=workspace, previous=reports)
+            else:
+                validate_report(report, round_number=round_number, pid=worker.pid,
+                                objects=args.objects, budget=args.raw_entry_budget)
+        if args.checkpoint_debt:
+            reports.append(report)
+            print(json.dumps(report, sort_keys=True), flush=True)
+            if round_number == 3:
+                print("PASS: four distinct processes retained Unicode coverage and failure debt across two handoffs; "
+                      "debt cleared after actual metadata repair")
+                return 0
+            continue
         if reports and replays_raw_window(reports[-1], report):
             replayed_raw_window = True
         reports.append(report)
@@ -198,6 +297,8 @@ def main():
     parser.add_argument("--test-binary", type=Path, required=True,
                         help="compiled rustfs-scanner libtest executable")
     parser.add_argument("--output", type=Path, required=True, help="new evidence directory (must not exist)")
+    parser.add_argument("--checkpoint-debt", action="store_true",
+                        help="run four fixed processes for Unicode frontier, failure debt, handoffs and metadata repair")
     parser.add_argument("--objects", type=bounded_int(1, 1024), default=128)
     parser.add_argument("--raw-entry-budget", type=bounded_int(1, 4096), default=8)
     parser.add_argument("--rounds", type=bounded_int(1, 64), default=8)