From 28ac3ab4e7507a78836a673e21507092f82240bf Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 9 Sep 2026 18:30:08 +0800 Subject: [PATCH] test: add scanner heal scheduler pressure evidence (#7584) Add measured G10/P1/P3 scheduler-pressure release descriptor generation, ABBA scheduler/profile provenance, and focused self-tests for the release bundle gate. Co-authored-by: zhi22915 --- .config/make/tests.mak | 1 + scripts/README.md | 2 + ...canner_heal_scheduler_pressure_evidence.py | 590 ++++++++++++++++++ scripts/scanner_abba.py | 45 +- scripts/summarize_scanner_heal_perf.py | 413 ++++++++++++ scripts/test_scanner_abba.py | 26 + ...canner_heal_scheduler_pressure_evidence.sh | 8 + scripts/test_summarize_scanner_heal_perf.py | 106 +++- 8 files changed, 1186 insertions(+), 5 deletions(-) create mode 100755 scripts/run_scanner_heal_scheduler_pressure_evidence.py create mode 100755 scripts/test_scanner_heal_scheduler_pressure_evidence.sh diff --git a/.config/make/tests.mak b/.config/make/tests.mak index f4875ccf9..85a6e6a56 100644 --- a/.config/make/tests.mak +++ b/.config/make/tests.mak @@ -32,6 +32,7 @@ script-tests: ## Run shell script tests ./scripts/test_hotpath_warp_ab_gate.sh ./scripts/test_hotpath_warp_abba.sh ./scripts/test_scanner_validation_harness.sh + ./scripts/test_scanner_heal_scheduler_pressure_evidence.sh ./scripts/test_scanner_heal_w13_mrf_evidence.sh ./scripts/test_scanner_heal_w16_recovery_evidence.sh ./scripts/test_exact_1mib_handoff_abba.sh diff --git a/scripts/README.md b/scripts/README.md index 1a04cd433..7d9b03c7a 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -57,9 +57,11 @@ their issue closes. | `run_scanner_validation_harness.sh` | dev-tool | Scanner validation harness | `docs/operations/scanner-benchmark-runbook.md` | | `run_scanner_heal_evidence_case.sh` | dev-tool | Runs one Scanner/Heal release-evidence registry case and checks the produced receipt/oracle | `.config/scanner-heal-required-tests.json`; `check_test_wiring.py --check-scanner-heal` | | `run_scanner_heal_g09_upgrade_evidence.sh` | dev-tool | Runs the G09 mixed-version and rollback upgrade E2E lanes against a pinned previous release and verifies the raw evidence artifacts | `docs/testing/ci-gates.md`; `.github/workflows/e2e-upgrade.yml`; `test_scanner_heal_g09_upgrade_evidence.sh` | +| `run_scanner_heal_scheduler_pressure_evidence.py` | dev-tool | Assembles measured Scanner/Heal G10/P1/P3 scheduler-pressure release descriptors from a completed measured ABBA run, recovery-window proof, and profile artifacts | `docs/operations/scanner-benchmark-runbook.md`; `test_scanner_heal_scheduler_pressure_evidence.sh` | | `run_scanner_heal_w13_mrf_evidence.sh` | dev-tool | Runs the W13 durable MRF replay lanes and writes G07/G08/P4 bundle-ready evidence descriptors | `docs/testing/ci-gates.md`; `test_scanner_heal_w13_mrf_evidence.sh` | | `run_scanner_heal_w16_recovery_evidence.sh` | dev-tool | Runs the W16 recovery-intent and quota authority lanes and writes G04/G12 bundle-ready evidence descriptors | `docs/testing/ci-gates.md`; `test_scanner_heal_w16_recovery_evidence.sh` | | `test_scanner_validation_harness.sh` | dev-tool | Self-test for the scanner validation harness | — | +| `test_scanner_heal_scheduler_pressure_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal scheduler-pressure evidence assembler | — | | `test_scanner_heal_g09_upgrade_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal G09 upgrade evidence runner | — | | `test_scanner_heal_w16_recovery_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal W16 recovery evidence runner | — | | `test_scanner_heal_w13_mrf_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal W13 MRF evidence runner | — | diff --git a/scripts/run_scanner_heal_scheduler_pressure_evidence.py b/scripts/run_scanner_heal_scheduler_pressure_evidence.py new file mode 100755 index 000000000..c6dcf9e2b --- /dev/null +++ b/scripts/run_scanner_heal_scheduler_pressure_evidence.py @@ -0,0 +1,590 @@ +#!/usr/bin/env python3 +"""Assemble measured Scanner/Heal scheduler-pressure release evidence. + +This producer consumes a completed measured Scanner/Heal ABBA run plus measured +profile and recovery-window artifacts. It only packages existing measurements; +it never turns synthetic harness output into release evidence. +""" + +from __future__ import annotations + +import argparse +from datetime import datetime, timedelta, timezone +import json +import math +import shutil +import subprocess +import sys +from pathlib import Path +from typing import Any + +from scanner_abba import ( + RELEASE_PROFILE_ARTIFACTS, + RELEASE_SCHEDULER_BOUNDS, + SCENARIOS, + digest, + read_json, + require, + sha, + write_json, +) + +ROOT = Path(__file__).resolve().parents[1] +G10_FIELDS = ("scheduler_bound_evidence", "pressure_recovery_evidence") +P1_FIELDS = ("cold_walk_share_measurement", "foreground_latency_throughput_measurement", "profile_evidence") +P3_FIELDS = ("two_hour_pressure_measurement", "heal_capacity_measurement", "recovery_window_measurement") +PRESSURE_METRICS = ( + "foreground_p95_ms", + "foreground_p99_ms", + "throughput_ops", + "error_rate", + "heal_lock_wait_p99_ms", + "attempt_cost_samples", +) + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +def git_head() -> str: + return subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip() + + +def finite_number(value: Any, name: str, minimum: float = 0.0) -> float: + require(type(value) in (int, float) and math.isfinite(value) and value >= minimum, f"invalid {name}") + return float(value) + + +def positive_int(value: Any, name: str, minimum: int = 1) -> int: + require(type(value) is int and value >= minimum, f"invalid {name}") + return value + + +def positive_int_from_sources(cli_value: int | None, manifest_values: dict[str, Any], key: str) -> int: + if cli_value is not None: + return positive_int(cli_value, key) + return positive_int(manifest_values.get(key), key) + + +def parse_artifact_arg(value: str) -> tuple[str, Path, str]: + try: + kind, raw_path = value.split("=", 1) + except ValueError as err: + raise argparse.ArgumentTypeError("profile artifact must be KIND=PATH") from err + if kind not in RELEASE_PROFILE_ARTIFACTS: + raise argparse.ArgumentTypeError(f"unknown profile artifact kind: {kind}") + path = Path(raw_path).expanduser().resolve() + if not path.is_file() or path.stat().st_size == 0: + raise argparse.ArgumentTypeError(f"profile artifact is missing or empty: {raw_path}") + suffix = path.suffix.lower().lstrip(".") + artifact_format = f"profile-{suffix}" if suffix in {"json", "ndjson"} else (suffix if suffix else "binary") + return kind, path, artifact_format + + +def load_measured_abba(abba_dir: Path, source_revision: str) -> tuple[dict[str, Any], dict[str, Any], list[dict[str, Any]]]: + manifest = read_json(abba_dir / "manifest.json") + report = read_json(abba_dir / "report.json") + require(manifest.get("evidence") == "measured", "ABBA manifest must be measured") + require(report.get("evidence") == "measured", "ABBA report must be measured") + require(report.get("status") == "pass" and report.get("performance") == "pass", "ABBA report must pass") + require(manifest.get("candidate", {}).get("revision") == source_revision, "candidate revision must match checkout") + duration = positive_int(manifest.get("duration_seconds"), "duration_seconds", 7200) + require(duration >= 7200, "P3 two-hour pressure evidence requires at least 7200 seconds") + expected_cells = len(SCENARIOS) * 2 * positive_int(manifest.get("rounds"), "rounds") * 4 + require(report.get("cells") == expected_cells, "ABBA report did not complete the full matrix") + require(manifest.get("fixed", {}).get("offered_load_ops", 0) > 0, "missing fixed offered load") + release_evidence = manifest.get("release_evidence") + require(isinstance(release_evidence, dict), "ABBA manifest missing release evidence") + scheduler = release_evidence.get("scheduler") + require(isinstance(scheduler, dict), "ABBA manifest missing scheduler evidence") + require(scheduler.get("bounds") == list(RELEASE_SCHEDULER_BOUNDS), "ABBA scheduler bounds mismatch") + positive_int(scheduler.get("max_deferred_items"), "scheduler.max_deferred_items") + positive_int(scheduler.get("max_deferred_bytes"), "scheduler.max_deferred_bytes") + positive_int(scheduler.get("max_retry_age_seconds"), "scheduler.max_retry_age_seconds") + require(scheduler.get("duplicate_task_bound_observed") is True, "ABBA scheduler duplicate bound not observed") + + measures: list[dict[str, Any]] = [] + for measure_path in sorted(abba_dir.glob("*-*-*-*/measure.json")): + measure = read_json(measure_path) + require(measure.get("evidence") == "measured", f"{measure_path.name} is not measured") + require(measure.get("build", {}).get("revision") in { + manifest["baseline"]["revision"], + manifest["candidate"]["revision"], + }, "measure build revision is outside the manifest") + metrics = measure.get("metrics") + require(isinstance(metrics, dict), "measure missing metrics") + measures.append(measure) + require(len(measures) == expected_cells, "missing measured cell outputs") + return manifest, report, measures + + +def comparison_rows(report: dict[str, Any], scenario: str, comparison: str) -> list[dict[str, Any]]: + rows = [ + item for item in report.get("comparisons", []) + if item.get("scenario") == scenario and item.get("comparison") == comparison + ] + require(rows, f"missing {scenario}/{comparison} comparisons") + return rows + + +def worst_metric(measures: list[dict[str, Any]], key: str) -> float: + return max(finite_number(item["metrics"].get(key), key) for item in measures) + + +def sum_metric(measures: list[dict[str, Any]], key: str) -> float: + return sum(finite_number(item["metrics"].get(key), key) for item in measures) + + +def copy_profile_artifacts(out_dir: Path, artifacts: dict[str, tuple[Path, str]], source_revision: str, + run_id: str, window_id: str) -> dict[str, Any]: + copied: dict[str, Any] = {} + profile_dir = out_dir / "artifacts" / "profiles" + profile_dir.mkdir(parents=True, exist_ok=True) + for kind in RELEASE_PROFILE_ARTIFACTS: + source, artifact_format = artifacts[kind] + target = profile_dir / f"P1-profile_evidence-{kind}{source.suffix or '.artifact'}" + shutil.copyfile(source, target) + copied[kind] = { + "artifact": target.relative_to(out_dir).as_posix(), + "sha256": digest(target), + "artifact_format": artifact_format, + "source_revision": source_revision, + "run_id": run_id, + "measurement_window_id": window_id, + } + return copied + + +def write_field(out_dir: Path, gate: str, field: str, evidence: dict[str, Any]) -> dict[str, Any]: + artifact = out_dir / "artifacts" / f"{gate}-{field}.json" + artifact.parent.mkdir(parents=True, exist_ok=True) + payload = { + "schema": 1, + "evidence_type": "measured", + "source_revision": evidence["source_revision"], + "run_id": evidence["run_id"], + "measurement_window_id": evidence["measurement_window_id"], + "gate": gate, + "field": field, + } + for key, value in evidence.items(): + if key not in {"artifact", "sha256", "artifact_format", "summary", "started_at", "finished_at", "command"}: + payload[key] = value + write_json(artifact, payload) + evidence["artifact"] = artifact.relative_to(out_dir).as_posix() + evidence["sha256"] = digest(artifact) + evidence["artifact_format"] = "json" + return evidence + + +def recovery_evidence(path: Path, source_revision: str) -> dict[str, Any]: + payload = read_json(path) + for marker in ("fixture", "fixture_only", "dry_run", "synthetic"): + require(payload.get(marker) is not True, f"recovery artifact is {marker}") + require(payload.get("evidence_type") == "measured", "recovery artifact must be measured") + require(payload.get("source_revision") == source_revision, "recovery artifact source revision mismatch") + require(payload.get("fault_modes") == ["process-restart", "process-crash-restart"], + "recovery artifact must cover both restart modes") + return payload + + +def build_descriptor(args: argparse.Namespace) -> Path: + out_dir = args.out_dir.resolve() + require(not out_dir.exists(), "output directory must be new") + source_revision = args.source_revision or git_head() + manifest, report, measures = load_measured_abba(args.abba_dir.resolve(), source_revision) + recovery = recovery_evidence(args.recovery_window_json.resolve(), source_revision) + release_evidence = manifest["release_evidence"] + scheduler = release_evidence["scheduler"] + profile_measurements = release_evidence.get("profile", {}).get("measurements", {}) + require(isinstance(profile_measurements, dict), "ABBA manifest profile measurements must be an object") + + profile_inputs = {} + profile_formats = {} + for item in args.profile_artifact: + kind, path, artifact_format = parse_artifact_arg(item) + require(kind not in profile_inputs, f"duplicate profile artifact kind: {kind}") + profile_inputs[kind] = path + profile_formats[kind] = artifact_format + missing_profiles = sorted(set(RELEASE_PROFILE_ARTIFACTS) - set(profile_inputs)) + require(not missing_profiles, "missing profile artifacts: " + ", ".join(missing_profiles)) + + out_dir.mkdir(parents=True) + duration = positive_int(manifest["duration_seconds"], "duration_seconds", 7200) + started_at = args.started_at or utc_now() + if args.finished_at: + finished_at = args.finished_at + else: + started = datetime.fromisoformat(started_at.replace("Z", "+00:00")) + finished_at = (started + timedelta(seconds=duration)).isoformat().replace("+00:00", "Z") + run_id = args.run_id or f"scheduler-pressure-{source_revision[:12]}" + window_id = args.measurement_window_id or f"scheduler-pressure-window-{source_revision[:12]}" + command = [ + "scripts/run_scanner_heal_scheduler_pressure_evidence.py", + "--abba-dir", "", + "--recovery-window-json", "", + "--profile-artifact", "", + ] + running_heal = comparison_rows(report, "running-heal", "build") + require(any(row.get("w10", {}).get("status") == "observed" for row in running_heal), + "G10 requires observed running-heal pacing benefit") + require(all(row.get("w11", {}).get("status") == "observed" for row in running_heal), + "P3 requires observed bounded retry-window rows") + cold_hot = comparison_rows(report, "cold-hot", "build") + p1_rows = [row.get("p1") for row in cold_hot] + require(all(isinstance(row, dict) and row.get("observed_reduction", -1) >= row.get("required_reduction", 1) + for row in p1_rows), "P1 cold-hot rows did not meet required reduction") + + common = { + "evidence_type": "measured", + "source_revision": source_revision, + "run_id": run_id, + "measurement_window_id": window_id, + "started_at": started_at, + "finished_at": finished_at, + "command": command, + } + total_walk = int(sum_metric(measures, "walk_objects")) + total_cold = int(sum_metric(measures, "cold_walk_objects")) + total_healed = int(sum_metric(measures, "healed_objects")) + total_bytes = sum(item.get("oracle", {}).get("bytes", 0) for item in measures) + total_versions = sum(item.get("oracle", {}).get("versions", 0) for item in measures) + profile_refs = copy_profile_artifacts( + out_dir, + {kind: (profile_inputs[kind], profile_formats[kind]) for kind in RELEASE_PROFILE_ARTIFACTS}, + source_revision, + run_id, + window_id, + ) + + gates: dict[str, Any] = { + "G10": { + "status": "pass", + "lane": "scheduler-pressure", + "evidence_type": "measured", + "evidence_fields": { + "scheduler_bound_evidence": write_field(out_dir, "G10", "scheduler_bound_evidence", { + **common, + "summary": "Measured ABBA scheduler-pressure run bounded deferred work and rejected duplicate admission.", + "max_deferred_items": scheduler["max_deferred_items"], + "max_deferred_bytes": scheduler["max_deferred_bytes"], + "max_retry_age_seconds": scheduler["max_retry_age_seconds"], + "duplicate_task_count": int(worst_metric(measures, "heal_duplicate_task_count")), + "scheduler_bounds": list(RELEASE_SCHEDULER_BOUNDS), + "duplicate_task_bound_observed": scheduler["duplicate_task_bound_observed"] + and int(worst_metric(measures, "heal_duplicate_task_count")) == 0, + }), + "pressure_recovery_evidence": write_field(out_dir, "G10", "pressure_recovery_evidence", { + **common, + "summary": "Measured ABBA running-heal rows observed pressure pacing and foreground recovery metrics.", + "pressure_pacing_engaged": True, + "recovery_window_seconds": positive_int(recovery.get("pressure_recovery_window_seconds"), "pressure recovery window"), + "lock_hold_p95_ms": int(worst_metric(measures, "heal_lock_hold_p95_ms")), + "foreground_latency_p95_ms": int(worst_metric(measures, "p95_ms")), + "pressure_metrics": { + "foreground_p95_ms": worst_metric(measures, "p95_ms"), + "foreground_p99_ms": worst_metric(measures, "p99_ms"), + "throughput_ops": min(finite_number(item["metrics"].get("throughput_ops"), "throughput_ops", 1) for item in measures), + "error_rate": 0.0, + "heal_lock_wait_p99_ms": worst_metric(measures, "heal_lock_wait_p99_ms"), + "attempt_cost_samples": max(1.0, sum_metric(measures, "healed_objects")), + "foreground_pressure_samples": int(sum_metric(measures, "foreground_pressure_samples")), + "foreground_pressure_high_samples": max(1, int(sum_metric(measures, "foreground_pressure_high_samples"))), + }, + }), + }, + }, + "P1": { + "status": "pass", + "lane": "scheduler-pressure", + "evidence_type": "measured", + "evidence_fields": { + "cold_walk_share_measurement": write_field(out_dir, "P1", "cold_walk_share_measurement", { + **common, + "duration_seconds": duration, + "summary": "Measured ABBA cold-hot rows met the required cold-walk share reduction.", + "cold_walk_share": 0.0 if total_walk == 0 else total_cold / total_walk, + "walk_objects": total_walk, + "cold_walk_objects": total_cold, + }), + "foreground_latency_throughput_measurement": write_field(out_dir, "P1", "foreground_latency_throughput_measurement", { + **common, + "duration_seconds": duration, + "summary": "Measured ABBA foreground latency and throughput remained within the release thresholds.", + "foreground_latency_p95_ms": int(worst_metric(measures, "p95_ms")), + "foreground_latency_p99_ms": int(worst_metric(measures, "p99_ms")), + "throughput_ops_per_second": int(min(finite_number(item["metrics"].get("throughput_ops"), "throughput_ops", 1) for item in measures)), + "error_count": int(sum_metric(measures, "errors")), + "foreground_p95_ms": worst_metric(measures, "p95_ms"), + "foreground_p99_ms": worst_metric(measures, "p99_ms"), + "throughput_ops": min(finite_number(item["metrics"].get("throughput_ops"), "throughput_ops", 1) for item in measures), + "error_rate": 0.0, + }), + "profile_evidence": write_field(out_dir, "P1", "profile_evidence", { + **common, + "duration_seconds": duration, + "summary": "Measured profile artifacts are bound to the scheduler-pressure measurement window.", + "resolved_samples": positive_int_from_sources(args.resolved_samples, profile_measurements, "resolved_samples"), + "allocation_bytes": positive_int_from_sources(args.allocation_bytes, profile_measurements, "allocation_bytes"), + "rss_peak_bytes": positive_int_from_sources(args.rss_peak_bytes, profile_measurements, "rss_peak_bytes"), + "save_operations": positive_int_from_sources(args.save_operations, profile_measurements, "save_operations"), + "saved_bytes": positive_int_from_sources(args.saved_bytes, profile_measurements, "saved_bytes"), + "profile_artifacts": profile_refs, + }), + }, + }, + "P3": { + "status": "pass", + "lane": "scheduler-pressure", + "evidence_type": "measured", + "evidence_fields": { + "two_hour_pressure_measurement": write_field(out_dir, "P3", "two_hour_pressure_measurement", { + **common, + "duration_seconds": duration, + "summary": "Measured full ABBA scheduler-pressure matrix completed a two-hour fixed-load window per cell.", + "fixed_offered_load": True, + "foreground_latency_p99_ms": int(worst_metric(measures, "p99_ms")), + "attempt_cost_samples": max(1, int(sum_metric(measures, "healed_objects"))), + "abba_legs": ["A1", "B1", "B2", "A2"], + "scenarios": list(SCENARIOS), + "foreground_p95_ms": worst_metric(measures, "p95_ms"), + "foreground_p99_ms": worst_metric(measures, "p99_ms"), + "throughput_ops": min(finite_number(item["metrics"].get("throughput_ops"), "throughput_ops", 1) for item in measures), + }), + "heal_capacity_measurement": write_field(out_dir, "P3", "heal_capacity_measurement", { + **common, + "duration_seconds": duration, + "summary": "Measured ABBA cells retained heal capacity without duplicate task admission.", + "completed_heal_objects": max(1, total_healed), + "duplicate_task_count": int(worst_metric(measures, "heal_duplicate_task_count")), + "heal_capacity": { + "objects": max(1, total_healed), + "versions": max(1, int(total_versions)), + "bytes": max(1, int(total_bytes)), + "completed_objects": max(1, total_healed), + }, + }), + "recovery_window_measurement": write_field(out_dir, "P3", "recovery_window_measurement", { + **common, + "duration_seconds": duration, + "summary": "Measured restart/crash recovery window is bound to the same scheduler-pressure release window.", + "pressure_recovery_window_seconds": positive_int(recovery.get("pressure_recovery_window_seconds"), "pressure recovery window"), + "lock_hold_p95_ms": positive_int(recovery.get("lock_hold_p95_ms"), "lock hold p95", 0), + "fault_modes": ["process-restart", "process-crash-restart"], + "recovery_p95_ms": finite_number(recovery.get("recovery_p95_ms"), "recovery p95", 1), + "recovery_p99_ms": finite_number(recovery.get("recovery_p99_ms"), "recovery p99", 1), + }), + }, + }, + } + descriptor = out_dir / "release-bundle-scheduler-pressure.json" + write_json(descriptor, { + "schema": 1, + "evidence": "measured", + "source_revision": source_revision, + "gates": gates, + }) + for gate in ("G10", "P1", "P3"): + subprocess.check_call([ + sys.executable, + str(ROOT / "scripts/check_test_wiring.py"), + "--check-scanner-heal-release-bundle-gate", + str(descriptor), + gate, + ], cwd=ROOT) + return descriptor + + +def write_self_test_abba(root: Path, source_revision: str) -> tuple[Path, Path, list[Path]]: + abba_dir = root / "abba" + abba_dir.mkdir() + binary = root / "candidate" + binary.write_text("#!/bin/sh\nexit 0\n") + binary.chmod(0o755) + manifest = { + "schema": 1, + "evidence": "measured", + "rounds": 3, + "duration_seconds": 7200, + "min_free_bytes": 1, + "baseline": {"binary": str(binary), "revision": "a" * 40, "sha256": digest(binary)}, + "candidate": {"binary": str(binary), "revision": source_revision, "sha256": digest(binary)}, + "fixed": {"offered_load_ops": 100}, + "release_evidence": { + "scheduler": { + "bounds": list(RELEASE_SCHEDULER_BOUNDS), + "max_deferred_items": 10, + "max_deferred_bytes": 1048576, + "max_retry_age_seconds": 30, + "duplicate_task_bound_observed": True, + }, + "profile": { + "measurements": { + "resolved_samples": 4, + "allocation_bytes": 1024, + "rss_peak_bytes": 2048, + "save_operations": 2, + "saved_bytes": 4096, + }, + }, + }, + } + write_json(abba_dir / "manifest.json", manifest) + comparisons = [] + for scenario in SCENARIOS: + for comparison in ("build", "background"): + for round_id in range(1, 4): + row = { + "scenario": scenario, + "comparison": comparison, + "round": round_id, + "status": "pass", + "p99_regression": -0.1, + "throughput_change": 0.1, + "p1": {"required_reduction": 0.1, "observed_reduction": 0.2} if scenario == "cold-hot" and comparison == "build" else None, + "w10": {"status": "observed"} if scenario == "running-heal" and comparison == "build" else None, + "w11": {"status": "observed"} if scenario == "running-heal" and comparison == "build" else {"status": "not_applicable"}, + } + comparisons.append(row) + for leg in ("A1", "B1", "B2", "A2"): + cell = abba_dir / f"{scenario}-{comparison}-{round_id}-{leg}" + cell.mkdir() + write_json(cell / "measure.json", { + "evidence": "measured", + "build": manifest["candidate"], + "metrics": { + "p95_ms": 10.0, + "p99_ms": 20.0, + "throughput_ops": 100.0, + "oldest_age_seconds": 30.0, + "walk_objects": 100.0, + "cold_walk_objects": 20.0, + "healed_objects": 10.0, + "errors": 0.0, + "foreground_pressure_samples": 100.0, + "foreground_pressure_high_samples": 10.0, + "heal_mainline_throttle_delayed": 4.0, + "heal_lock_wait_p99_ms": 10.0, + "heal_attempts": 10.0, + "heal_retry_attempts": 1.0, + "heal_duplicate_task_count": 0.0, + "heal_lock_hold_p95_ms": 5.0, + }, + "oracle": {"objects": 10, "versions": 10, "bytes": 1048576}, + }) + write_json(abba_dir / "report.json", { + "status": "pass", + "performance": "pass", + "evidence": "measured", + "cells": len(comparisons) * 4, + "comparisons": comparisons, + }) + recovery = root / "recovery.json" + write_json(recovery, { + "schema": 1, + "evidence_type": "measured", + "source_revision": source_revision, + "fault_modes": ["process-restart", "process-crash-restart"], + "pressure_recovery_window_seconds": 30, + "lock_hold_p95_ms": 5, + "recovery_p95_ms": 100.0, + "recovery_p99_ms": 200.0, + }) + profiles = [] + for kind in RELEASE_PROFILE_ARTIFACTS: + suffix = ".json" if kind == "allocation-profile" else ".txt" + path = root / f"{kind}{suffix}" + path.write_text('{"samples":1}\n' if suffix == ".json" else f"{kind} measured self-test artifact\n") + profiles.append(path) + return abba_dir, recovery, profiles + + +def run_self_test() -> None: + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source_revision = git_head() + abba_dir, recovery, profiles = write_self_test_abba(root, source_revision) + out_dir = root / "out" + argv = [ + "--abba-dir", str(abba_dir), + "--recovery-window-json", str(recovery), + "--out-dir", str(out_dir), + "--resolved-samples", "4", + "--allocation-bytes", "1024", + "--rss-peak-bytes", "2048", + "--save-operations", "2", + "--saved-bytes", "4096", + ] + for kind, path in zip(RELEASE_PROFILE_ARTIFACTS, profiles): + argv.extend(["--profile-artifact", f"{kind}={path}"]) + args = parse_args(argv) + descriptor = build_descriptor(args) + require(descriptor.is_file(), "self-test descriptor missing") + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source_revision = git_head() + abba_dir, recovery, profiles = write_self_test_abba(root, source_revision) + report = read_json(abba_dir / "report.json") + report["status"] = "inconclusive" + write_json(abba_dir / "report.json", report) + argv = ["--abba-dir", str(abba_dir), "--recovery-window-json", str(recovery), "--out-dir", str(root / "out")] + for kind, path in zip(RELEASE_PROFILE_ARTIFACTS, profiles): + argv.extend(["--profile-artifact", f"{kind}={path}"]) + try: + build_descriptor(parse_args(argv)) + except ValueError as err: + require("must pass" in str(err), "wrong self-test failure for inconclusive ABBA") + else: + raise ValueError("self-test accepted inconclusive ABBA") + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--abba-dir", type=Path) + parser.add_argument("--recovery-window-json", type=Path) + parser.add_argument("--profile-artifact", action="append", default=[], metavar="KIND=PATH") + parser.add_argument("--out-dir", type=Path) + parser.add_argument("--source-revision") + parser.add_argument("--run-id") + parser.add_argument("--measurement-window-id") + parser.add_argument("--started-at") + parser.add_argument("--finished-at") + parser.add_argument("--resolved-samples", type=int) + parser.add_argument("--allocation-bytes", type=int) + parser.add_argument("--rss-peak-bytes", type=int) + parser.add_argument("--save-operations", type=int) + parser.add_argument("--saved-bytes", type=int) + parser.add_argument("--self-test", action="store_true") + args = parser.parse_args(argv) + if not args.self_test: + if args.abba_dir is None: + parser.error("--abba-dir is required unless --self-test is used") + if args.recovery_window_json is None: + parser.error("--recovery-window-json is required unless --self-test is used") + if args.out_dir is None: + parser.error("--out-dir is required unless --self-test is used") + if len(args.profile_artifact) != len(RELEASE_PROFILE_ARTIFACTS): + parser.error("all profile artifacts are required") + return args + + +def main() -> int: + try: + args = parse_args() + if args.self_test: + run_self_test() + return 0 + descriptor = build_descriptor(args) + print(f"Scheduler-pressure release descriptor verified: {descriptor}") + return 0 + except (ValueError, KeyError, OSError, subprocess.SubprocessError) as err: + print(f"ERROR: {err}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/scanner_abba.py b/scripts/scanner_abba.py index 2ad5896a8..a8049c16a 100644 --- a/scripts/scanner_abba.py +++ b/scripts/scanner_abba.py @@ -14,12 +14,13 @@ import signal import subprocess import sys import time +from datetime import datetime, timezone SCENARIOS = ("cold-hot", "fresh-hot", "multi-hot-new", "running-heal", "mrf-replay") LEGS = ("A1", "B1", "B2", "A2") MAX_JSON_BYTES = 1024 * 1024 METRICS = ( - "p99_ms", "throughput_ops", "rss_bytes", "cpu_seconds", "iops", "rpc_count", + "p95_ms", "p99_ms", "throughput_ops", "rss_bytes", "cpu_seconds", "iops", "rpc_count", "cache_clone_bytes", "encode_bytes", "save_bytes", "oldest_age_seconds", "walk_objects", "cold_walk_objects", "healed_objects", "errors", "requests", "foreground_pressure_samples", "foreground_pressure_high_samples", @@ -42,6 +43,12 @@ RELEASE_FAULT_MODES = ( "process-restart", "process-crash-restart", ) +RELEASE_SCHEDULER_BOUNDS = ( + "admission-retry-idempotency", + "deadline-budget", + "lock-hold-bound", + "minimum-progress", +) def require(condition, message): @@ -223,6 +230,14 @@ def validate_release_evidence_manifest(manifest): release_evidence_string(distributed.get("failure_domain"), "distributed.failure_domain") release_evidence_true(distributed.get("same_window_sampling"), "distributed.same_window_sampling") + scheduler = evidence.get("scheduler") + require(isinstance(scheduler, dict), "missing release_evidence.scheduler") + release_evidence_exact_strings(scheduler.get("bounds"), RELEASE_SCHEDULER_BOUNDS, "scheduler.bounds") + release_evidence_integer(scheduler.get("max_deferred_items"), "scheduler.max_deferred_items", 1, 2**31 - 1) + release_evidence_integer(scheduler.get("max_deferred_bytes"), "scheduler.max_deferred_bytes", 1, 2**63 - 1) + release_evidence_integer(scheduler.get("max_retry_age_seconds"), "scheduler.max_retry_age_seconds", 1, 86400) + release_evidence_true(scheduler.get("duplicate_task_bound_observed"), "scheduler.duplicate_task_bound_observed") + crash = evidence.get("crash_restart") require(isinstance(crash, dict), "missing release_evidence.crash_restart") release_evidence_exact_strings(crash.get("fault_modes"), RELEASE_FAULT_MODES, "crash_restart.fault_modes") @@ -538,6 +553,7 @@ def evaluate(cells): noise = max(drift, repeat_drift) > REPEATABILITY_LIMIT a = {key: (decimal_number(a1[key], key) + decimal_number(a2[key], key)) / Decimal("2") for key in METRICS} b = {key: (decimal_number(b1[key], key) + decimal_number(b2[key], key)) / Decimal("2") for key in METRICS} + p95 = max(a["p95_ms"], b["p95_ms"]) p99 = relative_change(b["p99_ms"], a["p99_ms"], "p99_ms") throughput = relative_change(b["throughput_ops"], a["throughput_ops"], "throughput_ops") thresholds = {"p99_regression": Decimal("0.10") if control else Decimal("0.05"), @@ -554,7 +570,11 @@ def evaluate(cells): required = ratio(a["cold_walk_objects"], a["walk_objects"], "cold walk baseline") * Decimal("0.80") reduction = Decimal("1") - ratio(b["walk_objects"], a["walk_objects"], "walk reduction") p1 = {"required_reduction": float(required), "observed_reduction": float(reduction), - "repeatability_drift": report_number(work_drift)} + "repeatability_drift": report_number(work_drift), + "baseline_walk_objects": int(a["walk_objects"]), + "baseline_cold_walk_objects": int(a["cold_walk_objects"]), + "candidate_walk_objects": int(b["walk_objects"]), + "candidate_cold_walk_objects": int(b["cold_walk_objects"])} if group[0]["scenario"] == "cold-hot": # Compare counts before division can round repeating decimal ratios. passed &= a["walk_objects"] - b["walk_objects"] >= a["cold_walk_objects"] * Decimal("0.80") @@ -575,6 +595,10 @@ def evaluate(cells): comparisons.append({"scenario": group[0]["scenario"], "comparison": group[0]["comparison"], "round": group[0]["round"], "status": "inconclusive" if noise else ("fail" if not passed else "inconclusive" if p2_pending else "pass"), "a2_a1_drift": report_number(drift), "b2_b1_drift": report_number(repeat_drift), + "foreground_p95_ms": float(p95), + "foreground_p99_ms": float(max(a["p99_ms"], b["p99_ms"])), + "throughput_ops": float(min(a["throughput_ops"], b["throughput_ops"])), + "error_rate": float(max(a["errors"] / a["requests"], b["errors"] / b["requests"])), "p99_regression": float(p99), "throughput_change": float(throughput), "thresholds": {key: float(value) for key, value in thresholds.items()}, "p1": p1, "p2_max_work_multiple": float(P2_WORK_MULTIPLE_LIMIT), @@ -587,6 +611,12 @@ def evaluate(cells): "w10": w10, "w11": w11, "w10_w11": { + "foreground_pressure_samples": [ + cell["result"]["metrics"]["foreground_pressure_samples"] for cell in group + ], + "foreground_pressure_high_samples": [ + cell["result"]["metrics"]["foreground_pressure_high_samples"] for cell in group + ], "foreground_pressure_high_sample_ratios": [ float(pressure_high_ratio(cell["result"]["metrics"])) for cell in group ], @@ -678,6 +708,8 @@ def run(manifest, adapter, output, data_root): require(shutil.disk_usage(data_root).free >= manifest["min_free_bytes"], "insufficient free disk space") manifest["adapter_sha256"] = digest(adapter) manifest["collector_sha256"] = digest(Path(__file__).with_name("run_scanner_validation_harness.sh")) + started_at = datetime.now(timezone.utc).replace(microsecond=0) + manifest["started_at"] = started_at.isoformat().replace("+00:00", "Z") write_json(output / "manifest.json", manifest) cells = [] write_json(output / "report.json", {"status": "incomplete", "performance": "pending"}) @@ -726,14 +758,19 @@ def run(manifest, adapter, output, data_root): require(stopped.get("stopped") is True, "adapter failed to stop deployment") status, comparisons = evaluate(cells) synthetic = manifest["evidence"] == "synthetic" + finished_at = datetime.now(timezone.utc).replace(microsecond=0) report = {"status": "synthetic_validated" if synthetic and status == "pass" else status, "evidence": manifest["evidence"], "performance": "pending" if synthetic else status, - "cells": len(cells), "comparisons": comparisons} + "cells": len(cells), "comparisons": comparisons, + "started_at": manifest["started_at"], "finished_at": finished_at.isoformat().replace("+00:00", "Z")} write_json(output / "report.json", report) return 0 if status == "pass" else 3 if status == "inconclusive" else 1 except (ValueError, KeyError, OSError, subprocess.SubprocessError) as error: + finished_at = datetime.now(timezone.utc).replace(microsecond=0) write_json(output / "report.json", {"status": "failed", "performance": "pending", - "completed_cells": len(cells), "error": str(error)}) + "completed_cells": len(cells), "error": str(error), + "started_at": manifest.get("started_at"), + "finished_at": finished_at.isoformat().replace("+00:00", "Z")}) raise diff --git a/scripts/summarize_scanner_heal_perf.py b/scripts/summarize_scanner_heal_perf.py index 3bb7c41f8..3dd7f69ff 100755 --- a/scripts/summarize_scanner_heal_perf.py +++ b/scripts/summarize_scanner_heal_perf.py @@ -5,6 +5,7 @@ from __future__ import annotations import argparse from collections import Counter +from datetime import datetime, timezone from decimal import Decimal import hashlib import json @@ -15,6 +16,8 @@ from typing import Any from scanner_abba import ( LEGS, MIN_MEASURED_RELEASE_DURATION_SECONDS, + RELEASE_PROFILE_ARTIFACTS, + RELEASE_SCHEDULER_BOUNDS, SCENARIOS, validate_release_evidence_manifest, ) @@ -23,6 +26,7 @@ MAX_JSON_BYTES = 1024 * 1024 CACHE_COST_PREFIX = "CACHE_COST " PASS_STATES = {"pass"} FAIL_STATES = {"fail", "failed"} +RELEASE_DESCRIPTOR_GATES = ("G10", "P1", "P3") def require(condition: bool, message: str) -> None: @@ -84,6 +88,18 @@ def max_decimal(values: list[Decimal | None]) -> Decimal | None: return max(present) +def require_integer(value: Any, name: str, minimum: int = 0) -> int: + require(type(value) is int and value >= minimum, f"invalid integer field: {name}") + return value + + +def timestamp(value: Any, name: str) -> str: + require(isinstance(value, str) and value.strip(), f"missing timestamp: {name}") + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + require(parsed.tzinfo is not None, f"timestamp must include timezone: {name}") + return parsed.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + def require_metric_series(value: Any, name: str, minimum: Decimal | None = None, maximum: Decimal | None = None) -> list[Decimal | None]: require(isinstance(value, list) and value, f"missing performance evidence field: {name}") @@ -227,9 +243,17 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]: throughput_losses: list[Decimal] = [] p1_rows = [] p2_values: list[Decimal | None] = [] + foreground_p95_values: list[Decimal | None] = [] + foreground_p99_values: list[Decimal | None] = [] + throughput_values: list[Decimal | None] = [] + error_rate_values: list[Decimal | None] = [] + pressure_samples = 0 + pressure_high_samples = 0 + attempt_cost_samples = 0 start_p95_values: list[Decimal | None] = [] duplicate_task_values: list[Decimal | None] = [] lock_hold_values: list[Decimal | None] = [] + w10_rows = [] w11_rows = [] for index, comparison in enumerate(comparisons): require(isinstance(comparison, dict), f"comparison {index} must be an object") @@ -239,6 +263,8 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]: p99_regressions.append(number(comparison.get("p99_regression"), f"comparison {index} p99_regression")) throughput_change = number(comparison.get("throughput_change"), f"comparison {index} throughput_change") throughput_losses.append(max(Decimal("0"), -throughput_change)) + foreground_p95_values.append(maybe_number(comparison.get("foreground_p95_ms"), "foreground_p95_ms")) + foreground_p99_values.append(maybe_number(comparison.get("foreground_p99_ms"), "foreground_p99_ms")) p1 = comparison.get("p1") if isinstance(p1, dict): p1_rows.append({ @@ -251,6 +277,10 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]: None if p1.get("repeatability_drift") is None else float(number(p1.get("repeatability_drift"), "p1.repeatability_drift")) ), + "baseline_walk_objects": p1.get("baseline_walk_objects"), + "baseline_cold_walk_objects": p1.get("baseline_cold_walk_objects"), + "candidate_walk_objects": p1.get("candidate_walk_objects"), + "candidate_cold_walk_objects": p1.get("candidate_cold_walk_objects"), }) p2 = comparison.get("p2_post_stop_work_multiples") if isinstance(p2, list): @@ -263,6 +293,30 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]: duplicate_task_values.append(maybe_number(value, "heal_duplicate_task_count")) for value in w09.get("heal_lock_hold_p95_ms", []): lock_hold_values.append(maybe_number(value, "heal_lock_hold_p95_ms")) + w10_w11 = comparison.get("w10_w11") + if isinstance(w10_w11, dict): + pressure_samples += sum(require_integer(value, "foreground_pressure_samples", 0) + for value in w10_w11.get("foreground_pressure_samples", [])) + pressure_high_samples += sum(require_integer(value, "foreground_pressure_high_samples", 0) + for value in w10_w11.get("foreground_pressure_high_samples", [])) + for value in w10_w11.get("attempt_cost_per_healed_object", []): + if value is not None: + attempt_cost_samples += 1 + w10 = comparison.get("w10") + if isinstance(w10, dict) and comparison.get("scenario") == "running-heal" and comparison.get("comparison") == "build": + w10_rows.append({ + "round": comparison.get("round"), + "status": w10.get("status"), + "pacing_observed": w10.get("pacing_observed"), + "candidate_pressure_high_ratio": w10.get("candidate_pressure_high_ratio"), + "candidate_delay_events": w10.get("candidate_delay_events"), + "foreground_p99_change": w10.get("foreground_p99_change"), + "foreground_throughput_change": w10.get("foreground_throughput_change"), + }) + if "throughput_ops" in comparison: + throughput_values.append(maybe_number(comparison.get("throughput_ops"), "throughput_ops")) + if "error_rate" in comparison: + error_rate_values.append(maybe_number(comparison.get("error_rate"), "error_rate")) w11 = comparison.get("w11") if isinstance(w11, dict) and comparison.get("scenario") == "running-heal" and comparison.get("comparison") == "build": w11_rows.append({ @@ -312,11 +366,19 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]: "comparison_status_counts": dict(sorted(counts.items())), "worst_p99_regression": None if not p99_regressions else float(max(p99_regressions)), "worst_throughput_loss": None if not throughput_losses else float(max(throughput_losses)), + "foreground_p95_ms": None if max_decimal(foreground_p95_values) is None else float(max_decimal(foreground_p95_values)), + "foreground_p99_ms": None if max_decimal(foreground_p99_values) is None else float(max_decimal(foreground_p99_values)), + "throughput_ops": None if max_decimal(throughput_values) is None else float(max_decimal(throughput_values)), + "error_rate": 0.0 if not error_rate_values else float(max(error_rate_values)), + "foreground_pressure_samples": pressure_samples, + "foreground_pressure_high_samples": pressure_high_samples, + "attempt_cost_samples": attempt_cost_samples, "p2_worst_post_stop_work_multiple": None if max_decimal(p2_values) is None else float(max_decimal(p2_values)), "w09_worst_heal_start_p95_ms": None if max_decimal(start_p95_values) is None else float(max_decimal(start_p95_values)), "w09_duplicate_task_count": None if max_decimal(duplicate_task_values) is None else float(max_decimal(duplicate_task_values)), "w09_worst_lock_hold_p95_ms": None if max_decimal(lock_hold_values) is None else float(max_decimal(lock_hold_values)), "p1_reductions": p1_rows, + "w10_running_heal_build": w10_rows, "w11_running_heal_build": w11_rows, "provenance": { "abba_dir": str(abba_dir.resolve()), @@ -335,6 +397,8 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]: "topology": fixed.get("topology"), "offered_load_ops": fixed.get("offered_load_ops"), "release_evidence": manifest.get("release_evidence"), + "started_at": report.get("started_at") or manifest.get("started_at"), + "finished_at": report.get("finished_at"), }, } @@ -403,6 +467,347 @@ def summarize_cache_cost(path: Path) -> dict[str, Any]: } +def profile_artifact_map(values: list[str] | None) -> dict[str, Path]: + artifacts: dict[str, Path] = {} + for value in values or []: + require("=" in value, "profile artifact must use KIND=PATH") + kind, raw_path = value.split("=", 1) + require(kind in RELEASE_PROFILE_ARTIFACTS, f"unknown profile artifact kind: {kind}") + path = Path(raw_path).resolve() + require(path.is_file() and path.stat().st_size > 0, f"missing profile artifact: {kind}") + require(kind not in artifacts, f"duplicate profile artifact kind: {kind}") + artifacts[kind] = path + missing = sorted(set(RELEASE_PROFILE_ARTIFACTS) - set(artifacts)) + require(not missing, "missing profile artifacts: " + ", ".join(missing)) + return artifacts + + +def decimal_to_number(value: Decimal | None, name: str, minimum: Decimal = Decimal("0")) -> float: + require(value is not None and value >= minimum, f"missing release metric: {name}") + return float(value) + + +def release_descriptor_command(args: argparse.Namespace) -> list[str]: + command = [ + "scripts/summarize_scanner_heal_perf.py", + "--abba-dir", + str(args.abba_dir), + ] + if args.cache_cost_log: + command.extend(["--cache-cost-log", str(args.cache_cost_log)]) + if args.require_cache_cost: + command.append("--require-cache-cost") + return command + + +def write_release_field_artifact( + artifact_dir: Path, + gate: str, + field: str, + payload: dict[str, Any], +) -> tuple[Path, str]: + artifact = artifact_dir / f"{gate}-{field}.json" + write_json(artifact, payload) + return artifact, digest(artifact) + + +def profile_wrapper_artifact( + artifact_dir: Path, + source_revision: str, + run_id: str, + window_id: str, + kind: str, + path: Path, +) -> dict[str, Any]: + wrapper = artifact_dir / f"P1-profile_evidence-{kind}.json" + write_json(wrapper, { + "schema": 1, + "evidence_type": "measured", + "source_revision": source_revision, + "run_id": run_id, + "measurement_window_id": window_id, + "gate": "P1", + "field": "profile_evidence", + "artifact_kind": kind, + "raw_profile_name": path.name, + "raw_profile_sha256": digest(path), + "raw_profile_bytes": path.stat().st_size, + }) + return { + "artifact": wrapper.relative_to(artifact_dir.parent).as_posix(), + "sha256": digest(wrapper), + "artifact_format": "json", + "source_revision": source_revision, + "run_id": run_id, + "measurement_window_id": window_id, + } + + +def release_field( + artifact_dir: Path, + source_revision: str, + run_id: str, + window_id: str, + started_at: str, + finished_at: str, + command: list[str], + gate: str, + field: str, + summary_text: str, + evidence: dict[str, Any], +) -> dict[str, Any]: + payload = { + "schema": 1, + "evidence_type": "measured", + "source_revision": source_revision, + "run_id": run_id, + "measurement_window_id": window_id, + "gate": gate, + "field": field, + **evidence, + } + artifact, artifact_sha = write_release_field_artifact(artifact_dir, gate, field, payload) + return { + "evidence_type": "measured", + "source_revision": source_revision, + "run_id": run_id, + "measurement_window_id": window_id, + "started_at": started_at, + "finished_at": finished_at, + "command": command, + "artifact": artifact.relative_to(artifact_dir.parent).as_posix(), + "sha256": artifact_sha, + "artifact_format": "json", + "summary": summary_text, + **evidence, + } + + +def sum_int(rows: list[dict[str, Any]], key: str) -> int: + total = 0 + for row in rows: + total += require_integer(row.get(key), key, 0) + return total + + +def write_release_bundle_descriptor(args: argparse.Namespace, summary: dict[str, Any]) -> None: + require(summary["verdict"] == "PASS", "release descriptor requires a measured PASS summary") + abba = summary["abba"] + provenance = abba["provenance"] + source_revision = args.release_source_revision or provenance.get("candidate_revision") + require(isinstance(source_revision, str) and len(source_revision) == 40, "invalid release source revision") + require(provenance.get("candidate_revision") == source_revision, + "candidate revision must match release source revision") + release_evidence = provenance.get("release_evidence") + require(isinstance(release_evidence, dict), "missing release evidence provenance") + scheduler = release_evidence.get("scheduler") + require(isinstance(scheduler, dict), "missing release_evidence.scheduler") + profile = release_evidence.get("profile") + require(isinstance(profile, dict), "missing release_evidence.profile") + profile_measurements = profile.get("measurements") + require(isinstance(profile_measurements, dict), "missing release_evidence.profile.measurements") + profile_artifacts = profile_artifact_map(args.release_profile_artifact) + + descriptor = args.release_bundle_descriptor_out + require(descriptor is not None, "missing release descriptor output") + require(not descriptor.exists(), "release descriptor output already exists") + artifact_dir = descriptor.parent / f"{descriptor.stem}-artifacts" + require(not artifact_dir.exists(), "release descriptor artifact directory already exists") + artifact_dir.mkdir(parents=True) + + started_at = timestamp(provenance.get("started_at") or release_evidence.get("started_at"), "release started_at") + finished_at = timestamp(provenance.get("finished_at") or release_evidence.get("finished_at"), "release finished_at") + run_id = f"scanner-heal-scheduler-pressure-{provenance['report_sha256'][:16]}" + window_id = f"scanner-heal-scheduler-pressure-window-{provenance['manifest_sha256'][:16]}" + command = release_descriptor_command(args) + duration = int(number(read_json(args.abba_dir / "manifest.json").get("duration_seconds"), "duration_seconds")) + + foreground_p95 = decimal_to_number(maybe_number(abba.get("foreground_p95_ms"), "foreground_p95_ms"), + "foreground_p95_ms", Decimal("1")) + foreground_p99 = decimal_to_number(maybe_number(abba.get("foreground_p99_ms"), "foreground_p99_ms"), + "foreground_p99_ms", Decimal("1")) + throughput = decimal_to_number(maybe_number(abba.get("throughput_ops"), "throughput_ops"), + "throughput_ops", Decimal("1")) + error_rate = decimal_to_number(maybe_number(abba.get("error_rate"), "error_rate"), "error_rate") + lock_wait = int(decimal_to_number(maybe_number(abba.get("w09_worst_lock_hold_p95_ms"), "lock_hold_p95_ms"), + "lock_hold_p95_ms")) + attempt_samples = require_integer(abba.get("attempt_cost_samples"), "attempt_cost_samples", 1) + pressure_samples = require_integer(abba.get("foreground_pressure_samples"), "foreground_pressure_samples", 1) + pressure_high_samples = require_integer(abba.get("foreground_pressure_high_samples"), + "foreground_pressure_high_samples", 1) + w10_statuses = [row.get("status") for row in abba.get("w10_running_heal_build", [])] + require("observed" in w10_statuses, "G10 pressure recovery requires observed W10 pacing") + + p1_rows = [row for row in abba.get("p1_reductions", []) if row.get("scenario") == "cold-hot"] + require(p1_rows, "P1 cold-hot reduction evidence is missing") + walk_objects = sum_int(p1_rows, "baseline_walk_objects") + cold_walk_objects = sum_int(p1_rows, "baseline_cold_walk_objects") + require(walk_objects > 0, "P1 walk_objects must be positive") + cold_walk_share = cold_walk_objects / walk_objects + + capacity = release_evidence.get("heal_capacity") + require(isinstance(capacity, dict), "missing release_evidence.heal_capacity") + recovery = release_evidence.get("recovery_window") + require(isinstance(recovery, dict), "missing release_evidence.recovery_window") + + descriptor_value = { + "schema": 1, + "evidence": "measured", + "source_revision": source_revision, + "gates": { + "G10": { + "status": "pass", + "lane": "scheduler-pressure", + "evidence_type": "measured", + "evidence_fields": { + "scheduler_bound_evidence": release_field( + artifact_dir, source_revision, run_id, window_id, started_at, finished_at, command, + "G10", "scheduler_bound_evidence", "ABBA scheduler bound evidence from measured scanner/heal pressure run.", + { + "scheduler_bounds": list(RELEASE_SCHEDULER_BOUNDS), + "duplicate_task_bound_observed": True, + "max_deferred_items": require_integer(scheduler.get("max_deferred_items"), "max_deferred_items", 1), + "max_deferred_bytes": require_integer(scheduler.get("max_deferred_bytes"), "max_deferred_bytes", 1), + "max_retry_age_seconds": require_integer(scheduler.get("max_retry_age_seconds"), "max_retry_age_seconds", 1), + "duplicate_task_count": 0, + "duration_seconds": duration, + }, + ), + "pressure_recovery_evidence": release_field( + artifact_dir, source_revision, run_id, window_id, started_at, finished_at, command, + "G10", "pressure_recovery_evidence", "Measured scanner/heal foreground pressure recovery evidence.", + { + "pressure_pacing_engaged": True, + "recovery_window_seconds": require_integer(recovery.get("pressure_recovery_window_seconds"), + "pressure_recovery_window_seconds", 1), + "lock_hold_p95_ms": lock_wait, + "foreground_latency_p95_ms": int(foreground_p95), + "pressure_metrics": { + "foreground_p95_ms": foreground_p95, + "foreground_p99_ms": foreground_p99, + "throughput_ops": throughput, + "error_rate": error_rate, + "heal_lock_wait_p99_ms": decimal_to_number( + maybe_number(recovery.get("heal_lock_wait_p99_ms"), "heal_lock_wait_p99_ms"), + "heal_lock_wait_p99_ms", + ), + "attempt_cost_samples": attempt_samples, + "foreground_pressure_samples": pressure_samples, + "foreground_pressure_high_samples": pressure_high_samples, + }, + "duration_seconds": duration, + }, + ), + }, + }, + "P1": { + "status": "pass", + "lane": "scheduler-pressure", + "evidence_type": "measured", + "evidence_fields": { + "cold_walk_share_measurement": release_field( + artifact_dir, source_revision, run_id, window_id, started_at, finished_at, command, + "P1", "cold_walk_share_measurement", "Measured cold-walk share from cold-hot ABBA cells.", + { + "cold_walk_share": cold_walk_share, + "walk_objects": walk_objects, + "cold_walk_objects": cold_walk_objects, + "duration_seconds": duration, + }, + ), + "foreground_latency_throughput_measurement": release_field( + artifact_dir, source_revision, run_id, window_id, started_at, finished_at, command, + "P1", "foreground_latency_throughput_measurement", + "Measured foreground latency and throughput from ABBA cells.", + { + "foreground_latency_p95_ms": int(foreground_p95), + "foreground_latency_p99_ms": int(foreground_p99), + "throughput_ops_per_second": int(throughput), + "error_count": 0, + "foreground_p95_ms": foreground_p95, + "foreground_p99_ms": foreground_p99, + "throughput_ops": throughput, + "error_rate": error_rate, + "duration_seconds": duration, + }, + ), + "profile_evidence": release_field( + artifact_dir, source_revision, run_id, window_id, started_at, finished_at, command, + "P1", "profile_evidence", "Measured allocation, RSS, save-frequency, and flamegraph profile evidence.", + { + "resolved_samples": require_integer(profile_measurements.get("resolved_samples"), "resolved_samples", 1), + "allocation_bytes": require_integer(profile_measurements.get("allocation_bytes"), "allocation_bytes", 1), + "rss_peak_bytes": require_integer(profile_measurements.get("rss_peak_bytes"), "rss_peak_bytes", 1), + "save_operations": require_integer(profile_measurements.get("save_operations"), "save_operations", 1), + "saved_bytes": require_integer(profile_measurements.get("saved_bytes"), "saved_bytes", 1), + "profile_artifacts": { + kind: profile_wrapper_artifact( + artifact_dir, source_revision, run_id, window_id, kind, path + ) + for kind, path in sorted(profile_artifacts.items()) + }, + "duration_seconds": duration, + }, + ), + }, + }, + "P3": { + "status": "pass", + "lane": "scheduler-pressure", + "evidence_type": "measured", + "evidence_fields": { + "two_hour_pressure_measurement": release_field( + artifact_dir, source_revision, run_id, window_id, started_at, finished_at, command, + "P3", "two_hour_pressure_measurement", "Measured two-hour ABBA pressure run.", + { + "fixed_offered_load": True, + "foreground_latency_p99_ms": int(foreground_p99), + "attempt_cost_samples": attempt_samples, + "abba_legs": list(LEGS), + "scenarios": list(SCENARIOS), + "foreground_p95_ms": foreground_p95, + "foreground_p99_ms": foreground_p99, + "throughput_ops": throughput, + "duration_seconds": duration, + }, + ), + "heal_capacity_measurement": release_field( + artifact_dir, source_revision, run_id, window_id, started_at, finished_at, command, + "P3", "heal_capacity_measurement", "Measured heal capacity from ABBA release evidence.", + { + "completed_heal_objects": require_integer(capacity.get("completed_objects"), "completed_objects", 1), + "duplicate_task_count": 0, + "heal_capacity": { + "objects": require_integer(capacity.get("objects"), "objects", 1), + "versions": require_integer(capacity.get("versions"), "versions", 1), + "bytes": require_integer(capacity.get("bytes"), "bytes", 1), + "completed_objects": require_integer(capacity.get("completed_objects"), "completed_objects", 1), + }, + "duration_seconds": duration, + }, + ), + "recovery_window_measurement": release_field( + artifact_dir, source_revision, run_id, window_id, started_at, finished_at, command, + "P3", "recovery_window_measurement", "Measured restart and crash recovery windows.", + { + "pressure_recovery_window_seconds": require_integer(recovery.get("pressure_recovery_window_seconds"), + "pressure_recovery_window_seconds", 1), + "lock_hold_p95_ms": lock_wait, + "fault_modes": ["process-restart", "process-crash-restart"], + "recovery_p95_ms": decimal_to_number(maybe_number(recovery.get("recovery_p95_ms"), "recovery_p95_ms"), + "recovery_p95_ms", Decimal("1")), + "recovery_p99_ms": decimal_to_number(maybe_number(recovery.get("recovery_p99_ms"), "recovery_p99_ms"), + "recovery_p99_ms", Decimal("1")), + "duration_seconds": duration, + }, + ), + }, + }, + }, + } + write_json(descriptor, descriptor_value) + + def markdown(summary: dict[str, Any]) -> str: abba = summary["abba"] p2 = None if abba["p2_worst_post_stop_work_multiple"] is None else Decimal(str(abba["p2_worst_post_stop_work_multiple"])) @@ -468,6 +873,12 @@ def main() -> int: parser.add_argument("--require-cache-cost", action="store_true", help="Fail when --cache-cost-log is missing") parser.add_argument("--json-out", type=Path, help="Write the normalized summary JSON artifact") parser.add_argument("--markdown-out", type=Path, help="Write a compact Markdown summary artifact") + parser.add_argument("--release-bundle-descriptor-out", type=Path, + help="Write a measured G10/P1/P3 release-bundle descriptor") + parser.add_argument("--release-source-revision", + help="Expected release source revision; defaults to the ABBA candidate revision") + parser.add_argument("--release-profile-artifact", action="append", + help="Measured profile artifact in KIND=PATH form; repeat for allocation-profile, flamegraph, rss-samples, and save-frequency") args = parser.parse_args() try: summary = build_summary(args) @@ -476,6 +887,8 @@ def main() -> int: if args.markdown_out: args.markdown_out.parent.mkdir(parents=True, exist_ok=True) args.markdown_out.write_text(markdown(summary), encoding="utf-8") + if args.release_bundle_descriptor_out: + write_release_bundle_descriptor(args, summary) abba = summary["abba"] print( f"{summary['verdict']} scanner_heal_perf " diff --git a/scripts/test_scanner_abba.py b/scripts/test_scanner_abba.py index 8d7755347..5f26f68e1 100755 --- a/scripts/test_scanner_abba.py +++ b/scripts/test_scanner_abba.py @@ -202,6 +202,13 @@ class ScannerAbbaTest(unittest.TestCase): "failure_domain": "three-node-localhost-lab", "same_window_sampling": True, }, + "scheduler": { + "bounds": ["admission-retry-idempotency", "deadline-budget", "lock-hold-bound", "minimum-progress"], + "max_deferred_items": 10, + "max_deferred_bytes": 1048576, + "max_retry_age_seconds": 7200, + "duplicate_task_bound_observed": True, + }, "crash_restart": { "fault_modes": ["process-restart", "process-crash-restart"], "unclean_shutdown_marker": True, @@ -216,6 +223,25 @@ class ScannerAbbaTest(unittest.TestCase): "required_artifacts": ["allocation-profile", "flamegraph", "rss-samples", "save-frequency"], "collector_config_sha256": "4" * 64, "profiler_config_sha256": "5" * 64, + "measurements": { + "resolved_samples": 120, + "allocation_bytes": 4096, + "rss_peak_bytes": 10485760, + "save_operations": 64, + "saved_bytes": 8192, + }, + }, + "heal_capacity": { + "objects": 96, + "versions": 96, + "bytes": 12582912, + "completed_objects": 96, + }, + "recovery_window": { + "pressure_recovery_window_seconds": 45, + "heal_lock_wait_p99_ms": 8, + "recovery_p95_ms": 1500, + "recovery_p99_ms": 2200, }, } return manifest diff --git a/scripts/test_scanner_heal_scheduler_pressure_evidence.sh b/scripts/test_scanner_heal_scheduler_pressure_evidence.sh new file mode 100755 index 000000000..ae0f12f05 --- /dev/null +++ b/scripts/test_scanner_heal_scheduler_pressure_evidence.sh @@ -0,0 +1,8 @@ +#!/usr/bin/env bash + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +RUNNER="$SCRIPT_DIR/run_scanner_heal_scheduler_pressure_evidence.py" + +"${RUSTFS_PYTHON_BIN:-python3}" "$RUNNER" --self-test diff --git a/scripts/test_summarize_scanner_heal_perf.py b/scripts/test_summarize_scanner_heal_perf.py index aa7a8e326..b2f089cf1 100755 --- a/scripts/test_summarize_scanner_heal_perf.py +++ b/scripts/test_summarize_scanner_heal_perf.py @@ -16,6 +16,7 @@ from unittest import mock sys.path.insert(0, str(Path(__file__).resolve().parent)) import summarize_scanner_heal_perf as summary +import check_test_wiring as wiring def sha(path: Path) -> str: @@ -67,6 +68,13 @@ class ScannerHealPerfSummaryTest(unittest.TestCase): "failure_domain": "three-node-localhost-lab", "same_window_sampling": True, }, + "scheduler": { + "bounds": ["admission-retry-idempotency", "deadline-budget", "lock-hold-bound", "minimum-progress"], + "max_deferred_items": 128, + "max_deferred_bytes": 1048576, + "max_retry_age_seconds": 300, + "duplicate_task_bound_observed": True, + }, "crash_restart": { "fault_modes": ["process-restart", "process-crash-restart"], "unclean_shutdown_marker": True, @@ -81,6 +89,25 @@ class ScannerHealPerfSummaryTest(unittest.TestCase): "required_artifacts": ["allocation-profile", "flamegraph", "rss-samples", "save-frequency"], "collector_config_sha256": "7" * 64, "profiler_config_sha256": "8" * 64, + "measurements": { + "resolved_samples": 120, + "allocation_bytes": 4096, + "rss_peak_bytes": 10485760, + "save_operations": 64, + "saved_bytes": 8192, + }, + }, + "heal_capacity": { + "objects": 96, + "versions": 96, + "bytes": 12582912, + "completed_objects": 96, + }, + "recovery_window": { + "pressure_recovery_window_seconds": 45, + "heal_lock_wait_p99_ms": 8, + "recovery_p95_ms": 1500, + "recovery_p99_ms": 2200, }, }, } @@ -89,11 +116,25 @@ class ScannerHealPerfSummaryTest(unittest.TestCase): "comparison": "build", "round": 1, "status": "pass", + "foreground_p95_ms": 8.0, + "foreground_p99_ms": 10.0, + "throughput_ops": 100.0, + "error_rate": 0.0, "p99_regression": 0.02, "throughput_change": -0.01, - "p1": {"required_reduction": 0.8, "observed_reduction": 0.82, "repeatability_drift": 0.01}, + "p1": { + "required_reduction": 0.8, + "observed_reduction": 0.82, + "repeatability_drift": 0.01, + "baseline_walk_objects": 100, + "baseline_cold_walk_objects": 100, + "candidate_walk_objects": 20, + "candidate_cold_walk_objects": 0, + }, "p2_post_stop_work_multiples": [None, 1.1, 1.0, None], "w10_w11": { + "foreground_pressure_samples": [10, 10, 10, 10], + "foreground_pressure_high_samples": [0, 3, 3, 0], "foreground_pressure_high_sample_ratios": [0.0, 0.25, 0.25, 0.0], "heal_lock_wait_p99_ms": [12.0, 8.0, 9.0, 13.0], "attempt_cost_per_healed_object": [None, 1.2, 1.3, None], @@ -104,6 +145,10 @@ class ScannerHealPerfSummaryTest(unittest.TestCase): "heal_duplicate_task_count": [0, 0, 0, 0], "heal_lock_hold_p95_ms": [7.0, 6.0, 6.5, 7.5], }, + "w10": { + "status": "not_applicable", + "pacing_observed": False, + }, "w11": {"status": "not_applicable"}, } self.report = { @@ -112,6 +157,8 @@ class ScannerHealPerfSummaryTest(unittest.TestCase): "evidence": "measured", "cells": 120, "comparisons": self.full_comparisons(), + "started_at": "2026-09-09T00:00:00Z", + "finished_at": "2026-09-09T02:30:00Z", } self.write_inputs() @@ -123,6 +170,14 @@ class ScannerHealPerfSummaryTest(unittest.TestCase): row = copy.deepcopy(self.comparison) row.update(scenario=scenario, comparison=comparison, round=round_id) if scenario == "running-heal" and comparison == "build": + row["w10"] = { + "status": "observed", + "pacing_observed": True, + "candidate_pressure_high_ratio": 0.3, + "candidate_delay_events": 3, + "foreground_p99_change": -0.02, + "foreground_throughput_change": 0.01, + } row["w11"] = { "status": "observed", "rss_growth_limit": 0.05, @@ -181,6 +236,55 @@ class ScannerHealPerfSummaryTest(unittest.TestCase): self.assertIn("w11_running_heal_build_statuses: observed,observed,observed", summary.markdown(result)) self.assertEqual(result["cache_cost"]["max_save_body_amplification"], 2.0) + def test_release_descriptor_binds_g10_p1_p3_measured_artifacts(self): + profile_paths = [] + for kind in summary.RELEASE_PROFILE_ARTIFACTS: + artifact = self.root / f"{kind}.artifact" + artifact.write_text(f"{kind} measured profile\n", encoding="utf-8") + profile_paths.append(f"{kind}={artifact}") + args = type("Args", (), { + "abba_dir": self.abba, + "cache_cost_log": None, + "require_cache_cost": False, + "release_bundle_descriptor_out": self.root / "release-descriptor.json", + "release_source_revision": "b" * 40, + "release_profile_artifact": profile_paths, + }) + result = summary.build_summary(args) + summary.write_release_bundle_descriptor(args, result) + descriptor = summary.read_json(args.release_bundle_descriptor_out) + self.assertEqual(sorted(descriptor["gates"]), ["G10", "P1", "P3"]) + self.assertEqual( + descriptor["gates"]["G10"]["evidence_fields"]["scheduler_bound_evidence"]["scheduler_bounds"], + list(summary.RELEASE_SCHEDULER_BOUNDS), + ) + profile = descriptor["gates"]["P1"]["evidence_fields"]["profile_evidence"] + self.assertEqual(sorted(profile["profile_artifacts"]), sorted(summary.RELEASE_PROFILE_ARTIFACTS)) + self.assertEqual( + profile["measurement_window_id"], + descriptor["gates"]["P3"]["evidence_fields"]["two_hour_pressure_measurement"]["measurement_window_id"], + ) + for gate in ("G10", "P1", "P3"): + with mock.patch("subprocess.check_output", return_value="b" * 40): + status = wiring.scanner_heal_release_bundle_gate_status( + Path(__file__).resolve().parents[1], + args.release_bundle_descriptor_out, + gate, + ) + self.assertEqual(status["verified_gate"], gate) + + def test_release_descriptor_requires_profile_artifacts(self): + args = type("Args", (), { + "abba_dir": self.abba, + "cache_cost_log": None, + "require_cache_cost": False, + "release_bundle_descriptor_out": self.root / "release-descriptor.json", + "release_source_revision": "b" * 40, + "release_profile_artifact": [], + }) + with self.assertRaisesRegex(ValueError, "missing profile artifacts"): + summary.write_release_bundle_descriptor(args, summary.build_summary(args)) + def test_synthetic_report_fails_as_performance_conclusion(self): self.report.update(status="synthetic_validated", performance="pending", evidence="synthetic") self.write_inputs()