crank/scripts/check-metrics-boundaries.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import re
import subprocess
import sys
from pathlib import Path


MAX_FILES = 20_000
MAX_FILE_BYTES = 4 * 1024 * 1024
PATTERNS = (
    re.compile(
        r"(?<![A-Za-z0-9_])(?:::)?metrics\s*::\s*(?:counter|gauge|histogram|describe_counter|describe_gauge|describe_histogram|recorder|with_local_recorder|Key|Recorder)\b"
    ),
    re.compile(
        r"\b(?:pub\s+)?use\s+(?:::)?metrics\s*(?:\s+as\s+\w+|::\s*(?:Unit|counter|gauge|histogram|Recorder|Key|\{\s*(?:counter|gauge|histogram|Recorder|Key)))"
    ),
    re.compile(r"\bextern\s+crate\s+metrics(?:\s+as\s+\w+)?"),
    re.compile(r"\b(?:set_global_recorder|set_boxed_recorder|PrometheusRecorder|DebuggingRecorder|Recorder\s*::|register_counter|register_gauge|register_histogram)\b"),
)
DESCRIBE_ONLY = re.compile(
    r"(?<![A-Za-z0-9_])metrics\s*::\s*describe_(?:counter|gauge|histogram)\s*!"
)


def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Enforce crank-metrics ownership.")
    parser.add_argument("--root", type=Path, default=Path.cwd())
    parser.add_argument("--files", nargs="*")
    return parser.parse_args(argv)


def discover(root: Path) -> list[str]:
    result = subprocess.run(
        ["git", "ls-files", "--cached", "--others", "--exclude-standard", "-z", "--", "apps/**/*.rs", "crates/**/*.rs"],
        cwd=root,
        check=True,
        stdout=subprocess.PIPE,
    )
    return sorted(path.decode("utf-8") for path in result.stdout.split(b"\0") if path)


def resolve(root: Path, supplied: str) -> tuple[str, Path] | None:
    if "\0" in supplied:
        return None
    logical = Path(supplied)
    if logical.is_absolute() or ".." in logical.parts:
        return None
    path = root / logical
    if path.is_symlink() or not path.is_file():
        return None
    try:
        canonical = path.resolve(strict=True)
        relative = canonical.relative_to(root)
    except (OSError, ValueError):
        return None
    if canonical != path.absolute():
        return None
    return relative.as_posix(), canonical


def strip_non_code(text: str) -> str:
    """Remove comments and literals while preserving newlines and token separation."""
    output = list(text)
    index = 0
    block_depth = 0
    while index < len(text):
        if block_depth:
            if text.startswith("/*", index):
                block_depth += 1
                output[index:index + 2] = "  "
                index += 2
            elif text.startswith("*/", index):
                block_depth -= 1
                output[index:index + 2] = "  "
                index += 2
            else:
                if text[index] != "\n":
                    output[index] = " "
                index += 1
            continue
        if text.startswith("//", index):
            end = text.find("\n", index)
            end = len(text) if end < 0 else end
            for cursor in range(index, end):
                output[cursor] = " "
            index = end
            continue
        if text.startswith("/*", index):
            block_depth = 1
            output[index:index + 2] = "  "
            index += 2
            continue
        raw = re.match(r"(?:br|r)(?P<hashes>#{0,16})\"", text[index:])
        if raw:
            terminator = '"' + raw.group("hashes")
            end = text.find(terminator, index + raw.end())
            end = len(text) if end < 0 else end + len(terminator)
            for cursor in range(index, end):
                if text[cursor] != "\n":
                    output[cursor] = " "
            index = end
            continue
        prefix = 1 if text.startswith("b\"", index) or text.startswith("b'", index) else 0
        quote_index = index + prefix
        is_string = quote_index < len(text) and text[quote_index] == '"'
        is_char = quote_index < len(text) and text[quote_index] == "'" and re.match(
            r"'(?:\\.|[^\\'\n])'", text[quote_index:]
        )
        if is_string or is_char:
            quote = text[quote_index]
            end = quote_index + 1
            escaped = False
            while end < len(text):
                char = text[end]
                end += 1
                if escaped:
                    escaped = False
                elif char == "\\":
                    escaped = True
                elif char == quote:
                    break
            for cursor in range(index, end):
                if text[cursor] != "\n":
                    output[cursor] = " "
            index = end
            continue
        index += 1
    return "".join(output)


def is_test_path(logical: str) -> bool:
    parts = Path(logical).parts
    return bool(parts and parts[0] == "tests") or (
        len(parts) >= 4 and parts[0] in {"apps", "crates"} and parts[2] == "tests"
    )


def scan(logical: str, path: Path) -> bool:
    if logical.startswith("crates/crank-metrics/"):
        return False
    if is_test_path(logical):
        return False
    try:
        if path.stat().st_size > MAX_FILE_BYTES:
            return True
        text = strip_non_code(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError):
        return True
    matches = [match for pattern in PATTERNS for match in pattern.finditer(text)]
    if not matches:
        return False
    if logical == "crates/crank-observability/src/instrumentation.rs":
        residual = DESCRIBE_ONLY.sub("", text)
        residual = re.sub(r"\buse\s+metrics\s*::\s*Unit\s*;", "", residual)
        return any(pattern.search(residual) for pattern in PATTERNS)
    if logical == "crates/crank-observability/src/prometheus.rs":
        allowed = ("PrometheusRecorder",)
        residual = text
        for token in allowed:
            residual = residual.replace(token, "")
        return any(pattern.search(residual) for pattern in PATTERNS)
    return True
    return False


def main(argv: list[str]) -> int:
    args = parse_args(argv)
    root = args.root.resolve()
    try:
        supplied = args.files if args.files is not None else discover(root)
    except (OSError, subprocess.SubprocessError, UnicodeError):
        print("error: metrics boundary discovery failed", file=sys.stderr)
        return 1
    if len(supplied) > MAX_FILES:
        print("error: metrics boundary input limit exceeded", file=sys.stderr)
        return 1
    if args.files is not None and not supplied:
        print("error: metrics boundary explicit file list is empty", file=sys.stderr)
        return 1
    violations: list[str] = []
    for index, item in enumerate(sorted(set(supplied))):
        resolved = resolve(root, item)
        if resolved is None:
            violations.append(f"input[{index}]: invalid path")
            continue
        logical, path = resolved
        if path.suffix == ".rs" and scan(logical, path):
            violations.append(f"{logical}: direct metrics declaration")
    if violations:
        for violation in violations[:1000]:
            print(f"error: {violation}", file=sys.stderr)
        if len(violations) > 1000:
            print(f"error: diagnostics omitted={len(violations) - 1000}", file=sys.stderr)
        return 1
    print(f"Metrics boundary check passed: files={len(supplied)}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))