Consolidate required CI gate (#115)

tools/normalize_git_mtimes.py

@@ -0,0 +1,164 @@
+#!/usr/bin/env python3
+"""Give tracked files stable mtimes derived from their immutable Git blobs."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path, PurePath
+import subprocess
+import tempfile
+import time
+
+
+WINDOWS_TICK_NANOSECONDS = 100
+
+
+def tracked_entries(root: Path) -> list[tuple[bytes, Path]]:
+    output = subprocess.run(
+        ["git", "ls-files", "--stage", "-z"],
+        cwd=root,
+        check=True,
+        stdout=subprocess.PIPE,
+    ).stdout
+    entries: list[tuple[bytes, Path]] = []
+    for record in output.split(b"\0"):
+        if not record:
+            continue
+        metadata, separator, raw_path = record.partition(b"\t")
+        if not separator:
+            raise RuntimeError("git ls-files produced a record without a path")
+        mode, _object_id, stage = metadata.split(b" ")
+        if stage != b"0":
+            raise RuntimeError(
+                f"cannot normalize an unmerged index entry: {os.fsdecode(raw_path)}"
+            )
+        relative = PurePath(os.fsdecode(raw_path))
+        if relative.is_absolute() or ".." in relative.parts:
+            raise RuntimeError(f"unsafe tracked path: {relative}")
+        entries.append((mode, root.joinpath(*relative.parts)))
+    return entries
+
+
+def content_identity(mode: bytes, path: Path) -> str:
+    digest = hashlib.sha256(mode + b"\0")
+    if path.is_symlink():
+        digest.update(os.fsencode(os.readlink(path)))
+    else:
+        with path.open("rb") as source:
+            while chunk := source.read(1024 * 1024):
+                digest.update(chunk)
+    return digest.hexdigest()
+
+
+def default_state_path(root: Path) -> Path:
+    return root / "target" / ".metacrate-mtimes.json"
+
+
+def read_state(path: Path) -> dict[str, dict[str, int | str]]:
+    if not path.is_file():
+        return {}
+    value = json.loads(path.read_text(encoding="utf-8"))
+    if value.get("schema") != 1 or not isinstance(value.get("files"), dict):
+        raise RuntimeError(f"invalid tracked-mtime state: {path}")
+    return value["files"]
+
+
+def write_state(path: Path, files: dict[str, dict[str, int | str]]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
+    temporary.write_text(
+        json.dumps({"schema": 1, "files": files}, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+    os.replace(temporary, path)
+
+
+def normalize(root: Path, state_path: Path, check: bool) -> int:
+    previous = read_state(state_path)
+    current: dict[str, dict[str, int | str]] = {}
+    changed = 0
+    prior_max = max(
+        (int(value["mtime_ns"]) for value in previous.values()), default=0
+    )
+    wall_time = time.time_ns() // WINDOWS_TICK_NANOSECONDS * WINDOWS_TICK_NANOSECONDS
+    now = max(wall_time, prior_max + WINDOWS_TICK_NANOSECONDS)
+    for mode, path in tracked_entries(root):
+        if not (path.is_file() or path.is_symlink()):
+            continue
+        relative = path.relative_to(root).as_posix()
+        identity = content_identity(mode, path)
+        prior = previous.get(relative)
+        expected = (
+            int(prior["mtime_ns"])
+            if prior is not None and prior.get("identity") == identity
+            else now
+        )
+        current[relative] = {"identity": identity, "mtime_ns": expected}
+        actual = os.stat(path, follow_symlinks=False).st_mtime_ns
+        if actual == expected:
+            continue
+        if check:
+            raise RuntimeError(f"tracked mtime is not normalized: {path}")
+        os.utime(path, ns=(expected, expected), follow_symlinks=False)
+        changed += 1
+    if not check:
+        write_state(state_path, current)
+    return changed
+
+
+def self_test() -> None:
+    with tempfile.TemporaryDirectory(prefix="metacrate-mtimes-") as directory:
+        root = Path(directory)
+        subprocess.run(["git", "init", "--quiet"], cwd=root, check=True)
+        subprocess.run(
+            ["git", "config", "user.email", "ci@example.invalid"], cwd=root, check=True
+        )
+        subprocess.run(
+            ["git", "config", "user.name", "MetaCrate CI"], cwd=root, check=True
+        )
+        first = root / "first.txt"
+        second = root / "second.txt"
+        first.write_text("first\n", encoding="utf-8")
+        second.write_text("second\n", encoding="utf-8")
+        subprocess.run(["git", "add", "."], cwd=root, check=True)
+        subprocess.run(["git", "commit", "--quiet", "-m", "fixture"], cwd=root, check=True)
+        state = root / "state.json"
+
+        assert normalize(root, state, False) == 2
+        assert normalize(root, state, True) == 0
+        first_mtime = first.stat().st_mtime_ns
+        second_mtime = second.stat().st_mtime_ns
+
+        first.write_text("changed\n", encoding="utf-8")
+        assert normalize(root, state, False) == 1
+        assert first.stat().st_mtime_ns > first_mtime
+        assert second.stat().st_mtime_ns == second_mtime
+        assert normalize(root, state, True) == 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--check", action="store_true")
+    parser.add_argument("--self-test", action="store_true")
+    parser.add_argument("--root", type=Path, default=Path.cwd())
+    parser.add_argument("--state", type=Path)
+    args = parser.parse_args()
+    if args.self_test:
+        self_test()
+        print("tracked Git mtime self-test: ok")
+        return 0
+    root = args.root.resolve()
+    state_path = args.state or default_state_path(root)
+    if not state_path.is_absolute():
+        state_path = root / state_path
+    changed = normalize(root, state_path, args.check)
+    verb = "verified" if args.check else "normalized"
+    print(f"{verb} tracked Git mtimes ({changed} changed)")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())