20260721 metrics 服务新增 CCR rpki-client 格式检查指标与 Grafana 面板（#125）

20260714 补充 sync worker 消融实验远端运行脚本（#115 遗留）
20260721 rpctl ours RP 显式设置全局 repo sync workers 为 8
2026-07-21 18:07:05 +08:00 · 2026-07-21 16:10:09 +08:00 · 2026-07-21 15:55:15 +08:00 · 2026-07-21 15:55:15 +08:00
16 changed files with 2194 additions and 102 deletions
--- a/.dockerignore
+++ b/.dockerignore
@ -1,6 +1,7 @@
 target/
 .git/
 .gitignore
+docker/
 perf.*
 **/* copy.excalidraw
 ui/rpki-explorer/node_modules/
--- a/deploy/docker-installer/.env.example
+++ b/deploy/docker-installer/.env.example
@ -2,6 +2,8 @@
 # `build_docker_installer_package.sh` rewrites package-specific placeholders
 # when producing amd64/arm64 release packages.
 # 中文说明见 docs/README.zh-CN.md。English guide: docs/README.en.md
+# Every assignment is a required non-empty configuration key with a package
+# default. During upgrade: .env.example < reused old .env < command environment.

 # Package metadata and architecture guardrails.
 PACKAGE_ARCH=__PACKAGE_ARCH__
@ -86,6 +88,12 @@ RPKI_PROGRESS_CONTROL_LOOP_SLOW_MS=2000
 METRICS_INSTANCE=__PACKAGE_ARCH__-installer
 METRICS_PORT=9556
 METRICS_POLL_SECS=10
+# CCR rpki-client format check: the artifact-metrics container runs
+# `rpki-client -f result.ccr` once for every newly completed run and exports
+# ours_rp_ccr_format_check_* metrics. The binary is packaged inside the
+# metrics image; adjust only when deliberately testing another binary.
+CCR_CHECK_BIN=/opt/ours-rp/bin/rpki-client
+CCR_CHECK_TIMEOUT_SECS=120
 # Optional external RTR report directory produced by a separately deployed RTR service.
 # Default points to an installer-managed empty fallback directory. To enable real RTR
 # metrics, set this to the host-side report directory, for example:
--- a/deploy/docker-installer/compose/docker-compose.yml
+++ b/deploy/docker-installer/compose/docker-compose.yml
@ -39,6 +39,10 @@ services:
      - ${METRICS_POLL_SECS:-10}
      - --instance
      - ${METRICS_INSTANCE:-package-installer}
+      - --ccr-check-bin
+      - ${CCR_CHECK_BIN:-/opt/ours-rp/bin/rpki-client}
+      - --ccr-check-timeout-secs
+      - ${CCR_CHECK_TIMEOUT_SECS:-120}
    ports:
      - "${METRICS_PORT:-9556}:9556"
    volumes:
--- a/deploy/docker-installer/compose/grafana/dashboards/ours-rp-soak-overview.json
+++ b/deploy/docker-installer/compose/grafana/dashboards/ours-rp-soak-overview.json
@ -850,6 +850,264 @@
      ],
      "title": "Fresh PP / Object Counts by Run",
      "type": "timeseries"
+    },
+    {
+      "datasource": {
+        "type": "prometheus",
+        "uid": "Prometheus"
+      },
+      "fieldConfig": {
+        "defaults": {
+          "decimals": 0,
+          "unit": "none",
+          "mappings": [
+            {
+              "type": "value",
+              "options": {
+                "1": {
+                  "text": "PASS",
+                  "color": "green"
+                },
+                "0": {
+                  "text": "FAIL",
+                  "color": "red"
+                }
+              }
+            }
+          ]
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "x": 0,
+        "y": 56,
+        "w": 6,
+        "h": 4
+      },
+      "id": 19,
+      "options": {
+        "colorMode": "value",
+        "graphMode": "area",
+        "justifyMode": "auto",
+        "orientation": "auto",
+        "reduceOptions": {
+          "calcs": [
+            "lastNotNull"
+          ],
+          "fields": "",
+          "values": false
+        },
+        "textMode": "auto",
+        "wideLayout": true
+      },
+      "pluginVersion": "11.3.1",
+      "targets": [
+        {
+          "expr": "max(ours_rp_ccr_format_check_last_success and on (exported_instance) topk(1, ours_rp_run_sequence))",
+          "legendFormat": "CCR check",
+          "refId": "A"
+        }
+      ],
+      "title": "Latest CCR Format Check",
+      "type": "stat"
+    },
+    {
+      "datasource": {
+        "type": "prometheus",
+        "uid": "Prometheus"
+      },
+      "fieldConfig": {
+        "defaults": {
+          "decimals": 0,
+          "unit": "none"
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "x": 6,
+        "y": 56,
+        "w": 6,
+        "h": 4
+      },
+      "id": 20,
+      "options": {
+        "colorMode": "value",
+        "graphMode": "area",
+        "justifyMode": "auto",
+        "orientation": "auto",
+        "reduceOptions": {
+          "calcs": [
+            "lastNotNull"
+          ],
+          "fields": "",
+          "values": false
+        },
+        "textMode": "auto",
+        "wideLayout": true
+      },
+      "pluginVersion": "11.3.1",
+      "targets": [
+        {
+          "expr": "max(ours_rp_ccr_format_check_last_run_sequence and on (exported_instance) topk(1, ours_rp_run_sequence))",
+          "legendFormat": "run seq",
+          "refId": "A"
+        }
+      ],
+      "title": "CCR Checked Run Sequence",
+      "type": "stat"
+    },
+    {
+      "datasource": {
+        "type": "prometheus",
+        "uid": "Prometheus"
+      },
+      "fieldConfig": {
+        "defaults": {
+          "decimals": 1,
+          "unit": "s"
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "x": 12,
+        "y": 56,
+        "w": 6,
+        "h": 4
+      },
+      "id": 21,
+      "options": {
+        "colorMode": "value",
+        "graphMode": "area",
+        "justifyMode": "auto",
+        "orientation": "auto",
+        "reduceOptions": {
+          "calcs": [
+            "lastNotNull"
+          ],
+          "fields": "",
+          "values": false
+        },
+        "textMode": "auto",
+        "wideLayout": true
+      },
+      "pluginVersion": "11.3.1",
+      "targets": [
+        {
+          "expr": "max(ours_rp_ccr_format_check_last_duration_seconds and on (exported_instance) topk(1, ours_rp_run_sequence))",
+          "legendFormat": "duration",
+          "refId": "A"
+        }
+      ],
+      "title": "CCR Check Duration",
+      "type": "stat"
+    },
+    {
+      "datasource": {
+        "type": "prometheus",
+        "uid": "Prometheus"
+      },
+      "fieldConfig": {
+        "defaults": {
+          "decimals": 0,
+          "unit": "none",
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "green",
+                "value": null
+              },
+              {
+                "color": "red",
+                "value": 1
+              }
+            ]
+          }
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "x": 18,
+        "y": 56,
+        "w": 6,
+        "h": 4
+      },
+      "id": 22,
+      "options": {
+        "colorMode": "value",
+        "graphMode": "area",
+        "justifyMode": "auto",
+        "orientation": "auto",
+        "reduceOptions": {
+          "calcs": [
+            "lastNotNull"
+          ],
+          "fields": "",
+          "values": false
+        },
+        "textMode": "auto",
+        "wideLayout": true
+      },
+      "pluginVersion": "11.3.1",
+      "targets": [
+        {
+          "expr": "max(ours_rp_ccr_format_check_consecutive_failures and on (exported_instance) topk(1, ours_rp_run_sequence))",
+          "legendFormat": "failures",
+          "refId": "A"
+        }
+      ],
+      "title": "CCR Check Consecutive Failures",
+      "type": "stat"
+    },
+    {
+      "datasource": {
+        "type": "prometheus",
+        "uid": "Prometheus"
+      },
+      "fieldConfig": {
+        "defaults": {
+          "unit": "none",
+          "decimals": 0,
+          "min": 0
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "x": 0,
+        "y": 60,
+        "w": 24,
+        "h": 8
+      },
+      "id": 23,
+      "options": {
+        "legend": {
+          "calcs": [
+            "lastNotNull",
+            "max"
+          ],
+          "displayMode": "table",
+          "placement": "bottom",
+          "showLegend": true
+        },
+        "tooltip": {
+          "mode": "multi",
+          "sort": "none"
+        }
+      },
+      "targets": [
+        {
+          "expr": "sum by (result) (ours_rp_ccr_format_check_total)",
+          "legendFormat": "{{result}} checks",
+          "refId": "A"
+        },
+        {
+          "expr": "max(ours_rp_ccr_format_check_last_success and on (exported_instance) topk(1, ours_rp_run_sequence))",
+          "legendFormat": "latest check ok",
+          "refId": "B"
+        }
+      ],
+      "title": "CCR Format Check Results",
+      "type": "timeseries"
    }
  ],
  "refresh": "5s",
--- a/deploy/docker-installer/docs/README.en.md
+++ b/deploy/docker-installer/docs/README.en.md
@ -80,7 +80,7 @@ Supported values:
 - `rfc6487`: switch back to the original RFC 6487 resource containment behavior.

 The installer / soak runner passes `--resource-validation-mode` to every `rpki` child start.
-When an upgraded deployment reuses an older `.env` that does not contain this key, the installer appends the new default `validation-update-03` automatically.
+When an upgraded deployment reuses an older `.env` that does not contain this key, the new package `.env.example` default `validation-update-03` remains in the final `.env`; an empty final value fails before Docker operations.

 ## Architecture Guardrails

@ -125,6 +125,12 @@ If a separately deployed RTR service on the same host continuously writes `rtr-s

 The command accepts only an absolute path on the same host, validates the candidate Compose mount, and atomically updates `.env`. It recreates only `artifact-metrics`, Prometheus, and Grafana; it does not restart or recreate `ours-rp-soak`. The `artifact-metrics` container mounts the directory read-only and reads it through `RPKI_METRICS_RTR_REPORT_DIR`, producing `ours_rp_rtr_*` metrics. By default, `RTR_REPORT_DIR` points at an empty fallback directory under the installer data root, so the metrics container also starts cleanly without an RTR service.

+## CCR artifact format check metrics
+
+The metrics image ships the official `rpki-client 9.8`. For every newly completed run, `artifact-metrics` runs `rpki-client -f result.ccr` once and exports `ours_rp_ccr_format_check_*` metrics (cumulative pass/ccr_parse_error/tool_error counters, latest check result, check duration, consecutive failures). The Grafana `Ours RP Soak Overview` dashboard shows them in a dedicated panel row. On service start only the latest existing run is checked; older history is not backfilled.
+
+Related environment variables (normally no need to change): `CCR_CHECK_BIN` (defaults to the in-image `/opt/ours-rp/bin/rpki-client`) and `CCR_CHECK_TIMEOUT_SECS` (default 120 seconds).
+
 ## Data Directory

 Default host directory:
--- a/deploy/docker-installer/docs/README.zh-CN.md
+++ b/deploy/docker-installer/docs/README.zh-CN.md
@ -80,7 +80,7 @@ RESOURCE_VALIDATION_MODE=validation-update-03
 - `rfc6487`：切换为 RFC 6487 原始资源包含判定语义。

 installer / soak runner 每次启动 `rpki` 子进程时都会显式传入 `--resource-validation-mode`。
-如果升级时复用旧 `.env` 且缺少该变量，installer 会自动补入默认值 `validation-update-03`，不会破坏旧 `.env` 复用流程。
+如果升级时复用旧 `.env` 且缺少该变量，新包 `.env.example` 中的默认值 `validation-update-03` 会保留在最终 `.env` 中；最终值为空时升级会在 Docker 操作前失败。

 ## 架构检查

@ -125,6 +125,12 @@ GRAFANA_ADMIN_PASSWORD=admin

 该命令只接受本机绝对路径，会先校验候选 Compose 挂载，再原子更新 `.env`。随后只重建 `artifact-metrics`、Prometheus 和 Grafana，不重启或重建 `ours-rp-soak`。`artifact-metrics` 以只读方式挂载该目录，并通过 `RPKI_METRICS_RTR_REPORT_DIR` 读取 report，输出 `ours_rp_rtr_*` 指标。默认 `RTR_REPORT_DIR` 指向安装包数据目录下的空 fallback 目录，因此未接入 RTR 服务时 metrics 容器也能稳定启动。

+## CCR 产物格式检查监控
+
+metrics 镜像内置官方 `rpki-client 9.8`。`artifact-metrics` 会对每个新完成 run 的 `result.ccr` 执行一次 `rpki-client -f` 格式检查，并输出 `ours_rp_ccr_format_check_*` 指标（累计 pass/ccr_parse_error/tool_error、最近检查结果、检查耗时、连续失败数）。Grafana `Ours RP Soak Overview` 底部有对应面板。服务启动时只补查最新的一个历史 run，不回扫全部历史。
+
+相关环境变量（通常无需修改）：`CCR_CHECK_BIN`（默认为镜像内 `/opt/ours-rp/bin/rpki-client`）、`CCR_CHECK_TIMEOUT_SECS`（默认 120 秒）。
+
 ## 数据目录

 默认宿主机目录：
--- a/deploy/docker-installer/docs/operations.en.md
+++ b/deploy/docker-installer/docs/operations.en.md
@ -89,9 +89,26 @@ Extract the new package into a new directory and explicitly reuse the existing `
 ./scripts/upgrade.sh --reuse-env-from /path/to/old-installer/.env
 ```

-If the new package directory already has a `.env`, the upgrade script keeps it.
+Configuration is merged in this order, with a key on the right overriding the same key on the left:

-If an older `.env` does not contain `METRICS_IMAGE` or `METRICS_PLATFORM`, the install/upgrade flow backfills package-matched defaults automatically.
+```text
+new .env.example < old .env < environment variables on this upgrade command
+```
+
+For example, override the next-run interval and Grafana password for this upgrade:
+
+```bash
+INTERVAL_SECS=300 GRAFANA_ADMIN_PASSWORD='new-password' \
+  ./scripts/upgrade.sh --reuse-env-from /path/to/old-installer/.env
+```
+
+Before any image is loaded or container is started, upgrade verifies that every key declared by the new template is present and non-empty in the final `.env`. An explicit empty assignment fails, for example `INTERVAL_SECS= ./scripts/upgrade.sh ...`. Removed keys from an old `.env` are not copied, and the script reports key names only, never password-like values.
+
+The installer reads `.env` as Compose-compatible dotenv assignments. Do not put shell command substitutions or environment references such as `$HOME` in the file; use concrete values or command-level overrides instead.
+
+If the new package directory already has a `.env`, the upgrade script keeps it. In that case `--reuse-env-from` is not imported again, but environment variables from the current command still override the existing `.env`.
+
+When an older `.env` lacks a key introduced by the new package, the new `.env.example` default remains in the final `.env`. Upgrade fails before Docker operations if any final declared value is empty.

 Upgrade does not delete:

--- a/deploy/docker-installer/docs/operations.zh-CN.md
+++ b/deploy/docker-installer/docs/operations.zh-CN.md
@ -89,9 +89,26 @@ curl -s "http://127.0.0.1:${METRICS_PORT:-9556}/metrics" | grep '^ours_rp_rtr_'
 ./scripts/upgrade.sh --reuse-env-from /path/to/old-installer/.env
 ```

-如果新目录已经存在 `.env`，升级脚本会保留它，不覆盖。
+配置按以下优先级合并，右侧同名键覆盖左侧：

-如果旧 `.env` 缺失 `METRICS_IMAGE` 或 `METRICS_PLATFORM`，升级/安装链路会按当前包架构自动补齐默认值。
+```text
+新包 .env.example < 旧 .env < 本次 upgrade 命令的环境变量
+```
+
+例如，临时调整下轮触发间隔和 Grafana 密码：
+
+```bash
+INTERVAL_SECS=300 GRAFANA_ADMIN_PASSWORD='new-password' \
+  ./scripts/upgrade.sh --reuse-env-from /path/to/old-installer/.env
+```
+
+升级会在加载镜像或启动容器前检查：新模板声明的所有变量在最终 `.env` 中都必须存在且非空。显式传入空值会失败，例如 `INTERVAL_SECS= ./scripts/upgrade.sh ...`。旧 `.env` 中已不在新模板内的键不会复制，脚本只输出相关键名而不会输出密码等值。
+
+installer 按与 Compose 一致的 dotenv 赋值格式读取 `.env`。不要在文件中使用 Shell 命令替换或 `$HOME` 一类环境变量引用；请使用确定的值，或在执行 upgrade 时通过命令环境变量覆盖。
+
+如果新目录已经存在 `.env`，升级脚本会保留它，不覆盖；此时 `--reuse-env-from` 不会再次导入旧配置，但本次命令的环境变量仍会覆盖已存在的 `.env`。
+
+旧 `.env` 缺少新版本新增的键时，新包 `.env.example` 的默认值会保留在最终 `.env` 中；任一最终值为空则升级会在 Docker 操作前失败。

 升级不会删除以下数据：

--- a/deploy/docker-installer/scripts/common.sh
+++ b/deploy/docker-installer/scripts/common.sh
@ -6,6 +6,11 @@ ENV_FILE="${ENV_FILE:-$INSTALLER_ROOT/.env}"
 ENV_EXAMPLE="$INSTALLER_ROOT/.env.example"
 COMPOSE_FILE="$INSTALLER_ROOT/compose/docker-compose.yml"
 MANIFEST_FILE="${MANIFEST_FILE:-$INSTALLER_ROOT/PACKAGE-MANIFEST.env}"
+declare -a ENV_TEMPLATE_KEYS=()
+declare -A ENV_TEMPLATE_KEY_SET=()
+declare -A INVOCATION_ENV_OVERRIDES=()
+declare -A REUSED_ENV_KEYS=()
+declare -A ENV_PARSED_VALUES=()

 log() {
  printf '[ours-rp-installer] %s\n' "$*"
@ -48,20 +53,246 @@ env_file_has_key() {
  [[ -f "$env_path" ]] && grep -Eq "^${key}=" "$env_path"
 }

-persist_env_default() {
+load_env_template_keys() {
+  local env_path="$1"
+  local line
+  local key
+
+  [[ -f "$env_path" ]] || die "missing environment template: $env_path"
+  ENV_TEMPLATE_KEYS=()
+  ENV_TEMPLATE_KEY_SET=()
+
+  while IFS= read -r line || [[ -n "$line" ]]; do
+    line="${line%$'\r'}"
+    [[ "$line" =~ ^[[:space:]]*$ || "$line" =~ ^[[:space:]]*# ]] && continue
+    [[ "$line" =~ ^([A-Za-z_][A-Za-z0-9_]*)= ]] || die "invalid environment assignment in $env_path: $line"
+    key="${BASH_REMATCH[1]}"
+    [[ -z "${ENV_TEMPLATE_KEY_SET[$key]+x}" ]] || die "duplicate environment key in $env_path: $key"
+    ENV_TEMPLATE_KEYS+=("$key")
+    ENV_TEMPLATE_KEY_SET["$key"]=1
+  done < "$env_path"
+
+  (( ${#ENV_TEMPLATE_KEYS[@]} > 0 )) || die "no environment assignments found in $env_path"
+}
+
+template_has_env_key() {
+  local key="$1"
+  [[ -n "${ENV_TEMPLATE_KEY_SET[$key]+x}" ]]
+}
+
+env_file_assignment_keys() {
+  local env_path="$1"
+  local line
+
+  [[ -f "$env_path" ]] || die "missing environment file: $env_path"
+  while IFS= read -r line || [[ -n "$line" ]]; do
+    line="${line%$'\r'}"
+    [[ "$line" =~ ^([A-Za-z_][A-Za-z0-9_]*)= ]] || continue
+    printf '%s\n' "${BASH_REMATCH[1]}"
+  done < "$env_path"
+}
+
+dotenv_decode_value() {
+  local raw_value="$1"
+  local quote
+  local body
+  local decoded=""
+  local character
+  local next_character
+  local index=0
+
+  [[ -n "$raw_value" ]] || {
+    printf '\n'
+    return 0
+  }
+
+  quote="${raw_value:0:1}"
+  case "$quote" in
+    "'")
+      [[ "${raw_value: -1}" == "'" && ${#raw_value} -ge 2 ]] || return 1
+      body="${raw_value:1:${#raw_value}-2}"
+      while (( index < ${#body} )); do
+        character="${body:index:1}"
+        if [[ "$character" == "\\" && $((index + 1)) -lt ${#body} ]]; then
+          next_character="${body:index + 1:1}"
+          case "$next_character" in
+            "'")
+              decoded+="$next_character"
+              ((index += 2))
+              continue
+              ;;
+          esac
+        fi
+        decoded+="$character"
+        ((index += 1))
+      done
+      ;;
+    '"')
+      [[ "${raw_value: -1}" == '"' && ${#raw_value} -ge 2 ]] || return 1
+      body="${raw_value:1:${#raw_value}-2}"
+      while (( index < ${#body} )); do
+        character="${body:index:1}"
+        if [[ "$character" == '$' && $((index + 1)) -lt ${#body} && "${body:index + 1:1}" == '$' ]]; then
+          decoded+='$'
+          ((index += 2))
+          continue
+        fi
+        if [[ "$character" == "\\" && $((index + 1)) -lt ${#body} ]]; then
+          next_character="${body:index + 1:1}"
+          case "$next_character" in
+            '"'|\\|'$'|'`')
+              decoded+="$next_character"
+              ((index += 2))
+              continue
+              ;;
+            n)
+              decoded+=$'\n'
+              ((index += 2))
+              continue
+              ;;
+            r)
+              decoded+=$'\r'
+              ((index += 2))
+              continue
+              ;;
+            t)
+              decoded+=$'\t'
+              ((index += 2))
+              continue
+              ;;
+          esac
+        fi
+        decoded+="$character"
+        ((index += 1))
+      done
+      ;;
+    *)
+      decoded="$raw_value"
+      ;;
+  esac
+
+  printf '%s' "$decoded"
+}
+
+parse_env_file() {
+  local env_path="$1"
+  local line
+  local parsed_key
+  local raw_value
+  local parsed_value
+
+  [[ -f "$env_path" ]] || die "missing environment file: $env_path"
+  ENV_PARSED_VALUES=()
+  while IFS= read -r line || [[ -n "$line" ]]; do
+    line="${line%$'\r'}"
+    [[ "$line" =~ ^[[:space:]]*$ || "$line" =~ ^[[:space:]]*# ]] && continue
+    [[ "$line" =~ ^([A-Za-z_][A-Za-z0-9_]*)=(.*)$ ]] || return 2
+    parsed_key="${BASH_REMATCH[1]}"
+    raw_value="${BASH_REMATCH[2]}"
+    parsed_value="$(dotenv_decode_value "$raw_value")" || return 2
+    ENV_PARSED_VALUES["$parsed_key"]="$parsed_value"
+  done < "$env_path"
+}
+
+env_file_value() {
+  local env_path="$1"
+  local key="$2"
+
+  parse_env_file "$env_path"
+
+  [[ -n "${ENV_PARSED_VALUES[$key]+x}" ]] || return 3
+  printf '%s' "${ENV_PARSED_VALUES[$key]}"
+}
+
+env_file_set_value() {
  local env_path="$1"
  local key="$2"
  local value="$3"
  local tmp_env
-  if env_file_has_key "$env_path" "$key"; then
+  local encoded_value
+
+  [[ "$value" != *$'\n'* && "$value" != *$'\r'* ]] || die "environment value for $key contains an unsupported newline"
+  encoded_value="$(env_file_encode_value "$value")"
+  if ! env_file_has_key "$env_path" "$key"; then
+    printf '%s=%s\n' "$key" "$encoded_value" >> "$env_path"
    return 0
  fi
  tmp_env="$(mktemp)"
-  cat "$env_path" > "$tmp_env"
-  printf '%s=%s\n' "$key" "$value" >> "$tmp_env"
+  ENV_FILE_REPLACEMENT="$key=$encoded_value" awk -v key="$key" '
+    BEGIN { done=0 }
+    $0 ~ "^" key "=" { print ENVIRON["ENV_FILE_REPLACEMENT"]; done=1; next }
+    { print }
+    END { if (!done) print ENVIRON["ENV_FILE_REPLACEMENT"] }
+  ' "$env_path" > "$tmp_env"
  mv "$tmp_env" "$env_path"
 }

+env_file_encode_value() {
+  local value="$1"
+  local escaped="$value"
+
+  [[ "$value" != *$'\n'* && "$value" != *$'\r'* ]] || die "environment value contains an unsupported newline"
+  escaped="${escaped//\\/\\\\}"
+  escaped="${escaped//\"/\\\"}"
+  escaped="${escaped//\$/\$\$}"
+  printf '"%s"' "$escaped"
+}
+
+load_env_file() {
+  local env_path="$1"
+  local key
+
+  parse_env_file "$env_path" || die "unable to parse environment file: $env_path"
+  for key in "${!ENV_PARSED_VALUES[@]}"; do
+    printf -v "$key" '%s' "${ENV_PARSED_VALUES[$key]}"
+    export "$key"
+  done
+}
+
+validate_env_file_contract() {
+  local env_path="$1"
+  local invalid_keys=""
+  local key
+
+  [[ -f "$env_path" ]] || die "missing environment file: $env_path"
+  (( ${#ENV_TEMPLATE_KEYS[@]} > 0 )) || die "environment template keys were not loaded"
+
+  if ! parse_env_file "$env_path"; then
+    die "unable to parse environment file: $env_path"
+  fi
+
+  for key in "${ENV_TEMPLATE_KEYS[@]}"; do
+    if [[ -z "${ENV_PARSED_VALUES[$key]+x}" || -z "${ENV_PARSED_VALUES[$key]}" ]]; then
+      invalid_keys+="$key"$'\n'
+    fi
+  done
+
+  [[ -z "$invalid_keys" ]] || die "environment contract requires non-empty values in $env_path: ${invalid_keys//$'\n'/,}"
+}
+
+capture_invocation_env_overrides() {
+  local key
+
+  (( ${#ENV_TEMPLATE_KEYS[@]} > 0 )) || die "environment template keys were not loaded"
+  INVOCATION_ENV_OVERRIDES=()
+  for key in "${ENV_TEMPLATE_KEYS[@]}"; do
+    if [[ -v "$key" ]]; then
+      INVOCATION_ENV_OVERRIDES["$key"]="${!key}"
+    fi
+  done
+}
+
+log_env_key_group() {
+  local group="$1"
+  shift
+  local rendered="none"
+  if (( $# > 0 )); then
+    local IFS=,
+    rendered="$*"
+  fi
+  log "$group=$rendered"
+}
+
 env_flag_enabled() {
  case "${1:-0}" in
    1|true|TRUE|yes|YES|on|ON)
@ -156,15 +387,15 @@ assert_arch_compatibility() {

 load_env() {
  load_manifest
+  [[ -f "$ENV_EXAMPLE" ]] || die "missing $ENV_EXAMPLE"
+  load_env_template_keys "$ENV_EXAMPLE"
+  validate_env_file_contract "$ENV_EXAMPLE"
  if [[ ! -f "$ENV_FILE" ]]; then
-    [[ -f "$ENV_EXAMPLE" ]] || die "missing $ENV_EXAMPLE"
    cp "$ENV_EXAMPLE" "$ENV_FILE"
    log "created .env from .env.example"
  fi
-  set -a
-  # shellcheck disable=SC1090
-  source "$ENV_FILE"
-  set +a
+  validate_env_file_contract "$ENV_FILE"
+  load_env_file "$ENV_FILE"

  if [[ -n "${MANIFEST_PACKAGE_ARCH_RAW:-}" ]]; then
    manifest_package_arch="$(normalize_arch "$MANIFEST_PACKAGE_ARCH_RAW")" || die "unsupported manifest package arch: $MANIFEST_PACKAGE_ARCH_RAW"
@ -190,27 +421,6 @@ load_env() {
    PACKAGE_PLATFORM="${PACKAGE_PLATFORM:-$(platform_for_arch "$PACKAGE_ARCH")}"
  fi

-  HOST_DATA_DIR="${HOST_DATA_DIR:-/var/lib/ours-rp-${PACKAGE_ARCH}-installer}"
-  COMPOSE_PROJECT_NAME="${COMPOSE_PROJECT_NAME:-ours-rp-${PACKAGE_ARCH}-installer}"
-  RPKI_IMAGE="${RPKI_IMAGE:-ours-rp-runtime-${PACKAGE_ARCH}:${source_commit_short:-unknown}}"
-  RPKI_PLATFORM="${RPKI_PLATFORM:-$PACKAGE_PLATFORM}"
-  METRICS_IMAGE="${METRICS_IMAGE:-ours-rp-metrics-${PACKAGE_ARCH}:${source_commit_short:-unknown}}"
-  METRICS_PLATFORM="${METRICS_PLATFORM:-$PACKAGE_PLATFORM}"
-  RTR_REPORT_DIR="${RTR_REPORT_DIR:-$HOST_DATA_DIR/empty-rtr-report}"
-  RTR_REPORT_CONTAINER_DIR="${RTR_REPORT_CONTAINER_DIR:-/var/lib/ours-rp/rtr-report}"
-  MONITOR_PLATFORM="${MONITOR_PLATFORM:-$PACKAGE_PLATFORM}"
-  PROMETHEUS_IMAGE="${PROMETHEUS_IMAGE:-prom/prometheus:v2.55.1}"
-  GRAFANA_IMAGE="${GRAFANA_IMAGE:-grafana/grafana:11.3.1}"
-  METRICS_INSTANCE="${METRICS_INSTANCE:-${PACKAGE_ARCH}-installer}"
-  ALLOW_CROSS_ARCH="${ALLOW_CROSS_ARCH:-0}"
-  RESOURCE_VALIDATION_MODE="${RESOURCE_VALIDATION_MODE:-validation-update-03}"
-  FIRST_RUN_WAIT_TIMEOUT_SECS="${FIRST_RUN_WAIT_TIMEOUT_SECS:-7200}"
-
-  persist_env_default "$ENV_FILE" "METRICS_IMAGE" "$METRICS_IMAGE"
-  persist_env_default "$ENV_FILE" "METRICS_PLATFORM" "$METRICS_PLATFORM"
-  persist_env_default "$ENV_FILE" "RTR_REPORT_DIR" "$RTR_REPORT_DIR"
-  persist_env_default "$ENV_FILE" "RTR_REPORT_CONTAINER_DIR" "$RTR_REPORT_CONTAINER_DIR"
-  persist_env_default "$ENV_FILE" "RESOURCE_VALIDATION_MODE" "$RESOURCE_VALIDATION_MODE"
 }

 compose_cmd() {
@ -546,6 +756,18 @@ verify_metrics_image() {
  log "verifying metrics image $METRICS_IMAGE on $METRICS_PLATFORM"
  verify_image_platform "$METRICS_IMAGE" "$METRICS_PLATFORM" "metrics"
  verify_image_usage_output "$METRICS_IMAGE" "$METRICS_PLATFORM" "metrics" /opt/ours-rp/bin/rpki_artifact_metrics --help
+  verify_metrics_image_rpki_client
+}
+
+verify_metrics_image_rpki_client() {
+  local output
+  local status
+  set +e
+  output="$(docker run --rm --platform "$METRICS_PLATFORM" "$METRICS_IMAGE" /opt/ours-rp/bin/rpki-client -V 2>&1)"
+  status=$?
+  set -e
+  [[ "$status" == "0" ]] || die "metrics image rpki-client -V failed: image=$METRICS_IMAGE exit=$status output=$output"
+  grep -q "rpki-client" <<<"$output" || die "metrics image rpki-client -V unexpected output: $output"
 }

 verify_image_platform() {
--- a/deploy/docker-installer/scripts/upgrade.sh
+++ b/deploy/docker-installer/scripts/upgrade.sh
@ -13,6 +13,8 @@ Usage: ./scripts/upgrade.sh [--reuse-env-from /path/to/.env] [--keep-reused-imag

 By default, --reuse-env-from creates the new .env from this package's
 .env.example first, then overlays existing user settings from the old .env.
+For declared configuration keys, explicit invocation variables have highest
+priority, for example: INTERVAL_SECS=300 ./scripts/upgrade.sh ...
 Image tags are intentionally kept from the new package so the upgraded service
 actually runs the new packaged runtime and monitor images. Use
 --keep-reused-image only when you intentionally want to keep previous image tags.
@ -39,56 +41,80 @@ while [[ $# -gt 0 ]]; do
  esac
 done

-env_get_key() {
-  local env_path="$1"
-  local key="$2"
-  awk -F= -v key="$key" '$1 == key {sub(/^[^=]*=/, ""); print; exit}' "$env_path"
-}
-
-env_set_key() {
-  local env_path="$1"
-  local key="$2"
-  local value="$3"
-  local tmp_env
-  if grep -q "^${key}=" "$env_path"; then
-    tmp_env="$(mktemp)"
-    awk -v key="$key" -v value="$value" '
-      BEGIN { done=0 }
-      $0 ~ "^" key "=" { print key "=" value; done=1; next }
-      { print }
-      END { if (!done) print key "=" value }
-    ' "$env_path" > "$tmp_env"
-    mv "$tmp_env" "$env_path"
-  else
-    printf '%s=%s\n' "$key" "$value" >> "$env_path"
-  fi
-}
-
 overlay_reused_env() {
  local source_env="$1"
  local target_env="$2"
  local key
  local value
-  while IFS='=' read -r key _; do
-    [[ -n "$key" ]] || continue
-    [[ "$key" =~ ^[A-Za-z_][A-Za-z0-9_]*$ ]] || continue
+  local -a reused_keys=()
+  local -a ignored_removed_keys=()
+  local -a preserved_package_keys=()
+
+  REUSED_ENV_KEYS=()
+  parse_env_file "$source_env" || die "unable to parse reused environment: $source_env"
+  while IFS= read -r key; do
+    if ! template_has_env_key "$key"; then
+      ignored_removed_keys+=("$key")
+      continue
+    fi
    case "$key" in
      PACKAGE_ARCH|PACKAGE_PLATFORM|RPKI_PLATFORM|METRICS_PLATFORM|MONITOR_PLATFORM)
+        preserved_package_keys+=("$key")
        continue
        ;;
      RPKI_IMAGE|METRICS_IMAGE|PROMETHEUS_IMAGE|GRAFANA_IMAGE)
-        [[ "$UPDATE_PACKAGE_IMAGE" == "0" ]] || continue
+        if [[ "$UPDATE_PACKAGE_IMAGE" != "0" ]]; then
+          preserved_package_keys+=("$key")
+          continue
+        fi
        ;;
    esac
-    value="$(env_get_key "$source_env" "$key")"
-    env_set_key "$target_env" "$key" "$value"
-  done < <(grep -E '^[A-Za-z_][A-Za-z0-9_]*=' "$source_env" || true)
+    [[ -n "${ENV_PARSED_VALUES[$key]+x}" ]] || die "unable to read $key from reused environment: $source_env"
+    value="${ENV_PARSED_VALUES[$key]}"
+    env_file_set_value "$target_env" "$key" "$value"
+    REUSED_ENV_KEYS["$key"]=1
+    reused_keys+=("$key")
+  done < <(env_file_assignment_keys "$source_env")
+
+  log_env_key_group "reused_keys" "${reused_keys[@]}"
+  log_env_key_group "ignored_removed_old_keys" "${ignored_removed_keys[@]}"
+  log_env_key_group "preserved_package_keys" "${preserved_package_keys[@]}"
 }

+apply_invocation_env_overrides() {
+  local target_env="$1"
+  local key
+  local -a overridden_keys=()
+
+  for key in "${ENV_TEMPLATE_KEYS[@]}"; do
+    [[ -n "${INVOCATION_ENV_OVERRIDES[$key]+x}" ]] || continue
+    env_file_set_value "$target_env" "$key" "${INVOCATION_ENV_OVERRIDES[$key]}"
+    overridden_keys+=("$key")
+  done
+
+  log_env_key_group "command_override_keys" "${overridden_keys[@]}"
+}
+
+log_template_only_keys() {
+  local key
+  local -a template_only_keys=()
+
+  for key in "${ENV_TEMPLATE_KEYS[@]}"; do
+    [[ -n "${REUSED_ENV_KEYS[$key]+x}" ]] && continue
+    [[ -n "${INVOCATION_ENV_OVERRIDES[$key]+x}" ]] && continue
+    template_only_keys+=("$key")
+  done
+  log_env_key_group "template_only_keys" "${template_only_keys[@]}"
+}
+
+load_env_template_keys "$ENV_EXAMPLE"
+validate_env_file_contract "$ENV_EXAMPLE"
+capture_invocation_env_overrides
+REUSED_ENV_KEYS=()
+
 if [[ -n "$REUSE_ENV_FROM" ]]; then
  [[ -f "$REUSE_ENV_FROM" ]] || die "missing reuse env file: $REUSE_ENV_FROM"
  if [[ ! -f "$ENV_FILE" ]]; then
-    [[ -f "$ENV_EXAMPLE" ]] || die "missing $ENV_EXAMPLE"
    cp "$ENV_EXAMPLE" "$ENV_FILE"
    overlay_reused_env "$REUSE_ENV_FROM" "$ENV_FILE"
    log "created new package env from .env.example and overlaid user settings from $REUSE_ENV_FROM"
@ -97,6 +123,13 @@ if [[ -n "$REUSE_ENV_FROM" ]]; then
  fi
 fi

+if [[ ! -f "$ENV_FILE" ]]; then
+  cp "$ENV_EXAMPLE" "$ENV_FILE"
+  log "created new package env from .env.example"
+fi
+apply_invocation_env_overrides "$ENV_FILE"
+log_template_only_keys
+validate_env_file_contract "$ENV_FILE"
 load_env
 assert_arch_compatibility
 create_data_dirs
--- a/deploy/docker-installer/tests/upgrade_env_contract_test.sh
+++ b/deploy/docker-installer/tests/upgrade_env_contract_test.sh
@ -0,0 +1,206 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+INSTALLER_SOURCE="$(cd "$SCRIPT_DIR/.." && pwd)"
+TEST_ROOT="$(mktemp -d "${TMPDIR:-/tmp}/ours-rp-installer-env-contract.XXXXXX")"
+
+cleanup() {
+  rm -rf "$TEST_ROOT"
+}
+trap cleanup EXIT
+
+fail() {
+  printf 'FAIL: %s\n' "$*" >&2
+  exit 1
+}
+
+assert_equals() {
+  local expected="$1"
+  local actual="$2"
+  local message="$3"
+  [[ "$actual" == "$expected" ]] || fail "$message (expected=$expected actual=$actual)"
+}
+
+assert_not_contains() {
+  local needle="$1"
+  local path="$2"
+  ! grep -Fq "$needle" "$path" || fail "unexpected value in $path: $needle"
+}
+
+expect_exit() {
+  local expected_status="$1"
+  local output_path="$2"
+  shift 2
+  local actual_status
+
+  set +e
+  "$@" >"$output_path" 2>&1
+  actual_status=$?
+  set -e
+  [[ "$actual_status" == "$expected_status" ]] || {
+    cat "$output_path" >&2 || true
+    fail "unexpected exit status (expected=$expected_status actual=$actual_status)"
+  }
+}
+
+host_arch() {
+  case "$(uname -m)" in
+    x86_64) printf 'amd64\n' ;;
+    aarch64) printf 'arm64\n' ;;
+    *) fail "unsupported test host architecture: $(uname -m)" ;;
+  esac
+}
+
+env_value() {
+  local env_path="$1"
+  local key="$2"
+  (
+    # shellcheck disable=SC1090
+    source "$INSTALLER_SOURCE/scripts/common.sh"
+    env_file_value "$env_path" "$key"
+  )
+}
+
+create_fixture() {
+  local name="$1"
+  local fixture="$TEST_ROOT/$name"
+  local arch
+  arch="$(host_arch)"
+
+  mkdir -p "$fixture"
+  cp -a "$INSTALLER_SOURCE"/. "$fixture"/
+  rm -rf "$fixture/tests"
+  mkdir -p "$fixture/images" "$fixture/fake-bin"
+  sed -i \
+    -e "s|__PACKAGE_ARCH__|$arch|g" \
+    -e "s|__PACKAGE_PLATFORM__|linux/$arch|g" \
+    -e "s|__RUNTIME_IMAGE__|ours-rp-runtime-$arch:new|g" \
+    -e "s|__METRICS_IMAGE__|ours-rp-metrics-$arch:new|g" \
+    -e "s|__HOST_DATA_DIR__|$fixture/data|g" \
+    "$fixture/.env.example"
+  cat > "$fixture/PACKAGE-MANIFEST.env" <<EOF
+PACKAGE_ARCH=$arch
+PACKAGE_PLATFORM=linux/$arch
+EOF
+  cat > "$fixture/fake-bin/docker" <<'EOF'
+#!/usr/bin/env bash
+printf '%s\n' "$*" >> "$FAKE_DOCKER_LOG"
+if [[ "${1:-}" == "compose" && "${2:-}" == "version" ]]; then
+  exit 0
+fi
+exit 99
+EOF
+  chmod +x "$fixture/fake-bin/docker"
+  for command in jq rsync curl gzip tar; do
+    cat > "$fixture/fake-bin/$command" <<'EOF'
+#!/usr/bin/env bash
+exit 0
+EOF
+    chmod +x "$fixture/fake-bin/$command"
+  done
+  printf '%s\n' "$fixture"
+}
+
+run_upgrade() {
+  local fixture="$1"
+  local old_env="$2"
+  shift 2
+  (
+    cd "$fixture"
+    export PATH="$fixture/fake-bin:$PATH"
+    export FAKE_DOCKER_LOG="$fixture/fake-docker.log"
+    "$@" ./scripts/upgrade.sh --reuse-env-from "$old_env"
+  )
+}
+
+run_upgrade_without_reuse() {
+  local fixture="$1"
+  (
+    cd "$fixture"
+    export PATH="$fixture/fake-bin:$PATH"
+    export FAKE_DOCKER_LOG="$fixture/fake-docker.log"
+    ./scripts/upgrade.sh
+  )
+}
+
+new_fixture_with_old_env() {
+  local name="$1"
+  local fixture
+  fixture="$(create_fixture "$name")"
+  cat > "$fixture/old.env" <<'EOF'
+INTERVAL_SECS=120
+RPKI_IMAGE=ours-rp-runtime-old:old
+REMOVED_LEGACY_KEY=legacy-value
+EOF
+  printf '%s\n' "$fixture"
+}
+
+fixture="$(create_fixture template-defaults)"
+expect_exit 99 "$fixture/template.log" run_upgrade_without_reuse "$fixture"
+assert_equals "600" "$(env_value "$fixture/.env" INTERVAL_SECS)" "template default must be retained without reuse"
+
+fixture="$(new_fixture_with_old_env default-old)"
+expect_exit 99 "$fixture/default.log" run_upgrade "$fixture" "$fixture/old.env"
+assert_equals "120" "$(env_value "$fixture/.env" INTERVAL_SECS)" "old value must override template default"
+assert_equals "ours-rp-runtime-$(host_arch):new" "$(env_value "$fixture/.env" RPKI_IMAGE)" "standard upgrade must retain package image"
+assert_equals "10" "$(env_value "$fixture/.env" METRICS_POLL_SECS)" "template default must fill a key absent from old environment"
+! grep -q '^REMOVED_LEGACY_KEY=' "$fixture/.env" || fail "removed old key was copied into target environment"
+grep -q 'reused_keys=INTERVAL_SECS' "$fixture/default.log" || fail "reused key summary missing"
+grep -q 'ignored_removed_old_keys=REMOVED_LEGACY_KEY' "$fixture/default.log" || fail "removed key summary missing"
+grep -q 'template_only_keys=' "$fixture/default.log" || fail "template key summary missing"
+
+fixture="$(new_fixture_with_old_env invocation-overrides)"
+expect_exit 99 "$fixture/override.log" run_upgrade "$fixture" "$fixture/old.env" env INTERVAL_SECS=300 GRAFANA_ADMIN_PASSWORD='secret with space'
+assert_equals "300" "$(env_value "$fixture/.env" INTERVAL_SECS)" "invocation value must override old value"
+assert_equals "secret with space" "$(env_value "$fixture/.env" GRAFANA_ADMIN_PASSWORD)" "invocation value must preserve shell-sensitive content"
+grep -q 'command_override_keys=INTERVAL_SECS,GRAFANA_ADMIN_PASSWORD' "$fixture/override.log" || fail "override key summary missing"
+assert_not_contains 'secret with space' "$fixture/override.log"
+
+fixture="$(new_fixture_with_old_env invocation-special-characters)"
+special_password="O'Reilly"
+special_password+=' $literal `tick` slash\'
+expect_exit 99 "$fixture/special.log" run_upgrade "$fixture" "$fixture/old.env" env GRAFANA_ADMIN_PASSWORD="$special_password"
+assert_equals "$special_password" "$(env_value "$fixture/.env" GRAFANA_ADMIN_PASSWORD)" "invocation value with dotenv-sensitive characters must round-trip"
+assert_not_contains "$special_password" "$fixture/special.log"
+
+fixture="$(create_fixture existing-target-env)"
+cp "$fixture/.env.example" "$fixture/.env"
+sed -i 's/^INTERVAL_SECS=.*/INTERVAL_SECS=180/' "$fixture/.env"
+cat > "$fixture/old.env" <<'EOF'
+INTERVAL_SECS=120
+EOF
+expect_exit 99 "$fixture/existing.log" run_upgrade "$fixture" "$fixture/old.env" env INTERVAL_SECS=300
+assert_equals "300" "$(env_value "$fixture/.env" INTERVAL_SECS)" "invocation value must override an existing target environment"
+
+fixture="$(new_fixture_with_old_env explicit-empty)"
+expect_exit 1 "$fixture/empty.log" run_upgrade "$fixture" "$fixture/old.env" env INTERVAL_SECS=
+[[ ! -s "$fixture/fake-docker.log" ]] || fail "explicit empty value reached Docker"
+grep -q 'INTERVAL_SECS' "$fixture/empty.log" || fail "empty-value error did not identify key"
+
+fixture="$(create_fixture empty-old-value)"
+cat > "$fixture/old.env" <<'EOF'
+INTERVAL_SECS=
+EOF
+expect_exit 1 "$fixture/old-empty.log" run_upgrade "$fixture" "$fixture/old.env"
+[[ ! -s "$fixture/fake-docker.log" ]] || fail "empty reused value reached Docker"
+grep -q 'INTERVAL_SECS' "$fixture/old-empty.log" || fail "empty reused value error did not identify key"
+
+fixture="$(create_fixture empty-template)"
+sed -i 's/^INTERVAL_SECS=.*/INTERVAL_SECS=/' "$fixture/.env.example"
+cat > "$fixture/old.env" <<'EOF'
+INTERVAL_SECS=120
+EOF
+expect_exit 1 "$fixture/template-empty.log" run_upgrade "$fixture" "$fixture/old.env"
+[[ ! -s "$fixture/fake-docker.log" ]] || fail "empty template default reached Docker"
+grep -q 'INTERVAL_SECS' "$fixture/template-empty.log" || fail "empty template error did not identify key"
+
+fixture="$(new_fixture_with_old_env package-guard)"
+arch="$(host_arch)"
+other_arch="amd64"
+[[ "$arch" == "amd64" ]] && other_arch="arm64"
+expect_exit 1 "$fixture/arch.log" run_upgrade "$fixture" "$fixture/old.env" env PACKAGE_ARCH="$other_arch"
+[[ ! -s "$fixture/fake-docker.log" ]] || fail "package architecture mismatch reached Docker"
+grep -q 'mismatches manifest PACKAGE_ARCH' "$fixture/arch.log" || fail "package architecture guard did not reject override"
+
+printf 'PASS: Docker installer upgrade environment contract\n'
--- a/docker/ours-rp-metrics.Dockerfile
+++ b/docker/ours-rp-metrics.Dockerfile
@ -85,27 +85,66 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \
    mkdir -p /build-out/bin; \
    cp "target/$target_triple/release/rpki_artifact_metrics" /build-out/bin/

+FROM --platform=$TARGETPLATFORM ${RUNTIME_IMAGE} AS rpki-client-builder
+
+ARG RPKI_CLIENT_VERSION=9.8
+ARG RPKI_CLIENT_SHA256=42920aac5afd0996173fb9f7848691a2c49dd234e4f10478808c1aa475620861
+
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends \
+      build-essential \
+      ca-certificates \
+      curl \
+      libexpat1-dev \
+      libssl-dev \
+      libtls-dev \
+      pkg-config \
+      rsync \
+      zlib1g-dev \
+    && rm -rf /var/lib/apt/lists/* \
+    && useradd --system --no-create-home --shell /usr/sbin/nologin _rpki-client
+
+RUN set -eux; \
+    curl -fsSL "https://ftp.openbsd.org/pub/OpenBSD/rpki-client/rpki-client-${RPKI_CLIENT_VERSION}.tar.gz" -o /tmp/rpki-client.tar.gz; \
+    echo "${RPKI_CLIENT_SHA256}  /tmp/rpki-client.tar.gz" | sha256sum -c -; \
+    tar -xzf /tmp/rpki-client.tar.gz -C /tmp; \
+    cd "/tmp/rpki-client-${RPKI_CLIENT_VERSION}"; \
+    ./configure --prefix=/opt/rpki-client; \
+    make -j"$(nproc)"; \
+    make install; \
+    /opt/rpki-client/sbin/rpki-client -V
+
 FROM --platform=$TARGETPLATFORM ${RUNTIME_IMAGE} AS runtime

 RUN apt-get update \
    && apt-get install -y --no-install-recommends \
      ca-certificates \
+      libexpat1 \
+      libssl3 \
+      libtls26 \
      tzdata \
-    && rm -rf /var/lib/apt/lists/*
+      zlib1g \
+    && rm -rf /var/lib/apt/lists/* \
+    && useradd --system --no-create-home --shell /usr/sbin/nologin _rpki-client \
+    && mkdir -p /opt/rpki-client/var/cache/rpki-client

 WORKDIR /opt/ours-rp

 COPY --from=builder /build-out/bin/ /opt/ours-rp/bin/
+COPY --from=rpki-client-builder /opt/rpki-client/ /opt/rpki-client/
+COPY --from=rpki-client-builder /opt/rpki-client/sbin/rpki-client /opt/ours-rp/bin/rpki-client

 ARG SOURCE_COMMIT=unknown
 ARG SOURCE_DIRTY=unknown
 ARG BUILD_TIMESTAMP_UTC=unknown
+ARG RPKI_CLIENT_VERSION=9.8

 LABEL org.opencontainers.image.title="ours-rp-metrics" \
      org.opencontainers.image.description="Ours RP artifact metrics image for multi-arch Docker Compose deployment" \
      org.opencontainers.image.revision="${SOURCE_COMMIT}" \
      org.opencontainers.image.created="${BUILD_TIMESTAMP_UTC}" \
-      org.opencontainers.image.source-dirty="${SOURCE_DIRTY}"
+      org.opencontainers.image.source-dirty="${SOURCE_DIRTY}" \
+      org.opencontainers.image.rpki-client.version="${RPKI_CLIENT_VERSION}"

 RUN chmod +x /opt/ours-rp/bin/* \
    && mkdir -p /var/lib/ours-rp/state /var/lib/ours-rp/runs /var/lib/ours-rp/logs
--- a/scripts/benchmark/run_sync_worker_ablation.py
+++ b/scripts/benchmark/run_sync_worker_ablation.py
@ -0,0 +1,667 @@
+#!/usr/bin/env python3
+"""Run ours RP sync-worker-count ablation on a remote host.
+
+The experiment intentionally mirrors the Feature #100 ours-pp-object-cache
+profile and changes only --parallel-max-repo-sync-workers-global.
+"""
+
+from __future__ import annotations
+
+import argparse
+import csv
+import hashlib
+import html
+import json
+import os
+import re
+import shlex
+import subprocess
+import sys
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+
+PRODUCT_ROOT = Path(__file__).resolve().parents[2]
+DEFAULT_REMOTE = "root@47.77.204.233"
+DEFAULT_WORKERS = "1,2,4,8"
+DEFAULT_INTERVAL_SECS = 0
+DEFAULT_RUNS_PER_WORKER = 5
+
+RIRS = ["afrinic", "apnic", "arin", "lacnic", "ripe"]
+TAL_URLS = {
+    "afrinic": "https://rpki.afrinic.net/tal/afrinic.tal",
+    "apnic": "https://tal.apnic.net/apnic.tal",
+    "arin": "https://www.arin.net/resources/manage/rpki/arin.tal",
+    "lacnic": "https://www.lacnic.net/innovaportal/file/4983/1/lacnic.tal",
+    "ripe": "https://tal.rpki.ripe.net/ripe-ncc.tal",
+}
+
+
+REMOTE_RUNNER = r"""#!/usr/bin/env bash
+set -euo pipefail
+mode="$1"
+state_dir="$2"
+run_dir="$3"
+shift 3
+mkdir -p "$run_dir"
+if [[ "$mode" == "snapshot" ]]; then
+  rm -rf "$state_dir"
+fi
+mkdir -p "$state_dir" "$(dirname "$run_dir")"
+date -u +"%Y-%m-%dT%H:%M:%SZ" > "$run_dir/start-time.txt"
+printf '%q ' "$@" > "$run_dir/command.txt"
+printf '\n' >> "$run_dir/command.txt"
+set +e
+/usr/bin/time -v -o "$run_dir/process-time.txt" "$@" > "$run_dir/stdout.log" 2> "$run_dir/stderr.log"
+status=$?
+set -e
+date -u +"%Y-%m-%dT%H:%M:%SZ" > "$run_dir/end-time.txt"
+printf '%s\n' "$status" > "$run_dir/exit-code.txt"
+exit "$status"
+"""
+
+
+REMOTE_METRICS = r"""import csv
+import json
+import os
+import re
+import subprocess
+import sys
+from pathlib import Path
+
+run_dir = Path(os.environ["RUN_DIR"])
+state_dir = Path(os.environ["STATE_DIR"])
+workers = int(os.environ["WORKERS"])
+mode = os.environ["MODE"]
+run_index = int(os.environ["RUN_INDEX"])
+scheduled_epoch = float(os.environ["SCHEDULED_EPOCH"])
+started_epoch = float(os.environ["STARTED_EPOCH"])
+
+
+def read_text(path):
+    try:
+        return Path(path).read_text(errors="replace")
+    except FileNotFoundError:
+        return ""
+
+
+def read_json(path):
+    try:
+        with Path(path).open() as fh:
+            return json.load(fh)
+    except FileNotFoundError:
+        return None
+    except json.JSONDecodeError as exc:
+        return {"_json_error": str(exc)}
+
+
+def parse_time(path):
+    text = read_text(path)
+    out = {}
+    match = re.search(r"Elapsed \(wall clock\) time.*: (?:(\d+):)?(\d+):(\d+(?:\.\d+)?)", text)
+    if match:
+        hours = int(match.group(1) or 0)
+        minutes = int(match.group(2))
+        seconds = float(match.group(3))
+        out["wall_ms_timev"] = int(round(((hours * 60 + minutes) * 60 + seconds) * 1000))
+    for name, key in [
+        ("Maximum resident set size \\(kbytes\\)", "max_rss_kb"),
+        ("Percent of CPU this job got", "cpu_percent_text"),
+        ("Major \\(requiring I/O\\) page faults", "major_faults"),
+        ("Minor \\(reclaiming a frame\\) page faults", "minor_faults"),
+    ]:
+        found = re.search(name + r":\s+(.+)", text)
+        if found:
+            value = found.group(1).strip()
+            if key.endswith("_faults") or key == "max_rss_kb":
+                try:
+                    out[key] = int(value)
+                except ValueError:
+                    out[key] = value
+            else:
+                out[key] = value
+    return out
+
+
+def count_csv(path):
+    rows = 0
+    unique = set()
+    try:
+        with Path(path).open(newline="") as fh:
+            reader = csv.reader(fh)
+            header = next(reader, None)
+            for row in reader:
+                if not row:
+                    continue
+                rows += 1
+                unique.add(tuple(row))
+    except FileNotFoundError:
+        return {"rows": None, "unique_rows": None}
+    return {"rows": rows, "unique_rows": len(unique)}
+
+
+def dir_bytes(path):
+    path = Path(path)
+    if not path.exists():
+        return 0
+    total = 0
+    for item in path.rglob("*"):
+        try:
+            if item.is_file():
+                total += item.stat().st_size
+        except OSError:
+            pass
+    return total
+
+
+def file_bytes(path):
+    try:
+        return Path(path).stat().st_size
+    except FileNotFoundError:
+        return None
+
+
+stage = read_json(run_dir / "stage-timing.json") or {}
+process_time = parse_time(run_dir / "process-time.txt")
+exit_text = read_text(run_dir / "exit-code.txt").strip()
+start_text = read_text(run_dir / "start-time.txt").strip()
+end_text = read_text(run_dir / "end-time.txt").strip()
+
+df = subprocess.run(["df", "-P", "/"], text=True, capture_output=True)
+df_line = df.stdout.strip().splitlines()[-1].split() if df.returncode == 0 and len(df.stdout.strip().splitlines()) >= 2 else []
+df_root = {}
+if len(df_line) >= 5:
+    df_root = {
+        "filesystem": df_line[0],
+        "blocks_1k": int(df_line[1]),
+        "used_1k": int(df_line[2]),
+        "available_1k": int(df_line[3]),
+        "use_percent": int(df_line[4].rstrip("%")),
+    }
+
+metrics = {
+    "workers": workers,
+    "mode": mode,
+    "run_index": run_index,
+    "scheduled_epoch": scheduled_epoch,
+    "started_epoch": started_epoch,
+    "schedule_lag_ms": int(round((started_epoch - scheduled_epoch) * 1000)),
+    "start_time_utc": start_text,
+    "end_time_utc": end_text,
+    "exit_code": int(exit_text) if exit_text.isdigit() else None,
+    "run_dir": str(run_dir),
+    "state_dir": str(state_dir),
+    "stage": stage,
+    "process_time": process_time,
+    "vrps": count_csv(run_dir / "vrps.csv"),
+    "vaps": count_csv(run_dir / "vaps.csv"),
+    "artifact_bytes": {
+        "report_json": file_bytes(run_dir / "report.json"),
+        "result_ccr": file_bytes(run_dir / "result.ccr"),
+        "input_cir": file_bytes(run_dir / "input.cir"),
+        "vrps_csv": file_bytes(run_dir / "vrps.csv"),
+        "vaps_csv": file_bytes(run_dir / "vaps.csv"),
+    },
+    "state_bytes": dir_bytes(state_dir),
+    "run_bytes": dir_bytes(run_dir),
+    "df_root": df_root,
+}
+print(json.dumps(metrics, ensure_ascii=False, sort_keys=True))
+"""
+
+
+def utc_stamp() -> str:
+    return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+
+
+def run_cmd(cmd: list[str], *, cwd: Path | None = None, check: bool = True, input_text: str | None = None) -> subprocess.CompletedProcess[str]:
+    result = subprocess.run(cmd, cwd=cwd, input=input_text, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    if check and result.returncode != 0:
+        sys.stderr.write(result.stdout)
+        sys.stderr.write(result.stderr)
+        raise SystemExit(result.returncode)
+    return result
+
+
+def ssh(remote: str, command: str, *, check: bool = True, input_text: str | None = None) -> subprocess.CompletedProcess[str]:
+    return run_cmd(["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10", remote, command], check=check, input_text=input_text)
+
+
+def scp_to(remote: str, local: Path, remote_path: str) -> None:
+    ssh(remote, "mkdir -p " + shlex.quote(str(Path(remote_path).parent)))
+    run_cmd(["scp", "-q", str(local), f"{remote}:{remote_path}"])
+
+
+def parse_workers(value: str) -> list[int]:
+    workers = []
+    for part in value.split(","):
+        part = part.strip()
+        if not part:
+            continue
+        worker = int(part)
+        if worker <= 0:
+            raise argparse.ArgumentTypeError("worker count must be positive")
+        workers.append(worker)
+    if len(set(workers)) != len(workers):
+        raise argparse.ArgumentTypeError("worker counts must be unique")
+    return workers
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Run ours RP sync worker ablation.")
+    parser.add_argument("--remote", default=DEFAULT_REMOTE)
+    parser.add_argument("--workers", type=parse_workers, default=parse_workers(DEFAULT_WORKERS))
+    parser.add_argument("--interval-secs", type=int, default=DEFAULT_INTERVAL_SECS)
+    parser.add_argument("--runs-per-worker", type=int, default=DEFAULT_RUNS_PER_WORKER)
+    parser.add_argument("--remote-root")
+    parser.add_argument("--out-root", type=Path, default=Path("specs/develop/20260714/feature115_sync_worker_ablation_runs"))
+    parser.add_argument("--binary", type=Path)
+    parser.add_argument("--skip-build", action="store_true")
+    parser.add_argument("--dry-run", action="store_true")
+    return parser.parse_args()
+
+
+def ensure_binary(args: argparse.Namespace) -> Path:
+    if args.binary:
+        binary = args.binary.resolve()
+        if not binary.exists():
+            raise SystemExit(f"binary does not exist: {binary}")
+        return binary
+    binary = PRODUCT_ROOT / "target" / "release" / "rpki"
+    if not args.skip_build:
+        print("building release rpki binary", flush=True)
+        run_cmd(["cargo", "build", "--release", "--bin", "rpki"], cwd=PRODUCT_ROOT)
+    if not binary.exists():
+        raise SystemExit(f"release binary not found: {binary}")
+    return binary
+
+
+def fixed_args(remote_bin: str, state_dir: str, run_dir: str, workers: int) -> list[str]:
+    args = [
+        remote_bin,
+        "--db", f"{state_dir}/work-db",
+        "--repo-bytes-db", f"{state_dir}/repo-bytes.db",
+        "--rsync-mirror-root", f"{state_dir}/rsync-mirror",
+        "--rsync-scope", "module-root",
+        "--report-json", f"{run_dir}/report.json",
+        "--report-json-compact",
+        "--ccr-out", f"{run_dir}/result.ccr",
+        "--cir-enable",
+        "--cir-out", f"{run_dir}/input.cir",
+        "--vrps-csv-out", f"{run_dir}/vrps.csv",
+        "--vaps-csv-out", f"{run_dir}/vaps.csv",
+        "--compare-view-trust-anchor", "all5",
+        "--parallel-phase2-ready-batch-size", "256",
+        "--parallel-phase2-ready-batch-wall-time-budget-ms", "100",
+        "--parallel-phase2-result-drain-batch-size", "2048",
+        "--parallel-phase2-finalize-batch-size", "256",
+        "--parallel-phase2-finalize-batch-wall-time-budget-ms", "100",
+        "--parallel-max-repo-sync-workers-global", str(workers),
+        "--enable-publication-point-validation-cache",
+        "--enable-roa-validation-cache",
+        "--enable-child-certificate-validation-cache",
+    ]
+    for rir in RIRS:
+        url = TAL_URLS[rir]
+        args.extend(["--tal-url", url, "--cir-tal-uri", url])
+    return args
+
+
+def plan_runs(workers: list[int], runs_per_worker: int, interval_secs: int, remote_root: str, remote_bin: str) -> list[dict[str, Any]]:
+    if runs_per_worker < 2:
+        raise SystemExit("--runs-per-worker must be at least 2")
+    planned = []
+    ordinal = 0
+    for worker in workers:
+        case_root = f"{remote_root}/cases/workers-{worker:02d}"
+        state_dir = f"{case_root}/state"
+        for run_index in range(1, runs_per_worker + 1):
+            mode = "snapshot" if run_index == 1 else "delta"
+            run_dir = f"{case_root}/runs/run_{run_index:04d}_{mode}"
+            planned.append({
+                "ordinal": ordinal,
+                "workers": worker,
+                "mode": mode,
+                "run_index": run_index,
+                "state_dir": state_dir,
+                "run_dir": run_dir,
+                "interval_secs": interval_secs,
+                "command": fixed_args(remote_bin, state_dir, run_dir, worker),
+            })
+            ordinal += 1
+    return planned
+
+
+def write_json(path: Path, data: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(data, indent=2, ensure_ascii=False, sort_keys=True) + "\n")
+
+
+def write_text(path: Path, text: str) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(text)
+
+
+def prepare_remote(remote: str, remote_root: str, binary: Path) -> str:
+    remote_bin = f"{remote_root}/bin/rpki"
+    preflight = f"""
+set -euo pipefail
+mkdir -p {shlex.quote(remote_root)}/bin {shlex.quote(remote_root)}/env
+systemctl disable --now rpki-client.timer >/dev/null 2>&1 || true
+systemctl stop rpki-client.service >/dev/null 2>&1 || true
+pkill -x rpki-client >/dev/null 2>&1 || true
+pkill -x routinator >/dev/null 2>&1 || true
+pkill -x fort >/dev/null 2>&1 || true
+pkill -x rpki >/dev/null 2>&1 || true
+{{
+  date -u +"%Y-%m-%dT%H:%M:%SZ"
+  hostname
+  uname -a
+  nproc
+  df -h /
+  free -h || true
+  ps -eo pid,ppid,comm,%cpu,%mem,args --sort=-%cpu | head -30
+  systemctl is-active rpki-client.timer 2>/dev/null || true
+}} > {shlex.quote(remote_root)}/env/preflight.txt
+"""
+    ssh(remote, "bash -s", input_text=preflight)
+    scp_to(remote, binary, remote_bin)
+    ssh(remote, "chmod +x " + shlex.quote(remote_bin))
+    return remote_bin
+
+
+def run_remote_plan(remote: str, plan: dict[str, Any], scheduled_epoch: float) -> dict[str, Any]:
+    now = time.time()
+    if now < scheduled_epoch:
+        wait_secs = int(round(scheduled_epoch - now))
+        print(f"waiting {wait_secs}s before workers={plan['workers']} {plan['mode']}", flush=True)
+        time.sleep(scheduled_epoch - now)
+    started_epoch = time.time()
+    command = shlex.join(["bash", "-s", "--", plan["mode"], plan["state_dir"], plan["run_dir"], *plan["command"]])
+    print(f"start workers={plan['workers']} {plan['mode']} run={plan['run_index']} at {datetime.now(timezone.utc).isoformat()}", flush=True)
+    result = ssh(remote, command, input_text=REMOTE_RUNNER, check=False)
+    if result.returncode != 0:
+        sys.stderr.write(result.stdout)
+        sys.stderr.write(result.stderr)
+    metrics_script = "\n".join([
+        f"export RUN_DIR={shlex.quote(plan['run_dir'])}",
+        f"export STATE_DIR={shlex.quote(plan['state_dir'])}",
+        f"export WORKERS={plan['workers']}",
+        f"export MODE={shlex.quote(plan['mode'])}",
+        f"export RUN_INDEX={plan['run_index']}",
+        f"export SCHEDULED_EPOCH={scheduled_epoch}",
+        f"export STARTED_EPOCH={started_epoch}",
+        "python3 - <<'PY'",
+        REMOTE_METRICS,
+        "PY",
+    ])
+    metrics_result = ssh(remote, "bash -s", input_text=metrics_script, check=True)
+    metrics = json.loads(metrics_result.stdout)
+    metrics["ssh_return_code"] = result.returncode
+    metrics["command_digest"] = hashlib.sha256("\0".join(plan["command"]).encode()).hexdigest()
+    ssh(
+        remote,
+        "cat > " + shlex.quote(plan["run_dir"] + "/metrics.json"),
+        input_text=json.dumps(metrics, ensure_ascii=False, sort_keys=True) + "\n",
+    )
+    if result.returncode != 0:
+        raise RuntimeError(f"remote run failed: workers={plan['workers']} mode={plan['mode']} rc={result.returncode}")
+    validate_metrics(metrics)
+    print(f"done workers={plan['workers']} {plan['mode']} wall={metrics.get('wall_ms')}ms vrps={metrics['vrps'].get('rows')} vaps={metrics['vaps'].get('rows')}", flush=True)
+    return metrics
+
+
+def validate_metrics(metrics: dict[str, Any]) -> None:
+    stage = metrics.get("stage") or {}
+    expected = {
+        "enable_transport_request_prefetch": False,
+        "enable_publication_point_validation_cache": True,
+        "enable_roa_validation_cache": True,
+        "enable_child_certificate_validation_cache": True,
+    }
+    for key, value in expected.items():
+        if stage.get(key) is not value:
+            raise RuntimeError(f"unexpected {key}: {stage.get(key)!r}, expected {value!r}")
+    if metrics["vrps"].get("rows") in (None, 0):
+        raise RuntimeError("VRP CSV is missing or empty")
+    if metrics["vaps"].get("rows") is None:
+        raise RuntimeError("VAP CSV is missing")
+    df_root = metrics.get("df_root") or {}
+    if df_root.get("use_percent", 0) >= 90:
+        raise RuntimeError(f"remote disk is above threshold: {df_root}")
+    stage_total = stage.get("total_ms")
+    timev_total = metrics.get("process_time", {}).get("wall_ms_timev")
+    metrics["wall_ms"] = stage_total if isinstance(stage_total, int) else timev_total
+
+
+def append_metric(out_root: Path, metrics: dict[str, Any]) -> None:
+    path = out_root / "per_run_metrics.jsonl"
+    with path.open("a") as fh:
+        fh.write(json.dumps(metrics, ensure_ascii=False, sort_keys=True) + "\n")
+
+
+def load_metrics(out_root: Path) -> list[dict[str, Any]]:
+    path = out_root / "per_run_metrics.jsonl"
+    if not path.exists():
+        return []
+    return [json.loads(line) for line in path.read_text().splitlines() if line.strip()]
+
+
+def metric_row(metric: dict[str, Any]) -> dict[str, Any]:
+    stage = metric.get("stage") or {}
+    roa_cache = stage.get("roa_validation_cache") or {}
+    return {
+        "workers": metric["workers"],
+        "mode": metric["mode"],
+        "run_index": metric["run_index"],
+        "start_time_utc": metric.get("start_time_utc"),
+        "wall_ms": metric.get("wall_ms"),
+        "timev_wall_ms": metric.get("process_time", {}).get("wall_ms_timev"),
+        "max_rss_kb": metric.get("process_time", {}).get("max_rss_kb"),
+        "cpu_percent": metric.get("process_time", {}).get("cpu_percent_text"),
+        "validation_ms": stage.get("validation_ms"),
+        "repo_sync_ms_total": stage.get("repo_sync_ms_total"),
+        "rrdp_download_ms_total": stage.get("rrdp_download_ms_total"),
+        "rsync_download_ms_total": stage.get("rsync_download_ms_total"),
+        "download_bytes_total": stage.get("download_bytes_total"),
+        "publication_points": stage.get("publication_points"),
+        "pp_cache_hit_ratio": stage.get("publication_point_cache_hit_ratio"),
+        "roa_hit_roas": roa_cache.get("hit_roas"),
+        "roa_miss_roas": roa_cache.get("miss_roas"),
+        "vrps_rows": metric.get("vrps", {}).get("rows"),
+        "vrps_unique_rows": metric.get("vrps", {}).get("unique_rows"),
+        "vaps_rows": metric.get("vaps", {}).get("rows"),
+        "vaps_unique_rows": metric.get("vaps", {}).get("unique_rows"),
+        "state_bytes": metric.get("state_bytes"),
+        "run_bytes": metric.get("run_bytes"),
+        "remote_run_dir": metric.get("run_dir"),
+    }
+
+
+def write_csv_report(out_root: Path, metrics: list[dict[str, Any]]) -> None:
+    rows = [metric_row(item) for item in metrics]
+    if not rows:
+        return
+    with (out_root / "per_run_metrics.csv").open("w", newline="") as fh:
+        writer = csv.DictWriter(fh, fieldnames=list(rows[0]))
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+def fmt_ms(value: Any) -> str:
+    if value is None:
+        return "-"
+    try:
+        return f"{float(value) / 1000:.2f}s"
+    except (TypeError, ValueError):
+        return str(value)
+
+
+def fmt_bytes(value: Any) -> str:
+    if value is None:
+        return "-"
+    try:
+        value = float(value)
+    except (TypeError, ValueError):
+        return str(value)
+    for unit in ["B", "KiB", "MiB", "GiB"]:
+        if value < 1024 or unit == "GiB":
+            return f"{value:.1f} {unit}"
+        value /= 1024
+    return str(value)
+
+
+def write_markdown_report(out_root: Path, metrics: list[dict[str, Any]], provenance: dict[str, Any]) -> None:
+    rows = [metric_row(item) for item in metrics]
+    lines = [
+        "# Feature #115 Sync Worker Ablation Summary",
+        "",
+        "## Experiment",
+        "",
+        f"- Remote: {provenance['remote']}",
+        f"- Remote root: {provenance['remote_root']}",
+        f"- Local commit: {provenance['git_commit']}",
+        f"- Worker counts: {', '.join(str(w) for w in provenance['workers'])}",
+        f"- Runs per worker: {provenance['runs_per_worker']}",
+        f"- Interval seconds: {provenance['interval_secs']}",
+        "- Fixed profile: all5 live TAL URL, module-root rsync scope, PP cache enabled, ROA cache enabled, child-certificate cache enabled, transport prefetch disabled.",
+        "",
+        "## Per Run",
+        "",
+        "| workers | mode | wall | rss | validation | repo sync | RRDP | rsync | download | PP | VRPs | VAPs |",
+        "|---:|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|",
+    ]
+    for row in rows:
+        lines.append(
+            f"| {row['workers']} | {row['mode']} | {fmt_ms(row['wall_ms'])} | "
+            f"{fmt_bytes((row['max_rss_kb'] or 0) * 1024 if row['max_rss_kb'] else None)} | "
+            f"{fmt_ms(row['validation_ms'])} | {fmt_ms(row['repo_sync_ms_total'])} | "
+            f"{fmt_ms(row['rrdp_download_ms_total'])} | {fmt_ms(row['rsync_download_ms_total'])} | "
+            f"{fmt_bytes(row['download_bytes_total'])} | {row['publication_points']} | "
+            f"{row['vrps_rows']} | {row['vaps_rows']} |"
+        )
+    lines.extend(["", "## Artifacts", "", f"- Per-run metrics JSONL: {out_root / 'per_run_metrics.jsonl'}", f"- Per-run metrics CSV: {out_root / 'per_run_metrics.csv'}"])
+    write_text(out_root / "summary.md", "\n".join(lines) + "\n")
+
+
+def write_html_report(out_root: Path, metrics: list[dict[str, Any]], provenance: dict[str, Any]) -> None:
+    rows = [metric_row(item) for item in metrics]
+    if not rows:
+        return
+    max_wall = max(float(row["wall_ms"] or 0) for row in rows) or 1
+    bars = []
+    for row in rows:
+        width = max(1, int((float(row["wall_ms"] or 0) / max_wall) * 100))
+        bars.append(
+            f"<tr><td>{row['workers']}</td><td>{html.escape(str(row['mode']))}</td>"
+            f"<td>{fmt_ms(row['wall_ms'])}</td><td><div class='bar'><span style='width:{width}%'></span></div></td>"
+            f"<td>{fmt_bytes((row['max_rss_kb'] or 0) * 1024 if row['max_rss_kb'] else None)}</td>"
+            f"<td>{fmt_ms(row['validation_ms'])}</td><td>{fmt_ms(row['repo_sync_ms_total'])}</td>"
+            f"<td>{row['publication_points']}</td><td>{row['vrps_rows']}</td><td>{row['vaps_rows']}</td></tr>"
+        )
+    html_doc = f"""<!doctype html>
+<html lang="en">
+<head>
+<meta charset="utf-8">
+<title>Feature #115 Sync Worker Ablation</title>
+<style>
+body {{ font-family: system-ui, -apple-system, Segoe UI, sans-serif; margin: 32px; color: #1f2933; }}
+h1, h2 {{ margin-bottom: 0.35rem; }}
+.meta {{ display: grid; grid-template-columns: repeat(4, minmax(160px, 1fr)); gap: 12px; margin: 20px 0; }}
+.card {{ border: 1px solid #d8dee9; border-radius: 8px; padding: 14px; background: #f8fafc; }}
+.card b {{ display: block; font-size: 12px; color: #52606d; text-transform: uppercase; letter-spacing: .04em; }}
+.card span {{ font-size: 18px; }}
+table {{ border-collapse: collapse; width: 100%; margin-top: 12px; font-size: 13px; }}
+th, td {{ border: 1px solid #d8dee9; padding: 8px; text-align: left; vertical-align: middle; }}
+th {{ background: #eef2f7; }}
+.bar {{ width: 180px; height: 12px; background: #edf2f7; border-radius: 999px; overflow: hidden; }}
+.bar span {{ display: block; height: 100%; background: #2563eb; }}
+code {{ background: #edf2f7; padding: 2px 5px; border-radius: 4px; }}
+</style>
+</head>
+<body>
+<h1>Feature #115 Sync Worker Ablation</h1>
+<p>This report compares only the global repo-sync worker count while holding the Feature #100 ours-cache profile fixed.</p>
+<section class="meta">
+<div class="card"><b>Remote</b><span>{html.escape(provenance['remote'])}</span></div>
+<div class="card"><b>Remote Root</b><span>{html.escape(provenance['remote_root'])}</span></div>
+<div class="card"><b>Workers</b><span>{html.escape(', '.join(str(w) for w in provenance['workers']))}</span></div>
+<div class="card"><b>Runs</b><span>{provenance['runs_per_worker']} per worker, {provenance['interval_secs']}s interval</span></div>
+</section>
+<h2>Per Run Metrics</h2>
+<table>
+<thead><tr><th>Workers</th><th>Mode</th><th>Wall</th><th>Wall Bar</th><th>Max RSS</th><th>Validation</th><th>Repo Sync</th><th>PP</th><th>VRPs</th><th>VAPs</th></tr></thead>
+<tbody>
+{''.join(bars)}
+</tbody>
+</table>
+<h2>Configuration</h2>
+<p>all5 live TAL URL, module-root rsync scope, PP cache enabled, ROA cache enabled, child-certificate cache enabled, transport prefetch disabled.</p>
+<p>Local commit: <code>{html.escape(provenance['git_commit'])}</code>. Full JSONL and CSV evidence are stored beside this report.</p>
+</body>
+</html>
+"""
+    write_text(out_root / "summary.html", html_doc)
+
+
+def write_reports(out_root: Path, provenance: dict[str, Any]) -> None:
+    metrics = load_metrics(out_root)
+    write_csv_report(out_root, metrics)
+    write_markdown_report(out_root, metrics, provenance)
+    write_html_report(out_root, metrics, provenance)
+
+
+def git_metadata() -> dict[str, str]:
+    commit = run_cmd(["git", "rev-parse", "--short", "HEAD"], cwd=PRODUCT_ROOT).stdout.strip()
+    status = run_cmd(["git", "status", "--short"], cwd=PRODUCT_ROOT).stdout
+    return {"git_commit": commit, "git_status_short": status}
+
+
+def main() -> int:
+    args = parse_args()
+    stamp = utc_stamp()
+    out_root = (args.out_root / stamp).resolve()
+    remote_root = args.remote_root or f"/root/rpki_feature115_sync_worker_ablation_{stamp}"
+    remote_bin = f"{remote_root}/bin/rpki"
+    plans = plan_runs(args.workers, args.runs_per_worker, args.interval_secs, remote_root, remote_bin)
+
+    provenance = {
+        "created_at_utc": stamp,
+        "remote": args.remote,
+        "remote_root": remote_root,
+        "workers": args.workers,
+        "runs_per_worker": args.runs_per_worker,
+        "interval_secs": args.interval_secs,
+        "fixed_profile": "feature100 ours-pp-object-cache",
+        **git_metadata(),
+    }
+    write_json(out_root / "provenance.json", provenance)
+    write_text(out_root / "remote_root.txt", remote_root + "\n")
+    write_json(out_root / "execution_plan.json", plans)
+
+    if args.dry_run:
+        print(f"dry-run plan written to {out_root}")
+        return 0
+
+    binary = ensure_binary(args)
+    remote_bin = prepare_remote(args.remote, remote_root, binary)
+    plans = plan_runs(args.workers, args.runs_per_worker, args.interval_secs, remote_root, remote_bin)
+    write_json(out_root / "execution_plan.json", plans)
+    start_epoch = time.time()
+    for plan in plans:
+        write_json(
+            out_root / "run_configs" / f"workers-{plan['workers']:02d}-run-{plan['run_index']:04d}-{plan['mode']}.json",
+            plan,
+        )
+        scheduled_epoch = start_epoch + plan["ordinal"] * args.interval_secs
+        metrics = run_remote_plan(args.remote, plan, scheduled_epoch)
+        append_metric(out_root, metrics)
+        write_reports(out_root, provenance)
+    print(f"reports written under {out_root}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
--- a/scripts/rpctl/README.md
+++ b/scripts/rpctl/README.md
@ -17,6 +17,10 @@ ssh root@47.251.127.231 'cd /data/rpki_4rp_ctl && ./rpctl init \

 `--rpki-prover-image` 是可选项；不配置时原有四款 RP 仍可运行，但启动 rpki-prover 会明确报错。配置后 `init` 会复制并加载离线镜像，校验宿主机和镜像均为 `amd64`，确认 tag 为 `rpki-prover:6857d4bf`、版本为 `rpki-prover-0.10.1`，并将 image ID 与 archive SHA-256 写入 `config.json`。运行时不会在线拉取镜像。

+## ours RP 默认同步并发
+
+`ours` 和 `ours-no-cache` 两个 profile 都会显式传入 `--parallel-max-repo-sync-workers-global 8`，即默认使用 8 个全局仓库同步 worker。此设置只作用于 ours RP；Routinator、rpki-client 与 FORT 的运行参数保持不变。`show` 与 `view --command` 会在参数摘要中显示 `sync-workers=8`。
+
 ## 常用命令

 ```bash
--- a/scripts/rpctl/rpctl
+++ b/scripts/rpctl/rpctl
@ -415,6 +415,7 @@ def build_ours_args(profile: str, run_root: Path, run_dir: Path, state: Path, ri
        "--vrps-csv-out", str(run_dir / "vrps.csv"),
        "--vaps-csv-out", str(run_dir / "vaps.csv"),
        "--compare-view-trust-anchor", "all5" if len(rirs) > 1 else rirs[0],
+        "--parallel-max-repo-sync-workers-global", "8",
        "--parallel-phase2-ready-batch-size", "256",
        "--parallel-phase2-ready-batch-wall-time-budget-ms", "100",
        "--parallel-phase2-result-drain-batch-size", "2048",
@ -1270,6 +1271,10 @@ def parse_command_summary(command: str) -> str:
            keys.append(f"--rsync-scope={tokens[idx + 1]}")
        elif token.startswith("--rsync-scope="):
            keys.append(token)
+        elif token == "--parallel-max-repo-sync-workers-global" and idx + 1 < len(tokens):
+            keys.append(f"sync-workers={tokens[idx + 1]}")
+        elif token.startswith("--parallel-max-repo-sync-workers-global="):
+            keys.append(f"sync-workers={token.rsplit('=', 1)[1]}")
    if RPKI_PROVER_IMAGE_TAG in token_set:
        algorithm = "full-every-iteration" if "--no-incremental-validation" in token_set else "incremental"
        keys.append(f"algorithm={algorithm}")
@ -1868,6 +1873,10 @@ Cache difference for ours RP:

  --rp ours-no-cache does not add those flags.

+Ours RP sync parallelism:
+  Both ours RP profiles explicitly use 8 global repository sync workers.
+  This setting does not change Routinator, rpki-client, or FORT parameters.
+
 Typical usage:
  ./rpctl init --bundle /data/rpki_4rp_ctl_runtime_bundle \\
    --rpki-prover-image /data/rpki-prover-6857d4bf-linux-amd64.docker.tar.gz
@ -1974,6 +1983,10 @@ Cache mapping:
  --rp ours         enables PP cache + ROA cache + child certificate cache.
  --rp ours-no-cache disables those cache flags.

+Ours RP sync workers:
+  Both ours profiles pass --parallel-max-repo-sync-workers-global 8.
+  This does not alter the parameters of Routinator, rpki-client, or FORT.
+
 Safety:
  rpctl refuses to start a second RP while another rpctl runner is alive.
 """,
--- a/src/tools/rpki_artifact_metrics.rs
+++ b/src/tools/rpki_artifact_metrics.rs
@ -3,6 +3,7 @@ use std::fs;
 use std::io::{Read, Write};
 use std::net::{TcpListener, TcpStream};
 use std::path::{Path, PathBuf};
+use std::process::{Command, Stdio};
 use std::sync::{Arc, RwLock};
 use std::thread;
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
@ -26,10 +27,13 @@ struct Args {
    once: bool,
    out_metrics: Option<PathBuf>,
    out_status: Option<PathBuf>,
+    ccr_check_bin: Option<PathBuf>,
+    ccr_check_timeout_secs: u64,
+    ccr_check_max_per_scan: u64,
 }

 fn usage() -> &'static str {
-    "Usage: rpki_artifact_metrics --run-root <path> [--listen <addr:port>] [--poll-secs <n>] [--instance <name>] [--once] [--out-metrics <path>] [--out-status <path>]"
+    "Usage: rpki_artifact_metrics --run-root <path> [--listen <addr:port>] [--poll-secs <n>] [--instance <name>] [--once] [--out-metrics <path>] [--out-status <path>] [--ccr-check-bin <path>] [--ccr-check-timeout-secs <n>] [--ccr-check-max-per-scan <n>]"
 }

 pub fn main_entry() -> Result<(), String> {
@ -38,8 +42,16 @@ pub fn main_entry() -> Result<(), String> {

 fn real_main() -> Result<(), String> {
    let args = parse_args(&std::env::args().collect::<Vec<_>>())?;
+    let ccr_check_config = CcrCheckConfig::from_args(&args);
+    let rtr_report_dir = configured_rtr_report_dir();
    if args.once {
-        let snapshot = scan_run_root(&args.run_root, &args.instance)?;
+        let snapshot = scan_run_root_incremental_with_rtr_dir(
+            &args.run_root,
+            &args.instance,
+            None,
+            rtr_report_dir.as_deref(),
+            &ccr_check_config,
+        )?;
        let metrics = render_metrics(&snapshot);
        let status = render_status_json(&snapshot)?;
        if let Some(path) = args.out_metrics.as_ref() {
@ -53,7 +65,13 @@ fn real_main() -> Result<(), String> {
        return Ok(());
    }

-    let shared = Arc::new(RwLock::new(scan_run_root(&args.run_root, &args.instance)?));
+    let shared = Arc::new(RwLock::new(scan_run_root_incremental_with_rtr_dir(
+        &args.run_root,
+        &args.instance,
+        None,
+        rtr_report_dir.as_deref(),
+        &ccr_check_config,
+    )?));
    let scanner = Arc::clone(&shared);
    let run_root = args.run_root.clone();
    let instance = args.instance.clone();
@ -62,20 +80,25 @@ fn real_main() -> Result<(), String> {
        loop {
            thread::sleep(Duration::from_secs(poll_secs));
            let previous_snapshot = scanner.read().expect("metrics lock poisoned").clone();
-            let next =
-                match scan_run_root_incremental(&run_root, &instance, Some(&previous_snapshot)) {
-                    Ok(snapshot) => snapshot,
-                    Err(err) => {
-                        let mut previous = scanner.write().expect("metrics lock poisoned");
-                        previous
-                            .service
-                            .parse_errors
-                            .push(format!("scan failed: {err}"));
-                        previous.service.last_reload_success = false;
-                        previous.service.last_scan_timestamp_seconds = unix_now_seconds();
-                        continue;
-                    }
-                };
+            let next = match scan_run_root_incremental_with_rtr_dir(
+                &run_root,
+                &instance,
+                Some(&previous_snapshot),
+                rtr_report_dir.as_deref(),
+                &ccr_check_config,
+            ) {
+                Ok(snapshot) => snapshot,
+                Err(err) => {
+                    let mut previous = scanner.write().expect("metrics lock poisoned");
+                    previous
+                        .service
+                        .parse_errors
+                        .push(format!("scan failed: {err}"));
+                    previous.service.last_reload_success = false;
+                    previous.service.last_scan_timestamp_seconds = unix_now_seconds();
+                    continue;
+                }
+            };
            *scanner.write().expect("metrics lock poisoned") = next;
        }
    });
@ -91,6 +114,9 @@ fn parse_args(argv: &[String]) -> Result<Args, String> {
    let mut once = false;
    let mut out_metrics = None;
    let mut out_status = None;
+    let mut ccr_check_bin = None;
+    let mut ccr_check_timeout_secs = 120u64;
+    let mut ccr_check_max_per_scan = 2u64;
    let mut index = 1usize;
    while index < argv.len() {
        match argv[index].as_str() {
@ -122,6 +148,24 @@ fn parse_args(argv: &[String]) -> Result<Args, String> {
                index += 1;
                out_status = Some(PathBuf::from(value_at(argv, index, "--out-status")?));
            }
+            "--ccr-check-bin" => {
+                index += 1;
+                ccr_check_bin = Some(PathBuf::from(value_at(argv, index, "--ccr-check-bin")?));
+            }
+            "--ccr-check-timeout-secs" => {
+                index += 1;
+                let value = value_at(argv, index, "--ccr-check-timeout-secs")?;
+                ccr_check_timeout_secs = value
+                    .parse::<u64>()
+                    .map_err(|_| format!("invalid --ccr-check-timeout-secs: {value}"))?;
+            }
+            "--ccr-check-max-per-scan" => {
+                index += 1;
+                let value = value_at(argv, index, "--ccr-check-max-per-scan")?;
+                ccr_check_max_per_scan = value
+                    .parse::<u64>()
+                    .map_err(|_| format!("invalid --ccr-check-max-per-scan: {value}"))?;
+            }
            "-h" | "--help" => return Err(usage().to_string()),
            other => return Err(format!("unknown argument: {other}\n{}", usage())),
        }
@ -135,6 +179,9 @@ fn parse_args(argv: &[String]) -> Result<Args, String> {
        once,
        out_metrics,
        out_status,
+        ccr_check_bin,
+        ccr_check_timeout_secs,
+        ccr_check_max_per_scan,
    })
 }

@ -161,9 +208,67 @@ struct MetricsSnapshot {
    top_pp_by_sync_duration: Vec<TopPublicationPoint>,
    cir: Option<CirMetrics>,
    ccr: Option<CcrMetrics>,
+    ccr_check: CcrCheckMetrics,
    rtr: RtrMetrics,
 }

+#[derive(Clone, Debug, Default, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct CcrCheckMetrics {
+    enabled: bool,
+    up: bool,
+    bin: Option<String>,
+    pass_total: u64,
+    ccr_parse_error_total: u64,
+    tool_error_total: u64,
+    consecutive_failures: u64,
+    last: Option<CcrCheckLastResult>,
+    checked_run_dirs: Vec<String>,
+}
+
+#[derive(Clone, Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct CcrCheckLastResult {
+    run_dir: String,
+    run_seq: Option<u64>,
+    outcome: String,
+    duration_seconds: f64,
+    checked_at_seconds: f64,
+    detail: Option<String>,
+}
+
+#[derive(Clone, Debug)]
+struct CcrCheckConfig {
+    bin: Option<PathBuf>,
+    timeout_secs: u64,
+    max_per_scan: u64,
+}
+
+impl CcrCheckConfig {
+    fn disabled() -> Self {
+        Self {
+            bin: None,
+            timeout_secs: 120,
+            max_per_scan: 2,
+        }
+    }
+
+    fn from_args(args: &Args) -> Self {
+        let bin = args.ccr_check_bin.clone().or_else(|| {
+            std::env::var("RPKI_METRICS_CCR_CHECK_BIN")
+                .ok()
+                .map(|value| value.trim().to_string())
+                .filter(|value| !value.is_empty())
+                .map(PathBuf::from)
+        });
+        Self {
+            bin,
+            timeout_secs: args.ccr_check_timeout_secs,
+            max_per_scan: args.ccr_check_max_per_scan,
+        }
+    }
+}
+
 #[derive(Clone, Debug, Default, Serialize)]
 #[serde(rename_all = "camelCase")]
 struct ServiceMetrics {
@ -493,21 +598,13 @@ struct RunRecord {
 }

 fn scan_run_root(input_root: &Path, instance: &str) -> Result<MetricsSnapshot, String> {
-    let rtr_report_dir = configured_rtr_report_dir();
-    scan_run_root_incremental_with_rtr_dir(input_root, instance, None, rtr_report_dir.as_deref())
-}
-
-fn scan_run_root_incremental(
-    input_root: &Path,
-    instance: &str,
-    previous: Option<&MetricsSnapshot>,
-) -> Result<MetricsSnapshot, String> {
    let rtr_report_dir = configured_rtr_report_dir();
    scan_run_root_incremental_with_rtr_dir(
        input_root,
        instance,
-        previous,
+        None,
        rtr_report_dir.as_deref(),
+        &CcrCheckConfig::disabled(),
    )
 }

@ -516,6 +613,7 @@ fn scan_run_root_incremental_with_rtr_dir(
    instance: &str,
    previous: Option<&MetricsSnapshot>,
    rtr_report_dir: Option<&Path>,
+    ccr_check_config: &CcrCheckConfig,
 ) -> Result<MetricsSnapshot, String> {
    let started = Instant::now();
    let runs_root = resolve_runs_root(input_root);
@ -570,6 +668,7 @@ fn scan_run_root_incremental_with_rtr_dir(
    }

    snapshot.rtr = scan_rtr_reports(rtr_report_dir, previous.map(|snapshot| &snapshot.rtr));
+    run_pending_ccr_checks(&records, previous, ccr_check_config, &mut snapshot);
    snapshot.service.last_scan_timestamp_seconds = unix_now_seconds();
    snapshot.service.last_scan_duration_seconds = started.elapsed().as_secs_f64();
    snapshot.service.last_reload_success =
@ -975,6 +1074,218 @@ fn consecutive_failures(records: &[RunRecord]) -> u64 {
    count
 }

+const MAX_CHECKED_RUN_DIRS: usize = 512;
+
+fn run_pending_ccr_checks(
+    records: &[RunRecord],
+    previous: Option<&MetricsSnapshot>,
+    config: &CcrCheckConfig,
+    snapshot: &mut MetricsSnapshot,
+) {
+    let mut check = previous
+        .map(|prev| prev.ccr_check.clone())
+        .unwrap_or_default();
+    check.enabled = config.bin.is_some();
+    check.bin = config.bin.as_ref().map(|bin| bin.display().to_string());
+    check.up = false;
+
+    let Some(bin) = config.bin.as_ref() else {
+        snapshot.ccr_check = check;
+        return;
+    };
+    if !bin.is_file() {
+        snapshot
+            .service
+            .parse_errors
+            .push(format!("ccr check bin is not a file: {}", bin.display()));
+        snapshot.ccr_check = check;
+        return;
+    }
+    check.up = true;
+
+    let mut pending: Vec<&RunRecord> = records
+        .iter()
+        .filter(|record| record.status == "success")
+        .filter(|record| record.path.join("result.ccr").is_file())
+        .filter(|record| {
+            let dir = record.path.display().to_string();
+            !check.checked_run_dirs.iter().any(|seen| seen == &dir)
+        })
+        .collect();
+
+    if previous.is_none() {
+        // On service start only the latest existing run is checked; older
+        // history is marked as seen without executing checks.
+        let latest_path = pending.last().map(|record| record.path.clone());
+        for record in &pending {
+            if Some(&record.path) != latest_path.as_ref() {
+                check
+                    .checked_run_dirs
+                    .push(record.path.display().to_string());
+            }
+        }
+        pending.retain(|record| Some(&record.path) == latest_path.as_ref());
+    }
+
+    for record in pending
+        .into_iter()
+        .take(config.max_per_scan.max(1) as usize)
+    {
+        let ccr_path = record.path.join("result.ccr");
+        let outcome = execute_ccr_check(bin, &ccr_path, Duration::from_secs(config.timeout_secs));
+        let run_seq = record
+            .summary
+            .as_ref()
+            .and_then(|v| json_u64(v, &["runSeq"]))
+            .or_else(|| {
+                record
+                    .meta
+                    .as_ref()
+                    .and_then(|v| json_u64(v, &["run_index"]))
+            });
+        let passed = outcome.outcome == "pass";
+        match outcome.outcome.as_str() {
+            "pass" => check.pass_total += 1,
+            "ccr_parse_error" => check.ccr_parse_error_total += 1,
+            _ => check.tool_error_total += 1,
+        }
+        check.consecutive_failures = if passed {
+            0
+        } else {
+            check.consecutive_failures + 1
+        };
+        check.last = Some(CcrCheckLastResult {
+            run_dir: record.path.display().to_string(),
+            run_seq,
+            outcome: outcome.outcome,
+            duration_seconds: outcome.duration_seconds,
+            checked_at_seconds: unix_now_seconds(),
+            detail: outcome.detail,
+        });
+        check
+            .checked_run_dirs
+            .push(record.path.display().to_string());
+    }
+    if check.checked_run_dirs.len() > MAX_CHECKED_RUN_DIRS {
+        let drain = check.checked_run_dirs.len() - MAX_CHECKED_RUN_DIRS;
+        check.checked_run_dirs.drain(..drain);
+    }
+    snapshot.ccr_check = check;
+}
+
+struct CcrCheckOutcome {
+    outcome: String,
+    duration_seconds: f64,
+    detail: Option<String>,
+}
+
+fn execute_ccr_check(bin: &Path, ccr_path: &Path, timeout: Duration) -> CcrCheckOutcome {
+    let started = Instant::now();
+    let log_path = std::env::temp_dir().join(format!(
+        "rpki-artifact-metrics-ccr-check-{}-{}.log",
+        std::process::id(),
+        SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map(|d| d.as_nanos())
+            .unwrap_or(0)
+    ));
+    let (outcome, detail) = execute_ccr_check_logged(bin, ccr_path, timeout, &log_path);
+    let _ = fs::remove_file(&log_path);
+    CcrCheckOutcome {
+        outcome,
+        duration_seconds: started.elapsed().as_secs_f64(),
+        detail,
+    }
+}
+
+fn execute_ccr_check_logged(
+    bin: &Path,
+    ccr_path: &Path,
+    timeout: Duration,
+    log_path: &Path,
+) -> (String, Option<String>) {
+    let log_file = match fs::File::create(log_path) {
+        Ok(file) => file,
+        Err(err) => {
+            return (
+                "tool_error".to_string(),
+                Some(format!("create ccr check log failed: {err}")),
+            );
+        }
+    };
+    let log_err = match log_file.try_clone() {
+        Ok(file) => file,
+        Err(err) => {
+            return (
+                "tool_error".to_string(),
+                Some(format!("clone ccr check log handle failed: {err}")),
+            );
+        }
+    };
+    let mut child = match Command::new(bin)
+        .arg("-f")
+        .arg(ccr_path)
+        .stdin(Stdio::null())
+        .stdout(Stdio::from(log_file))
+        .stderr(Stdio::from(log_err))
+        .spawn()
+    {
+        Ok(child) => child,
+        Err(err) => {
+            return (
+                "tool_error".to_string(),
+                Some(format!("spawn ccr check failed: {err}")),
+            );
+        }
+    };
+    let started = Instant::now();
+    let exit_code = loop {
+        match child.try_wait() {
+            Ok(Some(status)) => break status.code(),
+            Ok(None) => {
+                if started.elapsed() >= timeout {
+                    let _ = child.kill();
+                    let _ = child.wait();
+                    return (
+                        "tool_error".to_string(),
+                        Some(format!("ccr check timed out after {}s", timeout.as_secs())),
+                    );
+                }
+                thread::sleep(Duration::from_millis(100));
+            }
+            Err(err) => {
+                return (
+                    "tool_error".to_string(),
+                    Some(format!("wait ccr check failed: {err}")),
+                );
+            }
+        }
+    };
+    let output = fs::read_to_string(log_path).unwrap_or_default();
+    classify_ccr_check_output(exit_code, ccr_path, &output)
+}
+
+fn classify_ccr_check_output(
+    exit_code: Option<i32>,
+    ccr_path: &Path,
+    output: &str,
+) -> (String, Option<String>) {
+    if exit_code != Some(0) {
+        return (
+            "tool_error".to_string(),
+            Some(format!("rpki-client exited with code {exit_code:?}")),
+        );
+    }
+    // rpki-client exits 0 even when the CCR file itself fails to parse, so
+    // the file-level error line is the authoritative failure signal. This
+    // mirrors scripts/ccr/check_rpki_client_ccr.sh.
+    let marker = format!("rpki-client: {}:", ccr_path.display());
+    if let Some(line) = output.lines().find(|line| line.contains(&marker)) {
+        return ("ccr_parse_error".to_string(), Some(line.trim().to_string()));
+    }
+    ("pass".to_string(), None)
+}
+
 fn read_json_optional(path: &Path, errors: &mut Vec<String>) -> Option<Value> {
    if !path.exists() {
        return None;
@ -1634,10 +1945,60 @@ fn render_metrics(snapshot: &MetricsSnapshot) -> String {
    if let Some(ccr) = snapshot.ccr.as_ref() {
        render_ccr_metrics(&mut writer, instance, ccr);
    }
+    render_ccr_check_metrics(&mut writer, instance, &snapshot.ccr_check);
    render_rtr_metrics(&mut writer, instance, snapshot);
    out
 }

+fn render_ccr_check_metrics(writer: &mut PromWriter<'_>, instance: &str, check: &CcrCheckMetrics) {
+    writer.gauge(
+        "ours_rp_ccr_format_check_up",
+        "Whether the CCR rpki-client format checker binary is configured and present",
+        &[label("instance", instance)],
+        bool_value(check.up),
+    );
+    for (result, value) in [
+        ("pass", check.pass_total),
+        ("ccr_parse_error", check.ccr_parse_error_total),
+        ("tool_error", check.tool_error_total),
+    ] {
+        writer.counter(
+            "ours_rp_ccr_format_check_total",
+            "CCR rpki-client format check results by classification",
+            &[label("instance", instance), label("result", result)],
+            value as f64,
+        );
+    }
+    writer.gauge(
+        "ours_rp_ccr_format_check_consecutive_failures",
+        "Consecutive failed CCR rpki-client format checks",
+        &[label("instance", instance)],
+        check.consecutive_failures as f64,
+    );
+    if let Some(last) = check.last.as_ref() {
+        writer.gauge(
+            "ours_rp_ccr_format_check_last_success",
+            "Whether the latest CCR rpki-client format check passed",
+            &[label("instance", instance)],
+            bool_value(last.outcome == "pass"),
+        );
+        if let Some(run_seq) = last.run_seq {
+            writer.gauge(
+                "ours_rp_ccr_format_check_last_run_sequence",
+                "Run sequence of the latest checked CCR artifact",
+                &[label("instance", instance)],
+                run_seq as f64,
+            );
+        }
+        writer.gauge(
+            "ours_rp_ccr_format_check_last_duration_seconds",
+            "Duration of the latest CCR rpki-client format check",
+            &[label("instance", instance)],
+            last.duration_seconds,
+        );
+    }
+}
+
 fn render_rtr_metrics(writer: &mut PromWriter<'_>, instance: &str, snapshot: &MetricsSnapshot) {
    let rtr = &snapshot.rtr;
    let scan_now = snapshot.service.last_scan_timestamp_seconds;
@ -2525,6 +2886,7 @@ fn render_status_json(snapshot: &MetricsSnapshot) -> Result<String, String> {
        "latestRun": snapshot.latest_run,
        "cir": snapshot.cir,
        "ccr": snapshot.ccr,
+        "ccrCheck": snapshot.ccr_check,
        "rtr": snapshot.rtr,
        "topRepositoriesBySyncDuration": snapshot.top_repos_by_sync_duration,
        "topPublicationPointsByObjectCount": snapshot.top_pp_by_object_count,
@ -3162,8 +3524,14 @@ mod tests {
        )
        .expect("clients");

-        let snapshot = scan_run_root_incremental_with_rtr_dir(td.path(), "test", None, Some(&rtr))
-            .expect("scan");
+        let snapshot = scan_run_root_incremental_with_rtr_dir(
+            td.path(),
+            "test",
+            None,
+            Some(&rtr),
+            &CcrCheckConfig::disabled(),
+        )
+        .expect("scan");
        assert!(snapshot.rtr.enabled);
        assert_eq!(snapshot.rtr.files["source"].seen, 1);
        assert_eq!(
@ -3233,8 +3601,14 @@ mod tests {
        )
        .expect("clients");

-        let snapshot = scan_run_root_incremental_with_rtr_dir(td.path(), "test", None, Some(&rtr))
-            .expect("scan");
+        let snapshot = scan_run_root_incremental_with_rtr_dir(
+            td.path(),
+            "test",
+            None,
+            Some(&rtr),
+            &CcrCheckConfig::disabled(),
+        )
+        .expect("scan");
        assert!(snapshot.rtr.enabled);
        assert_eq!(snapshot.rtr.files["source"].seen, 1);
        assert!(snapshot.rtr.source.is_none());
@ -3247,6 +3621,223 @@ mod tests {
        assert!(metrics.contains(r#"ours_rp_rtr_metrics_enabled{instance="test"} 1"#));
    }

+    #[test]
+    fn parse_args_accepts_ccr_check_options() {
+        let args = parse_args(&[
+            "rpki_artifact_metrics".to_string(),
+            "--run-root".to_string(),
+            "root".to_string(),
+            "--ccr-check-bin".to_string(),
+            "/opt/ours-rp/bin/rpki-client".to_string(),
+            "--ccr-check-timeout-secs".to_string(),
+            "60".to_string(),
+            "--ccr-check-max-per-scan".to_string(),
+            "3".to_string(),
+        ])
+        .expect("parse");
+        assert_eq!(
+            args.ccr_check_bin.as_deref(),
+            Some(Path::new("/opt/ours-rp/bin/rpki-client"))
+        );
+        assert_eq!(args.ccr_check_timeout_secs, 60);
+        assert_eq!(args.ccr_check_max_per_scan, 3);
+    }
+
+    #[test]
+    fn ccr_check_disabled_without_bin() {
+        let td = TempDir::new().expect("tempdir");
+        write_success_run(td.path(), "run_0001", 1);
+        let snapshot = scan_run_root(td.path(), "test").expect("scan");
+        assert!(!snapshot.ccr_check.enabled);
+        assert!(!snapshot.ccr_check.up);
+        assert!(snapshot.ccr_check.last.is_none());
+        let metrics = render_metrics(&snapshot);
+        assert!(metrics.contains(r#"ours_rp_ccr_format_check_up{instance="test"} 0"#));
+        assert!(!metrics.contains("ours_rp_ccr_format_check_last_success"));
+    }
+
+    #[test]
+    fn ccr_check_pass_records_metrics() {
+        let td = TempDir::new().expect("tempdir");
+        write_success_run(td.path(), "run_0001", 1);
+        let bin = write_fake_rpki_client(td.path(), "#!/bin/sh\nexit 0\n");
+        let snapshot = scan_with_ccr_check(td.path(), None, &bin, 30);
+        assert!(snapshot.ccr_check.up);
+        assert_eq!(snapshot.ccr_check.pass_total, 1);
+        assert_eq!(snapshot.ccr_check.consecutive_failures, 0);
+        let last = snapshot.ccr_check.last.as_ref().expect("last result");
+        assert_eq!(last.outcome, "pass");
+        assert_eq!(last.run_seq, Some(1));
+        let metrics = render_metrics(&snapshot);
+        assert!(
+            metrics.contains(r#"ours_rp_ccr_format_check_total{instance="test",result="pass"} 1"#)
+        );
+        assert!(metrics.contains(r#"ours_rp_ccr_format_check_last_success{instance="test"} 1"#));
+        assert!(
+            metrics.contains(r#"ours_rp_ccr_format_check_last_run_sequence{instance="test"} 1"#)
+        );
+        let status = render_status_json(&snapshot).expect("status");
+        assert!(status.contains(r#""ccrCheck""#));
+        assert!(status.contains(r#""passTotal": 1"#));
+    }
+
+    #[test]
+    fn ccr_check_parse_error_detected_from_output() {
+        let td = TempDir::new().expect("tempdir");
+        write_success_run(td.path(), "run_0001", 1);
+        let bin = write_fake_rpki_client(
+            td.path(),
+            "#!/bin/sh\necho \"rpki-client: $2: unexpected number of locations\" >&2\nexit 0\n",
+        );
+        let snapshot = scan_with_ccr_check(td.path(), None, &bin, 30);
+        assert_eq!(snapshot.ccr_check.ccr_parse_error_total, 1);
+        assert_eq!(snapshot.ccr_check.consecutive_failures, 1);
+        let last = snapshot.ccr_check.last.as_ref().expect("last result");
+        assert_eq!(last.outcome, "ccr_parse_error");
+        assert!(
+            last.detail
+                .as_deref()
+                .unwrap_or_default()
+                .contains("unexpected number of locations")
+        );
+        let metrics = render_metrics(&snapshot);
+        assert!(metrics.contains(
+            r#"ours_rp_ccr_format_check_total{instance="test",result="ccr_parse_error"} 1"#
+        ));
+        assert!(metrics.contains(r#"ours_rp_ccr_format_check_last_success{instance="test"} 0"#));
+    }
+
+    #[test]
+    fn ccr_check_nonzero_exit_is_tool_error() {
+        let td = TempDir::new().expect("tempdir");
+        write_success_run(td.path(), "run_0001", 1);
+        let bin = write_fake_rpki_client(td.path(), "#!/bin/sh\nexit 3\n");
+        let snapshot = scan_with_ccr_check(td.path(), None, &bin, 30);
+        assert_eq!(snapshot.ccr_check.tool_error_total, 1);
+        assert_eq!(
+            snapshot.ccr_check.last.as_ref().expect("last").outcome,
+            "tool_error"
+        );
+    }
+
+    #[test]
+    fn ccr_check_timeout_is_tool_error() {
+        let td = TempDir::new().expect("tempdir");
+        write_success_run(td.path(), "run_0001", 1);
+        let bin = write_fake_rpki_client(td.path(), "#!/bin/sh\nsleep 30\n");
+        let snapshot = scan_with_ccr_check(td.path(), None, &bin, 1);
+        let last = snapshot.ccr_check.last.as_ref().expect("last result");
+        assert_eq!(last.outcome, "tool_error");
+        assert!(
+            last.detail
+                .as_deref()
+                .unwrap_or_default()
+                .contains("timed out")
+        );
+    }
+
+    #[test]
+    fn ccr_check_startup_only_checks_latest_and_new_runs_once() {
+        let td = TempDir::new().expect("tempdir");
+        write_success_run(td.path(), "run_0001", 1);
+        write_success_run(td.path(), "run_0002", 2);
+        let bin = write_fake_rpki_client(td.path(), "#!/bin/sh\nexit 0\n");
+
+        // First scan at service start: only the latest existing run is
+        // checked, older history is marked as seen without executing checks.
+        let first = scan_with_ccr_check(td.path(), None, &bin, 30);
+        assert_eq!(first.ccr_check.pass_total, 1);
+        assert!(
+            first
+                .ccr_check
+                .last
+                .as_ref()
+                .expect("last")
+                .run_dir
+                .contains("run_0002")
+        );
+
+        // Second scan: pre-existing history must not be backfilled.
+        let second = scan_with_ccr_check(td.path(), Some(&first), &bin, 30);
+        assert_eq!(second.ccr_check.pass_total, 1);
+
+        // A newly completed run is checked exactly once.
+        write_success_run(td.path(), "run_0003", 3);
+        let third = scan_with_ccr_check(td.path(), Some(&second), &bin, 30);
+        assert_eq!(third.ccr_check.pass_total, 2);
+        assert_eq!(
+            third.ccr_check.last.as_ref().expect("last").run_seq,
+            Some(3)
+        );
+        let fourth = scan_with_ccr_check(td.path(), Some(&third), &bin, 30);
+        assert_eq!(fourth.ccr_check.pass_total, 2);
+    }
+
+    #[test]
+    fn ccr_check_missing_bin_marks_not_up() {
+        let td = TempDir::new().expect("tempdir");
+        write_success_run(td.path(), "run_0001", 1);
+        let missing = td.path().join("no-such-rpki-client");
+        let snapshot = scan_with_ccr_check(td.path(), None, &missing, 30);
+        assert!(snapshot.ccr_check.enabled);
+        assert!(!snapshot.ccr_check.up);
+        assert!(snapshot.ccr_check.last.is_none());
+        assert!(
+            snapshot
+                .service
+                .parse_errors
+                .iter()
+                .any(|error| error.contains("ccr check bin is not a file"))
+        );
+        let metrics = render_metrics(&snapshot);
+        assert!(metrics.contains(r#"ours_rp_ccr_format_check_up{instance="test"} 0"#));
+    }
+
+    fn write_success_run(root: &Path, name: &str, seq: u64) {
+        let run = root.join("runs").join(name);
+        fs::create_dir_all(&run).expect("create run");
+        fs::write(
+            run.join("run-meta.json"),
+            format!(
+                r#"{{"status":"success","run_index":{seq},"run_id":"{name}","sync_mode":"snapshot","snapshot_reason":"first_run","started_at_rfc3339_utc":"2026-07-21T00:00:00Z","completed_at_rfc3339_utc":"2026-07-21T00:00:10Z"}}"#
+            ),
+        )
+        .expect("meta");
+        fs::write(
+            run.join("run-summary.json"),
+            format!(
+                r#"{{"runSeq":{seq},"runId":"{name}","startedAtRfc3339Utc":"2026-07-21T00:00:00Z","finishedAtRfc3339Utc":"2026-07-21T00:00:10Z","wallMs":10000,"status":"success","exitCode":0}}"#
+            ),
+        )
+        .expect("summary");
+        fs::write(run.join("result.ccr"), sample_ccr()).expect("ccr");
+    }
+
+    fn write_fake_rpki_client(root: &Path, body: &str) -> PathBuf {
+        let path = root.join("fake-rpki-client.sh");
+        fs::write(&path, body).expect("write fake rpki-client");
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::PermissionsExt;
+            fs::set_permissions(&path, fs::Permissions::from_mode(0o755)).expect("chmod");
+        }
+        path
+    }
+
+    fn scan_with_ccr_check(
+        root: &Path,
+        previous: Option<&MetricsSnapshot>,
+        bin: &Path,
+        timeout_secs: u64,
+    ) -> MetricsSnapshot {
+        let config = CcrCheckConfig {
+            bin: Some(bin.to_path_buf()),
+            timeout_secs,
+            max_per_scan: 2,
+        };
+        scan_run_root_incremental_with_rtr_dir(root, "test", previous, None, &config).expect("scan")
+    }
+
    fn sample_cir() -> Vec<u8> {
        let rejected = vec![CirRejectedObject {
            object_uri: "rsync://repo.example/a/bad.roa".to_string(),
Author	SHA1	Message	Date
yuyr	f86fbd4939	20260721 metrics 服务新增 CCR rpki-client 格式检查指标与 Grafana 面板（#125）	2026-07-21 18:07:05 +08:00
yuyr	4074323302	20260714 补充 sync worker 消融实验远端运行脚本（#115 遗留）	2026-07-21 16:10:09 +08:00
yuyr	b00a5f920c	20260721 rpctl ours RP 显式设置全局 repo sync workers 为 8	2026-07-21 15:55:15 +08:00
yuyr	654b6d1c56	20260713_2 落地升级环境变量优先级契约（#114 主工作树收口）	2026-07-21 15:55:15 +08:00