test(scanner): require hard evidence ABBA manifest (#7459)

Co-authored-by: zhi22915 <qiuzgang@gmail.com>
2026-09-08 13:06:00 +00:00 · 2026-09-08 11:32:34 +08:00
parent df6981d88e
commit 6fe83f87a4
6 changed files with 309 additions and 0 deletions
@@ -66,6 +66,7 @@ The manifest has the following JSON contract (all fields are required):
 | `rounds`, `duration_seconds`, `min_free_bytes` | 3..10 groups, 900..86400 seconds for measured runs, and the independently estimated free-space reservation in bytes. Synthetic runs may use 1 second. |
 | `baseline`, `candidate` | Each contains executable `binary`, full 40-character `revision`, and verified `sha256`. The runner rehashes binaries before every leg. |
 | `fixed` | `config_sha256`, `dataset_sha256`, `release_flags`, `durability`, `disk_type`, `cache_state`, `load_command`, `resource_isolation`, `topology` (`EC8+4`), and positive `offered_load_ops`. Hashes use 64 lowercase hexadecimal characters. |
+| `release_evidence` | Required for `measured` runs. It binds the 3x4 EC8+4 topology, multi-pool/multi-set coverage, per-node metrics endpoints, same-window distributed sampling, process restart and crash-restart fault modes, mixed-version reader/writer/rollback participation, and allocation/flamegraph/RSS/save-frequency profile artifact requirements. Synthetic runs do not need this field and still cannot approve release evidence. |
 | `oracles` | A map with all five scenario names. Each value contains positive integer `objects`, `versions`, `bytes`, and `sha256` of the independently prepared canonical object/version/content manifest. |
 | `expected_healed_objects` | A map with all five scenario names and independently seeded repair counts. Running-heal and MRF-replay require a positive count. |

@@ -75,6 +76,14 @@ object/version/content result. Fix the foreground arrival rate (offered load),
 cache preparation procedure, configuration, and hardware across every leg.
 Do not include credentials in the manifest, adapter output, or saved commands;
 the collector reads `RUSTFS_ACCESS_KEY` and `RUSTFS_SECRET_KEY` from its environment.
+The adapter must echo the measured run's `release_evidence` object in every
+measurement response. A mismatch fails the cell because it means the deployment,
+mixed-version set, crash mode, or profiler contract no longer matches the
+operator-reviewed manifest. This echo is provenance binding only; it does not
+replace the independent correctness oracle, distributed metrics samples, profile
+artifacts, or ABBA comparison thresholds. The summary tool revalidates the same
+manifest contract before it can print a measured PASS result, so hand-built or
+trimmed reports without this provenance fail closed.

 #### Deployment Adapter Contract

@@ -157,6 +157,12 @@ feature-specific oracles, measurements and required topologies exist. Missing
 cases cannot be supplied by synthetic W20 results. W20's bounded JSON and
 file-hash helpers are reused; its ABBA performance contracts remain in
 `docs/operations/scanner-benchmark-runbook.md`.
+Measured ABBA manifests must also carry the runbook's `release_evidence`
+contract. The runner rejects reports that cannot bind the exact 3x4 EC8+4
+topology, multi-pool/multi-set shape, distributed same-window metrics endpoints,
+restart/crash modes, mixed-version reader/writer/rollback participation, and
+allocation/flamegraph/RSS/save-frequency profile artifact plan. Synthetic runs
+and manifests missing that contract remain harness-only evidence.

 ### Recording One Case

@@ -29,6 +29,16 @@ METRICS = (
 )
 REPEATABILITY_LIMIT = Decimal("0.05")
 P2_WORK_MULTIPLE_LIMIT = Decimal("1.2")
+RELEASE_PROFILE_ARTIFACTS = (
+    "allocation-profile",
+    "flamegraph",
+    "rss-samples",
+    "save-frequency",
+)
+RELEASE_FAULT_MODES = (
+    "process-restart",
+    "process-crash-restart",
+)


 def require(condition, message):
@@ -140,6 +150,95 @@ def validate_manifest(manifest):
        number(manifest["expected_healed_objects"].get(scenario), f"{scenario} expected repairs")
        if scenario in ("running-heal", "mrf-replay"):
            require(manifest["expected_healed_objects"][scenario] > 0, f"{scenario} requires repairs")
+    validate_release_evidence_manifest(manifest)
+
+
+def release_evidence_integer(value, name, minimum=1, maximum=1024):
+    require(type(value) is int and minimum <= value <= maximum, f"invalid release_evidence.{name}")
+    return value
+
+
+def release_evidence_string(value, name):
+    require(isinstance(value, str) and value.strip(), f"missing release_evidence.{name}")
+    return value
+
+
+def release_evidence_bool(value, name):
+    require(type(value) is bool, f"invalid release_evidence.{name}")
+    return value
+
+
+def release_evidence_true(value, name):
+    release_evidence_bool(value, name)
+    require(value is True, f"missing release_evidence.{name}")
+
+
+def validate_release_evidence_manifest(manifest):
+    if manifest["evidence"] != "measured":
+        return
+
+    evidence = manifest.get("release_evidence")
+    require(isinstance(evidence, dict), "missing release_evidence for measured ABBA")
+
+    topology = evidence.get("topology")
+    require(isinstance(topology, dict), "missing release_evidence.topology")
+    nodes = release_evidence_integer(topology.get("nodes"), "topology.nodes", 3, 3)
+    drives = release_evidence_integer(topology.get("drives_per_node"), "topology.drives_per_node", 4, 4)
+    data = release_evidence_integer(topology.get("erasure_data_blocks"), "topology.erasure_data_blocks", 8, 8)
+    parity = release_evidence_integer(topology.get("erasure_parity_blocks"), "topology.erasure_parity_blocks", 4, 4)
+    require(data + parity == nodes * drives, "release_evidence.topology must be 3x4 EC8+4")
+    release_evidence_integer(topology.get("pools"), "topology.pools", 2)
+    release_evidence_integer(topology.get("sets_total"), "topology.sets_total", 2)
+
+    distributed = evidence.get("distributed")
+    require(isinstance(distributed, dict), "missing release_evidence.distributed")
+    endpoints = distributed.get("metrics_endpoints")
+    require(isinstance(endpoints, list) and len(endpoints) >= nodes, "missing release_evidence.distributed.metrics_endpoints")
+    require(
+        all(isinstance(endpoint, str) and endpoint.strip() for endpoint in endpoints)
+        and len(set(endpoints)) == len(endpoints),
+        "invalid release_evidence.distributed.metrics_endpoints",
+    )
+    release_evidence_string(distributed.get("failure_domain"), "distributed.failure_domain")
+    release_evidence_true(distributed.get("same_window_sampling"), "distributed.same_window_sampling")
+
+    crash = evidence.get("crash_restart")
+    require(isinstance(crash, dict), "missing release_evidence.crash_restart")
+    fault_modes = crash.get("fault_modes")
+    require(
+        isinstance(fault_modes, list)
+        and all(mode in fault_modes for mode in RELEASE_FAULT_MODES)
+        and all(isinstance(mode, str) and mode.strip() for mode in fault_modes),
+        "missing release_evidence.crash_restart.fault_modes",
+    )
+    release_evidence_true(crash.get("unclean_shutdown_marker"), "crash_restart.unclean_shutdown_marker")
+
+    mixed = evidence.get("mixed_version")
+    require(isinstance(mixed, dict), "missing release_evidence.mixed_version")
+    revisions = mixed.get("participating_revisions")
+    require(
+        isinstance(revisions, list)
+        and len(set(revisions)) >= 2
+        and all(isinstance(revision, str) and len(revision) == 40 and all(c in "0123456789abcdef" for c in revision)
+                for revision in revisions),
+        "invalid release_evidence.mixed_version.participating_revisions",
+    )
+    for revision in (manifest["baseline"]["revision"], manifest["candidate"]["revision"]):
+        require(revision in revisions, "release_evidence.mixed_version omits tested build revision")
+    for key in ("reader", "writer", "rollback_payload"):
+        require(mixed.get(key) is True, f"missing release_evidence.mixed_version.{key}")
+
+    profile = evidence.get("profile")
+    require(isinstance(profile, dict), "missing release_evidence.profile")
+    artifacts = profile.get("required_artifacts")
+    require(
+        isinstance(artifacts, list)
+        and all(item in artifacts for item in RELEASE_PROFILE_ARTIFACTS)
+        and all(isinstance(item, str) and item.strip() for item in artifacts),
+        "missing release_evidence.profile.required_artifacts",
+    )
+    for key in ("collector_config_sha256", "profiler_config_sha256"):
+        require(sha(profile.get(key)), f"invalid release_evidence.profile.{key}")


 class OwnedCommand:
@@ -254,6 +353,8 @@ def validate_result(result, request, expected):
    require(result.get("build") == request["build"], "deployed build provenance mismatch")
    require(result.get("data_dir") == request["data_dir"], "adapter data isolation mismatch")
    require(result.get("background") == request["background"], "background mode mismatch")
+    if request["evidence"] == "measured":
+        require(result.get("release_evidence") == request["release_evidence"], "release evidence provenance mismatch")
    require(type(result.get("sample_count")) is int and 1 <= result["sample_count"] <= 3600,
            "sample_count must be 1..3600")
    number(result.get("elapsed_seconds"), "elapsed_seconds", request["duration_seconds"])
@@ -445,6 +546,9 @@ def collect_live(prepared, request, request_path, adapter):
    connection = prepared["collector"]
    require(set(connection) == {"alias", "endpoint", "metrics_endpoints"}, "invalid collector connection")
    require(all(isinstance(value, str) and value for value in connection.values()), "missing collector endpoint")
+    expected_metrics_endpoints = None
+    if request.get("evidence") == "measured":
+        expected_metrics_endpoints = request["release_evidence"]["distributed"]["metrics_endpoints"]
    output = request_path.parent / "telemetry"
    args = ["bash", str(collector), "--alias", connection["alias"], "--endpoint", connection["endpoint"],
            "--metrics-endpoints", connection["metrics_endpoints"], "--deployment", "distributed",
@@ -470,6 +574,9 @@ def collect_live(prepared, request, request_path, adapter):
                require(isinstance(status.get("healOperations"), dict) and status["healOperations"], "invalid heal status response")
            metrics = list((output / "metrics").glob("admin-metrics.*.ndjson"))
            endpoints = [endpoint for endpoint in connection["metrics_endpoints"].split(",") if endpoint]
+            if expected_metrics_endpoints is not None:
+                require(endpoints == expected_metrics_endpoints,
+                        "collector metrics endpoints do not match release evidence")
            require(metrics and len(metrics) == len(endpoints) * len(samples), "missing distributed metrics samples")
            for sample in metrics:
                # The collector requests n=1, so each file contains one final JSON record.
@@ -518,6 +625,8 @@ def run(manifest, adapter, output, data_root):
                                   "duration_seconds": manifest["duration_seconds"], "data_dir": str(data_dir),
                                   "expected_healed_objects": manifest["expected_healed_objects"][scenario],
                                   "expected_oracle": manifest["oracles"][scenario]}
+                        if manifest["evidence"] == "measured":
+                            request["release_evidence"] = manifest["release_evidence"]
                        require(digest(Path(request["build"]["binary"])) == request["build"]["sha256"], "binary changed during run")
                        require(digest(adapter) == manifest["adapter_sha256"], "adapter changed during run")
                        require(shutil.disk_usage(data_root).free >= manifest["min_free_bytes"], "insufficient free disk space")
@@ -12,6 +12,8 @@ from pathlib import Path
 import sys
 from typing import Any

+from scanner_abba import validate_release_evidence_manifest
+
 MAX_JSON_BYTES = 1024 * 1024
 CACHE_COST_PREFIX = "CACHE_COST "
 PASS_STATES = {"pass"}
@@ -168,6 +170,7 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
    measured = report.get("evidence") == "measured"
    passed = report_state in PASS_STATES and performance_state in PASS_STATES and measured
    if passed:
+        validate_release_evidence_manifest({**manifest, "evidence": "measured"})
        for index, comparison in enumerate(comparisons):
            require_measured_comparison_evidence(comparison, index)
    gate_state = "pass" if passed else "fail"
@@ -217,6 +220,7 @@ def summarize_abba(abba_dir: Path) -> dict[str, Any]:
            "durability": fixed.get("durability"),
            "topology": fixed.get("topology"),
            "offered_load_ops": fixed.get("offered_load_ops"),
+            "release_evidence": manifest.get("release_evidence"),
        },
    }

@@ -163,6 +163,42 @@ class ScannerAbbaTest(unittest.TestCase):
        build = {"binary": str(self.binary), "sha256": harness.digest(self.binary), "revision": "a" * 40}
        self.manifest.update(baseline=build.copy(), candidate=build.copy())

+    def measured_manifest(self):
+        manifest = copy.deepcopy(self.manifest)
+        manifest.update(evidence="measured", duration_seconds=900)
+        manifest["candidate"]["revision"] = "b" * 40
+        manifest["release_evidence"] = {
+            "topology": {
+                "nodes": 3,
+                "drives_per_node": 4,
+                "pools": 2,
+                "sets_total": 2,
+                "erasure_data_blocks": 8,
+                "erasure_parity_blocks": 4,
+            },
+            "distributed": {
+                "metrics_endpoints": ["https://node-1:9000", "https://node-2:9000", "https://node-3:9000"],
+                "failure_domain": "three-node-localhost-lab",
+                "same_window_sampling": True,
+            },
+            "crash_restart": {
+                "fault_modes": ["process-restart", "process-crash-restart"],
+                "unclean_shutdown_marker": True,
+            },
+            "mixed_version": {
+                "participating_revisions": ["a" * 40, "b" * 40],
+                "reader": True,
+                "writer": True,
+                "rollback_payload": True,
+            },
+            "profile": {
+                "required_artifacts": ["allocation-profile", "flamegraph", "rss-samples", "save-frequency"],
+                "collector_config_sha256": "4" * 64,
+                "profiler_config_sha256": "5" * 64,
+            },
+        }
+        return manifest
+
    def run_harness(self, fault=""):
        with patch.dict(os.environ, {"SCANNER_ABBA_TEST_FAULT": fault}), contextlib.redirect_stdout(io.StringIO()):
            return harness.run(copy.deepcopy(self.manifest), self.adapter, self.root / "out", self.root / "data")
@@ -440,6 +476,34 @@ class ScannerAbbaTest(unittest.TestCase):

                process.finish.assert_called_once_with(terminate=True)

+    def test_live_collector_binds_release_evidence_metrics_endpoints(self):
+        telemetry = self.root / "telemetry"
+        for name in ("status", "heal", "metrics"):
+            (telemetry / name).mkdir(parents=True)
+        (telemetry / "scanner-summary.csv").write_text("timestamp\n")
+        for index in range(16):
+            harness.write_json(telemetry / f"status/scanner-status.{index}.json", {"metrics": {"objects": 10}})
+            for node in ("node-a", "node-b"):
+                harness.write_json(telemetry / f"heal/background-heal-status.{node}.{index}.json",
+                                   {"healOperations": {"queueLength": 0}})
+                harness.write_json(telemetry / f"metrics/admin-metrics.{node}.{index}.ndjson",
+                                   {"errors": [], "final": True,
+                                    "by_host": {f"{node}:9000": {"scanner": {"objects": 10}}}})
+        prepared = {"collector": {"alias": "test", "endpoint": "http://node-a:9000",
+                                  "metrics_endpoints": "http://node-a:9000,http://node-b:9000"}}
+        request = {
+            "duration_seconds": 900,
+            "evidence": "measured",
+            "release_evidence": self.measured_manifest()["release_evidence"],
+        }
+        process = Mock(pid=123, wait=Mock(return_value=0))
+        with patch.object(harness, "OwnedCommand", return_value=process), \
+                patch.object(harness, "invoke", return_value={"sample_count": 10}), \
+                patch.object(harness.time, "monotonic", side_effect=(0, 900)):
+            with self.assertRaisesRegex(ValueError, "collector metrics endpoints"):
+                harness.collect_live(prepared, request, self.root / "request.json", self.adapter)
+        process.finish.assert_called_once_with(terminate=True)
+
    def test_unstable_p1_work_control_is_inconclusive(self):
        with patch.object(harness, "SCENARIOS", ("cold-hot",)):
            self.assertEqual(self.run_harness("unstable-p1-control"), 3)
@@ -463,6 +527,80 @@ class ScannerAbbaTest(unittest.TestCase):
        with self.assertRaisesRegex(ValueError, "rounds"):
            harness.validate_manifest(self.manifest)

+    def test_measured_manifest_requires_release_evidence_contract(self):
+        harness.validate_manifest(self.measured_manifest())
+        faults = {
+            "missing root": lambda manifest: manifest.pop("release_evidence"),
+            "single-set": lambda manifest: manifest["release_evidence"]["topology"].update(sets_total=1),
+            "wrong geometry": lambda manifest: manifest["release_evidence"]["topology"].update(nodes=4),
+            "duplicate endpoint": lambda manifest: manifest["release_evidence"]["distributed"].update(
+                metrics_endpoints=["https://node-1:9000", "https://node-1:9000", "https://node-3:9000"],
+            ),
+            "split sampling": lambda manifest: manifest["release_evidence"]["distributed"].update(
+                same_window_sampling=False,
+            ),
+            "missing crash": lambda manifest: manifest["release_evidence"]["crash_restart"].update(
+                fault_modes=["process-restart"],
+            ),
+            "clean crash marker": lambda manifest: manifest["release_evidence"]["crash_restart"].update(
+                unclean_shutdown_marker=False,
+            ),
+            "mixed version false": lambda manifest: manifest["release_evidence"]["mixed_version"].update(writer=False),
+            "missing candidate": lambda manifest: manifest["release_evidence"]["mixed_version"].update(
+                participating_revisions=["a" * 40, "c" * 40],
+            ),
+            "missing profile": lambda manifest: manifest["release_evidence"]["profile"].update(
+                required_artifacts=["allocation-profile", "flamegraph", "rss-samples"],
+            ),
+            "bad profile hash": lambda manifest: manifest["release_evidence"]["profile"].update(
+                profiler_config_sha256="not-a-sha",
+            ),
+        }
+        for name, mutate in faults.items():
+            with self.subTest(fault=name):
+                manifest = self.measured_manifest()
+                mutate(manifest)
+                with self.assertRaisesRegex(ValueError, "release_evidence"):
+                    harness.validate_manifest(manifest)
+
+    def test_measured_result_must_echo_release_evidence(self):
+        manifest = self.measured_manifest()
+        request = {
+            "schema": 1,
+            "scenario": "cold-hot",
+            "comparison": "build",
+            "round": 1,
+            "leg": "B1",
+            "background": "on",
+            "build": manifest["candidate"],
+            "evidence": manifest["evidence"],
+            "fixed": manifest["fixed"],
+            "release_evidence": manifest["release_evidence"],
+            "duration_seconds": manifest["duration_seconds"],
+            "data_dir": str(self.root / "data"),
+            "expected_healed_objects": manifest["expected_healed_objects"]["cold-hot"],
+        }
+        metrics = dict.fromkeys(harness.METRICS, 10)
+        metrics.update(p99_ms=10, throughput_ops=100, errors=0, requests=100,
+                       walk_objects=100, cold_walk_objects=20, healed_objects=10)
+        result = {
+            "evidence": request["evidence"],
+            "fixed": request["fixed"],
+            "build": request["build"],
+            "data_dir": request["data_dir"],
+            "background": request["background"],
+            "release_evidence": request["release_evidence"],
+            "sample_count": 10,
+            "elapsed_seconds": request["duration_seconds"],
+            "metrics": metrics,
+            "oracle": manifest["oracles"]["cold-hot"],
+        }
+        harness.validate_result(result, request, manifest["oracles"]["cold-hot"])
+        result["release_evidence"] = copy.deepcopy(result["release_evidence"])
+        result["release_evidence"]["profile"]["required_artifacts"].remove("flamegraph")
+        with self.assertRaisesRegex(ValueError, "release evidence provenance mismatch"):
+            harness.validate_result(result, request, manifest["oracles"]["cold-hot"])
+
    def test_existing_data_preserved(self):
        (self.root / "data").mkdir()
        marker = self.root / "data/keep"
@@ -45,6 +45,36 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
            "candidate": {"revision": "b" * 40, "sha256": "4" * 64},
            "adapter_sha256": "5" * 64,
            "collector_sha256": "6" * 64,
+            "release_evidence": {
+                "topology": {
+                    "nodes": 3,
+                    "drives_per_node": 4,
+                    "pools": 2,
+                    "sets_total": 2,
+                    "erasure_data_blocks": 8,
+                    "erasure_parity_blocks": 4,
+                },
+                "distributed": {
+                    "metrics_endpoints": ["https://node-1:9000", "https://node-2:9000", "https://node-3:9000"],
+                    "failure_domain": "three-node-localhost-lab",
+                    "same_window_sampling": True,
+                },
+                "crash_restart": {
+                    "fault_modes": ["process-restart", "process-crash-restart"],
+                    "unclean_shutdown_marker": True,
+                },
+                "mixed_version": {
+                    "participating_revisions": ["a" * 40, "b" * 40],
+                    "reader": True,
+                    "writer": True,
+                    "rollback_payload": True,
+                },
+                "profile": {
+                    "required_artifacts": ["allocation-profile", "flamegraph", "rss-samples", "save-frequency"],
+                    "collector_config_sha256": "7" * 64,
+                    "profiler_config_sha256": "8" * 64,
+                },
+            },
        }
        self.comparison = {
            "scenario": "cold-hot",
@@ -181,6 +211,19 @@ class ScannerHealPerfSummaryTest(unittest.TestCase):
                with self.assertRaisesRegex(ValueError, "W10/W11|performance evidence|length mismatch|above maximum"):
                    summary.build_summary(args)

+    def test_passing_measured_report_requires_release_evidence_manifest(self):
+        del self.manifest["release_evidence"]
+        self.write_inputs()
+        args = type("Args", (), {
+            "abba_dir": self.abba,
+            "cache_cost_log": None,
+            "require_cache_cost": False,
+            "json_out": None,
+            "markdown_out": None,
+        })
+        with self.assertRaisesRegex(ValueError, "release_evidence"):
+            summary.build_summary(args)
+
    def test_requires_cache_profile_when_requested(self):
        args = type("Args", (), {
            "abba_dir": self.abba,