| 1 | #!/usr/bin/env python3 |
| 2 | """Check the paused persistence backlog against one-way local ceilings.""" |
| 3 | |
| 4 | from __future__ import annotations |
| 5 | |
| 6 | import argparse |
| 7 | import json |
| 8 | import os |
| 9 | import re |
| 10 | import subprocess |
| 11 | import sys |
| 12 | from pathlib import Path |
| 13 | from typing import Any |
| 14 | |
| 15 | |
| 16 | ROOT = Path(__file__).resolve().parent.parent |
| 17 | MEASURE_SCRIPT = ROOT / "scripts" / "measure-persistence-backlog.py" |
| 18 | BUDGET_PATH = ROOT / "scripts" / "persistence-backlog-budget.json" |
| 19 | BASELINE_RECEIPT_PATH = ROOT / "scripts" / "persistence-backlog-baseline-receipt.json" |
| 20 | BASELINE_RECEIPT_REFERENCE = "scripts/persistence-backlog-baseline-receipt.json" |
| 21 | RECEIPT_KIND = "codewhale.persistence_backlog_receipt" |
| 22 | BUDGET_KIND = "codewhale.persistence_backlog_budget" |
| 23 | SCHEMA_VERSION = 2 |
| 24 | |
| 25 | FIXTURE = { |
| 26 | "fixture_id": "paused-production-channel-session-snapshot-v1", |
| 27 | "request_variant": "session_snapshot", |
| 28 | "payload_estimator": "retained-saved-session-json-bytes-v1", |
| 29 | "paused_consumer": True, |
| 30 | "requests_attempted": 128, |
| 31 | "content_bytes_per_request": 64 * 1024, |
| 32 | "single_session_id": True, |
| 33 | "expected_applied_version": 127, |
| 34 | } |
| 35 | |
| 36 | REQUIRED_RECEIPT_FIELDS = ( |
| 37 | "document_kind", |
| 38 | "schema_version", |
| 39 | "source_sha", |
| 40 | "source_dirty", |
| 41 | "rustc_version", |
| 42 | "cargo_version", |
| 43 | "build_profile", |
| 44 | "sample_count", |
| 45 | "fixture_id", |
| 46 | "platform", |
| 47 | "request_variant", |
| 48 | "payload_estimator", |
| 49 | "paused_consumer", |
| 50 | "requests_attempted", |
| 51 | "content_bytes_per_request", |
| 52 | "single_session_id", |
| 53 | "expected_applied_version", |
| 54 | "accepted_requests", |
| 55 | "retained_queued_requests", |
| 56 | "estimated_retained_payload_bytes", |
| 57 | "applied_version", |
| 58 | "final_version_applied", |
| 59 | "enqueue_elapsed_ns", |
| 60 | "rss_supported", |
| 61 | "rss_before_bytes", |
| 62 | "rss_during_bytes", |
| 63 | "rss_after_bytes", |
| 64 | "rss_during_delta_bytes", |
| 65 | "rss_after_delta_bytes", |
| 66 | "limitations", |
| 67 | ) |
| 68 | |
| 69 | CEILING_FIELDS = ( |
| 70 | "retained_queued_requests", |
| 71 | "estimated_retained_payload_bytes", |
| 72 | "enqueue_elapsed_ns", |
| 73 | "rss_during_delta_bytes", |
| 74 | "rss_after_delta_bytes", |
| 75 | ) |
| 76 | RSS_SAMPLE_FIELDS = ("rss_before_bytes", "rss_during_bytes", "rss_after_bytes") |
| 77 | RSS_DELTA_FIELDS = ("rss_during_delta_bytes", "rss_after_delta_bytes") |
| 78 | SUPPORTED_PLATFORMS = {"linux", "macos", "windows"} |
| 79 | SOURCE_SHA_PATTERN = re.compile(r"[0-9a-f]{40}") |
| 80 | |
| 81 | |
| 82 | class PersistenceBacklogError(ValueError): |
| 83 | """A receipt or budget broke the measurement contract.""" |
| 84 | |
| 85 | |
| 86 | def load_json(path: Path, label: str) -> dict[str, Any]: |
| 87 | try: |
| 88 | value = json.loads(path.read_text(encoding="utf-8")) |
| 89 | except (OSError, json.JSONDecodeError) as error: |
| 90 | raise PersistenceBacklogError(f"invalid {label} {path}: {error}") from error |
| 91 | if not isinstance(value, dict): |
| 92 | raise PersistenceBacklogError(f"{label} must be a JSON object") |
| 93 | return value |
| 94 | |
| 95 | |
| 96 | def non_negative_integer(value: Any, field: str) -> int: |
| 97 | if isinstance(value, bool) or not isinstance(value, int) or value < 0: |
| 98 | raise PersistenceBacklogError(f"{field} must be a non-negative integer") |
| 99 | return value |
| 100 | |
| 101 | |
| 102 | def validate_frozen_field(field: str, value: Any, expected: Any) -> None: |
| 103 | if type(value) is not type(expected) or value != expected: |
| 104 | raise PersistenceBacklogError( |
| 105 | f"receipt {field} must remain {expected!r}, got {value!r}" |
| 106 | ) |
| 107 | |
| 108 | |
| 109 | def current_source_identity() -> dict[str, Any]: |
| 110 | def run(command: list[str]) -> str: |
| 111 | result = subprocess.run( |
| 112 | command, |
| 113 | cwd=ROOT, |
| 114 | text=True, |
| 115 | capture_output=True, |
| 116 | check=False, |
| 117 | ) |
| 118 | if result.returncode != 0: |
| 119 | raise PersistenceBacklogError( |
| 120 | f"source provenance command failed: {' '.join(command)}" |
| 121 | ) |
| 122 | return result.stdout.strip() |
| 123 | |
| 124 | return { |
| 125 | "source_sha": run(["git", "rev-parse", "HEAD"]), |
| 126 | "source_dirty": bool( |
| 127 | run(["git", "status", "--porcelain", "--untracked-files=normal"]) |
| 128 | ), |
| 129 | "rustc_version": run(["rustc", "--version"]), |
| 130 | "cargo_version": run(["cargo", "--version"]), |
| 131 | "build_profile": "test", |
| 132 | "sample_count": 1, |
| 133 | } |
| 134 | |
| 135 | |
| 136 | def validate_receipt( |
| 137 | receipt: dict[str, Any], |
| 138 | *, |
| 139 | expected_source: dict[str, Any] | None = None, |
| 140 | require_clean_source: bool = False, |
| 141 | ) -> None: |
| 142 | missing = [field for field in REQUIRED_RECEIPT_FIELDS if field not in receipt] |
| 143 | if missing: |
| 144 | raise PersistenceBacklogError( |
| 145 | "receipt missing required field(s): " + ", ".join(missing) |
| 146 | ) |
| 147 | if receipt["document_kind"] != RECEIPT_KIND: |
| 148 | raise PersistenceBacklogError(f"receipt document_kind must be {RECEIPT_KIND}") |
| 149 | if receipt["schema_version"] != SCHEMA_VERSION: |
| 150 | raise PersistenceBacklogError("receipt schema_version changed") |
| 151 | for field, expected in FIXTURE.items(): |
| 152 | validate_frozen_field(field, receipt[field], expected) |
| 153 | if not isinstance(receipt["source_sha"], str) or not SOURCE_SHA_PATTERN.fullmatch( |
| 154 | receipt["source_sha"] |
| 155 | ): |
| 156 | raise PersistenceBacklogError("receipt source_sha must be an exact lowercase Git SHA") |
| 157 | if type(receipt["source_dirty"]) is not bool: |
| 158 | raise PersistenceBacklogError("receipt source_dirty must be boolean") |
| 159 | for field, prefix in (("rustc_version", "rustc "), ("cargo_version", "cargo ")): |
| 160 | if not isinstance(receipt[field], str) or not receipt[field].startswith(prefix): |
| 161 | raise PersistenceBacklogError(f"receipt {field} must be a version string") |
| 162 | validate_frozen_field("build_profile", receipt["build_profile"], "test") |
| 163 | validate_frozen_field("sample_count", receipt["sample_count"], 1) |
| 164 | if expected_source is not None: |
| 165 | for field in ( |
| 166 | "source_sha", |
| 167 | "source_dirty", |
| 168 | "rustc_version", |
| 169 | "cargo_version", |
| 170 | "build_profile", |
| 171 | "sample_count", |
| 172 | ): |
| 173 | if receipt[field] != expected_source[field]: |
| 174 | raise PersistenceBacklogError( |
| 175 | f"receipt {field} does not match the checked source" |
| 176 | ) |
| 177 | if require_clean_source and receipt["source_dirty"]: |
| 178 | raise PersistenceBacklogError("persistence measurement source tree is dirty") |
| 179 | platform = receipt["platform"] |
| 180 | if not isinstance(platform, str) or platform not in SUPPORTED_PLATFORMS: |
| 181 | raise PersistenceBacklogError("receipt platform is unsupported") |
| 182 | |
| 183 | attempted = non_negative_integer(receipt["requests_attempted"], "requests_attempted") |
| 184 | accepted = non_negative_integer(receipt["accepted_requests"], "accepted_requests") |
| 185 | if accepted != attempted: |
| 186 | raise PersistenceBacklogError( |
| 187 | "accepted_requests must equal requests_attempted; sender rejection is not backlog improvement" |
| 188 | ) |
| 189 | retained = non_negative_integer( |
| 190 | receipt["retained_queued_requests"], "retained_queued_requests" |
| 191 | ) |
| 192 | if retained > accepted: |
| 193 | raise PersistenceBacklogError("retained_queued_requests exceeds accepted_requests") |
| 194 | for field in ("estimated_retained_payload_bytes", "enqueue_elapsed_ns"): |
| 195 | non_negative_integer(receipt[field], field) |
| 196 | if retained == 0 or receipt["estimated_retained_payload_bytes"] == 0: |
| 197 | raise PersistenceBacklogError( |
| 198 | "the paused channel must retain the newest request and its payload" |
| 199 | ) |
| 200 | minimum_payload_bytes = retained * FIXTURE["content_bytes_per_request"] |
| 201 | if receipt["estimated_retained_payload_bytes"] < minimum_payload_bytes: |
| 202 | raise PersistenceBacklogError( |
| 203 | "estimated_retained_payload_bytes is smaller than the frozen retained content" |
| 204 | ) |
| 205 | applied = non_negative_integer( |
| 206 | receipt["applied_version"], "applied_version" |
| 207 | ) |
| 208 | if applied != FIXTURE["expected_applied_version"]: |
| 209 | raise PersistenceBacklogError("applied_version is not the final sent version") |
| 210 | if receipt["final_version_applied"] is not True: |
| 211 | raise PersistenceBacklogError("final_version_applied must be true") |
| 212 | |
| 213 | limitations = receipt["limitations"] |
| 214 | if not isinstance(limitations, list) or not limitations or not all( |
| 215 | isinstance(item, str) and item for item in limitations |
| 216 | ): |
| 217 | raise PersistenceBacklogError("limitations must be a non-empty string array") |
| 218 | |
| 219 | if not isinstance(receipt["rss_supported"], bool): |
| 220 | raise PersistenceBacklogError("rss_supported must be boolean") |
| 221 | if receipt["rss_supported"] != (platform == "macos"): |
| 222 | raise PersistenceBacklogError( |
| 223 | "rss_supported must be true exactly on the macOS measurement lane" |
| 224 | ) |
| 225 | rss_fields = RSS_SAMPLE_FIELDS + RSS_DELTA_FIELDS |
| 226 | if receipt["rss_supported"]: |
| 227 | for field in rss_fields: |
| 228 | non_negative_integer(receipt[field], field) |
| 229 | before = receipt["rss_before_bytes"] |
| 230 | if receipt["rss_during_delta_bytes"] != max( |
| 231 | 0, receipt["rss_during_bytes"] - before |
| 232 | ): |
| 233 | raise PersistenceBacklogError("rss_during_delta_bytes is inconsistent") |
| 234 | if receipt["rss_after_delta_bytes"] != max( |
| 235 | 0, receipt["rss_after_bytes"] - before |
| 236 | ): |
| 237 | raise PersistenceBacklogError("rss_after_delta_bytes is inconsistent") |
| 238 | elif any(receipt[field] is not None for field in rss_fields): |
| 239 | raise PersistenceBacklogError("unsupported RSS fields must be null") |
| 240 | |
| 241 | |
| 242 | def validate_budget(budget: dict[str, Any]) -> None: |
| 243 | if budget.get("document_kind") != BUDGET_KIND: |
| 244 | raise PersistenceBacklogError(f"budget document_kind must be {BUDGET_KIND}") |
| 245 | if budget.get("schema_version") != SCHEMA_VERSION: |
| 246 | raise PersistenceBacklogError("budget schema_version changed") |
| 247 | fixture = budget.get("fixture") |
| 248 | if not isinstance(fixture, dict) or set(fixture) != set(FIXTURE): |
| 249 | raise PersistenceBacklogError("budget fixture no longer matches the frozen workload") |
| 250 | for field, expected in FIXTURE.items(): |
| 251 | if type(fixture[field]) is not type(expected) or fixture[field] != expected: |
| 252 | raise PersistenceBacklogError( |
| 253 | f"budget fixture.{field} must remain {expected!r}" |
| 254 | ) |
| 255 | if budget.get("baseline_receipt") != BASELINE_RECEIPT_REFERENCE: |
| 256 | raise PersistenceBacklogError("budget baseline_receipt path changed") |
| 257 | ceilings = budget.get("ceilings") |
| 258 | baseline = budget.get("baseline_observation") |
| 259 | if not isinstance(ceilings, dict) or not isinstance(baseline, dict): |
| 260 | raise PersistenceBacklogError("budget needs ceilings and baseline_observation objects") |
| 261 | for field in CEILING_FIELDS: |
| 262 | ceiling = non_negative_integer(ceilings.get(field), f"ceilings.{field}") |
| 263 | observed = non_negative_integer( |
| 264 | baseline.get(field), f"baseline_observation.{field}" |
| 265 | ) |
| 266 | if observed > ceiling: |
| 267 | raise PersistenceBacklogError( |
| 268 | f"baseline_observation.{field} exceeds its ceiling" |
| 269 | ) |
| 270 | baseline_accepted = non_negative_integer( |
| 271 | baseline.get("accepted_requests"), "baseline_observation.accepted_requests" |
| 272 | ) |
| 273 | if baseline_accepted != FIXTURE["requests_attempted"]: |
| 274 | raise PersistenceBacklogError( |
| 275 | "baseline_observation.accepted_requests must equal requests_attempted" |
| 276 | ) |
| 277 | baseline_applied = non_negative_integer( |
| 278 | baseline.get("applied_version"), "baseline_observation.applied_version" |
| 279 | ) |
| 280 | if baseline_applied != FIXTURE["expected_applied_version"]: |
| 281 | raise PersistenceBacklogError( |
| 282 | "baseline_observation.applied_version must be the final sent version" |
| 283 | ) |
| 284 | baseline_retained = baseline["retained_queued_requests"] |
| 285 | baseline_payload = baseline["estimated_retained_payload_bytes"] |
| 286 | if baseline_retained == 0 or baseline_payload == 0: |
| 287 | raise PersistenceBacklogError( |
| 288 | "baseline_observation must retain the final request and payload" |
| 289 | ) |
| 290 | if baseline_retained > baseline_accepted: |
| 291 | raise PersistenceBacklogError( |
| 292 | "baseline_observation.retained_queued_requests exceeds accepted_requests" |
| 293 | ) |
| 294 | if baseline_payload < baseline_retained * FIXTURE["content_bytes_per_request"]: |
| 295 | raise PersistenceBacklogError( |
| 296 | "baseline_observation payload is smaller than frozen retained content" |
| 297 | ) |
| 298 | provenance = baseline.get("provenance") |
| 299 | if not isinstance(provenance, dict): |
| 300 | raise PersistenceBacklogError("baseline_observation needs provenance") |
| 301 | if provenance.get("platform") != "macos": |
| 302 | raise PersistenceBacklogError("baseline provenance platform must be macos") |
| 303 | if not isinstance(provenance.get("source_sha"), str) or not SOURCE_SHA_PATTERN.fullmatch( |
| 304 | provenance["source_sha"] |
| 305 | ): |
| 306 | raise PersistenceBacklogError("baseline provenance needs an exact source SHA") |
| 307 | if provenance.get("source_dirty") is not False: |
| 308 | raise PersistenceBacklogError("baseline provenance must identify a clean source tree") |
| 309 | for field, prefix in (("rustc_version", "rustc "), ("cargo_version", "cargo ")): |
| 310 | if not isinstance(provenance.get(field), str) or not provenance[field].startswith(prefix): |
| 311 | raise PersistenceBacklogError(f"baseline provenance needs {field}") |
| 312 | if provenance.get("build_profile") != "test" or not ( |
| 313 | type(provenance.get("sample_count")) is int |
| 314 | and provenance["sample_count"] == 1 |
| 315 | ): |
| 316 | raise PersistenceBacklogError("baseline provenance build profile/sample count changed") |
| 317 | |
| 318 | |
| 319 | def validate_baseline_receipt( |
| 320 | budget: dict[str, Any], baseline_receipt: dict[str, Any] |
| 321 | ) -> None: |
| 322 | validate_receipt(baseline_receipt, require_clean_source=True) |
| 323 | baseline = budget["baseline_observation"] |
| 324 | for field in ("accepted_requests", "applied_version", *CEILING_FIELDS): |
| 325 | if baseline_receipt[field] != baseline[field]: |
| 326 | raise PersistenceBacklogError( |
| 327 | f"baseline receipt {field} does not match baseline_observation" |
| 328 | ) |
| 329 | provenance = baseline["provenance"] |
| 330 | for field in ( |
| 331 | "platform", |
| 332 | "source_sha", |
| 333 | "source_dirty", |
| 334 | "rustc_version", |
| 335 | "cargo_version", |
| 336 | "build_profile", |
| 337 | "sample_count", |
| 338 | ): |
| 339 | if baseline_receipt[field] != provenance[field]: |
| 340 | raise PersistenceBacklogError( |
| 341 | f"baseline receipt {field} does not match baseline provenance" |
| 342 | ) |
| 343 | |
| 344 | |
| 345 | def compare( |
| 346 | receipt: dict[str, Any], |
| 347 | budget: dict[str, Any], |
| 348 | *, |
| 349 | expected_source: dict[str, Any] | None = None, |
| 350 | require_clean_source: bool = False, |
| 351 | ) -> tuple[list[tuple[str, int, int]], list[tuple[str, int, int]]]: |
| 352 | validate_receipt( |
| 353 | receipt, |
| 354 | expected_source=expected_source, |
| 355 | require_clean_source=require_clean_source, |
| 356 | ) |
| 357 | validate_budget(budget) |
| 358 | increases: list[tuple[str, int, int]] = [] |
| 359 | decreases: list[tuple[str, int, int]] = [] |
| 360 | for field in CEILING_FIELDS: |
| 361 | if field in RSS_DELTA_FIELDS and not receipt["rss_supported"]: |
| 362 | continue |
| 363 | current = receipt[field] |
| 364 | ceiling = budget["ceilings"][field] |
| 365 | if current > ceiling: |
| 366 | increases.append((field, current, ceiling)) |
| 367 | elif current < ceiling: |
| 368 | decreases.append((field, current, ceiling)) |
| 369 | return increases, decreases |
| 370 | |
| 371 | |
| 372 | def measure() -> dict[str, Any]: |
| 373 | env = os.environ.copy() |
| 374 | env["CARGO_NET_OFFLINE"] = "true" |
| 375 | result = subprocess.run( |
| 376 | [sys.executable, str(MEASURE_SCRIPT)], |
| 377 | cwd=ROOT, |
| 378 | env=env, |
| 379 | text=True, |
| 380 | capture_output=True, |
| 381 | check=False, |
| 382 | ) |
| 383 | sys.stderr.write(result.stderr) |
| 384 | if result.returncode != 0: |
| 385 | sys.stdout.write(result.stdout) |
| 386 | raise PersistenceBacklogError("measurement command failed") |
| 387 | try: |
| 388 | receipt = json.loads(result.stdout) |
| 389 | except json.JSONDecodeError as error: |
| 390 | raise PersistenceBacklogError(f"measurement emitted invalid JSON: {error}") from error |
| 391 | if not isinstance(receipt, dict): |
| 392 | raise PersistenceBacklogError("measurement receipt must be an object") |
| 393 | return receipt |
| 394 | |
| 395 | |
| 396 | def main() -> int: |
| 397 | parser = argparse.ArgumentParser(description=__doc__) |
| 398 | parser.add_argument("--receipt", type=Path, help="check an existing receipt") |
| 399 | parser.add_argument("--budget", type=Path, default=BUDGET_PATH) |
| 400 | args = parser.parse_args() |
| 401 | try: |
| 402 | expected_source = current_source_identity() |
| 403 | receipt = load_json(args.receipt, "receipt") if args.receipt else measure() |
| 404 | budget = load_json(args.budget, "budget") |
| 405 | baseline_receipt = load_json(BASELINE_RECEIPT_PATH, "baseline receipt") |
| 406 | validate_baseline_receipt(budget, baseline_receipt) |
| 407 | increases, decreases = compare( |
| 408 | receipt, |
| 409 | budget, |
| 410 | expected_source=expected_source, |
| 411 | require_clean_source=True, |
| 412 | ) |
| 413 | except PersistenceBacklogError as error: |
| 414 | print(f"[persistence-backlog-budget] ERROR: {error}", file=sys.stderr) |
| 415 | return 2 |
| 416 | if increases: |
| 417 | for field, current, ceiling in increases: |
| 418 | print( |
| 419 | f"[persistence-backlog-budget] FAIL: {field}={current} exceeds {ceiling}", |
| 420 | file=sys.stderr, |
| 421 | ) |
| 422 | return 1 |
| 423 | print("[persistence-backlog-budget] PASS: one-way ceilings respected") |
| 424 | for field, current, ceiling in decreases: |
| 425 | print(f" can tighten {field}: {current} < {ceiling}") |
| 426 | return 0 |
| 427 | |
| 428 | |
| 429 | if __name__ == "__main__": |
| 430 | raise SystemExit(main()) |
| 431 |