#!/usr/bin/env python3 """Enforce a ratcheting budget for swallowed-error-like Rust patterns. This is intentionally a broad guardrail. It tracks production occurrences of patterns that commonly hide failures and should either be removed, logged, propagated, or explicitly accepted as best-effort: - `let _ = ...` - `.ok()` - `.unwrap_or_default()` Policy: - Existing files may not increase their count. - New production files may not introduce these patterns. - Total count may not increase. - `--update` refreshes the baseline after intentional cleanup. """ from __future__ import annotations import argparse import json import re import sys from pathlib import Path from typing import Any REPO_ROOT = Path(__file__).resolve().parent.parent BASELINE_FILE = REPO_ROOT / "scripts" / "swallowed_error_budget.json" SCAN_ROOTS = (REPO_ROOT / "src", REPO_ROOT / "crates") PATTERNS = { "let_underscore": re.compile(r"\blet\s+_\s*="), "dot_ok": re.compile(r"\.ok\(\)"), "unwrap_or_default": re.compile(r"\.unwrap_or_default\(\)"), } CFG_TEST_RE = re.compile(r"^\s*#\s*\[\s*cfg\s*\(\s*test\s*\)\s*\]") ITEM_START_RE = re.compile(r"^\s*(?:pub(?:\([^)]*\))?\s+)?(?:mod|fn)\b") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--update", action="store_true", help="refresh the baseline") return parser.parse_args() def is_test_rust_file(path: Path) -> bool: rel = path.relative_to(REPO_ROOT).as_posix() if path.suffix != ".rs": return False parts = rel.split("/") if parts[0] == "tests" or any( part == "tests" or part.endswith("_tests") or part.endswith("_test") or part.startswith("tests_") for part in parts ): return True name = path.name return ( name == "tests.rs" or name.endswith("_tests.rs") or name.endswith("_test.rs") or name.startswith("tests_") ) def production_rust_files() -> list[Path]: files: list[Path] = [] for root in SCAN_ROOTS: if not root.exists(): continue for path in sorted(root.rglob("*.rs")): if path.suffix == ".rs" and not is_test_rust_file(path): files.append(path) return files def brace_delta(line: str) -> int: return line.count("{") - line.count("}") def production_lines(path: Path) -> list[str]: lines = path.read_text(encoding="utf-8", errors="ignore").splitlines() output: list[str] = [] skip_stack: list[int] = [] pending_cfg_test = False for line in lines: stripped = line.strip() current_depth = sum(skip_stack) if current_depth == 0: if pending_cfg_test and ITEM_START_RE.match(line): delta = brace_delta(line) if delta > 0: skip_stack.append(delta) pending_cfg_test = False continue if pending_cfg_test and stripped and not stripped.startswith("#"): pending_cfg_test = False if CFG_TEST_RE.match(line): pending_cfg_test = True continue output.append(line) else: skip_stack[-1] += brace_delta(line) if skip_stack[-1] <= 0: skip_stack.pop() return output def zero_counts() -> dict[str, int]: return {name: 0 for name in PATTERNS} def current_counts() -> dict[str, dict[str, int]]: counts: dict[str, dict[str, int]] = {} for path in production_rust_files(): file_counts = zero_counts() for line in production_lines(path): for name, pattern in PATTERNS.items(): if pattern.search(line): file_counts[name] += 1 if sum(file_counts.values()) > 0: counts[path.relative_to(REPO_ROOT).as_posix()] = file_counts return counts def file_total(counts: dict[str, int]) -> int: return sum(counts.values()) def total_counts(counts: dict[str, dict[str, int]]) -> dict[str, int]: totals = zero_counts() for file_counts in counts.values(): for name, count in file_counts.items(): totals[name] = totals.get(name, 0) + count return totals def grand_total(counts: dict[str, dict[str, int]]) -> int: return sum(file_total(file_counts) for file_counts in counts.values()) def load_baseline() -> dict[str, Any]: if not BASELINE_FILE.exists(): return {"version": 1, "total": 0, "totals_by_pattern": zero_counts(), "tracked_files": {}} data = json.loads(BASELINE_FILE.read_text(encoding="utf-8")) if not isinstance(data, dict): raise SystemExit(f"error: invalid baseline file format: {BASELINE_FILE}") tracked = data.get("tracked_files") totals_by_pattern = data.get("totals_by_pattern") total = data.get("total") if not isinstance(total, int) or total < 0: raise SystemExit(f"error: invalid total in {BASELINE_FILE}") if not isinstance(totals_by_pattern, dict): raise SystemExit(f"error: invalid totals_by_pattern in {BASELINE_FILE}") if not isinstance(tracked, dict): raise SystemExit(f"error: invalid tracked_files in {BASELINE_FILE}") for path, file_counts in tracked.items(): if not isinstance(path, str) or not isinstance(file_counts, dict): raise SystemExit(f"error: invalid tracked_files entry in {BASELINE_FILE}") if any(not isinstance(v, int) or v < 0 for v in file_counts.values()): raise SystemExit(f"error: invalid count in tracked_files entry for {path}") return data def write_baseline(counts: dict[str, dict[str, int]]) -> None: BASELINE_FILE.write_text( json.dumps( { "version": 1, "total": grand_total(counts), "totals_by_pattern": total_counts(counts), "tracked_files": counts, }, indent=2, sort_keys=True, ) + "\n", encoding="utf-8", ) def main() -> int: args = parse_args() baseline = load_baseline() current = current_counts() current_total = grand_total(current) current_pattern_totals = total_counts(current) if args.update: write_baseline(current) print( "Updated swallowed-error baseline: " f"total={baseline['total']} -> {current_total}, " f"files={len(baseline['tracked_files'])} -> {len(current)}" ) return 0 tracked: dict[str, dict[str, int]] = baseline["tracked_files"] regressions: list[str] = [] improvements: list[str] = [] if current_total > baseline["total"]: regressions.append(f"total swallowed-error-like count grew: {baseline['total']} -> {current_total}") elif current_total < baseline["total"]: improvements.append(f"total swallowed-error-like count shrank: {baseline['total']} -> {current_total}") baseline_pattern_totals: dict[str, int] = baseline["totals_by_pattern"] for name, count in sorted(current_pattern_totals.items()): old_count = baseline_pattern_totals.get(name, 0) if count > old_count: regressions.append(f"{name} count grew: {old_count} -> {count}") elif count < old_count: improvements.append(f"{name} count shrank: {old_count} -> {count}") for path, file_counts in sorted(current.items()): old_counts = tracked.get(path) if old_counts is None: regressions.append(f"new swallowed-error-like usage: {path} ({file_total(file_counts)})") continue old_total = file_total(old_counts) new_total = file_total(file_counts) if new_total > old_total: regressions.append(f"swallowed-error-like usage grew: {path} ({old_total} -> {new_total})") elif new_total < old_total: improvements.append(f"swallowed-error-like usage shrank: {path} ({old_total} -> {new_total})") for path, old_counts in sorted(tracked.items()): if path not in current: improvements.append(f"swallowed-error-like usage removed: {path} ({file_total(old_counts)} -> 0)") if regressions: print("Swallowed-error budget exceeded:", file=sys.stderr) for entry in regressions: print(f" - {entry}", file=sys.stderr) print("Run scripts/check_swallowed_error_budget.py --update only after intentional cleanup.", file=sys.stderr) return 1 if improvements: print("Swallowed-error budget improved:") for entry in improvements: print(f" - {entry}") print("Consider running: scripts/check_swallowed_error_budget.py --update") else: print( "Swallowed-error budget OK: " f"total={current_total} files={len(current)} patterns={current_pattern_totals}" ) return 0 if __name__ == "__main__": raise SystemExit(main())