"""Check the paused persistence backlog against one-way local ceilings."""
from __future__ import annotations
import argparse
import json
import os
import re
import subprocess
import sys
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parent.parent
MEASURE_SCRIPT = ROOT / "scripts" / "measure-persistence-backlog.py"
BUDGET_PATH = ROOT / "scripts" / "persistence-backlog-budget.json"
BASELINE_RECEIPT_PATH = ROOT / "scripts" / "persistence-backlog-baseline-receipt.json"
BASELINE_RECEIPT_REFERENCE = "scripts/persistence-backlog-baseline-receipt.json"
RECEIPT_KIND = "codewhale.persistence_backlog_receipt"
BUDGET_KIND = "codewhale.persistence_backlog_budget"
SCHEMA_VERSION = 2
FIXTURE = {
"fixture_id": "paused-production-channel-session-snapshot-v1",
"request_variant": "session_snapshot",
"payload_estimator": "retained-saved-session-json-bytes-v1",
"paused_consumer": True,
"requests_attempted": 128,
"content_bytes_per_request": 64 * 1024,
"single_session_id": True,
"expected_applied_version": 127,
}
REQUIRED_RECEIPT_FIELDS = (
"document_kind",
"schema_version",
"source_sha",
"source_dirty",
"rustc_version",
"cargo_version",
"build_profile",
"sample_count",
"fixture_id",
"platform",
"request_variant",
"payload_estimator",
"paused_consumer",
"requests_attempted",
"content_bytes_per_request",
"single_session_id",
"expected_applied_version",
"accepted_requests",
"retained_queued_requests",
"estimated_retained_payload_bytes",
"applied_version",
"final_version_applied",
"enqueue_elapsed_ns",
"rss_supported",
"rss_before_bytes",
"rss_during_bytes",
"rss_after_bytes",
"rss_during_delta_bytes",
"rss_after_delta_bytes",
"limitations",
)
CEILING_FIELDS = (
"retained_queued_requests",
"estimated_retained_payload_bytes",
"enqueue_elapsed_ns",
"rss_during_delta_bytes",
"rss_after_delta_bytes",
)
RSS_SAMPLE_FIELDS = ("rss_before_bytes", "rss_during_bytes", "rss_after_bytes")
RSS_DELTA_FIELDS = ("rss_during_delta_bytes", "rss_after_delta_bytes")
SUPPORTED_PLATFORMS = {"linux", "macos", "windows"}
SOURCE_SHA_PATTERN = re.compile(r"[0-9a-f]{40}")
class PersistenceBacklogError(ValueError):
"""A receipt or budget broke the measurement contract."""
def load_json(path: Path, label: str) -> dict[str, Any]:
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
raise PersistenceBacklogError(f"invalid {label} {path}: {error}") from error
if not isinstance(value, dict):
raise PersistenceBacklogError(f"{label} must be a JSON object")
return value
def non_negative_integer(value: Any, field: str) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
raise PersistenceBacklogError(f"{field} must be a non-negative integer")
return value
def validate_frozen_field(field: str, value: Any, expected: Any) -> None:
if type(value) is not type(expected) or value != expected:
raise PersistenceBacklogError(
f"receipt {field} must remain {expected!r}, got {value!r}"
)
def current_source_identity() -> dict[str, Any]:
def run(command: list[str]) -> str:
result = subprocess.run(
command,
cwd=ROOT,
text=True,
capture_output=True,
check=False,
)
if result.returncode != 0:
raise PersistenceBacklogError(
f"source provenance command failed: {' '.join(command)}"
)
return result.stdout.strip()
return {
"source_sha": run(["git", "rev-parse", "HEAD"]),
"source_dirty": bool(
run(["git", "status", "--porcelain", "--untracked-files=normal"])
),
"rustc_version": run(["rustc", "--version"]),
"cargo_version": run(["cargo", "--version"]),
"build_profile": "test",
"sample_count": 1,
}
def validate_receipt(
receipt: dict[str, Any],
*,
expected_source: dict[str, Any] | None = None,
require_clean_source: bool = False,
) -> None:
missing = [field for field in REQUIRED_RECEIPT_FIELDS if field not in receipt]
if missing:
raise PersistenceBacklogError(
"receipt missing required field(s): " + ", ".join(missing)
)
if receipt["document_kind"] != RECEIPT_KIND:
raise PersistenceBacklogError(f"receipt document_kind must be {RECEIPT_KIND}")
if receipt["schema_version"] != SCHEMA_VERSION:
raise PersistenceBacklogError("receipt schema_version changed")
for field, expected in FIXTURE.items():
validate_frozen_field(field, receipt[field], expected)
if not isinstance(receipt["source_sha"], str) or not SOURCE_SHA_PATTERN.fullmatch(
receipt["source_sha"]
):
raise PersistenceBacklogError("receipt source_sha must be an exact lowercase Git SHA")
if type(receipt["source_dirty"]) is not bool:
raise PersistenceBacklogError("receipt source_dirty must be boolean")
for field, prefix in (("rustc_version", "rustc "), ("cargo_version", "cargo ")):
if not isinstance(receipt[field], str) or not receipt[field].startswith(prefix):
raise PersistenceBacklogError(f"receipt {field} must be a version string")
validate_frozen_field("build_profile", receipt["build_profile"], "test")
validate_frozen_field("sample_count", receipt["sample_count"], 1)
if expected_source is not None:
for field in (
"source_sha",
"source_dirty",
"rustc_version",
"cargo_version",
"build_profile",
"sample_count",
):
if receipt[field] != expected_source[field]:
raise PersistenceBacklogError(
f"receipt {field} does not match the checked source"
)
if require_clean_source and receipt["source_dirty"]:
raise PersistenceBacklogError("persistence measurement source tree is dirty")
platform = receipt["platform"]
if not isinstance(platform, str) or platform not in SUPPORTED_PLATFORMS:
raise PersistenceBacklogError("receipt platform is unsupported")
attempted = non_negative_integer(receipt["requests_attempted"], "requests_attempted")
accepted = non_negative_integer(receipt["accepted_requests"], "accepted_requests")
if accepted != attempted:
raise PersistenceBacklogError(
"accepted_requests must equal requests_attempted; sender rejection is not backlog improvement"
)
retained = non_negative_integer(
receipt["retained_queued_requests"], "retained_queued_requests"
)
if retained > accepted:
raise PersistenceBacklogError("retained_queued_requests exceeds accepted_requests")
for field in ("estimated_retained_payload_bytes", "enqueue_elapsed_ns"):
non_negative_integer(receipt[field], field)
if retained == 0 or receipt["estimated_retained_payload_bytes"] == 0:
raise PersistenceBacklogError(
"the paused channel must retain the newest request and its payload"
)
minimum_payload_bytes = retained * FIXTURE["content_bytes_per_request"]
if receipt["estimated_retained_payload_bytes"] < minimum_payload_bytes:
raise PersistenceBacklogError(
"estimated_retained_payload_bytes is smaller than the frozen retained content"
)
applied = non_negative_integer(
receipt["applied_version"], "applied_version"
)
if applied != FIXTURE["expected_applied_version"]:
raise PersistenceBacklogError("applied_version is not the final sent version")
if receipt["final_version_applied"] is not True:
raise PersistenceBacklogError("final_version_applied must be true")
limitations = receipt["limitations"]
if not isinstance(limitations, list) or not limitations or not all(
isinstance(item, str) and item for item in limitations
):
raise PersistenceBacklogError("limitations must be a non-empty string array")
if not isinstance(receipt["rss_supported"], bool):
raise PersistenceBacklogError("rss_supported must be boolean")
if receipt["rss_supported"] != (platform == "macos"):
raise PersistenceBacklogError(
"rss_supported must be true exactly on the macOS measurement lane"
)
rss_fields = RSS_SAMPLE_FIELDS + RSS_DELTA_FIELDS
if receipt["rss_supported"]:
for field in rss_fields:
non_negative_integer(receipt[field], field)
before = receipt["rss_before_bytes"]
if receipt["rss_during_delta_bytes"] != max(
0, receipt["rss_during_bytes"] - before
):
raise PersistenceBacklogError("rss_during_delta_bytes is inconsistent")
if receipt["rss_after_delta_bytes"] != max(
0, receipt["rss_after_bytes"] - before
):
raise PersistenceBacklogError("rss_after_delta_bytes is inconsistent")
elif any(receipt[field] is not None for field in rss_fields):
raise PersistenceBacklogError("unsupported RSS fields must be null")
def validate_budget(budget: dict[str, Any]) -> None:
if budget.get("document_kind") != BUDGET_KIND:
raise PersistenceBacklogError(f"budget document_kind must be {BUDGET_KIND}")
if budget.get("schema_version") != SCHEMA_VERSION:
raise PersistenceBacklogError("budget schema_version changed")
fixture = budget.get("fixture")
if not isinstance(fixture, dict) or set(fixture) != set(FIXTURE):
raise PersistenceBacklogError("budget fixture no longer matches the frozen workload")
for field, expected in FIXTURE.items():
if type(fixture[field]) is not type(expected) or fixture[field] != expected:
raise PersistenceBacklogError(
f"budget fixture.{field} must remain {expected!r}"
)
if budget.get("baseline_receipt") != BASELINE_RECEIPT_REFERENCE:
raise PersistenceBacklogError("budget baseline_receipt path changed")
ceilings = budget.get("ceilings")
baseline = budget.get("baseline_observation")
if not isinstance(ceilings, dict) or not isinstance(baseline, dict):
raise PersistenceBacklogError("budget needs ceilings and baseline_observation objects")
for field in CEILING_FIELDS:
ceiling = non_negative_integer(ceilings.get(field), f"ceilings.{field}")
observed = non_negative_integer(
baseline.get(field), f"baseline_observation.{field}"
)
if observed > ceiling:
raise PersistenceBacklogError(
f"baseline_observation.{field} exceeds its ceiling"
)
baseline_accepted = non_negative_integer(
baseline.get("accepted_requests"), "baseline_observation.accepted_requests"
)
if baseline_accepted != FIXTURE["requests_attempted"]:
raise PersistenceBacklogError(
"baseline_observation.accepted_requests must equal requests_attempted"
)
baseline_applied = non_negative_integer(
baseline.get("applied_version"), "baseline_observation.applied_version"
)
if baseline_applied != FIXTURE["expected_applied_version"]:
raise PersistenceBacklogError(
"baseline_observation.applied_version must be the final sent version"
)
baseline_retained = baseline["retained_queued_requests"]
baseline_payload = baseline["estimated_retained_payload_bytes"]
if baseline_retained == 0 or baseline_payload == 0:
raise PersistenceBacklogError(
"baseline_observation must retain the final request and payload"
)
if baseline_retained > baseline_accepted:
raise PersistenceBacklogError(
"baseline_observation.retained_queued_requests exceeds accepted_requests"
)
if baseline_payload < baseline_retained * FIXTURE["content_bytes_per_request"]:
raise PersistenceBacklogError(
"baseline_observation payload is smaller than frozen retained content"
)
provenance = baseline.get("provenance")
if not isinstance(provenance, dict):
raise PersistenceBacklogError("baseline_observation needs provenance")
if provenance.get("platform") != "macos":
raise PersistenceBacklogError("baseline provenance platform must be macos")
if not isinstance(provenance.get("source_sha"), str) or not SOURCE_SHA_PATTERN.fullmatch(
provenance["source_sha"]
):
raise PersistenceBacklogError("baseline provenance needs an exact source SHA")
if provenance.get("source_dirty") is not False:
raise PersistenceBacklogError("baseline provenance must identify a clean source tree")
for field, prefix in (("rustc_version", "rustc "), ("cargo_version", "cargo ")):
if not isinstance(provenance.get(field), str) or not provenance[field].startswith(prefix):
raise PersistenceBacklogError(f"baseline provenance needs {field}")
if provenance.get("build_profile") != "test" or not (
type(provenance.get("sample_count")) is int
and provenance["sample_count"] == 1
):
raise PersistenceBacklogError("baseline provenance build profile/sample count changed")
def validate_baseline_receipt(
budget: dict[str, Any], baseline_receipt: dict[str, Any]
) -> None:
validate_receipt(baseline_receipt, require_clean_source=True)
baseline = budget["baseline_observation"]
for field in ("accepted_requests", "applied_version", *CEILING_FIELDS):
if baseline_receipt[field] != baseline[field]:
raise PersistenceBacklogError(
f"baseline receipt {field} does not match baseline_observation"
)
provenance = baseline["provenance"]
for field in (
"platform",
"source_sha",
"source_dirty",
"rustc_version",
"cargo_version",
"build_profile",
"sample_count",
):
if baseline_receipt[field] != provenance[field]:
raise PersistenceBacklogError(
f"baseline receipt {field} does not match baseline provenance"
)
def compare(
receipt: dict[str, Any],
budget: dict[str, Any],
*,
expected_source: dict[str, Any] | None = None,
require_clean_source: bool = False,
) -> tuple[list[tuple[str, int, int]], list[tuple[str, int, int]]]:
validate_receipt(
receipt,
expected_source=expected_source,
require_clean_source=require_clean_source,
)
validate_budget(budget)
increases: list[tuple[str, int, int]] = []
decreases: list[tuple[str, int, int]] = []
for field in CEILING_FIELDS:
if field in RSS_DELTA_FIELDS and not receipt["rss_supported"]:
continue
current = receipt[field]
ceiling = budget["ceilings"][field]
if current > ceiling:
increases.append((field, current, ceiling))
elif current < ceiling:
decreases.append((field, current, ceiling))
return increases, decreases
def measure() -> dict[str, Any]:
env = os.environ.copy()
env["CARGO_NET_OFFLINE"] = "true"
result = subprocess.run(
[sys.executable, str(MEASURE_SCRIPT)],
cwd=ROOT,
env=env,
text=True,
capture_output=True,
check=False,
)
sys.stderr.write(result.stderr)
if result.returncode != 0:
sys.stdout.write(result.stdout)
raise PersistenceBacklogError("measurement command failed")
try:
receipt = json.loads(result.stdout)
except json.JSONDecodeError as error:
raise PersistenceBacklogError(f"measurement emitted invalid JSON: {error}") from error
if not isinstance(receipt, dict):
raise PersistenceBacklogError("measurement receipt must be an object")
return receipt
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--receipt", type=Path, help="check an existing receipt")
parser.add_argument("--budget", type=Path, default=BUDGET_PATH)
args = parser.parse_args()
try:
expected_source = current_source_identity()
receipt = load_json(args.receipt, "receipt") if args.receipt else measure()
budget = load_json(args.budget, "budget")
baseline_receipt = load_json(BASELINE_RECEIPT_PATH, "baseline receipt")
validate_baseline_receipt(budget, baseline_receipt)
increases, decreases = compare(
receipt,
budget,
expected_source=expected_source,
require_clean_source=True,
)
except PersistenceBacklogError as error:
print(f"[persistence-backlog-budget] ERROR: {error}", file=sys.stderr)
return 2
if increases:
for field, current, ceiling in increases:
print(
f"[persistence-backlog-budget] FAIL: {field}={current} exceeds {ceiling}",
file=sys.stderr,
)
return 1
print("[persistence-backlog-budget] PASS: one-way ceilings respected")
for field, current, ceiling in decreases:
print(f" can tighten {field}: {current} < {ceiling}")
return 0
if __name__ == "__main__":
raise SystemExit(main())