From d6b0c379f11112b144407ee94ce21618f6a9f15e Mon Sep 17 00:00:00 2001 From: Arbousier1 Date: Fri, 17 Jul 2026 00:25:14 +0800 Subject: [PATCH 1/3] ci(perf): parallelize paired A/B shards --- .github/workflows/performance-ab.yml | 351 ++++++++++++++++++---- perf/ab/merge_manifests.py | 427 +++++++++++++++++++++++++++ perf/ab/run_matrix.py | 74 ++++- perf/ab/tests/test_gate.py | 250 ++++++++++++++++ 4 files changed, 1038 insertions(+), 64 deletions(-) create mode 100644 perf/ab/merge_manifests.py diff --git a/.github/workflows/performance-ab.yml b/.github/workflows/performance-ab.yml index d24dfe5..4700a18 100644 --- a/.github/workflows/performance-ab.yml +++ b/.github/workflows/performance-ab.yml @@ -62,7 +62,7 @@ concurrency: cancel-in-progress: true jobs: - benchmark: + prepare: if: >- github.event_name == 'workflow_dispatch' || contains(github.event.pull_request.labels.*.name, 'performance-ab') @@ -73,12 +73,22 @@ jobs: env: BASE_SHA: ${{ github.event.pull_request.base.sha || inputs.base_sha }} CANDIDATE_SHA: ${{ github.event.pull_request.head.sha || inputs.candidate_sha }} - PERF_RESULTS_ROOT: /tmp/mahjong-performance-ab-${{ github.run_id }}-${{ github.run_attempt }} - CI_FORKS: ${{ inputs.forks }} - CI_WARMUPS: ${{ inputs.warmup_iterations }} - CI_MEASUREMENTS: ${{ inputs.measurement_iterations }} + PREPARED_ROOT: /tmp/mahjong-performance-prepared-${{ github.run_id }}-${{ github.run_attempt }} steps: + - name: Require a same-repository optimization branch + if: github.event_name != 'workflow_dispatch' + shell: bash + env: + HEAD_REPOSITORY: ${{ github.event.pull_request.head.repo.full_name }} + TARGET_REPOSITORY: ${{ github.repository }} + run: | + set -euo pipefail + if [[ "$HEAD_REPOSITORY" != "$TARGET_REPOSITORY" ]]; then + echo "::error::The trusted performance gate cannot execute candidate code from a fork" + exit 1 + fi + - name: Check out base-owned harness uses: actions/checkout@v7 with: @@ -119,29 +129,6 @@ jobs: with: cache-disabled: true - - name: Create an unprivileged candidate benchmark user - shell: bash - run: | - set -euo pipefail - sudo useradd \ - --system \ - --user-group \ - --no-create-home \ - --shell /usr/sbin/nologin \ - mahjong-benchmark - getent passwd mahjong-benchmark - sudo install -d --owner root --group root --mode 0711 "$PERF_RESULTS_ROOT" - sudo install -d \ - --owner "$(id -u)" \ - --group "$(id -g)" \ - --mode 0711 \ - "$PERF_RESULTS_ROOT/evidence" - sudo install -d \ - --owner "$(id -u)" \ - --group "$(id -g)" \ - --mode 0711 \ - "$PERF_RESULTS_ROOT/runtime-tmp" - - name: Make wrappers executable shell: bash run: chmod +x base/gradlew candidate/gradlew @@ -188,7 +175,7 @@ jobs: --base-dir base \ --candidate-dir candidate \ --config base/perf/ab/gate-config.json \ - --report "$PERF_RESULTS_ROOT/evidence/protected-paths.json" + --report "$PREPARED_ROOT/protected-paths-prepare.json" - name: Test base-owned statistics shell: bash @@ -223,8 +210,7 @@ jobs: "-PmahjongPaperDevBundle=1.20.1-R0.1-SNAPSHOT" "-PmahjongJavaToolchain=25" "-PmahjongJavaTarget=17" - - name: Locate benchmark jars - id: jars + - name: Stage immutable benchmark inputs shell: bash run: | set -euo pipefail @@ -232,26 +218,176 @@ jobs: CANDIDATE_JAR="$(find candidate/build/libs -maxdepth 1 -name '*-jmh.jar' -print -quit)" test -n "$BASE_JAR" test -n "$CANDIDATE_JAR" - cp base/perf/ab/gate-config.json "$PERF_RESULTS_ROOT/evidence/gate-config.json" - mkdir -p "$PERF_RESULTS_ROOT/evidence/jars" - cp "$BASE_JAR" "$PERF_RESULTS_ROOT/evidence/jars/base-jmh.jar" - cp "$CANDIDATE_JAR" "$PERF_RESULTS_ROOT/evidence/jars/candidate-jmh.jar" + mkdir -p "$PREPARED_ROOT/jars" + cp base/perf/ab/gate-config.json "$PREPARED_ROOT/gate-config.json" + cp "$BASE_JAR" "$PREPARED_ROOT/jars/base-jmh.jar" + cp "$CANDIDATE_JAR" "$PREPARED_ROOT/jars/candidate-jmh.jar" chmod 0444 \ - "$PERF_RESULTS_ROOT/evidence/jars/base-jmh.jar" \ - "$PERF_RESULTS_ROOT/evidence/jars/candidate-jmh.jar" - echo "base=$PERF_RESULTS_ROOT/evidence/jars/base-jmh.jar" >> "$GITHUB_OUTPUT" - echo "candidate=$PERF_RESULTS_ROOT/evidence/jars/candidate-jmh.jar" >> "$GITHUB_OUTPUT" - sudo -H -u mahjong-benchmark -- test -r "$PERF_RESULTS_ROOT/evidence/jars/base-jmh.jar" - sudo -H -u mahjong-benchmark -- test -r "$PERF_RESULTS_ROOT/evidence/jars/candidate-jmh.jar" - sudo -H -u mahjong-benchmark -- test ! -w "$PERF_RESULTS_ROOT/evidence/jars/base-jmh.jar" - sudo -H -u mahjong-benchmark -- test ! -w "$PERF_RESULTS_ROOT/evidence/jars/candidate-jmh.jar" + "$PREPARED_ROOT/gate-config.json" \ + "$PREPARED_ROOT/jars/base-jmh.jar" \ + "$PREPARED_ROOT/jars/candidate-jmh.jar" + ( + cd "$PREPARED_ROOT" + sha256sum gate-config.json jars/base-jmh.jar jars/candidate-jmh.jar > SHA256SUMS + ) + + - name: Record prepared input provenance + shell: bash + env: + PROFILE: ${{ steps.profile.outputs.name }} + run: | + set -euo pipefail + python3 - "$PREPARED_ROOT" "$BASE_SHA" "$CANDIDATE_SHA" "$PROFILE" <<'PY' + import hashlib + import json + import os + import pathlib + import sys + + root = pathlib.Path(sys.argv[1]) + files = {} + for relative in ("gate-config.json", "jars/base-jmh.jar", "jars/candidate-jmh.jar"): + path = root / relative + files[relative] = hashlib.sha256(path.read_bytes()).hexdigest() + manifest = { + "schema_version": 1, + "github_run_id": os.environ["GITHUB_RUN_ID"], + "github_run_attempt": os.environ["GITHUB_RUN_ATTEMPT"], + "base_sha": sys.argv[2], + "candidate_sha": sys.argv[3], + "profile": sys.argv[4], + "files": files, + } + (root / "prepared-manifest.json").write_text( + json.dumps(manifest, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + PY + + - name: Upload immutable benchmark inputs + uses: actions/upload-artifact@v7 + with: + name: performance-ab-prepared-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ env.PREPARED_ROOT }}/** + if-no-files-found: error + retention-days: 1 + + benchmark: + needs: prepare + strategy: + fail-fast: false + max-parallel: 8 + matrix: + pair_index: + - 0 + - 1 + - 2 + - 3 + - 4 + - 5 + - 6 + - 7 + runs-on: ubuntu-24.04 + timeout-minutes: 30 + env: + BASE_SHA: ${{ github.event.pull_request.base.sha || inputs.base_sha }} + CANDIDATE_SHA: ${{ github.event.pull_request.head.sha || inputs.candidate_sha }} + PROFILE: ${{ needs.prepare.outputs.profile }} + PAIR_INDEX: ${{ matrix.pair_index }} + PREPARED_ROOT: ${{ runner.temp }}/performance-ab-prepared + PERF_RESULTS_ROOT: /tmp/mahjong-performance-ab-${{ github.run_id }}-${{ github.run_attempt }}-${{ matrix.pair_index }} + CI_FORKS: ${{ inputs.forks }} + CI_WARMUPS: ${{ inputs.warmup_iterations }} + CI_MEASUREMENTS: ${{ inputs.measurement_iterations }} + + steps: + - name: Check out base-owned benchmark runner + uses: actions/checkout@v7 + with: + ref: ${{ env.BASE_SHA }} + path: base + fetch-depth: 1 + persist-credentials: false + + - name: Set up JDK 25 + uses: actions/setup-java@v5 + with: + distribution: temurin + java-version: "25.0.2+10.0.LTS" + + - name: Set up Python + uses: actions/setup-python@v6 + with: + python-version: "3.13" + + - name: Download immutable benchmark inputs + uses: actions/download-artifact@v8 + with: + name: performance-ab-prepared-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ env.PREPARED_ROOT }} + + - name: Verify prepared inputs + shell: bash + run: | + set -euo pipefail + ( + cd "$PREPARED_ROOT" + sha256sum --check SHA256SUMS + ) + cmp base/perf/ab/gate-config.json "$PREPARED_ROOT/gate-config.json" + python3 - "$PREPARED_ROOT/prepared-manifest.json" \ + "$BASE_SHA" "$CANDIDATE_SHA" "$PROFILE" "$GITHUB_RUN_ID" "$GITHUB_RUN_ATTEMPT" <<'PY' + import json + import pathlib + import sys + + value = json.loads(pathlib.Path(sys.argv[1]).read_text(encoding="utf-8")) + expected = { + "base_sha": sys.argv[2], + "candidate_sha": sys.argv[3], + "profile": sys.argv[4], + "github_run_id": sys.argv[5], + "github_run_attempt": sys.argv[6], + } + for key, wanted in expected.items(): + if str(value.get(key)) != wanted: + raise SystemExit(f"prepared manifest {key} mismatch") + PY + + - name: Create isolated runner-local evidence directories + shell: bash + run: | + set -euo pipefail + sudo useradd \ + --system \ + --user-group \ + --no-create-home \ + --shell /usr/sbin/nologin \ + mahjong-benchmark + getent passwd mahjong-benchmark + sudo install -d --owner root --group root --mode 0711 "$PERF_RESULTS_ROOT" + sudo install -d \ + --owner "$(id -u)" \ + --group "$(id -g)" \ + --mode 0711 \ + "$PERF_RESULTS_ROOT/evidence" + sudo install -d \ + --owner "$(id -u)" \ + --group "$(id -g)" \ + --mode 0711 \ + "$PERF_RESULTS_ROOT/runtime-tmp" + RUNNER_SESSION_ID="$(python3 -c 'import secrets; print(secrets.token_hex(32))')" + echo "RUNNER_SESSION_ID=$RUNNER_SESSION_ID" >> "$GITHUB_ENV" + sudo -H -u mahjong-benchmark -- test -r "$PREPARED_ROOT/jars/base-jmh.jar" + sudo -H -u mahjong-benchmark -- test -r "$PREPARED_ROOT/jars/candidate-jmh.jar" + sudo -H -u mahjong-benchmark -- test ! -w "$PREPARED_ROOT/jars/base-jmh.jar" + sudo -H -u mahjong-benchmark -- test ! -w "$PREPARED_ROOT/jars/candidate-jmh.jar" sudo -H -u mahjong-benchmark -- test ! -w "$PERF_RESULTS_ROOT/evidence" sudo -H -u mahjong-benchmark -- test ! -w "$PERF_RESULTS_ROOT/runtime-tmp" - - name: Run A/A drift control + - name: Run runner-local A/A control pair shell: bash env: - BASE_JAR: ${{ steps.jars.outputs.base }} GITHUB_TOKEN: "" GH_TOKEN: "" ACTIONS_CACHE_URL: "" @@ -267,24 +403,25 @@ jobs: [[ -z "$CI_MEASUREMENTS" ]] || OVERRIDES+=(--measurement-iterations "$CI_MEASUREMENTS") python3 base/perf/ab/run_matrix.py \ --phase aa \ - --profile "${{ steps.profile.outputs.name }}" \ - --base-jar "$BASE_JAR" \ - --candidate-jar "$BASE_JAR" \ + --profile "$PROFILE" \ + --base-jar "$PREPARED_ROOT/jars/base-jmh.jar" \ + --candidate-jar "$PREPARED_ROOT/jars/base-jmh.jar" \ --base-sha "$BASE_SHA" \ --candidate-sha "$CANDIDATE_SHA" \ --config base/perf/ab/gate-config.json \ --output-dir "$PERF_RESULTS_ROOT/evidence/aa" \ --runtime-temp-root "$PERF_RESULTS_ROOT/runtime-tmp" \ --java "$JAVA_HOME/bin/java" \ + --pair-start "$PAIR_INDEX" \ + --pair-count 1 \ + --runner-session-id "$RUNNER_SESSION_ID" \ --candidate-command-prefix-json '["sudo","-H","-u","mahjong-benchmark","--"]' \ --candidate-file-user mahjong-benchmark \ "${OVERRIDES[@]}" - - name: Run four interleaved AB and four BA pairs + - name: Run runner-local A/B pair shell: bash env: - BASE_JAR: ${{ steps.jars.outputs.base }} - CANDIDATE_JAR: ${{ steps.jars.outputs.candidate }} GITHUB_TOKEN: "" GH_TOKEN: "" ACTIONS_CACHE_URL: "" @@ -300,39 +437,71 @@ jobs: [[ -z "$CI_MEASUREMENTS" ]] || OVERRIDES+=(--measurement-iterations "$CI_MEASUREMENTS") python3 base/perf/ab/run_matrix.py \ --phase ab \ - --profile "${{ steps.profile.outputs.name }}" \ - --base-jar "$BASE_JAR" \ - --candidate-jar "$CANDIDATE_JAR" \ + --profile "$PROFILE" \ + --base-jar "$PREPARED_ROOT/jars/base-jmh.jar" \ + --candidate-jar "$PREPARED_ROOT/jars/candidate-jmh.jar" \ --base-sha "$BASE_SHA" \ --candidate-sha "$CANDIDATE_SHA" \ --config base/perf/ab/gate-config.json \ --output-dir "$PERF_RESULTS_ROOT/evidence/ab" \ --runtime-temp-root "$PERF_RESULTS_ROOT/runtime-tmp" \ --java "$JAVA_HOME/bin/java" \ + --pair-start "$PAIR_INDEX" \ + --pair-count 1 \ + --runner-session-id "$RUNNER_SESSION_ID" \ --candidate-command-prefix-json '["sudo","-H","-u","mahjong-benchmark","--"]' \ --candidate-file-user mahjong-benchmark \ "${OVERRIDES[@]}" - - name: Upload isolated raw evidence for the trusted decision job + - name: Upload runner-local evidence if: always() uses: actions/upload-artifact@v7 with: - name: performance-ab-raw-${{ github.run_id }}-${{ github.run_attempt }} + name: performance-ab-shard-${{ github.run_id }}-${{ github.run_attempt }}-${{ matrix.pair_index }} path: ${{ env.PERF_RESULTS_ROOT }}/evidence/** if-no-files-found: error retention-days: 1 decide: - needs: benchmark + needs: + - prepare + - benchmark + if: always() runs-on: ubuntu-24.04 timeout-minutes: 20 env: BASE_SHA: ${{ github.event.pull_request.base.sha || inputs.base_sha }} CANDIDATE_SHA: ${{ github.event.pull_request.head.sha || inputs.candidate_sha }} - PROFILE: ${{ needs.benchmark.outputs.profile }} + PROFILE: ${{ needs.prepare.outputs.profile }} steps: + - name: Require a complete eligible performance run + shell: bash + env: + BENCHMARK_RESULT: ${{ needs.benchmark.result }} + GATE_REQUESTED: ${{ github.event_name == 'workflow_dispatch' || contains(github.event.pull_request.labels.*.name, 'performance-ab') }} + HEAD_REPOSITORY: ${{ github.event.pull_request.head.repo.full_name }} + PREPARE_RESULT: ${{ needs.prepare.result }} + TARGET_REPOSITORY: ${{ github.repository }} + run: | + set -euo pipefail + if [[ "$GATE_REQUESTED" != "true" ]]; then + echo "PERFORMANCE_GATE_ENABLED=false" >> "$GITHUB_ENV" + echo "Performance A/B was not requested for this pull request." + exit 0 + fi + if [[ "$GITHUB_EVENT_NAME" != "workflow_dispatch" && "$HEAD_REPOSITORY" != "$TARGET_REPOSITORY" ]]; then + echo "::error::The trusted performance gate cannot execute candidate code from a fork" + exit 1 + fi + if [[ "$PREPARE_RESULT" != "success" || "$BENCHMARK_RESULT" != "success" ]]; then + echo "::error::Performance preparation or one of the eight measurement shards failed" + exit 1 + fi + echo "PERFORMANCE_GATE_ENABLED=true" >> "$GITHUB_ENV" + - name: Check out a fresh base-owned decision gate + if: env.PERFORMANCE_GATE_ENABLED == 'true' uses: actions/checkout@v7 with: ref: ${{ env.BASE_SHA }} @@ -341,6 +510,7 @@ jobs: persist-credentials: false - name: Check out candidate only for a fresh protected-path comparison + if: env.PERFORMANCE_GATE_ENABLED == 'true' uses: actions/checkout@v7 with: repository: ${{ github.event.pull_request.head.repo.full_name || github.repository }} @@ -350,17 +520,28 @@ jobs: persist-credentials: false - name: Set up Python + if: env.PERFORMANCE_GATE_ENABLED == 'true' uses: actions/setup-python@v6 with: python-version: "3.13" - - name: Download raw benchmark evidence + - name: Download immutable benchmark inputs + if: env.PERFORMANCE_GATE_ENABLED == 'true' uses: actions/download-artifact@v8 with: - name: performance-ab-raw-${{ github.run_id }}-${{ github.run_attempt }} + name: performance-ab-prepared-${{ github.run_id }}-${{ github.run_attempt }} path: perf-results + - name: Download isolated runner-local evidence + if: env.PERFORMANCE_GATE_ENABLED == 'true' + uses: actions/download-artifact@v8 + with: + pattern: performance-ab-shard-${{ github.run_id }}-${{ github.run_attempt }}-* + path: perf-shards + merge-multiple: false + - name: Re-verify revisions and the protected decision surface + if: env.PERFORMANCE_GATE_ENABLED == 'true' shell: bash run: | set -euo pipefail @@ -373,7 +554,51 @@ jobs: --report perf-results/protected-paths.json python3 -m unittest discover -s base/perf/ab/tests -v + - name: Merge and verify the complete shard set + if: env.PERFORMANCE_GATE_ENABLED == 'true' + shell: bash + run: | + set -euo pipefail + mapfile -t AA_MANIFESTS < <( + find perf-shards -path '*/aa/run-manifest.json' -type f | sort + ) + mapfile -t AB_MANIFESTS < <( + find perf-shards -path '*/ab/run-manifest.json' -type f | sort + ) + if [[ "${#AA_MANIFESTS[@]}" -ne 8 || "${#AB_MANIFESTS[@]}" -ne 8 ]]; then + echo "::error::Expected exactly 8 A/A and 8 A/B shard manifests" + exit 1 + fi + + AA_ARGS=() + for manifest in "${AA_MANIFESTS[@]}"; do + AA_ARGS+=(--manifest "$manifest") + done + python3 base/perf/ab/merge_manifests.py \ + "${AA_ARGS[@]}" \ + --output-dir perf-results/aa + + AB_ARGS=() + for manifest in "${AB_MANIFESTS[@]}"; do + AB_ARGS+=(--manifest "$manifest") + done + python3 base/perf/ab/merge_manifests.py \ + "${AB_ARGS[@]}" \ + --output-dir perf-results/ab \ + --peer-manifest perf-results/aa/run-manifest.json + + mkdir -p perf-results/shards + for index in "${!AA_MANIFESTS[@]}"; do + printf -v shard '%02d' "$index" + cp "${AA_MANIFESTS[$index]}" "perf-results/shards/aa-$shard-run-manifest.json" + cp "${AB_MANIFESTS[$index]}" "perf-results/shards/ab-$shard-run-manifest.json" + chmod 0444 \ + "perf-results/shards/aa-$shard-run-manifest.json" \ + "perf-results/shards/ab-$shard-run-manifest.json" + done + - name: Decide from fresh base-owned thresholds + if: env.PERFORMANCE_GATE_ENABLED == 'true' id: decision shell: bash run: | @@ -387,7 +612,7 @@ jobs: cat perf-results/decision.md >> "$GITHUB_STEP_SUMMARY" - name: Write artifact manifest - if: always() + if: always() && env.PERFORMANCE_GATE_ENABLED == 'true' shell: bash run: | python3 base/perf/ab/artifact_manifest.py \ @@ -397,7 +622,7 @@ jobs: --candidate-sha "$CANDIDATE_SHA" - name: Upload raw and analyzed evidence - if: always() + if: always() && env.PERFORMANCE_GATE_ENABLED == 'true' uses: actions/upload-artifact@v7 with: name: performance-ab-${{ env.PROFILE }}-${{ env.BASE_SHA }}-${{ env.CANDIDATE_SHA }} @@ -406,7 +631,7 @@ jobs: retention-days: 30 - name: Enforce three-state decision - if: always() + if: always() && env.PERFORMANCE_GATE_ENABLED == 'true' shell: bash run: | set -euo pipefail diff --git a/perf/ab/merge_manifests.py b/perf/ab/merge_manifests.py new file mode 100644 index 0000000..8a037c8 --- /dev/null +++ b/perf/ab/merge_manifests.py @@ -0,0 +1,427 @@ +#!/usr/bin/env python3 +"""Safely merge consecutive run-matrix shards into one gate-compatible manifest.""" + +from __future__ import annotations + +import argparse +import copy +import json +import pathlib +import shutil +import tempfile +from dataclasses import dataclass +from typing import Any + +import run_matrix + + +INVARIANT_FIELDS = ( + "schema_version", + "phase", + "profile", + "base_sha", + "candidate_sha", + "config_sha256", + "base_jar_sha256", + "candidate_jar_sha256", + "jmh", + "ci_overrides", + "candidate_isolation", +) + + +@dataclass(frozen=True) +class Shard: + path: pathlib.Path + digest: str + payload: dict[str, Any] + pair_start: int + pair_count: int + total_pairs: int + + @property + def pair_end(self) -> int: + return self.pair_start + self.pair_count + + +def require_integer(value: Any, name: str) -> int: + if isinstance(value, bool) or not isinstance(value, int): + raise ValueError(f"{name} must be an integer") + return value + + +def evidence_path( + manifest_path: pathlib.Path, + relative_value: Any, + expected_directory: str, + expected_suffix: str, +) -> pathlib.Path: + if not isinstance(relative_value, str) or not relative_value: + raise ValueError(f"manifest has an invalid {expected_directory} evidence path") + relative = pathlib.Path(relative_value) + if ( + relative.is_absolute() + or len(relative.parts) != 2 + or relative.parts[0] != expected_directory + or relative.suffix != expected_suffix + or any(part in ("", ".", "..") for part in relative.parts) + ): + raise ValueError(f"unsafe {expected_directory} evidence path: {relative_value}") + root = manifest_path.parent.resolve() + requested = root / relative + if requested.is_symlink() or requested.parent.is_symlink(): + raise ValueError(f"unsafe {expected_directory} evidence file: {relative_value}") + source = requested.resolve(strict=True) + if root != source.parent.parent or not source.is_file(): + raise ValueError(f"unsafe {expected_directory} evidence file: {relative_value}") + return source + + +def verify_digest(path: pathlib.Path, expected: Any, label: str) -> str: + if not isinstance(expected, str) or len(expected) != 64: + raise ValueError(f"{label} omitted a valid SHA-256 digest") + actual = run_matrix.sha256_file(path) + if actual != expected: + raise ValueError(f"{label} digest mismatch: {path}") + return actual + + +def load_shard(path: pathlib.Path) -> Shard: + requested = path.absolute() + if requested.is_symlink() or not requested.is_file(): + raise FileNotFoundError(f"shard manifest is not a regular file: {requested}") + manifest_path = requested.resolve() + payload = json.loads(manifest_path.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError(f"shard manifest is not an object: {manifest_path}") + if payload.get("schema_version") != run_matrix.SCHEMA_VERSION: + raise ValueError(f"unsupported shard manifest schema: {manifest_path}") + if not payload.get("created_at") or not payload.get("completed_at"): + raise ValueError(f"shard manifest is incomplete: {manifest_path}") + pair_range = payload.get("pair_range") + if not isinstance(pair_range, dict): + raise ValueError(f"shard manifest omitted pair_range: {manifest_path}") + pair_start = require_integer(pair_range.get("start"), "pair_range.start") + pair_count = require_integer(pair_range.get("count"), "pair_range.count") + pair_end = require_integer(pair_range.get("end_exclusive"), "pair_range.end_exclusive") + total_pairs = require_integer(pair_range.get("total_pairs"), "pair_range.total_pairs") + if pair_start < 0 or pair_count <= 0 or total_pairs <= 0: + raise ValueError(f"invalid pair_range in {manifest_path}") + if pair_end != pair_start + pair_count or pair_end > total_pairs: + raise ValueError(f"inconsistent pair_range in {manifest_path}") + return Shard( + path=manifest_path, + digest=run_matrix.sha256_file(manifest_path), + payload=payload, + pair_start=pair_start, + pair_count=pair_count, + total_pairs=total_pairs, + ) + + +def validate_shard_schedule(shard: Shard, full_schedule: list[dict[str, Any]]) -> None: + expected = run_matrix.select_pair_range( + full_schedule, + shard.pair_start, + shard.pair_count, + ) + if shard.payload.get("schedule") != expected: + raise ValueError(f"shard schedule does not match its declared pair range: {shard.path}") + executions = shard.payload.get("executions") + if not isinstance(executions, list) or len(executions) != len(expected): + raise ValueError(f"shard has incomplete executions: {shard.path}") + for local_index, (execution, scheduled) in enumerate(zip(executions, expected, strict=True)): + if not isinstance(execution, dict): + raise ValueError(f"shard execution {local_index} is not an object: {shard.path}") + for field in ("pair_index", "order", "position", "role"): + if execution.get(field) != scheduled[field]: + raise ValueError( + f"shard execution {local_index} changed scheduled {field}: {shard.path}" + ) + if execution.get("execution_index") != local_index: + raise ValueError(f"shard execution indices are not sequential: {shard.path}") + if ( + execution.get("exit_code") != 0 + or execution.get("fork_jvm_args_verified") is not True + or execution.get("validation_error") is not None + ): + raise ValueError(f"shard contains unvalidated execution {local_index}: {shard.path}") + if execution.get("result_mode_after_lock") != 0o444: + raise ValueError(f"shard result was not locked read-only: {shard.path}") + if execution.get("log_mode_after_lock") != 0o444: + raise ValueError(f"shard log was not locked read-only: {shard.path}") + + +def validate_shards(shards: list[Shard]) -> tuple[list[Shard], list[dict[str, Any]]]: + if not shards: + raise ValueError("at least one shard manifest is required") + if len({shard.path for shard in shards}) != len(shards): + raise ValueError("the same shard manifest was supplied more than once") + ordered = sorted(shards, key=lambda shard: shard.pair_start) + first = ordered[0] + if first.total_pairs % 2 != 0: + raise ValueError("total_pairs must contain equal forward and reverse orders") + full_schedule = run_matrix.build_schedule( + str(first.payload.get("phase")), + first.total_pairs // 2, + ) + cursor = 0 + runner_session_ids: set[str] = set() + for shard in ordered: + for field in INVARIANT_FIELDS: + if shard.payload.get(field) != first.payload.get(field): + raise ValueError(f"shard invariant {field} differs: {shard.path}") + if shard.total_pairs != first.total_pairs: + raise ValueError(f"shard invariant total_pairs differs: {shard.path}") + if shard.pair_start < cursor: + raise ValueError(f"shard pair range overlaps pair {shard.pair_start}: {shard.path}") + if shard.pair_start > cursor: + raise ValueError(f"missing pair range starting at {cursor}") + sharded = shard.pair_start != 0 or shard.pair_count != shard.total_pairs + runner_session_id = run_matrix.validate_runner_session_id( + shard.payload.get("runner_session_id"), + sharded, + ) + if runner_session_id is not None: + if runner_session_id in runner_session_ids: + raise ValueError(f"runner_session_id is reused across shards: {runner_session_id}") + runner_session_ids.add(runner_session_id) + validate_shard_schedule(shard, full_schedule) + cursor = shard.pair_end + if cursor != first.total_pairs: + raise ValueError(f"missing pair range starting at {cursor}") + return ordered, full_schedule + + +def copy_verified(source: pathlib.Path, target: pathlib.Path, expected_sha256: str) -> None: + verify_digest(source, expected_sha256, "source evidence") + target.parent.mkdir(parents=True, exist_ok=True) + with source.open("rb") as input_stream, target.open("xb") as output_stream: + shutil.copyfileobj(input_stream, output_stream, length=1024 * 1024) + target.chmod(0o444) + verify_digest(target, expected_sha256, "copied evidence") + + +def build_merged_manifest( + ordered: list[Shard], + full_schedule: list[dict[str, Any]], + destination: pathlib.Path, +) -> dict[str, Any]: + first = ordered[0] + merged_executions: list[dict[str, Any]] = [] + runtime_temp_dirs: set[str] = set() + evidence_names: set[tuple[str, str]] = set() + shard_provenance: list[dict[str, Any]] = [] + for shard_index, shard in enumerate(ordered): + shard_provenance.append( + { + "index": shard_index, + "manifest_sha256": shard.digest, + "pair_range": copy.deepcopy(shard.payload["pair_range"]), + "created_at": shard.payload.get("created_at"), + "completed_at": shard.payload.get("completed_at"), + "runner_session_id": shard.payload.get("runner_session_id"), + "environment": copy.deepcopy(shard.payload.get("environment", {})), + } + ) + for execution in shard.payload["executions"]: + runtime_temp_dir = execution.get("runtime_temp_dir") + if not isinstance(runtime_temp_dir, str) or not runtime_temp_dir: + raise ValueError(f"execution omitted runtime_temp_dir: {shard.path}") + if runtime_temp_dir in runtime_temp_dirs: + raise ValueError(f"runtime_temp_dir is reused across shards: {runtime_temp_dir}") + runtime_temp_dirs.add(runtime_temp_dir) + + result_source = evidence_path(shard.path, execution.get("result_file"), "raw", ".json") + log_source = evidence_path(shard.path, execution.get("log_file"), "logs", ".log") + result_name = ("raw", result_source.name) + log_name = ("logs", log_source.name) + for evidence_name in (result_name, log_name): + if evidence_name in evidence_names: + raise ValueError(f"duplicate evidence destination: {'/'.join(evidence_name)}") + evidence_names.add(evidence_name) + + result_digest = verify_digest( + result_source, + execution.get("result_sha256"), + "result evidence", + ) + log_digest = verify_digest( + log_source, + execution.get("log_sha256"), + "log evidence", + ) + copy_verified(result_source, destination / "raw" / result_source.name, result_digest) + copy_verified(log_source, destination / "logs" / log_source.name, log_digest) + + merged_execution = copy.deepcopy(execution) + merged_execution["source_manifest_sha256"] = shard.digest + merged_execution["source_execution_index"] = execution["execution_index"] + merged_execution["execution_index"] = len(merged_executions) + merged_execution["result_file"] = f"raw/{result_source.name}" + merged_execution["log_file"] = f"logs/{log_source.name}" + merged_executions.append(merged_execution) + + return { + "schema_version": first.payload["schema_version"], + "phase": first.payload["phase"], + "profile": first.payload["profile"], + "created_at": min(str(shard.payload.get("created_at")) for shard in ordered), + "completed_at": max(str(shard.payload.get("completed_at")) for shard in ordered), + "base_sha": first.payload["base_sha"], + "candidate_sha": first.payload["candidate_sha"], + "config_path": first.payload.get("config_path"), + "config_sha256": first.payload["config_sha256"], + "base_jar_sha256": first.payload["base_jar_sha256"], + "candidate_jar_sha256": first.payload["candidate_jar_sha256"], + "environment": { + "sharded": len(ordered) > 1, + "shards": [copy.deepcopy(entry["environment"]) for entry in shard_provenance], + }, + "jmh": copy.deepcopy(first.payload["jmh"]), + "ci_overrides": copy.deepcopy(first.payload.get("ci_overrides", {})), + "candidate_isolation": copy.deepcopy(first.payload["candidate_isolation"]), + "runner_session_ids": [ + entry["runner_session_id"] + for entry in shard_provenance + if entry["runner_session_id"] is not None + ], + "pair_range": { + "start": 0, + "count": first.total_pairs, + "end_exclusive": first.total_pairs, + "total_pairs": first.total_pairs, + }, + "schedule": copy.deepcopy(full_schedule), + "executions": merged_executions, + "source_shards": shard_provenance, + } + + +def shard_session_index(payload: dict[str, Any], label: str) -> dict[tuple[int, int, int], str]: + source_shards = payload.get("source_shards") + if not isinstance(source_shards, list) or not source_shards: + raise ValueError(f"{label} manifest has no source_shards metadata") + sessions: dict[tuple[int, int, int], str] = {} + seen_session_ids: set[str] = set() + for entry in source_shards: + if not isinstance(entry, dict) or not isinstance(entry.get("pair_range"), dict): + raise ValueError(f"{label} manifest has malformed source_shards metadata") + pair_range = entry["pair_range"] + start = require_integer(pair_range.get("start"), f"{label} pair_range.start") + count = require_integer(pair_range.get("count"), f"{label} pair_range.count") + total = require_integer(pair_range.get("total_pairs"), f"{label} pair_range.total_pairs") + session_id = run_matrix.validate_runner_session_id( + entry.get("runner_session_id"), + True, + ) + key = (start, count, total) + if key in sessions: + raise ValueError(f"{label} manifest repeats pair range {key}") + if session_id in seen_session_ids: + raise ValueError(f"{label} manifest reuses runner session {session_id}") + sessions[key] = session_id + seen_session_ids.add(session_id) + return sessions + + +def validate_cross_phase_sessions( + first_payload: dict[str, Any], + second_payload: dict[str, Any], +) -> None: + by_phase = { + first_payload.get("phase"): first_payload, + second_payload.get("phase"): second_payload, + } + if set(by_phase) != {"aa", "ab"}: + raise ValueError("cross-phase validation requires one A/A and one A/B manifest") + aa_payload = by_phase["aa"] + ab_payload = by_phase["ab"] + for field in ( + "schema_version", + "profile", + "base_sha", + "candidate_sha", + "config_sha256", + "base_jar_sha256", + "jmh", + "ci_overrides", + ): + if aa_payload.get(field) != ab_payload.get(field): + raise ValueError(f"cross-phase invariant {field} differs") + aa_sessions = shard_session_index(aa_payload, "A/A") + ab_sessions = shard_session_index(ab_payload, "A/B") + if aa_sessions.keys() != ab_sessions.keys(): + raise ValueError("A/A and A/B shard pair ranges differ") + for pair_range, aa_session in aa_sessions.items(): + if ab_sessions[pair_range] != aa_session: + raise ValueError( + f"A/A and A/B runner sessions differ for pair range {pair_range}" + ) + + +def read_merged_manifest(path: pathlib.Path) -> dict[str, Any]: + requested = path.absolute() + if requested.is_symlink() or not requested.is_file(): + raise FileNotFoundError(f"peer manifest is not a regular file: {requested}") + payload = json.loads(requested.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError(f"peer manifest is not an object: {requested}") + return payload + + +def merge( + manifest_paths: list[pathlib.Path], + output_dir: pathlib.Path, + peer_manifest: pathlib.Path | None = None, +) -> pathlib.Path: + output = output_dir.resolve() + if output.exists(): + raise FileExistsError(f"merge output already exists: {output}") + output.parent.mkdir(parents=True, exist_ok=True) + ordered, full_schedule = validate_shards([load_shard(path) for path in manifest_paths]) + temporary = pathlib.Path( + tempfile.mkdtemp(prefix=f".{output.name}.merge-", dir=output.parent) + ) + try: + manifest = build_merged_manifest(ordered, full_schedule, temporary) + if peer_manifest is not None: + validate_cross_phase_sessions(read_merged_manifest(peer_manifest), manifest) + run_matrix.write_json(temporary / "run-manifest.json", manifest) + temporary.replace(output) + except Exception: + shutil.rmtree(temporary, ignore_errors=True) + raise + return output / "run-manifest.json" + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--manifest", + dest="manifests", + action="append", + type=pathlib.Path, + required=True, + help="shard run-manifest.json (repeat for every consecutive shard)", + ) + parser.add_argument("--output-dir", type=pathlib.Path, required=True) + parser.add_argument( + "--peer-manifest", + type=pathlib.Path, + help="already-merged opposite phase; validates identical per-shard runner sessions", + ) + return parser.parse_args() + + +if __name__ == "__main__": + try: + arguments = parse_args() + raise SystemExit( + 0 + if merge(arguments.manifests, arguments.output_dir, arguments.peer_manifest) + else 1 + ) + except Exception as error: + print(f"error: {error}") + raise SystemExit(1) from error diff --git a/perf/ab/run_matrix.py b/perf/ab/run_matrix.py index e1e2264..c0bf8a0 100644 --- a/perf/ab/run_matrix.py +++ b/perf/ab/run_matrix.py @@ -23,6 +23,7 @@ SCHEMA_VERSION = 1 TIME_PATTERN = re.compile(r"^[1-9][0-9]*(?:ns|us|ms|s)$") USER_PATTERN = re.compile(r"^[a-z_][a-z0-9_-]{0,31}$") +RUNNER_SESSION_PATTERN = re.compile(r"^[A-Za-z0-9._:-]{1,128}$") def sha256_file(path: pathlib.Path) -> str: @@ -72,6 +73,48 @@ def build_schedule(phase: str, repetitions_per_order: int) -> list[dict[str, Any return schedule +def select_pair_range( + schedule: list[dict[str, Any]], + pair_start: int = 0, + pair_count: int | None = None, +) -> list[dict[str, Any]]: + """Select a consecutive range of complete pairs from a full matrix schedule.""" + pair_indices = sorted({int(entry["pair_index"]) for entry in schedule}) + if pair_indices != list(range(len(pair_indices))): + raise ValueError("schedule pair indices must be sequential from zero") + total_pairs = len(pair_indices) + if pair_start < 0 or pair_start >= total_pairs: + raise ValueError(f"pair_start must be between 0 and {total_pairs - 1}") + effective_count = total_pairs - pair_start if pair_count is None else pair_count + if effective_count <= 0: + raise ValueError("pair_count must be positive") + pair_end = pair_start + effective_count + if pair_end > total_pairs: + raise ValueError( + f"pair range [{pair_start}, {pair_end}) exceeds the {total_pairs}-pair schedule" + ) + selected = [ + entry + for entry in schedule + if pair_start <= int(entry["pair_index"]) < pair_end + ] + expected_entries = effective_count * 2 + if len(selected) != expected_entries: + raise ValueError( + f"pair range [{pair_start}, {pair_end}) contains {len(selected)} entries, " + f"expected {expected_entries}" + ) + return selected + + +def validate_runner_session_id(runner_session_id: str | None, sharded: bool) -> str | None: + if sharded and not runner_session_id: + raise ValueError("runner_session_id is required for a partial pair shard") + if runner_session_id is not None and not RUNNER_SESSION_PATTERN.fullmatch(runner_session_id): + raise ValueError("runner_session_id contains unsupported characters or length") + return runner_session_id + + def java_version(java: str) -> str: completed = subprocess.run( [java, "-version"], @@ -346,7 +389,14 @@ def run(args: argparse.Namespace) -> int: log_dir.mkdir(parents=True, exist_ok=True) manifest_path = output_dir / "run-manifest.json" repetitions = config["matrix"]["repetitions_per_order"] - schedule = build_schedule(args.phase, repetitions) + full_schedule = build_schedule(args.phase, repetitions) + schedule = select_pair_range(full_schedule, args.pair_start, args.pair_count) + total_pairs = repetitions * 2 + selected_pair_indices = sorted({int(entry["pair_index"]) for entry in schedule}) + selected_pair_start = selected_pair_indices[0] + selected_pair_count = len(selected_pair_indices) + sharded = selected_pair_start != 0 or selected_pair_count != total_pairs + runner_session_id = validate_runner_session_id(args.runner_session_id, sharded) jar_by_role = { "base": base_jar, "candidate": candidate_jar, @@ -391,6 +441,13 @@ def run(args: argparse.Namespace) -> int: "command_prefix": candidate_prefix, "file_user": candidate_file_user, }, + "runner_session_id": runner_session_id, + "pair_range": { + "start": selected_pair_start, + "count": selected_pair_count, + "end_exclusive": selected_pair_start + selected_pair_count, + "total_pairs": total_pairs, + }, "schedule": schedule, "executions": [], } @@ -514,6 +571,21 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--measurement-iterations", type=int) parser.add_argument("--warmup-time") parser.add_argument("--measurement-time") + parser.add_argument( + "--pair-start", + type=int, + default=0, + help="zero-based first pair to execute (default: first pair)", + ) + parser.add_argument( + "--pair-count", + type=int, + help="number of consecutive complete pairs to execute (default: all remaining pairs)", + ) + parser.add_argument( + "--runner-session-id", + help="runner identity shared by this shard's A/A and A/B phases (required for partial shards)", + ) parser.add_argument("--candidate-command-prefix-json") parser.add_argument("--candidate-file-user") return parser.parse_args() diff --git a/perf/ab/tests/test_gate.py b/perf/ab/tests/test_gate.py index eed01ce..1044894 100644 --- a/perf/ab/tests/test_gate.py +++ b/perf/ab/tests/test_gate.py @@ -13,6 +13,7 @@ sys.path.insert(0, str(AB_DIR)) import gate # noqa: E402 +import merge_manifests # noqa: E402 import run_matrix # noqa: E402 import verify_protected # noqa: E402 @@ -306,6 +307,41 @@ def test_aa_schedule_is_order_balanced(self) -> None: orders = [schedule[index]["order"] for index in range(0, len(schedule), 2)] self.assertEqual(["A1A2", "A2A1"] * 4, orders) + def test_consecutive_pair_shards_preserve_complete_global_pairs(self) -> None: + full = run_matrix.build_schedule("ab", 4) + shards = [run_matrix.select_pair_range(full, start, 1) for start in range(8)] + self.assertEqual(full, [entry for shard in shards for entry in shard]) + for shard_index, shard in enumerate(shards): + self.assertEqual( + [shard_index], + sorted({entry["pair_index"] for entry in shard}), + ) + self.assertEqual("AB" if shard_index % 2 == 0 else "BA", shard[0]["order"]) + + def test_pair_range_defaults_to_every_remaining_complete_pair(self) -> None: + full = run_matrix.build_schedule("ab", 4) + self.assertEqual(full, run_matrix.select_pair_range(full)) + tail = run_matrix.select_pair_range(full, 6) + self.assertEqual([6, 7], sorted({entry["pair_index"] for entry in tail})) + + def test_pair_range_rejects_empty_negative_or_overflowing_slices(self) -> None: + full = run_matrix.build_schedule("ab", 4) + for start, count in ((-1, 2), (0, 0), (7, 2), (8, 1)): + with self.subTest(start=start, count=count): + with self.assertRaises(ValueError): + run_matrix.select_pair_range(full, start, count) + + def test_partial_shard_requires_a_safe_runner_session_id(self) -> None: + with self.assertRaisesRegex(ValueError, "required for a partial pair shard"): + run_matrix.validate_runner_session_id(None, True) + self.assertEqual( + "run-42.shard-1", + run_matrix.validate_runner_session_id("run-42.shard-1", True), + ) + self.assertIsNone(run_matrix.validate_runner_session_id(None, False)) + with self.assertRaisesRegex(ValueError, "unsupported characters"): + run_matrix.validate_runner_session_id("bad session", True) + def test_pending_ray_profile_is_rejected_without_production_classes(self) -> None: config = json.loads((AB_DIR / "gate-config.json").read_text(encoding="utf-8")) with tempfile.TemporaryDirectory() as temporary: @@ -409,6 +445,220 @@ def test_fork_log_must_confirm_every_fixed_jvm_option(self) -> None: run_matrix.verify_fork_jvm_args(log, ["-Xms1g", "-Dfile.encoding=UTF-8"]) +class ManifestMergeTest(unittest.TestCase): + @staticmethod + def write_shard( + root: pathlib.Path, + phase: str, + pair_start: int, + pair_count: int, + runner_session_id: str, + *, + total_pairs: int = 8, + candidate_sha: str = "candidate", + directory_name: str | None = None, + ) -> pathlib.Path: + directory = root / (directory_name or f"{phase}-{pair_start}") + raw = directory / "raw" + logs = directory / "logs" + raw.mkdir(parents=True) + logs.mkdir() + full_schedule = run_matrix.build_schedule(phase, total_pairs // 2) + schedule = run_matrix.select_pair_range(full_schedule, pair_start, pair_count) + executions = [] + for local_index, scheduled in enumerate(schedule): + role = scheduled["role"] + stem = ( + f"{phase}-pair-{scheduled['pair_index'] + 1:02d}-" + f"pos-{scheduled['position'] + 1}-{role}" + ) + result = raw / f"{stem}.json" + result.write_text("[]\n", encoding="utf-8") + log = logs / f"{stem}.log" + log.write_text("# VM options: -Xms1g -Xmx1g\n", encoding="utf-8") + is_candidate = role == "candidate" + executions.append( + { + **scheduled, + "execution_index": local_index, + "revision_sha": candidate_sha if is_candidate else "base", + "jar_sha256": "candidate-jar" if is_candidate else "base-jar", + "fork_jvm_args_verified": True, + "validation_error": None, + "isolated_candidate": is_candidate, + "result_mode_after_lock": 0o444, + "log_mode_after_lock": 0o444, + "started_at": "2026-01-01T00:00:00Z", + "completed_at": "2026-01-01T00:00:01Z", + "elapsed_seconds": 1.0, + "exit_code": 0, + "result_file": result.relative_to(directory).as_posix(), + "result_sha256": run_matrix.sha256_file(result), + "log_file": log.relative_to(directory).as_posix(), + "log_sha256": run_matrix.sha256_file(log), + "runtime_temp_dir": f"/tmp/{runner_session_id}/{stem}", + "command": ["java", f"-Djava.io.tmpdir=/tmp/{runner_session_id}/{stem}"], + } + ) + manifest = { + "schema_version": 1, + "phase": phase, + "profile": "infra", + "created_at": "2026-01-01T00:00:00Z", + "completed_at": "2026-01-01T00:01:00Z", + "base_sha": "base", + "candidate_sha": candidate_sha, + "config_path": "/base/perf/ab/gate-config.json", + "config_sha256": "base-config", + "base_jar_sha256": "base-jar", + "candidate_jar_sha256": "base-jar" if phase == "aa" else "candidate-jar", + "environment": {"runner": runner_session_id}, + "jmh": {"mode": "avgt", "jvm_args": ["-Xms1g", "-Xmx1g"]}, + "ci_overrides": {}, + "candidate_isolation": { + "enabled": True, + "command_prefix": ["sudo", "-u", "candidate", "--"], + "file_user": "candidate", + }, + "runner_session_id": runner_session_id, + "pair_range": { + "start": pair_start, + "count": pair_count, + "end_exclusive": pair_start + pair_count, + "total_pairs": total_pairs, + }, + "schedule": schedule, + "executions": executions, + } + manifest_path = directory / "run-manifest.json" + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + return manifest_path + + @classmethod + def write_matrix_shards( + cls, + root: pathlib.Path, + phase: str, + *, + prefix: str = "", + sessions: list[str] | None = None, + omit: set[int] | None = None, + candidate_sha_overrides: dict[int, str] | None = None, + ) -> list[pathlib.Path]: + runner_sessions = sessions or [f"runner-{index}" for index in range(8)] + if len(runner_sessions) != 8: + raise ValueError("tests require one runner session for each of eight pairs") + omitted = omit or set() + overrides = candidate_sha_overrides or {} + return [ + cls.write_shard( + root, + phase, + pair_index, + 1, + runner_sessions[pair_index], + candidate_sha=overrides.get(pair_index, "candidate"), + directory_name=f"{prefix}{phase}-{pair_index}", + ) + for pair_index in range(8) + if pair_index not in omitted + ] + + def test_merges_consecutive_shards_into_one_gate_compatible_manifest(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + root = pathlib.Path(temporary) + manifests = self.write_matrix_shards(root, "ab") + output = root / "merged-ab" + merged_path = merge_manifests.merge(manifests, output) + merged = json.loads(merged_path.read_text(encoding="utf-8")) + + self.assertEqual( + {"start": 0, "count": 8, "end_exclusive": 8, "total_pairs": 8}, + merged["pair_range"], + ) + self.assertEqual( + [f"runner-{index}" for index in range(8)], + merged["runner_session_ids"], + ) + self.assertEqual(list(range(16)), [entry["execution_index"] for entry in merged["executions"]]) + self.assertEqual( + [pair_index for pair_index in range(8) for _ in range(2)], + [entry["pair_index"] for entry in merged["executions"]], + ) + self.assertEqual(16, len(list((output / "raw").glob("*.json")))) + self.assertEqual(16, len(list((output / "logs").glob("*.log")))) + + def test_rejects_duplicate_or_missing_pair_ranges(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + root = pathlib.Path(temporary) + duplicate = [ + self.write_shard(root, "ab", 0, 1, "runner-0", directory_name="first"), + self.write_shard(root, "ab", 0, 1, "runner-1", directory_name="duplicate"), + ] + with self.assertRaisesRegex(ValueError, "overlaps"): + merge_manifests.merge(duplicate, root / "duplicate-output") + + missing = self.write_matrix_shards(root, "ab", prefix="missing-", omit={2}) + with self.assertRaisesRegex(ValueError, "missing pair range starting at 2"): + merge_manifests.merge(missing, root / "missing-output") + + def test_rejects_cross_shard_identity_changes_and_reused_runner_sessions(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + root = pathlib.Path(temporary) + changed_identity = self.write_matrix_shards( + root, + "ab", + prefix="identity-", + candidate_sha_overrides={7: "different-candidate"}, + ) + with self.assertRaisesRegex(ValueError, "candidate_sha"): + merge_manifests.merge(changed_identity, root / "identity-output") + + reused_session = self.write_matrix_shards( + root, + "ab", + prefix="session-", + sessions=["same-runner", "same-runner", *[f"runner-{index}" for index in range(2, 8)]], + ) + with self.assertRaisesRegex(ValueError, "reused across shards"): + merge_manifests.merge(reused_session, root / "session-output") + + def test_cross_phase_validation_requires_the_same_session_for_each_pair_range(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + root = pathlib.Path(temporary) + aa = merge_manifests.merge( + self.write_matrix_shards(root, "aa", prefix="good-"), + root / "merged-aa", + ) + ab = merge_manifests.merge( + self.write_matrix_shards(root, "ab", prefix="good-"), + root / "merged-ab", + aa, + ) + merge_manifests.validate_cross_phase_sessions( + json.loads(aa.read_text(encoding="utf-8")), + json.loads(ab.read_text(encoding="utf-8")), + ) + + mismatched = self.write_matrix_shards( + root, + "ab", + prefix="bad-", + sessions=[*[f"runner-{index}" for index in range(7)], "other-runner"], + ) + with self.assertRaisesRegex(ValueError, "runner sessions differ"): + merge_manifests.merge(mismatched, root / "bad-merged-ab", aa) + + def test_rejects_tampered_raw_evidence(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + root = pathlib.Path(temporary) + manifests = self.write_matrix_shards(root, "ab") + raw_file = next((manifests[0].parent / "raw").glob("*.json")) + raw_file.write_text("tampered\n", encoding="utf-8") + with self.assertRaisesRegex(ValueError, "digest mismatch"): + merge_manifests.merge(manifests, root / "tampered-output") + + class ProtectedPathTest(unittest.TestCase): def test_candidate_change_is_detected(self) -> None: with tempfile.TemporaryDirectory() as temporary: From 077c57f2e312b522d2f5790c0dab36afc1f75802 Mon Sep 17 00:00:00 2001 From: Arbousier1 Date: Fri, 17 Jul 2026 00:25:22 +0800 Subject: [PATCH 2/3] docs(perf): document parallel A/B evidence --- CHANGELOG.md | 4 ++-- docs/performance-testing.md | 34 +++++++++++++++++++++------------- 2 files changed, 23 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7117e08..2d45a51 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,7 +14,7 @@ Feature release for table interaction, river viewing, rules accuracy, storage in - **三种玩法校正**：GB 模式对齐 144 张牌、花牌、8 个非花番门槛、单和与固定 16 局流程；立直模式修正鸣牌优先级、振听、杠宝牌时机、流局与延长赛边界；四川模式对齐 T/TFMJ 的直接定缺、无换三张、三番封顶、血战到底、杠分转移和查叫流程。 - **段位与服务器集成**：新增可选 InvSync 2.x 玩家段位存储，并在 InvSync 缺失或不兼容时仅向已启用且健康的 SQL 后端回退。当前自动段位与统计更新仍只适用于四名真人完成的立直对局。 - **资源与本地化**：加入 GB、立直和四川模式对应的摸牌、出牌、鸣牌、流局与和牌音效；新增完整日语游戏消息、安装文档和三种玩法说明。 -- **性能与稳定性**：减少牌桌区域更新、展示实体复用、观察者集合和 GB Bot 决策热路径中的排序、装箱及临时分配；缓存调度器反射解析，并加入 GitHub Actions 配对 A/B 性能回归门禁。 +- **性能与稳定性**：减少牌桌区域更新、展示实体复用、观察者集合和 GB Bot 决策热路径中的排序、装箱及临时分配；缓存调度器反射解析，并加入由八个 runner 并行采样、可信汇总的 GitHub Actions 配对 A/B 性能回归门禁。 - **分发说明**：发行 JAR 现在内嵌并重定位 Sparrow Reflection 与 ASM。第三方声明和运行软件分发条款已更新，重新分发时应保留对应许可证与源码义务说明。 English Release Notes: @@ -27,7 +27,7 @@ English Release Notes: - **Rules corrections**: GB now follows the 144-tile flower flow, eight non-flower-fan floor, single-ron priority, and fixed 16-hand profile. Riichi fixes cover reaction priority, furiten, kan-dora timing, draws, and match extensions. Sichuan now follows the documented T/TFMJ direct ding-que, no exchange-three, three-fan cap, Bloody Battle, kong-transfer, and cha-jiao flow. - **Rank and server integrations**: Added optional InvSync 2.x player-rank storage, with fallback only to an enabled and healthy SQL backend. Automatic rank and personal-stat updates remain limited to completed four-human Riichi matches. - **Assets and localization**: Added mode-specific draw, discard, reaction, draw-result, and win sounds for GB, Riichi, and Sichuan, plus complete Japanese game messages and player documentation. -- **Performance and reliability**: Reduced sorting, boxing, and temporary allocations in region updates, display reconciliation, viewer membership handling, and GB bot decisions; cached scheduler reflection resolution and added paired GitHub Actions A/B performance gates. +- **Performance and reliability**: Reduced sorting, boxing, and temporary allocations in region updates, display reconciliation, viewer membership handling, and GB bot decisions; cached scheduler reflection resolution and added a paired GitHub Actions A/B gate with eight parallel measurement runners and trusted aggregation. - **Distribution notice**: Release jars now embed and relocate Sparrow Reflection and ASM. Third-party notices and runnable-distribution terms were updated; redistributors must preserve the applicable license and source-obligation notices. ## 1.4.1 - 2026-06-18 diff --git a/docs/performance-testing.md b/docs/performance-testing.md index 10a4ccd..41dc009 100644 --- a/docs/performance-testing.md +++ b/docs/performance-testing.md @@ -73,8 +73,10 @@ time and normalized allocation remain focused on production code. The infrastruc fingerprint profile applies the same rule to its exact delimited string and now treats normalized allocation as a secondary guardrail. -The workflow is loaded through `pull_request_target`, uses only `contents: read`, persists no -checkout credentials, disables Gradle's shared cache, and clears GitHub/Actions runtime +The workflow is loaded through `pull_request_target` and executes candidate bytecode only for +same-repository optimization branches; forked PRs are not eligible for this trusted gate. It +uses only `contents: read`, persists no checkout credentials, disables Gradle's shared cache, +and clears GitHub/Actions runtime credentials from every shell step that executes candidate bytecode. Candidate JMH runs use a dedicated no-login UID: only the current candidate JSON path receives candidate ownership, while stdout is confined to the current runner-opened log and the jars, parent directories, @@ -100,21 +102,26 @@ classes, `performance-ray-proxy` fails preflight instead of benchmarking the fal Once present, the same protected benchmark automatically invokes the real package-private coordinator and verifies that an unchanged second replace emits no additional logical spawns. -The workflow uses two fresh GitHub-hosted runners: +The workflow builds once, fans out to eight independent measurement runners, and finishes on +a fresh trusted decision runner: -1. Check out the PR base and candidate commits side by side. +1. In the prepare job, check out the PR base and candidate commits side by side. 2. Byte-compare all `protected_paths` from the base config. The candidate cannot change the benchmark source, Gradle harness, wrapper, decision scripts, workflow or thresholds. -3. Build one JMH jar from each revision with identical Java/Paper settings. -4. Run an order-balanced A/A control: four `A1,A2` pairs and four `A2,A1` pairs, - interleaved in ABBA execution order. -5. Run exactly four `base,candidate` and four `candidate,base` pairs in the same interleaved - ABBA order, with candidate bytecode confined to the unprivileged UID. -6. Upload the raw evidence and end the runner that executed candidate code. -7. On a fresh runner, check out the base and candidate again and repeat the protected-path +3. Build one JMH jar from each revision with identical Java/Paper settings and upload those + immutable inputs once. +4. Start eight matrix shards in parallel. Each shard owns exactly one pair index and runs both + its A/A control pair and its A/B pair on the same runner under one recorded runner session. +5. Even pair indices run `A1,A2` then `base,candidate`; odd indices run `A2,A1` then + `candidate,base`. Candidate bytecode remains confined to the unprivileged UID. Matrix + completion order is irrelevant; the trusted merger restores deterministic pair-index order. +6. Upload each runner's isolated raw evidence separately and end every runner that executed + candidate code. +7. On a fresh decision runner, check out the base and candidate again, repeat the protected-path comparison. -8. Analyze with the fresh base-owned gate, then upload every raw result, log, run manifest, - digest and decision for review. +8. Reject missing, duplicate, overlapping, cross-session or digest-invalid shards; merge the + complete 0..7 set and analyze it with the fresh base-owned gate. +9. Upload every raw result, log, shard/merged manifest, digest and decision for review. If a new benchmark or threshold is needed, merge that harness change first. The subsequent optimization PR must not contain benchmark-infrastructure changes. This split prevents a @@ -170,6 +177,7 @@ gate-config.json exact base-owned policy snapshot protected-paths.json byte-comparison result jars/base-jmh.jar exact retained base benchmark bytecode jars/candidate-jmh.jar exact retained candidate benchmark bytecode +shards/*-run-manifest.json original runner-local manifests and digests aa/run-manifest.json A/A schedule, environment, commands and hashes aa/raw/*.json raw JMH output for every control execution aa/logs/*.log complete JMH logs From 061106e7bb44c4ceaaf53ad965451108e1703813 Mon Sep 17 00:00:00 2001 From: Arbousier1 Date: Fri, 17 Jul 2026 00:27:45 +0800 Subject: [PATCH 3/3] fix(ci): use a job-safe prepared path --- .github/workflows/performance-ab.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/performance-ab.yml b/.github/workflows/performance-ab.yml index 4700a18..7876ad5 100644 --- a/.github/workflows/performance-ab.yml +++ b/.github/workflows/performance-ab.yml @@ -294,7 +294,7 @@ jobs: CANDIDATE_SHA: ${{ github.event.pull_request.head.sha || inputs.candidate_sha }} PROFILE: ${{ needs.prepare.outputs.profile }} PAIR_INDEX: ${{ matrix.pair_index }} - PREPARED_ROOT: ${{ runner.temp }}/performance-ab-prepared + PREPARED_ROOT: /tmp/mahjong-performance-prepared-${{ github.run_id }}-${{ github.run_attempt }}-${{ matrix.pair_index }} PERF_RESULTS_ROOT: /tmp/mahjong-performance-ab-${{ github.run_id }}-${{ github.run_attempt }}-${{ matrix.pair_index }} CI_FORKS: ${{ inputs.forks }} CI_WARMUPS: ${{ inputs.warmup_iterations }}