20260828 收口Feature151 TA constraints优化与验证

deploy/docker-installer/compose/grafana/dashboards/ours-rp-soak-overview.json

@@ -1108,6 +1108,85 @@
       ],
       "title": "CCR Format Check Results",
       "type": "timeseries"
+    },
+    {
+      "datasource": {
+        "type": "prometheus",
+        "uid": "Prometheus"
+      },
+      "fieldConfig": {
+        "defaults": {
+          "color": {
+            "mode": "palette-classic"
+          },
+          "decimals": 0,
+          "min": 0,
+          "unit": "none"
+        },
+        "overrides": [
+          {
+            "matcher": {
+              "id": "byName",
+              "options": "ber_compatible_cms_encoding"
+            },
+            "properties": [
+              {
+                "id": "color",
+                "value": {
+                  "fixedColor": "orange",
+                  "mode": "fixed"
+                }
+              }
+            ]
+          },
+          {
+            "matcher": {
+              "id": "byName",
+              "options": "unclassified"
+            },
+            "properties": [
+              {
+                "id": "color",
+                "value": {
+                  "fixedColor": "gray",
+                  "mode": "fixed"
+                }
+              }
+            ]
+          }
+        ]
+      },
+      "gridPos": {
+        "x": 0,
+        "y": 68,
+        "w": 24,
+        "h": 8
+      },
+      "id": 24,
+      "options": {
+        "legend": {
+          "calcs": [
+            "lastNotNull",
+            "max"
+          ],
+          "displayMode": "table",
+          "placement": "bottom",
+          "showLegend": true
+        },
+        "tooltip": {
+          "mode": "multi",
+          "sort": "none"
+        }
+      },
+      "targets": [
+        {
+          "expr": "sum by (warning_category) (ours_rp_warnings_by_category)",
+          "legendFormat": "{{warning_category}}",
+          "refId": "A"
+        }
+      ],
+      "title": "Warnings by Category",
+      "type": "timeseries"
     }
   ],
   "refresh": "5s",
@@ -1128,6 +1207,6 @@
   "timezone": "browser",
   "title": "Ours RP Soak Overview",
   "uid": "ours-rp-soak-overview",
-  "version": 4,
+  "version": 5,
   "weekStart": ""
 }

docker/ours-rp-runtime.Dockerfile

@@ -81,15 +81,24 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \
           ;; \
       esac; \
     fi; \
+    # The public crate CDN occasionally rejects a burst of parallel fetches
+    # from a BuildKit worker.  Keep the release build resilient without
+    # changing the produced binaries or their dependency graph.
+    export \
+      CARGO_NET_RETRY=10 \
+      CARGO_HTTP_TIMEOUT=120 \
+      CARGO_HTTP_MULTIPLEXING=false; \
     cargo build --release --target "$target_triple" \
       --bin rpki \
       --bin rpki_daemon \
-      --bin db_stats; \
+      --bin db_stats \
+      --bin rpki_der_ber_audit; \
     mkdir -p /build-out/bin; \
     cp \
       "target/$target_triple/release/rpki" \
       "target/$target_triple/release/rpki_daemon" \
       "target/$target_triple/release/db_stats" \
+      "target/$target_triple/release/rpki_der_ber_audit" \
       /build-out/bin/
 
 FROM --platform=$TARGETPLATFORM ${RUNTIME_IMAGE} AS runtime

scripts/docker/build_docker_runtime_image.sh

@@ -14,6 +14,10 @@ BUILDER_NAME="${BUILDER_NAME:-default}"
 INSTALL_BINFMT="${INSTALL_BINFMT:-1}"
 SAVE_IMAGE="${SAVE_IMAGE:-1}"
 LOAD_IMAGE="${LOAD_IMAGE:-1}"
+# BuildKit's default bridge network can differ from the Docker daemon proxy
+# configuration.  Set BUILD_NETWORK=host when the build container needs the
+# host's known-good registry connectivity.
+BUILD_NETWORK="${BUILD_NETWORK:-default}"
 
 usage() {
   cat <<'USAGE'
@@ -216,6 +220,7 @@ echo "builder_source_image: $builder_source_image"
 echo "runtime_source_image: $runtime_source_image"
 echo "builder_local_image: $BUILDER_IMAGE"
 echo "runtime_local_image: $RUNTIME_IMAGE"
+echo "build_network: $BUILD_NETWORK"
 echo "source_commit: $SOURCE_COMMIT_FULL"
 echo "source_dirty: $SOURCE_DIRTY"
 echo "build_timestamp_utc: $BUILD_TIMESTAMP_UTC"
@@ -225,6 +230,7 @@ if [[ "$LOAD_IMAGE" == "1" ]]; then
   docker buildx build \
     --platform "$target_platform" \
     --builder "$BUILDER_NAME" \
+    --network "$BUILD_NETWORK" \
     --pull=false \
     --load \
     --build-arg BUILDKIT_INLINE_CACHE=1 \
@@ -242,6 +248,7 @@ else
   docker buildx build \
     --platform "$target_platform" \
     --builder "$BUILDER_NAME" \
+    --network "$BUILD_NETWORK" \
     --pull=false \
     --output "type=docker,dest=$raw_tar_path" \
     --build-arg BUILDKIT_INLINE_CACHE=1 \

scripts/soak/build_portable_soak_package.sh

@@ -88,6 +88,9 @@ mkdir -p "$STAGE_DIR/bin" "$STAGE_DIR/fixtures" "$STAGE_DIR/scripts" "$STAGE_DIR
 install -m 0755 "$SCRIPT_DIR/run_soak.sh" "$STAGE_DIR/run_soak.sh"
 install -m 0755 "$SCRIPT_DIR/run_24h_soak_with_metrics.sh" "$STAGE_DIR/run_24h_soak_with_metrics.sh"
 install -m 0755 "$SCRIPT_DIR/run_cache_ablation_experiment.sh" "$STAGE_DIR/scripts/soak/run_cache_ablation_experiment.sh"
+install -m 0755 "$SCRIPT_DIR/run_feature149_cohort.sh" "$STAGE_DIR/scripts/soak/run_feature149_cohort.sh"
+install -m 0755 "$SCRIPT_DIR/summarize_feature149_cohort.py" "$STAGE_DIR/scripts/soak/summarize_feature149_cohort.py"
+install -m 0755 "$SCRIPT_DIR/render_feature149_report.py" "$STAGE_DIR/scripts/soak/render_feature149_report.py"
 install -m 0755 "$SCRIPT_DIR/fixed_phase_loop.sh" "$STAGE_DIR/scripts/soak/fixed_phase_loop.sh"
 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"
@@ -118,6 +121,7 @@ done
 
 cp -a "$REPO_ROOT/tests/fixtures/tal" "$STAGE_DIR/fixtures/"
 cp -a "$REPO_ROOT/tests/fixtures/ta" "$STAGE_DIR/fixtures/"
+cp -a "$REPO_ROOT/tests/fixtures/ta_constraints" "$STAGE_DIR/fixtures/"
 cp -a "$REPO_ROOT/scripts/periodic" "$STAGE_DIR/scripts/"
 cp -a "$REPO_ROOT/scripts/cir" "$STAGE_DIR/scripts/"
 cp -a "$REPO_ROOT/scripts/inter_rp" "$STAGE_DIR/scripts/"

scripts/soak/portable-soak.env.example

@@ -100,6 +100,15 @@ ENABLE_CHILD_CERTIFICATE_VALIDATION_CACHE=0
 # 传给 rpki 子进程的额外参数。多个参数用空格分隔。
 # 示例：RPKI_EXTRA_ARGS="--enable-roa-validation-cache"
 # 实验性 transport 预热：RPKI_EXTRA_ARGS="--enable-transport-request-prefetch --enable-roa-validation-cache"
+# TA constraints 的 RIPE 单 TA soak 示例（policy 并非默认启用）：
+# RIRS=ripe
+# RPKI_EXTRA_ARGS="--ta-constraints ripe-ncc=${PACKAGE_ROOT}/fixtures/ta_constraints/ripe-ncc-afrinic-deny.constraints"
+# all5 性能 A/B 的完整 AFRINIC IPv4 deny 示例（仅四个非 AFRINIC TA）：
+# RIRS=afrinic,apnic,arin,lacnic,ripe
+# RPKI_EXTRA_ARGS="--ta-constraints apnic-rfc7730-https=${PACKAGE_ROOT}/fixtures/ta_constraints/afrinic-full-ipv4-deny.constraints --ta-constraints arin=${PACKAGE_ROOT}/fixtures/ta_constraints/afrinic-full-ipv4-deny.constraints --ta-constraints lacnic=${PACKAGE_ROOT}/fixtures/ta_constraints/afrinic-full-ipv4-deny.constraints --ta-constraints ripe-ncc=${PACKAGE_ROOT}/fixtures/ta_constraints/afrinic-full-ipv4-deny.constraints"
+# M7 current-only AFRINIC holdings（delegated status=allocated/assigned；不含 legacy/ERX）：
+# RIRS=afrinic,apnic,arin,lacnic,ripe
+# RPKI_EXTRA_ARGS="--ta-constraints apnic-rfc7730-https=${PACKAGE_ROOT}/fixtures/ta_constraints/afrinic-current-ipv4-deny.constraints --ta-constraints arin=${PACKAGE_ROOT}/fixtures/ta_constraints/afrinic-current-ipv4-deny.constraints --ta-constraints lacnic=${PACKAGE_ROOT}/fixtures/ta_constraints/afrinic-current-ipv4-deny.constraints --ta-constraints ripe-ncc=${PACKAGE_ROOT}/fixtures/ta_constraints/afrinic-current-ipv4-deny.constraints"
 RPKI_EXTRA_ARGS=""
 
 # 所有出站 HTTP 请求（RRDP / TAL / TA / 死 repo 探针）的 User-Agent。

scripts/soak/render_feature149_report.py

@@ -0,0 +1,390 @@
+#!/usr/bin/env python3
+"""Render a self-contained final report for Feature #149."""
+
+from __future__ import annotations
+
+import argparse
+import html
+import json
+import statistics
+from pathlib import Path
+from typing import Any
+
+
+EXPECTED_COHORTS = (
+    "prefetch_pp_object_system",
+    "prefetch_pp_object_data",
+    "pp_object_system",
+    "pp_object_data",
+)
+
+
+def number(value: Any, digits: int = 0) -> str:
+    if not isinstance(value, (int, float)):
+        return "n/a"
+    return f"{value:,.{digits}f}"
+
+
+def load(path: Path) -> dict[str, Any]:
+    value = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(value, dict):
+        raise ValueError(f"{path} is not an object")
+    return value
+
+
+def latest(record: dict[str, Any], key: str) -> Any:
+    return (record.get("latestCounts") or {}).get(key)
+
+
+def stats(record: dict[str, Any], key: str) -> dict[str, Any]:
+    value = record.get(key) or {}
+    return value if isinstance(value, dict) else {}
+
+
+def delta_median(record: dict[str, Any]) -> float | None:
+    value = stats(record, "deltaWallMs").get("median")
+    return float(value) if isinstance(value, (int, float)) else None
+
+
+def stable_median(record: dict[str, Any]) -> float | None:
+    value = stats(record, "stableDeltaWallMs").get("median")
+    return float(value) if isinstance(value, (int, float)) else None
+
+
+def delta_mean(record: dict[str, Any]) -> float | None:
+    value = stats(record, "deltaWallMs").get("mean")
+    return float(value) if isinstance(value, (int, float)) else None
+
+
+def stable_mean(record: dict[str, Any]) -> float | None:
+    value = stats(record, "stableDeltaWallMs").get("mean")
+    return float(value) if isinstance(value, (int, float)) else None
+
+
+def percent_change(baseline: float | None, value: float | None) -> float | None:
+    if baseline in (None, 0) or value is None:
+        return None
+    return (value - baseline) / baseline * 100
+
+
+def percent_text(baseline: float | None, value: float | None) -> str:
+    change = percent_change(baseline, value)
+    return "n/a" if change is None else f"{change:+.1f}%"
+
+
+def stat_value(record: dict[str, Any], field: str, key: str) -> float | None:
+    value = stats(record, field).get(key)
+    return float(value) if isinstance(value, (int, float)) else None
+
+
+def svg_chart(records: list[dict[str, Any]]) -> str:
+    values = [delta_median(record) or 0 for record in records]
+    maximum = max(values) if values else 1
+    width, height, left, baseline = 860, 250, 62, 212
+    bar_width, gap = 112, 56
+    bars: list[str] = []
+    for index, (record, value) in enumerate(zip(records, values)):
+        x = left + index * (bar_width + gap)
+        bar_height = 0 if maximum == 0 else value / maximum * 164
+        y = baseline - bar_height
+        colour = "#0f766e" if record.get("disk") == "data" else "#2563eb"
+        label = html.escape(str(record.get("cohortId")))
+        bars.append(
+            f'<rect x="{x}" y="{y:.1f}" width="{bar_width}" height="{bar_height:.1f}" rx="7" fill="{colour}"/>'
+            f'<text x="{x + bar_width / 2:.1f}" y="{y - 8:.1f}" text-anchor="middle" class="value">{number(value / 1000, 2)}s</text>'
+            f'<text x="{x + bar_width / 2:.1f}" y="{baseline + 20}" text-anchor="middle" class="label">{label}</text>'
+        )
+    return f'''<svg viewBox="0 0 {width} {height}" role="img" aria-label="Delta median wall clock by cohort">
+      <line x1="{left}" y1="{baseline}" x2="{width - 28}" y2="{baseline}" stroke="#94a3b8"/>
+      <text x="8" y="24" class="axis">delta median wall clock</text>
+      {''.join(bars)}
+    </svg>'''
+
+
+def timeline_svg(records: list[dict[str, Any]]) -> str:
+    """Render the fixed serial order and the six run slots in each cohort."""
+    width, row_height, top = 930, 48, 36
+    height = top + row_height * len(records) + 18
+    parts = [
+        f'<svg viewBox="0 0 {width} {height}" role="img" aria-label="Serial cohort timeline">',
+        '<text x="8" y="18" class="axis">each row: snapshot (S) then five delta runs (D1–D5); adjacent starts are 600 s apart</text>',
+    ]
+    for row, record in enumerate(records):
+        y = top + row * row_height
+        label = html.escape(str(record.get("cohortId")))
+        parts.append(f'<text x="8" y="{y + 5}" class="label">{label}</text>')
+        parts.append(f'<line x1="310" y1="{y}" x2="890" y2="{y}" stroke="#cbd5e1" stroke-width="2"/>')
+        for index in range(6):
+            x = 310 + index * 116
+            mode = "S" if index == 0 else f"D{index}"
+            fill = "#2563eb" if index == 0 else "#0f766e"
+            parts.append(f'<circle cx="{x}" cy="{y}" r="11" fill="{fill}"/>')
+            parts.append(f'<text x="{x}" y="{y + 4}" text-anchor="middle" class="circle-label">{mode}</text>')
+    parts.append("</svg>")
+    return "".join(parts)
+
+
+def milestone_svg() -> str:
+    labels = (
+        ("M1", "data disk init"),
+        ("M2", "package + dry-run"),
+        ("M3", "prefetch cohorts"),
+        ("M4", "no-prefetch cohorts"),
+        ("M5", "evidence + report"),
+    )
+    width, y = 900, 44
+    parts = [f'<svg viewBox="0 0 {width} 90" role="img" aria-label="Milestone timeline"><line x1="72" y1="{y}" x2="828" y2="{y}" stroke="#cbd5e1" stroke-width="3"/>']
+    for index, (milestone, label) in enumerate(labels):
+        x = 72 + index * 189
+        parts.append(f'<circle cx="{x}" cy="{y}" r="17" fill="#0f766e"/><text x="{x}" y="{y + 5}" text-anchor="middle" class="circle-label">{milestone}</text><text x="{x}" y="78" text-anchor="middle" class="label">{label}</text>')
+    parts.append("</svg>")
+    return "".join(parts)
+
+
+def paired_rows(records: dict[str, dict[str, Any]]) -> list[tuple[str, dict[str, Any], dict[str, Any]]]:
+    return [
+        ("prefetch + PP + object", records["prefetch_pp_object_system"], records["prefetch_pp_object_data"]),
+        ("PP + object", records["pp_object_system"], records["pp_object_data"]),
+    ]
+
+
+def prefetch_rows(records: dict[str, dict[str, Any]]) -> list[tuple[str, dict[str, Any], dict[str, Any]]]:
+    return [
+        ("system disk", records["pp_object_system"], records["prefetch_pp_object_system"]),
+        ("data disk", records["pp_object_data"], records["prefetch_pp_object_data"]),
+    ]
+
+
+def run_rows(records: list[dict[str, Any]]) -> list[tuple[str, dict[str, Any]]]:
+    return [(str(record.get("cohortId")), run) for record in records for run in (record.get("runs") or [])]
+
+
+def all_delta_stats(records: list[dict[str, Any]]) -> dict[str, float]:
+    values = [
+        float(run["wallMs"])
+        for _, run in run_rows(records)
+        if run.get("syncMode") == "delta" and isinstance(run.get("wallMs"), (int, float))
+    ]
+    if not values:
+        return {}
+    return {
+        "count": float(len(values)),
+        "min": min(values),
+        "max": max(values),
+        "mean": statistics.mean(values),
+        "median": statistics.median(values),
+        "p90": statistics.quantiles(values, n=10, method="inclusive")[8] if len(values) > 1 else values[0],
+    }
+
+
+def conclusion_lines(indexed: dict[str, dict[str, Any]]) -> list[str]:
+    prefetch_system = indexed["prefetch_pp_object_system"]
+    prefetch_data = indexed["prefetch_pp_object_data"]
+    no_prefetch_system = indexed["pp_object_system"]
+    no_prefetch_data = indexed["pp_object_data"]
+    return [
+        f"两种磁盘的差异没有在两个 profile 上呈现同一方向：prefetch profile 的稳定 delta median 为 system {number(stable_median(prefetch_system) / 1000 if stable_median(prefetch_system) else None, 2)}s、data {number(stable_median(prefetch_data) / 1000 if stable_median(prefetch_data) else None, 2)}s（data vs system {percent_text(stable_median(prefetch_system), stable_median(prefetch_data))}）；无 prefetch profile 为 system {number(stable_median(no_prefetch_system) / 1000 if stable_median(no_prefetch_system) else None, 2)}s、data {number(stable_median(no_prefetch_data) / 1000 if stable_median(no_prefetch_data) else None, 2)}s（{percent_text(stable_median(no_prefetch_system), stable_median(no_prefetch_data))}）。",
+        f"在同一磁盘内，开启 transport prefetch 相对关闭 prefetch 的稳定 delta median 下降：system {percent_text(stable_median(no_prefetch_system), stable_median(prefetch_system))}，data {percent_text(stable_median(no_prefetch_data), stable_median(prefetch_data))}；这是本实验最稳定的方向性信号。",
+        f"snapshot 不与 delta 同向：system 开启 prefetch 后为 {percent_text((no_prefetch_system.get('snapshot') or {}).get('wallMs'), (prefetch_system.get('snapshot') or {}).get('wallMs'))}，data 为 {percent_text((no_prefetch_data.get('snapshot') or {}).get('wallMs'), (prefetch_data.get('snapshot') or {}).get('wallMs'))}；因此不能把单次 snapshot wall 解释为纯磁盘因果。",
+        "VRP/VAP/PP 产物量级和 warnings 随 live 发布窗口小幅变化，全部 run 成功且没有 recovery snapshot；网络同步长尾与 live 数据演进是主要限制。",
+    ]
+
+
+def write_markdown(records: list[dict[str, Any]], source_commit: str, package_sha256: str, remote_host: str, input_paths: list[Path], path: Path) -> None:
+    indexed = {str(record.get("cohortId")): record for record in records}
+    all_stats = all_delta_stats(records)
+    lines = [
+        "# Feature #149 远端系统盘/数据盘性能实验报告",
+        "",
+        f"- 远端：`{remote_host}`；源码：`{source_commit}`；portable archive SHA-256：`{package_sha256}`",
+        "- 结论口径：snapshot 单样本；全部 delta 为 delta1–delta5；稳定 delta 为 delta2–delta5。",
+        "",
+        "## 结论摘要",
+        "",
+    ]
+    lines.extend(f"- {line}" for line in conclusion_lines(indexed))
+    lines += [
+        "",
+        "## Cohort 摘要",
+        "",
+        "| cohort | disk | profile | total elapsed | snapshot wall | all delta min/mean/median/p90/max | stable median | max RSS | latest VRP/VAP/PP/warnings |",
+        "| --- | --- | --- | ---: | --- | ---: | ---: | ---: | --- |",
+    ]
+    for record in records:
+        delta = stats(record, "deltaWallMs")
+        lines.append(
+            "| {id} | {disk} | {profile} | {elapsed}s | {snapshot}ms | {min}/{mean}/{median}/{p90}/{max}ms | {stable}ms | {rss} KiB | {vrp}/{vap}/{pp}/{warnings} |".format(
+                id=record.get("cohortId"), disk=record.get("disk"), profile=record.get("profile"), elapsed=record.get("elapsedSeconds"),
+                snapshot=number((record.get("snapshot") or {}).get("wallMs"), 1), min=number(delta.get("min"), 1), mean=number(delta.get("mean"), 1),
+                median=number(delta.get("median"), 1), p90=number(delta.get("p90"), 1), max=number(delta.get("max"), 1), stable=number(stable_median(record), 1),
+                rss=number(record.get("maxRssKb")), vrp=latest(record, "vrps"), vap=latest(record, "vaps"), pp=latest(record, "publicationPoints"), warnings=latest(record, "warnings"),
+            )
+        )
+    lines += [
+        "",
+        "## 全部 delta 汇总",
+        "",
+        f"20 个 delta 样本：min/mean/median/p90/max = {number(all_stats.get('min'), 1)}/{number(all_stats.get('mean'), 1)}/{number(all_stats.get('median'), 1)}/{number(all_stats.get('p90'), 1)}/{number(all_stats.get('max'), 1)} ms。",
+        "",
+        "## 磁盘配对与 prefetch 对比",
+        "",
+        "| profile | system all/stable median | data all/stable median | data vs system（all / stable） |",
+        "| --- | ---: | ---: | ---: |",
+    ]
+    for label, system, data in paired_rows(indexed):
+        lines.append(f"| {label} | {number(delta_median(system), 1)} / {number(stable_median(system), 1)} ms | {number(delta_median(data), 1)} / {number(stable_median(data), 1)} ms | {percent_text(delta_median(system), delta_median(data))} / {percent_text(stable_median(system), stable_median(data))} |")
+    lines += [
+        "",
+        "| profile | system all/stable mean | data all/stable mean | data vs system（all / stable） |",
+        "| --- | ---: | ---: | ---: |",
+    ]
+    for label, system, data in paired_rows(indexed):
+        lines.append(f"| {label} | {number(delta_mean(system), 1)} / {number(stable_mean(system), 1)} ms | {number(delta_mean(data), 1)} / {number(stable_mean(data), 1)} ms | {percent_text(delta_mean(system), delta_mean(data))} / {percent_text(stable_mean(system), stable_mean(data))} |")
+    lines += [
+        "",
+        "| disk | no-prefetch stable median | prefetch stable median | prefetch vs no-prefetch |",
+        "| --- | ---: | ---: | ---: |",
+    ]
+    for label, no_prefetch, prefetch in prefetch_rows(indexed):
+        lines.append(f"| {label} | {number(stable_median(no_prefetch), 1)} ms | {number(stable_median(prefetch), 1)} ms | {percent_text(stable_median(no_prefetch), stable_median(prefetch))} |")
+    lines += [
+        "",
+        "| disk | no-prefetch stable mean | prefetch stable mean | prefetch vs no-prefetch |",
+        "| --- | ---: | ---: | ---: |",
+    ]
+    for label, no_prefetch, prefetch in prefetch_rows(indexed):
+        lines.append(f"| {label} | {number(stable_mean(no_prefetch), 1)} ms | {number(stable_mean(prefetch), 1)} ms | {percent_text(stable_mean(no_prefetch), stable_mean(prefetch))} |")
+    lines += ["", "## 逐 run 证据", "", "| cohort | run | mode | actual start UTC | wall ms | max RSS KiB | schedule lag ms | VRP/VAP/PP/warnings | artifacts |", "| --- | --- | --- | --- | ---: | ---: | ---: | --- | --- |"]
+    for cohort, run in run_rows(records):
+        lines.append(f"| {cohort} | {run.get('runId')} | {run.get('syncMode')} | {run.get('actualStartRfc3339Utc')} | {number(run.get('wallMs'), 1)} | {number(run.get('maxRssKb'))} | {number(run.get('scheduleLagMs'), 1)} | {run.get('vrps')}/{run.get('vaps')}/{run.get('publicationPoints')}/{run.get('warnings')} | {'complete' if run.get('artifactsComplete') and not run.get('missingArtifacts') else 'INCOMPLETE'} |")
+    lines += [
+        "",
+        "## 验收、异常与证据",
+        "",
+        "- 4 个 cohort 均为 6/6 success、首轮 snapshot 后连续五轮 delta，实际相邻启动节拍为 600 秒，`scheduleLagMs=0`；没有 recovery snapshot 或缺失核心产物。",
+        "- warnings 的最新值范围为 524–540（各 run 约 518–544），VRP/VAP/PP 在 live 窗口内小幅变化；这些属于可解释的 live/网络窗口变化，不是失败状态。",
+        "- 完整 run root 保留在远端对应系统盘或 `/data`；本地只回传四份 `cohort-summary.json/.md` 轻量证据。",
+        "",
+        "本报告由四份 `cohort-summary.json` 自动生成；由于 live RPKI 数据、网络和运行时间窗口不同，磁盘结果仅作方向性比较。",
+        "",
+        "### 本地证据",
+        "",
+    ]
+    lines.extend(f"- `{path}`" for path in input_paths)
+    lines += ["", "### 远端原始证据根", ""]
+    for record in records:
+        package_root = record.get("packageRoot", "n/a")
+        cohort = record.get("cohortId", "unknown")
+        lines.append(f"- `{cohort}` package/run root：`{package_root}`；cohort summary：`{package_root}/experiments/feature149-{cohort}/cohort-summary.json`")
+    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--cohort-summary", action="append", required=True, type=Path)
+    parser.add_argument("--out-html", required=True, type=Path)
+    parser.add_argument("--out-markdown", required=True, type=Path)
+    parser.add_argument("--remote-host", required=True)
+    parser.add_argument("--source-commit", required=True)
+    parser.add_argument("--package-sha256", default="see M2 package report")
+    args = parser.parse_args()
+    input_paths = list(args.cohort_summary)
+    records = [load(path) for path in input_paths]
+    indexed = {str(record.get("cohortId")): record for record in records}
+    if set(indexed) != set(EXPECTED_COHORTS):
+        raise ValueError(f"expected cohorts {EXPECTED_COHORTS}, got {tuple(indexed)}")
+    ordered = [indexed[name] for name in EXPECTED_COHORTS]
+    for record in ordered:
+        runs = record.get("runs") or []
+        if record.get("successCount") != 6 or record.get("runCount") != 6 or record.get("artifactStatus") != "complete" or len(runs) != 6:
+            raise ValueError(f"incomplete cohort {record.get('cohortId')}")
+        if runs[0].get("syncMode") != "snapshot" or any(run.get("syncMode") != "delta" for run in runs[1:]):
+            raise ValueError(f"unexpected sync modes in {record.get('cohortId')}")
+
+    all_stats = all_delta_stats(ordered)
+    cards = [
+        ("Cohorts", "4", "all complete"),
+        ("Runs", "24 / 24", "snapshot + 5 delta per cohort"),
+        ("All delta median", f"{number(all_stats.get('median', 0) / 1000, 2)}s", "20 live delta samples"),
+        ("Source", html.escape(args.source_commit), html.escape(args.remote_host)),
+    ]
+    card_html = "".join(f'<article class="card"><div>{title}</div><strong>{value}</strong><small>{detail}</small></article>' for title, value, detail in cards)
+
+    cohort_rows_html = "".join(
+        "<tr><td>{id}</td><td>{disk}</td><td>{profile}</td><td>{elapsed}s</td><td>{snapshot} ms</td><td>{min}/{mean}/{median}/{p90}/{max} ms</td><td>{stable} ms</td><td>{rss} KiB</td><td>{vrp}</td><td>{vap}</td><td>{pp}</td><td>{warnings}</td></tr>".format(
+            id=html.escape(str(record.get("cohortId"))), disk=html.escape(str(record.get("disk"))), profile=html.escape(str(record.get("profile"))), elapsed=number(record.get("elapsedSeconds")),
+            snapshot=number((record.get("snapshot") or {}).get("wallMs"), 1), min=number(stat_value(record, "deltaWallMs", "min"), 1), mean=number(stat_value(record, "deltaWallMs", "mean"), 1),
+            median=number(stat_value(record, "deltaWallMs", "median"), 1), p90=number(stat_value(record, "deltaWallMs", "p90"), 1), max=number(stat_value(record, "deltaWallMs", "max"), 1), stable=number(stable_median(record), 1),
+            rss=number(record.get("maxRssKb")), vrp=number(latest(record, "vrps")), vap=number(latest(record, "vaps")), pp=number(latest(record, "publicationPoints")), warnings=number(latest(record, "warnings")),
+        )
+        for record in ordered
+    )
+    paired_html = "".join(
+        "<tr><td>{label}</td><td>{system_all} / {system_stable}</td><td>{data_all} / {data_stable}</td><td>{all_change} / {stable_change}</td></tr>".format(
+            label=html.escape(label), system_all=number(delta_median(system), 1), system_stable=number(stable_median(system), 1), data_all=number(delta_median(data), 1), data_stable=number(stable_median(data), 1),
+            all_change=percent_text(delta_median(system), delta_median(data)), stable_change=percent_text(stable_median(system), stable_median(data)),
+        )
+        for label, system, data in paired_rows(indexed)
+    )
+    paired_mean_html = "".join(
+        "<tr><td>{label}</td><td>{system_all} / {system_stable}</td><td>{data_all} / {data_stable}</td><td>{all_change} / {stable_change}</td></tr>".format(
+            label=html.escape(label), system_all=number(delta_mean(system), 1), system_stable=number(stable_mean(system), 1), data_all=number(delta_mean(data), 1), data_stable=number(stable_mean(data), 1),
+            all_change=percent_text(delta_mean(system), delta_mean(data)), stable_change=percent_text(stable_mean(system), stable_mean(data)),
+        )
+        for label, system, data in paired_rows(indexed)
+    )
+    prefetch_html = "".join(
+        "<tr><td>{label}</td><td>{no_prefetch}</td><td>{prefetch}</td><td>{change}</td></tr>".format(
+            label=html.escape(label), no_prefetch=number(stable_median(no_prefetch), 1), prefetch=number(stable_median(prefetch), 1), change=percent_text(stable_median(no_prefetch), stable_median(prefetch)),
+        )
+        for label, no_prefetch, prefetch in prefetch_rows(indexed)
+    )
+    prefetch_mean_html = "".join(
+        "<tr><td>{label}</td><td>{no_prefetch}</td><td>{prefetch}</td><td>{change}</td></tr>".format(
+            label=html.escape(label), no_prefetch=number(stable_mean(no_prefetch), 1), prefetch=number(stable_mean(prefetch), 1), change=percent_text(stable_mean(no_prefetch), stable_mean(prefetch)),
+        )
+        for label, no_prefetch, prefetch in prefetch_rows(indexed)
+    )
+    run_rows_html = "".join(
+        "<tr><td>{cohort}</td><td>{run}</td><td>{mode}</td><td>{start}</td><td>{wall}</td><td>{rss}</td><td>{lag}</td><td>{vrp}/{vap}/{pp}/{warnings}</td><td>{artifact}</td></tr>".format(
+            cohort=html.escape(cohort), run=html.escape(str(run.get("runId"))), mode=html.escape(str(run.get("syncMode"))), start=html.escape(str(run.get("actualStartRfc3339Utc"))), wall=number(run.get("wallMs"), 1), rss=number(run.get("maxRssKb")), lag=number(run.get("scheduleLagMs"), 1),
+            vrp=number(run.get("vrps")), vap=number(run.get("vaps")), pp=number(run.get("publicationPoints")), warnings=number(run.get("warnings")), artifact="complete" if run.get("artifactsComplete") and not run.get("missingArtifacts") else "INCOMPLETE",
+        )
+        for cohort, run in run_rows(ordered)
+    )
+    evidence_rows_html = "".join(
+        "<tr><td>{cohort}</td><td><code>{root}</code></td><td><code>{summary}</code></td></tr>".format(
+            cohort=html.escape(str(record.get("cohortId"))),
+            root=html.escape(str(record.get("packageRoot", "n/a"))),
+            summary=html.escape(f"{record.get('packageRoot', 'n/a')}/experiments/feature149-{record.get('cohortId')}/cohort-summary.json"),
+        )
+        for record in ordered
+    )
+    warning_values = [latest(record, "warnings") for record in ordered if isinstance(latest(record, "warnings"), (int, float))]
+    max_delta = max((run.get("wallMs", 0) for _, run in run_rows(ordered) if run.get("syncMode") == "delta"), default=0)
+    report_html = f'''<!doctype html><html lang="zh-CN"><meta charset="utf-8"><title>Feature #149 性能实验报告</title>
+<style>
+body{{font-family:Inter,system-ui,sans-serif;margin:0;background:#f8fafc;color:#0f172a}}main{{max-width:1380px;margin:auto;padding:34px}}h1{{margin-bottom:4px}}.sub{{color:#475569}}.cards{{display:grid;grid-template-columns:repeat(4,1fr);gap:14px;margin:26px 0}}.card{{background:white;border:1px solid #e2e8f0;border-radius:12px;padding:16px}}.card strong{{display:block;font-size:25px;margin:8px 0}}.card small{{color:#64748b}}section{{background:white;border:1px solid #e2e8f0;border-radius:12px;padding:20px;margin-top:18px;overflow:auto}}table{{border-collapse:collapse;width:100%;font-size:13px}}th,td{{padding:9px;border-bottom:1px solid #e2e8f0;text-align:left;white-space:nowrap}}th{{background:#f1f5f9}}svg{{width:100%;height:auto;min-width:760px}}.axis,.label{{font-size:12px;fill:#475569}}.value{{font-size:13px;font-weight:700;fill:#0f172a}}.circle-label{{font-size:10px;fill:#fff;font-weight:700}}.note{{color:#475569;line-height:1.6}}li{{margin:7px 0;line-height:1.5}}code{{background:#f1f5f9;padding:2px 5px;border-radius:4px}}
+</style>
+<main><h1>Feature #149：系统盘 / 数据盘缓存性能实验</h1><p class="sub">远端 {html.escape(args.remote_host)} · source {html.escape(args.source_commit)} · archive SHA-256 {html.escape(args.package_sha256)} · 4 cohorts / 24 serial all5 runs</p><div class="cards">{card_html}</div>
+<section><h2>实验拓扑与时序</h2><p class="note">amd64 远端主机的 system cohort 将 portable package、state、数据和日志全部写在系统盘；data cohort 将对应内容全部写在 <code>/data</code>。四个 cohort 按 A-system → A-data → B-system → B-data 串行执行，每个 row 是 1 snapshot + 5 delta，相邻启动节拍 600 秒。</p>{timeline_svg(ordered)}</section>
+<section><h2>Milestone timeline</h2>{milestone_svg()}<p class="note">M1–M5 均已通过：磁盘初始化、同 checksum package 部署与 dry-run、A/B 两组正式运行、证据归档和报告收口。</p></section>
+<section><h2>主机、磁盘与 provenance</h2><p class="note"><code>{html.escape(args.remote_host)}</code>；system 为远端根文件系统（/dev/vda3），data 为 /data（/dev/vdb1、ext4、rpki-data）。同一 archive checksum 为 <code>{html.escape(args.package_sha256)}</code>，源码 commit 为 <code>{html.escape(args.source_commit)}</code>。对象 cache 固定包含 ROA 与 child-certificate validation cache；两 profile 的唯一 cache 差异是 transport prefetch。</p></section>
+<section><h2>结论摘要</h2><ul>{''.join(f'<li>{html.escape(line)}</li>' for line in conclusion_lines(indexed))}</ul></section>
+<section><h2>Delta median wall clock</h2>{svg_chart(ordered)}<p class="note">颜色：蓝色 system，青绿色 data。全体 delta 的 median 为 {number(all_stats.get('median'), 1)} ms；最大 delta 为 {number(max_delta, 1)} ms，长尾包含网络同步窗口。</p></section>
+<section><h2>逐 cohort KPI</h2><table><tr><th>cohort</th><th>disk</th><th>profile</th><th>elapsed</th><th>snapshot</th><th>delta min/mean/median/p90/max</th><th>stable median</th><th>max RSS</th><th>VRP</th><th>VAP</th><th>PP</th><th>warnings</th></tr>{cohort_rows_html}</table></section>
+<section><h2>磁盘配对（all delta / stable delta median）</h2><table><tr><th>profile</th><th>system ms</th><th>data ms</th><th>data vs system（all / stable）</th></tr>{paired_html}</table><p class="note">负值表示 data median 更快，正值表示更慢；snapshot 单样本另行列在 cohort KPI 表中。</p></section>
+<section><h2>磁盘配对（all delta / stable delta mean）</h2><table><tr><th>profile</th><th>system ms</th><th>data ms</th><th>data vs system（all / stable）</th></tr>{paired_mean_html}</table><p class="note">这里的 mean 是对应 cohort 的 deltaWallMs 平均值；负值表示 data mean 更快。</p></section>
+<section><h2>Prefetch 开关对比（stable delta median）</h2><table><tr><th>disk</th><th>no prefetch ms</th><th>prefetch ms</th><th>prefetch vs no prefetch</th></tr>{prefetch_html}</table></section>
+<section><h2>Prefetch 开关对比（stable delta mean）</h2><table><tr><th>disk</th><th>no prefetch ms</th><th>prefetch ms</th><th>prefetch vs no prefetch</th></tr>{prefetch_mean_html}</table></section>
+<section><h2>逐 run 原始指标</h2><table><tr><th>cohort</th><th>run</th><th>mode</th><th>actual start UTC</th><th>wall ms</th><th>max RSS KiB</th><th>schedule lag ms</th><th>VRP/VAP/PP/warnings</th><th>artifacts</th></tr>{run_rows_html}</table></section>
+<section><h2>验收与异常诊断</h2><ul><li>4 个 cohort 均为 6/6 success、首轮 snapshot 后连续五轮 delta；summary 中每 run 的核心产物均完整，实际启动间隔校验为 600 秒。</li><li>最新 warnings 范围为 {number(min(warning_values) if warning_values else None)}–{number(max(warning_values) if warning_values else None)}；VRP/VAP/PP 在 live 窗口内小幅变化，未出现输出塌缩或 recovery snapshot。</li><li>结果受 live RPKI 发布变化、RRDP/rsync 网络长尾和顺序时间窗口影响；不能仅凭四组 wall 值声称 IOPS 的严格因果效应。</li></ul><table><tr><th>cohort</th><th>远端 package/run root</th><th>远端 cohort summary</th></tr>{evidence_rows_html}</table><p class="note">本地输入证据为：{'; '.join(f'<code>{html.escape(str(path))}</code>' for path in input_paths)}。</p></section>
+<section><h2>统计口径</h2><p class="note">snapshot 统计只包含每个 cohort 的首轮；all delta 统计包含 delta1–delta5 共 20 个样本；stable delta 统计包含 delta2–delta5 共 16 个样本。详细 cache 命中、stage timing、命令 argv、run-meta/run-summary 和完整产物仍在远端 run root。</p></section></main></html>'''
+    args.out_html.parent.mkdir(parents=True, exist_ok=True)
+    args.out_markdown.parent.mkdir(parents=True, exist_ok=True)
+    args.out_html.write_text(report_html, encoding="utf-8")
+    write_markdown(ordered, args.source_commit, args.package_sha256, args.remote_host, input_paths, args.out_markdown)
+
+
+if __name__ == "__main__":
+    main()

scripts/soak/run_cache_ablation_experiment.sh

@@ -38,7 +38,8 @@ Environment:
   EXPERIMENT_RUN_ROOT          shared run root/state root; default PACKAGE_ROOT
   EXPERIMENT_DIR               experiment metadata output directory
   CASE_RUNS                    delta runs per case; default 10
-  EXPERIMENT_CASE_SET          default, cache-only, crypto-sig or sig-pp-compare; default runs the original 4-case matrix
+  EXPERIMENT_CASE_SET          default, cache-only, crypto-sig, sig-pp-compare,
+                               feature149-prefetch or feature149-no-prefetch
   RUN_START_INTERVAL_SECS      fixed start cadence for all runs; default 600
   FIRST_RUN_DELAY_SECS         delay before the first scheduled run; default 0
   SNAPSHOT_EXTRA_ARGS          extra rpki args for snapshot warmup
@@ -98,7 +99,8 @@ case_count() {
     crypto-sig) printf '%s' 4 ;;
     sig-pp-compare) printf '%s' 2 ;;
     baseline-combo) printf '%s' 2 ;;
-    *) die "EXPERIMENT_CASE_SET must be default, cache-only, crypto-sig or sig-pp-compare: $EXPERIMENT_CASE_SET" ;;
+    feature149-prefetch|feature149-no-prefetch) printf '%s' 1 ;;
+    *) die "unknown EXPERIMENT_CASE_SET: $EXPERIMENT_CASE_SET" ;;
   esac
 }
 
@@ -119,6 +121,8 @@ case_id_for_index() {
     sig-pp-compare:2) printf '%s' "pp-only" ;;
     baseline-combo:1) printf '%s' "baseline" ;;
     baseline-combo:2) printf '%s' "pp-object-sig" ;;
+    feature149-prefetch:1) printf '%s' "prefetch-pp-object" ;;
+    feature149-no-prefetch:1) printf '%s' "pp-object" ;;
     *) die "unknown case index: $1 for set $EXPERIMENT_CASE_SET" ;;
   esac
 }
@@ -140,6 +144,8 @@ case_name_for_index() {
     sig-pp-compare:2) printf '%s' "pp-cache-only" ;;
     baseline-combo:1) printf '%s' "all-cache-off" ;;
     baseline-combo:2) printf '%s' "pp-cache-object-crypto-sig-cache" ;;
+    feature149-prefetch:1) printf '%s' "prefetch-pp-object-cache" ;;
+    feature149-no-prefetch:1) printf '%s' "pp-object-cache" ;;
     *) die "unknown case index: $1 for set $EXPERIMENT_CASE_SET" ;;
   esac
 }
@@ -161,6 +167,8 @@ case_extra_args_for_index() {
     sig-pp-compare:2) printf '%s' "--enable-publication-point-validation-cache" ;;
     baseline-combo:1) printf '%s' "" ;;
     baseline-combo:2) printf '%s' "--enable-publication-point-validation-cache --enable-roa-validation-cache --enable-crypto-signature-cache" ;;
+    feature149-prefetch:1) printf '%s' "--enable-transport-request-prefetch --enable-publication-point-validation-cache --enable-roa-validation-cache --parallel-max-repo-sync-workers-global 4 --parallel-phase2-object-workers 4 --memory-trim-after-validation" ;;
+    feature149-no-prefetch:1) printf '%s' "--enable-publication-point-validation-cache --enable-roa-validation-cache --parallel-max-repo-sync-workers-global 4 --parallel-phase2-object-workers 4 --memory-trim-after-validation" ;;
     *) die "unknown case index: $1 for set $EXPERIMENT_CASE_SET" ;;
   esac
 }
@@ -168,6 +176,7 @@ case_extra_args_for_index() {
 case_child_cert_cache_for_index() {
   case "$EXPERIMENT_CASE_SET:$1" in
     default:4|cache-only:2|cache-only:3|crypto-sig:3|crypto-sig:4|baseline-combo:2) printf '%s' "1" ;;
+    feature149-prefetch:1|feature149-no-prefetch:1) printf '%s' "1" ;;
     default:1|default:2|default:3|cache-only:1|crypto-sig:1|crypto-sig:2|sig-pp-compare:1|sig-pp-compare:2|baseline-combo:1) printf '%s' "0" ;;
     *) die "unknown case index: $1 for set $EXPERIMENT_CASE_SET" ;;
   esac
@@ -207,6 +216,14 @@ elif case_set == "baseline-combo":
         {"caseId": "baseline", "caseName": "all-cache-off", "extraArgs": "", "enableChildCertificateValidationCache": False},
         {"caseId": "pp-object-sig", "caseName": "pp-cache-object-crypto-sig-cache", "extraArgs": "--enable-publication-point-validation-cache --enable-roa-validation-cache --enable-crypto-signature-cache", "enableChildCertificateValidationCache": True},
     ]
+elif case_set == "feature149-prefetch":
+    cases = [
+        {"caseId": "prefetch-pp-object", "caseName": "prefetch-pp-object-cache", "extraArgs": "--enable-transport-request-prefetch --enable-publication-point-validation-cache --enable-roa-validation-cache --parallel-max-repo-sync-workers-global 4 --parallel-phase2-object-workers 4 --memory-trim-after-validation", "enableChildCertificateValidationCache": True},
+    ]
+elif case_set == "feature149-no-prefetch":
+    cases = [
+        {"caseId": "pp-object", "caseName": "pp-object-cache", "extraArgs": "--enable-publication-point-validation-cache --enable-roa-validation-cache --parallel-max-repo-sync-workers-global 4 --parallel-phase2-object-workers 4 --memory-trim-after-validation", "enableChildCertificateValidationCache": True},
+    ]
 else:
     raise SystemExit(f"unknown case set: {case_set}")
 print(json.dumps(cases, ensure_ascii=False, indent=8))

scripts/soak/run_feature149_cohort.sh

@@ -0,0 +1,79 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PACKAGE_ROOT="${PACKAGE_ROOT:-$(cd "$SCRIPT_DIR/../.." && pwd)}"
+COHORT_ID="${COHORT_ID:?COHORT_ID is required}"
+COHORT_DISK="${COHORT_DISK:?COHORT_DISK is required (system or data)}"
+COHORT_PROFILE="${COHORT_PROFILE:?COHORT_PROFILE is required (prefetch-pp-object or pp-object)}"
+RUN_START_INTERVAL_SECS="${RUN_START_INTERVAL_SECS:-600}"
+CASE_RUNS="${CASE_RUNS:-5}"
+DRY_RUN="${DRY_RUN:-0}"
+FIRST_RUN_NOT_BEFORE_EPOCH="${FIRST_RUN_NOT_BEFORE_EPOCH:-0}"
+
+die() {
+  echo "error: $*" >&2
+  exit 2
+}
+
+is_true() {
+  case "${1:-}" in
+    1|true|TRUE|yes|YES|on|ON) return 0 ;;
+    *) return 1 ;;
+  esac
+}
+
+case "$COHORT_DISK" in
+  system|data) ;;
+  *) die "COHORT_DISK must be system or data: $COHORT_DISK" ;;
+esac
+
+case "$COHORT_PROFILE" in
+  prefetch-pp-object) case_set="feature149-prefetch" ;;
+  pp-object) case_set="feature149-no-prefetch" ;;
+  *) die "COHORT_PROFILE must be prefetch-pp-object or pp-object: $COHORT_PROFILE" ;;
+esac
+
+[[ "$CASE_RUNS" == "5" ]] || die "Feature #149 requires CASE_RUNS=5, got $CASE_RUNS"
+[[ "$RUN_START_INTERVAL_SECS" == "600" ]] || die "Feature #149 requires RUN_START_INTERVAL_SECS=600, got $RUN_START_INTERVAL_SECS"
+[[ -x "$SCRIPT_DIR/run_cache_ablation_experiment.sh" ]] || die "missing cache ablation driver"
+[[ -x "$SCRIPT_DIR/summarize_feature149_cohort.py" ]] || die "missing Feature #149 summarizer"
+[[ -f "$PACKAGE_ROOT/.env" ]] || die "missing package environment: $PACKAGE_ROOT/.env"
+
+experiment_dir="$PACKAGE_ROOT/experiments/feature149-$COHORT_ID"
+summary_jsonl="$experiment_dir/experiment-summary.jsonl"
+
+if (( FIRST_RUN_NOT_BEFORE_EPOCH > 0 )) && ! is_true "$DRY_RUN"; then
+  now_epoch="$(date +%s)"
+  if (( now_epoch < FIRST_RUN_NOT_BEFORE_EPOCH )); then
+    wait_secs=$(( FIRST_RUN_NOT_BEFORE_EPOCH - now_epoch ))
+    echo "[$(date -u +%Y-%m-%dT%H:%M:%SZ)] waiting ${wait_secs}s before cohort=$COHORT_ID" >&2
+    sleep "$wait_secs"
+  fi
+fi
+
+env \
+  PACKAGE_ROOT="$PACKAGE_ROOT" \
+  ENV_FILE="$PACKAGE_ROOT/.env" \
+  EXPERIMENT_RUN_ROOT="$PACKAGE_ROOT" \
+  EXPERIMENT_DIR="$experiment_dir" \
+  EXPERIMENT_CASE_SET="$case_set" \
+  CASE_RUNS="$CASE_RUNS" \
+  RUN_SNAPSHOT=1 \
+  RUN_START_INTERVAL_SECS="$RUN_START_INTERVAL_SECS" \
+  FIRST_RUN_DELAY_SECS=0 \
+  RETAIN_RUNS=6 \
+  DRY_RUN="$DRY_RUN" \
+  "$SCRIPT_DIR/run_cache_ablation_experiment.sh"
+
+if ! is_true "$DRY_RUN"; then
+  "$SCRIPT_DIR/summarize_feature149_cohort.py" \
+    --cohort-id "$COHORT_ID" \
+    --cohort-disk "$COHORT_DISK" \
+    --cohort-profile "$COHORT_PROFILE" \
+    --package-root "$PACKAGE_ROOT" \
+    --summary-jsonl "$summary_jsonl" \
+    --out-dir "$experiment_dir"
+fi
+
+printf 'feature149_cohort_dir=%s\n' "$experiment_dir"

scripts/soak/summarize_feature149_cohort.py

@@ -0,0 +1,169 @@
+#!/usr/bin/env python3
+"""Validate and summarize one Feature #149 six-run cohort."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import statistics
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+REQUIRED_ARTIFACTS = {
+    "run-meta.json",
+    "run-summary.json",
+    "input.cir",
+    "result.ccr",
+    "report.json",
+    "vrps.csv",
+    "vaps.csv",
+    "stage-timing.json",
+    "process-time.txt",
+}
+
+
+def load_jsonl(path: Path) -> list[dict[str, Any]]:
+    records: list[dict[str, Any]] = []
+    for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
+        if line.strip():
+            value = json.loads(line)
+            if not isinstance(value, dict):
+                raise ValueError(f"{path}:{line_number} is not an object")
+            records.append(value)
+    return records
+
+
+def require_success(records: list[dict[str, Any]], package_root: Path) -> list[dict[str, Any]]:
+    if len(records) != 6:
+        raise ValueError(f"expected 6 records (snapshot + 5 delta), found {len(records)}")
+    for index, record in enumerate(records):
+        expected_mode = "snapshot" if index == 0 else "delta"
+        if record.get("status") != "success":
+            raise ValueError(f"run {index + 1} status is {record.get('status')!r}")
+        if record.get("syncMode") != expected_mode:
+            raise ValueError(
+                f"run {index + 1} syncMode is {record.get('syncMode')!r}, expected {expected_mode!r}"
+            )
+        run_dir = package_root / "runs" / f"run_{int(record['maxRunIndexAfter']):04d}"
+        missing = sorted(name for name in REQUIRED_ARTIFACTS if not (run_dir / name).is_file())
+        record["runDir"] = str(run_dir)
+        record["artifactsComplete"] = not missing
+        record["missingArtifacts"] = missing
+        if missing:
+            raise ValueError(f"run {index + 1} missing required artifacts: {', '.join(missing)}")
+    return records
+
+
+def percentile(values: list[float], fraction: float) -> float | None:
+    if not values:
+        return None
+    ordered = sorted(values)
+    position = (len(ordered) - 1) * fraction
+    lower = int(position)
+    upper = min(lower + 1, len(ordered) - 1)
+    return ordered[lower] + (ordered[upper] - ordered[lower]) * (position - lower)
+
+
+def stats(values: list[int | float | None]) -> dict[str, float | None]:
+    nums = [float(value) for value in values if isinstance(value, (int, float))]
+    if not nums:
+        return {"count": 0, "min": None, "median": None, "p90": None, "max": None, "mean": None}
+    return {
+        "count": len(nums),
+        "min": min(nums),
+        "median": statistics.median(nums),
+        "p90": percentile(nums, 0.9),
+        "max": max(nums),
+        "mean": statistics.fmean(nums),
+    }
+
+
+def iso(epoch: Any) -> str | None:
+    if not isinstance(epoch, (int, float)):
+        return None
+    return datetime.fromtimestamp(epoch, timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--cohort-id", required=True)
+    parser.add_argument("--cohort-disk", required=True, choices=("system", "data"))
+    parser.add_argument("--cohort-profile", required=True, choices=("prefetch-pp-object", "pp-object"))
+    parser.add_argument("--package-root", required=True, type=Path)
+    parser.add_argument("--summary-jsonl", required=True, type=Path)
+    parser.add_argument("--out-dir", required=True, type=Path)
+    args = parser.parse_args()
+
+    records = require_success(load_jsonl(args.summary_jsonl), args.package_root)
+    delta = records[1:]
+    elapsed_seconds = records[-1]["completedEpoch"] - records[0]["actualStartEpoch"]
+    report = {
+        "feature": "149",
+        "createdAtUtc": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
+        "cohortId": args.cohort_id,
+        "disk": args.cohort_disk,
+        "profile": args.cohort_profile,
+        "packageRoot": str(args.package_root),
+        "runCount": len(records),
+        "successCount": len(records),
+        "artifactStatus": "complete",
+        "elapsedSeconds": elapsed_seconds,
+        "actualStartedAtUtc": iso(records[0].get("actualStartEpoch")),
+        "completedAtUtc": iso(records[-1].get("completedEpoch")),
+        "maxRssKb": max(int(record["maxRssKb"]) for record in records if isinstance(record.get("maxRssKb"), (int, float))),
+        "snapshot": {"wallMs": records[0].get("wallMs"), "maxRssKb": records[0].get("maxRssKb")},
+        "deltaWallMs": stats([record.get("wallMs") for record in delta]),
+        "stableDeltaWallMs": stats([record.get("wallMs") for record in delta[1:]]),
+        "latestCounts": {
+            key: records[-1].get(key)
+            for key in ("vrps", "vaps", "publicationPoints", "warnings")
+        },
+        "runs": records,
+    }
+    args.out_dir.mkdir(parents=True, exist_ok=True)
+    json_path = args.out_dir / "cohort-summary.json"
+    markdown_path = args.out_dir / "cohort-summary.md"
+    json_path.write_text(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    latest = report["latestCounts"]
+    markdown_path.write_text(
+        "\n".join(
+            [
+                f"# Feature #149 cohort: {args.cohort_id}",
+                "",
+                f"- 磁盘：`{args.cohort_disk}`",
+                f"- profile：`{args.cohort_profile}`",
+                f"- 状态：`success`，6/6 run 成功，必要产物完整",
+                f"- 总执行耗时：`{elapsed_seconds}s`",
+                f"- 最高 RSS：`{report['maxRssKb']} KiB`",
+                f"- snapshot wall：`{records[0].get('wallMs')} ms`",
+                f"- delta median / p90：`{report['deltaWallMs']['median']} ms` / `{report['deltaWallMs']['p90']} ms`",
+                f"- 最新产物：VRP `{latest['vrps']}`、VAP `{latest['vaps']}`、PP `{latest['publicationPoints']}`、warnings `{latest['warnings']}`",
+                "",
+                "| run | mode | wall ms | max RSS KiB | VRP | VAP | PP | warnings | artifacts |",
+                "| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | --- |",
+                *[
+                    "| {run} | {mode} | {wall} | {rss} | {vrps} | {vaps} | {pps} | {warnings} | {artifacts} |".format(
+                        run=index + 1,
+                        mode=record.get("syncMode"),
+                        wall=record.get("wallMs"),
+                        rss=record.get("maxRssKb"),
+                        vrps=record.get("vrps"),
+                        vaps=record.get("vaps"),
+                        pps=record.get("publicationPoints"),
+                        warnings=record.get("warnings"),
+                        artifacts="complete" if record.get("artifactsComplete") else "missing",
+                    )
+                    for index, record in enumerate(records)
+                ],
+                "",
+            ]
+        ),
+        encoding="utf-8",
+    )
+    print(json.dumps({"cohortSummaryJson": str(json_path), "cohortSummaryMarkdown": str(markdown_path), "elapsedSeconds": elapsed_seconds, "maxRssKb": report["maxRssKb"]}, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()

scripts/ta_constraints/generate_afrinic_current_holdings.py

@@ -0,0 +1,178 @@
+#!/usr/bin/env python3
+"""Generate a current-only AFRINIC IPv4 holdings constraints fixture.
+
+The AFRINIC delegated extended format contains one IPv4 allocation/assignment
+per line.  This tool keeps only active registry holdings (``allocated`` and
+``assigned``), computes their exact union, and emits an allow-all policy with
+the union as IPv4 deny entries.  ``reserved`` and ``available`` records are
+intentionally excluded.
+"""
+
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import hashlib
+import ipaddress
+import pathlib
+import sys
+import urllib.error
+import urllib.request
+from collections import Counter
+from typing import Iterable
+
+
+DEFAULT_SOURCE_URL = (
+    "https://ftp.afrinic.net/pub/stats/afrinic/"
+    "delegated-afrinic-extended-latest"
+)
+ACTIVE_STATUSES = {"allocated", "assigned"}
+
+
+def merge_intervals(
+    intervals: Iterable[tuple[int, int]],
+) -> list[tuple[int, int]]:
+    merged: list[tuple[int, int]] = []
+    for start, end in sorted(intervals):
+        if merged and start <= merged[-1][1] + 1:
+            merged[-1] = (merged[-1][0], max(merged[-1][1], end))
+        else:
+            merged.append((start, end))
+    return merged
+
+
+def parse_source(raw: bytes) -> tuple[list[tuple[int, int]], Counter[str], Counter[str]]:
+    status_counts: Counter[str] = Counter()
+    ipv4_status_counts: Counter[str] = Counter()
+    active: list[tuple[int, int]] = []
+    for line_number, raw_line in enumerate(raw.decode("utf-8", "replace").splitlines(), 1):
+        if not raw_line or raw_line.startswith("#"):
+            continue
+        fields = raw_line.split("|")
+        if len(fields) < 7 or fields[2] != "ipv4":
+            continue
+        status = fields[6]
+        status_counts[status] += 1
+        ipv4_status_counts[status] += 1
+        if status not in ACTIVE_STATUSES:
+            continue
+        try:
+            start = int(ipaddress.IPv4Address(fields[3]))
+            count = int(fields[4])
+        except (ValueError, TypeError) as error:
+            raise ValueError(f"line {line_number}: invalid IPv4 record: {raw_line}") from error
+        if count <= 0 or start + count > 1 << 32:
+            raise ValueError(f"line {line_number}: invalid IPv4 range: {raw_line}")
+        active.append((start, start + count - 1))
+    return active, status_counts, ipv4_status_counts
+
+
+def render(
+    source_url: str,
+    retrieved: str,
+    source_sha256: str,
+    source_ipv4_count: int,
+    status_counts: Counter[str],
+    active_source: list[tuple[int, int]],
+) -> tuple[str, dict[str, int]]:
+    merged = merge_intervals(active_source)
+    networks = [
+        network
+        for start, end in merged
+        for network in ipaddress.summarize_address_range(
+            ipaddress.IPv4Address(start), ipaddress.IPv4Address(end)
+        )
+    ]
+    networks.sort(key=lambda network: int(network.network_address))
+    active_addresses = sum(end - start + 1 for start, end in merged)
+    counts = {
+        "source_ipv4_records": source_ipv4_count,
+        "allocated": status_counts["allocated"],
+        "assigned": status_counts["assigned"],
+        "reserved": status_counts["reserved"],
+        "available": status_counts["available"],
+        "selected_source_records": len(active_source),
+        "selected_addresses": active_addresses,
+        "merged_intervals": len(merged),
+        "cidr_rules": len(networks),
+    }
+    lines = [
+        "# Feature #151 M7 current AFRINIC IPv4 holdings deny fixture.",
+        "#",
+        "# Source: AFRINIC delegated extended latest.",
+        f"# URL: {source_url}",
+        f"# Retrieved: {retrieved}",
+        f"# Source SHA-256: {source_sha256}",
+        "#",
+        "# Selection: IPv4 records with status allocated or assigned only.",
+        "# Excluded statuses: reserved and available (not current registry holdings).",
+        (
+            "# Source IPv4 records: {source_ipv4_records}; allocated={allocated}; "
+            "assigned={assigned}; reserved={reserved}; available={available}."
+        ).format(**counts),
+        "# Selected source records: {selected_source_records}; selected addresses: {selected_addresses}.".format(
+            **counts
+        ),
+        "# Overlap/adjacency union intervals: {merged_intervals}; exact CIDR deny rules: {cidr_rules}.".format(
+            **counts
+        ),
+        "# IPv6 and ASN remain allow-all to isolate the current AFRINIC IPv4 policy.",
+        "allow 0.0.0.0/0",
+        "allow ::/0",
+        "allow 0 - 4294967295",
+        "",
+    ]
+    lines.extend(f"deny {network}" for network in networks)
+    return "\n".join(lines) + "\n", counts
+
+
+def fetch_source(url: str, timeout: float) -> bytes:
+    request = urllib.request.Request(
+        url,
+        headers={"User-Agent": "rpki-dev-feature151-afrinic-holdings/1.0"},
+    )
+    with urllib.request.urlopen(request, timeout=timeout) as response:
+        return response.read()
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    source = parser.add_mutually_exclusive_group()
+    source.add_argument("--url", default=DEFAULT_SOURCE_URL, help="AFRINIC source URL")
+    source.add_argument("--input", type=pathlib.Path, help="local delegated file for replay")
+    parser.add_argument("--output", type=pathlib.Path, required=True)
+    parser.add_argument("--retrieved", help="UTC date recorded in the fixture header")
+    parser.add_argument("--timeout", type=float, default=60.0)
+    args = parser.parse_args()
+
+    source_url = args.url if args.input is None else f"file://{args.input.resolve()}"
+    raw = args.input.read_bytes() if args.input is not None else fetch_source(args.url, args.timeout)
+    retrieved = args.retrieved or dt.datetime.now(dt.timezone.utc).date().isoformat()
+    active, status_counts, ipv4_status_counts = parse_source(raw)
+    rendered, counts = render(
+        source_url,
+        retrieved,
+        hashlib.sha256(raw).hexdigest(),
+        sum(ipv4_status_counts.values()),
+        status_counts,
+        active,
+    )
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(rendered, encoding="utf-8")
+    print(
+        "generated {output}: {cidr_rules} CIDRs from {selected_source_records} active "
+        "records ({selected_addresses} addresses); source_sha256={sha256}".format(
+            output=args.output,
+            sha256=hashlib.sha256(raw).hexdigest(),
+            **counts,
+        )
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        raise SystemExit(main())
+    except (OSError, ValueError, urllib.error.URLError) as error:
+        print(f"error: {error}", file=sys.stderr)
+        raise SystemExit(2)

src/audit.rs

@@ -36,6 +36,7 @@ pub struct ObjectAuditEntry {
 #[derive(Clone, Debug, PartialEq, Eq, Serialize)]
 pub struct AuditWarning {
     pub message: String,
+    pub category: String,
     pub rfc_refs: Vec<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub context: Option<String>,
@@ -102,6 +103,7 @@ impl From<&crate::report::Warning> for AuditWarning {
     fn from(w: &crate::report::Warning) -> Self {
         Self {
             message: w.message.clone(),
+            category: w.category.as_str().to_string(),
             rfc_refs: w.rfc_refs.iter().map(|r| r.0.to_string()).collect(),
             context: w.context.clone(),
         }
@@ -238,6 +240,15 @@ mod tests {
         assert!(value.get("cir_fresh_objects").is_none());
         assert!(value.get("cir_cached_objects").is_none());
     }
+
+    #[test]
+    fn warning_audit_keeps_the_warning_category() {
+        let warning = crate::report::Warning::new("BER-compatible CMS accepted")
+            .with_category(crate::report::WarningCategory::BerCompatibleCmsEncoding);
+
+        let audit_warning = AuditWarning::from(&warning);
+        assert_eq!(audit_warning.category, "ber_compatible_cms_encoding");
+    }
 }
 
 #[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)]

src/bin/rpki_der_ber_audit.rs

@@ -0,0 +1,437 @@
+use rocksdb::{DB, IteratorMode, Options};
+use rpki::data_model::crl::RpkixCrl;
+use rpki::data_model::rc::ResourceCertificate;
+use rpki::data_model::signed_object::RpkiSignedObject;
+use serde::Serialize;
+use sha2::{Digest, Sha256};
+use std::collections::{BTreeMap, BTreeSet};
+use std::path::{Path, PathBuf};
+
+#[derive(Debug, Default)]
+struct Args {
+    input: Option<PathBuf>,
+    repo_bytes_db: Option<PathBuf>,
+    output: Option<PathBuf>,
+    pretty: bool,
+}
+
+#[derive(Clone, Debug, Serialize)]
+#[serde(rename_all = "snake_case")]
+enum EncodingStatus {
+    Der,
+    BerCompatibleOnly,
+    ParseFailed,
+    NotRpki,
+}
+
+#[derive(Clone, Debug, Serialize)]
+struct ObjectRecord {
+    sha256: String,
+    object_type: String,
+    encoding_status: EncodingStatus,
+    bytes: u64,
+    path: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    detail: Option<String>,
+}
+
+#[derive(Debug, Serialize)]
+struct AuditReport {
+    generated_by: &'static str,
+    input: String,
+    files_seen: u64,
+    recognized_rpki_objects: u64,
+    duplicate_content_files: u64,
+    non_rpki_entries: u64,
+    scan_errors: Vec<String>,
+    status_counts: BTreeMap<String, u64>,
+    object_type_status_counts: BTreeMap<String, BTreeMap<String, u64>>,
+    objects: Vec<ObjectRecord>,
+}
+
+fn usage() -> &'static str {
+    "Usage: rpki_der_ber_audit (--input <raw-rpki-object-directory>|--repo-bytes-db <repo-bytes.db>) [--output <audit.json>] [--compact]"
+}
+
+fn main() {
+    if let Err(err) = real_main() {
+        eprintln!("{err}");
+        std::process::exit(1);
+    }
+}
+
+fn real_main() -> Result<(), String> {
+    let args = parse_args(&std::env::args().collect::<Vec<_>>())?;
+    let input_label = args
+        .input
+        .as_ref()
+        .or(args.repo_bytes_db.as_ref())
+        .expect("validated input");
+    let mut report = AuditReport {
+        generated_by: "rpki_der_ber_audit",
+        input: input_label.display().to_string(),
+        files_seen: 0,
+        recognized_rpki_objects: 0,
+        duplicate_content_files: 0,
+        non_rpki_entries: 0,
+        scan_errors: Vec::new(),
+        status_counts: BTreeMap::new(),
+        object_type_status_counts: BTreeMap::new(),
+        objects: Vec::new(),
+    };
+    let mut seen_hashes = BTreeSet::new();
+    if let Some(input) = args.input.as_ref() {
+        scan_directory(input, &mut report, &mut seen_hashes)?;
+    } else if let Some(repo_bytes_db) = args.repo_bytes_db.as_ref() {
+        scan_repo_bytes_db(repo_bytes_db, &mut report, &mut seen_hashes)?;
+    }
+    report
+        .objects
+        .sort_by(|left, right| left.sha256.cmp(&right.sha256));
+    let rendered = if args.pretty {
+        serde_json::to_string_pretty(&report)
+    } else {
+        serde_json::to_string(&report)
+    }
+    .map_err(|err| format!("serialize audit report: {err}"))?;
+
+    if let Some(output) = args.output {
+        if let Some(parent) = output.parent() {
+            std::fs::create_dir_all(parent)
+                .map_err(|err| format!("create output directory {}: {err}", parent.display()))?;
+        }
+        std::fs::write(&output, rendered)
+            .map_err(|err| format!("write audit report {}: {err}", output.display()))?;
+    } else {
+        println!("{rendered}");
+    }
+    Ok(())
+}
+
+fn parse_args(argv: &[String]) -> Result<Args, String> {
+    let mut args = Args {
+        pretty: true,
+        ..Args::default()
+    };
+    let mut index = 1usize;
+    while index < argv.len() {
+        match argv[index].as_str() {
+            "--input" | "--in" => {
+                index += 1;
+                args.input = Some(PathBuf::from(
+                    argv.get(index).ok_or("--input requires a value")?,
+                ));
+            }
+            "--repo-bytes-db" => {
+                index += 1;
+                args.repo_bytes_db = Some(PathBuf::from(
+                    argv.get(index).ok_or("--repo-bytes-db requires a value")?,
+                ));
+            }
+            "--output" | "--out" => {
+                index += 1;
+                args.output = Some(PathBuf::from(
+                    argv.get(index).ok_or("--output requires a value")?,
+                ));
+            }
+            "--compact" => args.pretty = false,
+            "--pretty" => args.pretty = true,
+            "-h" | "--help" => return Err(usage().to_string()),
+            other => return Err(format!("unknown argument: {other}\n{}", usage())),
+        }
+        index += 1;
+    }
+    if args.input.is_some() == args.repo_bytes_db.is_some() {
+        return Err(format!("select exactly one input source\n{}", usage()));
+    }
+    if let Some(input) = args.input.as_ref()
+        && !input.is_dir()
+    {
+        return Err(format!("--input must be a directory: {}", input.display()));
+    }
+    if let Some(repo_bytes_db) = args.repo_bytes_db.as_ref()
+        && !repo_bytes_db.is_dir()
+    {
+        return Err(format!(
+            "--repo-bytes-db must be a RocksDB directory: {}",
+            repo_bytes_db.display()
+        ));
+    }
+    Ok(args)
+}
+
+fn scan_directory(
+    directory: &Path,
+    report: &mut AuditReport,
+    seen_hashes: &mut BTreeSet<String>,
+) -> Result<(), String> {
+    let entries = std::fs::read_dir(directory)
+        .map_err(|err| format!("read directory {}: {err}", directory.display()))?;
+    for entry in entries {
+        let entry = match entry {
+            Ok(entry) => entry,
+            Err(err) => {
+                report.scan_errors.push(format!(
+                    "read directory entry under {}: {err}",
+                    directory.display()
+                ));
+                continue;
+            }
+        };
+        let path = entry.path();
+        let file_type = match entry.file_type() {
+            Ok(file_type) => file_type,
+            Err(err) => {
+                report
+                    .scan_errors
+                    .push(format!("read file type {}: {err}", path.display()));
+                continue;
+            }
+        };
+        if file_type.is_dir() {
+            scan_directory(&path, report, seen_hashes)?;
+        } else if file_type.is_file() {
+            scan_file(&path, report, seen_hashes);
+        }
+    }
+    Ok(())
+}
+
+fn scan_file(path: &Path, report: &mut AuditReport, seen_hashes: &mut BTreeSet<String>) {
+    report.files_seen += 1;
+    let Some(object_type) = object_type_from_path(path) else {
+        return;
+    };
+    let bytes = match std::fs::read(path) {
+        Ok(bytes) => bytes,
+        Err(err) => {
+            report
+                .scan_errors
+                .push(format!("read {}: {err}", path.display()));
+            return;
+        }
+    };
+    let sha256 = hex::encode(Sha256::digest(&bytes));
+    let (encoding_status, detail) = classify_object(&object_type, &bytes);
+    record_object(
+        report,
+        seen_hashes,
+        sha256,
+        object_type,
+        encoding_status,
+        bytes.len() as u64,
+        path.display().to_string(),
+        detail,
+    );
+}
+
+fn scan_repo_bytes_db(
+    repo_bytes_db: &Path,
+    report: &mut AuditReport,
+    seen_hashes: &mut BTreeSet<String>,
+) -> Result<(), String> {
+    let mut options = Options::default();
+    options.create_if_missing(false);
+    let db = DB::open_for_read_only(&options, repo_bytes_db, false)
+        .map_err(|err| format!("open repo-bytes DB {}: {err}", repo_bytes_db.display()))?;
+    for entry in db.iterator(IteratorMode::Start) {
+        let (key, bytes) = entry.map_err(|err| format!("iterate repo-bytes DB: {err}"))?;
+        let Ok(key) = std::str::from_utf8(&key) else {
+            continue;
+        };
+        let Some(sha256) = key.strip_prefix("sha256:") else {
+            continue;
+        };
+        report.files_seen += 1;
+        let (object_type, encoding_status, detail) = classify_repo_blob(&bytes);
+        record_object(
+            report,
+            seen_hashes,
+            sha256.to_string(),
+            object_type,
+            encoding_status,
+            bytes.len() as u64,
+            format!("rocksdb://{}/{}", repo_bytes_db.display(), key),
+            detail,
+        );
+    }
+    Ok(())
+}
+
+#[allow(clippy::too_many_arguments)]
+fn record_object(
+    report: &mut AuditReport,
+    seen_hashes: &mut BTreeSet<String>,
+    sha256: String,
+    object_type: String,
+    encoding_status: EncodingStatus,
+    bytes: u64,
+    path: String,
+    detail: Option<String>,
+) {
+    if matches!(encoding_status, EncodingStatus::NotRpki) {
+        report.non_rpki_entries += 1;
+    } else {
+        report.recognized_rpki_objects += 1;
+    }
+    if !seen_hashes.insert(sha256.clone()) {
+        report.duplicate_content_files += 1;
+        return;
+    }
+    let status_name = encoding_status_name(&encoding_status).to_string();
+    *report.status_counts.entry(status_name.clone()).or_default() += 1;
+    *report
+        .object_type_status_counts
+        .entry(object_type.clone())
+        .or_default()
+        .entry(status_name)
+        .or_default() += 1;
+    report.objects.push(ObjectRecord {
+        sha256,
+        object_type,
+        encoding_status,
+        bytes,
+        path,
+        detail,
+    });
+}
+
+fn object_type_from_path(path: &Path) -> Option<String> {
+    let extension = path.extension()?.to_str()?.to_ascii_lowercase();
+    match extension.as_str() {
+        "mft" | "roa" | "asa" | "aspa" | "gbr" | "cer" | "crl" => Some(extension),
+        _ => None,
+    }
+}
+
+fn classify_object(object_type: &str, bytes: &[u8]) -> (EncodingStatus, Option<String>) {
+    match object_type {
+        "mft" | "roa" | "asa" | "aspa" | "gbr" => match RpkiSignedObject::parse_der(bytes) {
+            Ok(_) => match RpkiSignedObject::strict_cms_der_error(bytes) {
+                Some(err) => (EncodingStatus::BerCompatibleOnly, Some(err.to_string())),
+                None => (EncodingStatus::Der, None),
+            },
+            Err(err) => (EncodingStatus::ParseFailed, Some(err.to_string())),
+        },
+        "cer" => match ResourceCertificate::parse_der(bytes) {
+            Ok(_) => (EncodingStatus::Der, None),
+            Err(err) => (EncodingStatus::ParseFailed, Some(err.to_string())),
+        },
+        "crl" => match RpkixCrl::parse_der(bytes) {
+            Ok(_) => (EncodingStatus::Der, None),
+            Err(err) => (EncodingStatus::ParseFailed, Some(err.to_string())),
+        },
+        _ => unreachable!("object type was filtered before classification"),
+    }
+}
+
+fn classify_repo_blob(bytes: &[u8]) -> (String, EncodingStatus, Option<String>) {
+    if looks_like_tal_metadata(bytes) {
+        return (
+            "tal".to_string(),
+            EncodingStatus::NotRpki,
+            Some("TAL text metadata, not an RPKI object".to_string()),
+        );
+    }
+    if RpkiSignedObject::parse_der(bytes).is_ok() {
+        return match RpkiSignedObject::strict_cms_der_error(bytes) {
+            Some(err) => (
+                "cms_signed_object".to_string(),
+                EncodingStatus::BerCompatibleOnly,
+                Some(err.to_string()),
+            ),
+            None => ("cms_signed_object".to_string(), EncodingStatus::Der, None),
+        };
+    }
+    if ResourceCertificate::parse_der(bytes).is_ok() {
+        return ("cer".to_string(), EncodingStatus::Der, None);
+    }
+    if RpkixCrl::parse_der(bytes).is_ok() {
+        return ("crl".to_string(), EncodingStatus::Der, None);
+    }
+    (
+        "unknown".to_string(),
+        EncodingStatus::ParseFailed,
+        Some("not parsed as CMS, RPKI resource certificate, or CRL".to_string()),
+    )
+}
+
+fn encoding_status_name(status: &EncodingStatus) -> &'static str {
+    match status {
+        EncodingStatus::Der => "der",
+        EncodingStatus::BerCompatibleOnly => "ber_compatible_only",
+        EncodingStatus::ParseFailed => "parse_failed",
+        EncodingStatus::NotRpki => "not_rpki",
+    }
+}
+
+/// Return true for the TAL text format persisted by the live-TA refresh.
+///
+/// A TAL is intentionally not DER: it contains an rsync URI, an HTTPS URI,
+/// a blank separator, and the base64-encoded trust-anchor certificate.  It
+/// can share the repo-bytes DB with downloaded RPKI objects, so the audit must
+/// keep it out of the object encoding verdict rather than report a false
+/// parse failure.
+fn looks_like_tal_metadata(bytes: &[u8]) -> bool {
+    let Ok(text) = std::str::from_utf8(bytes) else {
+        return false;
+    };
+    let mut lines = text.lines();
+    let Some(rsync_uri) = lines.next() else {
+        return false;
+    };
+    let Some(https_uri) = lines.next() else {
+        return false;
+    };
+    if !rsync_uri.starts_with("rsync://") || !https_uri.starts_with("https://") {
+        return false;
+    }
+    // `str::lines` removes the blank line, so inspect the raw prefix to make
+    // sure the URI pair is followed by the TAL separator.
+    let Some(separator) = text.find("\n\n") else {
+        return false;
+    };
+    let base64 = &text[separator + 2..];
+    !base64.is_empty()
+        && base64.bytes().all(|byte| {
+            byte.is_ascii_alphanumeric() || matches!(byte, b'+' | b'/' | b'=' | b'\n' | b'\r')
+        })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn classifies_known_der_manifest_fixture() {
+        let path = Path::new(env!("CARGO_MANIFEST_DIR")).join(
+            "tests/fixtures/repository/ca.rg.net/rpki/RGnet-OU/bW-_qXU9uNhGQz21NR2ansB8lr0.mft",
+        );
+        let bytes = std::fs::read(path).expect("read manifest fixture");
+        assert!(matches!(
+            classify_object("mft", &bytes).0,
+            EncodingStatus::Der
+        ));
+    }
+
+    #[test]
+    fn recognizes_supported_object_extensions() {
+        assert_eq!(
+            object_type_from_path(Path::new("a.roa")),
+            Some("roa".to_string())
+        );
+        assert_eq!(object_type_from_path(Path::new("a.txt")), None);
+    }
+
+    #[test]
+    fn classifies_tal_metadata_as_not_rpki() {
+        let tal = b"rsync://rpki.example/ta.cer\nhttps://rpki.example/ta.cer\n\nMIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8A\n";
+        let (object_type, status, detail) = classify_repo_blob(tal);
+        assert_eq!(object_type, "tal");
+        assert!(matches!(status, EncodingStatus::NotRpki));
+        assert_eq!(
+            detail.as_deref(),
+            Some("TAL text metadata, not an RPKI object")
+        );
+    }
+}

src/cli.rs

@@ -26,6 +26,7 @@ use crate::parallel::config::{ParallelPhase1Config, ParallelPhase2Config};
 use crate::parallel::types::TalInputSpec;
 use crate::policy::{Policy, ResourceValidationMode, StrictPolicy};
 use crate::storage::{RocksStore, VcirStorageSummary};
+use crate::ta_constraints::TaConstraintsByTal;
 use crate::validation::run_tree_from_tal::{
     RunTreeFromTalAuditOutput, run_tree_from_multiple_tals_parallel_phase2_audit,
     run_tree_from_multiple_tals_parallel_phase2_audit_with_timing,
@@ -131,6 +132,7 @@ pub struct CliArgs {
     pub parallel_phase1_config: ParallelPhase1Config,
     pub parallel_phase2_config: ParallelPhase2Config,
     pub tal_inputs: Vec<TalInputSpec>,
+    pub ta_constraints: TaConstraintsByTal,
 
     pub db_path: PathBuf,
     pub raw_store_db: Option<PathBuf>,
@@ -210,6 +212,8 @@ Options:
   --raw-store-db <path>      External raw-by-hash store DB path (optional)
   --repo-bytes-db <path>     External repo object bytes DB path (optional)
   --policy <path>            Policy TOML path (optional)
+  --ta-constraints <tal-id>=<path>
+                             Apply local EE-resource constraints to one TAL (repeatable); adjacent <tal>.constraints files are auto-discovered
   --strict [policies]        Enable strict policies (default all; comma list: name,cms-der,signed-attrs; none disables)
   --resource-validation-mode <validation-update-03|rfc6487>
                              Resource certificate validation mode (default: validation-update-03)
@@ -318,6 +322,7 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
     let mut tal_urls: Vec<String> = Vec::new();
     let mut tal_paths: Vec<PathBuf> = Vec::new();
     let mut ta_paths: Vec<PathBuf> = Vec::new();
+    let mut ta_constraint_specs: Vec<String> = Vec::new();
     let mut parallel_phase1_cfg = ParallelPhase1Config::default();
     let mut parallel_phase2_cfg = ParallelPhase2Config::default();
     let mut dead_repo_blacklist_path: Option<PathBuf> = None;
@@ -427,6 +432,13 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
                 let v = argv.get(i).ok_or("--ta-path requires a value")?;
                 ta_paths.push(PathBuf::from(v));
             }
+            "--ta-constraints" => {
+                i += 1;
+                let v = argv
+                    .get(i)
+                    .ok_or("--ta-constraints requires <tal-id>=<path>")?;
+                ta_constraint_specs.push(v.clone());
+            }
             "--parallel-max-repo-sync-workers-global" => {
                 i += 1;
                 let v = argv
@@ -989,6 +1001,12 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
             ));
         }
     }
+    if verification_only && !ta_constraint_specs.is_empty() {
+        return Err(format!(
+            "--ta-constraints is not supported with --verification-only\\n\\n{}",
+            usage()
+        ));
+    }
     let tal_url = tal_urls.first().cloned();
     let tal_path = tal_paths.first().cloned();
     let ta_path = ta_paths.first().cloned();
@@ -1147,6 +1165,7 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
             tal_inputs.extend(tal_paths.iter().cloned().map(TalInputSpec::from_file_path));
         }
     }
+    let ta_constraints = TaConstraintsByTal::load_for_tals(&tal_inputs, &ta_constraint_specs)?;
 
     if dead_repo_blacklist_fail_threshold.is_some() && dead_repo_blacklist_path.is_none() {
         return Err(
@@ -1183,6 +1202,7 @@ pub fn parse_args(argv: &[String]) -> Result<CliArgs, String> {
         parallel_phase1_config: parallel_phase1_cfg,
         parallel_phase2_config: parallel_phase2_cfg,
         tal_inputs,
+        ta_constraints,
         db_path,
         raw_store_db,
         repo_bytes_db,
@@ -2090,7 +2110,11 @@ where
     H: crate::sync::rrdp::Fetcher + Clone + 'static,
     R: crate::fetch::rsync::RsyncFetcher + Clone + 'static,
 {
-    if args.verification_only || args.tal_inputs.len() > 1 {
+    // The multi-TAL entry point preserves the TAL id supplied by the CLI.
+    // A single-file TAL may otherwise derive its id from the embedded TA URI,
+    // which is intentionally different from the local filename used for
+    // adjacent <tal>.constraints discovery.
+    if args.verification_only || args.tal_inputs.len() > 1 || !policy.ta_constraints.is_empty() {
         return if let Some(t) = timing {
             run_tree_from_multiple_tals_parallel_phase2_audit_with_timing(
                 store,
@@ -2322,6 +2346,10 @@ pub fn run(argv: &[String]) -> Result<(), String> {
         if args.disable_rrdp {
             policy.sync_preference = crate::policy::SyncPreference::RsyncOnly;
         }
+        policy.ta_constraints = args.ta_constraints.clone();
+        for warning in policy.ta_constraints.configuration_warnings() {
+            eprintln!("warning: {warning}");
+        }
     }
     let validation_time = args
         .validation_time
@@ -3050,6 +3078,10 @@ pub fn run(argv: &[String]) -> Result<(), String> {
                 },
                 args.rsync_scope_policy,
             )?;
+            if !policy.ta_constraints.is_empty() {
+                contract.ta_constraints_fingerprint =
+                    Some(policy.ta_constraints.fingerprint_sha256_hex());
+            }
             contract.fallback_replay_selections =
                 crate::verification_only::collect_fallback_replay_selections(
                     store.as_ref(),

src/cli/tests.rs

@@ -23,6 +23,7 @@ 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("--ta-constraints"), "{err}");
     assert!(err.contains("--max-ca-depth"), "{err}");
     assert!(err.contains("default: 32"), "{err}");
     assert!(
@@ -37,6 +38,26 @@ fn parse_help_returns_usage() {
     assert!(!err.contains("--parallel-phase2          "), "{err}");
 }
 
+#[test]
+fn parse_accepts_explicit_ta_constraints_for_known_tal() {
+    let dir = tempfile::tempdir().expect("tmpdir");
+    let constraints_path = dir.path().join("example.constraints");
+    std::fs::write(&constraints_path, "allow 192.0.2.0/24\n").expect("write constraints");
+    let argv = vec![
+        "rpki".to_string(),
+        "--db".to_string(),
+        "db".to_string(),
+        "--tal-path".to_string(),
+        "example.tal".to_string(),
+        "--ta-path".to_string(),
+        "example.cer".to_string(),
+        "--ta-constraints".to_string(),
+        format!("example={}", constraints_path.display()),
+    ];
+    let args = parse_args(&argv).expect("parse args");
+    assert!(!args.ta_constraints.is_empty());
+}
+
 #[test]
 fn parse_rejects_unknown_argument() {
     let argv = vec![
@@ -119,6 +140,7 @@ fn verification_policy_preserves_contract_identity_and_selects_only_bound_fallba
             sync_preference: crate::policy::SyncPreference::RrdpThenRsync,
             ..Policy::default()
         },
+        ta_constraints_fingerprint: None,
         max_ca_depth: 32,
         max_instances: None,
         cache: crate::verification_only::ValidationCacheContract {

src/data_model/signed_object.rs

@@ -330,6 +330,16 @@ impl RpkiSignedObject {
         parse_signed_object_content_info(der, der, CmsParseMode::DerStrict)
     }
 
+    /// Return the strict-DER CMS parse error for an object that was otherwise
+    /// accepted through the normal BER-compatible CMS parser.
+    ///
+    /// Callers must only surface this as a compatibility warning after normal
+    /// decoding and validation have succeeded; a strict parse failure alone
+    /// does not prove that an arbitrary byte string is an RPKI signed object.
+    pub fn strict_cms_der_error(der: &[u8]) -> Option<SignedObjectParseError> {
+        Self::parse_der_strict_cms(der).err()
+    }
+
     /// Decode a DER-encoded RPKI Signed Object (CMS ContentInfo wrapping SignedData) and enforce
     /// the profile constraints from RFC 6488 §2-§3 and RFC 9589 §4.
     pub fn decode_der(der: &[u8]) -> Result<Self, SignedObjectDecodeError> {
@@ -1541,6 +1551,10 @@ mod tests {
         let err = RpkiSignedObject::parse_der_strict_cms(&mutated)
             .expect_err("DER strict rejects constructed OCTET STRING");
         assert!(err.to_string().contains("DER"), "{err}");
+        let compatibility_error = RpkiSignedObject::strict_cms_der_error(&mutated)
+            .expect("BER-compatible fixture must report strict-DER incompatibility");
+        assert_eq!(compatibility_error.to_string(), err.to_string());
+        assert!(RpkiSignedObject::strict_cms_der_error(&der).is_none());
     }
 
     fn replace_first_subslice(input: &[u8], from: &[u8], to: &[u8]) -> Option<Vec<u8>> {

src/lib.rs

@@ -40,6 +40,8 @@ pub mod storage;
 #[cfg(feature = "full")]
 pub mod sync;
 #[cfg(feature = "full")]
+pub mod ta_constraints;
+#[cfg(feature = "full")]
 pub mod tools;
 #[cfg(feature = "full")]
 pub mod validation;

src/policy.rs

@@ -1,6 +1,8 @@
 use serde::{Deserialize, Serialize};
 use std::collections::BTreeSet;
 
+use crate::ta_constraints::TaConstraintsByTal;
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum SyncPreference {
@@ -133,6 +135,10 @@ pub struct Policy {
     /// the already contract-bound VCIR fallback branch while replaying a run.
     #[serde(skip)]
     pub verification_forced_vcir_reuse_manifest_uris: BTreeSet<String>,
+    /// Locally configured, per-TAL EE certificate resource constraints.  They
+    /// are intentionally CLI/runtime-only rather than policy-file input.
+    #[serde(skip)]
+    pub ta_constraints: TaConstraintsByTal,
 }
 
 impl Default for Policy {
@@ -144,6 +150,7 @@ impl Default for Policy {
             resource_validation_mode: ResourceValidationMode::default(),
             strict: StrictPolicy::default(),
             verification_forced_vcir_reuse_manifest_uris: BTreeSet::new(),
+            ta_constraints: TaConstraintsByTal::default(),
         }
     }
 }

src/report.rs

@@ -1,9 +1,31 @@
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub struct RfcRef(pub &'static str);
 
+/// Stable categories for warnings written to audit reports and exported metrics.
+///
+/// New warning sites should use a specific category when one is available.  The
+/// default preserves the behaviour and schema of existing callers while making
+/// their metric label explicit.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub enum WarningCategory {
+    #[default]
+    Unclassified,
+    BerCompatibleCmsEncoding,
+}
+
+impl WarningCategory {
+    pub const fn as_str(self) -> &'static str {
+        match self {
+            Self::Unclassified => "unclassified",
+            Self::BerCompatibleCmsEncoding => "ber_compatible_cms_encoding",
+        }
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct Warning {
     pub message: String,
+    pub category: WarningCategory,
     pub rfc_refs: Vec<RfcRef>,
     pub context: Option<String>,
 }
@@ -12,11 +34,17 @@ impl Warning {
     pub fn new(message: impl Into<String>) -> Self {
         Self {
             message: message.into(),
+            category: WarningCategory::Unclassified,
             rfc_refs: Vec::new(),
             context: None,
         }
     }
 
+    pub fn with_category(mut self, category: WarningCategory) -> Self {
+        self.category = category;
+        self
+    }
+
     pub fn with_rfc_refs(mut self, refs: &[RfcRef]) -> Self {
         self.rfc_refs.extend_from_slice(refs);
         self

src/tools/rpki_artifact_metrics.rs

@@ -424,6 +424,7 @@ struct LatestRunMetrics {
     vaps: u64,
     publication_points: u64,
     warnings: u64,
+    warning_categories: BTreeMap<String, u64>,
     tree_instances_processed: Option<u64>,
     tree_instances_failed: Option<u64>,
     stage_seconds: BTreeMap<String, f64>,
@@ -1477,13 +1478,9 @@ fn parse_report(path: &Path, snapshot: &mut MetricsSnapshot, latest: &mut Latest
             .map(|a| a.len() as u64)
             .unwrap_or(0);
     }
-    latest.warnings = latest.warnings.max(
+    let (reported_warning_count, warning_categories) = collect_warning_categories(&report);
-        report
+    latest.warnings = latest.warnings.max(reported_warning_count);
-            .pointer("/tree/warnings")
+    latest.warning_categories = warning_categories;
-            .and_then(|v| v.as_array())
-            .map(|a| a.len() as u64)
-            .unwrap_or(0),
-    );
     if let Some(processed) = json_u64(&report, &["tree", "instances_processed"]) {
         latest.tree_instances_processed = Some(processed);
     }
@@ -1504,6 +1501,47 @@ fn parse_report(path: &Path, snapshot: &mut MetricsSnapshot, latest: &mut Latest
     }
 }
 
+fn collect_warning_categories(report: &Value) -> (u64, BTreeMap<String, u64>) {
+    let mut total = 0u64;
+    let mut categories = BTreeMap::new();
+    total += count_warning_categories(
+        report.pointer("/tree/warnings").and_then(Value::as_array),
+        &mut categories,
+    );
+    if let Some(publication_points) = report.get("publication_points").and_then(Value::as_array) {
+        for publication_point in publication_points {
+            total += count_warning_categories(
+                publication_point.get("warnings").and_then(Value::as_array),
+                &mut categories,
+            );
+        }
+    }
+    // Export the baseline and BER-compatibility series even when their count is
+    // zero, so dashboards can distinguish a clean run from a missing metric.
+    for category in ["unclassified", "ber_compatible_cms_encoding"] {
+        categories.entry(category.to_string()).or_insert(0);
+    }
+    (total, categories)
+}
+
+fn count_warning_categories(
+    warnings: Option<&Vec<Value>>,
+    categories: &mut BTreeMap<String, u64>,
+) -> u64 {
+    let Some(warnings) = warnings else {
+        return 0;
+    };
+    for warning in warnings {
+        let category = warning
+            .get("category")
+            .and_then(Value::as_str)
+            .filter(|value| !value.is_empty())
+            .unwrap_or("unclassified");
+        *categories.entry(category.to_string()).or_default() += 1;
+    }
+    warnings.len() as u64
+}
+
 fn extract_publication_point_metrics(
     pps: &[Value],
     downloads: Option<&Value>,
@@ -2409,6 +2447,17 @@ fn render_latest_metrics(writer: &mut PromWriter<'_>, instance: &str, latest: &L
         &[label("instance", instance)],
         latest.warnings as f64,
     );
+    for (category, count) in &latest.warning_categories {
+        writer.gauge(
+            "ours_rp_warnings_by_category",
+            "Latest run warning count by category",
+            &[
+                label("instance", instance),
+                label("warning_category", category),
+            ],
+            *count as f64,
+        );
+    }
     if let Some(value) = latest.tree_instances_processed {
         writer.gauge(
             "ours_rp_tree_instances",
@@ -3364,6 +3413,48 @@ mod tests {
         assert_eq!(args.out_metrics.as_deref(), Some(Path::new("metrics.prom")));
     }
 
+    #[test]
+    fn warning_categories_cover_tree_and_publication_point_warnings() {
+        let report = json!({
+            "tree": {
+                "warnings": [
+                    {"category": "ber_compatible_cms_encoding"},
+                    {"category": "ber_compatible_cms_encoding"}
+                ]
+            },
+            "publication_points": [
+                {"warnings": [{"category": "repository_sync"}]},
+                {"warnings": [{"message": "legacy warning without category"}]}
+            ]
+        });
+
+        let (total, categories) = collect_warning_categories(&report);
+        assert_eq!(total, 4);
+        assert_eq!(categories["ber_compatible_cms_encoding"], 2);
+        assert_eq!(categories["repository_sync"], 1);
+        assert_eq!(categories["unclassified"], 1);
+    }
+
+    #[test]
+    fn renders_warning_category_metrics() {
+        let mut warning_categories = BTreeMap::new();
+        warning_categories.insert("ber_compatible_cms_encoding".to_string(), 2);
+        let snapshot = MetricsSnapshot {
+            instance: "test".to_string(),
+            latest_run: Some(LatestRunMetrics {
+                warnings: 2,
+                warning_categories,
+                ..LatestRunMetrics::default()
+            }),
+            ..MetricsSnapshot::default()
+        };
+
+        let metrics = render_metrics(&snapshot);
+        assert!(metrics.contains(
+            r#"ours_rp_warnings_by_category{instance="test",warning_category="ber_compatible_cms_encoding"} 2"#
+        ));
+    }
+
     #[test]
     fn scan_fixture_exports_repo_pp_cir_and_ccr_metrics() {
         let td = TempDir::new().expect("tempdir");

src/validation/objects.rs

@@ -8,13 +8,13 @@ use crate::data_model::rc::{
     ResourceCertificate,
 };
 use crate::data_model::roa::{IpPrefix, RoaAfi, RoaDecodeError, RoaObject, RoaValidateError};
-use crate::data_model::signed_object::SignedObjectVerifyError;
+use crate::data_model::signed_object::{RpkiSignedObject, SignedObjectVerifyError};
 use crate::parallel::config::ParallelPhase2Config;
 use crate::parallel::object_worker::{
     ObjectTaskExecutor, ObjectWorkerPool, ObjectWorkerSubmitError,
 };
 use crate::policy::{Policy, ResourceValidationMode, SignedObjectFailurePolicy};
-use crate::report::{RfcRef, Warning};
+use crate::report::{RfcRef, Warning, WarningCategory};
 use crate::storage::{
     PackFile, PackTime, RoaCacheObjectMeta, RoaCacheProjection, VcirLocalOutput,
     VcirLocalOutputPayload, VcirOutputType, VcirSourceObjectType,
@@ -34,6 +34,22 @@ const RFC_CRLDP: &[RfcRef] = &[RfcRef("RFC 6487 §4.8.6")];
 const RFC_CRLDP_AND_LOCKED_PACK: &[RfcRef] =
     &[RfcRef("RFC 6487 §4.8.6"), RfcRef("RFC 9286 §4.2.1")];
 
+fn ber_compatible_cms_warning(der: &[u8], rsync_uri: &str, object_kind: &str) -> Option<Warning> {
+    let strict_error = RpkiSignedObject::strict_cms_der_error(der)?;
+    Some(
+        Warning::new(format!(
+            "accepted BER-compatible CMS encoding for {object_kind}: {rsync_uri}: {strict_error}"
+        ))
+        .with_category(WarningCategory::BerCompatibleCmsEncoding)
+        .with_rfc_refs(&[RfcRef("X.690 §10"), RfcRef("RFC 6488 §2")])
+        .with_context(rsync_uri),
+    )
+}
+
+fn ber_compatible_cms_warning_for_file(file: &PackFile, object_kind: &str) -> Option<Warning> {
+    ber_compatible_cms_warning(file.bytes().ok()?, &file.rsync_uri, object_kind)
+}
+
 fn sha256_hex_to_32(hex_value: &str) -> [u8; 32] {
     let bytes = hex::decode(hex_value).expect("internal sha256 hex should decode");
     let mut out = [0u8; 32];
@@ -945,6 +961,38 @@ pub fn process_publication_point_for_issuer_with_cache_options<P: PublicationPoi
     timing: Option<&TimingHandle>,
     collect_vcir_local_outputs: bool,
     roa_cache: RoaValidationCacheInput<'_>,
+) -> ObjectsOutput {
+    process_publication_point_for_issuer_with_cache_options_and_ta_constraints(
+        publication_point,
+        policy,
+        issuer_ca_der,
+        issuer_ca_rsync_uri,
+        issuer_effective_ip,
+        issuer_effective_as,
+        validation_time,
+        timing,
+        collect_vcir_local_outputs,
+        roa_cache,
+        None,
+    )
+}
+
+/// Serial signed-object processing with an optional, locally configured
+/// constraint set for the TA that owns this publication-point tree.
+pub fn process_publication_point_for_issuer_with_cache_options_and_ta_constraints<
+    P: PublicationPointData,
+>(
+    publication_point: &P,
+    policy: &Policy,
+    issuer_ca_der: &[u8],
+    issuer_ca_rsync_uri: Option<&str>,
+    issuer_effective_ip: Option<&crate::data_model::rc::IpResourceSet>,
+    issuer_effective_as: Option<&crate::data_model::rc::AsResourceSet>,
+    validation_time: time::OffsetDateTime,
+    timing: Option<&TimingHandle>,
+    collect_vcir_local_outputs: bool,
+    roa_cache: RoaValidationCacheInput<'_>,
+    ta_constraints: Option<&crate::ta_constraints::TaConstraints>,
 ) -> ObjectsOutput {
     let manifest_rsync_uri = publication_point.manifest_rsync_uri();
     let manifest_bytes = publication_point.manifest_bytes();
@@ -1015,6 +1063,12 @@ pub fn process_publication_point_for_issuer_with_cache_options<P: PublicationPoi
         }
     };
 
+    if let Some(warning) =
+        ber_compatible_cms_warning(manifest_bytes, manifest_rsync_uri, "manifest")
+    {
+        warnings.push(warning);
+    }
+
     // Decode issuer CA once; if it fails we cannot validate ROA/ASPA EE certificates.
     let issuer_ca = match decode_resource_certificate_with_policy(issuer_ca_der, policy) {
         Ok(v) => v,
@@ -1249,6 +1303,7 @@ pub fn process_publication_point_for_issuer_with_cache_options<P: PublicationPoi
                             policy.strict.cms_der,
                             policy.strict.name,
                             policy.resource_validation_mode,
+                            ta_constraints,
                         )
                     }
                     blocked @ (RoaCacheLookupResult::HashBlocked
@@ -1275,6 +1330,7 @@ pub fn process_publication_point_for_issuer_with_cache_options<P: PublicationPoi
                             policy.strict.cms_der,
                             policy.strict.name,
                             policy.resource_validation_mode,
+                            ta_constraints,
                         )
                     }
                 }
@@ -1298,6 +1354,7 @@ pub fn process_publication_point_for_issuer_with_cache_options<P: PublicationPoi
                     policy.strict.cms_der,
                     policy.strict.name,
                     policy.resource_validation_mode,
+                    ta_constraints,
                 )
             };
             match result.outcome {
@@ -1317,6 +1374,9 @@ pub fn process_publication_point_for_issuer_with_cache_options<P: PublicationPoi
                         result: AuditObjectResult::Ok,
                         detail: None,
                     });
+                    if let Some(warning) = ber_compatible_cms_warning_for_file(file, "ROA") {
+                        warnings.push(warning);
+                    }
                 }
                 Err(e) => match policy.signed_object_failure_policy {
                     SignedObjectFailurePolicy::DropObject => {
@@ -1412,6 +1472,7 @@ pub fn process_publication_point_for_issuer_with_cache_options<P: PublicationPoi
                 policy.strict.cms_der,
                 policy.strict.name,
                 policy.resource_validation_mode,
+                ta_constraints,
             ) {
                 Ok((att, local_output)) => {
                     stats.aspa_ok += 1;
@@ -1426,6 +1487,9 @@ pub fn process_publication_point_for_issuer_with_cache_options<P: PublicationPoi
                         result: AuditObjectResult::Ok,
                         detail: None,
                     });
+                    if let Some(warning) = ber_compatible_cms_warning_for_file(file, "ASPA") {
+                        warnings.push(warning);
+                    }
                 }
                 Err(e) => match policy.signed_object_failure_policy {
                     SignedObjectFailurePolicy::DropObject => {
@@ -1762,6 +1826,10 @@ pub(crate) struct RoaTaskShared {
     issuer_effective_ip: Option<Arc<crate::data_model::rc::IpResourceSet>>,
     issuer_effective_as: Option<Arc<crate::data_model::rc::AsResourceSet>>,
     resource_validation_mode: ResourceValidationMode,
+    /// Immutable constraints snapshot selected by the owning TAL.  Every
+    /// ROA task for a publication point shares this Arc, so workers do not
+    /// need a mutable/global policy lookup or a copy of the interval rules.
+    ta_constraints: Option<Arc<crate::ta_constraints::TaConstraints>>,
 }
 
 #[derive(Clone)]
@@ -1883,6 +1951,7 @@ fn validate_owned_roa_task(worker_index: usize, task: OwnedRoaTask) -> RoaTaskRe
         task.strict_cms_der,
         task.strict_name,
         task.resource_validation_mode,
+        shared.ta_constraints.as_deref(),
     )
     .map(|(vrps, local_outputs, cache_object_meta)| RoaTaskOk {
         vrps,
@@ -1981,6 +2050,40 @@ pub(crate) fn prepare_publication_point_for_parallel_roa_with_cache<P: Publicati
     validation_time: time::OffsetDateTime,
     collect_vcir_local_outputs: bool,
     roa_cache: RoaValidationCacheInput<'_>,
+) -> ParallelObjectsPrepare {
+    prepare_publication_point_for_parallel_roa_with_cache_and_ta_constraints(
+        publication_point_id,
+        publication_point,
+        policy,
+        issuer_ca_der,
+        issuer_ca_rsync_uri,
+        issuer_effective_ip,
+        issuer_effective_as,
+        validation_time,
+        collect_vcir_local_outputs,
+        roa_cache,
+        None,
+    )
+}
+
+/// Prepare a publication point for parallel ROA validation with the
+/// immutable constraints snapshot belonging to its TAL.  The snapshot is
+/// moved into the stage-owned shared payload and therefore remains available
+/// to detached ROA workers after the scoped phase-2 stage worker returns.
+pub(crate) fn prepare_publication_point_for_parallel_roa_with_cache_and_ta_constraints<
+    P: PublicationPointData,
+>(
+    publication_point_id: u64,
+    publication_point: &P,
+    policy: &Policy,
+    issuer_ca_der: &[u8],
+    issuer_ca_rsync_uri: Option<&str>,
+    issuer_effective_ip: Option<&crate::data_model::rc::IpResourceSet>,
+    issuer_effective_as: Option<&crate::data_model::rc::AsResourceSet>,
+    validation_time: time::OffsetDateTime,
+    collect_vcir_local_outputs: bool,
+    roa_cache: RoaValidationCacheInput<'_>,
+    ta_constraints: Option<Arc<crate::ta_constraints::TaConstraints>>,
 ) -> ParallelObjectsPrepare {
     let manifest_rsync_uri = publication_point.manifest_rsync_uri();
     let manifest_bytes = publication_point.manifest_bytes();
@@ -2050,6 +2153,12 @@ pub(crate) fn prepare_publication_point_for_parallel_roa_with_cache<P: Publicati
         }
     };
 
+    if let Some(warning) =
+        ber_compatible_cms_warning(manifest_bytes, manifest_rsync_uri, "manifest")
+    {
+        warnings.push(warning);
+    }
+
     let issuer_ca = match decode_resource_certificate_with_policy(issuer_ca_der, policy) {
         Ok(v) => v,
         Err(e) => {
@@ -2289,6 +2398,7 @@ pub(crate) fn prepare_publication_point_for_parallel_roa_with_cache<P: Publicati
             issuer_effective_ip: issuer_effective_ip.cloned().map(Arc::new),
             issuer_effective_as: issuer_effective_as.cloned().map(Arc::new),
             resource_validation_mode: policy.resource_validation_mode,
+            ta_constraints,
         }),
         validation_time,
         collect_vcir_local_outputs,
@@ -2367,6 +2477,9 @@ pub(crate) fn reduce_parallel_roa_stage(
                         result: AuditObjectResult::Ok,
                         detail: None,
                     });
+                    if let Some(warning) = ber_compatible_cms_warning_for_file(file, "ROA") {
+                        warnings.push(warning);
+                    }
                 }
                 Err(e) => {
                     audit.push(ObjectAuditEntry {
@@ -2404,6 +2517,7 @@ pub(crate) fn reduce_parallel_roa_stage(
                 strict_cms_der,
                 strict_name,
                 shared.resource_validation_mode,
+                shared.ta_constraints.as_deref(),
             ) {
                 Ok((att, local_output)) => {
                     stats.aspa_ok += 1;
@@ -2418,6 +2532,9 @@ pub(crate) fn reduce_parallel_roa_stage(
                         result: AuditObjectResult::Ok,
                         detail: None,
                     });
+                    if let Some(warning) = ber_compatible_cms_warning_for_file(file, "ASPA") {
+                        warnings.push(warning);
+                    }
                 }
                 Err(e) => {
                     audit.push(ObjectAuditEntry {
@@ -2620,6 +2737,9 @@ pub(crate) enum ObjectValidateError {
         "EE certificate resources are not a subset of issuer effective resources (RFC 6487 §7.2; RFC 3779)"
     )]
     EeResourcesNotSubset,
+
+    #[error("EE certificate violates locally configured TA constraints: {0}")]
+    TaConstraints(#[from] crate::ta_constraints::TaConstraintsViolation),
 }
 
 pub(crate) fn validate_roa_task_serial(
@@ -2639,6 +2759,7 @@ pub(crate) fn validate_roa_task_serial(
     strict_cms_der: bool,
     strict_name: bool,
     resource_validation_mode: ResourceValidationMode,
+    ta_constraints: Option<&crate::ta_constraints::TaConstraints>,
 ) -> RoaTaskResult {
     let outcome = process_roa_with_issuer(
         task.file,
@@ -2657,6 +2778,7 @@ pub(crate) fn validate_roa_task_serial(
         strict_cms_der,
         strict_name,
         resource_validation_mode,
+        ta_constraints,
     )
     .map(|(vrps, local_outputs, cache_object_meta)| RoaTaskOk {
         vrps,
@@ -2692,6 +2814,7 @@ fn process_roa_with_issuer(
     strict_cms_der: bool,
     strict_name: bool,
     resource_validation_mode: ResourceValidationMode,
+    ta_constraints: Option<&crate::ta_constraints::TaConstraints>,
 ) -> Result<(Vec<Vrp>, Vec<VcirLocalOutput>, Option<RoaCacheObjectMeta>), ObjectValidateError> {
     let _decode = timing
         .as_ref()
@@ -2752,6 +2875,10 @@ fn process_roa_with_issuer(
     )?;
     drop(_subset);
 
+    if let Some(ta_constraints) = ta_constraints {
+        ta_constraints.validate_ee_certificate(&ee.resource_cert)?;
+    }
+
     let vrps = roa_to_vrps_with_vrs(&roa, ee_vrs.ip.as_ref())?;
     let cache_object_meta = RoaCacheObjectMeta {
         source_object_uri: file.rsync_uri.clone(),
@@ -2821,6 +2948,7 @@ fn process_roa_with_issuer_parallel_cached(
     strict_cms_der: bool,
     strict_name: bool,
     resource_validation_mode: ResourceValidationMode,
+    ta_constraints: Option<&crate::ta_constraints::TaConstraints>,
 ) -> Result<(Vec<Vrp>, Vec<VcirLocalOutput>, Option<RoaCacheObjectMeta>), ObjectValidateError> {
     let _decode = timing
         .as_ref()
@@ -2887,6 +3015,10 @@ fn process_roa_with_issuer_parallel_cached(
     )?;
     drop(_subset);
 
+    if let Some(ta_constraints) = ta_constraints {
+        ta_constraints.validate_ee_certificate(&ee.resource_cert)?;
+    }
+
     let vrps = roa_to_vrps_with_vrs(&roa, ee_vrs.ip.as_ref())?;
     let cache_object_meta = RoaCacheObjectMeta {
         source_object_uri: file.rsync_uri.clone(),
@@ -2956,6 +3088,7 @@ fn process_aspa_with_issuer(
     strict_cms_der: bool,
     strict_name: bool,
     resource_validation_mode: ResourceValidationMode,
+    ta_constraints: Option<&crate::ta_constraints::TaConstraints>,
 ) -> Result<(AspaAttestation, Option<VcirLocalOutput>), ObjectValidateError> {
     let _decode = timing
         .as_ref()
@@ -3016,6 +3149,10 @@ fn process_aspa_with_issuer(
     )?;
     drop(_subset);
 
+    if let Some(ta_constraints) = ta_constraints {
+        ta_constraints.validate_ee_certificate(&ee.resource_cert)?;
+    }
+
     validate_aspa_customer_in_vrs(&aspa, ee_vrs.asn.as_ref())?;
 
     let attestation = AspaAttestation {
@@ -5465,6 +5602,13 @@ mod tests {
 
     #[test]
     fn parallel_stage_roa_tasks_share_stage_owned_payloads() {
+        let ta_constraints = Arc::new(
+            crate::ta_constraints::TaConstraints::from_file(
+                &std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+                    .join("tests/fixtures/ta_constraints/afrinic-current-ipv4-deny.constraints"),
+            )
+            .expect("load constraints fixture"),
+        );
         let stage = ParallelObjectsStage {
             publication_point_id: 7,
             shared: Arc::new(RoaTaskShared {
@@ -5495,6 +5639,7 @@ mod tests {
                 issuer_effective_ip: None,
                 issuer_effective_as: None,
                 resource_validation_mode: ResourceValidationMode::default(),
+                ta_constraints: Some(Arc::clone(&ta_constraints)),
             }),
             validation_time: OffsetDateTime::now_utc(),
             collect_vcir_local_outputs: false,
@@ -5514,6 +5659,14 @@ mod tests {
         let tasks = stage.build_roa_tasks();
         assert_eq!(tasks.len(), 2);
         assert!(Arc::ptr_eq(&tasks[0].shared, &tasks[1].shared));
+        assert!(Arc::ptr_eq(
+            tasks[0]
+                .shared
+                .ta_constraints
+                .as_ref()
+                .expect("task constraints snapshot"),
+            &ta_constraints
+        ));
     }
 
     #[test]

src/validation/run_tree_from_tal.rs

@@ -901,6 +901,10 @@ where
     H: Fetcher + Clone + 'static,
     R: crate::fetch::rsync::RsyncFetcher + Clone + 'static,
 {
+    // Constraints are an immutable per-run policy snapshot.  The phase-2
+    // ready-stage binds the snapshot to each CA's TAL and moves an Arc into
+    // ROA/ASPA worker state, so constrained multi-TAL runs retain the same
+    // parallel scheduler as unconstrained runs.
     let phase2_enabled = phase2_config.is_some();
     if tal_inputs.is_empty() {
         return Err(RunTreeFromTalError::Replay(

src/validation/tree_parallel.rs

@@ -16,7 +16,8 @@ use crate::validation::manifest::PublicationPointData;
 use crate::validation::manifest::PublicationPointSource;
 use crate::validation::objects::{
     ObjectsOutput, OwnedRoaTask, ParallelObjectsPrepare, ParallelObjectsStage,
-    RoaValidationCacheInput, prepare_publication_point_for_parallel_roa_with_cache,
+    RoaValidationCacheInput,
+    prepare_publication_point_for_parallel_roa_with_cache_and_ta_constraints,
     reduce_parallel_roa_stage,
 };
 use crate::validation::tree::{
@@ -1367,7 +1368,16 @@ fn compute_ready_publication_point_stage(
     } else {
         RoaValidationCacheInput::disabled()
     };
-    match prepare_publication_point_for_parallel_roa_with_cache(
+    let ta_constraints = runner
+        .policy
+        .ta_constraints
+        .shared_for_tal(&ready.node.handle.tal_id);
+    if ta_constraints.is_some() {
+        if let Some(timing) = runner.timing.as_ref() {
+            timing.record_count("ta_constraints_parallel_publication_points", 1);
+        }
+    }
+    match prepare_publication_point_for_parallel_roa_with_cache_and_ta_constraints(
         ready.node.id,
         &fresh_stage.fresh_point,
         runner.policy,
@@ -1378,6 +1388,7 @@ fn compute_ready_publication_point_stage(
         runner.validation_time,
         runner.persist_vcir,
         roa_cache,
+        ta_constraints,
     ) {
         ParallelObjectsPrepare::Complete(mut objects) => {
             metrics.prepare_ms = elapsed_ms(prepare_started);
@@ -3013,6 +3024,7 @@ mod tests {
         });
         sample.audit.warnings.push(crate::audit::AuditWarning {
             message: "warning".to_string(),
+            category: "unclassified".to_string(),
             rfc_refs: Vec::new(),
             context: None,
         });

src/validation/tree_runner.rs

@@ -53,6 +53,7 @@ use crate::validation::objects::{
     AspaAttestation, ParallelRoaWorkerPool, RoaValidationCacheInput, RoaValidationCacheView,
     RouterKeyPayload, Vrp, process_publication_point_for_issuer_parallel_roa_with_cache_options,
     process_publication_point_for_issuer_parallel_roa_with_pool_cache_options,
+    process_publication_point_for_issuer_with_cache_options_and_ta_constraints,
 };
 use crate::validation::publication_point::PublicationPointSnapshot;
 use crate::validation::tree::{
@@ -1551,12 +1552,30 @@ impl<'a> PublicationPointRunner for Rpkiv1PublicationPointRunner<'a> {
                     RoaValidationCacheInput::disabled()
                 };
                 let objects_processing_started = std::time::Instant::now();
+                let ta_constraints = self.policy.ta_constraints.for_tal(&ca.tal_id);
                 let mut objects = {
                     let _objects_total = self
                         .timing
                         .as_ref()
                         .map(|t| t.span_phase("objects_processing_total"));
-                    if let Some(phase2_pool) = self.parallel_roa_worker_pool.as_ref() {
+                    if let Some(ta_constraints) = ta_constraints {
+                        // This method is the serial/fallback publication-point path.  The
+                        // phase-2 scheduler uses the stage-specific parallel prepare path,
+                        // which carries the same immutable per-TAL snapshot into workers.
+                        process_publication_point_for_issuer_with_cache_options_and_ta_constraints(
+                            &fresh_point,
+                            self.policy,
+                            issuer_ca_der.as_ref(),
+                            ca.ca_certificate_rsync_uri.as_deref(),
+                            ca.effective_ip_resources.as_ref(),
+                            ca.effective_as_resources.as_ref(),
+                            self.validation_time,
+                            self.timing.as_ref(),
+                            false,
+                            roa_cache,
+                            Some(ta_constraints),
+                        )
+                    } else if let Some(phase2_pool) = self.parallel_roa_worker_pool.as_ref() {
                         process_publication_point_for_issuer_parallel_roa_with_pool_cache_options(
                             &fresh_point,
                             self.policy,
@@ -1971,13 +1990,13 @@ fn sha256_digest_32(bytes: impl AsRef<[u8]>) -> [u8; 32] {
     out
 }
 
-fn hash_serialized_parts(parts: &[(&str, Vec<u8>)]) -> [u8; 32] {
+fn hash_serialized_parts(parts: &[(&str, &[u8])]) -> [u8; 32] {
     let mut hasher = sha2::Sha256::new();
     for (label, value) in parts {
         hasher.update((label.len() as u64).to_be_bytes());
         hasher.update(label.as_bytes());
         hasher.update((value.len() as u64).to_be_bytes());
-        hasher.update(value);
+        hasher.update(*value);
     }
     let digest = hasher.finalize();
     let mut out = [0u8; 32];
@@ -1991,41 +2010,34 @@ fn cbor_or_debug_bytes<T: serde::Serialize + std::fmt::Debug>(value: &T) -> Vec<
 
 fn ta_context_digest_for_ca(ca: &CaInstanceHandle) -> [u8; 32] {
     hash_serialized_parts(&[
-        ("version", b"publication-point-cache-ta-v1".to_vec()),
+        ("version", b"publication-point-cache-ta-v1"),
-        ("tal_id", ca.tal_id.as_bytes().to_vec()),
+        ("tal_id", ca.tal_id.as_bytes()),
     ])
 }
 
 pub(crate) fn ca_validation_context_digest_for_ca(ca: &CaInstanceHandle) -> [u8; 32] {
-    hash_serialized_parts(&[
+    let parent_manifest = ca
-        (
+        .parent_manifest_rsync_uri
-            "version",
-            b"publication-point-cache-parent-context-v1".to_vec(),
-        ),
-        ("tal_id", ca.tal_id.as_bytes().to_vec()),
-        (
-            "parent_manifest",
-            ca.parent_manifest_rsync_uri
         .as_deref()
         .unwrap_or("")
-                .as_bytes()
+        .as_bytes();
-                .to_vec(),
+    let effective_ip = cbor_or_debug_bytes(&ca.effective_ip_resources);
-        ),
+    let effective_as = cbor_or_debug_bytes(&ca.effective_as_resources);
-        (
+    hash_serialized_parts(&[
-            "effective_ip",
+        ("version", b"publication-point-cache-parent-context-v1"),
-            cbor_or_debug_bytes(&ca.effective_ip_resources),
+        ("tal_id", ca.tal_id.as_bytes()),
-        ),
+        ("parent_manifest", parent_manifest),
-        (
+        ("effective_ip", effective_ip.as_slice()),
-            "effective_as",
+        ("effective_as", effective_as.as_slice()),
-            cbor_or_debug_bytes(&ca.effective_as_resources),
-        ),
     ])
 }
 
 pub(crate) fn publication_point_cache_policy_fingerprint(policy: &Policy) -> [u8; 32] {
+    let policy_bytes = cbor_or_debug_bytes(policy);
     hash_serialized_parts(&[
-        ("version", b"publication-point-cache-policy-v1".to_vec()),
+        ("version", b"publication-point-cache-policy-v3"),
-        ("policy", cbor_or_debug_bytes(policy)),
+        ("policy", policy_bytes.as_slice()),
+        ("ta_constraints", policy.ta_constraints.fingerprint_bytes()),
     ])
 }
 
@@ -2084,16 +2096,13 @@ fn child_certificate_cache_key_sha256_hex(
     policy_fingerprint: &[u8; 32],
 ) -> String {
     let digest = hash_serialized_parts(&[
-        ("version", b"child-certificate-cache-key-v1".to_vec()),
+        ("version", b"child-certificate-cache-key-v1"),
-        ("child_cert_uri", child_cert_uri.as_bytes().to_vec()),
+        ("child_cert_uri", child_cert_uri.as_bytes()),
-        ("child_cert_sha256", child_cert_sha256.to_vec()),
+        ("child_cert_sha256", child_cert_sha256),
-        ("issuer_ca_sha256", issuer_ca_sha256.to_vec()),
+        ("issuer_ca_sha256", issuer_ca_sha256),
         // Keep the persisted cache-key label stable; this change only renames Rust identifiers.
-        (
+        ("parent_context_digest", ca_validation_context_digest),
-            "parent_context_digest",
+        ("policy_fingerprint", policy_fingerprint),
-            ca_validation_context_digest.to_vec(),
-        ),
-        ("policy_fingerprint", policy_fingerprint.to_vec()),
     ]);
     sha256_hex_from_32(&digest)
 }
@@ -2986,6 +2995,24 @@ fn discover_children_from_fresh_snapshot_with_audit_cached_with_issuer_der<
 
                 match router_result {
                     Ok(router) => {
+                        if let Some(ta_constraints) = policy.ta_constraints.for_tal(&issuer.tal_id)
+                        {
+                            if let Err(error) =
+                                ta_constraints.validate_ee_certificate(&router.resource_cert)
+                            {
+                                router_error = router_error.saturating_add(1);
+                                audits.push(ObjectAuditEntry {
+                                    rsync_uri: f.rsync_uri.clone(),
+                                    sha256_hex: sha256_hex_from_32(&f.sha256),
+                                    kind: AuditObjectKind::RouterCertificate,
+                                    result: AuditObjectResult::Error,
+                                    detail: Some(format!(
+                                        "router certificate violates TA constraints: {error}"
+                                    )),
+                                });
+                                continue;
+                            }
+                        }
                         let router_asns = match router_asns_for_resource_mode(
                             &router.asns,
                             issuer.effective_as_resources.as_ref(),

src/verification_only.rs

@@ -73,6 +73,11 @@ pub struct ValidationContract {
     pub validation_time: String,
     pub binary_sha256: String,
     pub policy: Policy,
+    /// Constraints are currently normal-run-only.  Persist their semantic
+    /// fingerprint so verification-only can fail closed rather than replay a
+    /// constrained run with an empty runtime-only policy map.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub ta_constraints_fingerprint: Option<String>,
     pub max_ca_depth: usize,
     pub max_instances: Option<usize>,
     pub cache: ValidationCacheContract,
@@ -140,6 +145,7 @@ impl ValidationContract {
             validation_time: format_validation_time(validation_time)?,
             binary_sha256: current_binary_sha256()?,
             policy,
+            ta_constraints_fingerprint: None,
             max_ca_depth,
             max_instances,
             cache,
@@ -163,6 +169,9 @@ impl ValidationContract {
         }
         self.validation_time()?;
         validate_sha256_hex("validation contract binarySha256", &self.binary_sha256)?;
+        if let Some(fingerprint) = self.ta_constraints_fingerprint.as_deref() {
+            validate_sha256_hex("validation contract taConstraintsFingerprint", fingerprint)?;
+        }
         if self.max_ca_depth == 0 {
             return Err("validation contract maxCaDepth must be greater than zero".to_string());
         }
@@ -600,6 +609,12 @@ pub fn prepare_verification(
 
     let cir = read_cir(&artifacts.source_cir)?;
     let contract = read_validation_contract(&artifacts.source_contract)?;
+    if contract.ta_constraints_fingerprint.is_some() {
+        return Err(
+            "verification-only is not supported for a run with TA constraints; rerun normal validation with the same constraints"
+                .to_string(),
+        );
+    }
     if cir.validation_time.to_offset(time::UtcOffset::UTC) != contract.validation_time()? {
         return Err(format!(
             "source CIR validation time differs from validation contract: cir={}, contract={}",
@@ -895,6 +910,7 @@ mod tests {
             validation_time: "2026-07-16T00:00:00Z".to_string(),
             binary_sha256: "ab".repeat(32),
             policy: Policy::default(),
+            ta_constraints_fingerprint: None,
             max_ca_depth: 32,
             max_instances: None,
             cache: ValidationCacheContract {

tests/fixtures/ta_constraints/afrinic-full-ipv4-deny.constraints

@@ -0,0 +1,394 @@
+# Feature #151 remote all5 performance fixture.
+#
+# Source: official rpki-client 9.8 afrinic.constraints
+# ($OpenBSD: afrinic.constraints,v 1.4 2026/03/11 21:46:36 job Exp $).
+# Every IPv4 allow entry from that file (including recovered and transfer
+# ranges) is converted to deny. IPv6 and ASN entries are intentionally not
+# included because this fixture measures the full AFRINIC IPv4 policy.
+allow 0.0.0.0/0
+allow ::/0
+allow 0 - 4294967295
+
+deny 41.0.0.0/8
+deny 102.0.0.0/8
+deny 105.0.0.0/8
+deny 154.0.0.0/16
+deny 154.16.0.0/16
+deny 154.65.0.0 - 154.255.255.255
+deny 196.0.0.0 - 196.1.0.255
+deny 196.1.4.0/24
+deny 196.1.7.0 - 196.1.63.255
+deny 196.1.71.0/24
+deny 196.1.74.0 - 196.1.103.255
+deny 196.1.115.0 - 196.1.133.255
+deny 196.1.137.0/24
+deny 196.1.143.0 - 196.1.159.255
+deny 196.1.176.0 - 196.1.255.255
+deny 196.2.2.0/23
+deny 196.2.8.0 - 196.2.255.255
+deny 196.3.14.0/23
+deny 196.3.57.0 - 196.3.64.255
+deny 196.3.90.0/24
+deny 196.3.92.0 - 196.3.94.255
+deny 196.3.96.0/21
+deny 196.3.105.0/24
+deny 196.3.107.0 - 196.3.131.255
+deny 196.3.148.0/22
+deny 196.3.154.0 - 196.3.183.255
+deny 196.3.224.0 - 196.4.45.255
+deny 196.4.71.0 - 196.11.171.255
+deny 196.11.174.0 - 196.11.239.255
+deny 196.11.248.0/21
+deny 196.12.10.0 - 196.12.31.255
+deny 196.12.128.0/19
+deny 196.12.192.0 - 196.15.15.255
+deny 196.15.64.0 - 196.26.255.255
+deny 196.27.64.0 - 196.28.47.255
+deny 196.28.64.0 - 196.29.63.255
+deny 196.29.96.0 - 196.31.255.255
+deny 196.32.8.0 - 196.32.31.255
+deny 196.32.96.0/19
+deny 196.32.160.0 - 196.39.255.255
+deny 196.40.96.0 - 196.41.255.255
+deny 196.42.64.0 - 196.216.0.255
+deny 196.216.2.0 - 197.255.255.255
+deny 45.96.0.0 - 45.111.255.255
+deny 45.192.0.0 - 45.222.255.255
+deny 45.240.0.0 - 45.247.255.255
+deny 66.251.128.0 - 66.251.191.255
+deny 139.26.0.0 - 139.26.255.255
+deny 146.196.128.0 - 146.196.255.255
+deny 160.19.36.0 - 160.19.39.255
+deny 160.19.60.0 - 160.19.63.255
+deny 160.19.96.0 - 160.19.103.255
+deny 160.19.112.0 - 160.19.143.255
+deny 160.19.152.0 - 160.19.155.255
+deny 160.19.188.0 - 160.19.191.255
+deny 160.19.192.0 - 160.19.199.255
+deny 160.19.232.0 - 160.19.239.255
+deny 160.20.24.0 - 160.20.31.255
+deny 160.20.112.0 - 160.20.115.255
+deny 160.20.213.0 - 160.20.213.255
+deny 160.20.217.0 - 160.20.217.255
+deny 160.20.221.0 - 160.20.221.255
+deny 160.20.226.0 - 160.20.227.255
+deny 160.20.252.0 - 160.20.255.255
+deny 160.238.11.0 - 160.238.11.255
+deny 160.238.48.0 - 160.238.49.255
+deny 160.238.50.0 - 160.238.50.255
+deny 160.238.57.0 - 160.238.57.255
+deny 160.238.101.0 - 160.238.101.255
+deny 161.123.0.0 - 161.123.255.255
+deny 164.160.0.0 - 164.160.255.255
+deny 192.12.110.0 - 192.12.111.255
+deny 192.12.116.0 - 192.12.117.255
+deny 192.47.36.0 - 192.47.36.255
+deny 192.51.240.0 - 192.51.240.255
+deny 192.70.200.0 - 192.70.201.255
+deny 192.75.236.0 - 192.75.236.255
+deny 192.83.208.0 - 192.83.215.255
+deny 192.91.200.0 - 192.91.200.255
+deny 192.142.0.0 - 192.143.255.255
+deny 192.145.128.0 - 192.145.191.255
+deny 192.145.230.0 - 192.145.230.255
+deny 204.8.204.0 - 204.8.207.255
+deny 208.85.156.0 - 208.85.159.255
+deny 83.143.24.0 - 83.143.31.255
+deny 84.205.96.0 - 84.205.127.255
+deny 131.176.0.0 - 131.176.255.255
+deny 163.121.0.0 - 163.121.255.255
+deny 165.231.0.0 - 165.231.255.255
+deny 192.52.232.0 - 192.52.232.255
+deny 193.17.215.0 - 193.17.215.255
+deny 193.19.232.0 - 193.19.235.255
+deny 193.41.146.0 - 193.41.147.255
+deny 193.108.23.0 - 193.108.23.255
+deny 193.108.28.0 - 193.108.28.255
+deny 193.109.66.0 - 193.109.67.255
+deny 193.110.104.0 - 193.110.105.255
+deny 193.194.128.0 - 193.194.128.255
+deny 193.227.128.0 - 193.227.128.255
+deny 194.9.64.0 - 194.9.65.255
+deny 194.9.82.0 - 194.9.83.255
+deny 195.24.80.0 - 195.24.87.255
+deny 195.39.218.0 - 195.39.219.255
+deny 195.234.120.0 - 195.234.123.255
+deny 195.234.168.0 - 195.234.168.255
+deny 195.234.185.0 - 195.234.185.255
+deny 195.234.252.0 - 195.234.255.255
+deny 193.188.7.0/24
+deny 193.189.0.0/18
+deny 193.189.128.0/24
+deny 193.194.160.0/19
+deny 193.221.218.0/24
+deny 64.57.112.0 - 64.57.127.255
+deny 66.8.0.0 - 66.8.127.255
+deny 66.18.64.0 - 66.18.95.255
+deny 69.63.64.0 - 69.63.79.255
+deny 69.67.32.0 - 69.67.47.255
+deny 137.158.0.0 - 137.158.255.255
+deny 137.214.0.0 - 137.214.255.255
+deny 137.215.0.0 - 137.215.255.255
+deny 139.53.0.0 - 139.53.255.255
+deny 143.128.0.0 - 143.128.255.255
+deny 143.160.0.0 - 143.160.255.255
+deny 146.64.0.0 - 146.64.255.255
+deny 146.141.0.0 - 146.141.255.255
+deny 146.182.0.0 - 146.182.255.255
+deny 146.230.0.0 - 146.230.255.255
+deny 146.231.0.0 - 146.231.255.255
+deny 146.232.0.0 - 146.232.255.255
+deny 147.110.0.0 - 147.110.255.255
+deny 152.106.0.0 - 152.106.255.255
+deny 152.107.0.0 - 152.107.255.255
+deny 152.108.0.0 - 152.108.255.255
+deny 152.109.0.0 - 152.109.255.255
+deny 152.110.0.0 - 152.110.255.255
+deny 152.111.0.0 - 152.111.255.255
+deny 152.112.0.0 - 152.112.255.255
+deny 155.159.0.0 - 155.159.255.255
+deny 155.232.0.0 - 155.232.255.255
+deny 155.233.0.0 - 155.233.255.255
+deny 155.234.0.0 - 155.234.255.255
+deny 155.235.0.0 - 155.235.255.255
+deny 155.236.0.0 - 155.236.255.255
+deny 155.237.0.0 - 155.237.255.255
+deny 155.238.0.0 - 155.238.255.255
+deny 155.239.0.0 - 155.239.255.255
+deny 155.240.0.0 - 155.240.255.255
+deny 156.8.0.0 - 156.8.255.255
+deny 160.115.0.0 - 160.115.255.255
+deny 160.116.0.0 - 160.116.255.255
+deny 160.117.0.0 - 160.117.255.255
+deny 160.118.0.0 - 160.118.255.255
+deny 160.119.0.0 - 160.119.255.255
+deny 160.120.0.0 - 160.120.255.255
+deny 160.121.0.0 - 160.121.255.255
+deny 160.122.0.0 - 160.122.255.255
+deny 160.123.0.0 - 160.123.255.255
+deny 160.124.0.0 - 160.124.255.255
+deny 163.195.0.0 - 163.195.255.255
+deny 163.196.0.0 - 163.196.255.255
+deny 163.197.0.0 - 163.197.255.255
+deny 163.198.0.0 - 163.198.255.255
+deny 163.199.0.0 - 163.199.255.255
+deny 163.200.0.0 - 163.200.255.255
+deny 163.201.0.0 - 163.201.255.255
+deny 163.202.0.0 - 163.202.255.255
+deny 163.203.0.0 - 163.203.255.255
+deny 164.88.0.0 - 164.88.255.255
+deny 164.146.0.0 - 164.151.255.255
+deny 164.155.0.0 - 164.155.255.255
+deny 165.3.0.0 - 165.5.255.255
+deny 165.8.0.0 - 165.11.255.255
+deny 165.25.0.0 - 165.25.255.255
+deny 165.143.0.0 - 165.149.255.255
+deny 165.165.0.0 - 165.165.255.255
+deny 165.180.0.0 - 165.180.255.255
+deny 165.233.0.0 - 165.233.255.255
+deny 166.85.0.0 - 166.85.255.255
+deny 168.76.0.0 - 168.76.255.255
+deny 168.80.0.0 - 168.81.255.255
+deny 168.89.0.0 - 168.89.255.255
+deny 168.128.0.0 - 168.128.255.255
+deny 168.142.0.0 - 168.142.255.255
+deny 168.155.0.0 - 168.155.255.255
+deny 168.164.0.0 - 168.164.255.255
+deny 168.167.0.0 - 168.167.255.255
+deny 168.172.0.0 - 168.172.255.255
+deny 168.206.0.0 - 168.206.255.255
+deny 168.209.0.0 - 168.210.255.255
+deny 169.129.0.0 - 169.129.255.255
+deny 169.202.0.0 - 169.202.255.255
+deny 192.33.10.0 - 192.33.10.255
+deny 192.42.99.0 - 192.42.99.255
+deny 192.48.253.0 - 192.48.253.255
+deny 192.68.138.0 - 192.68.138.255
+deny 192.70.237.0 - 192.70.237.255
+deny 192.82.142.0 - 192.82.142.255
+deny 192.84.244.0 - 192.84.244.255
+deny 192.94.61.0 - 192.94.61.255
+deny 192.94.210.0 - 192.94.210.255
+deny 192.94.240.0 - 192.94.240.255
+deny 192.94.241.0 - 192.94.241.255
+deny 192.94.246.0 - 192.94.246.255
+deny 192.96.0.0 - 192.96.255.255
+deny 192.100.1.0 - 192.100.1.255
+deny 192.101.142.0 - 192.101.142.255
+deny 192.102.9.0 - 192.102.9.255
+deny 192.133.250.0 - 192.133.250.255
+deny 192.136.55.0 - 192.136.55.255
+deny 192.136.56.0 - 192.136.56.255
+deny 192.136.57.0 - 192.136.57.255
+deny 192.157.190.0 - 192.157.190.255
+deny 192.188.164.0 - 192.188.167.255
+deny 192.189.75.0 - 192.189.75.255
+deny 192.189.139.0 - 192.189.140.255
+deny 192.231.237.0 - 192.231.237.255
+deny 192.231.254.0 - 192.231.254.255
+deny 192.245.148.0 - 192.245.148.255
+deny 192.251.202.0 - 192.251.202.255
+deny 198.54.0.0 - 198.54.255.255
+deny 200.16.8.0 - 200.16.15.255
+deny 204.12.128.0 - 204.12.143.255
+deny 204.87.179.0 - 204.87.179.255
+deny 204.152.14.0 - 204.152.15.255
+deny 204.235.32.0 - 204.235.43.255
+deny 205.159.79.0 - 205.159.79.255
+deny 206.223.136.0 - 206.223.136.255
+deny 209.203.0.0 - 209.203.63.255
+deny 209.212.96.0 - 209.212.127.255
+deny 216.236.176.0 - 216.236.191.255
+deny 202.123.0.0/19
+deny 62.8.64.0/19
+deny 62.12.96.0/19
+deny 62.24.96.0/19
+deny 62.61.192.0/18
+deny 62.68.32.0/19
+deny 62.68.224.0/19
+deny 62.114.0.0/16
+deny 62.117.32.0/19
+deny 62.135.0.0/17
+deny 62.139.0.0/16
+deny 62.140.64.0/18
+deny 62.173.32.0/19
+deny 62.193.64.0/18
+deny 62.193.160.0/19
+deny 62.240.32.0/19
+deny 62.240.96.0/19
+deny 62.241.128.0/19
+deny 62.251.128.0/17
+deny 77.220.0.0/19
+deny 80.67.128.0/20
+deny 80.72.96.0/20
+deny 80.75.160.0/19
+deny 80.87.64.0/19
+deny 80.88.0.0/20
+deny 80.95.0.0/20
+deny 80.240.192.0/20
+deny 80.246.0.0/20
+deny 80.248.0.0/20
+deny 80.248.64.0/20
+deny 80.249.64.0/20
+deny 80.250.32.0/20
+deny 81.4.0.0/18
+deny 81.10.0.0/17
+deny 81.21.96.0/20
+deny 81.22.64.0/19
+deny 81.26.64.0/20
+deny 81.29.96.0/20
+deny 81.91.224.0/20
+deny 81.192.0.0/16
+deny 82.101.128.0/18
+deny 82.128.0.0/17
+deny 82.129.128.0/17
+deny 82.151.64.0/19
+deny 82.201.128.0/17
+deny 84.36.0.0/16
+deny 84.233.0.0/17
+deny 87.255.96.0/19
+deny 193.95.0.0/17
+deny 193.108.214.0/24
+deny 193.108.252.0/22
+deny 193.189.64.0 - 193.189.65.255
+deny 193.194.1.0 - 193.194.5.255
+deny 193.194.32.0 - 193.194.95.255
+deny 193.227.0.0/18
+deny 194.6.224.0/24
+deny 194.79.96.0/19
+deny 194.204.192.0/18
+deny 195.24.192.0/19
+deny 195.43.0.0/19
+deny 195.166.224.0/19
+deny 195.202.64.0/19
+deny 195.246.32.0/19
+deny 212.0.128.0/19
+deny 212.12.224.0/19
+deny 212.22.160.0/19
+deny 212.49.64.0/19
+deny 212.52.128.0/19
+deny 212.60.64.0/19
+deny 212.85.192.0/19
+deny 212.88.96.0/19
+deny 212.96.0.0/19
+deny 212.100.64.0/19
+deny 212.103.160.0/19
+deny 212.122.224.0/19
+deny 212.217.0.0/17
+deny 213.55.64.0/18
+deny 213.131.64.0/19
+deny 213.136.96.0/19
+deny 213.147.64.0/19
+deny 213.150.96.0/19
+deny 213.150.160.0 - 213.150.223.255
+deny 213.152.64.0/19
+deny 213.154.32.0 - 213.154.95.255
+deny 213.158.160.0/19
+deny 213.172.128.0/19
+deny 213.179.160.0/19
+deny 213.181.224.0/19
+deny 213.193.32.0/19
+deny 213.212.192.0/18
+deny 213.247.0.0/19
+deny 213.255.128.0/19
+deny 217.14.80.0/20
+deny 217.20.224.0/20
+deny 217.21.112.0/20
+deny 217.29.128.0/20
+deny 217.29.208.0/20
+deny 217.52.0.0/14
+deny 217.64.96.0/20
+deny 217.77.64.0/20
+deny 217.78.64.0/20
+deny 217.117.0.0/20
+deny 217.139.0.0/16
+deny 217.170.144.0/20
+deny 217.199.144.0/20
+deny 129.0.0.0/16
+deny 129.18.0.0/16
+deny 129.45.0.0/16
+deny 129.56.0.0/16
+deny 129.122.0.0/16
+deny 129.140.0.0/16
+deny 129.205.0.0/16
+deny 129.232.0.0/16
+deny 137.63.0.0 - 137.64.255.255
+deny 137.115.0.0/16
+deny 137.171.0.0/16
+deny 137.196.0.0/16
+deny 137.255.0.0/16
+deny 155.0.0.0/16
+deny 155.11.0.0 - 155.12.255.255
+deny 155.89.0.0/16
+deny 155.93.0.0/16
+deny 155.196.0.0/16
+deny 155.251.0.0/16
+deny 155.255.0.0 - 156.0.255.255
+deny 156.38.0.0/16
+deny 156.155.0.0 - 156.255.255.255
+deny 160.0.0.0/16
+deny 160.77.0.0/16
+deny 160.89.0.0 - 160.90.255.255
+deny 160.105.0.0/16
+deny 160.113.0.0/16
+deny 160.152.0.0/16
+deny 160.154.0.0 - 160.179.255.255
+deny 160.181.0.0 - 160.184.255.255
+deny 160.224.0.0 - 160.226.255.255
+deny 160.242.0.0/16
+deny 160.255.0.0/16
+deny 165.0.0.0/16
+deny 165.16.0.0/16
+deny 165.49.0.0 - 165.63.255.255
+deny 165.73.0.0/16
+deny 165.90.0.0/16
+deny 165.169.0.0/16
+deny 165.210.0.0/15
+deny 165.255.0.0/16
+deny 168.211.0.0 - 168.211.255.255
+deny 168.253.0.0/16
+deny 169.0.0.0/15
+deny 169.159.0.0/16
+deny 169.239.0.0/16
+deny 169.255.0.0/16
+deny 192.109.242.0/24

tests/fixtures/ta_constraints/local-custom-allow.constraints

@@ -0,0 +1,2 @@
+# local baseline-v1: the single valid ROA EE certificate has this IPv4 block
+allow 203.0.113.0/24

tests/fixtures/ta_constraints/local-custom-deny.constraints

@@ -0,0 +1,3 @@
+# local baseline-v1 negative control: deny wins over a covering allow rule
+allow 0.0.0.0/0
+deny 203.0.113.0/24

tests/fixtures/ta_constraints/ripe-ncc-afrinic-deny.constraints

@@ -0,0 +1,12 @@
+# Remote RIPE TA soak policy for feature #151.
+#
+# Start permissive for every INR kind, then carve out several /8 blocks
+# allocated to AFRINIC.  These denies should not match normal RIPE NCC
+# EE resources, while exercising deny-overrides-allow in a live run.
+allow 0.0.0.0/0
+allow ::/0
+allow 0 - 4294967295
+
+deny 41.0.0.0/8
+deny 102.0.0.0/8
+deny 105.0.0.0/8