diff --git a/scripts/verification/observe_remote231_verification_only.sh b/scripts/verification/observe_remote231_verification_only.sh new file mode 100755 index 0000000..98c3d85 --- /dev/null +++ b/scripts/verification/observe_remote231_verification_only.sh @@ -0,0 +1,563 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Observe a Docker-managed soak from the host without joining its compose project. +# The verifier binary must be copied byte-for-byte from the running runtime image; +# validation-contract.json rejects a different binary SHA-256. + +usage() { + cat <<'USAGE' +Usage: + observe_remote231_verification_only.sh \ + --runs-root \ + --state-root \ + --rpki-bin \ + --work-root \ + --webhook-url-file <0600_file> \ + [--target-runs 20] [--delay-secs 30] [--poll-secs 5] \ + [--verification-timeout-secs 480] [--normal-interval-secs 600] \ + [--normal-run-safety-secs 60] [--verification-object-workers 4] \ + [--verification-child-path ] + +This is a host-side, finite observer. On its first start it records the latest +already-terminal normal run and observes only later terminal runs. For each one +it waits DELAY seconds, then uses a non-blocking state lock. It never waits for +or blocks the Docker soak. Successful normal runs are verified with a copied +host-side rpki binary; failed, stale, or lock-busy runs are recorded as skipped. + +The verification timeout is additionally bounded by the source run completion +time plus NORMAL_INTERVAL_SECS minus NORMAL_RUN_SAFETY_SECS. This preserves a +normal-run safety window even after an observer restart or delayed scan. + +Only compact JSON results are retained under WORK_ROOT/results/. The large +verification output and any retained scratch DB are always deleted afterwards. + +When --verification-child-path is supplied, it replaces PATH only for the +verification binary. Passing a directory without rsync (for example +/nonexistent) is an adversarial acceptance check: an accidental SystemRsync +fallback must fail instead of accessing the network. +USAGE +} + +RUNS_ROOT="" +STATE_ROOT="" +RPKI_BIN="" +WORK_ROOT="" +WEBHOOK_URL_FILE="" +TARGET_RUNS=20 +DELAY_SECS=30 +POLL_SECS=5 +VERIFICATION_TIMEOUT_SECS=480 +NORMAL_INTERVAL_SECS=600 +NORMAL_RUN_SAFETY_SECS=60 +VERIFICATION_OBJECT_WORKERS=4 +VERIFICATION_CHILD_PATH="" + +while [[ $# -gt 0 ]]; do + case "$1" in + --runs-root) RUNS_ROOT="${2:?missing value}"; shift 2 ;; + --state-root) STATE_ROOT="${2:?missing value}"; shift 2 ;; + --rpki-bin) RPKI_BIN="${2:?missing value}"; shift 2 ;; + --work-root) WORK_ROOT="${2:?missing value}"; shift 2 ;; + --webhook-url-file) WEBHOOK_URL_FILE="${2:?missing value}"; shift 2 ;; + --target-runs) TARGET_RUNS="${2:?missing value}"; shift 2 ;; + --delay-secs) DELAY_SECS="${2:?missing value}"; shift 2 ;; + --poll-secs) POLL_SECS="${2:?missing value}"; shift 2 ;; + --verification-timeout-secs) VERIFICATION_TIMEOUT_SECS="${2:?missing value}"; shift 2 ;; + --normal-interval-secs) NORMAL_INTERVAL_SECS="${2:?missing value}"; shift 2 ;; + --normal-run-safety-secs) NORMAL_RUN_SAFETY_SECS="${2:?missing value}"; shift 2 ;; + --verification-object-workers) VERIFICATION_OBJECT_WORKERS="${2:?missing value}"; shift 2 ;; + --verification-child-path) VERIFICATION_CHILD_PATH="${2:?missing value}"; shift 2 ;; + -h|--help) usage; exit 0 ;; + *) echo "unknown argument: $1" >&2; usage >&2; exit 2 ;; + esac +done + +for value_name in RUNS_ROOT STATE_ROOT RPKI_BIN WORK_ROOT WEBHOOK_URL_FILE; do + [[ -n "${!value_name}" ]] || { echo "$value_name is required" >&2; exit 2; } +done +for integer_name in TARGET_RUNS DELAY_SECS POLL_SECS VERIFICATION_TIMEOUT_SECS NORMAL_INTERVAL_SECS VERIFICATION_OBJECT_WORKERS; do + [[ "${!integer_name}" =~ ^[0-9]+$ ]] && [[ "${!integer_name}" != 0 ]] || { + echo "$integer_name must be a positive integer" >&2 + exit 2 + } +done +[[ "$NORMAL_RUN_SAFETY_SECS" =~ ^[0-9]+$ ]] || { + echo "NORMAL_RUN_SAFETY_SECS must be a non-negative integer" >&2 + exit 2 +} +(( NORMAL_RUN_SAFETY_SECS < NORMAL_INTERVAL_SECS )) || { + echo "NORMAL_RUN_SAFETY_SECS must be smaller than NORMAL_INTERVAL_SECS" >&2 + exit 2 +} +[[ -d "$RUNS_ROOT" ]] || { echo "runs root is not a directory: $RUNS_ROOT" >&2; exit 2; } +[[ -d "$STATE_ROOT" ]] || { echo "state root is not a directory: $STATE_ROOT" >&2; exit 2; } +[[ -x "$RPKI_BIN" ]] || { echo "rpki binary is not executable: $RPKI_BIN" >&2; exit 2; } +[[ -f "$WEBHOOK_URL_FILE" ]] || { echo "webhook URL file is absent: $WEBHOOK_URL_FILE" >&2; exit 2; } +for required_command in curl flock ionice jq nice python3 sha256sum timeout; do + command -v "$required_command" >/dev/null || { + echo "missing required command: $required_command" >&2 + exit 2 + } +done + +mkdir -p "$WORK_ROOT/results" "$WORK_ROOT/scratch" +STATE_FILE="$WORK_ROOT/observer-state.json" +OBSERVER_LOCK="$WORK_ROOT/observer.lock" +RESULTS_DIR="$WORK_ROOT/results" +SCRATCH_ROOT="$WORK_ROOT/scratch" +WEBHOOK_URL="$(tr -d '\r\n' < "$WEBHOOK_URL_FILE")" +[[ "$WEBHOOK_URL" == https://open.feishu.cn/open-apis/bot/v2/hook/* ]] || { + echo "webhook URL does not have the expected Feishu incoming-webhook form" >&2 + exit 2 +} + +now_utc() { date -u +%Y-%m-%dT%H:%M:%SZ; } +now_ms() { python3 -c 'import time; print(time.time_ns() // 1_000_000)'; } + +remaining_verification_window_secs() { + local completed_at="$1" + python3 - "$completed_at" "$NORMAL_INTERVAL_SECS" "$NORMAL_RUN_SAFETY_SECS" <<'PY' +import datetime +import sys +import time + +completed = datetime.datetime.fromisoformat(sys.argv[1].replace("Z", "+00:00")) +interval = int(sys.argv[2]) +safety = int(sys.argv[3]) +print(int(completed.timestamp()) + interval - safety - int(time.time())) +PY +} + +atomic_replace() { + local destination="$1" + local temporary="${destination}.tmp.$$" + cat > "$temporary" + mv "$temporary" "$destination" +} + +terminal_run_ids() { + local run_dir meta + for run_dir in "$RUNS_ROOT"/run_[0-9]*; do + [[ -d "$run_dir" ]] || continue + meta="$run_dir/run-meta.json" + [[ -f "$meta" ]] || continue + jq -e '(.completed_at_rfc3339_utc? // .completedAt? // null) != null' "$meta" >/dev/null 2>&1 || continue + basename "$run_dir" + done | sort -t_ -k2,2n +} + +run_number() { + local run_id="$1" + printf '%d\n' "$((10#${run_id#run_}))" +} + +latest_terminal_run_id() { + terminal_run_ids | tail -n 1 +} + +result_path() { + printf '%s/%s.json\n' "$RESULTS_DIR" "$1" +} + +read_json_or_empty() { + local path="$1" + if [[ -f "$path" ]] && jq -e . "$path" >/dev/null 2>&1; then + jq -c . "$path" + else + printf '{}' + fi +} + +update_cumulative() { + local observed normal_success passes mismatches errors skipped pending latest + local -a result_files + shopt -s nullglob + result_files=("$RESULTS_DIR"/run_*.json) + shopt -u nullglob + observed="${#result_files[@]}" + if (( observed == 0 )); then + normal_success=0 + passes=0 + mismatches=0 + errors=0 + skipped=0 + pending=0 + else + normal_success="$(jq -r 'select(.normalRun.status == "success") | 1' "${result_files[@]}" | wc -l)" + passes="$(jq -r 'select(.verification.status == "pass") | 1' "${result_files[@]}" | wc -l)" + mismatches="$(jq -r 'select(.verification.status == "mismatch") | 1' "${result_files[@]}" | wc -l)" + errors="$(jq -r 'select(.verification.status == "error") | 1' "${result_files[@]}" | wc -l)" + skipped="$(jq -r 'select(.verification.status | startswith("skipped_")) | 1' "${result_files[@]}" | wc -l)" + pending="$(jq -r 'select(.notification.status != "sent") | 1' "${result_files[@]}" | wc -l)" + fi + latest="$(find "$RESULTS_DIR" -maxdepth 1 -type f -name 'run_*.json' -printf '%f\n' | sed 's/.json$//' | sort -t_ -k2,2n | tail -n 1)" + jq -n \ + --arg updated_at "$(now_utc)" \ + --arg initial_run_id "$(jq -r '.initialLatestTerminalRunId' "$STATE_FILE")" \ + --arg latest_run_id "$latest" \ + --argjson target_runs "$TARGET_RUNS" \ + --argjson observed_runs "$observed" \ + --argjson normal_success_runs "$normal_success" \ + --argjson verification_passes "$passes" \ + --argjson verification_mismatches "$mismatches" \ + --argjson verification_errors "$errors" \ + --argjson verification_skipped "$skipped" \ + --argjson notification_pending "$pending" \ + '{schemaVersion: 1, updatedAt: $updated_at, initialLatestTerminalRunId: $initial_run_id, + latestProcessedRunId: $latest_run_id, targetRuns: $target_runs, observedRuns: $observed_runs, + normalSuccessRuns: $normal_success_runs, verificationPasses: $verification_passes, + verificationMismatches: $verification_mismatches, verificationErrors: $verification_errors, + verificationSkipped: $verification_skipped, notificationPending: $notification_pending}' \ + | atomic_replace "$RESULTS_DIR/cumulative.json" +} + +notify_result() { + local path="$1" run_id status observed target passes mismatches errors skipped analysis_available top_phase message payload response + run_id="$(jq -r '.runId' "$path")" + status="$(jq -r '.verification.status' "$path")" + observed="$(jq -r '.observedRuns' "$RESULTS_DIR/cumulative.json")" + target="$(jq -r '.targetRuns' "$RESULTS_DIR/cumulative.json")" + passes="$(jq -r '.verificationPasses' "$RESULTS_DIR/cumulative.json")" + mismatches="$(jq -r '.verificationMismatches' "$RESULTS_DIR/cumulative.json")" + errors="$(jq -r '.verificationErrors' "$RESULTS_DIR/cumulative.json")" + skipped="$(jq -r '.verificationSkipped' "$RESULTS_DIR/cumulative.json")" + analysis_available="$(jq -r '.verification.analysis.available // false' "$path")" + top_phase="$(jq -r '(.verification.analysis.phaseTotals // [])[0] | if . == null then "n/a" else "\(.name)=\(.totalMs)ms" end' "$path")" + message="From codex: #137 远端231 verification-only 旁路复核\n${run_id}: ${status}\n" + message+="CCR state digest match=$(jq -r 'if .verification.comparison | has("stateDigestMatch") then (.verification.comparison.stateDigestMatch | tostring) else "n/a" end' "$path"); fallback replay match=$(jq -r 'if .verification.comparison.fallbackReplaySelection | has("matches") then (.verification.comparison.fallbackReplaySelection.matches | tostring) else "n/a" end' "$path"); verificationProcessMs=$(jq -r '.verification.timing.verificationProcessMs // .verification.wallMs // "n/a"' "$path"); preValidationPreparationMs=$(jq -r '.verification.timing.preValidationPreparationMs // "n/a"' "$path"); lockWaitMs=$(jq -r '.verification.lockWaitMs // "n/a"' "$path")\n" + message+="analysisAvailable=${analysis_available}; topPhase=${top_phase}\n" + message+="累计 ${observed}/${target}: pass=${passes}, mismatch=${mismatches}, error=${errors}, skipped=${skipped}\n" + message+="结果: ${path}" + payload="$(jq -n --arg text "$message" '{msg_type: "text", content: {text: $text}}')" + if response="$(curl --fail-with-body --silent --show-error --connect-timeout 10 --max-time 30 \ + -H 'Content-Type: application/json' --data "$payload" "$WEBHOOK_URL" 2>&1)"; then + if jq -e '((.code // .StatusCode // 0) == 0)' <<< "$response" >/dev/null 2>&1; then + jq --arg sent_at "$(now_utc)" --arg response "$response" \ + '.notification = {status: "sent", sentAt: $sent_at, response: $response}' "$path" \ + | atomic_replace "$path" + return 0 + fi + response="Feishu rejected payload: $response" + fi + jq --arg attempted_at "$(now_utc)" --arg error "$response" \ + '.notification = {status: "pending", lastAttemptAt: $attempted_at, lastError: $error}' "$path" \ + | atomic_replace "$path" + return 1 +} + +deliver_pending_notifications() { + local path + for path in "$RESULTS_DIR"/run_*.json; do + [[ -f "$path" ]] || continue + [[ "$(jq -r '.notification.status // "pending"' "$path")" == "sent" ]] && continue + update_cumulative + notify_result "$path" || true + done + update_cumulative +} + +write_result() { + local run_id="$1" normal_json="$2" verification_json="$3" result + result="$(result_path "$run_id")" + jq -n \ + --arg recorded_at "$(now_utc)" \ + --arg run_id "$run_id" \ + --argjson normal_run "$normal_json" \ + --argjson verification "$verification_json" \ + '{schemaVersion: 1, recordedAt: $recorded_at, runId: $run_id, normalRun: $normal_run, + verification: $verification, notification: {status: "pending"}}' \ + | atomic_replace "$result" + update_cumulative + notify_result "$result" || true + update_cumulative +} + +normal_run_summary() { + local run_dir="$1" + if [[ ! -f "$run_dir/run-meta.json" ]]; then + printf '{"status":"invalid-meta"}' + return + fi + jq -cn --slurpfile meta "$run_dir/run-meta.json" --slurpfile summary "$run_dir/run-summary.json" \ + '{status: ($meta[0].status // "unknown"), syncMode: ($meta[0].sync_mode // $meta[0].syncMode // "unknown"), + runIndex: ($meta[0].run_index // $meta[0].runIndex // null), + completedAt: ($meta[0].completed_at_rfc3339_utc // $meta[0].completedAt // null), + wallMs: ($summary[0].wall_ms // $summary[0].wallMs // null), + exitCode: ($summary[0].exit_code // $summary[0].exitCode // $meta[0].daemon_exit_code // null)}' 2>/dev/null \ + || printf '{"status":"invalid-meta"}' +} + +process_terminal_run() { + local run_id="$1" run_dir="$RUNS_ROOT/$run_id" normal_json status temporary_dir state_lock_fd lock_started_ms lock_acquired_ms + local binding_run_id binary_sha source_sha run_started_ms run_finished_ms exit_code comparison_json comparison_state_digest_match fallback_replay_comparison_json fallback_replay_selection_match verification_meta_json stage_timing_json analysis_json analysis_summary_json error_tail max_rss_kb verification_status verification_json timeout_secs completed_at + local -a verification_child_exec + local observer_started_ms delay_finished_ms preparation_started_ms binding_started_ms binding_finished_ms binary_check_started_ms binary_check_finished_ms window_started_ms window_finished_ms scratch_cleanup_started_ms scratch_cleanup_finished_ms post_validation_started_ms post_validation_collection_finished_ms result_cleanup_started_ms result_cleanup_finished_ms + normal_json="$(normal_run_summary "$run_dir")" + status="$(jq -r '.status' <<< "$normal_json")" + observer_started_ms="$(now_ms)" + sleep "$DELAY_SECS" + delay_finished_ms="$(now_ms)" + if [[ "$status" != "success" ]]; then + write_result "$run_id" "$normal_json" '{"status":"skipped_normal_run_not_success"}' + return + fi + + preparation_started_ms="$delay_finished_ms" + mkdir -p "$STATE_ROOT/locks" + exec {state_lock_fd}>"$STATE_ROOT/locks/state-execution.lock" + lock_started_ms="$(now_ms)" + if ! flock -n "$state_lock_fd"; then + exec {state_lock_fd}>&- + write_result "$run_id" "$normal_json" '{"status":"skipped_state_lock_busy"}' + return + fi + lock_acquired_ms="$(now_ms)" + binding_started_ms="$(now_ms)" + binding_run_id="$(jq -r '.currentRunId // empty' "$STATE_ROOT/meta/current-state-binding.json" 2>/dev/null || true)" + binding_finished_ms="$(now_ms)" + if [[ "$binding_run_id" != "$run_id" ]]; then + flock -u "$state_lock_fd" + exec {state_lock_fd}>&- + verification_json="$(jq -n --arg bound_run "$binding_run_id" --argjson lock_wait_ms "$((lock_acquired_ms - lock_started_ms))" \ + '{status: "skipped_source_no_longer_current", boundRunId: $bound_run, lockWaitMs: $lock_wait_ms}')" + write_result "$run_id" "$normal_json" "$verification_json" + return + fi + binary_check_started_ms="$(now_ms)" + source_sha="$(jq -r '.binarySha256 // empty' "$run_dir/validation-contract.json" 2>/dev/null || true)" + binary_sha="$(sha256sum "$RPKI_BIN" | awk '{print $1}')" + binary_check_finished_ms="$(now_ms)" + if [[ -z "$source_sha" || "$source_sha" != "$binary_sha" ]]; then + flock -u "$state_lock_fd" + exec {state_lock_fd}>&- + verification_json="$(jq -n --arg expected "$source_sha" --arg actual "$binary_sha" --argjson lock_wait_ms "$((lock_acquired_ms - lock_started_ms))" \ + '{status: "error_binary_sha256_mismatch", expectedBinarySha256: $expected, actualBinarySha256: $actual, lockWaitMs: $lock_wait_ms}')" + write_result "$run_id" "$normal_json" "$verification_json" + return + fi + + window_started_ms="$(now_ms)" + completed_at="$(jq -r '.completedAt // empty' <<< "$normal_json")" + timeout_secs="$(remaining_verification_window_secs "$completed_at")" + window_finished_ms="$(now_ms)" + if (( timeout_secs <= 0 )); then + flock -u "$state_lock_fd" + exec {state_lock_fd}>&- + verification_json="$(jq -n --arg completed_at "$completed_at" --argjson lock_wait_ms "$((lock_acquired_ms - lock_started_ms))" \ + '{status: "skipped_insufficient_interval", sourceCompletedAt: $completed_at, lockWaitMs: $lock_wait_ms}')" + write_result "$run_id" "$normal_json" "$verification_json" + return + fi + if (( timeout_secs > VERIFICATION_TIMEOUT_SECS )); then + timeout_secs="$VERIFICATION_TIMEOUT_SECS" + fi + + temporary_dir="$SCRATCH_ROOT/$run_id" + scratch_cleanup_started_ms="$(now_ms)" + rm -rf "$temporary_dir" + scratch_cleanup_finished_ms="$(now_ms)" + run_started_ms="$(now_ms)" + verification_child_exec=() + if [[ -n "$VERIFICATION_CHILD_PATH" ]]; then + verification_child_exec=(/usr/bin/env "PATH=$VERIFICATION_CHILD_PATH") + fi + set +e + RPKI_STATE_EXECUTION_LOCK_HELD=1 /usr/bin/time -v -o "$temporary_dir.process-time" -- \ + /usr/bin/timeout --foreground --signal=TERM --kill-after=30s "$timeout_secs" \ + /usr/bin/ionice -c2 -n0 "${verification_child_exec[@]}" "$RPKI_BIN" \ + --verification-only --verification-source-run "$run_dir" \ + --verification-source-state "$STATE_ROOT" --verification-out "$temporary_dir" \ + --analyze --analysis-out "$temporary_dir/analyze" \ + --parallel-phase2-object-workers "$VERIFICATION_OBJECT_WORKERS" \ + >"$temporary_dir.stdout" 2>"$temporary_dir.stderr" + exit_code=$? + set -e + run_finished_ms="$(now_ms)" + post_validation_started_ms="$run_finished_ms" + flock -u "$state_lock_fd" + exec {state_lock_fd}>&- + + comparison_json="$(read_json_or_empty "$temporary_dir/compare/ccr-state-digest.json")" + fallback_replay_comparison_json="$(read_json_or_empty "$temporary_dir/compare/fallback-replay-selection.json")" + verification_meta_json="$(read_json_or_empty "$temporary_dir/verification-meta.json")" + stage_timing_json="$(read_json_or_empty "$temporary_dir/stage-timing.json")" + analysis_json="$(read_json_or_empty "$temporary_dir/analyze/timing.json")" + analysis_summary_json="$(jq -cn --argjson stage "$stage_timing_json" --argjson analysis "$analysis_json" ' + {enabled: true, + available: (($analysis | type) == "object" and ($analysis | length) > 0), + stageTiming: { + totalMs: ($stage.total_ms // null), + validationMs: ($stage.validation_ms // null), + repoSyncMsTotal: ($stage.repo_sync_ms_total // null), + publicationPointRepoSyncMsTotal: ($stage.publication_point_repo_sync_ms_total // null), + rrdpDownloadMsTotal: ($stage.rrdp_download_ms_total // null), + rsyncDownloadMsTotal: ($stage.rsync_download_ms_total // null), + reportBuildMs: ($stage.report_build_ms // null), + reportWriteMs: ($stage.report_write_ms // null), + ccrBuildMs: ($stage.ccr_build_ms // null), + ccrWriteMs: ($stage.ccr_write_ms // null), + cirTotalMs: ($stage.cir_total_ms // null) + }, + phaseTotals: (($analysis.phases // {}) + | to_entries + | map({name: .key, count: (.value.count // 0), totalMs: (((.value.total_nanos // 0) / 1000000) | floor)}) + | sort_by(-.totalMs) + | .[0:20]), + topPublicationPoints: (($analysis.top_publication_points // [])[0:10]), + topPublicationPointSteps: (($analysis.top_publication_point_steps // [])[0:10]), + topRrdpRepos: (($analysis.top_rrdp_repos // [])[0:10]), + topRrdpRepoSteps: (($analysis.top_rrdp_repo_steps // [])[0:10])}')" + error_tail="$(tail -n 20 "$temporary_dir.stderr" 2>/dev/null || true)" + max_rss_kb="$(awk -F: '/Maximum resident set size/{gsub(/^[[:space:]]+/, "", $2); print $2}' "$temporary_dir.process-time" 2>/dev/null || true)" + comparison_state_digest_match="$(jq -r 'if has("stateDigestMatch") then (.stateDigestMatch | tostring) else "missing" end' <<< "$comparison_json")" + fallback_replay_selection_match="$(jq -r 'if has("matches") then (.matches | tostring) else "missing" end' <<< "$fallback_replay_comparison_json")" + if [[ "$comparison_state_digest_match" == "true" && "$fallback_replay_selection_match" == "true" && "$exit_code" -eq 0 ]]; then + verification_status="pass" + elif [[ "$comparison_state_digest_match" == "false" || "$fallback_replay_selection_match" == "false" ]]; then + verification_status="mismatch" + else + verification_status="error" + fi + post_validation_collection_finished_ms="$(now_ms)" + result_cleanup_started_ms="$(now_ms)" + rm -rf "$temporary_dir" "$temporary_dir.stdout" "$temporary_dir.stderr" "$temporary_dir.process-time" + result_cleanup_finished_ms="$(now_ms)" + verification_json="$(jq -n \ + --arg status "$verification_status" \ + --arg verification_child_path "$VERIFICATION_CHILD_PATH" \ + --argjson exit_code "$exit_code" \ + --argjson wall_ms "$((run_finished_ms - run_started_ms))" \ + --argjson lock_wait_ms "$((lock_acquired_ms - lock_started_ms))" \ + --argjson timeout_limit_secs "$timeout_secs" \ + --argjson object_workers "$VERIFICATION_OBJECT_WORKERS" \ + --arg max_rss_kb "${max_rss_kb:-}" \ + --arg error_tail "$error_tail" \ + --argjson comparison "$comparison_json" \ + --argjson fallback_replay_comparison "$fallback_replay_comparison_json" \ + --argjson meta "$verification_meta_json" \ + --argjson analysis "$analysis_summary_json" \ + --argjson post_run_delay_ms "$((delay_finished_ms - observer_started_ms))" \ + --argjson observer_to_verification_start_ms "$((run_started_ms - observer_started_ms))" \ + --argjson pre_validation_preparation_ms "$((run_started_ms - preparation_started_ms))" \ + --argjson binding_check_ms "$((binding_finished_ms - binding_started_ms))" \ + --argjson binary_contract_check_ms "$((binary_check_finished_ms - binary_check_started_ms))" \ + --argjson interval_window_calculation_ms "$((window_finished_ms - window_started_ms))" \ + --argjson scratch_preparation_cleanup_ms "$((scratch_cleanup_finished_ms - scratch_cleanup_started_ms))" \ + --argjson verification_process_ms "$((run_finished_ms - run_started_ms))" \ + --argjson post_validation_collection_ms "$((post_validation_collection_finished_ms - post_validation_started_ms))" \ + --argjson result_cleanup_ms "$((result_cleanup_finished_ms - result_cleanup_started_ms))" \ + '{status: $status, exitCode: $exit_code, wallMs: $wall_ms, lockWaitMs: $lock_wait_ms, + timeoutLimitSecs: $timeout_limit_secs, objectWorkers: $object_workers, + verificationChildPath: (if $verification_child_path == "" then null else $verification_child_path end), + timing: {postRunDelayMs: $post_run_delay_ms, + observerToVerificationStartMs: $observer_to_verification_start_ms, + preValidationPreparationMs: $pre_validation_preparation_ms, + preparationBreakdownMs: {stateLockWaitMs: $lock_wait_ms, + bindingCheckMs: $binding_check_ms, + binaryContractCheckMs: $binary_contract_check_ms, + intervalWindowCalculationMs: $interval_window_calculation_ms, + scratchPreparationCleanupMs: $scratch_preparation_cleanup_ms}, + verificationProcessMs: $verification_process_ms, + postValidationCollectionMs: $post_validation_collection_ms, + resultCleanupMs: $result_cleanup_ms}, + maxRssKb: (if $max_rss_kb == "" then null else ($max_rss_kb | tonumber) end), + comparison: {stateDigestMatch: ($comparison | if has("stateDigestMatch") then .stateDigestMatch else null end), + mismatchedStates: ($comparison.mismatchedStates // []), + states: (($comparison.states // []) | map({name, sourcePresent, verificationPresent, + sourceHashHex, verificationHashHex, matches})), + fallbackReplaySelection: { + matches: ($fallback_replay_comparison | if has("matches") then .matches else null end), + expectedCount: (($fallback_replay_comparison.expected // []) | length), + actualCount: (($fallback_replay_comparison.actual // []) | length), + missing: ($fallback_replay_comparison.missing // []), + unexpected: ($fallback_replay_comparison.unexpected // []) + }}, + verificationMeta: {status: ($meta.status // null), wallMs: ($meta.wallMs // null), + repositoryObjectsChecked: ($meta.repositoryObjectsChecked // null), + repositoryBytesChecked: ($meta.repositoryBytesChecked // null)}, + analysis: $analysis, + errorTail: $error_tail}')" + write_result "$run_id" "$normal_json" "$verification_json" +} + +exec {observer_lock_fd}>"$OBSERVER_LOCK" +if ! flock -n "$observer_lock_fd"; then + echo "another observer already owns $OBSERVER_LOCK" >&2 + exit 3 +fi + +find "$SCRATCH_ROOT" -mindepth 1 -maxdepth 1 -exec rm -rf {} + + +if [[ ! -f "$STATE_FILE" ]]; then + initial_run_id="$(latest_terminal_run_id)" + [[ -n "$initial_run_id" ]] || { echo "no terminal normal run found under $RUNS_ROOT" >&2; exit 2; } + jq -n \ + --arg started_at "$(now_utc)" \ + --arg initial_run_id "$initial_run_id" \ + --arg runs_root "$RUNS_ROOT" \ + --arg state_root "$STATE_ROOT" \ + --arg rpki_sha256 "$(sha256sum "$RPKI_BIN" | awk '{print $1}')" \ + --argjson target_runs "$TARGET_RUNS" \ + --argjson delay_secs "$DELAY_SECS" \ + --argjson verification_timeout_secs "$VERIFICATION_TIMEOUT_SECS" \ + --argjson normal_interval_secs "$NORMAL_INTERVAL_SECS" \ + --argjson normal_run_safety_secs "$NORMAL_RUN_SAFETY_SECS" \ + --argjson verification_object_workers "$VERIFICATION_OBJECT_WORKERS" \ + --arg verification_child_path "$VERIFICATION_CHILD_PATH" \ + '{schemaVersion: 1, startedAt: $started_at, initialLatestTerminalRunId: $initial_run_id, + runsRoot: $runs_root, stateRoot: $state_root, verifierBinarySha256: $rpki_sha256, + targetRuns: $target_runs, delaySecs: $delay_secs, verificationTimeoutSecs: $verification_timeout_secs, + normalIntervalSecs: $normal_interval_secs, normalRunSafetySecs: $normal_run_safety_secs, + verificationObjectWorkers: $verification_object_workers, + verificationChildPath: (if $verification_child_path == "" then null else $verification_child_path end), + verificationAnalyze: true}' \ + | atomic_replace "$STATE_FILE" + update_cumulative +fi + +# Preserve the original baseline run while recording the live configuration after +# an observer restart. This is particularly important for the normal-run safety +# window, whose length depends on the soak interval. +jq --arg updated_at "$(now_utc)" \ + --argjson target_runs "$TARGET_RUNS" \ + --argjson delay_secs "$DELAY_SECS" \ + --argjson verification_timeout_secs "$VERIFICATION_TIMEOUT_SECS" \ + --argjson normal_interval_secs "$NORMAL_INTERVAL_SECS" \ + --argjson normal_run_safety_secs "$NORMAL_RUN_SAFETY_SECS" \ + --argjson verification_object_workers "$VERIFICATION_OBJECT_WORKERS" \ + --arg verification_child_path "$VERIFICATION_CHILD_PATH" \ + '.updatedAt = $updated_at + | .targetRuns = $target_runs + | .delaySecs = $delay_secs + | .verificationTimeoutSecs = $verification_timeout_secs + | .normalIntervalSecs = $normal_interval_secs + | .normalRunSafetySecs = $normal_run_safety_secs + | .verificationObjectWorkers = $verification_object_workers + | .verificationChildPath = (if $verification_child_path == "" then null else $verification_child_path end) + | .verificationAnalyze = true' "$STATE_FILE" \ + | atomic_replace "$STATE_FILE" + +initial_run_number="$(run_number "$(jq -r '.initialLatestTerminalRunId' "$STATE_FILE")")" +while true; do + deliver_pending_notifications + observed_count="$(find "$RESULTS_DIR" -maxdepth 1 -type f -name 'run_*.json' -printf '%f\n' | wc -l)" + pending_count="$(jq -r '.notificationPending' "$RESULTS_DIR/cumulative.json")" + if (( observed_count >= TARGET_RUNS && pending_count == 0 )); then + break + fi + if (( observed_count < TARGET_RUNS )); then + while IFS= read -r run_id; do + [[ -n "$run_id" ]] || continue + (( $(run_number "$run_id") > initial_run_number )) || continue + [[ -f "$(result_path "$run_id")" ]] && continue + process_terminal_run "$run_id" + observed_count="$(find "$RESULTS_DIR" -maxdepth 1 -type f -name 'run_*.json' -printf '%f\n' | wc -l)" + (( observed_count < TARGET_RUNS )) || break + done < <(terminal_run_ids) + fi + sleep "$POLL_SECS" +done + +update_cumulative +echo "completed $(jq -r '.observedRuns' "$RESULTS_DIR/cumulative.json") observed runs; results: $RESULTS_DIR/cumulative.json" diff --git a/src/cli.rs b/src/cli.rs index ddd6620..a33d74b 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -2173,6 +2173,36 @@ where } } +fn policy_for_verification_contract( + contract: &crate::verification_only::ValidationContract, +) -> crate::policy::Policy { + // Preserve every contracted policy field, including sync_preference. The + // verification fetchers themselves are network-disabled; changing this + // field would change the VCIR reuse-identity fingerprint and falsely make + // a contract-selected fallback ineligible. + let mut policy = contract.policy.clone(); + policy.ca_failed_fetch_policy = if contract.fallback_replay_selections.is_empty() { + crate::policy::CaFailedFetchPolicy::StopAllOutput + } else { + crate::policy::CaFailedFetchPolicy::ReuseCurrentInstanceVcir + }; + policy.verification_forced_vcir_reuse_manifest_uris = contract + .fallback_replay_selections + .iter() + .map(|selection| selection.manifest_rsync_uri.clone()) + .collect(); + policy +} + +fn transport_prefetch_for_verification_contract( + contract: &crate::verification_only::ValidationContract, +) -> bool { + // This only restores the source run's transport scheduling and request + // deduplication topology. It does not enable validation-result reuse or + // permit network access. + contract.cache.transport_prefetch +} + pub fn run(argv: &[String]) -> Result<(), String> { let mut args = parse_args(argv)?; let mut prepared_verification = if args.verification_only { @@ -2199,7 +2229,9 @@ pub fn run(argv: &[String]) -> Result<(), String> { args.enable_publication_point_validation_cache = false; args.crypto_signature_cache_observe_only = false; args.enable_crypto_signature_cache = false; - args.enable_transport_request_prefetch = false; + args.enable_transport_request_prefetch = + transport_prefetch_for_verification_contract(&prepared.contract); + args.rsync_scope_policy = prepared.contract.rsync_scope_policy; args.ccr_out_path = Some(prepared.artifacts.result_ccr.clone()); args.vrps_csv_out_path = Some(prepared.artifacts.vrps_csv.clone()); args.vaps_csv_out_path = Some(prepared.artifacts.vaps_csv.clone()); @@ -2219,7 +2251,7 @@ pub fn run(argv: &[String]) -> Result<(), String> { }; let mut policy = if let Some(prepared) = prepared_verification.as_ref() { - prepared.contract.policy.clone() + policy_for_verification_contract(&prepared.contract) } else { read_policy(args.policy_path.as_deref())? }; @@ -2233,9 +2265,6 @@ pub fn run(argv: &[String]) -> Result<(), String> { if args.disable_rrdp { policy.sync_preference = crate::policy::SyncPreference::RsyncOnly; } - } else { - policy.sync_preference = crate::policy::SyncPreference::RsyncOnly; - policy.ca_failed_fetch_policy = crate::policy::CaFailedFetchPolicy::StopAllOutput; } let validation_time = args .validation_time @@ -2243,22 +2272,6 @@ pub fn run(argv: &[String]) -> Result<(), String> { let validation_time = time::OffsetDateTime::from_unix_timestamp(validation_time.unix_timestamp()) .map_err(|error| format!("normalize validation time failed: {error}"))?; - if let Some(path) = args.validation_contract_out_path.as_deref() { - let contract = crate::verification_only::ValidationContract::for_current_binary( - validation_time, - policy.clone(), - args.max_ca_depth, - args.max_instances, - crate::verification_only::ValidationCacheContract { - publication_point: args.enable_publication_point_validation_cache, - roa: args.enable_roa_validation_cache, - child_certificate: args.enable_child_certificate_validation_cache, - transport_prefetch: args.enable_transport_request_prefetch, - crypto_signature: args.enable_crypto_signature_cache, - }, - )?; - crate::verification_only::write_validation_contract(path, &contract)?; - } let http_root_certificates_pem = args .http_root_cert_paths .iter() @@ -2375,24 +2388,24 @@ pub fn run(argv: &[String]) -> Result<(), String> { store.as_ref(), ); let validation_started = std::time::Instant::now(); - let crypto_sig_cache = if args.crypto_signature_cache_observe_only - || args.enable_crypto_signature_cache - { - let cache_file = crate::crypto_sig_cache::default_cache_file_path(&args.db_path); - let cache = Arc::new(crate::crypto_sig_cache::CryptoSigCache::load_or_rebuild( - cache_file, - args.enable_crypto_signature_cache, - )); - crate::crypto_sig_cache::install_global(Arc::clone(&cache)); - Some(cache) - } else { - None - }; + let crypto_sig_cache = + if args.crypto_signature_cache_observe_only || args.enable_crypto_signature_cache { + let cache_file = crate::crypto_sig_cache::default_cache_file_path(&args.db_path); + let cache = Arc::new(crate::crypto_sig_cache::CryptoSigCache::load_or_rebuild( + cache_file, + args.enable_crypto_signature_cache, + )); + crate::crypto_sig_cache::install_global(Arc::clone(&cache)); + Some(cache) + } else { + None + }; let collect_current_repo_objects = false; let out = if args.verification_only { let http = crate::verification_only::NetworkDisabledHttpFetcher; let rsync = crate::fetch::current_repository::CurrentRepositoryViewRsyncFetcher::new( Arc::clone(&store), + args.rsync_scope_policy, ); run_online_validation_with_fetchers( Arc::clone(&store), @@ -2828,7 +2841,8 @@ pub fn run(argv: &[String]) -> Result<(), String> { crate::crypto_sig_cache::clear_global(); cache.summary() }); - if let (Some((_, t)), Some(summary)) = (timing.as_ref(), crypto_signature_cache_observe.as_ref()) + if let (Some((_, t)), Some(summary)) = + (timing.as_ref(), crypto_signature_cache_observe.as_ref()) { let (mut calls, mut would_hit, mut new_keys, mut executed, mut skipped) = (0u64, 0u64, 0u64, 0u64, 0u64); @@ -2919,6 +2933,36 @@ pub fn run(argv: &[String]) -> Result<(), String> { .or(args.vrps_csv_out_path.as_deref()); write_stage_timing(stage_timing_anchor_path, &stage_timing)?; + // A normal run's contract is intentionally finalized only after its + // validation outputs have all been produced successfully. This prevents a + // failed/incomplete run from leaving a contract that appears eligible for + // verification-only, and captures the fallback set actually exercised by + // the completed traversal rather than a prediction made before it starts. + if !args.verification_only { + if let Some(path) = args.validation_contract_out_path.as_deref() { + let mut contract = crate::verification_only::ValidationContract::for_current_binary( + validation_time, + policy.clone(), + args.max_ca_depth, + args.max_instances, + crate::verification_only::ValidationCacheContract { + publication_point: args.enable_publication_point_validation_cache, + roa: args.enable_roa_validation_cache, + child_certificate: args.enable_child_certificate_validation_cache, + transport_prefetch: args.enable_transport_request_prefetch, + crypto_signature: args.enable_crypto_signature_cache, + }, + args.rsync_scope_policy, + )?; + contract.fallback_replay_selections = + crate::verification_only::collect_fallback_replay_selections( + store.as_ref(), + shared.publication_points.as_ref(), + )?; + crate::verification_only::write_validation_contract(path, &contract)?; + } + } + if let Some((out_dir, t)) = timing.as_ref() { t.record_count("vrps", shared.vrps.len() as u64); t.record_count("aspas", shared.aspas.len() as u64); @@ -2981,7 +3025,26 @@ pub fn run(argv: &[String]) -> Result<(), String> { Some(&prepared.artifacts.compare_dir.join("ccr-state-digest.md")), &comparison, )?; - let scratch_retained = !comparison.state_digest_match || args.keep_verification_scratch; + let actual_fallback_replay_selections = + crate::verification_only::collect_fallback_replay_selections( + store.as_ref(), + shared.publication_points.as_ref(), + )?; + let fallback_replay_selection_comparison = + crate::verification_only::compare_fallback_replay_selections( + &prepared.contract.fallback_replay_selections, + &actual_fallback_replay_selections, + ); + crate::verification_only::write_fallback_replay_selection_comparison( + &prepared + .artifacts + .compare_dir + .join("fallback-replay-selection.json"), + &fallback_replay_selection_comparison, + )?; + let verification_match = + comparison.state_digest_match && fallback_replay_selection_comparison.matches; + let scratch_retained = !verification_match || args.keep_verification_scratch; let source_run_id = prepared .artifacts .source_run_dir @@ -2993,7 +3056,7 @@ pub fn run(argv: &[String]) -> Result<(), String> { &prepared.artifacts.verification_meta, &crate::verification_only::VerificationRunMeta { schema_version: 1, - status: if comparison.state_digest_match { + status: if verification_match { "success".to_string() } else { "mismatch".to_string() @@ -3004,14 +3067,27 @@ pub fn run(argv: &[String]) -> Result<(), String> { repository_objects_checked: prepared.blob_summary.current_objects, repository_bytes_checked: prepared.blob_summary.bytes_verified, state_digest_match: comparison.state_digest_match, + fallback_replay_selection_match: fallback_replay_selection_comparison.matches, + fallback_replay_materialization: prepared.fallback_replay_materialization.clone(), scratch_retained, }, )?; - if !comparison.state_digest_match { - return Err(format!( - "verification-only CCR state digest mismatch: {}", - comparison.mismatched_states.join(",") - )); + if !verification_match { + let mut mismatch_parts = Vec::new(); + if !comparison.state_digest_match { + mismatch_parts.push(format!( + "CCR state digest mismatch: {}", + comparison.mismatched_states.join(",") + )); + } + if !fallback_replay_selection_comparison.matches { + mismatch_parts.push(format!( + "fallback replay selection mismatch: missing={}, unexpected={}", + fallback_replay_selection_comparison.missing.len(), + fallback_replay_selection_comparison.unexpected.len() + )); + } + return Err(format!("verification-only {}", mismatch_parts.join("; "))); } if !args.keep_verification_scratch { drop(shared); diff --git a/src/cli/tests.rs b/src/cli/tests.rs index ec3d85e..4253ea5 100644 --- a/src/cli/tests.rs +++ b/src/cli/tests.rs @@ -108,6 +108,73 @@ fn parse_accepts_normal_validation_contract_output() { ); } +#[test] +fn verification_policy_preserves_contract_identity_and_selects_only_bound_fallbacks() { + let manifest_rsync_uri = "rsync://example.test/repo/current.mft".to_string(); + let contract = crate::verification_only::ValidationContract { + schema_version: crate::verification_only::VALIDATION_CONTRACT_SCHEMA_VERSION, + validation_time: "2026-07-16T00:00:00Z".to_string(), + binary_sha256: "11".repeat(32), + policy: Policy { + sync_preference: crate::policy::SyncPreference::RrdpThenRsync, + ..Policy::default() + }, + max_ca_depth: 32, + max_instances: None, + cache: crate::verification_only::ValidationCacheContract { + publication_point: true, + roa: false, + child_certificate: false, + transport_prefetch: false, + crypto_signature: false, + }, + rsync_scope_policy: crate::fetch::rsync_system::RsyncScopePolicy::ModuleRoot, + fallback_replay_selections: vec![crate::verification_only::FallbackReplaySelection { + manifest_rsync_uri: manifest_rsync_uri.clone(), + vcir_sha256: "22".repeat(32), + reuse_identity_sha256: "33".repeat(32), + }], + }; + + let policy = policy_for_verification_contract(&contract); + assert_eq!( + policy.sync_preference, + crate::policy::SyncPreference::RrdpThenRsync, + "sync preference is part of the stored VCIR reuse-identity fingerprint" + ); + assert_eq!( + policy.ca_failed_fetch_policy, + crate::policy::CaFailedFetchPolicy::ReuseCurrentInstanceVcir + ); + assert!( + policy + .verification_forced_vcir_reuse_manifest_uris + .contains(&manifest_rsync_uri) + ); + assert!( + !serde_json::to_string(&policy) + .expect("serialize policy") + .contains("verification_forced_vcir_reuse_manifest_uris"), + "the runtime-only override must not perturb the contracted policy fingerprint" + ); + + assert!( + !transport_prefetch_for_verification_contract(&contract), + "verification-only must preserve a disabled source transport prefetch setting" + ); + let mut contract_with_prefetch = contract; + contract_with_prefetch.cache.transport_prefetch = true; + assert_eq!( + contract_with_prefetch.rsync_scope_policy, + crate::fetch::rsync_system::RsyncScopePolicy::ModuleRoot, + "transport scope is part of the source contract" + ); + assert!( + transport_prefetch_for_verification_contract(&contract_with_prefetch), + "verification-only must preserve a source transport prefetch setting" + ); +} + #[test] fn parse_rejects_both_tal_url_and_tal_path() { let argv = vec![ diff --git a/src/fetch/current_repository.rs b/src/fetch/current_repository.rs index 5e494b2..55034a1 100644 --- a/src/fetch/current_repository.rs +++ b/src/fetch/current_repository.rs @@ -1,24 +1,33 @@ use std::sync::Arc; -use crate::fetch::rsync::{ - RsyncFetchError, RsyncFetchResult, RsyncFetcher, normalize_rsync_base_uri, +use crate::fetch::rsync::{RsyncFetchError, RsyncFetchResult, RsyncFetcher}; +use crate::fetch::rsync_system::{ + RsyncScopePolicy, scoped_rsync_failure_dedup_key, scoped_rsync_fetch_uri, }; use crate::storage::{RepositoryViewState, RocksStore}; #[derive(Clone)] pub struct CurrentRepositoryViewRsyncFetcher { store: Arc, + scope_policy: RsyncScopePolicy, } impl CurrentRepositoryViewRsyncFetcher { - pub fn new(store: Arc) -> Self { - Self { store } + pub fn new(store: Arc, scope_policy: RsyncScopePolicy) -> Self { + Self { + store, + scope_policy, + } } } impl RsyncFetcher for CurrentRepositoryViewRsyncFetcher { fn fetch_objects(&self, rsync_base_uri: &str) -> RsyncFetchResult)>> { - let base = normalize_rsync_base_uri(rsync_base_uri); + // The source run's live fetcher may have fetched a module root rather + // than the publication-point URI that triggered it. Read that same + // frozen-view prefix here so transport prefetch request identities and + // their materialized object sets stay equivalent. + let base = scoped_rsync_fetch_uri(self.scope_policy, rsync_base_uri); let entries = self .store .list_repository_view_entries_with_prefix(&base) @@ -75,7 +84,11 @@ impl RsyncFetcher for CurrentRepositoryViewRsyncFetcher { } fn dedup_key(&self, rsync_base_uri: &str) -> String { - normalize_rsync_base_uri(rsync_base_uri) + scoped_rsync_fetch_uri(self.scope_policy, rsync_base_uri) + } + + fn failure_dedup_key(&self, rsync_base_uri: &str) -> Option { + scoped_rsync_failure_dedup_key(self.scope_policy, rsync_base_uri) } } @@ -107,7 +120,7 @@ mod tests { }) .expect("put view"); } - let fetcher = CurrentRepositoryViewRsyncFetcher::new(store); + let fetcher = CurrentRepositoryViewRsyncFetcher::new(store, RsyncScopePolicy::default()); let objects = fetcher .fetch_objects("rsync://example.test/repo/") .expect("fetch objects"); @@ -168,11 +181,54 @@ mod tests { current_hash: Some(missing_hash), }) .expect("put missing view"); - let fetcher = CurrentRepositoryViewRsyncFetcher::new(store); + let fetcher = CurrentRepositoryViewRsyncFetcher::new(store, RsyncScopePolicy::default()); let error = fetcher .fetch_objects("rsync://example.test/repo/") .unwrap_err() .to_string(); assert!(error.contains("blob bytes missing")); } + + #[test] + fn current_repository_fetcher_replays_module_scope_and_failure_dedup() { + let dir = tempfile::tempdir().expect("tempdir"); + let store = Arc::new(RocksStore::open(dir.path()).expect("open store")); + for (uri, bytes) in [ + ("rsync://example.test/repo/ca/a.mft", b"a".as_slice()), + ("rsync://example.test/repo/other/b.roa", b"b".as_slice()), + ] { + let hash = hex::encode(crate::cir::sha256(bytes)); + store + .put_blob_bytes_batch(&[(hash.clone(), bytes.to_vec())]) + .expect("put blob"); + store + .put_repository_view_entry(&RepositoryViewEntry { + rsync_uri: uri.to_string(), + repository_source: Some("fixture".to_string()), + object_type: Some("fixture".to_string()), + state: RepositoryViewState::Present, + current_hash: Some(hash), + }) + .expect("put view"); + } + + let module_fetcher = CurrentRepositoryViewRsyncFetcher::new( + Arc::clone(&store), + RsyncScopePolicy::ModuleRoot, + ); + let objects = module_fetcher + .fetch_objects("rsync://example.test/repo/ca/") + .expect("fetch module scope"); + assert_eq!(objects.len(), 2); + assert_eq!( + module_fetcher.dedup_key("rsync://example.test/repo/ca/"), + "rsync://example.test/repo/" + ); + + let host_fetcher = CurrentRepositoryViewRsyncFetcher::new(store, RsyncScopePolicy::Host); + assert_eq!( + host_fetcher.failure_dedup_key("rsync://example.test/repo/ca/"), + Some("rsync://example.test/".to_string()) + ); + } } diff --git a/src/fetch/rsync_system.rs b/src/fetch/rsync_system.rs index aeb7913..98a3766 100644 --- a/src/fetch/rsync_system.rs +++ b/src/fetch/rsync_system.rs @@ -13,7 +13,8 @@ use crate::fetch::rsync::{ RsyncFetchError, RsyncFetchResult, RsyncFetcher, normalize_rsync_base_uri, }; -#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "kebab-case")] pub enum RsyncScopePolicy { Host, PublicationPoint, @@ -297,13 +298,7 @@ impl SystemRsyncFetcher { } fn scope_fetch_uri(&self, rsync_base_uri: &str) -> String { - match self.config.scope_policy { - RsyncScopePolicy::Host | RsyncScopePolicy::PublicationPoint => { - normalize_rsync_base_uri(rsync_base_uri) - } - RsyncScopePolicy::ModuleRoot => rsync_module_root_uri(rsync_base_uri) - .unwrap_or_else(|| normalize_rsync_base_uri(rsync_base_uri)), - } + scoped_rsync_fetch_uri(self.config.scope_policy, rsync_base_uri) } } @@ -382,10 +377,35 @@ impl RsyncFetcher for SystemRsyncFetcher { } fn failure_dedup_key(&self, rsync_base_uri: &str) -> Option { - match self.config.scope_policy { - RsyncScopePolicy::Host => rsync_host_scope_uri(rsync_base_uri), - RsyncScopePolicy::PublicationPoint | RsyncScopePolicy::ModuleRoot => None, + scoped_rsync_failure_dedup_key(self.config.scope_policy, rsync_base_uri) + } +} + +/// Return the exact successful-fetch scope used by the live rsync fetcher. +/// +/// Verification-only reuses this function while reading the frozen repository +/// view, so its request keys and object projection scope remain equivalent to +/// the source run without granting it network access. +pub fn scoped_rsync_fetch_uri(scope_policy: RsyncScopePolicy, rsync_base_uri: &str) -> String { + match scope_policy { + RsyncScopePolicy::Host | RsyncScopePolicy::PublicationPoint => { + normalize_rsync_base_uri(rsync_base_uri) } + RsyncScopePolicy::ModuleRoot => rsync_module_root_uri(rsync_base_uri) + .unwrap_or_else(|| normalize_rsync_base_uri(rsync_base_uri)), + } +} + +/// Return the live fetcher's failure-deduplication key for the configured +/// scope. This is part of the transport request identity, even though host +/// scope deliberately does not widen the successful fetch URI. +pub fn scoped_rsync_failure_dedup_key( + scope_policy: RsyncScopePolicy, + rsync_base_uri: &str, +) -> Option { + match scope_policy { + RsyncScopePolicy::Host => rsync_host_scope_uri(rsync_base_uri), + RsyncScopePolicy::PublicationPoint | RsyncScopePolicy::ModuleRoot => None, } } diff --git a/src/policy.rs b/src/policy.rs index 59a9c31..556a683 100644 --- a/src/policy.rs +++ b/src/policy.rs @@ -1,4 +1,5 @@ use serde::{Deserialize, Serialize}; +use std::collections::BTreeSet; #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] @@ -127,6 +128,11 @@ pub struct Policy { pub signed_object_failure_policy: SignedObjectFailurePolicy, pub resource_validation_mode: ResourceValidationMode, pub strict: StrictPolicy, + /// Ephemeral verification-only override. This is deliberately excluded + /// from policy files, contracts, and policy fingerprints: it only selects + /// the already contract-bound VCIR fallback branch while replaying a run. + #[serde(skip)] + pub verification_forced_vcir_reuse_manifest_uris: BTreeSet, } impl Default for Policy { @@ -137,6 +143,7 @@ impl Default for Policy { signed_object_failure_policy: SignedObjectFailurePolicy::default(), resource_validation_mode: ResourceValidationMode::default(), strict: StrictPolicy::default(), + verification_forced_vcir_reuse_manifest_uris: BTreeSet::new(), } } } diff --git a/src/storage.rs b/src/storage.rs index c88f9d8..19ddc35 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -613,6 +613,12 @@ pub struct VcirFailedFetchReuseIdentity { pub effective_until: PackTime, } +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] +pub struct VcirReuseRecordsCleared { + pub vcir_records: u64, + pub failed_fetch_identity_records: u64, +} + impl VcirFailedFetchReuseIdentity { pub fn validate_internal(&self) -> StorageResult<()> { let effective_not_before = parse_time( @@ -4533,6 +4539,36 @@ impl RocksStore { Ok(out) } + /// Remove every reusable VCIR and its failed-fetch identity while retaining + /// manifest replay metadata. The replay metadata is a fresh-validation + /// guard, not a reusable validation result; removing it changes the normal + /// traversal semantics for manifests that are not selected for replay. + /// + /// This is intentionally narrower than [`Self::delete_vcir`]. It is used + /// by verification-only after cloning a work-db so that it can import an + /// explicit, contract-bound allowlist of fallback projections. + pub fn clear_vcir_reuse_records(&self) -> StorageResult { + let vcir_cf = self.cf(CF_VCIR)?; + let identity_cf = self.cf(CF_VCIR_FAILED_FETCH_REUSE_IDENTITY)?; + let mut batch = WriteBatch::default(); + let mut summary = VcirReuseRecordsCleared::default(); + + for entry in self.db.iterator_cf(vcir_cf, IteratorMode::Start) { + let (key, _value) = entry.map_err(|error| StorageError::RocksDb(error.to_string()))?; + batch.delete_cf(vcir_cf, key); + summary.vcir_records += 1; + } + for entry in self.db.iterator_cf(identity_cf, IteratorMode::Start) { + let (key, _value) = entry.map_err(|error| StorageError::RocksDb(error.to_string()))?; + batch.delete_cf(identity_cf, key); + summary.failed_fetch_identity_records += 1; + } + if summary.vcir_records > 0 || summary.failed_fetch_identity_records > 0 { + self.write_batch(batch)?; + } + Ok(summary) + } + pub fn summarize_vcir_storage(&self) -> StorageResult { let cf = self.cf(CF_VCIR)?; let mode = IteratorMode::Start; diff --git a/src/storage/tests.rs b/src/storage/tests.rs index 45ff050..32672dc 100644 --- a/src/storage/tests.rs +++ b/src/storage/tests.rs @@ -1310,6 +1310,111 @@ fn vcir_roundtrip_and_validation_failures_are_reported() { ); } +#[test] +fn clear_vcir_reuse_records_preserves_manifest_replay_meta() { + let td = tempfile::tempdir().expect("tempdir"); + let store = RocksStore::open(td.path()).expect("open rocksdb"); + let vcir = sample_vcir("rsync://example.test/repo/current.mft"); + store.put_vcir(&vcir).expect("put vcir"); + let replay_meta = store + .get_manifest_replay_meta(&vcir.manifest_rsync_uri) + .expect("get replay meta") + .expect("replay meta exists"); + + let cleared = store + .clear_vcir_reuse_records() + .expect("clear reusable VCIR records"); + assert_eq!(cleared.vcir_records, 1); + assert_eq!(cleared.failed_fetch_identity_records, 0); + assert!(store + .get_vcir(&vcir.manifest_rsync_uri) + .expect("get cleared vcir") + .is_none()); + assert_eq!( + store + .get_manifest_replay_meta(&vcir.manifest_rsync_uri) + .expect("get preserved replay meta"), + Some(replay_meta) + ); +} + +#[test] +fn verification_only_materializes_only_contract_selected_fallback_vcirs() { + fn reuse_identity() -> VcirFailedFetchReuseIdentity { + VcirFailedFetchReuseIdentity { + current_ca_sha256: [0x11; 32], + ta_context_digest: [0x22; 32], + ca_validation_context_digest: [0x33; 32], + policy_fingerprint: [0x44; 32], + effective_not_before: pack_time(0), + effective_until: pack_time(12), + } + } + + let td = tempfile::tempdir().expect("tempdir"); + let store = RocksStore::open(td.path()).expect("open rocksdb"); + let selected = sample_vcir("rsync://example.test/repo/selected.mft"); + let unselected = sample_vcir("rsync://example.test/repo/unselected.mft"); + store + .put_vcir_with_failed_fetch_reuse_identity(&selected, &reuse_identity()) + .expect("put selected fallback vcir"); + store + .put_vcir_with_failed_fetch_reuse_identity(&unselected, &reuse_identity()) + .expect("put unselected fallback vcir"); + let selection = crate::verification_only::fallback_replay_selection_for_manifest( + &store, + &selected.manifest_rsync_uri, + ) + .expect("select fallback vcir"); + let normal_audits = vec![crate::audit::PublicationPointAudit { + manifest_rsync_uri: selected.manifest_rsync_uri.clone(), + source: "vcir_current_instance".to_string(), + ..crate::audit::PublicationPointAudit::default() + }]; + assert_eq!( + crate::verification_only::collect_fallback_replay_selections(&store, &normal_audits) + .expect("collect normal fallback selection"), + vec![selection.clone()] + ); + + let mut tampered = selection.clone(); + tampered.vcir_sha256 = "00".repeat(32); + assert!(crate::verification_only::materialize_fallback_replay_selections( + &store, + &[tampered] + ) + .expect_err("tampered contract hash must fail before cleanup") + .contains("hash mismatch")); + assert_eq!(store.list_vcirs().expect("VCIRs remain after hash rejection").len(), 2); + + let summary = crate::verification_only::materialize_fallback_replay_selections( + &store, + &[selection.clone()], + ) + .expect("materialize contract fallback selection"); + assert_eq!(summary.contract_selection_count, 1); + assert_eq!(summary.source_vcir_records_cleared, 2); + assert_eq!(summary.source_reuse_identity_records_cleared, 2); + assert_eq!(summary.imported_selection_count, 1); + assert_eq!(summary.scratch_vcir_count, 1); + assert_eq!( + crate::verification_only::fallback_replay_selection_for_manifest( + &store, + &selected.manifest_rsync_uri, + ) + .expect("selected fallback remains"), + selection + ); + assert!(store + .get_vcir(&unselected.manifest_rsync_uri) + .expect("get unselected fallback") + .is_none()); + assert!(store + .get_manifest_replay_meta(&unselected.manifest_rsync_uri) + .expect("get unselected replay meta") + .is_some()); +} + #[test] fn transport_prefetch_snapshot_roundtrips() { use crate::parallel::transport_prefetch::{ diff --git a/src/validation/manifest.rs b/src/validation/manifest.rs index 00979ed..f36a0b3 100644 --- a/src/validation/manifest.rs +++ b/src/validation/manifest.rs @@ -135,6 +135,11 @@ impl PublicationPointData for PublicationPointSnapshot { #[derive(Debug, thiserror::Error)] pub enum ManifestFreshError { + #[error( + "verification-only fallback replay selected current-instance VCIR: {manifest_rsync_uri}" + )] + VerificationOnlyForcedCurrentInstanceVcir { manifest_rsync_uri: String }, + #[error("repo sync failed: {detail} (RFC 8182 §3.4.5; RFC 9286 §6.6)")] RepoSyncFailed { detail: String }, @@ -222,7 +227,8 @@ impl ManifestFreshError { pub(crate) fn should_warn_when_current_instance_reused(&self) -> bool { !matches!( self, - ManifestFreshError::RepoSyncFailed { .. } + ManifestFreshError::VerificationOnlyForcedCurrentInstanceVcir { .. } + | ManifestFreshError::RepoSyncFailed { .. } | ManifestFreshError::MissingManifest { .. } | ManifestFreshError::MissingFile { .. } ) diff --git a/src/validation/tree_runner.rs b/src/validation/tree_runner.rs index 653bf47..cfd07e9 100644 --- a/src/validation/tree_runner.rs +++ b/src/validation/tree_runner.rs @@ -807,6 +807,21 @@ impl<'a> Rpkiv1PublicationPointRunner<'a> { repo_sync_ok: bool, repo_sync_err: Option<&str>, ) -> Result { + if self + .policy + .verification_forced_vcir_reuse_manifest_uris + .contains(&ca.manifest_rsync_uri) + { + if let Some(timing) = self.timing.as_ref() { + timing.record_count("verification_forced_vcir_reuse_publication_points", 1); + } + return Err(FreshPublicationPointStageError { + error: ManifestFreshError::VerificationOnlyForcedCurrentInstanceVcir { + manifest_rsync_uri: ca.manifest_rsync_uri.clone(), + }, + snapshot_prepare_ms: 0, + }); + } let snapshot_prepare_started = std::time::Instant::now(); let issuer_ca_der = ca_certificate_der_for_validation(ca, self.store, self.timing.as_ref()) .map_err(|detail| FreshPublicationPointStageError { diff --git a/src/validation/tree_runner/tests.rs b/src/validation/tree_runner/tests.rs index 9aeb767..af343cd 100644 --- a/src/validation/tree_runner/tests.rs +++ b/src/validation/tree_runner/tests.rs @@ -3426,6 +3426,45 @@ fn runner_when_repo_sync_fails_uses_current_instance_vcir_and_keeps_children_emp .any(|w| w.message.contains("repo sync failed")), "expected warning about repo sync failure" ); + + // Verification-only replays have the frozen repository bytes available, + // so a fresh read would succeed and never naturally enter the fallback + // branch. A contract-selected URI must therefore select the same VCIR + // projection without attempting fresh manifest/object validation. + let mut forced_policy = policy.clone(); + forced_policy + .verification_forced_vcir_reuse_manifest_uris + .insert(manifest_rsync_uri.clone()); + let forced_runner = Rpkiv1PublicationPointRunner { + store: &store, + policy: &forced_policy, + http_fetcher: &NeverHttpFetcher, + rsync_fetcher: &LocalDirRsyncFetcher::new(&fixture_dir), + validation_time, + timing: None, + download_log: None, + replay_archive_index: None, + replay_delta_index: None, + rrdp_dedup: false, + rrdp_repo_cache: Mutex::new(HashMap::new()), + rsync_dedup: false, + rsync_repo_cache: Mutex::new(HashMap::new()), + current_repo_index: None, + repo_sync_runtime: None, + parallel_phase2_config: None, + parallel_roa_worker_pool: None, + ccr_accumulator: None, + persist_vcir: true, + enable_roa_validation_cache: false, + enable_child_certificate_validation_cache: false, + publication_point_cache_observe_only: false, + enable_publication_point_validation_cache: false, + }; + let forced = forced_runner + .run_publication_point(&handle) + .expect("contract-selected fallback should reuse current-instance VCIR"); + assert_eq!(forced.source, PublicationPointSource::VcirCurrentInstance); + assert!(forced.discovered_children.is_empty()); } #[test] @@ -4807,17 +4846,18 @@ fn build_publication_point_audit_from_vcir_restores_reject_reason_with_legacy_fa assert!(audit.objects.iter().any(|entry| { entry.rsync_uri == "rsync://example.test/repo/issuer/rejected-with-reason.roa" && matches!(entry.result, AuditObjectResult::Error) - && entry.detail.as_deref() - == Some("EE certificate path validation failed: test") + && entry.detail.as_deref() == Some("EE certificate path validation failed: test") })); assert!(audit.objects.iter().any(|entry| { entry.rsync_uri == "rsync://example.test/repo/issuer/rejected-legacy.roa" && matches!(entry.result, AuditObjectResult::Error) && entry.detail.as_deref() == Some(CACHED_REJECT_REASON_NOT_RECORDED) })); - assert!(audit.objects.iter().all(|entry| { - !matches!(entry.result, AuditObjectResult::Ok) || entry.detail.is_none() - })); + assert!( + audit.objects.iter().all(|entry| { + !matches!(entry.result, AuditObjectResult::Ok) || entry.detail.is_none() + }) + ); } #[test] @@ -4897,8 +4937,7 @@ fn build_publication_point_audit_from_pp_cache_projection_restores_reject_reason assert!(audit.objects.iter().any(|entry| { entry.rsync_uri == "rsync://example.test/repo/issuer/rejected-with-reason.roa" && matches!(entry.result, AuditObjectResult::Error) - && entry.detail.as_deref() - == Some("EE certificate path validation failed: test") + && entry.detail.as_deref() == Some("EE certificate path validation failed: test") })); assert!(audit.objects.iter().any(|entry| { entry.rsync_uri == "rsync://example.test/repo/issuer/rejected-legacy.roa" diff --git a/src/verification_only.rs b/src/verification_only.rs index fe4f2b1..1ef4a1b 100644 --- a/src/verification_only.rs +++ b/src/verification_only.rs @@ -8,8 +8,12 @@ use sha2::{Digest, Sha256}; use time::format_description::well_known::Rfc3339; use crate::parallel::types::TalInputSpec; -use crate::policy::Policy; -use crate::storage::{RepositoryBlobVerificationSummary, RocksStore}; +use crate::policy::{CaFailedFetchPolicy, Policy}; +use crate::fetch::rsync_system::RsyncScopePolicy; +use crate::storage::{ + RepositoryBlobVerificationSummary, RocksStore, ValidatedCaInstanceResult, + VcirFailedFetchReuseIdentity, VcirReuseRecordsCleared, +}; pub const VALIDATION_CONTRACT_SCHEMA_VERSION: u32 = 1; pub const CURRENT_STATE_BINDING_SCHEMA_VERSION: u32 = 1; @@ -25,6 +29,37 @@ pub struct ValidationCacheContract { pub crypto_signature: bool, } +/// The normal-run fallback inputs which verification-only is permitted to +/// replay. The payload remains in the source work-db; the contract binds the +/// selected VCIR and its failed-fetch identity by canonical CBOR hashes. +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct FallbackReplaySelection { + pub manifest_rsync_uri: String, + pub vcir_sha256: String, + pub reuse_identity_sha256: String, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct FallbackReplayMaterialization { + pub contract_selection_count: u64, + pub source_vcir_records_cleared: u64, + pub source_reuse_identity_records_cleared: u64, + pub imported_selection_count: u64, + pub scratch_vcir_count: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct FallbackReplaySelectionComparison { + pub matches: bool, + pub expected: Vec, + pub actual: Vec, + pub missing: Vec, + pub unexpected: Vec, +} + impl ValidationCacheContract { pub fn any_validation_cache_enabled(&self) -> bool { self.publication_point || self.roa || self.child_certificate || self.crypto_signature @@ -41,6 +76,13 @@ pub struct ValidationContract { pub max_ca_depth: usize, pub max_instances: Option, pub cache: ValidationCacheContract, + /// Source transport scope. With `transportPrefetch=true`, verification-only + /// replays this scope over the frozen repository view rather than deriving + /// a new publication-point scope. + #[serde(default)] + pub rsync_scope_policy: RsyncScopePolicy, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub fallback_replay_selections: Vec, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -91,6 +133,7 @@ impl ValidationContract { max_ca_depth: usize, max_instances: Option, cache: ValidationCacheContract, + rsync_scope_policy: RsyncScopePolicy, ) -> Result { Ok(Self { schema_version: VALIDATION_CONTRACT_SCHEMA_VERSION, @@ -100,6 +143,8 @@ impl ValidationContract { max_ca_depth, max_instances, cache, + rsync_scope_policy, + fallback_replay_selections: Vec::new(), }) } @@ -121,6 +166,40 @@ impl ValidationContract { if self.max_ca_depth == 0 { return Err("validation contract maxCaDepth must be greater than zero".to_string()); } + if !self.fallback_replay_selections.is_empty() + && self.policy.ca_failed_fetch_policy != CaFailedFetchPolicy::ReuseCurrentInstanceVcir + { + return Err( + "validation contract fallbackReplaySelections require caFailedFetchPolicy=reuse_current_instance_vcir" + .to_string(), + ); + } + let mut previous: Option<&str> = None; + for selection in &self.fallback_replay_selections { + if !selection.manifest_rsync_uri.starts_with("rsync://") { + return Err(format!( + "validation contract fallbackReplaySelections manifestRsyncUri must be an rsync URI: {}", + selection.manifest_rsync_uri + )); + } + validate_sha256_hex( + "validation contract fallbackReplaySelections vcirSha256", + &selection.vcir_sha256, + )?; + validate_sha256_hex( + "validation contract fallbackReplaySelections reuseIdentitySha256", + &selection.reuse_identity_sha256, + )?; + if let Some(previous) = previous { + if previous >= selection.manifest_rsync_uri.as_str() { + return Err( + "validation contract fallbackReplaySelections must be sorted and unique by manifestRsyncUri" + .to_string(), + ); + } + } + previous = Some(&selection.manifest_rsync_uri); + } Ok(()) } @@ -197,6 +276,7 @@ pub struct PreparedVerification { pub cir: crate::cir::CanonicalInputRepresentation, pub contract: ValidationContract, pub blob_summary: RepositoryBlobVerificationSummary, + pub fallback_replay_materialization: FallbackReplayMaterialization, pub store: Option>, } @@ -234,6 +314,8 @@ pub struct VerificationRunMeta { pub repository_objects_checked: u64, pub repository_bytes_checked: u64, pub state_digest_match: bool, + pub fallback_replay_selection_match: bool, + pub fallback_replay_materialization: FallbackReplayMaterialization, pub scratch_retained: bool, } @@ -305,6 +387,174 @@ pub fn cir_tal_uris(cir: &crate::cir::CanonicalInputRepresentation) -> Vec Result, String> { + let mut selections = Vec::new(); + for publication_point in publication_points { + if publication_point.source != "vcir_current_instance" { + continue; + } + selections.push(fallback_replay_selection_for_manifest( + store, + &publication_point.manifest_rsync_uri, + )?); + } + selections.sort(); + for pair in selections.windows(2) { + if pair[0].manifest_rsync_uri == pair[1].manifest_rsync_uri { + return Err(format!( + "multiple vcir_current_instance publication points use the same manifest URI: {}", + pair[0].manifest_rsync_uri + )); + } + } + Ok(selections) +} + +pub(crate) fn fallback_replay_selection_for_manifest( + store: &RocksStore, + manifest_rsync_uri: &str, +) -> Result { + let vcir = store + .get_vcir(manifest_rsync_uri) + .map_err(|error| format!("load fallback VCIR failed for {manifest_rsync_uri}: {error}"))? + .ok_or_else(|| format!("fallback VCIR is missing for {manifest_rsync_uri}"))?; + let reuse_identity = store + .get_vcir_failed_fetch_reuse_identity(manifest_rsync_uri) + .map_err(|error| { + format!("load fallback VCIR reuse identity failed for {manifest_rsync_uri}: {error}") + })? + .ok_or_else(|| format!("fallback VCIR reuse identity is missing for {manifest_rsync_uri}"))?; + fallback_replay_selection_from_values(&vcir, &reuse_identity) +} + +fn fallback_replay_selection_from_values( + vcir: &ValidatedCaInstanceResult, + reuse_identity: &VcirFailedFetchReuseIdentity, +) -> Result { + Ok(FallbackReplaySelection { + manifest_rsync_uri: vcir.manifest_rsync_uri.clone(), + vcir_sha256: canonical_cbor_sha256("fallback VCIR", vcir)?, + reuse_identity_sha256: canonical_cbor_sha256("fallback VCIR reuse identity", reuse_identity)?, + }) +} + +fn canonical_cbor_sha256(label: &str, value: &T) -> Result { + let bytes = serde_cbor::to_vec(value) + .map_err(|error| format!("serialize {label} for SHA-256 failed: {error}"))?; + Ok(hex::encode(Sha256::digest(bytes))) +} + +pub(crate) fn materialize_fallback_replay_selections( + store: &RocksStore, + selections: &[FallbackReplaySelection], +) -> Result { + let mut extracted = Vec::with_capacity(selections.len()); + for selection in selections { + let actual = fallback_replay_selection_for_manifest(store, &selection.manifest_rsync_uri)?; + if actual != *selection { + return Err(format!( + "fallback replay selection hash mismatch for {}: contract vcir={}, work-db vcir={}, contract identity={}, work-db identity={}", + selection.manifest_rsync_uri, + selection.vcir_sha256, + actual.vcir_sha256, + selection.reuse_identity_sha256, + actual.reuse_identity_sha256, + )); + } + let vcir = store + .get_vcir(&selection.manifest_rsync_uri) + .map_err(|error| format!("reload selected fallback VCIR failed: {error}"))? + .expect("fallback selection was just verified as present"); + let reuse_identity = store + .get_vcir_failed_fetch_reuse_identity(&selection.manifest_rsync_uri) + .map_err(|error| format!("reload selected fallback reuse identity failed: {error}"))? + .expect("fallback selection was just verified as present"); + extracted.push((vcir, reuse_identity)); + } + + let VcirReuseRecordsCleared { + vcir_records, + failed_fetch_identity_records, + } = store + .clear_vcir_reuse_records() + .map_err(|error| format!("clear scratch VCIR fallback records failed: {error}"))?; + for (vcir, reuse_identity) in &extracted { + store + .put_vcir_with_failed_fetch_reuse_identity(vcir, reuse_identity) + .map_err(|error| format!("import selected fallback VCIR failed: {error}"))?; + } + + let actual = store + .list_vcirs() + .map_err(|error| format!("list scratch fallback VCIR records failed: {error}"))? + .iter() + .map(|vcir| { + let reuse_identity = store + .get_vcir_failed_fetch_reuse_identity(&vcir.manifest_rsync_uri) + .map_err(|error| format!("read materialized fallback reuse identity failed: {error}"))? + .ok_or_else(|| { + format!( + "materialized fallback reuse identity is missing for {}", + vcir.manifest_rsync_uri + ) + })?; + fallback_replay_selection_from_values(vcir, &reuse_identity) + }) + .collect::, String>>()?; + if actual != selections { + return Err(format!( + "scratch fallback replay allowlist differs from validation contract: expected={}, actual={}", + selections.len(), + actual.len() + )); + } + + Ok(FallbackReplayMaterialization { + contract_selection_count: selections.len() as u64, + source_vcir_records_cleared: vcir_records, + source_reuse_identity_records_cleared: failed_fetch_identity_records, + imported_selection_count: extracted.len() as u64, + scratch_vcir_count: actual.len() as u64, + }) +} + +pub fn compare_fallback_replay_selections( + expected: &[FallbackReplaySelection], + actual: &[FallbackReplaySelection], +) -> FallbackReplaySelectionComparison { + let missing: Vec = expected + .iter() + .filter(|selection| !actual.contains(selection)) + .cloned() + .collect(); + let unexpected: Vec = actual + .iter() + .filter(|selection| !expected.contains(selection)) + .cloned() + .collect(); + FallbackReplaySelectionComparison { + matches: missing.is_empty() && unexpected.is_empty(), + expected: expected.to_vec(), + actual: actual.to_vec(), + missing, + unexpected, + } +} + +pub fn write_fallback_replay_selection_comparison( + path: &Path, + comparison: &FallbackReplaySelectionComparison, +) -> Result<(), String> { + let bytes = serde_json::to_vec_pretty(comparison) + .map_err(|error| format!("encode fallback replay selection comparison failed: {error}"))?; + atomic_write(path, &bytes) +} + pub fn prepare_verification( source_run_dir: &Path, source_state_root: &Path, @@ -391,6 +641,10 @@ pub fn prepare_verification( ) .map_err(|error| format!("open frozen verification store failed: {error}"))?, ); + let fallback_replay_materialization = materialize_fallback_replay_selections( + store.as_ref(), + &contract.fallback_replay_selections, + )?; let blob_summary = store .verify_current_repository_blobs(1024) .map_err(|error| format!("verify frozen repository blobs failed: {error}"))?; @@ -401,6 +655,7 @@ pub fn prepare_verification( cir, contract, blob_summary, + fallback_replay_materialization, store: Some(store), }) } @@ -649,6 +904,8 @@ mod tests { transport_prefetch: false, crypto_signature: false, }, + rsync_scope_policy: RsyncScopePolicy::default(), + fallback_replay_selections: Vec::new(), } } @@ -756,6 +1013,7 @@ mod tests { transport_prefetch: true, crypto_signature: false, }, + RsyncScopePolicy::default(), ) .expect("contract"); write_validation_contract(&source_run.join("validation-contract.json"), &contract) @@ -799,6 +1057,60 @@ mod tests { assert!(contract.validate().is_err()); } + #[test] + fn fallback_replay_contract_requires_reuse_policy_and_sorted_unique_selections() { + let selection = FallbackReplaySelection { + manifest_rsync_uri: "rsync://example.test/repo/current.mft".to_string(), + vcir_sha256: "11".repeat(32), + reuse_identity_sha256: "22".repeat(32), + }; + let mut contract = sample_contract(); + contract.policy.ca_failed_fetch_policy = CaFailedFetchPolicy::StopAllOutput; + contract.fallback_replay_selections = vec![selection.clone()]; + assert!(contract + .validate() + .unwrap_err() + .contains("caFailedFetchPolicy")); + + contract.policy.ca_failed_fetch_policy = CaFailedFetchPolicy::ReuseCurrentInstanceVcir; + contract.fallback_replay_selections = vec![ + FallbackReplaySelection { + manifest_rsync_uri: "rsync://example.test/repo/z.mft".to_string(), + ..selection.clone() + }, + FallbackReplaySelection { + manifest_rsync_uri: "rsync://example.test/repo/a.mft".to_string(), + ..selection.clone() + }, + ]; + assert!(contract.validate().unwrap_err().contains("sorted and unique")); + contract.fallback_replay_selections.sort(); + contract.validate().expect("sorted selection contract"); + } + + #[test] + fn fallback_replay_selection_comparison_reports_exact_differences() { + let expected = vec![FallbackReplaySelection { + manifest_rsync_uri: "rsync://example.test/repo/a.mft".to_string(), + vcir_sha256: "11".repeat(32), + reuse_identity_sha256: "22".repeat(32), + }]; + let mut actual = expected.clone(); + actual[0].vcir_sha256 = "33".repeat(32); + let comparison = compare_fallback_replay_selections(&expected, &actual); + assert!(!comparison.matches); + assert_eq!(comparison.missing, expected); + assert_eq!(comparison.unexpected, actual); + + let temp = tempfile::tempdir().expect("tempdir"); + let path = temp.path().join("fallback-replay-selection.json"); + write_fallback_replay_selection_comparison(&path, &comparison) + .expect("write fallback selection comparison"); + assert!(fs::read_to_string(path) + .expect("read fallback selection comparison") + .contains("reuseIdentitySha256")); + } + #[test] fn cache_contract_distinguishes_prefetch_from_validation_caches() { let cache = ValidationCacheContract { @@ -811,6 +1123,18 @@ mod tests { assert!(!cache.any_validation_cache_enabled()); } + #[test] + fn validation_contract_defaults_missing_rsync_scope_to_module_root() { + let mut value = serde_json::to_value(sample_contract()).expect("serialize contract"); + value + .as_object_mut() + .expect("contract object") + .remove("rsyncScopePolicy"); + let decoded: ValidationContract = + serde_json::from_value(value).expect("decode older contract"); + assert_eq!(decoded.rsync_scope_policy, RsyncScopePolicy::ModuleRoot); + } + #[test] fn validation_contract_and_binding_validation_cover_rejection_paths() { let time = time::OffsetDateTime::parse( @@ -824,6 +1148,7 @@ mod tests { 32, Some(10), sample_contract().cache, + RsyncScopePolicy::default(), ) .expect("current contract"); assert_eq!(contract.validation_time().expect("contract time"), time); @@ -919,6 +1244,8 @@ mod tests { repository_objects_checked: 3, repository_bytes_checked: 4, state_digest_match: true, + fallback_replay_selection_match: true, + fallback_replay_materialization: FallbackReplayMaterialization::default(), scratch_retained: false, }; let meta_path = temp.path().join("meta/result.json"); @@ -1008,6 +1335,16 @@ mod tests { assert_eq!(prepared.blob_summary.current_objects, 0); assert_eq!(prepared.blob_summary.bytes_verified, 0); assert_eq!(prepared.contract.validation_time, "2026-07-16T00:00:00Z"); + assert_eq!( + prepared + .fallback_replay_materialization + .contract_selection_count, + 0 + ); + assert_eq!( + prepared.fallback_replay_materialization.scratch_vcir_count, + 0 + ); assert_eq!(cir_tal_uris(&prepared.cir).len(), 1); assert!(prepared.artifacts.scratch_work_db.is_dir()); assert!(prepared.artifacts.result_contract.is_file());