20260717 rpctl接入rpki-prover实验控制

scripts/rpctl/README.md

@@ -1,18 +1,22 @@
 # rpctl
 
-`rpctl` 是四款 RP 本地控制 CLI，用于在实验服务器本地独立启动、停止和观察单个 RP/profile 的 all5 实验。一个实验可包含一次 snapshot 和多次 delta 子 run。
+`rpctl` 是五款 RP 本地控制 CLI，用于在实验服务器本地独立启动、停止和观察单个 RP/profile 的 all5 实验。一个实验可包含一次 snapshot 和多次复用 state 的子 run。
 
 ## 部署
 
 ```bash
-mkdir -p /root/rpki_4rp_ctl
+mkdir -p /data/rpki_4rp_ctl
-rsync -a rpki_2/rpki/scripts/rpctl/ root@47.77.237.41:/root/rpki_4rp_ctl/
+rsync -a rpki_2/rpki/scripts/rpctl/ root@47.251.127.231:/data/rpki_4rp_ctl/
-rsync -a specs/develop/20260707_3/feature100_four_rp_interval_archive/runtime_bundle/ root@47.77.237.41:/root/rpki_4rp_ctl_runtime_bundle/
+rsync -a artifacts/rpki-prover/rpki-prover-6857d4bf-linux-amd64.docker.tar.gz root@47.251.127.231:/data/
-ssh root@47.77.237.41 'cd /root/rpki_4rp_ctl && ./rpctl init --bundle /root/rpki_4rp_ctl_runtime_bundle'
+ssh root@47.251.127.231 'cd /data/rpki_4rp_ctl && ./rpctl init \
+  --bundle /data/rpki_4rp_ctl_runtime_bundle \
+  --rpki-prover-image /data/rpki-prover-6857d4bf-linux-amd64.docker.tar.gz'
 ```
 
 `--bundle` 目录需要包含 #100 runtime bundle 的 `bin/`、`lib/`、`fixtures/`、`scripts/` 和 `versions.json`。不要使用早期 precheck 目录作为标准 bundle，因为它可能缺少 FORT 运行所需动态库。
 
+`--rpki-prover-image` 是可选项；不配置时原有四款 RP 仍可运行，但启动 rpki-prover 会明确报错。配置后 `init` 会复制并加载离线镜像，校验宿主机和镜像均为 `amd64`，确认 tag 为 `rpki-prover:6857d4bf`、版本为 `rpki-prover-0.10.1`，并将 image ID 与 archive SHA-256 写入 `config.json`。运行时不会在线拉取镜像。
+
 ## 常用命令
 
 ```bash
@@ -20,6 +24,8 @@ ssh root@47.77.237.41 'cd /root/rpki_4rp_ctl && ./rpctl init --bundle /root/rpki
 ./rpctl status
 ./rpctl start --rp ours-pp-object-cache --interval 10m --runs 6 --rirs all5
 ./rpctl start --rp routinator --interval 0 --runs 1 --rirs apnic
+./rpctl start --rp rpki-prover --interval 0 --runs 2 --rirs all5
+./rpctl start --rp rpki-prover-full --interval 0 --runs 1 --rirs all5
 ./rpctl continue --exp-id <successful-exp-id> --runs 3 --interval 10m
 ./rpctl tail
 ./rpctl show
@@ -40,6 +46,10 @@ ssh root@47.77.237.41 'cd /root/rpki_4rp_ctl && ./rpctl init --bundle /root/rpki
 | `routinator` | `routinator-latest-release` |
 | `rpki-client` | `rpki-client-latest-release` |
 | `fort` | `fort-latest-release` |
+| `rpki-prover`, `rpki-prover-incremental` | `rpki-prover-incremental` |
+| `rpki-prover-full`, `rpki-prover-no-incremental` | `rpki-prover-full` |
+
+`rpki-prover-incremental` 使用上游默认增量算法；`rpki-prover-full` 显式传入 `--no-incremental-validation`。两者的 state 位于各自实验/profile 目录，互不复用。首轮显示为 `snapshot`；后续和 `continue` 追加轮显示为 `warm`，表示保留了上轮 state，实际 RRDP snapshot/delta 决策仍由 rpki-prover 自身完成。
 
 ## 数据目录
 
@@ -53,7 +63,9 @@ ssh root@47.77.237.41 'cd /root/rpki_4rp_ctl && ./rpctl init --bundle /root/rpki
   runs/<profile>/<exp-id>/profiles/<profile>/runs/run_0001/
 ```
 
-每个子 run 目录包含 `stdout.log`、`stderr.log`、`process-time.txt`、`exit-code.txt`、`run-meta.json` 以及各 RP 产物。
+每个子 run 目录包含 `stdout.log`、`stderr.log`、`exit-code.txt`、`run-meta.json` 以及各 RP 产物。传统四 RP 使用 `process-time.txt`；rpki-prover 额外保存 `container-inspect.json`、`rss-samples.tsv` 和 `vrps.csv`。
+
+rpki-prover 的 `run-meta.json` 和 `view` 同时记录 VRP 原始行数与按 `(ASN, prefix, maxLength)` 去重数。其当前 CLI 不输出 VAP，所以统一显示 `n/a`，不是 `0`。max RSS 是容器全部相关 PID 的 RSS 峰值汇总，不是短暂 Docker client 进程的 RSS。
 
 ## 历史查看与删除
 

scripts/rpctl/rpctl

@@ -4,8 +4,10 @@ from __future__ import annotations
 import argparse
 import csv
 import gzip
+import hashlib
 import json
 import os
+import platform
 import re
 import shutil
 import signal
@@ -37,6 +39,8 @@ PROFILES = {
     "routinator-latest-release": "routinator",
     "rpki-client-latest-release": "rpki-client",
     "fort-latest-release": "fort",
+    "rpki-prover-incremental": "rpki-prover",
+    "rpki-prover-full": "rpki-prover",
 }
 
 ALIASES = {
@@ -52,8 +56,15 @@ ALIASES = {
     "rpki-client-latest-release": "rpki-client-latest-release",
     "fort": "fort-latest-release",
     "fort-latest-release": "fort-latest-release",
+    "rpki-prover": "rpki-prover-incremental",
+    "rpki-prover-incremental": "rpki-prover-incremental",
+    "rpki-prover-full": "rpki-prover-full",
+    "rpki-prover-no-incremental": "rpki-prover-full",
 }
 
+RPKI_PROVER_IMAGE_TAG = "rpki-prover:6857d4bf"
+RPKI_PROVER_VERSION = "rpki-prover-0.10.1"
+
 DEFAULT_CONFIG = {
     "root": "/root/rpki_4rp_ctl",
     "bundle": "",
@@ -207,6 +218,14 @@ def command_string(argv: list[str]) -> str:
     return " ".join(shlex.quote(x) for x in argv)
 
 
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
 def check_bundle(root: Path) -> None:
     required = ["bin/rpki", "bin/routinator", "bin/rpki-client", "bin/fort", "fixtures/tal", "scripts/normalize_and_collect.py", "scripts/write_group_report.py", "versions.json"]
     missing = [item for item in required if not (root / item).exists()]
@@ -232,6 +251,102 @@ def install_from_bundle(bundle: Path) -> None:
                 item.chmod(item.stat().st_mode | 0o111)
 
 
+def validate_rpki_prover_host() -> None:
+    arch = platform.machine().lower()
+    if arch not in {"x86_64", "amd64"}:
+        raise SystemExit(f"rpki-prover image requires amd64 host, found: {arch}")
+    if shutil.which("docker") is None:
+        raise SystemExit("docker is required for rpki-prover profiles")
+    result = subprocess.run(
+        ["docker", "info"],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.PIPE,
+        text=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        raise SystemExit(f"docker is not available: {result.stderr.strip()}")
+
+
+def inspect_rpki_prover_image() -> dict[str, str]:
+    inspect = subprocess.run(
+        ["docker", "image", "inspect", RPKI_PROVER_IMAGE_TAG],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+        check=False,
+    )
+    if inspect.returncode != 0:
+        raise SystemExit(f"loaded image is missing tag {RPKI_PROVER_IMAGE_TAG}: {inspect.stderr.strip()}")
+    data = json.loads(inspect.stdout)[0]
+    architecture = str(data.get("Architecture") or "")
+    if architecture != "amd64":
+        raise SystemExit(f"rpki-prover image architecture mismatch: expected amd64, found {architecture or '<unknown>'}")
+    version = subprocess.run(
+        ["docker", "run", "--rm", "--entrypoint", "/opt/app/rpki-prover", RPKI_PROVER_IMAGE_TAG, "--version"],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+        check=False,
+    )
+    version_text = version.stdout.strip()
+    if version.returncode != 0 or version_text != RPKI_PROVER_VERSION:
+        raise SystemExit(
+            f"rpki-prover version check failed: expected {RPKI_PROVER_VERSION}, "
+            f"exit={version.returncode} output={version_text or version.stderr.strip()}"
+        )
+    return {
+        "tag": RPKI_PROVER_IMAGE_TAG,
+        "imageId": str(data.get("Id") or ""),
+        "architecture": architecture,
+        "version": version_text,
+    }
+
+
+def install_rpki_prover_image(archive: Path) -> dict[str, str]:
+    if not archive.is_file():
+        raise SystemExit(f"rpki-prover image archive does not exist: {archive}")
+    validate_rpki_prover_host()
+    image_dir = ctl_root() / "images"
+    image_dir.mkdir(parents=True, exist_ok=True)
+    destination = image_dir / archive.name
+    if archive != destination:
+        shutil.copy2(archive, destination)
+    load = subprocess.run(
+        ["docker", "load", "--input", str(destination)],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+        check=False,
+    )
+    if load.returncode != 0:
+        raise SystemExit(f"failed to load rpki-prover image archive: {load.stderr.strip()}")
+    metadata = inspect_rpki_prover_image()
+    metadata.update({
+        "archive": str(destination),
+        "archiveSha256": sha256_file(destination),
+        "loadedAtUtc": utc_now(),
+    })
+    return metadata
+
+
+def require_rpki_prover_image() -> dict[str, Any]:
+    metadata = load_config().get("rpkiProverImage")
+    if not isinstance(metadata, dict) or metadata.get("tag") != RPKI_PROVER_IMAGE_TAG:
+        raise SystemExit(
+            "rpki-prover image is not initialized; run ./rpctl init --bundle <bundle> "
+            "--rpki-prover-image <docker.tar.gz>"
+        )
+    validate_rpki_prover_host()
+    inspected = inspect_rpki_prover_image()
+    if metadata.get("imageId") and metadata.get("imageId") != inspected.get("imageId"):
+        raise SystemExit(
+            f"rpki-prover image id changed: configured={metadata.get('imageId')} "
+            f"loaded={inspected.get('imageId')}"
+        )
+    return metadata
+
+
 def cmd_init(args: argparse.Namespace) -> None:
     ensure_dirs()
     bundle = Path(args.bundle).resolve()
@@ -239,9 +354,16 @@ def cmd_init(args: argparse.Namespace) -> None:
     cfg = load_config()
     cfg["bundle"] = str(bundle)
     cfg["initializedAtUtc"] = utc_now()
+    if args.rpki_prover_image:
+        cfg["rpkiProverImage"] = install_rpki_prover_image(Path(args.rpki_prover_image).resolve())
     save_config(cfg)
     print(f"initialized root={ctl_root()}")
     print(f"bundle={bundle}")
+    if cfg.get("rpkiProverImage"):
+        image = cfg["rpkiProverImage"]
+        print(f"rpki_prover_image={image.get('tag')}")
+        print(f"rpki_prover_image_id={image.get('imageId')}")
+        print(f"rpki_prover_archive_sha256={image.get('archiveSha256')}")
 
 
 def timed_run(run_dir: Path, argv: list[str], env: dict[str, str] | None = None) -> int:
@@ -396,6 +518,240 @@ def build_fort_args(run_dir: Path, state: Path, rirs: list[str]) -> tuple[list[s
     return argv, env
 
 
+def prepare_rpki_prover_state(state: Path, rirs: list[str]) -> None:
+    tal_dir = state / "tals"
+    tal_dir.mkdir(parents=True, exist_ok=True)
+    for old_tal in tal_dir.glob("*.tal"):
+        old_tal.unlink()
+    for rir in rirs:
+        shutil.copy2(ctl_root() / "fixtures" / "tal" / RIR_TAL[rir], tal_dir / RIR_TAL[rir])
+
+
+def rpki_prover_arguments(profile: str) -> list[str]:
+    argv = [
+        "--once",
+        "--vrp-output", "/outputs/vrps.csv",
+        "--no-rir-tals",
+        "--rpki-root-directory", "/rpki-data",
+        "--cpu-count", "4",
+        "--fetcher-count", "8",
+        "--log-level", "info",
+    ]
+    if profile == "rpki-prover-full":
+        argv.append("--no-incremental-validation")
+    return argv
+
+
+def build_rpki_prover_command(profile: str, run_dir: Path, state: Path, container_name: str) -> list[str]:
+    return [
+        "docker", "run", "--name", container_name,
+        "-v", f"{state.resolve()}:/rpki-data",
+        "-v", f"{run_dir.resolve()}:/outputs",
+        RPKI_PROVER_IMAGE_TAG,
+        *rpki_prover_arguments(profile),
+    ]
+
+
+def container_rss_kb(container_name: str) -> tuple[int, int]:
+    top = subprocess.run(
+        ["docker", "top", container_name, "-eo", "pid"],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.DEVNULL,
+        text=True,
+        check=False,
+    )
+    total = 0
+    maximum = 0
+    for line in top.stdout.splitlines():
+        pid = line.strip()
+        if not pid.isdigit():
+            continue
+        try:
+            status = Path(f"/proc/{pid}/status").read_text(encoding="utf-8", errors="replace")
+        except OSError:
+            continue
+        match = re.search(r"^VmRSS:\s+(\d+)\s+kB$", status, re.MULTILINE)
+        if not match:
+            continue
+        rss = int(match.group(1))
+        total += rss
+        maximum = max(maximum, rss)
+    return total, maximum
+
+
+def timed_rpki_prover_run(
+    profile: str,
+    run_dir: Path,
+    state: Path,
+    container_name: str,
+) -> tuple[int, int, int, int, dict[str, Any]]:
+    run_dir.mkdir(parents=True, exist_ok=True)
+    start_utc = utc_now()
+    start_ns = time.monotonic_ns()
+    (run_dir / "start-utc.txt").write_text(start_utc + "\n", encoding="utf-8")
+    create_argv = [
+        "docker", "create", "--name", container_name,
+        "-v", f"{state.resolve()}:/rpki-data",
+        "-v", f"{run_dir.resolve()}:/outputs",
+        RPKI_PROVER_IMAGE_TAG,
+        *rpki_prover_arguments(profile),
+    ]
+    create = subprocess.run(create_argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False)
+    if create.returncode != 0:
+        (run_dir / "stderr.log").write_text(create.stderr, encoding="utf-8")
+        (run_dir / "stdout.log").write_text(create.stdout, encoding="utf-8")
+        (run_dir / "end-utc.txt").write_text(utc_now() + "\n", encoding="utf-8")
+        (run_dir / "exit-code.txt").write_text(str(create.returncode) + "\n", encoding="utf-8")
+        return create.returncode, 0, 0, (time.monotonic_ns() - start_ns) // 1_000_000, {}
+
+    previous_sigterm = signal.getsignal(signal.SIGTERM)
+
+    def terminate_container(signum: int, _frame: Any) -> None:
+        subprocess.run(
+            ["docker", "rm", "--force", container_name],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            check=False,
+        )
+        raise SystemExit(128 + signum)
+
+    signal.signal(signal.SIGTERM, terminate_container)
+
+    max_total_rss_kb = 0
+    max_process_rss_kb = 0
+    inspect_data: dict[str, Any] = {}
+    samples_path = run_dir / "rss-samples.tsv"
+    with (run_dir / "stdout.log").open("wb") as stdout, (run_dir / "stderr.log").open("wb") as stderr, samples_path.open("w", encoding="utf-8") as samples:
+        samples.write("epoch_ms\ttotal_rss_kb\tmax_process_rss_kb\n")
+        start = subprocess.Popen(["docker", "start", "--attach", container_name], stdout=stdout, stderr=stderr)
+        while start.poll() is None:
+            total_rss_kb, process_rss_kb = container_rss_kb(container_name)
+            max_total_rss_kb = max(max_total_rss_kb, total_rss_kb)
+            max_process_rss_kb = max(max_process_rss_kb, process_rss_kb)
+            samples.write(f"{int(time.time() * 1000)}\t{total_rss_kb}\t{process_rss_kb}\n")
+            samples.flush()
+            time.sleep(0.25)
+        start.wait()
+
+    try:
+        inspect = subprocess.run(
+            ["docker", "inspect", container_name],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            check=False,
+        )
+        if inspect.returncode == 0:
+            parsed = json.loads(inspect.stdout)
+            if parsed:
+                inspect_data = parsed[0]
+                write_json(run_dir / "container-inspect.json", inspect_data)
+        state_data = inspect_data.get("State", {}) if inspect_data else {}
+        exit_code = int(state_data.get("ExitCode", start.returncode if start.returncode is not None else 1))
+    finally:
+        subprocess.run(["docker", "rm", "--force", container_name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)
+        signal.signal(signal.SIGTERM, previous_sigterm)
+    wall_ms = (time.monotonic_ns() - start_ns) // 1_000_000
+    (run_dir / "end-utc.txt").write_text(utc_now() + "\n", encoding="utf-8")
+    (run_dir / "exit-code.txt").write_text(str(exit_code) + "\n", encoding="utf-8")
+    return exit_code, max_total_rss_kb, max_process_rss_kb, wall_ms, inspect_data
+
+
+def collect_rpki_prover_vrps(path: Path) -> tuple[int, set[str]]:
+    unique: set[str] = set()
+    raw = 0
+    if not path.is_file():
+        return raw, unique
+    with path.open("r", encoding="utf-8", errors="replace", newline="") as handle:
+        for row in csv.reader(handle):
+            if not row or row[0].strip().lower() == "asn":
+                continue
+            if len(row) < 3:
+                continue
+            raw += 1
+            unique.add(",".join(value.strip() for value in row[:3]))
+    return raw, unique
+
+
+def write_rpki_prover_run_metadata(
+    run_root: Path,
+    profile: str,
+    run_seq: int,
+    mode: str,
+    wall_ms: int,
+    max_rss_kb: int,
+    max_process_rss_kb: int,
+    exit_code: int,
+) -> dict[str, Any]:
+    run_dir = run_root / "profiles" / profile / "runs" / f"run_{run_seq:04d}"
+    raw_vrps, unique_vrps = collect_rpki_prover_vrps(run_dir / "vrps.csv")
+    with gzip.open(run_dir / "normalized-vrps.txt.gz", "wt", encoding="utf-8") as handle:
+        for row in sorted(unique_vrps):
+            handle.write(row + "\n")
+    image = load_config().get("rpkiProverImage", {})
+    meta = {
+        "profile": profile,
+        "rp": "rpki-prover",
+        "runSeq": run_seq,
+        "syncMode": mode,
+        "startUtc": read_text_strip(run_dir / "start-utc.txt"),
+        "endUtc": read_text_strip(run_dir / "end-utc.txt"),
+        "exitCode": exit_code,
+        "timing": {
+            "wallMs": wall_ms,
+            "maxRssKb": max_rss_kb,
+            "maxProcessRssKb": max_process_rss_kb,
+        },
+        "counts": {
+            "vrps": raw_vrps,
+            "vrpsRaw": raw_vrps,
+            "vrpsUnique": len(unique_vrps),
+            "vaps": "n/a",
+        },
+        "vapsAvailable": False,
+        "validationAlgorithm": "full-every-iteration" if profile == "rpki-prover-full" else "incremental",
+        "image": image,
+        "remoteRunDir": str(run_dir),
+    }
+    write_json(run_dir / "run-meta.json", meta)
+    return meta
+
+
+def write_rpki_prover_group_report(run_root: Path, profile: str) -> None:
+    rows: list[dict[str, Any]] = []
+    run_dirs = sorted((run_root / "profiles" / profile / "runs").glob("run_*"))
+    for run_dir in run_dirs:
+        meta = read_json(run_dir / "run-meta.json", {})
+        if not meta:
+            continue
+        rows.append({
+            "profile": profile,
+            "rp": "rpki-prover",
+            "run_seq": meta.get("runSeq", ""),
+            "sync_mode": meta.get("syncMode", ""),
+            "start_utc": meta.get("startUtc", ""),
+            "end_utc": meta.get("endUtc", ""),
+            "exit_code": meta.get("exitCode", ""),
+            "wall_ms": meta.get("timing", {}).get("wallMs", ""),
+            "max_rss_kb": meta.get("timing", {}).get("maxRssKb", ""),
+            "max_process_rss_kb": meta.get("timing", {}).get("maxProcessRssKb", ""),
+            "vrps": meta.get("counts", {}).get("vrpsRaw", ""),
+            "vrps_unique": meta.get("counts", {}).get("vrpsUnique", ""),
+            "vaps": "n/a",
+            "vaps_available": False,
+            "validation_algorithm": meta.get("validationAlgorithm", ""),
+            "remote_run_dir": str(run_dir),
+        })
+    reports = run_root / "reports"
+    reports.mkdir(parents=True, exist_ok=True)
+    if rows:
+        with (reports / "per_run_metrics.csv").open("w", encoding="utf-8", newline="") as handle:
+            writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
+            writer.writeheader()
+            writer.writerows(rows)
+    write_json(reports / "per_run_metrics.json", {"generatedAtUtc": utc_now(), "rows": rows})
+
+
 def normalize_one(run_root: Path, profile: str, run_seq: int) -> None:
     script = ctl_root() / "scripts" / "normalize_and_collect.py"
     if not script.exists():
@@ -460,6 +816,18 @@ def parse_entry_rirs(entry: dict[str, Any]) -> list[str]:
     return parse_rirs(str(value))
 
 
+def write_experiment_versions(run_root: Path) -> None:
+    versions = read_json(ctl_root() / "versions.json", {})
+    image = load_config().get("rpkiProverImage")
+    if image:
+        versions["rpki-prover"] = image
+        profiles = versions.setdefault("profiles", [])
+        for profile in ["rpki-prover-incremental", "rpki-prover-full"]:
+            if profile not in profiles:
+                profiles.append(profile)
+    write_json(run_root / "versions.json", versions)
+
+
 def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs: list[str], append: bool = False) -> int:
     ensure_dirs()
     rp = PROFILES[profile]
@@ -467,7 +835,7 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs:
     profile_root = run_root / "profiles" / profile
     (run_root / "logs").mkdir(parents=True, exist_ok=True)
     (run_root / "reports").mkdir(parents=True, exist_ok=True)
-    (run_root / "versions.json").write_text((ctl_root() / "versions.json").read_text(encoding="utf-8"), encoding="utf-8")
+    write_experiment_versions(run_root)
     now = utc_now()
     if append:
         run_info = normalize_experiment_metadata(read_json(run_root / "run-info.json", {}))
@@ -504,6 +872,9 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs:
             "experimentRoot": str(run_root),
             "stateRoot": str(profile_root / "state"),
         }
+        if rp == "rpki-prover":
+            run_info["validationAlgorithm"] = cache_profile_name(profile)
+            run_info["rpkiProverImage"] = load_config().get("rpkiProverImage", {})
     write_json(run_root / "run-info.json", run_info)
     status = {
         "state": "running",
@@ -535,7 +906,7 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs:
                 time.sleep(sleep_seconds)
         last_start = time.time()
         snapshot = run_seq == 1
-        mode = "snapshot" if snapshot else "delta"
+        mode = "snapshot" if snapshot else ("warm" if rp == "rpki-prover" else "delta")
         run_dir = profile_root / "runs" / f"run_{run_seq:04d}"
         if run_dir.exists() and any(run_dir.iterdir()):
             raise RuntimeError(f"refusing to overwrite existing run directory: {run_dir}")
@@ -550,13 +921,34 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs:
             argv, env = build_rpki_client_args(run_dir, state, rirs)
         elif rp == "fort":
             argv, env = build_fort_args(run_dir, state, rirs)
+        elif rp == "rpki-prover":
+            prepare_rpki_prover_state(state, rirs)
+            container_suffix = hashlib.sha1(f"{run_root}:{run_seq}".encode("utf-8")).hexdigest()[:12]
+            container_name = f"rpctl-prover-{container_suffix}"
+            argv = build_rpki_prover_command(profile, run_dir, state, container_name)
         else:
             raise RuntimeError(rp)
         run_dir.mkdir(parents=True, exist_ok=True)
         (run_dir / "command.txt").write_text(command_string(argv) + "\n", encoding="utf-8")
-        code = timed_run(run_dir, argv, env=env)
+        if rp == "rpki-prover":
-        normalize_one(run_root, profile, run_seq)
+            code, max_rss_kb, max_process_rss_kb, wall_ms, _ = timed_rpki_prover_run(
-        meta = parse_latest_meta(run_root, profile, run_seq)
+                profile, run_dir, state, container_name
+            )
+            meta = write_rpki_prover_run_metadata(
+                run_root,
+                profile,
+                run_seq,
+                mode,
+                wall_ms,
+                max_rss_kb,
+                max_process_rss_kb,
+                code,
+            )
+            write_rpki_prover_group_report(run_root, profile)
+        else:
+            code = timed_run(run_dir, argv, env=env)
+            normalize_one(run_root, profile, run_seq)
+            meta = parse_latest_meta(run_root, profile, run_seq)
         update_current(
             state="running",
             lastRunSeq=run_seq,
@@ -571,7 +963,10 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs:
         if code != 0:
             exit_code = code
             break
-    write_group_report(run_root, interval_secs, completed_seq)
+    if rp == "rpki-prover":
+        write_rpki_prover_group_report(run_root, profile)
+    else:
+        write_group_report(run_root, interval_secs, completed_seq)
     final_state = "success" if exit_code == 0 else "failed"
     update_current(state=final_state, finishedAtUtc=utc_now(), exitCode=exit_code)
     run_info = normalize_experiment_metadata(read_json(run_root / "run-info.json", {}))
@@ -645,6 +1040,10 @@ def cmd_paths(args: argparse.Namespace) -> None:
     print(f"runs={runs_dir()}")
     print(f"logs={logs_dir()}")
     print(f"bundle={cfg.get('bundle','')}")
+    image = cfg.get("rpkiProverImage", {})
+    print(f"rpki_prover_image={image.get('tag', '')}")
+    print(f"rpki_prover_image_id={image.get('imageId', '')}")
+    print(f"rpki_prover_image_archive={image.get('archive', '')}")
     cur = current_live_status()
     if cur.get("experimentRoot"):
         print(f"current_experiment_root={cur['experimentRoot']}")
@@ -668,6 +1067,8 @@ def cmd_start(args: argparse.Namespace) -> None:
         raise SystemExit("--runs must be >= 1")
     if not (ctl_root() / "versions.json").exists():
         raise SystemExit("rpctl is not initialized; run ./rpctl init --bundle <bundle-root> first")
+    if PROFILES[profile] == "rpki-prover":
+        require_rpki_prover_image()
     exp_id = args.exp_id or f"{utc_id()}_{profile}_{args.runs}runs_{interval_secs}s"
     runner_args = [
         sys.executable,
@@ -717,6 +1118,8 @@ def cmd_continue(args: argparse.Namespace) -> None:
     if entry["state"] != "success":
         raise SystemExit(f"can only continue a successful experiment: exp_id={args.exp_id} state={entry['state']}")
     profile = entry["profile"]
+    if PROFILES[profile] == "rpki-prover":
+        require_rpki_prover_image()
     rirs = parse_entry_rirs(entry)
     interval_secs = parse_interval(args.interval)
     runner_args = [
@@ -867,6 +1270,9 @@ def parse_command_summary(command: str) -> str:
             keys.append(f"--rsync-scope={tokens[idx + 1]}")
         elif token.startswith("--rsync-scope="):
             keys.append(token)
+    if RPKI_PROVER_IMAGE_TAG in token_set:
+        algorithm = "full-every-iteration" if "--no-incremental-validation" in token_set else "incremental"
+        keys.append(f"algorithm={algorithm}")
     return " ".join(keys)
 
 
@@ -945,6 +1351,10 @@ def cache_profile_name(profile: str) -> str:
         return "pp+roa+crt"
     if profile == "ours-no-cache":
         return "off"
+    if profile == "rpki-prover-incremental":
+        return "incremental"
+    if profile == "rpki-prover-full":
+        return "full-every-iteration"
     return "n/a"
 
 
@@ -971,6 +1381,8 @@ def profile_label(profile: str) -> str:
         "routinator-latest-release": "routinator",
         "rpki-client-latest-release": "rpki-client",
         "fort-latest-release": "fort",
+        "rpki-prover-incremental": "prover-inc",
+        "rpki-prover-full": "prover-full",
     }
     return labels.get(profile, profile)
 
@@ -1043,6 +1455,9 @@ def infer_output_counts(profile: str, run_dir: Path) -> tuple[str, str]:
         return vrps, vaps
     if rp == "fort":
         return count_csv_data_rows(run_dir / "vrps.csv"), "0" if (run_dir / "vrps.csv").exists() else ""
+    if rp == "rpki-prover":
+        raw, _ = collect_rpki_prover_vrps(run_dir / "vrps.csv")
+        return str(raw) if (run_dir / "vrps.csv").exists() else "", "n/a" if (run_dir / "vrps.csv").exists() else ""
     return "", ""
 
 
@@ -1355,6 +1770,7 @@ def fallback_run_details(entry: dict[str, Any]) -> list[dict[str, Any]]:
             "wall_ms": timing.get("wallMs", ""),
             "max_rss_kb": timing.get("maxRssKb", ""),
             "vrps": first_non_empty(meta.get("counts", {}).get("vrps") if meta else "", inferred_vrps),
+            "vrps_unique": first_non_empty(meta.get("counts", {}).get("vrpsUnique") if meta else ""),
             "vaps": first_non_empty(meta.get("counts", {}).get("vaps") if meta else "", inferred_vaps),
             "remote_run_dir": str(run_dir),
         })
@@ -1391,7 +1807,7 @@ def cmd_view(args: argparse.Namespace) -> None:
         print("command=" + entry["command"])
     if rows:
         print("\nper-run:")
-        headers = ["run", "mode", "exit", "start", "end", "wall_ms", "rss_kb", "vrps", "vaps", "path"]
+        headers = ["run", "mode", "exit", "start", "end", "wall_ms", "rss_kb", "vrps", "vrps_unique", "vaps", "path"]
         table_rows = []
         for row in rows:
             table_rows.append([
@@ -1403,6 +1819,7 @@ def cmd_view(args: argparse.Namespace) -> None:
                 first_non_empty(row.get("wall_ms"), row.get("wallMs")),
                 first_non_empty(row.get("max_rss_kb"), row.get("maxRssKb")),
                 first_non_empty(row.get("vrps"), row.get("VRPs")),
+                first_non_empty(row.get("vrps_unique"), row.get("vrpsUnique")),
                 first_non_empty(row.get("vaps"), row.get("VAPs")),
                 first_non_empty(row.get("remote_run_dir"), row.get("remoteRunDir")),
             ])
@@ -1412,7 +1829,7 @@ def cmd_view(args: argparse.Namespace) -> None:
 def build_parser() -> argparse.ArgumentParser:
     formatter = argparse.RawDescriptionHelpFormatter
     parser = argparse.ArgumentParser(
-        description="Control one local four-RP performance run on this server.",
+        description="Control one local five-RP performance run on this server.",
         formatter_class=formatter,
         epilog="""
 RP/profile naming:
@@ -1435,6 +1852,14 @@ RP/profile naming:
   fort
       Alias of fort-latest-release. Uses archived FORT validator release binary.
 
+  rpki-prover / rpki-prover-incremental
+      Uses the archived rpki-prover Docker image with its default incremental
+      validation algorithm and persistent state.
+
+  rpki-prover-full / rpki-prover-no-incremental
+      Uses the same image with --no-incremental-validation. Its state is kept
+      separate from the incremental profile.
+
 Cache difference for ours RP:
   --rp ours adds:
     --enable-publication-point-validation-cache
@@ -1444,10 +1869,12 @@ Cache difference for ours RP:
   --rp ours-no-cache does not add those flags.
 
 Typical usage:
-  ./rpctl init --bundle /root/rpki_4rp_ctl_runtime_bundle
+  ./rpctl init --bundle /data/rpki_4rp_ctl_runtime_bundle \\
+    --rpki-prover-image /data/rpki-prover-6857d4bf-linux-amd64.docker.tar.gz
   ./rpctl list
   ./rpctl start --rp ours --interval 10m --runs 6 --rirs all5
   ./rpctl start --rp ours-no-cache --interval 10m --runs 6 --rirs all5
+  ./rpctl start --rp rpki-prover --interval 0 --runs 2 --rirs all5
   ./rpctl status
   ./rpctl show
   ./rpctl show --failed
@@ -1458,7 +1885,8 @@ Typical usage:
 
 Data model:
   - Experiment data stays under ./runs/<profile>/<exp-id>/.
-  - Each experiment starts with a snapshot; later child runs are deltas and reuse the profile state.
+  - Each experiment starts with a snapshot; later child runs reuse the profile state.
+  - rpki-prover labels reused-state runs as warm because its native storage decides snapshot/delta behavior.
   - By default only one RP/profile can run at a time.
 """,
     )
@@ -1481,14 +1909,24 @@ The bundle must contain:
   scripts/write_group_report.py
   versions.json
 
+Optional rpki-prover image archive:
+  Pass --rpki-prover-image to copy and load the saved linux/amd64 Docker image.
+  Init verifies image architecture, tag and rpki-prover version without pulling online.
+
 Recommended source on 47.77.237.41:
   /root/rpki_4rp_ctl_runtime_bundle
 
 Example:
-  ./rpctl init --bundle /root/rpki_4rp_ctl_runtime_bundle
+  ./rpctl init --bundle /data/rpki_4rp_ctl_runtime_bundle \\
+    --rpki-prover-image /data/rpki-prover-6857d4bf-linux-amd64.docker.tar.gz
 """,
     )
     p.add_argument("--bundle", required=True, help="Path to the archived #100 runtime bundle root.")
+    p.add_argument(
+        "--rpki-prover-image",
+        default="",
+        help="Optional saved rpki-prover linux/amd64 Docker image archive. Existing four-RP profiles do not require it.",
+    )
     p.set_defaults(func=cmd_init)
 
     p = sub.add_parser(
@@ -1502,6 +1940,10 @@ Use this command when unsure which value to pass to --rp.
 Important ours RP profiles:
   ours-pp-object-cache  cache-enabled profile
   ours-no-cache         no-cache baseline profile
+
+rpki-prover profiles:
+  rpki-prover-incremental  default incremental validation
+  rpki-prover-full         --no-incremental-validation baseline
 """,
     )
     p.set_defaults(func=cmd_list)
@@ -1525,6 +1967,8 @@ RP/profile examples:
   ./rpctl start --rp routinator --interval 15m --runs 2 --rirs apnic
   ./rpctl start --rp rpki-client --interval 0 --runs 1 --rirs apnic,arin
   ./rpctl start --rp fort --interval 0 --runs 1 --rirs all5
+  ./rpctl start --rp rpki-prover --interval 0 --runs 2 --rirs all5
+  ./rpctl start --rp rpki-prover-full --interval 0 --runs 1 --rirs all5
 
 Cache mapping:
   --rp ours         enables PP cache + ROA cache + child certificate cache.
@@ -1537,7 +1981,7 @@ Safety:
     p.add_argument(
         "--rp",
         required=True,
-        help="RP/profile or alias. Common values: ours, ours-no-cache, routinator, rpki-client, fort.",
+        help="RP/profile or alias. Common values: ours, ours-no-cache, routinator, rpki-client, fort, rpki-prover, rpki-prover-full.",
     )
     p.add_argument(
         "--interval",
@@ -1578,6 +2022,7 @@ Run semantics:
 Examples:
   ./rpctl continue --exp-id 20260709T081122Z_ours-pp-object-cache_2runs_600s --runs 3 --interval 10m
   ./rpctl continue --rp routinator --exp-id m18_237_routinator_all5_snapshot_delta_20260710T002726Z --runs 1 --interval 0
+  ./rpctl continue --rp rpki-prover --exp-id prover_all5 --runs 1 --interval 0
 """,
     )
     p.add_argument("--exp-id", required=True, help="Existing successful experiment id to append to.")
@@ -1692,7 +2137,7 @@ Examples:
   ./rpctl show --json
 """,
     )
-    p.add_argument("--rp", default="", help="Optional RP/profile filter, e.g. ours, ours-no-cache, routinator, rpki-client, fort.")
+    p.add_argument("--rp", default="", help="Optional RP/profile filter, e.g. ours, ours-no-cache, routinator, rpki-client, fort, rpki-prover.")
     p.add_argument("--failed", action="store_true", help="Only show failed, partial, incomplete or exited experiments.")
     p.add_argument("--limit", type=int, default=50, help="Maximum rows to show. 0 means no limit. Default: 50.")
     p.add_argument("--paths", action="store_true", help="Also show full experiment root paths. Hidden by default to keep the table readable.")

scripts/rpctl/test_rpctl.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+import importlib.machinery
+import importlib.util
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+
+def load_rpctl():
+    path = Path(__file__).with_name("rpctl")
+    loader = importlib.machinery.SourceFileLoader("rpctl_module", str(path))
+    spec = importlib.util.spec_from_loader(loader.name, loader)
+    module = importlib.util.module_from_spec(spec)
+    loader.exec_module(module)
+    return module
+
+
+class RpkiProverProfileTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.rpctl = load_rpctl()
+
+    def test_profile_aliases_and_algorithms(self) -> None:
+        self.assertEqual(self.rpctl.normalize_profile("rpki-prover"), "rpki-prover-incremental")
+        self.assertEqual(self.rpctl.normalize_profile("rpki-prover-no-incremental"), "rpki-prover-full")
+        self.assertNotIn("--no-incremental-validation", self.rpctl.rpki_prover_arguments("rpki-prover-incremental"))
+        self.assertIn("--no-incremental-validation", self.rpctl.rpki_prover_arguments("rpki-prover-full"))
+        incremental = self.rpctl.parse_command_summary(
+            f"docker run {self.rpctl.RPKI_PROVER_IMAGE_TAG} --once"
+        )
+        full = self.rpctl.parse_command_summary(
+            f"docker run {self.rpctl.RPKI_PROVER_IMAGE_TAG} --once --no-incremental-validation"
+        )
+        self.assertEqual(incremental, "algorithm=incremental")
+        self.assertEqual(full, "algorithm=full-every-iteration")
+
+    def test_vrp_count_preserves_raw_and_unique(self) -> None:
+        with tempfile.TemporaryDirectory() as temp:
+            csv_path = Path(temp) / "vrps.csv"
+            csv_path.write_text(
+                "ASN,IP Prefix,Max Length,Trust Anchor\n"
+                "AS64496,192.0.2.0/24,24,TA-A\n"
+                "AS64496,192.0.2.0/24,24,TA-B\n"
+                "AS64497,2001:db8::/32,48,TA-A\n",
+                encoding="utf-8",
+            )
+            raw, unique = self.rpctl.collect_rpki_prover_vrps(csv_path)
+            self.assertEqual(raw, 3)
+            self.assertEqual(len(unique), 2)
+
+    def test_run_metadata_marks_vaps_not_applicable(self) -> None:
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            old_ctl_root = self.rpctl.ctl_root
+            self.rpctl.ctl_root = lambda: root
+            try:
+                self.rpctl.save_config({"rpkiProverImage": {"tag": self.rpctl.RPKI_PROVER_IMAGE_TAG}})
+                run_root = root / "runs" / "rpki-prover-incremental" / "test"
+                run_dir = run_root / "profiles" / "rpki-prover-incremental" / "runs" / "run_0001"
+                run_dir.mkdir(parents=True)
+                (run_dir / "start-utc.txt").write_text("2026-07-17T00:00:00Z\n", encoding="utf-8")
+                (run_dir / "end-utc.txt").write_text("2026-07-17T00:01:00Z\n", encoding="utf-8")
+                (run_dir / "vrps.csv").write_text(
+                    "ASN,IP Prefix,Max Length,Trust Anchor\nAS64496,192.0.2.0/24,24,TA-A\n",
+                    encoding="utf-8",
+                )
+                meta = self.rpctl.write_rpki_prover_run_metadata(
+                    run_root, "rpki-prover-incremental", 1, "snapshot", 60_000, 1234, 1200, 0
+                )
+                self.assertEqual(meta["counts"]["vrpsRaw"], 1)
+                self.assertEqual(meta["counts"]["vrpsUnique"], 1)
+                self.assertEqual(meta["counts"]["vaps"], "n/a")
+                self.assertFalse(meta["vapsAvailable"])
+                self.rpctl.write_rpki_prover_group_report(run_root, "rpki-prover-incremental")
+                report = (run_root / "reports" / "per_run_metrics.csv").read_text(encoding="utf-8")
+                self.assertIn("vrps_unique", report)
+                self.assertIn("n/a", report)
+            finally:
+                self.rpctl.ctl_root = old_ctl_root
+
+    def test_missing_image_configuration_fails_before_docker(self) -> None:
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            old_ctl_root = self.rpctl.ctl_root
+            self.rpctl.ctl_root = lambda: root
+            try:
+                self.rpctl.save_config({})
+                with self.assertRaisesRegex(SystemExit, "not initialized"):
+                    self.rpctl.require_rpki_prover_image()
+            finally:
+                self.rpctl.ctl_root = old_ctl_root
+
+    def test_versions_include_prover_profiles(self) -> None:
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            old_ctl_root = self.rpctl.ctl_root
+            self.rpctl.ctl_root = lambda: root
+            try:
+                (root / "versions.json").write_text(json.dumps({"profiles": ["fort-latest-release"]}), encoding="utf-8")
+                self.rpctl.save_config({"rpkiProverImage": {"tag": self.rpctl.RPKI_PROVER_IMAGE_TAG}})
+                run_root = root / "experiment"
+                self.rpctl.write_experiment_versions(run_root)
+                versions = json.loads((run_root / "versions.json").read_text(encoding="utf-8"))
+                self.assertIn("rpki-prover-incremental", versions["profiles"])
+                self.assertIn("rpki-prover-full", versions["profiles"])
+            finally:
+                self.rpctl.ctl_root = old_ctl_root
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/soak/build_portable_soak_package.sh

@@ -14,7 +14,7 @@ Usage:
   scripts/soak/build_portable_soak_package.sh [--out-dir <path>] [--profile <profile>]
 
 Requires release binaries to already exist. Build them first, for example:
-  cargo build --release --bin rpki --bin rpki_daemon --bin db_stats --bin rpki_artifact_metrics --bin rpki_query_service --bin rpki_query_indexer
+  cargo build --release --bin rpki --bin rpki_daemon --bin db_stats --bin rpki_artifact_metrics --bin rpki_query_service --bin rpki_query_indexer --bin verification_ccr_compare
 USAGE
 }
 
@@ -53,7 +53,7 @@ else
   TARGET_BIN_DIR="$REPO_ROOT/target/$PROFILE"
 fi
 
-REQUIRED_BINS=(rpki rpki_daemon db_stats rpki_artifact_metrics rpki_query_service rpki_query_indexer)
+REQUIRED_BINS=(rpki rpki_daemon db_stats rpki_artifact_metrics rpki_query_service rpki_query_indexer verification_ccr_compare)
 OPTIONAL_BINS=(
   ccr_dump
   ccr_state_compare
@@ -82,7 +82,7 @@ STAGE_DIR="$BUILD_ROOT/$PACKAGE_DIR_NAME"
 ARCHIVE_PATH="$OUT_DIR/$PACKAGE_NAME.tar.gz"
 
 rm -rf "$STAGE_DIR" "$ARCHIVE_PATH"
-mkdir -p "$STAGE_DIR/bin" "$STAGE_DIR/fixtures" "$STAGE_DIR/scripts" "$STAGE_DIR/scripts/soak" \
+mkdir -p "$STAGE_DIR/bin" "$STAGE_DIR/fixtures" "$STAGE_DIR/scripts" "$STAGE_DIR/scripts/soak" "$STAGE_DIR/scripts/verification" \
   "$STAGE_DIR/runs" "$STAGE_DIR/state" "$STAGE_DIR/logs" "$STAGE_DIR/tmp"
 
 install -m 0755 "$SCRIPT_DIR/run_soak.sh" "$STAGE_DIR/run_soak.sh"
@@ -92,6 +92,8 @@ install -m 0755 "$SCRIPT_DIR/fixed_phase_loop.sh" "$STAGE_DIR/scripts/soak/fixed
 install -m 0755 "$SCRIPT_DIR/hourly_soak_report.py" "$STAGE_DIR/scripts/soak/hourly_soak_report.py"
 install -m 0755 "$SCRIPT_DIR/publish_remote231.sh" "$STAGE_DIR/scripts/soak/publish_remote231.sh"
 install -m 0755 "$SCRIPT_DIR/publish_remote231_full.sh" "$STAGE_DIR/scripts/soak/publish_remote231_full.sh"
+install -m 0755 "$REPO_ROOT/scripts/verification/run_verification_only.sh" "$STAGE_DIR/scripts/verification/run_verification_only.sh"
+install -m 0644 "$REPO_ROOT/scripts/verification/README.md" "$STAGE_DIR/scripts/verification/README.md"
 install -m 0644 "$SCRIPT_DIR/portable-soak.env.example" "$STAGE_DIR/.env"
 install -m 0644 "$SCRIPT_DIR/portable-soak.env.example" "$STAGE_DIR/portable-soak.env.example"
 

scripts/soak/run_soak.sh

@@ -48,6 +48,7 @@ TMP_DIR="${TMP_DIR:-$RUN_ROOT/tmp}"
 RSYNC_MIRROR_ROOT="${RSYNC_MIRROR_ROOT:-$STATE_ROOT/rsync-mirror}"
 INVALID_ROOT="$STATE_ROOT/invalid"
 RESET_STAGING_ROOT="$STATE_ROOT/reset-staging"
+STATE_EXECUTION_LOCK_PATH="$STATE_ROOT/locks/state-execution.lock"
 LIVE_TA_REFRESH_DIR="${LIVE_TA_REFRESH_DIR:-$META_DIR/live-ta-refresh}"
 LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS="${LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS:-15}"
 LIVE_TA_REFRESH_MAX_TIME_SECS="${LIVE_TA_REFRESH_MAX_TIME_SECS:-120}"
@@ -1080,6 +1081,7 @@ build_child_args() {
 
   CHILD_ARGS+=(
     --report-json "{run_out}/report.json"
+    --validation-contract-out "{run_out}/validation-contract.json"
   )
   if is_true "$OUTPUT_COMPACT_REPORT"; then
     CHILD_ARGS+=(--report-json-compact)
@@ -1109,6 +1111,59 @@ build_child_args() {
   fi
 }
 
+write_current_state_binding() {
+  local run_dir="$1"
+  local run_id="$2"
+  local output_path="$META_DIR/current-state-binding.json"
+  python3 - "$run_dir" "$run_id" "$output_path" <<'PY'
+import hashlib
+import json
+import os
+import sys
+import tempfile
+from datetime import datetime, timezone
+from pathlib import Path
+
+run_dir = Path(sys.argv[1])
+run_id = sys.argv[2]
+output_path = Path(sys.argv[3])
+summary = json.loads((run_dir / "run-summary.json").read_text(encoding="utf-8"))
+if summary.get("status") != "success":
+    raise SystemExit(f"cannot bind unsuccessful source run: {run_id}")
+
+def digest(name):
+    path = run_dir / name
+    if not path.is_file():
+        raise SystemExit(f"current state binding input is missing: {path}")
+    hasher = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            hasher.update(chunk)
+    return hasher.hexdigest()
+
+binding = {
+    "schemaVersion": 1,
+    "currentRunId": run_id,
+    "sourceCirSha256": digest("input.cir"),
+    "sourceCcrSha256": digest("result.ccr"),
+    "validationContractSha256": digest("validation-contract.json"),
+    "updatedAtRfc3339Utc": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
+}
+output_path.parent.mkdir(parents=True, exist_ok=True)
+fd, tmp_name = tempfile.mkstemp(prefix=output_path.name + ".", dir=output_path.parent)
+try:
+    with os.fdopen(fd, "w", encoding="utf-8") as handle:
+        json.dump(binding, handle, indent=2, sort_keys=True)
+        handle.write("\n")
+        handle.flush()
+        os.fsync(handle.fileno())
+    os.replace(tmp_name, output_path)
+finally:
+    if os.path.exists(tmp_name):
+        os.unlink(tmp_name)
+PY
+}
+
 copy_inner_run_outputs() {
   local daemon_state_root="$1"
   local run_dir="$2"
@@ -1304,6 +1359,12 @@ run_one_round() {
       final_status="failed"
     fi
   fi
+  if [[ "$final_status" == "success" ]]; then
+    if ! write_current_state_binding "$run_dir" "$run_id"; then
+      warn "failed to bind successful run to current state: $run_id"
+      final_status="failed"
+    fi
+  fi
   write_run_meta "$run_dir/run-meta.json" "$final_status" "$run_index" "$run_id" "$sync_mode" \
     "$snapshot_reason" "$previous_run_id" "$previous_success_value" "$started_at" "$completed_at" \
     "$INVALID_DB_PATH" "$INVALID_STATE_PATH" "$INVALID_TMP_PATH" "$daemon_exit_code" "$PACKAGE_ROOT" "$ENV_FILE" \
@@ -1327,6 +1388,7 @@ main() {
   require_command python3
   require_command date
   require_command find
+  require_command flock
   if [[ "$TAL_INPUT_MODE" == "file-live-ta" ]]; then
     require_command curl
     validate_positive_int "LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS" "$LIVE_TA_REFRESH_CONNECT_TIMEOUT_SECS"
@@ -1360,7 +1422,7 @@ main() {
     fi
   done
 
-  mkdir -p "$RUNS_ROOT" "$LOG_ROOT" "$DB_DIR" "$META_DIR" "$TMP_DIR" "$INVALID_ROOT" "$RESET_STAGING_ROOT" "$LIVE_TA_REFRESH_DIR"
+  mkdir -p "$RUNS_ROOT" "$LOG_ROOT" "$DB_DIR" "$META_DIR" "$TMP_DIR" "$INVALID_ROOT" "$RESET_STAGING_ROOT" "$LIVE_TA_REFRESH_DIR" "$(dirname "$STATE_EXECUTION_LOCK_PATH")"
   if is_true "$ALLOW_RSYNC_MIRROR_REUSE"; then
     mkdir -p "$RSYNC_MIRROR_ROOT"
   fi
@@ -1386,6 +1448,9 @@ main() {
   local any_failed=0
 
   while (( run_forever == 1 || next_index <= stop_index )); do
+    local state_lock_fd
+    exec {state_lock_fd}>"$STATE_EXECUTION_LOCK_PATH"
+    flock "$state_lock_fd"
     INVALID_DB_PATH=""
     INVALID_STATE_PATH=""
     INVALID_TMP_PATH=""
@@ -1472,6 +1537,8 @@ main() {
       echo "completed run $(printf 'run_%04d' "$next_index") status=failed" >&2
       any_failed=1
     fi
+    flock -u "$state_lock_fd"
+    exec {state_lock_fd}>&-
     if (( STOP_AFTER_SECS > 0 )); then
       elapsed_secs=$(( $(date +%s) - started_epoch ))
       if (( elapsed_secs >= STOP_AFTER_SECS )); then

scripts/verification/README.md

@@ -0,0 +1,50 @@
+# verification-only 使用说明
+
+`verification-only` 用于复核启用验证缓存的最新成功 normal run。它冻结当前仓库视图，关闭 PP、ROA、child certificate cache、prefetch、VCIR 持久化和网络访问，在源 CIR 的同一验证时间重新执行完整验证，并比较 CCR 的 MFT、VRP、VAP、TA、router key 五类 state digest。
+
+## 前提
+
+- 源 run 必须是当前 state DB 对应的最新成功 normal run。
+- 源 run 必须包含 `input.cir`、`result.ccr`、`run-summary.json` 和 `validation-contract.json`。
+- state root 必须包含 `db/work-db`、`db/repo-bytes.db` 和 `meta/current-state-binding.json`。
+- 源 normal run 必须启用至少一种验证缓存；仅启用 transport prefetch 不算验证缓存。
+- verification-only 与 normal soak 共用 `state/locks/state-execution.lock`，不能绕过 wrapper 并发执行。
+
+## 执行
+
+在 portable package 根目录执行：
+
+```bash
+./scripts/verification/run_verification_only.sh \
+  --source-run-dir /data/ours-rp/runs/run_0123 \
+  --source-state-root /data/ours-rp/state \
+  --out-dir /data/ours-rp/verification-runs/run_0123
+```
+
+验证时间自动从源 `input.cir` 读取，不接受命令行覆盖。成功后 scratch work-db 默认删除；排障时可加 `--keep-scratch`。
+
+## 输出
+
+- `input.cir`、`result.ccr`、`report.json`
+- `vrps.csv`、`vaps.csv`
+- `stage-timing.json`、`validation-events.jsonl`
+- `validation-contract.json`、`verification-meta.json`
+- `compare/ccr-state-compare.json`、`compare/ccr-state-compare.md`
+- `stdout.log`、`stderr.log`、`process-time.txt`、`wrapper-timing.json`
+
+通过条件是 compare JSON 中 `stateDigestMatch=true`。`producedAt` 不参与比较。
+
+## 隔离保证
+
+- 不分配 normal run id，不更新 lifecycle、periodic snapshot 计数和 retention。
+- 源 work-db 只用于创建 checkpoint，后续写入只发生在独立 scratch work-db。
+- 源 repo-bytes.db 以只读模式打开。
+- 输出目录必须不存在，防止覆盖旧验证证据。
+
+## 常见失败
+
+- `current state binding`：所选 run 已不是当前 state 对应的最新成功 run，需选择 binding 指向的 run。
+- `differs from current state binding`：源 CIR、CCR 或 contract 已变化，不能继续复核。
+- `blob is missing`：repository view 引用的原始 bytes 不完整，验证不会启动。
+- `did not enable a validation cache`：源 run 不是缓存正确性对比的有效基准。
+- `CCR state digest mismatch`：查看 `compare/` 下 JSON/Markdown，按不匹配状态集合继续排查。

scripts/verification/run_verification_only.sh

@@ -0,0 +1,102 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+usage() {
+  cat <<'USAGE'
+Usage:
+  run_verification_only.sh \
+    --source-run-dir <run_dir> \
+    --source-state-root <state_dir> \
+    --out-dir <verification_dir> \
+    [--rpki-bin <path>] [--keep-scratch]
+
+validation time is always read from the source CIR. The wrapper serializes access
+with normal soak runs and never writes into the normal run sequence.
+USAGE
+}
+
+SOURCE_RUN_DIR=""
+SOURCE_STATE_ROOT=""
+OUT_DIR=""
+RPKI_BIN=""
+KEEP_SCRATCH=0
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --source-run-dir) SOURCE_RUN_DIR="${2:?missing value}"; shift 2 ;;
+    --source-state-root) SOURCE_STATE_ROOT="${2:?missing value}"; shift 2 ;;
+    --out-dir) OUT_DIR="${2:?missing value}"; shift 2 ;;
+    --rpki-bin) RPKI_BIN="${2:?missing value}"; shift 2 ;;
+    --keep-scratch) KEEP_SCRATCH=1; shift ;;
+    -h|--help) usage; exit 0 ;;
+    *) echo "unknown argument: $1" >&2; usage >&2; exit 2 ;;
+  esac
+done
+
+[[ -n "$SOURCE_RUN_DIR" ]] || { echo "--source-run-dir is required" >&2; exit 2; }
+[[ -n "$SOURCE_STATE_ROOT" ]] || { echo "--source-state-root is required" >&2; exit 2; }
+[[ -n "$OUT_DIR" ]] || { echo "--out-dir is required" >&2; exit 2; }
+if [[ -z "$RPKI_BIN" ]]; then
+  RPKI_BIN="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)/target/release/rpki"
+fi
+[[ -x "$RPKI_BIN" ]] || { echo "rpki binary is not executable: $RPKI_BIN" >&2; exit 2; }
+command -v flock >/dev/null || { echo "flock is required" >&2; exit 2; }
+command -v python3 >/dev/null || { echo "python3 is required" >&2; exit 2; }
+
+now_ms() {
+  python3 -c 'import time; print(time.time_ns() // 1_000_000)'
+}
+
+mkdir -p "$SOURCE_STATE_ROOT/locks" "$(dirname "$OUT_DIR")"
+[[ ! -e "$OUT_DIR" ]] || { echo "verification output already exists: $OUT_DIR" >&2; exit 2; }
+LOCK_PATH="$SOURCE_STATE_ROOT/locks/state-execution.lock"
+STDOUT_TMP="${OUT_DIR}.stdout.tmp"
+STDERR_TMP="${OUT_DIR}.stderr.tmp"
+PROCESS_TIME_TMP="${OUT_DIR}.process-time.tmp"
+WAIT_STARTED_MS="$(now_ms)"
+exec {LOCK_FD}>"$LOCK_PATH"
+flock "$LOCK_FD"
+LOCK_ACQUIRED_MS="$(now_ms)"
+
+ARGS=(
+  --verification-only
+  --verification-source-run "$SOURCE_RUN_DIR"
+  --verification-source-state "$SOURCE_STATE_ROOT"
+  --verification-out "$OUT_DIR"
+)
+if [[ "$KEEP_SCRATCH" -eq 1 ]]; then
+  ARGS+=(--keep-verification-scratch)
+fi
+
+set +e
+if [[ -x /usr/bin/time ]]; then
+  RPKI_STATE_EXECUTION_LOCK_HELD=1 /usr/bin/time -v -o "$PROCESS_TIME_TMP" -- \
+    "$RPKI_BIN" "${ARGS[@]}" >"$STDOUT_TMP" 2>"$STDERR_TMP"
+else
+  RPKI_STATE_EXECUTION_LOCK_HELD=1 "$RPKI_BIN" "${ARGS[@]}" >"$STDOUT_TMP" 2>"$STDERR_TMP"
+fi
+EXIT_CODE=$?
+set -e
+LOCK_RELEASED_MS="$(now_ms)"
+flock -u "$LOCK_FD"
+exec {LOCK_FD}>&-
+
+mkdir -p "$OUT_DIR"
+mv "$STDOUT_TMP" "$OUT_DIR/stdout.log"
+mv "$STDERR_TMP" "$OUT_DIR/stderr.log"
+if [[ -f "$PROCESS_TIME_TMP" ]]; then
+  mv "$PROCESS_TIME_TMP" "$OUT_DIR/process-time.txt"
+fi
+python3 - "$OUT_DIR/wrapper-timing.json" "$WAIT_STARTED_MS" "$LOCK_ACQUIRED_MS" "$LOCK_RELEASED_MS" "$EXIT_CODE" <<'PY'
+import json
+import sys
+from pathlib import Path
+
+path = Path(sys.argv[1])
+wait_started_ms, lock_acquired_ms, lock_released_ms, exit_code = map(int, sys.argv[2:])
+path.write_text(json.dumps({
+    "lockWaitMs": lock_acquired_ms - wait_started_ms,
+    "lockHeldMs": lock_released_ms - lock_acquired_ms,
+    "exitCode": exit_code,
+}, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+PY
+exit "$EXIT_CODE"

src/bin/verification_ccr_compare.rs

@@ -0,0 +1,171 @@
+use std::path::PathBuf;
+
+fn usage() -> &'static str {
+    "Usage: verification_ccr_compare --source-ccr <path> --verification-ccr <path> --out-json <path> [--out-md <path>]"
+}
+
+#[derive(Debug, PartialEq, Eq)]
+struct Args {
+    source_ccr: PathBuf,
+    verification_ccr: PathBuf,
+    out_json: PathBuf,
+    out_md: Option<PathBuf>,
+}
+
+fn parse_args(argv: &[String]) -> Result<Args, String> {
+    let mut source_ccr = None;
+    let mut verification_ccr = None;
+    let mut out_json = None;
+    let mut out_md = None;
+    let mut index = 1usize;
+    while index < argv.len() {
+        let target = match argv[index].as_str() {
+            "--source-ccr" => &mut source_ccr,
+            "--verification-ccr" => &mut verification_ccr,
+            "--out-json" => &mut out_json,
+            "--out-md" => &mut out_md,
+            "-h" | "--help" => return Err(usage().to_string()),
+            other => return Err(format!("unknown argument: {other}\n{}", usage())),
+        };
+        index += 1;
+        *target =
+            Some(PathBuf::from(argv.get(index).ok_or_else(|| {
+                format!("{} requires a value", argv[index - 1])
+            })?));
+        index += 1;
+    }
+    let source_ccr = source_ccr.ok_or_else(|| format!("--source-ccr is required\n{}", usage()))?;
+    let verification_ccr =
+        verification_ccr.ok_or_else(|| format!("--verification-ccr is required\n{}", usage()))?;
+    let out_json = out_json.ok_or_else(|| format!("--out-json is required\n{}", usage()))?;
+    Ok(Args {
+        source_ccr,
+        verification_ccr,
+        out_json,
+        out_md,
+    })
+}
+
+fn run(args: &Args) -> Result<(), String> {
+    let comparison =
+        rpki::verification_only::compare_ccr_files(&args.source_ccr, &args.verification_ccr)?;
+    rpki::verification_only::write_ccr_comparison(
+        &args.out_json,
+        args.out_md.as_deref(),
+        &comparison,
+    )?;
+    if !comparison.state_digest_match {
+        return Err(format!(
+            "CCR state digest mismatch: {}",
+            comparison.mismatched_states.join(",")
+        ));
+    }
+    println!("CCR state digests match: {}", args.out_json.display());
+    Ok(())
+}
+
+fn main() -> Result<(), String> {
+    let argv = std::env::args().collect::<Vec<_>>();
+    run(&parse_args(&argv)?)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rpki::ccr::{
+        CcrContentInfo, CcrDigestAlgorithm, RpkiCanonicalCacheRepresentation,
+        build_roa_payload_state, encode_content_info,
+    };
+    use rpki::data_model::roa::{IpPrefix, RoaAfi};
+    use rpki::validation::objects::Vrp;
+    use std::fs;
+
+    fn argv(values: &[&str]) -> Vec<String> {
+        std::iter::once("verification_ccr_compare")
+            .chain(values.iter().copied())
+            .map(str::to_string)
+            .collect()
+    }
+
+    fn sample_ccr(asn: u32) -> Vec<u8> {
+        let vrps = build_roa_payload_state(&[Vrp {
+            asn,
+            prefix: IpPrefix {
+                afi: RoaAfi::Ipv4,
+                prefix_len: 24,
+                addr: [192, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+            },
+            max_length: 24,
+        }])
+        .expect("build VRP state");
+        encode_content_info(&CcrContentInfo::new(RpkiCanonicalCacheRepresentation {
+            version: 0,
+            hash_alg: CcrDigestAlgorithm::Sha256,
+            produced_at: time::OffsetDateTime::UNIX_EPOCH,
+            mfts: None,
+            vrps: Some(vrps),
+            vaps: None,
+            tas: None,
+            rks: None,
+        }))
+        .expect("encode CCR")
+    }
+
+    #[test]
+    fn parse_args_accepts_complete_input_and_rejects_bad_forms() {
+        let parsed = parse_args(&argv(&[
+            "--source-ccr",
+            "source.ccr",
+            "--verification-ccr",
+            "verification.ccr",
+            "--out-json",
+            "result.json",
+            "--out-md",
+            "result.md",
+        ]))
+        .expect("parse");
+        assert_eq!(parsed.source_ccr, PathBuf::from("source.ccr"));
+        assert_eq!(parsed.out_md, Some(PathBuf::from("result.md")));
+        assert!(parse_args(&argv(&[])).unwrap_err().contains("source-ccr"));
+        assert!(
+            parse_args(&argv(&["--source-ccr"]))
+                .unwrap_err()
+                .contains("requires a value")
+        );
+        assert!(
+            parse_args(&argv(&["--unknown"]))
+                .unwrap_err()
+                .contains("unknown argument")
+        );
+        assert!(parse_args(&argv(&["--help"])).is_err());
+    }
+
+    #[test]
+    fn run_writes_matching_outputs_and_rejects_mismatch() {
+        let temp = tempfile::tempdir().expect("tempdir");
+        let source = temp.path().join("source.ccr");
+        let verification = temp.path().join("verification.ccr");
+        let out_json = temp.path().join("out/result.json");
+        let out_md = temp.path().join("out/result.md");
+        fs::write(&source, sample_ccr(64496)).expect("source");
+        fs::write(&verification, sample_ccr(64496)).expect("verification");
+        let args = Args {
+            source_ccr: source.clone(),
+            verification_ccr: verification.clone(),
+            out_json: out_json.clone(),
+            out_md: Some(out_md.clone()),
+        };
+        run(&args).expect("matching run");
+        assert!(out_json.is_file());
+        assert!(out_md.is_file());
+
+        fs::write(&verification, sample_ccr(64497)).expect("mismatch");
+        let error = run(&args).unwrap_err();
+        assert!(error.contains("vrps"));
+        assert!(
+            fs::read_to_string(out_json)
+                .expect("comparison JSON")
+                .contains("mismatchedStates")
+        );
+    }
+}

src/blob_store.rs

@@ -77,6 +77,7 @@ pub struct ExternalRawStoreDb {
 pub struct ExternalRepoBytesDb {
     path: PathBuf,
     db: Arc<DB>,
+    read_only: bool,
 }
 
 impl ExternalRawStoreDb {
@@ -189,6 +190,7 @@ impl ExternalRepoBytesDb {
         Ok(Self {
             path,
             db: Arc::new(db),
+            read_only: false,
         })
     }
 
@@ -201,6 +203,7 @@ impl ExternalRepoBytesDb {
         Ok(Self {
             path,
             db: Arc::new(db),
+            read_only: true,
         })
     }
 
@@ -208,6 +211,12 @@ impl ExternalRepoBytesDb {
         if blobs.is_empty() {
             return Ok(());
         }
+        if self.read_only {
+            return Err(StorageError::RocksDb(format!(
+                "repo-bytes DB is read-only: {}",
+                self.path.display()
+            )));
+        }
         let mut batch = WriteBatch::default();
         for (sha256_hex, bytes) in blobs {
             validate_blob_sha256_hex(sha256_hex)?;
@@ -221,6 +230,42 @@ impl ExternalRepoBytesDb {
         Ok(())
     }
 
+    pub(crate) fn is_read_only(&self) -> bool {
+        self.read_only
+    }
+
+    pub(crate) fn require_existing_blob_bytes_batch(
+        &self,
+        blobs: &[(String, Vec<u8>)],
+    ) -> StorageResult<()> {
+        if blobs.is_empty() {
+            return Ok(());
+        }
+        let hashes = blobs
+            .iter()
+            .map(|(sha256_hex, _)| sha256_hex.clone())
+            .collect::<Vec<_>>();
+        let existing = self.get_blob_bytes_batch(&hashes)?;
+        for ((sha256_hex, expected_bytes), actual_bytes) in blobs.iter().zip(existing) {
+            match actual_bytes {
+                Some(actual_bytes) if actual_bytes == *expected_bytes => {}
+                Some(_) => {
+                    return Err(StorageError::InvalidData {
+                        entity: "read_only_repo_bytes",
+                        detail: format!("existing bytes differ for SHA-256 {sha256_hex}"),
+                    });
+                }
+                None => {
+                    return Err(StorageError::InvalidData {
+                        entity: "read_only_repo_bytes",
+                        detail: format!("blob is missing for SHA-256 {sha256_hex}"),
+                    });
+                }
+            }
+        }
+        Ok(())
+    }
+
     pub fn get_blob_bytes(&self, sha256_hex: &str) -> StorageResult<Option<Vec<u8>>> {
         validate_blob_sha256_hex(sha256_hex)?;
         let key = repo_bytes_key(sha256_hex);

src/cli.rs

@@ -38,7 +38,7 @@ use crate::validation::run_tree_from_tal::{
     run_tree_from_tal_url_parallel_phase2_audit,
     run_tree_from_tal_url_parallel_phase2_audit_with_timing,
 };
-use crate::validation::tree::TreeRunConfig;
+use crate::validation::tree::{DEFAULT_MAX_CA_DEPTH, TreeRunConfig};
 #[cfg(test)]
 use output::write_json;
 use output::{
@@ -115,6 +115,12 @@ fn vcir_storage_summary_enabled() -> bool {
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct CliArgs {
+    pub verification_only: bool,
+    pub verification_source_run_dir: Option<PathBuf>,
+    pub verification_source_state_root: Option<PathBuf>,
+    pub verification_out_dir: Option<PathBuf>,
+    pub keep_verification_scratch: bool,
+    pub validation_contract_out_path: Option<PathBuf>,
     pub tal_urls: Vec<String>,
     pub tal_paths: Vec<PathBuf>,
     pub ta_paths: Vec<PathBuf>,
@@ -168,7 +174,7 @@ pub struct CliArgs {
     pub rsync_mirror_root: Option<PathBuf>,
     pub rsync_scope_policy: RsyncScopePolicy,
 
-    pub max_depth: Option<usize>,
+    pub max_ca_depth: usize,
     pub max_instances: Option<usize>,
     pub validation_time: Option<time::OffsetDateTime>,
 
@@ -184,8 +190,19 @@ fn usage() -> String {
 Usage:
   {bin} --db <path> --tal-url <url> [--tal-url <url> ...] [options]
   {bin} --db <path> --tal-path <path> --ta-path <path> [--tal-path <path> --ta-path <path> ...] [options]
+  {bin} --verification-only --verification-source-run <path> --verification-source-state <path> --verification-out <path> [options]
 
 Options:
+  --verification-only       Revalidate the latest successful run without validation caches or network access
+  --verification-source-run <path>
+                             Source normal run directory containing input.cir/result.ccr/validation-contract.json
+  --verification-source-state <path>
+                             Source state root containing db/work-db and db/repo-bytes.db
+  --verification-out <path> Independent output directory for the one-shot verification run
+  --keep-verification-scratch
+                             Keep the private work-db checkpoint after a successful verification
+  --validation-contract-out <path>
+                             Write the effective normal-run validation contract (not valid in verification-only mode)
   --db <path>                RocksDB directory path (required)
   --raw-store-db <path>      External raw-by-hash store DB path (optional)
   --repo-bytes-db <path>     External repo object bytes DB path (optional)
@@ -259,7 +276,8 @@ Options:
   --rsync-timeout-secs <n>   rsync I/O timeout seconds (default: 60)
   --rsync-mirror-root <path> Persist rsync mirrors under this directory (default: disabled)
   --rsync-scope <policy>     rsync scope policy: host, publication-point, or module-root (default: module-root)
-  --max-depth <n>            Max CA instance depth (0 = root only)
+  --max-ca-depth <n>         Maximum CA depth from a trust anchor (root = 0, default: {DEFAULT_MAX_CA_DEPTH})
+  --max-depth <n>            Deprecated alias for --max-ca-depth
   --max-instances <n>        Max number of CA instances to process
   --validation-time <rfc3339> Validation time in RFC3339 (default: now UTC)
   --analyze                  Write timing analysis JSON under target/live/analyze/<timestamp>/
@@ -272,6 +290,12 @@ Options:
 }
 
 pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
+    let mut verification_only = false;
+    let mut verification_source_run_dir: Option<PathBuf> = None;
+    let mut verification_source_state_root: Option<PathBuf> = None;
+    let mut verification_out_dir: Option<PathBuf> = None;
+    let mut keep_verification_scratch = false;
+    let mut validation_contract_out_path: Option<PathBuf> = None;
     let mut tal_urls: Vec<String> = Vec::new();
     let mut tal_paths: Vec<PathBuf> = Vec::new();
     let mut ta_paths: Vec<PathBuf> = Vec::new();
@@ -319,7 +343,8 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
     let mut rsync_timeout_secs: u64 = 30;
     let mut rsync_mirror_root: Option<PathBuf> = None;
     let mut rsync_scope_policy = RsyncScopePolicy::default();
-    let mut max_depth: Option<usize> = None;
+    let mut max_ca_depth: Option<usize> = None;
+    let mut max_ca_depth_option: Option<&'static str> = None;
     let mut max_instances: Option<usize> = None;
     let mut validation_time: Option<time::OffsetDateTime> = None;
     let mut analyze: bool = false;
@@ -331,6 +356,39 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
         let arg = argv[i].as_str();
         match arg {
             "--help" | "-h" => return Err(usage()),
+            "--verification-only" => {
+                verification_only = true;
+            }
+            "--verification-source-run" => {
+                i += 1;
+                verification_source_run_dir = Some(PathBuf::from(
+                    argv.get(i)
+                        .ok_or("--verification-source-run requires a value")?,
+                ));
+            }
+            "--verification-source-state" => {
+                i += 1;
+                verification_source_state_root = Some(PathBuf::from(
+                    argv.get(i)
+                        .ok_or("--verification-source-state requires a value")?,
+                ));
+            }
+            "--verification-out" => {
+                i += 1;
+                verification_out_dir = Some(PathBuf::from(
+                    argv.get(i).ok_or("--verification-out requires a value")?,
+                ));
+            }
+            "--keep-verification-scratch" => {
+                keep_verification_scratch = true;
+            }
+            "--validation-contract-out" => {
+                i += 1;
+                validation_contract_out_path = Some(PathBuf::from(
+                    argv.get(i)
+                        .ok_or("--validation-contract-out requires a value")?,
+                ));
+            }
             "--tal-url" => {
                 i += 1;
                 let v = argv.get(i).ok_or("--tal-url requires a value")?;
@@ -661,13 +719,26 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
                 let v = argv.get(i).ok_or("--rsync-scope requires a value")?;
                 rsync_scope_policy = RsyncScopePolicy::parse_cli_value(v)?;
             }
-            "--max-depth" => {
+            "--max-ca-depth" | "--max-depth" => {
                 i += 1;
-                let v = argv.get(i).ok_or("--max-depth requires a value")?;
+                let option = arg;
-                max_depth = Some(
+                let v = argv
+                    .get(i)
+                    .ok_or_else(|| format!("{option} requires a value"))?;
+                if let Some(previous_option) = max_ca_depth_option {
+                    return Err(format!(
+                        "{option} cannot be combined with {previous_option}; use only --max-ca-depth"
+                    ));
+                }
+                max_ca_depth = Some(
                     v.parse::<usize>()
-                        .map_err(|_| format!("invalid --max-depth: {v}"))?,
+                        .map_err(|_| format!("invalid {option}: {v}"))?,
                 );
+                max_ca_depth_option = Some(match option {
+                    "--max-ca-depth" => "--max-ca-depth",
+                    "--max-depth" => "--max-depth",
+                    _ => unreachable!("matched CA depth option"),
+                });
             }
             "--max-instances" => {
                 i += 1;
@@ -702,10 +773,91 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
         i += 1;
     }
 
+    if verification_only {
+        if verification_source_run_dir.is_none() {
+            return Err(format!(
+                "--verification-only requires --verification-source-run\n\n{}",
+                usage()
+            ));
+        }
+        if verification_source_state_root.is_none() {
+            return Err(format!(
+                "--verification-only requires --verification-source-state\n\n{}",
+                usage()
+            ));
+        }
+        if verification_out_dir.is_none() {
+            return Err(format!(
+                "--verification-only requires --verification-out\n\n{}",
+                usage()
+            ));
+        }
+        if db_path.is_some()
+            || raw_store_db.is_some()
+            || repo_bytes_db.is_some()
+            || !tal_urls.is_empty()
+            || !tal_paths.is_empty()
+            || !ta_paths.is_empty()
+            || policy_path.is_some()
+            || strict_policy.is_some()
+            || resource_validation_mode.is_some()
+            || validation_time.is_some()
+            || max_ca_depth_option.is_some()
+            || max_instances.is_some()
+            || enable_roa_validation_cache
+            || enable_child_certificate_validation_cache
+            || publication_point_cache_observe_only
+            || enable_publication_point_validation_cache
+            || enable_transport_request_prefetch
+            || rsync_local_dir.is_some()
+            || rsync_command.is_some()
+            || !http_root_cert_paths.is_empty()
+            || payload_replay_archive.is_some()
+            || payload_replay_locks.is_some()
+            || payload_base_archive.is_some()
+            || payload_base_locks.is_some()
+            || payload_delta_archive.is_some()
+            || payload_delta_locks.is_some()
+            || validation_contract_out_path.is_some()
+        {
+            return Err(format!(
+                "verification-only derives DB, TAL, validation time, policy and cache settings from the source run; conflicting normal-run options are not allowed\n\n{}",
+                usage()
+            ));
+        }
+        if report_json_path.is_some()
+            || ccr_out_path.is_some()
+            || vrps_csv_out_path.is_some()
+            || vaps_csv_out_path.is_some()
+            || cir_enabled
+            || cir_out_path.is_some()
+        {
+            return Err(format!(
+                "verification-only output paths are derived from --verification-out\n\n{}",
+                usage()
+            ));
+        }
+        db_path = Some(
+            verification_out_dir
+                .as_ref()
+                .expect("checked verification out")
+                .join("scratch/work-db"),
+        );
+    } else if verification_source_run_dir.is_some()
+        || verification_source_state_root.is_some()
+        || verification_out_dir.is_some()
+        || keep_verification_scratch
+    {
+        return Err(format!(
+            "verification source/output options require --verification-only\n\n{}",
+            usage()
+        ));
+    }
+
     let db_path = db_path.ok_or_else(|| format!("--db is required\n\n{}", usage()))?;
 
     let tal_mode_count = (!tal_urls.is_empty()) as u8 + (!tal_paths.is_empty()) as u8;
-    if tal_mode_count != 1 {
+    if !verification_only && tal_mode_count != 1 {
         return Err(format!(
             "must specify either one-or-more --tal-url or one-or-more --tal-path/--ta-path pairs\n\n{}",
             usage()
@@ -759,13 +911,13 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
             usage()
         ));
     }
-    if !tal_urls.is_empty() && !ta_paths.is_empty() {
+    if !verification_only && !tal_urls.is_empty() && !ta_paths.is_empty() {
         return Err(format!(
             "--ta-path cannot be used with --tal-url mode\n\n{}",
             usage()
         ));
     }
-    if !tal_paths.is_empty() {
+    if !verification_only && !tal_paths.is_empty() {
         if !ta_paths.is_empty() {
             if ta_paths.len() != tal_paths.len() {
                 return Err(format!(
@@ -940,6 +1092,12 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
     }
 
     Ok(CliArgs {
+        verification_only,
+        verification_source_run_dir,
+        verification_source_state_root,
+        verification_out_dir,
+        keep_verification_scratch,
+        validation_contract_out_path,
         tal_urls,
         tal_paths,
         ta_paths,
@@ -989,7 +1147,7 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
         rsync_timeout_secs,
         rsync_mirror_root,
         rsync_scope_policy,
-        max_depth,
+        max_ca_depth: max_ca_depth.unwrap_or(DEFAULT_MAX_CA_DEPTH),
         max_instances,
         validation_time,
         analyze,
@@ -1854,7 +2012,7 @@ where
     H: crate::sync::rrdp::Fetcher + Clone + 'static,
     R: crate::fetch::rsync::RsyncFetcher + Clone + 'static,
 {
-    if args.tal_inputs.len() > 1 {
+    if args.verification_only || args.tal_inputs.len() > 1 {
         return if let Some(t) = timing {
             run_tree_from_multiple_tals_parallel_phase2_audit_with_timing(
                 store,
@@ -1995,21 +2153,88 @@ where
 }
 
 pub fn run(argv: &[String]) -> Result<(), String> {
-    let args = parse_args(argv)?;
+    let mut args = parse_args(argv)?;
+    let mut prepared_verification = if args.verification_only {
+        let prepared = crate::verification_only::prepare_verification(
+            args.verification_source_run_dir
+                .as_deref()
+                .expect("validated by parse_args"),
+            args.verification_source_state_root
+                .as_deref()
+                .expect("validated by parse_args"),
+            args.verification_out_dir
+                .as_deref()
+                .expect("validated by parse_args"),
+        )?;
+        args.db_path = prepared.artifacts.scratch_work_db.clone();
+        args.repo_bytes_db = Some(prepared.artifacts.source_repo_bytes_db.clone());
+        args.report_json_path = Some(prepared.artifacts.report_json.clone());
+        args.report_json_compact = true;
+        args.skip_report_build = false;
+        args.skip_vcir_persist = true;
+        args.enable_roa_validation_cache = false;
+        args.enable_child_certificate_validation_cache = false;
+        args.publication_point_cache_observe_only = false;
+        args.enable_publication_point_validation_cache = false;
+        args.enable_transport_request_prefetch = false;
+        args.ccr_out_path = Some(prepared.artifacts.result_ccr.clone());
+        args.vrps_csv_out_path = Some(prepared.artifacts.vrps_csv.clone());
+        args.vaps_csv_out_path = Some(prepared.artifacts.vaps_csv.clone());
+        args.compare_view_trust_anchor = Some("verification-only".to_string());
+        args.cir_enabled = true;
+        args.cir_out_path = Some(prepared.artifacts.result_cir.clone());
+        args.cir_tal_uris = crate::verification_only::cir_tal_uris(&prepared.cir);
+        args.tal_inputs = crate::verification_only::tal_inputs_from_cir(&prepared.cir);
+        args.validation_time = Some(prepared.cir.validation_time);
+        args.max_ca_depth = prepared.contract.max_ca_depth;
+        args.max_instances = prepared.contract.max_instances;
+        args.disable_rrdp = true;
+        args.rsync_local_dir = None;
+        Some(prepared)
+    } else {
+        None
+    };
 
-    let mut policy = read_policy(args.policy_path.as_deref())?;
+    let mut policy = if let Some(prepared) = prepared_verification.as_ref() {
-    if let Some(strict_policy) = args.strict_policy {
+        prepared.contract.policy.clone()
-        policy.strict = strict_policy;
+    } else {
-    }
+        read_policy(args.policy_path.as_deref())?
-    if let Some(resource_validation_mode) = args.resource_validation_mode {
+    };
-        policy.resource_validation_mode = resource_validation_mode;
+    if !args.verification_only {
-    }
+        if let Some(strict_policy) = args.strict_policy {
-    if args.disable_rrdp {
+            policy.strict = strict_policy;
+        }
+        if let Some(resource_validation_mode) = args.resource_validation_mode {
+            policy.resource_validation_mode = resource_validation_mode;
+        }
+        if args.disable_rrdp {
+            policy.sync_preference = crate::policy::SyncPreference::RsyncOnly;
+        }
+    } else {
         policy.sync_preference = crate::policy::SyncPreference::RsyncOnly;
+        policy.ca_failed_fetch_policy = crate::policy::CaFailedFetchPolicy::StopAllOutput;
     }
     let validation_time = args
         .validation_time
         .unwrap_or_else(time::OffsetDateTime::now_utc);
+    let validation_time =
+        time::OffsetDateTime::from_unix_timestamp(validation_time.unix_timestamp())
+            .map_err(|error| format!("normalize validation time failed: {error}"))?;
+    if let Some(path) = args.validation_contract_out_path.as_deref() {
+        let contract = crate::verification_only::ValidationContract::for_current_binary(
+            validation_time,
+            policy.clone(),
+            args.max_ca_depth,
+            args.max_instances,
+            crate::verification_only::ValidationCacheContract {
+                publication_point: args.enable_publication_point_validation_cache,
+                roa: args.enable_roa_validation_cache,
+                child_certificate: args.enable_child_certificate_validation_cache,
+                transport_prefetch: args.enable_transport_request_prefetch,
+            },
+        )?;
+        crate::verification_only::write_validation_contract(path, &contract)?;
+    }
     let http_root_certificates_pem = args
         .http_root_cert_paths
         .iter()
@@ -2019,7 +2244,12 @@ pub fn run(argv: &[String]) -> Result<(), String> {
         })
         .collect::<Result<Vec<_>, _>>()?;
 
-    let store = if args.raw_store_db.is_some() || args.repo_bytes_db.is_some() {
+    let store = if let Some(prepared) = prepared_verification.as_mut() {
+        prepared
+            .store
+            .take()
+            .expect("prepared verification store must be available")
+    } else if args.raw_store_db.is_some() || args.repo_bytes_db.is_some() {
         Arc::new(
             RocksStore::open_with_external_stores(
                 &args.db_path,
@@ -2032,7 +2262,7 @@ pub fn run(argv: &[String]) -> Result<(), String> {
         Arc::new(RocksStore::open(&args.db_path).map_err(|e| e.to_string())?)
     };
     let config = TreeRunConfig {
-        max_depth: args.max_depth,
+        max_depth: Some(args.max_ca_depth),
         max_instances: args.max_instances,
         compact_audit: args.skip_report_build
             && args.report_json_path.is_none()
@@ -2122,7 +2352,23 @@ pub fn run(argv: &[String]) -> Result<(), String> {
     );
     let validation_started = std::time::Instant::now();
     let collect_current_repo_objects = false;
-    let out = if delta_replay_mode {
+    let out = if args.verification_only {
+        let http = crate::verification_only::NetworkDisabledHttpFetcher;
+        let rsync = crate::fetch::current_repository::CurrentRepositoryViewRsyncFetcher::new(
+            Arc::clone(&store),
+        );
+        run_online_validation_with_fetchers(
+            Arc::clone(&store),
+            &policy,
+            &args,
+            &http,
+            &rsync,
+            validation_time,
+            &config,
+            collect_current_repo_objects,
+            timing.as_ref().map(|(_, t)| t),
+        )?
+    } else if delta_replay_mode {
         let tal_path = args
             .tal_path
             .as_ref()
@@ -2651,6 +2897,62 @@ pub fn run(argv: &[String]) -> Result<(), String> {
     }
 
     print_summary_from_shared(validation_time, &shared);
+    if let Some(prepared) = prepared_verification.as_ref() {
+        std::fs::create_dir_all(&prepared.artifacts.compare_dir).map_err(|error| {
+            format!(
+                "create verification compare directory failed: {}: {error}",
+                prepared.artifacts.compare_dir.display()
+            )
+        })?;
+        let comparison = crate::verification_only::compare_ccr_files(
+            &prepared.artifacts.source_ccr,
+            &prepared.artifacts.result_ccr,
+        )?;
+        crate::verification_only::write_ccr_comparison(
+            &prepared.artifacts.compare_dir.join("ccr-state-digest.json"),
+            Some(&prepared.artifacts.compare_dir.join("ccr-state-digest.md")),
+            &comparison,
+        )?;
+        let scratch_retained = !comparison.state_digest_match || args.keep_verification_scratch;
+        let source_run_id = prepared
+            .artifacts
+            .source_run_dir
+            .file_name()
+            .and_then(|value| value.to_str())
+            .unwrap_or("unknown")
+            .to_string();
+        crate::verification_only::write_verification_meta(
+            &prepared.artifacts.verification_meta,
+            &crate::verification_only::VerificationRunMeta {
+                schema_version: 1,
+                status: if comparison.state_digest_match {
+                    "success".to_string()
+                } else {
+                    "mismatch".to_string()
+                },
+                source_run_id,
+                validation_time: crate::verification_only::format_validation_time(validation_time)?,
+                wall_ms: total_started.elapsed().as_millis() as u64,
+                repository_objects_checked: prepared.blob_summary.current_objects,
+                repository_bytes_checked: prepared.blob_summary.bytes_verified,
+                state_digest_match: comparison.state_digest_match,
+                scratch_retained,
+            },
+        )?;
+        if !comparison.state_digest_match {
+            return Err(format!(
+                "verification-only CCR state digest mismatch: {}",
+                comparison.mismatched_states.join(",")
+            ));
+        }
+        if !args.keep_verification_scratch {
+            drop(shared);
+            drop(store);
+            std::fs::remove_dir_all(prepared.artifacts.output_dir.join("scratch")).map_err(
+                |error| format!("remove verification scratch directory failed: {error}"),
+            )?;
+        }
+    }
     Ok(())
 }
 

src/cli/tests.rs

@@ -21,6 +21,8 @@ fn parse_help_returns_usage() {
     assert!(err.contains("--memory-trim-after-validation"), "{err}");
     assert!(err.contains("--enable-roa-validation-cache"), "{err}");
     assert!(err.contains("--resource-validation-mode"), "{err}");
+    assert!(err.contains("--max-ca-depth"), "{err}");
+    assert!(err.contains("default: 32"), "{err}");
     assert!(
         err.contains("--publication-point-cache-observe-only"),
         "{err}"
@@ -47,6 +49,63 @@ fn parse_rejects_unknown_argument() {
     assert!(err.contains("unknown argument"), "{err}");
 }
 
+#[test]
+fn parse_accepts_verification_only_without_manual_validation_time() {
+    let argv = vec![
+        "rpki".to_string(),
+        "--verification-only".to_string(),
+        "--verification-source-run".to_string(),
+        "/run/run_0002".to_string(),
+        "--verification-source-state".to_string(),
+        "/run/state".to_string(),
+        "--verification-out".to_string(),
+        "/verification/check_0002".to_string(),
+    ];
+    let args = parse_args(&argv).expect("parse verification-only args");
+    assert!(args.verification_only);
+    assert!(args.validation_time.is_none());
+    assert_eq!(
+        args.db_path,
+        PathBuf::from("/verification/check_0002/scratch/work-db")
+    );
+}
+
+#[test]
+fn parse_rejects_manual_validation_time_in_verification_only() {
+    let argv = vec![
+        "rpki".to_string(),
+        "--verification-only".to_string(),
+        "--verification-source-run".to_string(),
+        "/run/run_0002".to_string(),
+        "--verification-source-state".to_string(),
+        "/run/state".to_string(),
+        "--verification-out".to_string(),
+        "/verification/check_0002".to_string(),
+        "--validation-time".to_string(),
+        "2026-07-16T00:00:00Z".to_string(),
+    ];
+    let error = parse_args(&argv).expect_err("manual time must be rejected");
+    assert!(error.contains("validation time"), "{error}");
+}
+
+#[test]
+fn parse_accepts_normal_validation_contract_output() {
+    let argv = vec![
+        "rpki".to_string(),
+        "--db".to_string(),
+        "db".to_string(),
+        "--tal-url".to_string(),
+        "https://example.test/root.tal".to_string(),
+        "--validation-contract-out".to_string(),
+        "out/validation-contract.json".to_string(),
+    ];
+    let args = parse_args(&argv).expect("parse normal contract output");
+    assert_eq!(
+        args.validation_contract_out_path,
+        Some(PathBuf::from("out/validation-contract.json"))
+    );
+}
+
 #[test]
 fn parse_rejects_both_tal_url_and_tal_path() {
     let argv = vec![
@@ -80,6 +139,51 @@ fn parse_rejects_invalid_max_depth() {
     assert!(err.contains("invalid --max-depth"), "{err}");
 }
 
+#[test]
+fn parse_rejects_invalid_max_ca_depth() {
+    let argv = vec![
+        "rpki".to_string(),
+        "--db".to_string(),
+        "db".to_string(),
+        "--tal-url".to_string(),
+        "https://example.test/x.tal".to_string(),
+        "--max-ca-depth".to_string(),
+        "nope".to_string(),
+    ];
+    let err = parse_args(&argv).unwrap_err();
+    assert!(err.contains("invalid --max-ca-depth"), "{err}");
+}
+
+#[test]
+fn parse_rejects_both_ca_depth_options() {
+    let argv = vec![
+        "rpki".to_string(),
+        "--db".to_string(),
+        "db".to_string(),
+        "--tal-url".to_string(),
+        "https://example.test/x.tal".to_string(),
+        "--max-ca-depth".to_string(),
+        "32".to_string(),
+        "--max-depth".to_string(),
+        "12".to_string(),
+    ];
+    let err = parse_args(&argv).unwrap_err();
+    assert!(err.contains("cannot be combined"), "{err}");
+}
+
+#[test]
+fn parse_defaults_max_ca_depth_to_32() {
+    let argv = vec![
+        "rpki".to_string(),
+        "--db".to_string(),
+        "db".to_string(),
+        "--tal-url".to_string(),
+        "https://example.test/x.tal".to_string(),
+    ];
+    let args = parse_args(&argv).expect("parse");
+    assert_eq!(args.max_ca_depth, 32);
+}
+
 #[test]
 fn parse_accepts_ccr_out_path() {
     let argv = vec![
@@ -1103,7 +1207,7 @@ fn parse_accepts_offline_mode_requires_ta() {
     assert_eq!(args.ta_paths, vec![PathBuf::from("ta.cer")]);
     assert_eq!(args.tal_path.as_deref(), Some(Path::new("a.tal")));
     assert_eq!(args.ta_path.as_deref(), Some(Path::new("ta.cer")));
-    assert_eq!(args.max_depth, Some(0));
+    assert_eq!(args.max_ca_depth, 0);
 }
 
 #[test]

src/fetch/current_repository.rs

@@ -0,0 +1,178 @@
+use std::sync::Arc;
+
+use crate::fetch::rsync::{
+    RsyncFetchError, RsyncFetchResult, RsyncFetcher, normalize_rsync_base_uri,
+};
+use crate::storage::{RepositoryViewState, RocksStore};
+
+#[derive(Clone)]
+pub struct CurrentRepositoryViewRsyncFetcher {
+    store: Arc<RocksStore>,
+}
+
+impl CurrentRepositoryViewRsyncFetcher {
+    pub fn new(store: Arc<RocksStore>) -> Self {
+        Self { store }
+    }
+}
+
+impl RsyncFetcher for CurrentRepositoryViewRsyncFetcher {
+    fn fetch_objects(&self, rsync_base_uri: &str) -> RsyncFetchResult<Vec<(String, Vec<u8>)>> {
+        let base = normalize_rsync_base_uri(rsync_base_uri);
+        let entries = self
+            .store
+            .list_repository_view_entries_with_prefix(&base)
+            .map_err(|error| {
+                RsyncFetchError::Fetch(format!(
+                    "list frozen repository view failed for {base}: {error}"
+                ))
+            })?;
+        let mut objects = Vec::with_capacity(entries.len());
+        for entry in entries {
+            if !matches!(
+                entry.state,
+                RepositoryViewState::Present | RepositoryViewState::Replaced
+            ) {
+                continue;
+            }
+            let bytes = self
+                .store
+                .load_current_object_bytes_by_uri(&entry.rsync_uri)
+                .map_err(|error| {
+                    RsyncFetchError::Fetch(format!(
+                        "load frozen repository object failed for {}: {error}",
+                        entry.rsync_uri
+                    ))
+                })?
+                .ok_or_else(|| {
+                    RsyncFetchError::Fetch(format!(
+                        "frozen repository object missing for {}",
+                        entry.rsync_uri
+                    ))
+                })?;
+            objects.push((entry.rsync_uri, bytes));
+        }
+        objects.sort_by(|left, right| left.0.cmp(&right.0));
+        if objects.is_empty() {
+            return Err(RsyncFetchError::Fetch(format!(
+                "frozen repository view contains no current objects under {base}"
+            )));
+        }
+        Ok(objects)
+    }
+
+    fn fetch_object(&self, rsync_uri: &str) -> RsyncFetchResult<Vec<u8>> {
+        self.store
+            .load_current_object_bytes_by_uri(rsync_uri)
+            .map_err(|error| {
+                RsyncFetchError::Fetch(format!(
+                    "load frozen repository object failed for {rsync_uri}: {error}"
+                ))
+            })?
+            .ok_or_else(|| {
+                RsyncFetchError::Fetch(format!("frozen repository object not found: {rsync_uri}"))
+            })
+    }
+
+    fn dedup_key(&self, rsync_base_uri: &str) -> String {
+        normalize_rsync_base_uri(rsync_base_uri)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::storage::{RepositoryViewEntry, RocksStore};
+
+    #[test]
+    fn current_repository_fetcher_returns_sorted_present_objects() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let store = Arc::new(RocksStore::open(dir.path()).expect("open store"));
+        let entries = [
+            ("rsync://example.test/repo/b.roa", b"b".as_slice()),
+            ("rsync://example.test/repo/a.mft", b"a".as_slice()),
+        ];
+        for (uri, bytes) in entries {
+            let hash = hex::encode(crate::cir::sha256(bytes));
+            store
+                .put_blob_bytes_batch(&[(hash.clone(), bytes.to_vec())])
+                .expect("put blob");
+            store
+                .put_repository_view_entry(&RepositoryViewEntry {
+                    rsync_uri: uri.to_string(),
+                    repository_source: Some("fixture".to_string()),
+                    object_type: Some("roa".to_string()),
+                    state: RepositoryViewState::Present,
+                    current_hash: Some(hash),
+                })
+                .expect("put view");
+        }
+        let fetcher = CurrentRepositoryViewRsyncFetcher::new(store);
+        let objects = fetcher
+            .fetch_objects("rsync://example.test/repo/")
+            .expect("fetch objects");
+        assert_eq!(
+            objects
+                .iter()
+                .map(|(uri, _)| uri.as_str())
+                .collect::<Vec<_>>(),
+            vec![
+                "rsync://example.test/repo/a.mft",
+                "rsync://example.test/repo/b.roa"
+            ]
+        );
+        assert_eq!(
+            fetcher
+                .fetch_object("rsync://example.test/repo/a.mft")
+                .expect("fetch one"),
+            b"a"
+        );
+        assert!(
+            fetcher
+                .fetch_object("rsync://example.test/repo/missing.roa")
+                .is_err()
+        );
+        assert_eq!(
+            fetcher.dedup_key("rsync://example.test/repo"),
+            "rsync://example.test/repo/"
+        );
+        assert!(
+            fetcher
+                .fetch_objects("rsync://empty.example/repo/")
+                .unwrap_err()
+                .to_string()
+                .contains("no current objects")
+        );
+    }
+
+    #[test]
+    fn current_repository_fetcher_ignores_withdrawn_and_reports_missing_blob() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let store = Arc::new(RocksStore::open(dir.path()).expect("open store"));
+        store
+            .put_repository_view_entry(&RepositoryViewEntry {
+                rsync_uri: "rsync://example.test/repo/withdrawn.roa".to_string(),
+                repository_source: Some("fixture".to_string()),
+                object_type: Some("roa".to_string()),
+                state: RepositoryViewState::Withdrawn,
+                current_hash: None,
+            })
+            .expect("put withdrawn view");
+        let missing_hash = "ab".repeat(32);
+        store
+            .put_repository_view_entry(&RepositoryViewEntry {
+                rsync_uri: "rsync://example.test/repo/missing.roa".to_string(),
+                repository_source: Some("fixture".to_string()),
+                object_type: Some("roa".to_string()),
+                state: RepositoryViewState::Present,
+                current_hash: Some(missing_hash),
+            })
+            .expect("put missing view");
+        let fetcher = CurrentRepositoryViewRsyncFetcher::new(store);
+        let error = fetcher
+            .fetch_objects("rsync://example.test/repo/")
+            .unwrap_err()
+            .to_string();
+        assert!(error.contains("blob bytes missing"));
+    }
+}

src/fetch/mod.rs

@@ -1,3 +1,4 @@
+pub mod current_repository;
 pub mod http;
 pub mod rsync;
 pub mod rsync_system;

src/lib.rs

@@ -42,3 +42,5 @@ pub mod sync;
 pub mod tools;
 #[cfg(feature = "full")]
 pub mod validation;
+#[cfg(feature = "full")]
+pub mod verification_only;

src/storage.rs

@@ -8,6 +8,7 @@ use std::path::{Path, PathBuf};
 use std::sync::{Arc, Mutex};
 
 use base64::Engine;
+use rocksdb::checkpoint::Checkpoint;
 use rocksdb::{ColumnFamily, DB, Direction, IteratorMode, Options, WriteBatch};
 use serde::{Deserialize, Serialize};
 
@@ -53,6 +54,13 @@ pub enum StorageError {
 
 pub type StorageResult<T> = Result<T, StorageError>;
 
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)]
+pub struct RepositoryBlobVerificationSummary {
+    pub current_objects: u64,
+    pub bytes_verified: u64,
+    pub batches: u64,
+}
+
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)]
 pub struct ChildCertificateCacheMmapLookup {
     pub projections: Vec<Option<ChildCertificateCacheProjection>>,
@@ -595,7 +603,7 @@ pub struct VcirFailedFetchReuseIdentity {
     pub ta_context_digest: [u8; 32],
     #[serde(rename = "p")]
     #[serde(with = "serde_bytes_32")]
-    pub parent_context_digest: [u8; 32],
+    pub ca_validation_context_digest: [u8; 32],
     #[serde(rename = "f")]
     #[serde(with = "serde_bytes_32")]
     pub policy_fingerprint: [u8; 32],
@@ -1117,7 +1125,7 @@ pub struct RoaCacheObjectMeta {
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct RoaCacheProjectionContext {
-    pub parent_context_digest: [u8; 32],
+    pub ca_validation_context_digest: [u8; 32],
     pub policy_fingerprint: [u8; 32],
     pub object_meta: Vec<RoaCacheObjectMeta>,
 }
@@ -1258,7 +1266,7 @@ pub struct RoaCacheProjection {
     pub issuer_ca_sha256_hex: Option<String>,
     #[serde(rename = "p", default, skip_serializing_if = "Option::is_none")]
     #[serde(with = "serde_optional_bytes_32")]
-    pub parent_context_digest: Option<[u8; 32]>,
+    pub ca_validation_context_digest: Option<[u8; 32]>,
     #[serde(rename = "f", default, skip_serializing_if = "Option::is_none")]
     #[serde(with = "serde_optional_bytes_32")]
     pub policy_fingerprint: Option<[u8; 32]>,
@@ -1400,7 +1408,7 @@ pub struct ChildCertificateCacheProjection {
     pub issuer_crl_sha256_hex: String,
     #[serde(rename = "pc")]
     #[serde(with = "serde_bytes_32")]
-    pub parent_context_digest: [u8; 32],
+    pub ca_validation_context_digest: [u8; 32],
     #[serde(rename = "pf")]
     #[serde(with = "serde_bytes_32")]
     pub validation_policy_fingerprint: [u8; 32],
@@ -1707,7 +1715,7 @@ pub struct PublicationPointCacheProjection {
     pub ta_context_digest: [u8; 32],
     #[serde(rename = "pc")]
     #[serde(with = "serde_bytes_32")]
-    pub parent_context_digest: [u8; 32],
+    pub ca_validation_context_digest: [u8; 32],
     #[serde(rename = "pf")]
     #[serde(with = "serde_bytes_32")]
     pub validation_policy_fingerprint: [u8; 32],
@@ -1743,7 +1751,7 @@ impl PublicationPointCacheProjection {
         ca_cert_sha256: [u8; 32],
         manifest_sha256: [u8; 32],
         ta_context_digest: [u8; 32],
-        parent_context_digest: [u8; 32],
+        ca_validation_context_digest: [u8; 32],
         validation_policy_fingerprint: [u8; 32],
     ) -> StorageResult<Self> {
         let projection = Self {
@@ -1756,7 +1764,7 @@ impl PublicationPointCacheProjection {
             manifest_sha256,
             tal_id: vcir.tal_id.clone(),
             ta_context_digest,
-            parent_context_digest,
+            ca_validation_context_digest,
             validation_policy_fingerprint,
             instance_effective_not_before: vcir.last_successful_validation_time.clone(),
             instance_effective_until: vcir.instance_gate.instance_effective_until.clone(),
@@ -1811,7 +1819,7 @@ impl PublicationPointCacheProjection {
             tal_id: self.tal_id.clone(),
             ca_subject_name: String::from("publication-point-cache"),
             ca_ski: hex::encode(self.ca_cert_sha256),
-            issuer_ski: hex::encode(self.parent_context_digest),
+            issuer_ski: hex::encode(self.ca_validation_context_digest),
             last_successful_validation_time: validation_time.clone(),
             current_manifest_rsync_uri: self.manifest_rsync_uri.clone(),
             current_crl_rsync_uri: self
@@ -2042,7 +2050,7 @@ impl RoaCacheProjection {
             manifest_rsync_uri: vcir.manifest_rsync_uri.clone(),
             instance_effective_until: vcir.instance_gate.instance_effective_until.clone(),
             issuer_ca_sha256_hex,
-            parent_context_digest: Some(context.parent_context_digest),
+            ca_validation_context_digest: Some(context.ca_validation_context_digest),
             policy_fingerprint: Some(context.policy_fingerprint),
             crl_sha256_by_uri,
             entries,
@@ -2063,10 +2071,10 @@ impl RoaCacheProjection {
         if let Some(hash) = &self.issuer_ca_sha256_hex {
             validate_sha256_hex("roa_cache_projection.issuer_ca_sha256_hex", hash)?;
         }
-        if self.parent_context_digest.is_some() != self.policy_fingerprint.is_some() {
+        if self.ca_validation_context_digest.is_some() != self.policy_fingerprint.is_some() {
             return Err(StorageError::InvalidData {
                 entity: "roa_cache_projection.context",
-                detail: "parent_context_digest and policy_fingerprint must be both present or both absent"
+                detail: "ca_validation_context_digest and policy_fingerprint must be both present or both absent"
                     .to_string(),
             });
         }
@@ -2893,6 +2901,38 @@ fn write_publication_point_cache_projection_to_batch(
 }
 
 impl RocksStore {
+    pub fn create_read_only_checkpoint(source: &Path, destination: &Path) -> StorageResult<()> {
+        if destination.exists() {
+            return Err(StorageError::InvalidData {
+                entity: "work_db_checkpoint",
+                detail: format!(
+                    "checkpoint destination already exists: {}",
+                    destination.display()
+                ),
+            });
+        }
+        if let Some(parent) = destination.parent() {
+            std::fs::create_dir_all(parent)
+                .map_err(|error| StorageError::RocksDb(error.to_string()))?;
+        }
+        let mut options = Options::default();
+        let blob_mode = work_db_blob_mode_from_env();
+        let memory_profile = work_db_memory_profile_from_env();
+        configure_work_db_options(&mut options, blob_mode, memory_profile);
+        let db = DB::open_cf_descriptors_read_only(
+            &options,
+            source,
+            column_family_descriptors_for_blob_mode(blob_mode),
+            false,
+        )
+        .map_err(|error| StorageError::RocksDb(error.to_string()))?;
+        let checkpoint =
+            Checkpoint::new(&db).map_err(|error| StorageError::RocksDb(error.to_string()))?;
+        checkpoint
+            .create_checkpoint(destination)
+            .map_err(|error| StorageError::RocksDb(error.to_string()))
+    }
+
     pub fn open(path: &Path) -> StorageResult<Self> {
         let mut base_opts = Options::default();
         base_opts.create_if_missing(true);
@@ -2931,6 +2971,15 @@ impl RocksStore {
         Self::open_with_external_stores(path, None, Some(repo_bytes_path))
     }
 
+    pub fn open_with_external_repo_bytes_read_only(
+        path: &Path,
+        repo_bytes_path: &Path,
+    ) -> StorageResult<Self> {
+        let mut store = Self::open(path)?;
+        store.external_repo_bytes = Some(ExternalRepoBytesDb::open_read_only(repo_bytes_path)?);
+        Ok(store)
+    }
+
     pub fn open_with_external_stores(
         path: &Path,
         raw_store_path: Option<&Path>,
@@ -3147,6 +3196,53 @@ impl RocksStore {
             .collect()
     }
 
+    pub fn verify_current_repository_blobs(
+        &self,
+        batch_size: usize,
+    ) -> StorageResult<RepositoryBlobVerificationSummary> {
+        if batch_size == 0 {
+            return Err(StorageError::InvalidData {
+                entity: "repository_blob_verification",
+                detail: "batch_size must be greater than zero".to_string(),
+            });
+        }
+        let cf = self.cf(CF_REPOSITORY_VIEW)?;
+        let prefix = repository_view_prefix("");
+        let mode = IteratorMode::From(prefix.as_bytes(), Direction::Forward);
+        let mut summary = RepositoryBlobVerificationSummary::default();
+        let mut batch = Vec::with_capacity(batch_size);
+        for item in self.db.iterator_cf(cf, mode) {
+            let (key, value) = item.map_err(|error| StorageError::RocksDb(error.to_string()))?;
+            if !key.starts_with(prefix.as_bytes()) {
+                break;
+            }
+            let entry = decode_cbor::<RepositoryViewEntry>(&value, "repository_view")?;
+            entry.validate_internal()?;
+            if !matches!(
+                entry.state,
+                RepositoryViewState::Present | RepositoryViewState::Replaced
+            ) {
+                continue;
+            }
+            let hash = entry
+                .current_hash
+                .clone()
+                .ok_or(StorageError::InvalidData {
+                    entity: "repository_blob_verification",
+                    detail: format!("current hash missing for {}", entry.rsync_uri),
+                })?;
+            batch.push((entry.rsync_uri, hash));
+            if batch.len() >= batch_size {
+                verify_repository_blob_batch(self, &batch, &mut summary)?;
+                batch.clear();
+            }
+        }
+        if !batch.is_empty() {
+            verify_repository_blob_batch(self, &batch, &mut summary)?;
+        }
+        Ok(summary)
+    }
+
     pub fn put_raw_by_hash_entry(&self, entry: &RawByHashEntry) -> StorageResult<()> {
         entry.validate_internal()?;
         if let Some(raw_store) = self.external_raw_store.as_ref() {
@@ -3206,6 +3302,9 @@ impl RocksStore {
             return Ok(());
         }
         if let Some(repo_bytes) = self.external_repo_bytes.as_ref() {
+            if repo_bytes.is_read_only() {
+                return repo_bytes.require_existing_blob_bytes_batch(blobs);
+            }
             return repo_bytes.put_blob_bytes_batch(blobs);
         }
         if let Some(raw_store) = self.external_raw_store.as_ref() {
@@ -4682,6 +4781,37 @@ impl RocksStore {
     }
 }
 
+fn verify_repository_blob_batch(
+    store: &RocksStore,
+    batch: &[(String, String)],
+    summary: &mut RepositoryBlobVerificationSummary,
+) -> StorageResult<()> {
+    let hashes = batch
+        .iter()
+        .map(|(_, hash)| hash.clone())
+        .collect::<Vec<_>>();
+    let blobs = store.get_blob_bytes_batch(&hashes)?;
+    for ((uri, expected_hash), blob) in batch.iter().zip(blobs.into_iter()) {
+        let bytes = blob.as_ref().ok_or(StorageError::InvalidData {
+            entity: "repository_blob_verification",
+            detail: format!("blob missing for URI {uri} (hash={expected_hash})"),
+        })?;
+        let actual_hash = hex::encode(compute_sha256_32(bytes));
+        if !actual_hash.eq_ignore_ascii_case(expected_hash) {
+            return Err(StorageError::InvalidData {
+                entity: "repository_blob_verification",
+                detail: format!(
+                    "blob hash mismatch for URI {uri}: expected={expected_hash}, actual={actual_hash}"
+                ),
+            });
+        }
+        summary.current_objects += 1;
+        summary.bytes_verified += bytes.len() as u64;
+    }
+    summary.batches += 1;
+    Ok(())
+}
+
 #[cfg(test)]
 #[path = "storage/tests.rs"]
 mod tests;

src/storage/tests.rs

@@ -29,7 +29,7 @@ fn sample_child_certificate_cache_projection(
         issuer_ca_sha256_hex: sha256_hex(b"issuer-ca"),
         issuer_crl_uri: "rsync://example.test/repo/issuer.crl".to_string(),
         issuer_crl_sha256_hex: sha256_hex(b"issuer-crl"),
-        parent_context_digest: sha256_32(b"parent-context"),
+        ca_validation_context_digest: sha256_32(b"parent-context"),
         validation_policy_fingerprint: sha256_32(b"policy"),
         effective_not_before: pack_time(0),
         effective_until: pack_time(24),
@@ -294,7 +294,7 @@ fn roa_cache_projection_context(vcir: &ValidatedCaInstanceResult) -> RoaCachePro
         })
         .expect("sample VCIR has a ROA output");
     RoaCacheProjectionContext {
-        parent_context_digest: [0x31; 32],
+        ca_validation_context_digest: [0x31; 32],
         policy_fingerprint: [0x32; 32],
         object_meta: vec![RoaCacheObjectMeta {
             source_object_uri: roa_output.source_object_uri.clone(),
@@ -2276,3 +2276,121 @@ fn pack_file_can_lazy_load_bytes_from_external_repo_bytes_store() {
     assert_eq!(file.bytes_cloned().expect("cloned repo bytes"), bytes);
     assert_eq!(file.compute_sha256().expect("compute sha256"), file.sha256);
 }
+
+#[test]
+fn read_only_checkpoint_isolated_from_source_work_db() {
+    let td = tempfile::tempdir().expect("tempdir");
+    let source_path = td.path().join("source-work-db");
+    let checkpoint_path = td.path().join("checkpoint-work-db");
+    let uri = "rsync://example.test/repo/a.roa";
+    let bytes = b"checkpoint-object";
+    let hash = sha256_hex(bytes);
+    {
+        let source = RocksStore::open(&source_path).expect("open source");
+        source
+            .put_blob_bytes_batch(&[(hash.clone(), bytes.to_vec())])
+            .expect("put blob");
+        source
+            .put_repository_view_entry(&RepositoryViewEntry {
+                rsync_uri: uri.to_string(),
+                current_hash: Some(hash),
+                repository_source: Some("fixture".to_string()),
+                object_type: Some("roa".to_string()),
+                state: RepositoryViewState::Present,
+            })
+            .expect("put view");
+    }
+
+    RocksStore::create_read_only_checkpoint(&source_path, &checkpoint_path)
+        .expect("create checkpoint");
+    let checkpoint = RocksStore::open(&checkpoint_path).expect("open checkpoint");
+    checkpoint
+        .delete_repository_view_entry(uri)
+        .expect("delete checkpoint entry");
+    drop(checkpoint);
+
+    let source = RocksStore::open(&source_path).expect("reopen source");
+    assert!(
+        source
+            .get_repository_view_entry(uri)
+            .expect("get source entry")
+            .is_some()
+    );
+}
+
+#[test]
+fn read_only_external_repo_bytes_rejects_writes() {
+    let td = tempfile::tempdir().expect("tempdir");
+    let path = td.path().join("repo-bytes.db");
+    let hash = sha256_hex(b"repo-object");
+    {
+        let writable = ExternalRepoBytesDb::open(&path).expect("open writable repo bytes");
+        writable
+            .put_blob_bytes_batch(&[(hash.clone(), b"repo-object".to_vec())])
+            .expect("seed repo bytes");
+    }
+    let read_only = ExternalRepoBytesDb::open_read_only(&path).expect("open read-only repo bytes");
+    assert_eq!(
+        read_only
+            .get_blob_bytes(&hash)
+            .expect("read repo bytes")
+            .expect("blob"),
+        b"repo-object"
+    );
+    assert!(
+        read_only
+            .put_blob_bytes_batch(&[(hash, b"repo-object".to_vec())])
+            .is_err()
+    );
+}
+
+#[test]
+fn read_only_store_accepts_only_idempotent_existing_blob_apply() {
+    let td = tempfile::tempdir().expect("tempdir");
+    let work_db_path = td.path().join("work-db");
+    let repo_bytes_path = td.path().join("repo-bytes.db");
+    let bytes = b"repo-object".to_vec();
+    let hash = sha256_hex(&bytes);
+    {
+        let repo_bytes = ExternalRepoBytesDb::open(&repo_bytes_path).expect("open repo bytes");
+        repo_bytes
+            .put_blob_bytes_batch(&[(hash.clone(), bytes.clone())])
+            .expect("seed repo bytes");
+    }
+    let store =
+        RocksStore::open_with_external_repo_bytes_read_only(&work_db_path, &repo_bytes_path)
+            .expect("open read-only store");
+    store
+        .put_blob_bytes_batch(&[(hash.clone(), bytes)])
+        .expect("idempotent apply");
+    assert!(
+        store
+            .put_blob_bytes_batch(&[(hash, b"different".to_vec())])
+            .is_err()
+    );
+}
+
+#[test]
+fn repository_blob_verification_detects_missing_external_blob() {
+    let td = tempfile::tempdir().expect("tempdir");
+    let repo_bytes_path = td.path().join("repo-bytes.db");
+    ExternalRepoBytesDb::open(&repo_bytes_path).expect("create repo bytes");
+    let store = RocksStore::open_with_external_repo_bytes_read_only(
+        &td.path().join("work-db"),
+        &repo_bytes_path,
+    )
+    .expect("open work db");
+    store
+        .put_repository_view_entry(&RepositoryViewEntry {
+            rsync_uri: "rsync://example.test/repo/missing.roa".to_string(),
+            current_hash: Some(sha256_hex(b"missing")),
+            repository_source: Some("fixture".to_string()),
+            object_type: Some("roa".to_string()),
+            state: RepositoryViewState::Present,
+        })
+        .expect("put view");
+    let error = store
+        .verify_current_repository_blobs(16)
+        .expect_err("missing blob must fail");
+    assert!(error.to_string().contains("missing.roa"), "{error}");
+}

src/validation/objects.rs

@@ -180,7 +180,7 @@ pub struct RoaValidationCacheStats {
 pub struct RoaValidationCacheInput<'a> {
     enabled: bool,
     view: Option<&'a RoaValidationCacheView>,
-    parent_context_digest: Option<[u8; 32]>,
+    ca_validation_context_digest: Option<[u8; 32]>,
     policy_fingerprint: Option<[u8; 32]>,
 }
 
@@ -189,7 +189,7 @@ impl<'a> RoaValidationCacheInput<'a> {
         Self {
             enabled: false,
             view: None,
-            parent_context_digest: None,
+            ca_validation_context_digest: None,
             policy_fingerprint: None,
         }
     }
@@ -198,20 +198,20 @@ impl<'a> RoaValidationCacheInput<'a> {
         Self {
             enabled: true,
             view,
-            parent_context_digest: None,
+            ca_validation_context_digest: None,
             policy_fingerprint: None,
         }
     }
 
     pub fn enabled_with_context(
         view: Option<&'a RoaValidationCacheView>,
-        parent_context_digest: [u8; 32],
+        ca_validation_context_digest: [u8; 32],
         policy_fingerprint: [u8; 32],
     ) -> Self {
         Self {
             enabled: true,
             view,
-            parent_context_digest: Some(parent_context_digest),
+            ca_validation_context_digest: Some(ca_validation_context_digest),
             policy_fingerprint: Some(policy_fingerprint),
         }
     }
@@ -495,7 +495,7 @@ fn evaluate_roa_cache_crl_gate(
 pub struct RoaValidationCacheView {
     entries_by_uri: HashMap<String, CachedRoaValidationResult>,
     issuer_ca_sha256_hex: Option<String>,
-    parent_context_digest: Option<[u8; 32]>,
+    ca_validation_context_digest: Option<[u8; 32]>,
     policy_fingerprint: Option<[u8; 32]>,
     crl_sha256_by_uri: HashMap<String, String>,
     blocked: bool,
@@ -519,7 +519,7 @@ impl RoaValidationCacheView {
         let mut entries_by_uri: HashMap<String, CachedRoaValidationResult> =
             HashMap::with_capacity(projection.entries.len());
         let issuer_ca_sha256_hex = projection.issuer_ca_sha256_hex.clone();
-        let parent_context_digest = projection.parent_context_digest;
+        let ca_validation_context_digest = projection.ca_validation_context_digest;
         let policy_fingerprint = projection.policy_fingerprint;
         let crl_sha256_by_uri = projection
             .crl_sha256_by_uri
@@ -536,7 +536,7 @@ impl RoaValidationCacheView {
             return Self {
                 entries_by_uri,
                 issuer_ca_sha256_hex,
-                parent_context_digest,
+                ca_validation_context_digest,
                 policy_fingerprint,
                 crl_sha256_by_uri,
                 blocked,
@@ -580,7 +580,7 @@ impl RoaValidationCacheView {
         Self {
             entries_by_uri,
             issuer_ca_sha256_hex,
-            parent_context_digest,
+            ca_validation_context_digest,
             policy_fingerprint,
             crl_sha256_by_uri,
             blocked,
@@ -590,7 +590,7 @@ impl RoaValidationCacheView {
     fn matches_current_context(
         &self,
         issuer_ca_der: &[u8],
-        parent_context_digest: Option<[u8; 32]>,
+        ca_validation_context_digest: Option<[u8; 32]>,
         policy_fingerprint: Option<[u8; 32]>,
     ) -> bool {
         if self.blocked {
@@ -604,8 +604,8 @@ impl RoaValidationCacheView {
             return false;
         }
 
-        if let Some(expected_parent_context) = self.parent_context_digest {
+        if let Some(expected_ca_validation_context) = self.ca_validation_context_digest {
-            if Some(expected_parent_context) != parent_context_digest {
+            if Some(expected_ca_validation_context) != ca_validation_context_digest {
                 return false;
             }
         } else {
@@ -797,7 +797,7 @@ fn active_roa_cache_view<'a>(
     let gate_started = Instant::now();
     if view.matches_current_context(
         issuer_ca_der,
-        roa_cache.parent_context_digest,
+        roa_cache.ca_validation_context_digest,
         roa_cache.policy_fingerprint,
     ) {
         stats.context_gate_nanos = stats
@@ -3914,7 +3914,7 @@ mod tests {
         OffsetDateTime::parse(value, &Rfc3339).expect("parse fixed test time")
     }
 
-    const TEST_PARENT_CONTEXT: [u8; 32] = [0x70; 32];
+    const TEST_CA_VALIDATION_CONTEXT: [u8; 32] = [0x70; 32];
     const TEST_POLICY_FINGERPRINT: [u8; 32] = [0x71; 32];
     const TEST_CRL_URI: &str = "rsync://example.test/repo/current.crl";
     const TEST_ROA_URI: &str = "rsync://example.test/repo/a.roa";
@@ -3933,7 +3933,7 @@ mod tests {
         RoaCacheProjection::from_vcir_with_context(
             vcir,
             Some(&RoaCacheProjectionContext {
-                parent_context_digest: TEST_PARENT_CONTEXT,
+                ca_validation_context_digest: TEST_CA_VALIDATION_CONTEXT,
                 policy_fingerprint: TEST_POLICY_FINGERPRINT,
                 object_meta: vec![RoaCacheObjectMeta {
                     source_object_uri: TEST_ROA_URI.to_string(),
@@ -4071,7 +4071,7 @@ mod tests {
 
         assert!(view.matches_current_context(
             issuer_der,
-            Some(TEST_PARENT_CONTEXT),
+            Some(TEST_CA_VALIDATION_CONTEXT),
             Some(TEST_POLICY_FINGERPRINT)
         ));
         let hit = view.lookup(&file, &mut crl_cache, issuer_der, validation_time);
@@ -4202,7 +4202,10 @@ mod tests {
             fixed_time("2026-06-08T00:00:00Z"),
         );
         let projection = sample_roa_cache_projection(&vcir, roa_hash);
-        assert_eq!(projection.parent_context_digest, Some(TEST_PARENT_CONTEXT));
+        assert_eq!(
+            projection.ca_validation_context_digest,
+            Some(TEST_CA_VALIDATION_CONTEXT)
+        );
         assert_eq!(projection.policy_fingerprint, Some(TEST_POLICY_FINGERPRINT));
         assert_eq!(
             projection.entries[0].outputs_effective_until_unix,
@@ -4245,7 +4248,7 @@ mod tests {
         let issuer_changed = RoaValidationCacheView::from_projection(&projection, validation_time);
         assert!(!issuer_changed.matches_current_context(
             issuer_der,
-            Some(TEST_PARENT_CONTEXT),
+            Some(TEST_CA_VALIDATION_CONTEXT),
             Some(TEST_POLICY_FINGERPRINT)
         ));
 
@@ -4258,7 +4261,7 @@ mod tests {
         ));
         assert!(!parent_changed.matches_current_context(
             issuer_der,
-            Some(TEST_PARENT_CONTEXT),
+            Some(TEST_CA_VALIDATION_CONTEXT),
             Some([0x98; 32])
         ));
 
@@ -4283,7 +4286,7 @@ mod tests {
             RoaValidationCacheView::from_projection(&expired_projection, validation_time);
         assert!(!expired_view.matches_current_context(
             issuer_der,
-            Some(TEST_PARENT_CONTEXT),
+            Some(TEST_CA_VALIDATION_CONTEXT),
             Some(TEST_POLICY_FINGERPRINT)
         ));
         let mut crl_cache = sample_crl_cache(b"current-crl".to_vec());
@@ -4563,7 +4566,7 @@ mod tests {
 
         assert!(view.matches_current_context(
             issuer_der,
-            Some(TEST_PARENT_CONTEXT),
+            Some(TEST_CA_VALIDATION_CONTEXT),
             Some(TEST_POLICY_FINGERPRINT)
         ));
         assert!(matches!(
@@ -4595,43 +4598,43 @@ mod tests {
         let mut view = RoaValidationCacheView {
             entries_by_uri: HashMap::new(),
             issuer_ca_sha256_hex: None,
-            parent_context_digest: Some(TEST_PARENT_CONTEXT),
+            ca_validation_context_digest: Some(TEST_CA_VALIDATION_CONTEXT),
             policy_fingerprint: Some(TEST_POLICY_FINGERPRINT),
             crl_sha256_by_uri: HashMap::new(),
             blocked: false,
         };
         assert!(!view.matches_current_context(
             b"issuer-ca",
-            Some(TEST_PARENT_CONTEXT),
+            Some(TEST_CA_VALIDATION_CONTEXT),
             Some(TEST_POLICY_FINGERPRINT)
         ));
 
         view.issuer_ca_sha256_hex = Some("00".repeat(32));
         assert!(!view.matches_current_context(
             b"issuer-ca",
-            Some(TEST_PARENT_CONTEXT),
+            Some(TEST_CA_VALIDATION_CONTEXT),
             Some(TEST_POLICY_FINGERPRINT)
         ));
 
         view.issuer_ca_sha256_hex = Some(sha256_hex(b"issuer-ca"));
         assert!(view.matches_current_context(
             b"issuer-ca",
-            Some(TEST_PARENT_CONTEXT),
+            Some(TEST_CA_VALIDATION_CONTEXT),
             Some(TEST_POLICY_FINGERPRINT)
         ));
         assert!(!view.matches_current_context(b"issuer-ca", None, Some(TEST_POLICY_FINGERPRINT)));
 
-        view.parent_context_digest = None;
+        view.ca_validation_context_digest = None;
         assert!(!view.matches_current_context(
             b"issuer-ca",
-            Some(TEST_PARENT_CONTEXT),
+            Some(TEST_CA_VALIDATION_CONTEXT),
             Some(TEST_POLICY_FINGERPRINT)
         ));
-        view.parent_context_digest = Some(TEST_PARENT_CONTEXT);
+        view.ca_validation_context_digest = Some(TEST_CA_VALIDATION_CONTEXT);
         view.policy_fingerprint = None;
         assert!(!view.matches_current_context(
             b"issuer-ca",
-            Some(TEST_PARENT_CONTEXT),
+            Some(TEST_CA_VALIDATION_CONTEXT),
             Some(TEST_POLICY_FINGERPRINT)
         ));
     }
@@ -4661,7 +4664,7 @@ mod tests {
         let mut view = RoaValidationCacheView {
             entries_by_uri: HashMap::new(),
             issuer_ca_sha256_hex: Some(sha256_hex(issuer_der)),
-            parent_context_digest: Some(TEST_PARENT_CONTEXT),
+            ca_validation_context_digest: Some(TEST_CA_VALIDATION_CONTEXT),
             policy_fingerprint: Some(TEST_POLICY_FINGERPRINT),
             crl_sha256_by_uri: HashMap::from([(TEST_CRL_URI.to_string(), crl_hash_hex.clone())]),
             blocked: false,
@@ -4849,7 +4852,7 @@ mod tests {
             false,
             RoaValidationCacheInput::enabled_with_context(
                 Some(&view),
-                TEST_PARENT_CONTEXT,
+                TEST_CA_VALIDATION_CONTEXT,
                 TEST_POLICY_FINGERPRINT,
             ),
         ) {

src/validation/tree.rs

@@ -10,6 +10,8 @@ use crate::validation::publication_point::PublicationPointSnapshot;
 use std::borrow::Cow;
 use std::sync::{Arc, OnceLock};
 
+pub const DEFAULT_MAX_CA_DEPTH: usize = 32;
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct TreeRunConfig {
     /// Max CA instance depth to process (0 = root only).
@@ -37,7 +39,7 @@ pub struct TreeRunConfig {
 impl Default for TreeRunConfig {
     fn default() -> Self {
         Self {
-            max_depth: None,
+            max_depth: Some(DEFAULT_MAX_CA_DEPTH),
             max_instances: None,
             compact_audit: false,
             persist_vcir: true,
@@ -51,6 +53,15 @@ impl Default for TreeRunConfig {
     }
 }
 
+pub(crate) fn ca_depth_is_allowed(config: &TreeRunConfig, depth: usize) -> bool {
+    config.max_depth.is_none_or(|max_depth| depth <= max_depth)
+}
+
+pub(crate) fn next_allowed_ca_depth(config: &TreeRunConfig, parent_depth: usize) -> Option<usize> {
+    let child_depth = parent_depth.checked_add(1)?;
+    ca_depth_is_allowed(config, child_depth).then_some(child_depth)
+}
+
 #[derive(Clone, Debug)]
 pub enum CaCertificateRef {
     InlineDer(Vec<u8>),
@@ -347,10 +358,8 @@ pub fn run_tree_serial_audit_multi_root(
             }
         }
 
-        if let Some(max_depth) = config.max_depth {
+        if !ca_depth_is_allowed(config, ca.depth) {
-            if ca.depth > max_depth {
+            continue;
-                continue;
-            }
         }
 
         let res = match runner.run_publication_point(ca) {
@@ -393,6 +402,9 @@ pub fn run_tree_serial_audit_multi_root(
         }
         publication_points.push(audit);
 
+        let Some(child_depth) = next_allowed_ca_depth(config, ca.depth) else {
+            continue;
+        };
         let mut children = res.discovered_children;
         children.sort_by(|a, b| {
             a.handle
@@ -413,7 +425,7 @@ pub fn run_tree_serial_audit_multi_root(
         for child in children {
             queue.push_back(QueuedCaInstance {
                 id: next_id,
-                handle: child.handle.with_depth(ca.depth + 1),
+                handle: child.handle.with_depth(child_depth),
                 parent_id: Some(node.id),
                 discovered_from: Some(child.discovered_from),
             });

src/validation/tree_parallel.rs

@@ -19,8 +19,8 @@ use crate::validation::objects::{
 };
 use crate::validation::tree::{
     CaInstanceHandle, DiscoveredChildCaInstance, PublicationPointRunResult, PublicationPointRunner,
-    TreeRunAuditOutput, TreeRunConfig, TreeRunError, TreeRunOutput,
+    TreeRunAuditOutput, TreeRunConfig, TreeRunError, TreeRunOutput, ca_depth_is_allowed,
-    run_tree_serial_audit_multi_root,
+    next_allowed_ca_depth, run_tree_serial_audit_multi_root,
 };
 use crate::validation::tree_runner::{
     FreshPublicationPointFinalizeOutput, FreshPublicationPointStage, Rpkiv1PublicationPointRunner,
@@ -659,6 +659,7 @@ pub fn run_tree_parallel_phase2_audit_multi_root(
                         &mut finished,
                         ready,
                         ready_queue_len_after_pop,
+                        config,
                         config.compact_audit,
                     );
                     ready_batch_metrics.record(metrics);
@@ -916,10 +917,8 @@ fn start_queued_ca_instances(
         if !visited_manifest_uris.insert(node.handle.manifest_rsync_uri.clone()) {
             continue;
         }
-        if let Some(max_depth) = config.max_depth {
+        if !ca_depth_is_allowed(config, node.handle.depth) {
-            if node.handle.depth > max_depth {
+            continue;
-                continue;
-            }
         }
         *instances_started += 1;
         match repo_runtime.request_publication_point_repo(&node.handle, 0) {
@@ -959,6 +958,7 @@ fn stage_ready_publication_point(
     finished: &mut Vec<FinishedPublicationPoint>,
     ready: ReadyCaInstance,
     ready_queue_len_after_pop: usize,
+    config: &TreeRunConfig,
     compact_audit: bool,
 ) -> ReadyStageMetrics {
     let publication_point_started = Instant::now();
@@ -991,6 +991,7 @@ fn stage_ready_publication_point(
             next_id,
             ca_queue,
             &ready.node,
+            config,
             result.discovered_children.clone(),
         );
         metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started);
@@ -1057,6 +1058,7 @@ fn stage_ready_publication_point(
                     next_id,
                     ca_queue,
                     &ready.node,
+                    config,
                     result.discovered_children.clone(),
                 );
                 metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started);
@@ -1142,6 +1144,7 @@ fn stage_ready_publication_point(
         next_id,
         ca_queue,
         &ready.node,
+        config,
         fresh_stage.discovered_children.clone(),
     );
     metrics.child_enqueue_ms = elapsed_ms(child_enqueue_started);
@@ -1175,7 +1178,7 @@ fn stage_ready_publication_point(
     let roa_cache = if runner.enable_roa_validation_cache && has_roa {
         RoaValidationCacheInput::enabled_with_context(
             roa_cache_view.as_ref(),
-            crate::validation::tree_runner::parent_context_digest_for_ca(&ready.node.handle),
+            crate::validation::tree_runner::ca_validation_context_digest_for_ca(&ready.node.handle),
             crate::validation::tree_runner::publication_point_cache_policy_fingerprint(
                 runner.policy,
             ),
@@ -1412,8 +1415,13 @@ fn enqueue_discovered_children(
     next_id: &mut u64,
     ca_queue: &mut VecDeque<QueuedCaInstance>,
     parent: &QueuedCaInstance,
+    config: &TreeRunConfig,
     mut children: Vec<DiscoveredChildCaInstance>,
 ) {
+    let Some(child_depth) = next_allowed_ca_depth(config, parent.handle.depth) else {
+        return;
+    };
+
     children.sort_by(|a, b| {
         a.handle
             .manifest_rsync_uri
@@ -1428,7 +1436,7 @@ fn enqueue_discovered_children(
         let _ = runtime.prefetch_discovered_children(&children);
     }
     for child in children {
-        let mut handle = child.handle.with_depth(parent.handle.depth + 1);
+        let mut handle = child.handle.with_depth(child_depth);
         handle.parent_manifest_rsync_uri = Some(parent.handle.manifest_rsync_uri.clone());
         ca_queue.push_back(QueuedCaInstance {
             id: *next_id,

src/validation/tree_runner.rs

@@ -304,7 +304,8 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
             self.finish_publication_point_cache_miss(ca, "ta_context_mismatch", lookup_started);
             return None;
         }
-        if projection.parent_context_digest != current_identity.parent_context_digest {
+        if projection.ca_validation_context_digest != current_identity.ca_validation_context_digest
+        {
             self.finish_publication_point_cache_miss(ca, "parent_context_mismatch", lookup_started);
             return None;
         }
@@ -505,7 +506,7 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
             ca_cert_sha256,
             manifest_sha256,
             ta_context_digest: ta_context_digest_for_ca(ca),
-            parent_context_digest: parent_context_digest_for_ca(ca),
+            ca_validation_context_digest: ca_validation_context_digest_for_ca(ca),
             policy_fingerprint: publication_point_cache_policy_fingerprint(self.policy),
         })
     }
@@ -938,7 +939,7 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> {
                     Some(ChildCertificateValidationCacheContext {
                         store: self.store,
                         issuer_ca_sha256: sha256_digest_32(issuer_ca_der.as_ref()),
-                        parent_context_digest: parent_context_digest_for_ca(ca),
+                        ca_validation_context_digest: ca_validation_context_digest_for_ca(ca),
                         policy_fingerprint: publication_point_cache_policy_fingerprint(self.policy),
                     })
                 } else {
@@ -1530,7 +1531,7 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
                 let roa_cache = if self.enable_roa_validation_cache && has_roa {
                     RoaValidationCacheInput::enabled_with_context(
                         roa_cache_view.as_ref(),
-                        parent_context_digest_for_ca(ca),
+                        ca_validation_context_digest_for_ca(ca),
                         publication_point_cache_policy_fingerprint(self.policy),
                     )
                 } else {
@@ -1946,7 +1947,7 @@ struct PublicationPointCacheIdentity {
     ca_cert_sha256: [u8; 32],
     manifest_sha256: [u8; 32],
     ta_context_digest: [u8; 32],
-    parent_context_digest: [u8; 32],
+    ca_validation_context_digest: [u8; 32],
     policy_fingerprint: [u8; 32],
 }
 
@@ -1982,7 +1983,7 @@ fn ta_context_digest_for_ca(ca: &CaInstanceHandle) -> [u8; 32] {
     ])
 }
 
-pub(crate) fn parent_context_digest_for_ca(ca: &CaInstanceHandle) -> [u8; 32] {
+pub(crate) fn ca_validation_context_digest_for_ca(ca: &CaInstanceHandle) -> [u8; 32] {
     hash_serialized_parts(&[
         (
             "version",
@@ -2058,7 +2059,7 @@ fn pack_time_window_contains(
 struct ChildCertificateValidationCacheContext<'a> {
     store: &'a RocksStore,
     issuer_ca_sha256: [u8; 32],
-    parent_context_digest: [u8; 32],
+    ca_validation_context_digest: [u8; 32],
     policy_fingerprint: [u8; 32],
 }
 
@@ -2066,7 +2067,7 @@ fn child_certificate_cache_key_sha256_hex(
     child_cert_uri: &str,
     child_cert_sha256: &[u8; 32],
     issuer_ca_sha256: &[u8; 32],
-    parent_context_digest: &[u8; 32],
+    ca_validation_context_digest: &[u8; 32],
     policy_fingerprint: &[u8; 32],
 ) -> String {
     let digest = hash_serialized_parts(&[
@@ -2074,7 +2075,11 @@ fn child_certificate_cache_key_sha256_hex(
         ("child_cert_uri", child_cert_uri.as_bytes().to_vec()),
         ("child_cert_sha256", child_cert_sha256.to_vec()),
         ("issuer_ca_sha256", issuer_ca_sha256.to_vec()),
-        ("parent_context_digest", parent_context_digest.to_vec()),
+        // Keep the persisted cache-key label stable; this change only renames Rust identifiers.
+        (
+            "parent_context_digest",
+            ca_validation_context_digest.to_vec(),
+        ),
         ("policy_fingerprint", policy_fingerprint.to_vec()),
     ]);
     sha256_hex_from_32(&digest)
@@ -2469,7 +2474,7 @@ fn discover_children_from_fresh_snapshot_with_audit_cached_with_issuer_der<
                 &f.rsync_uri,
                 &f.sha256,
                 &cache.issuer_ca_sha256,
-                &cache.parent_context_digest,
+                &cache.ca_validation_context_digest,
                 &cache.policy_fingerprint,
             ));
         }
@@ -3022,7 +3027,7 @@ fn discover_children_from_fresh_snapshot_with_audit_cached_with_issuer_der<
                                     &f.rsync_uri,
                                     &f.sha256,
                                     &cache.issuer_ca_sha256,
-                                    &cache.parent_context_digest,
+                                    &cache.ca_validation_context_digest,
                                     &cache.policy_fingerprint,
                                 );
                                 let projection = ChildCertificateCacheProjection {
@@ -3040,7 +3045,8 @@ fn discover_children_from_fresh_snapshot_with_audit_cached_with_issuer_der<
                                     ),
                                     issuer_crl_uri: issuer_crl_uri.clone(),
                                     issuer_crl_sha256_hex: verified_crl.sha256_hex.clone(),
-                                    parent_context_digest: cache.parent_context_digest,
+                                    ca_validation_context_digest: cache
+                                        .ca_validation_context_digest,
                                     validation_policy_fingerprint: cache.policy_fingerprint,
                                     effective_not_before,
                                     effective_until,
@@ -3239,7 +3245,7 @@ fn discover_children_from_fresh_snapshot_with_audit_cached_with_issuer_der<
                         &f.rsync_uri,
                         &f.sha256,
                         &cache.issuer_ca_sha256,
-                        &cache.parent_context_digest,
+                        &cache.ca_validation_context_digest,
                         &cache.policy_fingerprint,
                     );
                     let projection = ChildCertificateCacheProjection {
@@ -3255,7 +3261,7 @@ fn discover_children_from_fresh_snapshot_with_audit_cached_with_issuer_der<
                         issuer_ca_sha256_hex: sha256_hex_from_32(&cache.issuer_ca_sha256),
                         issuer_crl_uri: issuer_crl_uri.clone(),
                         issuer_crl_sha256_hex: verified_crl.sha256_hex.clone(),
-                        parent_context_digest: cache.parent_context_digest,
+                        ca_validation_context_digest: cache.ca_validation_context_digest,
                         validation_policy_fingerprint: cache.policy_fingerprint,
                         effective_not_before,
                         effective_until,
@@ -4395,7 +4401,7 @@ fn failed_fetch_reuse_identity_for_fresh_result(
     let identity = VcirFailedFetchReuseIdentity {
         current_ca_sha256,
         ta_context_digest: ta_context_digest_for_ca(ca),
-        parent_context_digest: parent_context_digest_for_ca(ca),
+        ca_validation_context_digest: ca_validation_context_digest_for_ca(ca),
         policy_fingerprint: publication_point_cache_policy_fingerprint(policy),
         effective_not_before: PackTime::from_utc_offset_datetime(validation_time),
         effective_until,
@@ -4418,7 +4424,7 @@ fn failed_fetch_reuse_identity_matches_current(
     };
     cached.current_ca_sha256 == current_ca_sha256
         && cached.ta_context_digest == ta_context_digest_for_ca(ca)
-        && cached.parent_context_digest == parent_context_digest_for_ca(ca)
+        && cached.ca_validation_context_digest == ca_validation_context_digest_for_ca(ca)
         && cached.policy_fingerprint == publication_point_cache_policy_fingerprint(policy)
         && cached.effective_until == vcir.instance_gate.instance_effective_until
         && cached.contains_validation_time(validation_time)
@@ -5347,7 +5353,7 @@ fn persist_vcir_for_fresh_result_with_timing(
         .replace_vcir_manifest_replay_meta_and_projection_action_with_failed_fetch_reuse_identity(
             &vcir,
             Some(&RoaCacheProjectionContext {
-                parent_context_digest: parent_context_digest_for_ca(ca),
+                ca_validation_context_digest: ca_validation_context_digest_for_ca(ca),
                 policy_fingerprint: publication_point_cache_policy_fingerprint(policy),
                 object_meta: objects.roa_cache_object_meta.clone(),
             }),
@@ -5378,7 +5384,7 @@ fn build_publication_point_cache_projection_from_fresh(
         ca_cert_sha256,
         manifest_sha256,
         ta_context_digest_for_ca(ca),
-        parent_context_digest_for_ca(ca),
+        ca_validation_context_digest_for_ca(ca),
         publication_point_cache_policy_fingerprint(policy),
     )
     .map_err(|e| e.to_string())

src/validation/tree_runner/tests.rs

@@ -1601,7 +1601,7 @@ fn discover_children_child_certificate_cache_reuses_successful_child_ca() {
     let cache_context = ChildCertificateValidationCacheContext {
         store: &store,
         issuer_ca_sha256: issuer.ca_certificate_sha256_32().unwrap(),
-        parent_context_digest: parent_context_digest_for_ca(&issuer),
+        ca_validation_context_digest: ca_validation_context_digest_for_ca(&issuer),
         policy_fingerprint: publication_point_cache_policy_fingerprint(&policy),
     };
     let validation_time = time::OffsetDateTime::now_utc();
@@ -1625,7 +1625,7 @@ fn discover_children_child_certificate_cache_reuses_successful_child_ca() {
         &child_file.rsync_uri,
         &child_file.sha256,
         &cache_context.issuer_ca_sha256,
-        &cache_context.parent_context_digest,
+        &cache_context.ca_validation_context_digest,
         &cache_context.policy_fingerprint,
     );
     assert!(
@@ -1724,7 +1724,7 @@ fn discover_children_child_certificate_cache_rechecks_changed_valid_crl() {
     let cache_context = ChildCertificateValidationCacheContext {
         store: &store,
         issuer_ca_sha256: issuer.ca_certificate_sha256_32().unwrap(),
-        parent_context_digest: parent_context_digest_for_ca(&issuer),
+        ca_validation_context_digest: ca_validation_context_digest_for_ca(&issuer),
         policy_fingerprint: publication_point_cache_policy_fingerprint(&policy),
     };
     let validation_time = time::OffsetDateTime::now_utc();
@@ -1745,7 +1745,7 @@ fn discover_children_child_certificate_cache_rechecks_changed_valid_crl() {
         &child_file.rsync_uri,
         &child_file.sha256,
         &cache_context.issuer_ca_sha256,
-        &cache_context.parent_context_digest,
+        &cache_context.ca_validation_context_digest,
         &cache_context.policy_fingerprint,
     );
     let projection = store
@@ -1832,7 +1832,7 @@ fn discover_children_child_certificate_cache_misses_when_current_crl_invalid() {
     let cache_context = ChildCertificateValidationCacheContext {
         store: &store,
         issuer_ca_sha256: issuer.ca_certificate_sha256_32().unwrap(),
-        parent_context_digest: parent_context_digest_for_ca(&issuer),
+        ca_validation_context_digest: ca_validation_context_digest_for_ca(&issuer),
         policy_fingerprint: publication_point_cache_policy_fingerprint(&policy),
     };
     let validation_time = time::OffsetDateTime::now_utc();
@@ -1907,7 +1907,7 @@ fn discover_children_child_certificate_cache_misses_when_unchanged_crl_expired()
     let cache_context = ChildCertificateValidationCacheContext {
         store: &store,
         issuer_ca_sha256: issuer.ca_certificate_sha256_32().unwrap(),
-        parent_context_digest: parent_context_digest_for_ca(&issuer),
+        ca_validation_context_digest: ca_validation_context_digest_for_ca(&issuer),
         policy_fingerprint: publication_point_cache_policy_fingerprint(&policy),
     };
     let validation_time = time::OffsetDateTime::now_utc();
@@ -2436,7 +2436,7 @@ fn seed_publication_point_cache_projection(
         ca.ca_certificate_sha256_32().unwrap(),
         sha256_32(b"manifest-bytes"),
         ta_context_digest_for_ca(ca),
-        parent_context_digest_for_ca(ca),
+        ca_validation_context_digest_for_ca(ca),
         publication_point_cache_policy_fingerprint(policy),
     )
     .expect("build publication point projection");

tests/test_tree_traversal_m14.rs

@@ -253,6 +253,7 @@ fn tree_enqueues_children_for_fresh_and_current_instance_vcir_results() {
 fn tree_respects_max_depth_and_max_instances() {
     let root_manifest = "rsync://example.test/repo/root.mft";
     let child_manifest = "rsync://example.test/repo/child.mft";
+    let grandchild_manifest = "rsync://example.test/repo/grandchild.mft";
 
     let runner = MockRunner::default()
         .with(
@@ -301,6 +302,32 @@ fn tree_respects_max_depth_and_max_instances() {
                 audit: PublicationPointAudit::default(),
                 cir_fresh_objects: Vec::new(),
                 cir_cached_objects: Vec::new(),
+                discovered_children: vec![discovered_child(child_manifest, grandchild_manifest)],
+            },
+        )
+        .with(
+            grandchild_manifest,
+            PublicationPointRunResult {
+                source: PublicationPointSource::Fresh,
+                snapshot: Some(empty_snapshot(
+                    grandchild_manifest,
+                    "rsync://example.test/repo/grandchild/",
+                )),
+                warnings: Vec::new(),
+                objects: ObjectsOutput {
+                    vrps: Vec::new(),
+                    aspas: Vec::new(),
+                    router_keys: Vec::new(),
+                    local_outputs_cache: Vec::new(),
+                    warnings: Vec::new(),
+                    stats: ObjectsStats::default(),
+                    audit: Vec::new(),
+                    roa_cache_stats: Default::default(),
+                    roa_cache_object_meta: Vec::new(),
+                },
+                audit: PublicationPointAudit::default(),
+                cir_fresh_objects: Vec::new(),
+                cir_cached_objects: Vec::new(),
                 discovered_children: Vec::new(),
             },
         );
@@ -324,6 +351,62 @@ fn tree_respects_max_depth_and_max_instances() {
     .expect("run tree depth-limited");
     assert_eq!(out.instances_processed, 1);
     assert_eq!(out.instances_failed, 0);
+    assert_eq!(runner.called(), vec![root_manifest]);
+
+    let out = run_tree_serial(
+        ca_handle(root_manifest),
+        &runner,
+        &TreeRunConfig {
+            max_depth: Some(1),
+            max_instances: None,
+            compact_audit: false,
+            persist_vcir: true,
+            build_ccr_accumulator: true,
+            enable_roa_validation_cache: false,
+            enable_child_certificate_validation_cache: false,
+            publication_point_cache_observe_only: false,
+            enable_publication_point_validation_cache: false,
+            enable_transport_request_prefetch: false,
+        },
+    )
+    .expect("run tree depth-one-limited");
+    assert_eq!(out.instances_processed, 2);
+    assert_eq!(out.instances_failed, 0);
+    assert_eq!(
+        runner.called(),
+        vec![root_manifest, root_manifest, child_manifest]
+    );
+
+    let out = run_tree_serial(
+        ca_handle(root_manifest),
+        &runner,
+        &TreeRunConfig {
+            max_depth: Some(2),
+            max_instances: None,
+            compact_audit: false,
+            persist_vcir: true,
+            build_ccr_accumulator: true,
+            enable_roa_validation_cache: false,
+            enable_child_certificate_validation_cache: false,
+            publication_point_cache_observe_only: false,
+            enable_publication_point_validation_cache: false,
+            enable_transport_request_prefetch: false,
+        },
+    )
+    .expect("run tree depth-two-limited");
+    assert_eq!(out.instances_processed, 3);
+    assert_eq!(out.instances_failed, 0);
+    assert_eq!(
+        runner.called(),
+        vec![
+            root_manifest,
+            root_manifest,
+            child_manifest,
+            root_manifest,
+            child_manifest,
+            grandchild_manifest,
+        ]
+    );
 
     let out = run_tree_serial(
         ca_handle(root_manifest),