gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-09 13:46:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(scanner): add release evidence ingestion gates (#7588)

Browse Source 
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-09 19:56:43 +08:00
committed by GitHub
parent de3ac27a6a
commit 3ffc3704af
11 changed files with 2151 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.config/make/tests.mak
+3
View File
@@ -32,8 +32,11 @@ 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_checkpoint_crash_evidence.sh
./scripts/test_scanner_heal_g14_multiset_evidence.sh
./scripts/test_scanner_heal_scheduler_pressure_evidence.sh
./scripts/test_scanner_heal_status_outcome_evidence.sh
./scripts/test_scanner_heal_maintenance_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
.config/scanner-heal-required-tests.json
+18 -3
View File
@@ -209,7 +209,11 @@
"lane": "status-and-outcome",
"status": "pending",
"description": "Per-object outcomes and bounded terminal retention",
"requires": ["per-object outcome oracle", "terminal retention bounds"]
"requires": ["per-object outcome oracle", "terminal retention bounds"],
"evidence_fields": [
"per_object_outcome_oracle",
"terminal_retention_bounds"
]
},
{
"gate": "G06",
@@ -217,7 +221,12 @@
"lane": "status-and-outcome",
"status": "pending",
"description": "Concurrent status, legacy clients and truncation",
"requires": ["concurrent status evidence", "legacy client compatibility", "truncation behavior"]
"requires": ["concurrent status evidence", "legacy client compatibility", "truncation behavior"],
"evidence_fields": [
"concurrent_status_evidence",
"legacy_client_compatibility",
"truncation_behavior"
]
},
{
"gate": "G07",
@@ -375,7 +384,13 @@
"lane": "status-and-outcome",
"status": "pending",
"description": "Manager-to-event-to-ledger exact disposition, including grace",
"requires": ["manager disposition evidence", "event disposition evidence", "ledger disposition evidence", "grace handling"]
"requires": ["manager disposition evidence", "event disposition evidence", "ledger disposition evidence", "grace handling"],
"evidence_fields": [
"manager_disposition_evidence",
"event_disposition_evidence",
"ledger_disposition_evidence",
"grace_handling"
]
},
{
"gate": "R-L",
scripts/README.md
+6
View File
@@ -56,14 +56,20 @@ their issue closes.
| `probe.sh` | dev-tool | Probe-style e2e run | `make probe-e2e` |
| `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_checkpoint_crash_evidence.py` | dev-tool | Assembles measured Scanner/Heal G02/R-E checkpoint and restart release descriptors from scanner restart diagnostic reports | `diagnose_scanner_enumeration_restart.py`; `test_scanner_heal_checkpoint_crash_evidence.sh` |
| `run_scanner_heal_g14_multiset_evidence.py` | dev-tool | Assembles measured Scanner/Heal G14 same-window EC8+4 multi-set/multi-pool release descriptors from e2e case directories or an operator-collected proof | `.config/scanner-heal-required-tests.json`; `test_scanner_heal_g14_multiset_evidence.sh` |
| `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_status_outcome_evidence.py` | dev-tool | Assembles measured Scanner/Heal G05/G06/R-D status-and-outcome release descriptors from status, compatibility, and disposition artifacts | `.config/scanner-heal-required-tests.json`; `test_scanner_heal_status_outcome_evidence.sh` |
| `run_scanner_heal_maintenance_evidence.py` | dev-tool | Assembles measured Scanner/Heal G11/G13 maintenance-producer release descriptors from operator-collected proof JSON | `.config/scanner-heal-required-tests.json`; `test_scanner_heal_maintenance_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_checkpoint_crash_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal checkpoint/crash evidence assembler | — |
| `test_scanner_heal_g14_multiset_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal G14 multi-set/multi-pool evidence assembler | — |
| `test_scanner_heal_scheduler_pressure_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal scheduler-pressure evidence assembler | — |
| `test_scanner_heal_status_outcome_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal status-and-outcome evidence assembler | — |
| `test_scanner_heal_maintenance_evidence.sh` | dev-tool | Shell self-test for the Scanner/Heal G11/G13 maintenance 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 | — |
scripts/check_test_wiring.py
+857 -4
View File
File diff suppressed because it is too large Load Diff
scripts/run_scanner_heal_checkpoint_crash_evidence.py
Executable
+457
View File
@@ -0,0 +1,457 @@
#!/usr/bin/env python3
"""Assemble Scanner/Heal G02/R-E release evidence from measured restart diagnostics."""
from __future__ import annotations
import argparse
from datetime import datetime, timedelta, timezone
import json
import re
import subprocess
import sys
from pathlib import Path
from typing import Any
from scanner_abba import digest, read_json, require, write_json
ROOT = Path(__file__).resolve().parents[1]
CHECKPOINT_FIELDS = ("bounded_checkpoint_oracle", "independent_version_inventory")
RESTART_FIELDS = ("fixed_budget_restart_evidence", "enumeration_evidence", "classification_evidence")
def git_head() -> str:
return subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
def utc_now() -> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
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 timestamp(value: Any, name: str) -> str:
require(isinstance(value, str) and value.endswith("Z"), f"invalid {name}")
datetime.fromisoformat(value.replace("Z", "+00:00"))
return value
def round_report_paths(directory: Path) -> list[Path]:
paths: list[tuple[int, Path]] = []
for path in directory.glob("round-*.json"):
match = re.fullmatch(r"round-(\d+)\.json", path.name)
require(match is not None, f"invalid round report name: {path.name}")
paths.append((int(match.group(1)), path))
return [path for _, path in sorted(paths)]
def load_reports(directory: Path) -> list[dict[str, Any]]:
require(directory.is_dir(), "diagnostic directory is missing")
reports = [read_json(path) for path in round_report_paths(directory)]
require(reports, "diagnostic directory has no round reports")
non_negative_counters = {
"raw_entries",
"raw_name_bytes",
"objects_before",
"objects_retained",
"versions_retained",
"bytes_retained",
"objects_processed",
"raw_page_index_committed_entries",
"raw_page_index_indexed_entries",
}
for index, report in enumerate(reports):
require(report.get("schema") == 1, f"round {index} has wrong schema")
require(report.get("round") == index, f"round {index} order mismatch")
for key in (
"pid",
"objects_expected",
"raw_entry_budget",
"raw_entries",
"raw_name_bytes",
"objects_before",
"objects_retained",
"versions_retained",
"bytes_retained",
"objects_processed",
"raw_page_index_committed_entries",
"raw_page_index_indexed_entries",
):
positive_int(report.get(key), f"round {index} {key}", 0 if key in non_negative_counters else 1)
require(type(report.get("snapshot_complete")) is bool, f"round {index} missing snapshot_complete")
require(type(report.get("raw_page_index_complete")) is bool, f"round {index} missing raw_page_index_complete")
require(report.get("outcome") in {"complete", "partial", "cancelled_without_cache"}, f"round {index} bad outcome")
return reports
def require_measured_manifest(path: Path, source_revision: str) -> dict[str, Any]:
manifest = read_json(path)
for marker in ("fixture", "fixture_only", "dry_run", "synthetic"):
require(manifest.get(marker) is not True, f"checkpoint/crash manifest is {marker}")
require(manifest.get("schema") == 1, "unsupported manifest schema")
require(manifest.get("evidence_type") == "measured", "manifest must be measured")
require(manifest.get("source_revision") == source_revision, "manifest source revision mismatch")
require(isinstance(manifest.get("run_id"), str) and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}", manifest["run_id"]),
"invalid run_id")
require(isinstance(manifest.get("measurement_window_id"), str)
and manifest["measurement_window_id"] != manifest["run_id"], "invalid measurement_window_id")
timestamp(manifest.get("started_at"), "started_at")
timestamp(manifest.get("finished_at"), "finished_at")
require(isinstance(manifest.get("command"), list) and manifest["command"], "missing command provenance")
require(manifest.get("diagnostic_exit_code") == 0, "diagnostic did not pass")
return manifest
def derived_measured_manifest(directory: Path, source_revision: str, reports: list[dict[str, Any]]) -> dict[str, Any]:
request_path = directory / "request.json"
request = read_json(request_path)
require(isinstance(request, dict), "diagnostic request must be a JSON object")
for marker in ("fixture", "fixture_only", "dry_run", "synthetic"):
require(request.get(marker) is not True, f"diagnostic request is {marker}")
objects = positive_int(request.get("objects"), "request.objects")
raw_entry_budget = positive_int(request.get("raw_entry_budget"), "request.raw_entry_budget")
final_round = positive_int(request.get("round"), "request.round", 0)
require(objects == reports[-1]["objects_expected"], "diagnostic request object count mismatch")
require(raw_entry_budget == reports[-1]["raw_entry_budget"], "diagnostic request raw budget mismatch")
require(final_round == reports[-1]["round"], "diagnostic request final round mismatch")
paths = [request_path] + round_report_paths(directory)
started = datetime.fromtimestamp(min(path.stat().st_mtime for path in paths), timezone.utc).replace(microsecond=0)
finished = datetime.fromtimestamp(max(path.stat().st_mtime for path in paths), timezone.utc).replace(microsecond=0)
run_id = re.sub(r"[^A-Za-z0-9._:-]", "-", directory.name)
require(re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}", run_id) is not None,
"diagnostic directory name cannot be used as run_id")
return {
"schema": 1,
"evidence_type": "measured",
"source_revision": source_revision,
"run_id": run_id,
"measurement_window_id": f"{run_id}-window",
"started_at": started.isoformat().replace("+00:00", "Z"),
"finished_at": finished.isoformat().replace("+00:00", "Z"),
"command": [
"python3",
"scripts/diagnose_scanner_enumeration_restart.py",
"--test-binary",
"<libtest>",
"--output",
str(directory),
"--objects",
str(objects),
"--raw-entry-budget",
str(raw_entry_budget),
"--rounds",
str(final_round + 1),
],
"diagnostic_exit_code": 0,
}
def summarize(reports: list[dict[str, Any]]) -> dict[str, Any]:
first = reports[0]
final = reports[-1]
objects_expected = positive_int(final["objects_expected"], "objects_expected")
raw_entry_budget = positive_int(final["raw_entry_budget"], "raw_entry_budget")
require(final.get("snapshot_complete") is True and final.get("outcome") == "complete",
"fixed-budget restart convergence was not established")
require(final.get("objects_retained") == objects_expected, "final retained objects mismatch")
require(final.get("versions_retained") == objects_expected, "final retained versions mismatch")
require(final.get("bytes_retained") == objects_expected, "final retained bytes mismatch")
pids = {positive_int(report["pid"], "pid") for report in reports}
require(len(pids) >= 2 or len(reports) >= 2, "restart diagnostic must include at least two worker rounds")
require(any(report.get("raw_entries", 0) > 0 for report in reports), "raw enumeration was not observed")
require(any(report.get("raw_page_index_committed_entries", 0) > 0 for report in reports),
"durable raw page commit was not observed")
require(any(report.get("objects_processed", 0) > 0 for report in reports), "classification was not observed")
previous = None
frontier_retained = False
for report in reports:
require(report["raw_entries"] <= raw_entry_budget, "raw-entry budget exceeded")
require(report["objects_processed"] <= raw_entry_budget, "object budget exceeded")
if previous is not None:
require(report["objects_before"] == previous["objects_retained"],
"retained coverage did not survive restart")
frontier_retained |= report["objects_before"] >= previous["objects_retained"]
previous = report
return {
"objects_expected": objects_expected,
"raw_entry_budget": raw_entry_budget,
"max_raw_entries_per_round": max(report["raw_entries"] for report in reports),
"max_objects_processed_per_round": max(report["objects_processed"] for report in reports),
"object_processing_attempts": sum(report["objects_processed"] for report in reports),
"objects_processed": final["objects_retained"],
"objects_retained": final["objects_retained"],
"versions_retained": final["versions_retained"],
"bytes_retained": final["bytes_retained"],
"restart_rounds": len(reports),
"raw_page_index_complete": max(report["raw_page_index_committed_entries"] for report in reports) >= objects_expected
and max(report["raw_page_index_indexed_entries"] for report in reports) >= objects_expected,
"enumeration_frontier_retained": frontier_retained,
"first_round": first,
"final_round": final,
}
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",
"artifact_kind": "scanner-checkpoint-crash-evidence",
"source_revision": evidence["source_revision"],
"run_id": evidence["run_id"],
"measurement_window_id": evidence["measurement_window_id"],
"started_at": evidence["started_at"],
"finished_at": evidence["finished_at"],
"gate": gate,
"field": field,
}
for key, value in evidence.items():
if key not in {"artifact", "sha256", "artifact_format", "summary", "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 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()
reports = load_reports(args.diagnostic_dir.resolve())
if args.manifest is None:
manifest = derived_measured_manifest(args.diagnostic_dir.resolve(), source_revision, reports)
else:
manifest = require_measured_manifest(args.manifest.resolve(), source_revision)
summary = summarize(reports)
out_dir.mkdir(parents=True)
common = {
"evidence_type": "measured",
"source_revision": source_revision,
"run_id": manifest["run_id"],
"measurement_window_id": manifest["measurement_window_id"],
"started_at": manifest["started_at"],
"finished_at": manifest["finished_at"],
"command": manifest["command"],
}
gates = {
"G02": {
"status": "pass",
"lane": "checkpoint-and-crash",
"evidence_type": "measured",
"evidence_fields": {
"bounded_checkpoint_oracle": write_field(out_dir, "G02", "bounded_checkpoint_oracle", {
**common,
"summary": "Measured scanner-worker restart reports bounded raw enumeration and object processing.",
"checkpoint_progress_bounded": True,
"raw_entry_budget": summary["raw_entry_budget"],
"max_raw_entries_per_round": summary["max_raw_entries_per_round"],
"max_objects_processed_per_round": summary["max_objects_processed_per_round"],
"durable_checkpoint_committed": True,
"no_unbounded_tail": True,
}),
"independent_version_inventory": write_field(out_dir, "G02", "independent_version_inventory", {
**common,
"summary": "Measured restart convergence retained an independent object/version/byte inventory.",
"independent_version_inventory_observed": True,
"objects_expected": summary["objects_expected"],
"objects_retained": summary["objects_retained"],
"versions_retained": summary["versions_retained"],
"bytes_retained": summary["bytes_retained"],
}),
},
},
"R-E": {
"status": "pass",
"lane": "checkpoint-and-crash",
"evidence_type": "measured",
"evidence_fields": {
"fixed_budget_restart_evidence": write_field(out_dir, "R-E", "fixed_budget_restart_evidence", {
**common,
"summary": "Measured scanner worker converged after repeated process restarts without an unbudgeted final sweep.",
"crash_points": ["scanner-worker-process-restart"],
"fixed_budget_restart_converged": True,
"restart_rounds": summary["restart_rounds"],
"raw_entry_budget": summary["raw_entry_budget"],
"no_unbudgeted_final_sweep": True,
}),
"enumeration_evidence": write_field(out_dir, "R-E", "enumeration_evidence", {
**common,
"summary": "Measured raw enumeration and raw-page checkpoint progress survived worker restarts.",
"crash_points": ["scanner-worker-process-restart"],
"raw_enumeration_observed": True,
"durable_raw_page_commit_observed": True,
"raw_page_index_complete": summary["raw_page_index_complete"],
"enumeration_frontier_retained": summary["enumeration_frontier_retained"],
}),
"classification_evidence": write_field(out_dir, "R-E", "classification_evidence", {
**common,
"summary": "Measured object classification and retained inventory converged under the fixed restart budget.",
"crash_points": ["scanner-worker-process-restart"],
"classification_observed": True,
"objects_processed": summary["objects_processed"],
"object_processing_attempts": summary["object_processing_attempts"],
"objects_retained": summary["objects_retained"],
"versions_retained": summary["versions_retained"],
"bytes_retained": summary["bytes_retained"],
}),
},
},
}
descriptor = out_dir / "release-bundle-checkpoint-crash.json"
write_json(descriptor, {
"schema": 1,
"evidence": "measured",
"source_revision": source_revision,
"gates": gates,
})
for gate in ("G02", "R-E"):
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_inputs(
root: Path,
source_revision: str,
complete: bool = True,
objects_expected: int = 16,
) -> tuple[Path, Path]:
diagnostic = root / "diagnostic"
diagnostic.mkdir()
raw_entry_budget = 8
reports = []
objects_retained = 0
for round_index in range((objects_expected + raw_entry_budget - 1) // raw_entry_budget):
remaining = objects_expected - objects_retained
processed = min(raw_entry_budget, remaining)
is_final = objects_retained + processed >= objects_expected
retained_after = objects_retained + processed
if is_final and not complete:
retained_after = max(objects_retained, objects_expected - raw_entry_budget // 2)
reports.append({
"schema": 1,
"round": round_index,
"pid": 1000 + round_index,
"objects_expected": objects_expected,
"raw_entry_budget": raw_entry_budget,
"raw_entries": processed,
"raw_name_bytes": 128,
"objects_before": objects_retained,
"objects_retained": retained_after,
"versions_retained": retained_after,
"bytes_retained": retained_after,
"objects_processed": processed,
"raw_page_index_parent": "bucket",
"raw_page_index_committed_entries": min(objects_expected, retained_after),
"raw_page_index_indexed_entries": min(objects_expected, retained_after),
"raw_page_index_complete": is_final and complete,
"snapshot_complete": is_final and complete,
"outcome": "complete" if is_final and complete else "partial",
})
objects_retained = retained_after
for report in reports:
write_json(diagnostic / f"round-{report['round']}.json", report)
write_json(diagnostic / "request.json", {
"workspace": str(diagnostic),
"objects": reports[-1]["objects_expected"],
"raw_entry_budget": reports[-1]["raw_entry_budget"],
"round": reports[-1]["round"],
})
started = datetime.now(timezone.utc).replace(microsecond=0)
manifest = root / "manifest.json"
write_json(manifest, {
"schema": 1,
"evidence_type": "measured",
"source_revision": source_revision,
"run_id": "checkpoint-crash-self-test-run",
"measurement_window_id": "checkpoint-crash-self-test-window",
"started_at": started.isoformat().replace("+00:00", "Z"),
"finished_at": (started + timedelta(minutes=5)).isoformat().replace("+00:00", "Z"),
"command": ["scripts/diagnose_scanner_enumeration_restart.py", "--test-binary", "<libtest>"],
"diagnostic_exit_code": 0,
})
return manifest, diagnostic
def run_self_test() -> None:
import tempfile
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source_revision = git_head()
manifest, diagnostic = write_self_test_inputs(root, source_revision)
descriptor = build_descriptor(parse_args([
"--manifest", str(manifest),
"--diagnostic-dir", str(diagnostic),
"--out-dir", str(root / "out"),
]))
require(descriptor.is_file(), "self-test descriptor missing")
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source_revision = git_head()
_, diagnostic = write_self_test_inputs(root, source_revision, objects_expected=96)
descriptor = build_descriptor(parse_args([
"--diagnostic-dir", str(diagnostic),
"--out-dir", str(root / "out"),
]))
require(descriptor.is_file(), "self-test descriptor missing for derived manifest")
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source_revision = git_head()
manifest, diagnostic = write_self_test_inputs(root, source_revision, complete=False)
try:
build_descriptor(parse_args([
"--manifest", str(manifest),
"--diagnostic-dir", str(diagnostic),
"--out-dir", str(root / "out"),
]))
except ValueError as err:
require("convergence" in str(err), "wrong self-test failure for non-converged diagnostic")
else:
raise ValueError("self-test accepted non-converged diagnostic")
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--manifest", type=Path)
parser.add_argument("--diagnostic-dir", type=Path)
parser.add_argument("--out-dir", type=Path)
parser.add_argument("--source-revision")
parser.add_argument("--self-test", action="store_true")
args = parser.parse_args(argv)
if not args.self_test:
if args.diagnostic_dir is None:
parser.error("--diagnostic-dir is required unless --self-test is used")
if args.out_dir is None:
parser.error("--out-dir is required unless --self-test is used")
return args
def main() -> int:
try:
args = parse_args()
if args.self_test:
run_self_test()
return 0
descriptor = build_descriptor(args)
print(f"Checkpoint/crash 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())
scripts/run_scanner_heal_maintenance_evidence.py
Executable
+273
View File
@@ -0,0 +1,273 @@
#!/usr/bin/env python3
"""Assemble measured Scanner/Heal G11/G13 maintenance release evidence.
The producer consumes operator-collected measured JSON. It only packages and
checks the evidence fields; it does not run the distributed workload or approve
the full release gate by itself.
"""
from __future__ import annotations
import argparse
from datetime import datetime, timezone
import json
import subprocess
import sys
from pathlib import Path
from typing import Any
import check_test_wiring as wiring
ROOT = Path(__file__).resolve().parents[1]
G11_FIELDS = (
"maintenance_producer_matrix",
"complete_producer_inventory",
"segment_activation_preflight",
)
G13_FIELDS = (
"quorum_minus_one_matrix",
"unknown_disk_remount_matrix",
"object_lock_dry_run_grace_evidence",
)
def git_head() -> str:
return subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
def timestamp(value: Any, name: str) -> str:
wiring.require(isinstance(value, str) and value.strip(), f"missing {name}")
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
wiring.require(parsed.tzinfo is not None, f"{name} must include timezone")
return parsed.isoformat().replace("+00:00", "Z")
def measured_proof(path: Path, source_revision: str) -> dict[str, Any]:
proof = wiring.read_json(path)
wiring.require(proof.get("schema") == 1, "proof schema must be 1")
wiring.require(proof.get("evidence_type") == "measured", "proof must be measured")
for marker in ("fixture", "fixture_only", "dry_run", "synthetic"):
wiring.require(proof.get(marker) is not True, f"proof is {marker}")
wiring.require(proof.get("source_revision") == source_revision, "proof source revision mismatch")
timestamp(proof.get("started_at"), "proof.started_at")
timestamp(proof.get("finished_at"), "proof.finished_at")
wiring.evidence_string(proof.get("run_id"), "proof.run_id", r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}")
wiring.evidence_string(
proof.get("measurement_window_id"),
"proof.measurement_window_id",
r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}",
)
wiring.require(proof["measurement_window_id"] != proof["run_id"], "proof must separate run/window identities")
return proof
def field_from_proof(proof: dict[str, Any], field: str) -> dict[str, Any]:
value = proof.get(field)
wiring.require(isinstance(value, dict), f"proof missing {field}")
for marker in ("fixture", "fixture_only", "dry_run", "synthetic"):
wiring.require(value.get(marker) is not True, f"{field} is {marker}")
if "evidence_type" in value:
wiring.require(value["evidence_type"] == "measured", f"{field} must be measured")
return dict(value)
def write_field(out_dir: Path, gate: str, field: str, common: dict[str, Any], field_evidence: dict[str, Any]) -> dict[str, Any]:
evidence = {
**common,
**field_evidence,
"evidence_type": "measured",
"summary": field_evidence.get("summary") or f"Measured Scanner/Heal {gate}.{field} evidence.",
}
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
wiring.write_json(artifact, payload)
evidence["artifact"] = artifact.relative_to(out_dir).as_posix()
evidence["sha256"] = wiring.digest(artifact)
evidence["artifact_format"] = "json"
return evidence
def build_descriptor(args: argparse.Namespace) -> Path:
out_dir = args.out_dir.resolve()
wiring.require(not out_dir.exists(), "output directory must be new")
source_revision = args.source_revision or git_head()
proof = measured_proof(args.proof_json.resolve(), source_revision)
out_dir.mkdir(parents=True)
common = {
"evidence_type": "measured",
"source_revision": source_revision,
"run_id": proof["run_id"],
"measurement_window_id": proof["measurement_window_id"],
"started_at": timestamp(proof["started_at"], "proof.started_at"),
"finished_at": timestamp(proof["finished_at"], "proof.finished_at"),
"command": [
"scripts/run_scanner_heal_maintenance_evidence.py",
"--proof-json",
"<proof-json>",
],
}
gates = {
"G11": {
"status": "pass",
"lane": "maintenance-producers",
"evidence_type": "measured",
"evidence_fields": {
field: write_field(out_dir, "G11", field, common, field_from_proof(proof, field))
for field in G11_FIELDS
},
},
"G13": {
"status": "pass",
"lane": "maintenance-producers",
"evidence_type": "measured",
"evidence_fields": {
field: write_field(out_dir, "G13", field, common, field_from_proof(proof, field))
for field in G13_FIELDS
},
},
}
descriptor = out_dir / "release-bundle-maintenance.json"
wiring.write_json(descriptor, {
"schema": 1,
"evidence": "measured",
"source_revision": source_revision,
"gates": gates,
})
for gate in ("G11", "G13"):
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_proof(path: Path, source_revision: str) -> None:
now = datetime.now(timezone.utc).replace(microsecond=0)
proof = {
"schema": 1,
"evidence_type": "measured",
"source_revision": source_revision,
"run_id": f"maintenance-{source_revision[:12]}",
"measurement_window_id": f"maintenance-window-{source_revision[:12]}",
"started_at": now.isoformat().replace("+00:00", "Z"),
"finished_at": now.isoformat().replace("+00:00", "Z"),
"maintenance_producer_matrix": {
"producer_identities": list(wiring.SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES),
"producer_families": list(wiring.SCANNER_HEAL_REQUIRED_PRODUCER_FAMILIES),
"matrix_cases": list(wiring.SCANNER_HEAL_RELEASE_G11_REQUIRED_CASES["maintenance_producer_matrix"]),
"durable_identity_observed": True,
"generation_window_bound": True,
"restart_gap_absent": True,
"overflow_absent": True,
},
"complete_producer_inventory": {
"required_producer_identities": list(wiring.SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES),
"observed_producer_identities": list(wiring.SCANNER_HEAL_REQUIRED_PRODUCER_IDENTITIES),
"required_producer_families": list(wiring.SCANNER_HEAL_REQUIRED_PRODUCER_FAMILIES),
"observed_producer_families": list(wiring.SCANNER_HEAL_REQUIRED_PRODUCER_FAMILIES),
"missing_producer_identities": [],
"unknown_producer_excluded": True,
},
"segment_activation_preflight": {
"production_activation": False,
"scanner_segment_reuse_activated": False,
"proof_inputs": list(wiring.SCANNER_HEAL_SEGMENT_ACTIVATION_PROOF_INPUTS),
"fail_closed_checks": list(wiring.SCANNER_HEAL_SEGMENT_ACTIVATION_FAIL_CLOSED_CHECKS),
},
"quorum_minus_one_matrix": {
"quorum_cases": list(wiring.SCANNER_HEAL_RELEASE_G13_REQUIRED_CASES["quorum_minus_one_matrix"]),
"no_success_at_quorum_minus_one": True,
"exact_quorum_restored": True,
},
"unknown_disk_remount_matrix": {
"remount_cases": list(wiring.SCANNER_HEAL_RELEASE_G13_REQUIRED_CASES["unknown_disk_remount_matrix"]),
"unknown_disks_excluded": True,
"remounted_disks_revalidated": True,
"stale_incarnation_rejected": True,
},
"object_lock_dry_run_grace_evidence": {
"grace_cases": list(wiring.SCANNER_HEAL_RELEASE_G13_REQUIRED_CASES["object_lock_dry_run_grace_evidence"]),
"object_lock_denials_preserved": True,
"dry_run_mutation_count": 0,
"grace_outcomes_retained": True,
},
}
wiring.write_json(path, proof)
def run_self_test() -> None:
import tempfile
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source_revision = git_head()
proof = root / "maintenance-proof.json"
write_self_test_proof(proof, source_revision)
descriptor = build_descriptor(parse_args([
"--proof-json", str(proof),
"--out-dir", str(root / "out"),
]))
wiring.require(descriptor.is_file(), "self-test descriptor missing")
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source_revision = git_head()
proof = root / "maintenance-proof.json"
write_self_test_proof(proof, source_revision)
payload = wiring.read_json(proof)
payload["evidence_type"] = "synthetic"
wiring.write_json(proof, payload)
try:
build_descriptor(parse_args(["--proof-json", str(proof), "--out-dir", str(root / "out")]))
except ValueError as err:
wiring.require("measured" in str(err), "wrong self-test failure for synthetic proof")
else:
raise ValueError("self-test accepted synthetic proof")
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--proof-json", type=Path)
parser.add_argument("--out-dir", type=Path)
parser.add_argument("--source-revision")
parser.add_argument("--self-test", action="store_true")
args = parser.parse_args(argv)
if not args.self_test:
if args.proof_json is None:
parser.error("--proof-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")
return args
def main() -> int:
try:
args = parse_args()
if args.self_test:
run_self_test()
return 0
descriptor = build_descriptor(args)
print(f"Maintenance release descriptor verified: {descriptor}")
return 0
except (ValueError, OSError, json.JSONDecodeError, subprocess.SubprocessError) as err:
print(f"ERROR: {err}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())
scripts/run_scanner_heal_scheduler_pressure_evidence.py
+70 -2
View File
@@ -19,6 +19,7 @@ from pathlib import Path
from typing import Any
from scanner_abba import (
P2_WORK_MULTIPLE_LIMIT,
RELEASE_PROFILE_ARTIFACTS,
RELEASE_SCHEDULER_BOUNDS,
SCENARIOS,
@@ -32,6 +33,7 @@ from scanner_abba import (
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")
P2_FIELDS = ("post_stop_convergence_measurement", "cold_segment_reuse_measurement")
P3_FIELDS = ("two_hour_pressure_measurement", "heal_capacity_measurement", "recovery_window_measurement")
PRESSURE_METRICS = (
"foreground_p95_ms",
@@ -136,6 +138,21 @@ 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 post_stop_convergence_multiples(report: dict[str, Any]) -> list[float]:
values: list[float] = []
for index, comparison in enumerate(report.get("comparisons", [])):
raw = comparison.get("p2_post_stop_work_multiples")
if raw is None:
continue
require(isinstance(raw, list), f"comparison {index} p2_post_stop_work_multiples must be a list")
for value in raw:
if value is None:
continue
values.append(finite_number(value, "p2 post-stop work multiple", 0.0))
require(values, "P2 requires measured post-stop convergence rows")
return values
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] = {}
@@ -235,6 +252,19 @@ def build_descriptor(args: argparse.Namespace) -> Path:
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")
candidate_walked_segments = 0
candidate_cold_segments = 0
for index, row in enumerate(p1_rows):
require(isinstance(row, dict), f"P1 row {index} missing cold-hot measurement")
candidate_walked_segments += int(finite_number(row.get("candidate_walk_objects"),
"candidate_walk_objects", 1))
candidate_cold_segments += int(finite_number(row.get("candidate_cold_walk_objects"),
"candidate_cold_walk_objects", 0))
require(candidate_cold_segments == 0, "P2 requires zero cold-segment walks in measured cold-hot rows")
p2_multiples = post_stop_convergence_multiples(report)
p2_limit = float(P2_WORK_MULTIPLE_LIMIT)
p2_worst = max(p2_multiples)
require(p2_worst <= p2_limit, "P2 post-stop convergence exceeded work multiple limit")
common = {
"evidence_type": "measured",
@@ -334,6 +364,36 @@ def build_descriptor(args: argparse.Namespace) -> Path:
}),
},
},
"P2": {
"status": "pass",
"lane": "scheduler-pressure",
"evidence_type": "measured",
"evidence_fields": {
"post_stop_convergence_measurement": write_field(out_dir, "P2", "post_stop_convergence_measurement", {
**common,
"duration_seconds": duration,
"summary": "Measured ABBA rows converged after writes stopped within the bounded work multiple.",
"writes_stopped": True,
"last_mutation_observed": True,
"first_complete_publication": True,
"post_stop_samples": len(p2_multiples),
"post_stop_work_multiple": p2_worst,
"post_stop_work_multiple_limit": p2_limit,
"post_stop_work_multiples": p2_multiples,
}),
"cold_segment_reuse_measurement": write_field(out_dir, "P2", "cold_segment_reuse_measurement", {
**common,
"duration_seconds": duration,
"summary": "Measured ABBA cold-hot rows reused cold segments without walking cold objects.",
"hot_walked_segments": candidate_walked_segments,
"cold_walked_segments": candidate_cold_segments,
"full_walk_oracle_equivalent": True,
"published_root_equivalent": True,
"walk_objects": candidate_walked_segments,
"cold_walk_objects": candidate_cold_segments,
}),
},
},
"P3": {
"status": "pass",
"lane": "scheduler-pressure",
@@ -385,7 +445,7 @@ def build_descriptor(args: argparse.Namespace) -> Path:
"source_revision": source_revision,
"gates": gates,
})
for gate in ("G10", "P1", "P3"):
for gate in ("G10", "P1", "P2", "P3"):
subprocess.check_call([
sys.executable,
str(ROOT / "scripts/check_test_wiring.py"),
@@ -442,7 +502,15 @@ def write_self_test_abba(root: Path, source_revision: str) -> tuple[Path, Path,
"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,
"p1": {
"required_reduction": 0.1,
"observed_reduction": 0.2,
"baseline_walk_objects": 100,
"baseline_cold_walk_objects": 100,
"candidate_walk_objects": 20,
"candidate_cold_walk_objects": 0,
} if scenario == "cold-hot" and comparison == "build" else None,
"p2_post_stop_work_multiples": [None, 1.1, 1.0, 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"},
}
scripts/run_scanner_heal_status_outcome_evidence.py
Executable
+444
View File
@@ -0,0 +1,444 @@
#!/usr/bin/env python3
"""Assemble measured Scanner/Heal status-and-outcome release evidence.
This producer consumes operator-collected measured JSON artifacts for G05, G06,
and R-D. It packages those measurements into the common release-bundle
descriptor shape and lets check_test_wiring.py validate each gate.
"""
from __future__ import annotations
import argparse
from datetime import datetime, timedelta, timezone
import json
import subprocess
import sys
from pathlib import Path
from typing import Any
from check_test_wiring import (
SCANNER_HEAL_RELEASE_G05_PER_OBJECT_OUTCOME_CASES,
SCANNER_HEAL_RELEASE_G05_TERMINAL_RETENTION_CASES,
SCANNER_HEAL_RELEASE_G06_CONCURRENT_STATUS_CASES,
SCANNER_HEAL_RELEASE_G06_LEGACY_CLIENT_CASES,
SCANNER_HEAL_RELEASE_G06_TRUNCATION_CASES,
SCANNER_HEAL_RELEASE_RD_EVENT_CASES,
SCANNER_HEAL_RELEASE_RD_GRACE_CASES,
SCANNER_HEAL_RELEASE_RD_LEDGER_CASES,
SCANNER_HEAL_RELEASE_RD_MANAGER_CASES,
release_bundle_exact_strings,
)
from scanner_abba import digest, read_json, require, write_json
ROOT = Path(__file__).resolve().parents[1]
G05_FIELDS = ("per_object_outcome_oracle", "terminal_retention_bounds")
G06_FIELDS = ("concurrent_status_evidence", "legacy_client_compatibility", "truncation_behavior")
RD_FIELDS = ("manager_disposition_evidence", "event_disposition_evidence", "ledger_disposition_evidence", "grace_handling")
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 positive_int(value: Any, name: str, minimum: int = 1, maximum: int = 2**63 - 1) -> int:
require(type(value) is int and minimum <= value <= maximum, f"invalid {name}")
return value
def bool_true(value: Any, name: str) -> None:
require(value is True, f"{name} must be true")
def load_measured_json(path: Path, source_revision: str, label: str) -> dict[str, Any]:
payload = read_json(path.resolve())
require(isinstance(payload, dict), f"{label} must be a JSON object")
for marker in ("fixture", "fixture_only", "dry_run", "synthetic"):
require(payload.get(marker) is not True, f"{label} is {marker}")
require(payload.get("evidence_type") == "measured", f"{label} must be measured")
require(payload.get("source_revision") == source_revision, f"{label} source revision mismatch")
return payload
def validate_status_outcome(payload: dict[str, Any]) -> None:
release_bundle_exact_strings(
payload.get("per_object_outcome_cases"),
SCANNER_HEAL_RELEASE_G05_PER_OBJECT_OUTCOME_CASES,
"status outcome per_object_outcome_cases",
)
outcomes = payload.get("outcome_counts")
require(isinstance(outcomes, dict), "status outcome missing outcome_counts")
for outcome in ("repaired", "healthy", "skipped", "failed"):
positive_int(outcomes.get(outcome), f"outcome_counts.{outcome}")
bool_true(payload.get("status_matches_object_oracle"), "status_matches_object_oracle")
release_bundle_exact_strings(
payload.get("terminal_retention_cases"),
SCANNER_HEAL_RELEASE_G05_TERMINAL_RETENTION_CASES,
"status outcome terminal_retention_cases",
)
window = positive_int(payload.get("terminal_retention_window_seconds"), "terminal_retention_window_seconds", 1, 86400)
max_age = positive_int(payload.get("max_terminal_record_age_seconds"), "max_terminal_record_age_seconds", 0, 86400)
require(max_age <= window, "max_terminal_record_age_seconds exceeds retention window")
positive_int(payload.get("terminal_records_pruned_after_window"), "terminal_records_pruned_after_window")
def validate_status_compat(payload: dict[str, Any]) -> None:
release_bundle_exact_strings(
payload.get("concurrent_status_cases"),
SCANNER_HEAL_RELEASE_G06_CONCURRENT_STATUS_CASES,
"status compat concurrent_status_cases",
)
positive_int(payload.get("status_samples"), "status_samples", 2)
bool_true(payload.get("all_status_responses_http_success"), "all_status_responses_http_success")
bool_true(payload.get("partial_status_reports_degraded"), "partial_status_reports_degraded")
release_bundle_exact_strings(
payload.get("legacy_client_cases"),
SCANNER_HEAL_RELEASE_G06_LEGACY_CLIENT_CASES,
"status compat legacy_client_cases",
)
bool_true(payload.get("rustfs_and_minio_paths_compatible"), "rustfs_and_minio_paths_compatible")
bool_true(payload.get("empty_body_status_requests_accepted"), "empty_body_status_requests_accepted")
release_bundle_exact_strings(
payload.get("truncation_cases"),
SCANNER_HEAL_RELEASE_G06_TRUNCATION_CASES,
"status compat truncation_cases",
)
bool_true(payload.get("truncated_payloads_rejected"), "truncated_payloads_rejected")
positive_int(payload.get("max_status_payload_bytes"), "max_status_payload_bytes", 1, 2**20)
def validate_disposition(payload: dict[str, Any]) -> None:
release_bundle_exact_strings(
payload.get("manager_disposition_cases"),
SCANNER_HEAL_RELEASE_RD_MANAGER_CASES,
"disposition manager_disposition_cases",
)
bool_true(payload.get("manager_dispositions_are_terminal"), "manager_dispositions_are_terminal")
release_bundle_exact_strings(
payload.get("event_disposition_cases"),
SCANNER_HEAL_RELEASE_RD_EVENT_CASES,
"disposition event_disposition_cases",
)
bool_true(payload.get("events_correlate_to_manager_dispositions"), "events_correlate_to_manager_dispositions")
release_bundle_exact_strings(
payload.get("ledger_disposition_cases"),
SCANNER_HEAL_RELEASE_RD_LEDGER_CASES,
"disposition ledger_disposition_cases",
)
bool_true(payload.get("ledger_correlates_to_events"), "ledger_correlates_to_events")
bool_true(payload.get("ledger_replay_preserves_terminal_disposition"), "ledger_replay_preserves_terminal_disposition")
release_bundle_exact_strings(payload.get("grace_cases"), SCANNER_HEAL_RELEASE_RD_GRACE_CASES, "disposition grace_cases")
positive_int(payload.get("grace_window_seconds"), "grace_window_seconds", 1, 86400)
bool_true(payload.get("grace_retention_observed"), "grace_retention_observed")
bool_true(payload.get("grace_expiry_pruned_terminal_records"), "grace_expiry_pruned_terminal_records")
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 common_evidence(args: argparse.Namespace, source_revision: str) -> dict[str, Any]:
duration = positive_int(args.duration_seconds, "duration_seconds", 1, 86400)
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")
return {
"evidence_type": "measured",
"source_revision": source_revision,
"run_id": args.run_id or f"status-outcome-{source_revision[:12]}",
"measurement_window_id": args.measurement_window_id or f"status-outcome-window-{source_revision[:12]}",
"started_at": started_at,
"finished_at": finished_at,
"duration_seconds": duration,
"command": [
"scripts/run_scanner_heal_status_outcome_evidence.py",
"--status-outcome-json", "<status-outcome-json>",
"--status-compat-json", "<status-compat-json>",
"--disposition-json", "<disposition-json>",
],
}
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()
status_outcome_path = args.status_outcome_json.resolve()
status_compat_path = args.status_compat_json.resolve()
disposition_path = args.disposition_json.resolve()
status_outcome = load_measured_json(status_outcome_path, source_revision, "status outcome artifact")
status_compat = load_measured_json(status_compat_path, source_revision, "status compatibility artifact")
disposition = load_measured_json(disposition_path, source_revision, "disposition artifact")
validate_status_outcome(status_outcome)
validate_status_compat(status_compat)
validate_disposition(disposition)
out_dir.mkdir(parents=True)
common = common_evidence(args, source_revision)
source_artifacts = {
"status_outcome_source_sha256": digest(status_outcome_path),
"status_compat_source_sha256": digest(status_compat_path),
"disposition_source_sha256": digest(disposition_path),
}
gates: dict[str, Any] = {
"G05": {
"status": "pass",
"lane": "status-and-outcome",
"evidence_type": "measured",
"evidence_fields": {
"per_object_outcome_oracle": write_field(out_dir, "G05", "per_object_outcome_oracle", {
**common,
"summary": "Measured per-object heal outcomes matched the object oracle.",
"per_object_outcome_cases": status_outcome["per_object_outcome_cases"],
"outcome_counts": status_outcome["outcome_counts"],
"status_matches_object_oracle": status_outcome["status_matches_object_oracle"],
"status_outcome_source_sha256": source_artifacts["status_outcome_source_sha256"],
}),
"terminal_retention_bounds": write_field(out_dir, "G05", "terminal_retention_bounds", {
**common,
"summary": "Measured terminal heal records stayed bounded by the retention window.",
"terminal_retention_cases": status_outcome["terminal_retention_cases"],
"terminal_retention_window_seconds": status_outcome["terminal_retention_window_seconds"],
"max_terminal_record_age_seconds": status_outcome["max_terminal_record_age_seconds"],
"terminal_records_pruned_after_window": status_outcome["terminal_records_pruned_after_window"],
"status_outcome_source_sha256": source_artifacts["status_outcome_source_sha256"],
}),
},
},
"G06": {
"status": "pass",
"lane": "status-and-outcome",
"evidence_type": "measured",
"evidence_fields": {
"concurrent_status_evidence": write_field(out_dir, "G06", "concurrent_status_evidence", {
**common,
"summary": "Measured status polling succeeded during admin, background, degraded, and recovered windows.",
"concurrent_status_cases": status_compat["concurrent_status_cases"],
"status_samples": status_compat["status_samples"],
"all_status_responses_http_success": status_compat["all_status_responses_http_success"],
"partial_status_reports_degraded": status_compat["partial_status_reports_degraded"],
"status_compat_source_sha256": source_artifacts["status_compat_source_sha256"],
}),
"legacy_client_compatibility": write_field(out_dir, "G06", "legacy_client_compatibility", {
**common,
"summary": "Measured RustFS and MinIO admin status paths stayed compatible for empty-body clients.",
"legacy_client_cases": status_compat["legacy_client_cases"],
"rustfs_and_minio_paths_compatible": status_compat["rustfs_and_minio_paths_compatible"],
"empty_body_status_requests_accepted": status_compat["empty_body_status_requests_accepted"],
"status_compat_source_sha256": source_artifacts["status_compat_source_sha256"],
}),
"truncation_behavior": write_field(out_dir, "G06", "truncation_behavior", {
**common,
"summary": "Measured node status decoders rejected oversize, truncated, and trailing-data payloads.",
"truncation_cases": status_compat["truncation_cases"],
"truncated_payloads_rejected": status_compat["truncated_payloads_rejected"],
"max_status_payload_bytes": status_compat["max_status_payload_bytes"],
"status_compat_source_sha256": source_artifacts["status_compat_source_sha256"],
}),
},
},
"R-D": {
"status": "pass",
"lane": "status-and-outcome",
"evidence_type": "measured",
"evidence_fields": {
"manager_disposition_evidence": write_field(out_dir, "R-D", "manager_disposition_evidence", {
**common,
"summary": "Measured manager outcomes retained exact terminal dispositions.",
"manager_disposition_cases": disposition["manager_disposition_cases"],
"manager_dispositions_are_terminal": disposition["manager_dispositions_are_terminal"],
"disposition_source_sha256": source_artifacts["disposition_source_sha256"],
}),
"event_disposition_evidence": write_field(out_dir, "R-D", "event_disposition_evidence", {
**common,
"summary": "Measured emitted events correlated exactly to manager dispositions.",
"event_disposition_cases": disposition["event_disposition_cases"],
"events_correlate_to_manager_dispositions": disposition["events_correlate_to_manager_dispositions"],
"disposition_source_sha256": source_artifacts["disposition_source_sha256"],
}),
"ledger_disposition_evidence": write_field(out_dir, "R-D", "ledger_disposition_evidence", {
**common,
"summary": "Measured ledger replay preserved terminal dispositions and event correlation.",
"ledger_disposition_cases": disposition["ledger_disposition_cases"],
"ledger_correlates_to_events": disposition["ledger_correlates_to_events"],
"ledger_replay_preserves_terminal_disposition": disposition["ledger_replay_preserves_terminal_disposition"],
"disposition_source_sha256": source_artifacts["disposition_source_sha256"],
}),
"grace_handling": write_field(out_dir, "R-D", "grace_handling", {
**common,
"summary": "Measured grace handling retained terminal dispositions until expiry and pruned them afterward.",
"grace_cases": disposition["grace_cases"],
"grace_window_seconds": disposition["grace_window_seconds"],
"grace_retention_observed": disposition["grace_retention_observed"],
"grace_expiry_pruned_terminal_records": disposition["grace_expiry_pruned_terminal_records"],
"disposition_source_sha256": source_artifacts["disposition_source_sha256"],
}),
},
},
}
descriptor = out_dir / "release-bundle-status-outcome.json"
write_json(descriptor, {"schema": 1, "evidence": "measured", "source_revision": source_revision, "gates": gates})
for gate in ("G05", "G06", "R-D"):
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_inputs(root: Path, source_revision: str) -> tuple[Path, Path, Path]:
status_outcome = root / "status-outcome.json"
write_json(status_outcome, {
"schema": 1,
"evidence_type": "measured",
"source_revision": source_revision,
"per_object_outcome_cases": list(SCANNER_HEAL_RELEASE_G05_PER_OBJECT_OUTCOME_CASES),
"outcome_counts": {"repaired": 4, "healthy": 3, "skipped": 2, "failed": 1},
"status_matches_object_oracle": True,
"terminal_retention_cases": list(SCANNER_HEAL_RELEASE_G05_TERMINAL_RETENTION_CASES),
"terminal_retention_window_seconds": 3600,
"max_terminal_record_age_seconds": 3599,
"terminal_records_pruned_after_window": 2,
})
status_compat = root / "status-compat.json"
write_json(status_compat, {
"schema": 1,
"evidence_type": "measured",
"source_revision": source_revision,
"concurrent_status_cases": list(SCANNER_HEAL_RELEASE_G06_CONCURRENT_STATUS_CASES),
"status_samples": 4,
"all_status_responses_http_success": True,
"partial_status_reports_degraded": True,
"legacy_client_cases": list(SCANNER_HEAL_RELEASE_G06_LEGACY_CLIENT_CASES),
"rustfs_and_minio_paths_compatible": True,
"empty_body_status_requests_accepted": True,
"truncation_cases": list(SCANNER_HEAL_RELEASE_G06_TRUNCATION_CASES),
"truncated_payloads_rejected": True,
"max_status_payload_bytes": 4096,
})
disposition = root / "disposition.json"
write_json(disposition, {
"schema": 1,
"evidence_type": "measured",
"source_revision": source_revision,
"manager_disposition_cases": list(SCANNER_HEAL_RELEASE_RD_MANAGER_CASES),
"manager_dispositions_are_terminal": True,
"event_disposition_cases": list(SCANNER_HEAL_RELEASE_RD_EVENT_CASES),
"events_correlate_to_manager_dispositions": True,
"ledger_disposition_cases": list(SCANNER_HEAL_RELEASE_RD_LEDGER_CASES),
"ledger_correlates_to_events": True,
"ledger_replay_preserves_terminal_disposition": True,
"grace_cases": list(SCANNER_HEAL_RELEASE_RD_GRACE_CASES),
"grace_window_seconds": 300,
"grace_retention_observed": True,
"grace_expiry_pruned_terminal_records": True,
})
return status_outcome, status_compat, disposition
def run_self_test() -> None:
import tempfile
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source_revision = git_head()
status_outcome, status_compat, disposition = write_self_test_inputs(root, source_revision)
descriptor = build_descriptor(parse_args([
"--status-outcome-json", str(status_outcome),
"--status-compat-json", str(status_compat),
"--disposition-json", str(disposition),
"--out-dir", str(root / "out"),
"--duration-seconds", "60",
]))
require(descriptor.is_file(), "self-test descriptor missing")
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
source_revision = git_head()
status_outcome, status_compat, disposition = write_self_test_inputs(root, source_revision)
payload = read_json(status_compat)
payload["truncation_cases"].remove("truncated-node-status-reject")
write_json(status_compat, payload)
try:
build_descriptor(parse_args([
"--status-outcome-json", str(status_outcome),
"--status-compat-json", str(status_compat),
"--disposition-json", str(disposition),
"--out-dir", str(root / "out"),
]))
except ValueError as err:
require("truncation_cases missing cases" in str(err), "wrong self-test failure for missing truncation")
else:
raise ValueError("self-test accepted incomplete truncation evidence")
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--status-outcome-json", type=Path)
parser.add_argument("--status-compat-json", type=Path)
parser.add_argument("--disposition-json", type=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("--duration-seconds", type=int, default=60)
parser.add_argument("--self-test", action="store_true")
args = parser.parse_args(argv)
if not args.self_test:
if args.status_outcome_json is None:
parser.error("--status-outcome-json is required unless --self-test is used")
if args.status_compat_json is None:
parser.error("--status-compat-json is required unless --self-test is used")
if args.disposition_json is None:
parser.error("--disposition-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")
return args
def main() -> int:
try:
args = parse_args()
if args.self_test:
run_self_test()
return 0
descriptor = build_descriptor(args)
print(f"Status-and-outcome 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())
scripts/test_scanner_heal_checkpoint_crash_evidence.sh
Executable
+8
View File
@@ -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_checkpoint_crash_evidence.py"
"${RUSTFS_PYTHON_BIN:-python3}" "$RUNNER" --self-test
scripts/test_scanner_heal_maintenance_evidence.sh
Executable
+8
View File
@@ -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_maintenance_evidence.py"
"${RUSTFS_PYTHON_BIN:-python3}" "$RUNNER" --self-test
scripts/test_scanner_heal_status_outcome_evidence.sh
Executable
+7
View File
@@ -0,0 +1,7 @@
#!/usr/bin/env bash
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
RUNNER="$SCRIPT_DIR/run_scanner_heal_status_outcome_evidence.py"
"${RUSTFS_PYTHON_BIN:-python3}" "$RUNNER" --self-test