diff --git a/automation/jobs.yml b/automation/jobs.yml new file mode 100644 index 000000000..4d5de8963 --- /dev/null +++ b/automation/jobs.yml @@ -0,0 +1,92 @@ +# Jobs scripts/autorun.py runs. A change here is a reviewed commit: the +# schedule, the gate and the paths a job may commit are all decided here. +# +# Commands run in the automation worktree (.cache/autorun/worktree) unless +# `worktree: false`, which runs them in the main checkout and forbids +# commits. `python` names the interpreter running autorun.py. A step is a +# command list, or a mapping with `run`, `ok_codes` and `stdout` (a file +# the command's output is written to). `twice` recomputes an artifact the +# gate wrote into `{out}` and compares both, timestamps aside. Each commit +# takes only the changed files its `paths` globs match, and a job that +# leaves any other change behind fails. `report` names a JSON file a step +# writes, kept in the job's status. +# +# Schedules are systemd OnCalendar expressions, written into user timers by +# `python scripts/autorun.py install-timers`. + +defaults: + min_free_gb: 30 + max_load_wait: 3600 + timeout: 21600 + +jobs: + cleanup: + schedule: "*-*-* 05:30" + worktree: false + min_free_gb: 0 + steps: + - [python, scripts/cleanup_disk.py, --apply, --hours, "72"] + + disk-check: + schedule: hourly + worktree: false + lock: false + min_free_gb: 0 + max_load_wait: 0 + steps: + - [python, scripts/cleanup_disk.py, --check] + + nightly: + schedule: "*-*-* 02:00" + network: true + min_free_gb: 120 + steps: + - run: [python, scripts/check_freshness.py, --json] + ok_codes: [0, 1] + stdout: tmp/autorun/freshness.json + - [python, scripts/refresh_stale.py, --findings, tmp/autorun/freshness.json, + --only, "data,catalogs"] + - [python, -m, scripts.scraper.mame_hash_scraper] + - [python, -m, scripts.scraper.fbneo_hash_scraper] + gate: + - [python, scripts/pipeline.py, --offline] + twice: + - artifact: database.json + run: [python, scripts/generate_db.py, --force, --bios-dir, bios, --output, "{out}"] + - artifact: install + glob: "*.json" + run: [python, scripts/generate_pack.py, --all, --manifest, --offline, + --output-dir, "{out}"] + commits: + - subject: "chore: refresh mame and fbneo hashes" + body: >- + Written by mame_hash_scraper and fbneo_hash_scraper from the + upstream sources, behind a green offline pipeline. + paths: [emulators/mame*.yml, emulators/fbneo*.yml] + - subject: "chore: refresh romset recipes" + body: >- + Snapshots refreshed by refresh_stale from the MAME and FBNeo DATs + check_freshness reported stale, behind a green offline pipeline. + paths: [recipes/*.json] + - subject: "chore: regenerate collection records" + body: >- + Regenerated by the offline pipeline, database.json and the + manifests computed twice with the same result. + paths: [database.json, install/*.json, install/targets/*.json, README.md, + CONTRIBUTING.md, website/src/data/*.json] + scratch: [dist, tmp/autorun] + + bios-integrity: + schedule: "Sun *-*-* 08:30" + worktree: false + max_load_wait: 14400 + steps: + - run: [python, scripts/bios_integrity.py, --apply, --json] + stdout: tmp/autorun/bios-integrity.json + report: tmp/autorun/bios-integrity.json + + assets: + schedule: "*-*-* 04:30" + network: true + steps: + - [python, scripts/sync_release_assets.py] diff --git a/scripts/autorun.py b/scripts/autorun.py new file mode 100644 index 000000000..9ee2cfac7 --- /dev/null +++ b/scripts/autorun.py @@ -0,0 +1,1139 @@ +#!/usr/bin/env python3 +"""Run the repository's unattended jobs, one at a time, in their own worktree. + +Jobs are declared in automation/jobs.yml: the commands, the gate they must +pass, the artifacts computed twice, the paths each commit may carry and the +schedule. A run takes one host-wide lock, waits for an idle host, refuses to +start on a full disk, and retries a step once on an idle host when it died +the way this host's CPU fails (exit 139, SystemError, "unknown opcode"). + +Jobs that commit run in .cache/autorun/worktree on the branch `autorun`, +never in the shared checkout and its index. The worktree is created on the +first run, fast-forwarded (or rebased) onto origin/main before each job, and +a job stops when it finds anything left uncommitted there. Commits name +their paths. Pushing is off unless --push is given: the batch is rebased on +origin/main, the gate runs again when origin/main moved, and the push never +forces. A red CI run on a pushed automation commit opens a circuit breaker +that stops every commit and push until a green run on main follows. + +State, logs and the worktree live under .cache/autorun/ of the main +checkout. The lock is ~/.local/state/retrobios/auto.lock (RETROBIOS_AUTO_LOCK +overrides it), shared with any other heavy job on this host. + +Usage: + python scripts/autorun.py run nightly + python scripts/autorun.py run nightly --dry-run + python scripts/autorun.py run nightly --push + python scripts/autorun.py status + python scripts/autorun.py status --json + python scripts/autorun.py install-timers + python scripts/autorun.py install-timers --enable +""" + +from __future__ import annotations + +import argparse +import contextlib +import datetime as dt +import fnmatch +import json +import os +import re +import shlex +import shutil +import subprocess +import sys +import time +from collections.abc import Callable, Iterator, Sequence +from dataclasses import dataclass, field +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +JOBS_FILE = REPO_ROOT / "automation" / "jobs.yml" +AUTORUN_DIR = Path(".cache") / "autorun" +DEFAULT_LOCK = Path.home() / ".local" / "state" / "retrobios" / "auto.lock" +LOCK_ENV = "RETROBIOS_AUTO_LOCK" + +BRANCH = "autorun" +REMOTE = "origin" +TARGET = "main" +REPO_SLUG = "Abdess/retrobios" + +OK = "ok" +FAILED = "failed" +DEFERRED = "deferred" +REFUSED = "refused" +BLOCKED = "blocked" +EXIT_CODES = {OK: 0, DEFERRED: 75} + +ALERT_FREE_GB = 100 +LOG_DAYS = 30 +SEGFAULT_CODES = (139, -11) +HOST_MISREAD_MARKERS = ("SystemError", "unknown opcode") +VOLATILE_KEYS = frozenset({"generated_at", "generated", "imported_at"}) + +SUBJECT_TYPES = re.compile(r"^(feat|refactor|chore|docs|fix): [a-z0-9]") +FORBIDDEN_WORDS = re.compile( + r"\b(bug|regression|revert|dead code|broken|wrong|hack|workaround|" + r"comprehensive|implement)", + re.IGNORECASE, +) +CI_RED = frozenset({"failure", "timed_out", "startup_failure"}) +CI_GREEN = frozenset({"success", "skipped", "neutral"}) + +JOB_KEYS = frozenset({ + "schedule", "steps", "gate", "twice", "commits", "worktree", "lock", + "network", "min_free_gb", "max_load_wait", "timeout", "scratch", "report", +}) + + +class AutorunError(RuntimeError): + """A condition that ends a job with a named result.""" + + def __init__(self, result: str, detail: str) -> None: + super().__init__(detail) + self.result = result + self.detail = detail + + +class LockBusy(RuntimeError): + """Another job holds the host-wide lock.""" + + +@dataclass(frozen=True) +class Step: + argv: tuple[str, ...] + ok_codes: tuple[int, ...] = (0,) + stdout: str = "" + + +@dataclass(frozen=True) +class Twice: + """An artifact the gate wrote, computed again into {out} and compared.""" + + artifact: str + argv: tuple[str, ...] + glob: str = "" + + +@dataclass(frozen=True) +class CommitRule: + subject: str + paths: tuple[str, ...] + body: str = "" + + +@dataclass(frozen=True) +class Job: + name: str + steps: tuple[Step, ...] = () + gate: tuple[Step, ...] = () + twice: tuple[Twice, ...] = () + commits: tuple[CommitRule, ...] = () + schedule: str = "" + worktree: bool = True + lock: bool = True + network: bool = False + min_free_gb: float = 30.0 + max_load_wait: int = 3600 + timeout: int = 21600 + scratch: tuple[str, ...] = () + report: str = "" + + +@dataclass +class Outcome: + returncode: int + text: str = "" + retried: bool = False + + +@dataclass +class Host: + """What a run reads from the machine, replaced by fakes in the tests.""" + + loadavg: Callable[[], tuple[float, float, float]] = os.getloadavg + sleep: Callable[[float], None] = time.sleep + clock: Callable[[], float] = time.monotonic + cpu_count: int = os.cpu_count() or 1 + poll: float = 60.0 + + @property + def max_load(self) -> float: + return self.cpu_count / 4 + + +@dataclass +class Context: + repo: Path + jobs: dict[str, Job] + lock_path: Path + host: Host = field(default_factory=Host) + gh: Callable[[list[str]], object] | None = None + dry_run: bool = False + push: bool = False + + @property + def home(self) -> Path: + return self.repo / AUTORUN_DIR + + @property + def worktree(self) -> Path: + return self.home / "worktree" + + @property + def state_path(self) -> Path: + return self.home / "state.json" + + +# Job registry + + +def check_subject(subject: str, body: str = "") -> list[str]: + """What the repository's commit-msg hook would refuse in this message.""" + problems = [] + if len(subject) > 50: + problems.append(f"subject is {len(subject)} characters, 50 at most") + if not SUBJECT_TYPES.match(subject): + problems.append("subject must read 'type: description'") + if FORBIDDEN_WORDS.search(f"{subject}\n{body}"): + problems.append("message uses a word the history does not carry") + if re.search(r"^co-authored-by", body, re.IGNORECASE | re.MULTILINE): + problems.append("no Co-Authored-By trailer") + return problems + + +def _step(raw: object, where: str) -> Step: + if isinstance(raw, list): + return Step(argv=tuple(str(a) for a in raw)) + if isinstance(raw, dict) and isinstance(raw.get("run"), list): + unknown = set(raw) - {"run", "ok_codes", "stdout"} + if unknown: + raise ValueError(f"{where}: unknown step key {sorted(unknown)}") + return Step( + argv=tuple(str(a) for a in raw["run"]), + ok_codes=tuple(int(c) for c in raw.get("ok_codes", (0,))), + stdout=str(raw.get("stdout") or ""), + ) + raise ValueError(f"{where}: a step is a command list or a mapping with 'run'") + + +def parse_jobs(data: dict) -> dict[str, Job]: + """Jobs from the registry mapping, refused whole on any unknown key.""" + defaults = data.get("defaults") or {} + jobs: dict[str, Job] = {} + for name, raw in (data.get("jobs") or {}).items(): + raw = {**defaults, **(raw or {})} + unknown = set(raw) - JOB_KEYS + if unknown: + raise ValueError(f"job {name}: unknown key {sorted(unknown)}") + twice = [] + for i, item in enumerate(raw.get("twice") or ()): + argv = tuple(str(a) for a in item["run"]) + if "{out}" not in argv: + raise ValueError(f"job {name}: twice[{i}] names no {{out}}") + twice.append(Twice(str(item["artifact"]), argv, str(item.get("glob") or ""))) + commits = [] + for item in raw.get("commits") or (): + rule = CommitRule( + subject=str(item["subject"]), + paths=tuple(str(p) for p in item["paths"]), + body=str(item.get("body") or "").strip(), + ) + problems = check_subject(rule.subject, rule.body) + if problems: + raise ValueError(f"job {name}: {rule.subject!r}: {', '.join(problems)}") + commits.append(rule) + if commits and raw.get("worktree") is False: + raise ValueError(f"job {name}: a job that commits runs in the worktree") + jobs[name] = Job( + name=name, + steps=tuple(_step(s, f"job {name}") for s in raw.get("steps") or ()), + gate=tuple(_step(s, f"job {name} gate") for s in raw.get("gate") or ()), + twice=tuple(twice), + commits=tuple(commits), + schedule=str(raw.get("schedule") or ""), + worktree=bool(raw.get("worktree", True)), + lock=bool(raw.get("lock", True)), + network=bool(raw.get("network", False)), + min_free_gb=float(raw.get("min_free_gb", 30)), + max_load_wait=int(raw.get("max_load_wait", 3600)), + timeout=int(raw.get("timeout", 21600)), + scratch=tuple(str(p) for p in raw.get("scratch") or ()), + report=str(raw.get("report") or ""), + ) + return jobs + + +def load_jobs(path: Path = JOBS_FILE) -> dict[str, Job]: + import yaml + + with path.open(encoding="utf-8") as handle: + return parse_jobs(yaml.safe_load(handle) or {}) + + +# Host guards + + +@contextlib.contextmanager +def global_lock(path: Path) -> Iterator[None]: + """Hold the host-wide automation lock, or raise LockBusy at once.""" + import fcntl + + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a+", encoding="utf-8") as handle: + try: + fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError as exc: + raise LockBusy(f"{path} is held by another job") from exc + handle.seek(0) + handle.truncate() + handle.write(f"{os.getpid()}\n") + handle.flush() + try: + yield + finally: + fcntl.flock(handle, fcntl.LOCK_UN) + + +def lock_busy(path: Path) -> bool: + try: + with global_lock(path): + return False + except LockBusy: + return True + + +def wait_for_load(host: Host, max_wait: float) -> bool: + """True once the 1-minute load is at most a quarter of the CPUs.""" + deadline = host.clock() + max_wait + while True: + if host.loadavg()[0] <= host.max_load: + return True + left = deadline - host.clock() + if left <= 0: + return False + host.sleep(min(host.poll, left)) + + +def free_gb(path: Path) -> float: + return shutil.disk_usage(path).free / 1024 ** 3 + + +def host_misread(returncode: int, text: str) -> bool: + """A failure this host's CPU produces under load, worth one idle retry.""" + if returncode in SEGFAULT_CODES: + return True + return returncode != 0 and any(marker in text for marker in HOST_MISREAD_MARKERS) + + +# Running commands + + +def resolve_argv(argv: Sequence[str], out: str = "") -> list[str]: + """`python` is the interpreter running autorun, {out} the scratch output.""" + resolved = [sys.executable if a == "python" else a for a in argv] + return [a.replace("{out}", out) for a in resolved] if out else resolved + + +def execute( + argv: Sequence[str], + cwd: Path, + log: Path, + env: dict[str, str], + timeout: float, + stdout: Path | None = None, +) -> Outcome: + """Run one command, its output appended to the job log.""" + log.parent.mkdir(parents=True, exist_ok=True) + with log.open("a", encoding="utf-8") as handle: + handle.write(f"$ {shlex.join(argv)}\n") + start = handle.tell() + try: + with log.open("a", encoding="utf-8") as handle: + if stdout is not None: + stdout.parent.mkdir(parents=True, exist_ok=True) + with stdout.open("w", encoding="utf-8") as out: + proc = subprocess.run( + list(argv), cwd=cwd, stdout=out, stderr=handle, + env=env, timeout=max(1, timeout), check=False, + ) + else: + proc = subprocess.run( + list(argv), cwd=cwd, stdout=handle, stderr=subprocess.STDOUT, + env=env, timeout=max(1, timeout), check=False, + ) + returncode = proc.returncode + except subprocess.TimeoutExpired: + returncode = 124 + with log.open("a", encoding="utf-8") as handle: + handle.write(f"\nTIMEOUT after {int(timeout)}s\n") + except OSError as exc: + returncode = 127 + with log.open("a", encoding="utf-8") as handle: + handle.write(f"\ncannot run: {exc}\n") + with log.open("rb") as handle: + handle.seek(start) + text = handle.read().decode("utf-8", errors="replace") + return Outcome(returncode, text) + + +def run_with_retry( + run: Callable[[], Outcome], + ok_codes: Sequence[int], + wait_idle: Callable[[], object], +) -> Outcome: + """Run once, and once more on an idle host when the failure is a misread.""" + first = run() + if first.returncode in ok_codes or not host_misread(first.returncode, first.text): + return first + wait_idle() + second = run() + second.retried = True + return second + + +# Double computation + + +def strip_volatile(value: object) -> object: + if isinstance(value, dict): + return {k: strip_volatile(v) for k, v in value.items() if k not in VOLATILE_KEYS} + if isinstance(value, list): + return [strip_volatile(v) for v in value] + return value + + +def comparable(path: Path) -> object: + if path.suffix == ".json": + with contextlib.suppress(ValueError): + return strip_volatile(json.loads(path.read_text(encoding="utf-8"))) + return path.read_bytes() + + +def _files(root: Path, glob: str) -> dict[str, Path]: + if not root.is_dir(): + return {} + found = root.glob(glob) if glob else root.rglob("*") + return {p.relative_to(root).as_posix(): p for p in found if p.is_file()} + + +def differences(a: Path, b: Path, glob: str = "") -> list[str]: + """Files that differ between two outputs, timestamps aside.""" + if a.is_dir() or b.is_dir(): + files_a, files_b = _files(a, glob), _files(b, glob) + return sorted( + name for name in set(files_a) | set(files_b) + if name not in files_a or name not in files_b + or comparable(files_a[name]) != comparable(files_b[name]) + ) + if not a.is_file() or not b.is_file(): + return [a.name] + return [] if comparable(a) == comparable(b) else [a.name] + + +def twice( + argv: Sequence[str], + out_a: Path, + out_b: Path, + run: Callable[[list[str]], Outcome], + glob: str = "", +) -> list[str]: + """Compute one output twice and name what the two runs disagree on.""" + for out in (out_a, out_b): + outcome = run(resolve_argv(argv, str(out))) + if outcome.returncode != 0: + raise AutorunError(FAILED, f"{shlex.join(argv)} exited {outcome.returncode}") + return differences(out_a, out_b, glob) + + +# Git + + +def git(cwd: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess: + proc = subprocess.run( + ["git", "-C", str(cwd), *args], capture_output=True, text=True, check=False, + ) + if check and proc.returncode != 0: + raise AutorunError( + FAILED, f"git {' '.join(args)}: {(proc.stderr or proc.stdout).strip()}" + ) + return proc + + +def changed_files(worktree: Path) -> list[str]: + """Every path git status reports, untracked included, ignored excluded.""" + out = git(worktree, "status", "--porcelain=v1", "-z", "--untracked-files=all").stdout + entries = out.split("\0") + paths: list[str] = [] + i = 0 + while i < len(entries): + entry = entries[i] + i += 1 + if len(entry) < 4: + continue + paths.append(entry[3:]) + if entry[0] in "RC": + i += 1 # the rename source follows + return sorted(set(paths)) + + +def path_matches(pattern: str, path: str) -> bool: + """A glob whose `*` stays inside one path segment.""" + parts, segments = pattern.split("/"), path.split("/") + return len(parts) == len(segments) and all( + fnmatch.fnmatchcase(s, p) for p, s in zip(parts, segments) + ) + + +def plan_commit(changed: Sequence[str], patterns: Sequence[str]) -> list[str]: + return sorted(p for p in changed if any(path_matches(g, p) for g in patterns)) + + +def commit_paths(worktree: Path, files: Sequence[str], rule: CommitRule) -> str: + """Commit exactly these files, and return the new commit.""" + if not files: + raise ValueError("no file to commit") + git(worktree, "add", "--", *files) + message = ["-m", rule.subject] + (["-m", rule.body] if rule.body else []) + git(worktree, "commit", "--quiet", *message, "--", *files) + return git(worktree, "rev-parse", "HEAD").stdout.strip() + + +def _hardlink_tree(src: Path, dst: Path) -> None: + if src.is_dir(): + shutil.copytree(src, dst, copy_function=os.link, symlinks=True) + else: + dst.parent.mkdir(parents=True, exist_ok=True) + os.link(src, dst) + + +def seed_worktree(repo: Path, worktree: Path) -> list[str]: + """Hard-link what git does not carry: a tool deleting a file drops a link.""" + from largefiles import load_registered_paths + + seeded = [] + for rel in [".cache/large", "data", *load_registered_paths(repo / ".gitignore")]: + src, dst = repo / rel, worktree / rel + if src.exists() and not dst.exists(): + _hardlink_tree(src, dst) + seeded.append(rel) + return seeded + + +def ensure_worktree(repo: Path, worktree: Path) -> None: + """Create the automation worktree on branch `autorun`, once.""" + if (worktree / ".git").exists(): + return + git(repo, "fetch", "--quiet", REMOTE, TARGET) + has_branch = git( + repo, "rev-parse", "--verify", "--quiet", f"refs/heads/{BRANCH}", check=False + ).returncode == 0 + worktree.parent.mkdir(parents=True, exist_ok=True) + if has_branch: + git(repo, "worktree", "add", str(worktree), BRANCH) + else: + git(repo, "worktree", "add", "-b", BRANCH, str(worktree), f"{REMOTE}/{TARGET}") + + +def sync_worktree(worktree: Path) -> str: + """Bring the worktree onto origin/main, and return the base it now sits on.""" + leftovers = changed_files(worktree) + if leftovers: + raise AutorunError( + BLOCKED, f"worktree holds {len(leftovers)} uncommitted paths: " + + ", ".join(leftovers[:5]) + ) + git(worktree, "fetch", "--quiet", REMOTE, TARGET) + upstream = f"{REMOTE}/{TARGET}" + ahead = int(git(worktree, "rev-list", "--count", f"{upstream}..HEAD").stdout) + if ahead == 0: + git(worktree, "merge", "--ff-only", "--quiet", upstream) + elif git(worktree, "merge-base", "--is-ancestor", upstream, "HEAD", + check=False).returncode != 0: + if git(worktree, "rebase", "--quiet", upstream, check=False).returncode != 0: + git(worktree, "rebase", "--abort", check=False) + raise AutorunError(BLOCKED, f"{ahead} automation commits conflict with {upstream}") + return git(worktree, "rev-parse", upstream).stdout.strip() + + +def unpushed(worktree: Path) -> int: + if not (worktree / ".git").exists(): + return 0 + proc = git(worktree, "rev-list", "--count", f"{REMOTE}/{TARGET}..HEAD", check=False) + return int(proc.stdout.strip() or 0) if proc.returncode == 0 else 0 + + +def push_batch(worktree: Path, gate_base: str, regate: Callable[[], bool]) -> str: + """Rebase the night's commits on origin/main, gate again if it moved, push.""" + git(worktree, "fetch", "--quiet", REMOTE, TARGET) + upstream = f"{REMOTE}/{TARGET}" + if int(git(worktree, "rev-list", "--count", f"{upstream}..HEAD").stdout) == 0: + return "" + if git(worktree, "merge-base", "--is-ancestor", upstream, "HEAD", + check=False).returncode != 0: + if git(worktree, "rebase", "--quiet", upstream, check=False).returncode != 0: + git(worktree, "rebase", "--abort", check=False) + raise AutorunError(BLOCKED, f"rebase onto {upstream} conflicts, batch kept") + if git(worktree, "rev-parse", upstream).stdout.strip() != gate_base and not regate(): + raise AutorunError(FAILED, f"gate failed on the batch rebased onto {upstream}") + git(worktree, "push", "--quiet", REMOTE, f"HEAD:{TARGET}") + return git(worktree, "rev-parse", "HEAD").stdout.strip() + + +# Circuit breaker + + +def gh_json(args: list[str]) -> object: + """Parsed output of a gh command, or None when gh cannot answer.""" + try: + proc = subprocess.run( + ["gh", *args], capture_output=True, text=True, timeout=120, check=False, + ) + except (OSError, subprocess.TimeoutExpired): + return None + if proc.returncode != 0: + return None + try: + return json.loads(proc.stdout) + except ValueError: + return None + + +RUN_FIELDS = "workflowName,status,conclusion,headSha,createdAt" + + +def ci_verdict(runs: Sequence[dict]) -> str: + """failure as soon as one run is red, success once all ended green.""" + if any(r.get("conclusion") in CI_RED for r in runs): + return "failure" + if not runs or any(r.get("status") != "completed" for r in runs): + return "pending" + if all(r.get("conclusion") in CI_GREEN for r in runs): + return "success" + return "pending" + + +def refresh_breaker(state: dict, gh: Callable[[list[str]], object], now: str) -> None: + """Read CI for the last automation push, open or close the breaker.""" + push = state.setdefault("push", {}) + breaker = state.setdefault("breaker", {"open": False}) + sha = push.get("sha") + if sha and push.get("ci") not in ("success", "failure"): + runs = gh(["run", "list", "-R", REPO_SLUG, "--commit", sha, "--json", RUN_FIELDS]) + if isinstance(runs, list): + verdict = ci_verdict(runs) + push["ci"] = verdict + if verdict == "failure": + red = [r for r in runs if r.get("conclusion") in CI_RED] + state["breaker"] = breaker = { + "open": True, + "sha": sha, + "opened_at": now, + "failed_run_at": max(str(r.get("createdAt") or "") for r in runs), + "reason": ", ".join(sorted({str(r.get("workflowName")) for r in red})), + } + state.setdefault("alerts", {})["ci"] = { + "since": now, "detail": f"red CI on {sha[:12]}: {breaker['reason']}", + } + if not breaker.get("open"): + return + runs = gh([ + "run", "list", "-R", REPO_SLUG, "--branch", TARGET, "--limit", "40", + "--json", RUN_FIELDS, + ]) + if not isinstance(runs, list): + return + after = [ + r for r in runs + if str(r.get("createdAt") or "") > breaker.get("failed_run_at", "") + and r.get("headSha") != breaker.get("sha") + ] + if not after: + return + newest = max(after, key=lambda r: str(r.get("createdAt") or ""))["headSha"] + if ci_verdict([r for r in after if r.get("headSha") == newest]) == "success": + state["breaker"] = {"open": False, "closed_at": now, "closed_by": newest} + state.get("alerts", {}).pop("ci", None) + + +def may_push(state: dict) -> str: + """Empty when a push may go out, else the reason it may not.""" + if state.get("breaker", {}).get("open"): + return "circuit breaker open" + push = state.get("push") or {} + if push.get("sha") and push.get("ci") != "success": + return f"CI of {push['sha'][:12]} has not ended green" + return "" + + +# State and logs + + +def now_iso() -> str: + return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat() + + +def load_state(path: Path) -> dict: + try: + return json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError: + return {} + except ValueError as exc: + raise AutorunError(FAILED, f"{path} is unreadable: {exc}") from exc + + +def save_state(path: Path, state: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + scratch = path.with_suffix(".tmp") + scratch.write_text(json.dumps(state, indent=2, sort_keys=True) + "\n", encoding="utf-8") + os.replace(scratch, path) + + +def prune_logs(logs: Path, days: int = LOG_DAYS, now: float | None = None) -> int: + limit = (now if now is not None else time.time()) - days * 86400 + removed = 0 + for path in logs.glob("*/*.log") if logs.is_dir() else (): + if path.stat().st_mtime < limit: + path.unlink() + removed += 1 + return removed + + +def set_alert(state: dict, name: str, active: bool, detail: str, now: str) -> None: + alerts = state.setdefault("alerts", {}) + if active: + alerts.setdefault(name, {"since": now})["detail"] = detail + else: + alerts.pop(name, None) + + +# Running a job + + +def job_env(job: Job) -> dict[str, str]: + env = os.environ.copy() + env["RETROBIOS_AUTO"] = "1" + if job.network: + sys.path.insert(0, str(REPO_ROOT / "scripts")) + from common import github_token + + token = github_token() + if token: + env["GITHUB_TOKEN"] = token + return env + + +def _remove_scratch(root: Path, rel: str) -> None: + target = (root / rel).resolve() + if root.resolve() not in target.parents: + raise AutorunError(FAILED, f"scratch path {rel} leaves {root}") + if target.is_dir(): + shutil.rmtree(target) + elif target.exists(): + target.unlink() + + +def read_report(root: Path, rel: str) -> object: + """The JSON report a job's step wrote, kept in the job's state.""" + if not rel: + return None + try: + return json.loads((root / rel).read_text(encoding="utf-8")) + except FileNotFoundError: + return None + except ValueError as exc: + return {"unreadable": str(exc)} + + +class Runner: + """One job run: guards, steps, gate, double computation, commits, push.""" + + def __init__(self, ctx: Context, job: Job, state: dict) -> None: + self.ctx = ctx + self.job = job + self.state = state + self.cwd = ctx.worktree if job.worktree else ctx.repo + self.log = ctx.home / "logs" / job.name / f"{dt.date.today().isoformat()}.log" + self.deadline = ctx.host.clock() + job.timeout + self.env: dict[str, str] = {} + self.commits: list[str] = [] + self.retried: list[str] = [] + self.report: object = None + + def note(self, text: str) -> None: + self.log.parent.mkdir(parents=True, exist_ok=True) + with self.log.open("a", encoding="utf-8") as handle: + handle.write(f"# {now_iso()} {text}\n") + + def wait_idle(self) -> None: + self.note("host misread suspected, retrying once on an idle host") + wait_for_load(self.ctx.host, self.job.max_load_wait) + + def run_step(self, step: Step, argv: list[str] | None = None) -> Outcome: + argv = argv if argv is not None else resolve_argv(step.argv) + stdout = self.cwd / step.stdout if step.stdout else None + + def once() -> Outcome: + left = self.deadline - self.ctx.host.clock() + if left <= 0: + return Outcome(124, "job timeout reached") + return execute(argv, self.cwd, self.log, self.env, left, stdout) + + outcome = run_with_retry(once, step.ok_codes, self.wait_idle) + if outcome.retried: + self.retried.append(shlex.join(argv)) + return outcome + + def run_all(self, steps: Sequence[Step], what: str) -> None: + for step in steps: + outcome = self.run_step(step) + if outcome.returncode not in step.ok_codes: + raise AutorunError( + FAILED, f"{what}: {shlex.join(step.argv)} exited {outcome.returncode}" + ) + + def regate(self) -> bool: + """The gate on the rebased batch. What it regenerates is computed + twice and committed under the same rules before the push.""" + try: + self.run_all(self.job.gate, "gate") + self.check_twice() + self.commit() + except AutorunError as exc: + self.note(f"gate on the rebased batch: {exc.detail}") + return False + return True + + def check_twice(self) -> None: + """Each shipped artifact the gate wrote, computed again and compared.""" + scratch = self.cwd / "tmp" / "autorun" / "twice" + for index, item in enumerate(self.job.twice): + artifact = self.cwd / item.artifact + for attempt in (1, 2): + slot = scratch / f"{index}-{attempt}" + if slot.exists(): + _remove_scratch(self.cwd, slot.relative_to(self.cwd).as_posix()) + if item.glob or artifact.is_dir(): + slot.parent.mkdir(parents=True, exist_ok=True) + out = slot + else: + slot.mkdir(parents=True) + out = slot / artifact.name + rel = out.relative_to(self.cwd).as_posix() + outcome = self.run_step(Step(item.argv), resolve_argv(item.argv, rel)) + if outcome.returncode != 0: + raise AutorunError(FAILED, f"recomputing {item.artifact} exited " + f"{outcome.returncode}") + diff = differences(artifact, out, item.glob) + if not diff: + break + self.note(f"{item.artifact}: second computation differs on " + f"{', '.join(diff[:10])}") + if attempt == 1: + wait_for_load(self.ctx.host, self.job.max_load_wait) + continue + set_alert(self.state, "host-integrity", True, + f"{item.artifact} computed twice differs", now_iso()) + raise AutorunError(FAILED, f"{item.artifact}: two computations disagree " + f"on {', '.join(diff[:5])}") + set_alert(self.state, "host-integrity", False, "", now_iso()) + + def commit(self) -> None: + changed = changed_files(self.cwd) + for rule in self.job.commits: + files = plan_commit(changed, rule.paths) + if not files: + continue + sha = commit_paths(self.cwd, files, rule) + self.commits.append(sha) + self.note(f"committed {sha[:12]} {rule.subject} ({len(files)} files)") + leftovers = changed_files(self.cwd) + if leftovers: + raise AutorunError( + FAILED, f"{len(leftovers)} paths changed outside every commit rule: " + + ", ".join(leftovers[:5]) + ) + + def execute(self) -> tuple[str, str]: + ctx, job = self.ctx, self.job + free = free_gb(ctx.repo) + set_alert(self.state, "disk", free < ALERT_FREE_GB, + f"{free:.0f} GB free, under {ALERT_FREE_GB} GB", now_iso()) + if free < job.min_free_gb: + return REFUSED, f"{free:.0f} GB free, {job.name} needs {job.min_free_gb:.0f} GB" + if not wait_for_load(ctx.host, job.max_load_wait): + return DEFERRED, f"load stayed above {ctx.host.max_load:.1f}" + if job.commits and ctx.gh is not None: + refresh_breaker(self.state, ctx.gh, now_iso()) + if job.commits and self.state.get("breaker", {}).get("open"): + return BLOCKED, "circuit breaker open: a pushed automation commit failed CI" + self.env = job_env(job) + base = "" + if job.worktree: + ensure_worktree(ctx.repo, ctx.worktree) + seeded = seed_worktree(ctx.repo, ctx.worktree) + if seeded: + self.note(f"hard-linked {', '.join(seeded[:5])}") + base = sync_worktree(ctx.worktree) + try: + self.run_all(job.steps, "step") + self.run_all(job.gate, "gate") + self.check_twice() + if job.commits: + self.commit() + if ctx.push and job.commits: + reason = may_push(self.state) + if reason: + return OK, f"{len(self.commits)} commits, push held: {reason}" + pushed = push_batch(ctx.worktree, base, self.regate) + if pushed: + self.state["push"] = {"sha": pushed, "at": now_iso(), "ci": "pending"} + return OK, f"pushed {pushed[:12]}" + finally: + self.report = read_report(self.cwd, job.report) + for rel in job.scratch: + _remove_scratch(self.cwd, rel) + return OK, f"{len(self.commits)} commits" + + +def describe(job: Job, ctx: Context) -> list[str]: + """What a run would do, for --dry-run.""" + lines = [f"job {job.name} in {ctx.worktree if job.worktree else ctx.repo}"] + lines += [f" step {shlex.join(s.argv)}" for s in job.steps] + lines += [f" gate {shlex.join(s.argv)}" for s in job.gate] + lines += [f" twice {t.artifact}: {shlex.join(t.argv)}" for t in job.twice] + changed = changed_files(ctx.worktree) if (ctx.worktree / ".git").exists() else [] + for rule in job.commits: + files = plan_commit(changed, rule.paths) + lines.append(f" commit {rule.subject!r} paths {', '.join(rule.paths)}") + lines += [f" would commit {f}" for f in files] + lines += [f" scratch {s}" for s in job.scratch] + return lines + + +def run_job(ctx: Context, name: str) -> tuple[str, str]: + """Run one job and record it, returning (result, detail).""" + job = ctx.jobs.get(name) + if job is None: + raise SystemExit(f"unknown job {name!r}: {', '.join(sorted(ctx.jobs))}") + if ctx.dry_run: + print("\n".join(describe(job, ctx))) + return OK, "dry run" + state = load_state(ctx.state_path) + runner = Runner(ctx, job, state) + started, clock = now_iso(), ctx.host.clock() + lock = global_lock(ctx.lock_path) if job.lock else contextlib.nullcontext() + try: + with lock: + runner.note(f"start {name}") + try: + result, detail = runner.execute() + except AutorunError as exc: + result, detail = exc.result, exc.detail + except OSError as exc: + result, detail = FAILED, f"{type(exc).__name__}: {exc}" + except LockBusy as exc: + result, detail = DEFERRED, str(exc) + runner.note(f"end {name}: {result} {detail}") + previous = state.setdefault("jobs", {}).get(name, {}) + record = { + "last_start": started, + "last_end": now_iso(), + "duration_sec": round(ctx.host.clock() - clock, 1), + "result": result, + "detail": detail, + "commits": runner.commits, + "retried": runner.retried, + "report": runner.report, + "log": str(runner.log), + "last_success": started if result == OK else previous.get("last_success"), + "consecutive_failures": 0 if result in (OK, DEFERRED) + else previous.get("consecutive_failures", 0) + 1, + } + state["jobs"][name] = record + prune_logs(ctx.home / "logs") + save_state(ctx.state_path, state) + return result, detail + + +# Status + + +def status(ctx: Context) -> dict: + state = load_state(ctx.state_path) + return { + "jobs": { + name: {"schedule": job.schedule, **state.get("jobs", {}).get(name, {})} + for name, job in sorted(ctx.jobs.items()) + }, + "breaker": state.get("breaker", {"open": False}), + "push": state.get("push", {}), + "alerts": state.get("alerts", {}), + "lock": {"path": str(ctx.lock_path), "busy": lock_busy(ctx.lock_path)}, + "free_gb": round(free_gb(ctx.repo), 1), + "worktree": { + "path": str(ctx.worktree), + "exists": (ctx.worktree / ".git").exists(), + "unpushed": unpushed(ctx.worktree), + }, + } + + +def render_status(report: dict) -> str: + lines = [] + for name, job in report["jobs"].items(): + lines.append( + f"{name:12} {job.get('result', 'never run'):9} {job.get('last_end', ''):25} " + f"{job.get('detail', '')}" + ) + counts = (job.get("report") or {}).get("counts") if isinstance( + job.get("report"), dict) else None + if counts: + lines.append(f"{'':12} " + ", ".join(f"{n} {v}" for v, n in sorted(counts.items()))) + wt = report["worktree"] + lines.append(f"worktree {wt['path']} ({'present' if wt['exists'] else 'absent'}, " + f"{wt['unpushed']} unpushed commits)") + lines.append(f"lock {report['lock']['path']} ({'busy' if report['lock']['busy'] else 'free'})") + lines.append(f"disk {report['free_gb']} GB free") + breaker = report["breaker"] + lines.append("breaker " + (f"OPEN since {breaker.get('opened_at')}: {breaker.get('reason')}" + if breaker.get("open") else "closed")) + for name, alert in sorted(report["alerts"].items()): + lines.append(f"ALERT {name}: {alert.get('detail', '')} (since {alert.get('since')})") + return "\n".join(lines) + + +# Timers + + +def render_units( + jobs: dict[str, Job], repo: Path, python: str, push: bool = False +) -> dict[str, str]: + """systemd user units: one service template and a timer per scheduled job.""" + flags = " --push" if push else "" + units = { + "retrobios@.service": ( + "[Unit]\n" + "Description=retrobios automation job %i\n" + "\n" + "[Service]\n" + "Type=oneshot\n" + f"WorkingDirectory={repo}\n" + "Environment=RETROBIOS_AUTO=1\n" + f"ExecStart={python} {repo / 'scripts' / 'autorun.py'} run %i{flags}\n" + "TimeoutStartSec=6h\n" + "Nice=10\n" + "IOSchedulingClass=idle\n" + "SuccessExitStatus=75\n" + ) + } + for name, job in sorted(jobs.items()): + if not job.schedule: + continue + units[f"retrobios-{name}.timer"] = ( + "[Unit]\n" + f"Description=retrobios automation: {name}\n" + "\n" + "[Timer]\n" + f"OnCalendar={job.schedule}\n" + "Persistent=true\n" + f"Unit=retrobios@{name}.service\n" + "\n" + "[Install]\n" + "WantedBy=timers.target\n" + ) + return units + + +def install_timers( + unit_dir: Path, + units: dict[str, str], + enable: bool, + systemctl: Callable[[list[str]], int] | None = None, +) -> list[str]: + """Write the units, and enable the timers only when asked.""" + unit_dir.mkdir(parents=True, exist_ok=True) + for name, text in units.items(): + (unit_dir / name).write_text(text, encoding="utf-8") + timers = sorted(n for n in units if n.endswith(".timer")) + if enable: + call = systemctl or (lambda args: subprocess.run( + ["systemctl", "--user", *args], check=False).returncode) + for args in (["daemon-reload"], ["enable", "--now", *timers]): + if call(args) != 0: + raise AutorunError(FAILED, f"systemctl --user {' '.join(args)} failed") + return timers + + +# Command line + + +def main_checkout() -> Path: + """The main checkout, also when this script runs from a linked worktree.""" + proc = git(REPO_ROOT, "rev-parse", "--path-format=absolute", "--git-common-dir", + check=False) + common = Path(proc.stdout.strip()) if proc.returncode == 0 else REPO_ROOT / ".git" + return common.parent + + +def main() -> int: + parser = argparse.ArgumentParser( + allow_abbrev=False, description=__doc__.split("\n\n")[0], + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + sub = parser.add_subparsers(dest="command", required=True) + run_p = sub.add_parser("run", help="run one job from automation/jobs.yml") + run_p.add_argument("job", help="a job name from automation/jobs.yml") + run_p.add_argument("--dry-run", action="store_true", + help="print the commands and the files each commit would take") + run_p.add_argument("--push", action="store_true", + help="push the committed batch to origin/main (off by default)") + status_p = sub.add_parser("status", help="last result of every job, alerts, breaker") + status_p.add_argument("--json", action="store_true", help="machine-readable status") + timers_p = sub.add_parser("install-timers", help="write systemd user units") + timers_p.add_argument("--enable", action="store_true", + help="also reload systemd and enable the timers") + timers_p.add_argument("--unit-dir", type=Path, + default=Path.home() / ".config" / "systemd" / "user", + help="where the units go (default: %(default)s)") + timers_p.add_argument("--with-push", action="store_true", + help="the units run jobs with --push") + args = parser.parse_args() + + try: + jobs = load_jobs() + except (OSError, ValueError, KeyError) as exc: + print(f"ERROR: automation/jobs.yml: {exc}", file=sys.stderr) + return 2 + repo = main_checkout() + ctx = Context( + repo=repo, + jobs=jobs, + lock_path=Path(os.environ.get(LOCK_ENV) or DEFAULT_LOCK), + gh=gh_json, + ) + if args.command == "run": + ctx.dry_run, ctx.push = args.dry_run, args.push + result, detail = run_job(ctx, args.job) + print(f"{args.job}: {result} ({detail})") + return EXIT_CODES.get(result, 1) + if args.command == "status": + report = status(ctx) + print(json.dumps(report, indent=2) if args.json else render_status(report)) + return 0 + units = render_units(jobs, repo, sys.executable, push=args.with_push) + timers = install_timers(args.unit_dir, units, args.enable) + print(f"wrote {len(units)} units to {args.unit_dir}") + if not args.enable: + print("enable with: python scripts/autorun.py install-timers --enable") + print(" (systemctl --user daemon-reload && systemctl --user enable --now " + + " ".join(timers) + ")") + return 0 + + +if __name__ == "__main__": + sys.path.insert(0, str(REPO_ROOT / "scripts")) + from common import run_cli + + run_cli(main) diff --git a/tests/test_autorun.py b/tests/test_autorun.py new file mode 100644 index 000000000..72bd0687e --- /dev/null +++ b/tests/test_autorun.py @@ -0,0 +1,838 @@ +"""autorun: the runner every unattended job goes through, and its first jobs. + +Everything here is offline. Jobs are fake commands, git repositories are +built in tmp/tests with a bare repository standing in for origin, and gh is +a function returning canned CI runs. +""" + +from __future__ import annotations + +import contextlib +import io +import json +import os +import shutil +import subprocess +import sys +import tempfile +import time +import unittest +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT / "scripts")) + +import autorun as ar # noqa: E402 +import bios_integrity as bib # noqa: E402 +import cleanup_disk as cd # noqa: E402 +import sync_release_assets as sra # noqa: E402 + +PY = "python" +_ENV_KEYS = ("GIT_CONFIG_GLOBAL", "GIT_CONFIG_NOSYSTEM") +_saved_env: dict[str, str | None] = {} +_config_dir: Path | None = None + + +def setUpModule() -> None: + global _config_dir + _config_dir = Path(tempfile.mkdtemp(dir=REPO_ROOT / "tmp" / "tests")) + config = _config_dir / "gitconfig" + config.write_text( + "[user]\n\tname = Test\n\temail = test@example.invalid\n" + "[init]\n\tdefaultBranch = main\n" + "[commit]\n\tgpgsign = false\n", + encoding="utf-8", + ) + for key in _ENV_KEYS: + _saved_env[key] = os.environ.get(key) + os.environ["GIT_CONFIG_GLOBAL"] = str(config) + os.environ["GIT_CONFIG_NOSYSTEM"] = "1" + + +def tearDownModule() -> None: + for key, value in _saved_env.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + if _config_dir is not None: + shutil.rmtree(_config_dir, ignore_errors=True) + + +def run_git(cwd: Path, *args: str) -> str: + return subprocess.run( + ["git", "-C", str(cwd), *args], capture_output=True, text=True, check=True, + ).stdout + + +class FakeHost(ar.Host): + def __init__(self, loads: list[float] | None = None, cpus: int = 4) -> None: + self.loads = list(loads or [0.0]) + self.now = 0.0 + self.slept: list[float] = [] + super().__init__( + loadavg=self._load, sleep=self._sleep, clock=lambda: self.now, + cpu_count=cpus, poll=10.0, + ) + + def _load(self) -> tuple[float, float, float]: + value = self.loads.pop(0) if len(self.loads) > 1 else self.loads[0] + return (value, value, value) + + def _sleep(self, seconds: float) -> None: + self.slept.append(seconds) + self.now += seconds + + +class Scratch(unittest.TestCase): + def setUp(self) -> None: + self.tmp = Path(tempfile.mkdtemp(dir=REPO_ROOT / "tmp" / "tests")) + self.addCleanup(shutil.rmtree, self.tmp, True) + + def context(self, jobs: dict, **kwargs) -> ar.Context: + repo = self.tmp / "repo" + repo.mkdir(exist_ok=True) + parsed = ar.parse_jobs({"jobs": jobs}) + kwargs.setdefault("host", FakeHost()) + return ar.Context(repo=repo, jobs=parsed, lock_path=self.tmp / "auto.lock", **kwargs) + + def state(self, ctx: ar.Context) -> dict: + return json.loads(ctx.state_path.read_text(encoding="utf-8")) + + +class Registry(Scratch): + def test_the_committed_registry_loads(self): + jobs = ar.load_jobs() + self.assertIn("nightly", jobs) + self.assertTrue(jobs["nightly"].commits) + self.assertFalse(jobs["disk-check"].lock) + + def test_an_unknown_key_is_refused(self): + with self.assertRaises(ValueError): + ar.parse_jobs({"jobs": {"x": {"steps": [], "schedul": "daily"}}}) + + def test_a_subject_the_hook_refuses_is_refused_at_load(self): + for subject in ("chore: fix the broken manifests", + "update things", + "chore: " + "a" * 60): + with self.subTest(subject=subject), self.assertRaises(ValueError): + ar.parse_jobs({"jobs": {"x": {"commits": [ + {"subject": subject, "paths": ["a"]}]}}}) + + def test_a_job_outside_the_worktree_cannot_commit(self): + with self.assertRaises(ValueError): + ar.parse_jobs({"jobs": {"x": {"worktree": False, "commits": [ + {"subject": "chore: add a", "paths": ["a"]}]}}}) + + def test_a_twice_entry_names_its_output(self): + with self.assertRaises(ValueError): + ar.parse_jobs({"jobs": {"x": {"twice": [ + {"artifact": "a.json", "run": [PY, "gen.py"]}]}}}) + + +class Lock(Scratch): + def test_a_second_holder_is_refused(self): + path = self.tmp / "auto.lock" + with ar.global_lock(path): + self.assertTrue(ar.lock_busy(path)) + with self.assertRaises(ar.LockBusy), ar.global_lock(path): + pass + self.assertFalse(ar.lock_busy(path)) + + def test_a_job_waiting_on_the_lock_is_deferred_and_recorded(self): + ctx = self.context({"x": {"worktree": False, "min_free_gb": 0, + "steps": [[PY, "-c", "open('ran', 'w')"]]}}) + with ar.global_lock(ctx.lock_path): + result, detail = ar.run_job(ctx, "x") + self.assertEqual(result, ar.DEFERRED) + self.assertIn("held", detail) + self.assertFalse((ctx.repo / "ran").exists()) + self.assertEqual(self.state(ctx)["jobs"]["x"]["result"], ar.DEFERRED) + + def test_a_job_that_takes_no_lock_runs_beside_a_holder(self): + ctx = self.context({"x": {"worktree": False, "lock": False, "min_free_gb": 0, + "steps": [[PY, "-c", "open('ran', 'w')"]]}}) + with ar.global_lock(ctx.lock_path): + self.assertEqual(ar.run_job(ctx, "x")[0], ar.OK) + self.assertTrue((ctx.repo / "ran").exists()) + + +class Guards(Scratch): + def test_the_load_guard_waits_then_lets_go(self): + host = FakeHost(loads=[3.0, 3.0, 0.5], cpus=4) + self.assertTrue(ar.wait_for_load(host, max_wait=600)) + self.assertEqual(host.slept, [10.0, 10.0]) + + def test_the_load_guard_gives_up_at_its_deadline(self): + host = FakeHost(loads=[8.0], cpus=4) + self.assertFalse(ar.wait_for_load(host, max_wait=25)) + self.assertEqual(sum(host.slept), 25) + + def test_a_busy_host_defers_the_job(self): + ctx = self.context({"x": {"worktree": False, "min_free_gb": 0, "max_load_wait": 0, + "steps": [[PY, "-c", "open('ran', 'w')"]]}}, + host=FakeHost(loads=[64.0])) + self.assertEqual(ar.run_job(ctx, "x")[0], ar.DEFERRED) + self.assertFalse((ctx.repo / "ran").exists()) + + def test_a_full_disk_refuses_the_job(self): + ctx = self.context({"x": {"worktree": False, "min_free_gb": 10 ** 9, + "steps": [[PY, "-c", "open('ran', 'w')"]]}}) + result, detail = ar.run_job(ctx, "x") + self.assertEqual(result, ar.REFUSED) + self.assertIn("GB free", detail) + self.assertFalse((ctx.repo / "ran").exists()) + + def test_a_failure_counts_and_a_success_resets(self): + ctx = self.context({ + "bad": {"worktree": False, "min_free_gb": 0, "steps": [[PY, "-c", "raise SystemExit(3)"]]}, + }) + ar.run_job(ctx, "bad") + ar.run_job(ctx, "bad") + record = self.state(ctx)["jobs"]["bad"] + self.assertEqual(record["result"], ar.FAILED) + self.assertEqual(record["consecutive_failures"], 2) + self.assertIsNone(record["last_success"]) + + +class Retry(Scratch): + def _runs(self, *outcomes: ar.Outcome): + queue = list(outcomes) + calls = [] + + def run() -> ar.Outcome: + calls.append(1) + return queue.pop(0) + return run, calls + + def test_a_segfault_is_retried_once_on_an_idle_host(self): + run, calls = self._runs(ar.Outcome(139), ar.Outcome(0)) + waited = [] + outcome = ar.run_with_retry(run, (0,), lambda: waited.append(1)) + self.assertEqual((outcome.returncode, outcome.retried, len(calls), waited), + (0, True, 2, [1])) + + def test_a_second_segfault_is_a_failure(self): + run, calls = self._runs(ar.Outcome(-11), ar.Outcome(-11), ar.Outcome(0)) + outcome = ar.run_with_retry(run, (0,), lambda: None) + self.assertEqual((outcome.returncode, len(calls)), (-11, 2)) + + def test_a_host_misread_message_is_retried(self): + run, calls = self._runs(ar.Outcome(1, "SystemError: unknown opcode"), ar.Outcome(0)) + self.assertEqual(ar.run_with_retry(run, (0,), lambda: None).returncode, 0) + self.assertEqual(len(calls), 2) + + def test_an_ordinary_failure_is_not_retried(self): + run, calls = self._runs(ar.Outcome(1, "KeyError: 'x'"), ar.Outcome(0)) + self.assertEqual(ar.run_with_retry(run, (0,), lambda: None).returncode, 1) + self.assertEqual(len(calls), 1) + + def test_a_real_segfault_runs_again_and_the_job_passes(self): + crash = ("import os, signal, pathlib\n" + "p = pathlib.Path('crashed')\n" + "if not p.exists():\n" + " p.touch()\n" + " os.kill(os.getpid(), signal.SIGSEGV)\n") + ctx = self.context({"x": {"worktree": False, "min_free_gb": 0, + "steps": [[PY, "-c", crash]]}}) + self.assertEqual(ar.run_job(ctx, "x")[0], ar.OK) + self.assertEqual(len(self.state(ctx)["jobs"]["x"]["retried"]), 1) + + +class DoubleComputation(Scratch): + GEN = ("import json, random, sys\n" + "json.dump({'generated_at': str(random.random()), 'value': %s}," + " open(sys.argv[1], 'w'))\n") + + def _run(self, cwd: Path): + def run(argv): + return ar.Outcome(subprocess.run(argv, cwd=cwd, check=False).returncode) + return run + + def test_timestamps_aside_two_equal_runs_agree(self): + script = self.tmp / "gen.py" + script.write_text(self.GEN % "1", encoding="utf-8") + diff = ar.twice([PY, str(script), "{out}"], self.tmp / "a.json", + self.tmp / "b.json", self._run(self.tmp)) + self.assertEqual(diff, []) + + def test_two_runs_that_disagree_are_named(self): + script = self.tmp / "gen.py" + script.write_text(self.GEN % "random.random()", encoding="utf-8") + diff = ar.twice([PY, str(script), "{out}"], self.tmp / "a.json", + self.tmp / "b.json", self._run(self.tmp)) + self.assertEqual(diff, ["a.json"]) + + def test_directories_compare_file_by_file_within_the_glob(self): + a, b = self.tmp / "a", self.tmp / "b" + (a / "targets").mkdir(parents=True) + b.mkdir() + (a / "x.json").write_text('{"generated": "1", "n": 1}') + (b / "x.json").write_text('{"generated": "2", "n": 1}') + (a / "targets" / "t.json").write_text("{}") + self.assertEqual(ar.differences(a, b, "*.json"), []) + (b / "y.json").write_text("{}") + self.assertEqual(ar.differences(a, b, "*.json"), ["y.json"]) + + def test_a_job_whose_artifact_disagrees_fails_with_an_alert(self): + script = self.tmp / "repo" / "gen.py" + script.parent.mkdir(parents=True, exist_ok=True) + script.write_text(self.GEN % "random.random()", encoding="utf-8") + ctx = self.context({"x": { + "worktree": False, "min_free_gb": 0, "max_load_wait": 0, + "gate": [[PY, "gen.py", "db.json"]], + "twice": [{"artifact": "db.json", "run": [PY, "gen.py", "{out}"]}], + }}) + result, detail = ar.run_job(ctx, "x") + self.assertEqual(result, ar.FAILED) + self.assertIn("two computations disagree", detail) + self.assertIn("host-integrity", self.state(ctx)["alerts"]) + + def test_a_job_whose_artifact_agrees_passes_and_clears_the_alert(self): + script = self.tmp / "repo" / "gen.py" + script.parent.mkdir(parents=True, exist_ok=True) + script.write_text(self.GEN % "1", encoding="utf-8") + ctx = self.context({"x": { + "worktree": False, "min_free_gb": 0, + "gate": [[PY, "gen.py", "db.json"]], + "twice": [{"artifact": "db.json", "run": [PY, "gen.py", "{out}"]}], + }}) + ar.save_state(ctx.state_path, {"alerts": {"host-integrity": {"since": "x"}}}) + self.assertEqual(ar.run_job(ctx, "x")[0], ar.OK) + self.assertNotIn("host-integrity", self.state(ctx).get("alerts", {})) + + +def ci_run(sha: str, conclusion: str | None, created: str, name: str = "Deploy Site") -> dict: + return {"workflowName": name, "headSha": sha, "createdAt": created, + "status": "completed" if conclusion else "in_progress", + "conclusion": conclusion or ""} + + +class CircuitBreaker(Scratch): + def _gh(self, by_commit: dict, on_main: list): + calls = [] + + def gh(args): + calls.append(args) + if "--commit" in args: + return by_commit.get(args[args.index("--commit") + 1], []) + return on_main + return gh, calls + + def test_a_red_run_on_the_pushed_commit_opens_it(self): + state = {"push": {"sha": "a" * 40, "ci": "pending"}} + gh, _ = self._gh({"a" * 40: [ci_run("a" * 40, "success", "2026-10-27T02:10:00Z"), + ci_run("a" * 40, "failure", "2026-10-27T02:10:00Z", + "Validate")]}, []) + ar.refresh_breaker(state, gh, "now") + self.assertTrue(state["breaker"]["open"]) + self.assertEqual(state["breaker"]["reason"], "Validate") + self.assertIn("ci", state["alerts"]) + self.assertEqual(ar.may_push(state), "circuit breaker open") + + def test_runs_still_going_keep_the_push_pending(self): + state = {"push": {"sha": "a" * 40, "ci": "pending"}} + gh, _ = self._gh({"a" * 40: [ci_run("a" * 40, None, "2026-10-27T02:10:00Z")]}, []) + ar.refresh_breaker(state, gh, "now") + self.assertFalse(state["breaker"]["open"]) + self.assertIn("has not ended green", ar.may_push(state)) + + def test_a_later_green_run_on_main_closes_it(self): + state = {"push": {"sha": "a" * 40, "ci": "failure"}, + "breaker": {"open": True, "sha": "a" * 40, + "failed_run_at": "2026-10-27T02:10:00Z"}, + "alerts": {"ci": {"since": "x"}}} + pending = [ci_run("b" * 40, None, "2026-10-27T09:00:00Z")] + gh, _ = self._gh({}, pending + [ci_run("a" * 40, "failure", "2026-10-27T02:10:00Z")]) + ar.refresh_breaker(state, gh, "now") + self.assertTrue(state["breaker"]["open"]) + green = [ci_run("c" * 40, "success", "2026-10-27T10:00:00Z"), + ci_run("c" * 40, "skipped", "2026-10-27T10:00:01Z", "Validate")] + gh, _ = self._gh({}, green + pending) + ar.refresh_breaker(state, gh, "later") + self.assertFalse(state["breaker"]["open"]) + self.assertNotIn("ci", state["alerts"]) + + def test_an_open_breaker_blocks_a_committing_job_before_any_git(self): + gh, _ = self._gh({"a" * 40: [ci_run("a" * 40, "failure", "t")]}, []) + ctx = self.context({"x": {"min_free_gb": 0, "commits": [ + {"subject": "chore: add a", "paths": ["a"]}]}}, gh=gh) + ar.save_state(ctx.state_path, {"push": {"sha": "a" * 40, "ci": "pending"}}) + result, _ = ar.run_job(ctx, "x") + self.assertEqual(result, ar.BLOCKED) + self.assertFalse(ctx.worktree.exists()) + + +class Origin(Scratch): + """A bare origin, a main checkout cloned from it, and a second clone + standing for another contributor.""" + + def setUp(self) -> None: + super().setUp() + self.origin = self.tmp / "origin.git" + run_git(self.tmp, "init", "--quiet", "--bare", "-b", "main", str(self.origin)) + self.repo = self.tmp / "repo" + run_git(self.tmp, "clone", "--quiet", str(self.origin), str(self.repo)) + (self.repo / ".gitignore").write_text(".cache/\ntmp/\n", encoding="utf-8") + (self.repo / "database.json").write_text("{}\n", encoding="utf-8") + (self.repo / "install").mkdir() + (self.repo / "install" / "a.json").write_text("{}\n", encoding="utf-8") + run_git(self.repo, "add", ".gitignore", "database.json", "install/a.json") + run_git(self.repo, "commit", "--quiet", "-m", "chore: start") + run_git(self.repo, "push", "--quiet", "origin", "main") + + def other_commit(self, name: str) -> None: + other = self.tmp / "other" + if not other.exists(): + run_git(self.tmp, "clone", "--quiet", str(self.origin), str(other)) + run_git(other, "pull", "--quiet", "origin", "main") + (other / name).write_text(name, encoding="utf-8") + run_git(other, "add", name) + run_git(other, "commit", "--quiet", "-m", f"chore: add {name}") + run_git(other, "push", "--quiet", "origin", "main") + + def worktree(self) -> Path: + wt = self.repo / ".cache" / "autorun" / "worktree" + ar.ensure_worktree(self.repo, wt) + return wt + + +class Commits(Origin): + def test_globs_stay_inside_one_segment(self): + self.assertTrue(ar.path_matches("install/*.json", "install/a.json")) + self.assertFalse(ar.path_matches("install/*.json", "install/targets/a.json")) + self.assertFalse(ar.path_matches("install/*.json", "x/install/a.json")) + + def test_the_dry_run_names_only_matching_files_and_stages_nothing(self): + wt = self.worktree() + (wt / "database.json").write_text('{"n": 1}\n', encoding="utf-8") + (wt / "install" / "b.json").write_text("{}\n", encoding="utf-8") + (wt / "notes.txt").write_text("x", encoding="utf-8") + ctx = ar.Context(repo=self.repo, lock_path=self.tmp / "l", dry_run=True, + jobs=ar.parse_jobs({"jobs": {"n": {"commits": [{ + "subject": "chore: regenerate collection records", + "paths": ["database.json", "install/*.json"]}]}}})) + with contextlib.redirect_stdout(io.StringIO()): + self.assertEqual(ar.run_job(ctx, "n"), (ar.OK, "dry run")) + lines = ar.describe(ctx.jobs["n"], ctx) + would = sorted(l.split("would commit ")[1] for l in lines if "would commit" in l) + self.assertEqual(would, ["database.json", "install/b.json"]) + self.assertEqual(run_git(wt, "diff", "--cached", "--name-only"), "") + self.assertFalse(ctx.state_path.exists()) + + def test_a_commit_takes_its_paths_and_nothing_staged_beside(self): + wt = self.worktree() + (wt / "database.json").write_text('{"n": 1}\n', encoding="utf-8") + (wt / "other.txt").write_text("x", encoding="utf-8") + run_git(wt, "add", "other.txt") + rule = ar.CommitRule("chore: regenerate collection records", ("database.json",)) + files = ar.plan_commit(ar.changed_files(wt), rule.paths) + ar.commit_paths(wt, files, rule) + shown = run_git(wt, "show", "--name-only", "--format=%s", "HEAD").split() + self.assertEqual(shown[-1], "database.json") + self.assertNotIn("other.txt", shown) + self.assertEqual(run_git(wt, "diff", "--cached", "--name-only").split(), ["other.txt"]) + + +class Worktree(Origin): + def test_it_is_created_on_its_own_branch_and_reused(self): + wt = self.worktree() + self.assertEqual(run_git(wt, "branch", "--show-current").strip(), ar.BRANCH) + self.assertEqual(run_git(self.repo, "branch", "--show-current").strip(), "main") + ar.ensure_worktree(self.repo, wt) + self.assertEqual(len(run_git(self.repo, "worktree", "list").splitlines()), 2) + + def test_it_follows_origin_and_stops_on_leftovers(self): + wt = self.worktree() + self.other_commit("x.txt") + base = ar.sync_worktree(wt) + self.assertTrue((wt / "x.txt").exists()) + self.assertEqual(base, run_git(wt, "rev-parse", "HEAD").strip()) + (wt / "database.json").write_text("changed", encoding="utf-8") + with self.assertRaises(ar.AutorunError) as caught: + ar.sync_worktree(wt) + self.assertEqual(caught.exception.result, ar.BLOCKED) + self.assertEqual((wt / "database.json").read_text(), "changed") + + def test_the_hard_links_share_bytes_with_the_main_checkout(self): + (self.repo / ".gitignore").write_text(".cache/\ntmp/\nbios/big.bin\n") + (self.repo / "bios").mkdir() + (self.repo / "bios" / "big.bin").write_bytes(b"x" * 10) + (self.repo / ".cache" / "large").mkdir(parents=True) + (self.repo / ".cache" / "large" / "a.zip").write_bytes(b"z") + wt = self.worktree() + seeded = ar.seed_worktree(self.repo, wt) + self.assertEqual(sorted(seeded), [".cache/large", "bios/big.bin"]) + self.assertTrue(os.path.samefile(wt / "bios" / "big.bin", self.repo / "bios" / "big.bin")) + self.assertEqual(ar.seed_worktree(self.repo, wt), []) + + +class Push(Origin): + def _commit(self, wt: Path) -> None: + (wt / "database.json").write_text('{"n": 2}\n', encoding="utf-8") + ar.commit_paths(wt, ["database.json"], + ar.CommitRule("chore: regenerate collection records", ("database.json",))) + + def test_a_batch_on_an_unmoved_origin_pushes_without_a_second_gate(self): + wt = self.worktree() + base = ar.sync_worktree(wt) + self._commit(wt) + gates = [] + pushed = ar.push_batch(wt, base, lambda: gates.append(1) or True) + self.assertEqual(gates, []) + self.assertEqual(run_git(self.origin, "rev-parse", "main").strip(), pushed) + + def test_a_moved_origin_rebases_and_gates_again(self): + wt = self.worktree() + base = ar.sync_worktree(wt) + self._commit(wt) + self.other_commit("y.txt") + gates = [] + pushed = ar.push_batch(wt, base, lambda: gates.append(1) or True) + self.assertEqual(gates, [1]) + log = run_git(self.origin, "log", "--format=%s", "main").splitlines() + self.assertEqual(log[:2], ["chore: regenerate collection records", "chore: add y.txt"]) + self.assertEqual(run_git(self.origin, "rev-parse", "main").strip(), pushed) + + def test_a_failed_second_gate_pushes_nothing(self): + wt = self.worktree() + base = ar.sync_worktree(wt) + self._commit(wt) + self.other_commit("y.txt") + before = run_git(self.origin, "rev-parse", "main") + with self.assertRaises(ar.AutorunError): + ar.push_batch(wt, base, lambda: False) + self.assertEqual(run_git(self.origin, "rev-parse", "main"), before) + + def test_a_conflicting_batch_is_kept_and_nothing_pushed(self): + wt = self.worktree() + base = ar.sync_worktree(wt) + self._commit(wt) + other = self.tmp / "other" + run_git(self.tmp, "clone", "--quiet", str(self.origin), str(other)) + (other / "database.json").write_text('{"n": 3}\n', encoding="utf-8") + run_git(other, "commit", "--quiet", "-am", "chore: other database") + run_git(other, "push", "--quiet", "origin", "main") + mine = run_git(wt, "rev-parse", "HEAD") + with self.assertRaises(ar.AutorunError) as caught: + ar.push_batch(wt, base, lambda: True) + self.assertEqual(caught.exception.result, ar.BLOCKED) + self.assertEqual(run_git(wt, "rev-parse", "HEAD"), mine) + self.assertEqual(ar.changed_files(wt), []) + + +class RunnerPush(Origin): + def _ctx(self, value: int, push: bool) -> ar.Context: + write = "open('database.json', 'w').write('{\"n\": %d}')" % value + jobs = {"n": {"min_free_gb": 0, "steps": [[PY, "-c", write]], "commits": [ + {"subject": "chore: regenerate collection records", "paths": ["database.json"]}]}} + return ar.Context(repo=self.repo, jobs=ar.parse_jobs({"jobs": jobs}), + lock_path=self.tmp / "auto.lock", host=FakeHost(), + gh=lambda _args: [], push=push) + + def test_without_push_the_commit_stays_local(self): + before = run_git(self.origin, "rev-parse", "main") + result, detail = ar.run_job(self._ctx(5, push=False), "n") + self.assertEqual((result, detail), (ar.OK, "1 commits")) + self.assertEqual(run_git(self.origin, "rev-parse", "main"), before) + self.assertEqual(ar.unpushed(self.repo / ".cache" / "autorun" / "worktree"), 1) + + def test_a_push_waits_for_the_previous_push_to_end_green(self): + ctx = self._ctx(5, push=True) + result, detail = ar.run_job(ctx, "n") + self.assertEqual(result, ar.OK) + self.assertIn("pushed", detail) + pushed = run_git(self.origin, "rev-parse", "main").strip() + self.assertEqual(self.state(ctx)["push"]["sha"], pushed) + result, detail = ar.run_job(self._ctx(6, push=True), "n") + self.assertEqual(result, ar.OK) + self.assertIn("push held", detail) + self.assertEqual(run_git(self.origin, "rev-parse", "main").strip(), pushed) + + +class Units(Scratch): + def test_one_service_template_and_a_timer_per_scheduled_job(self): + jobs = ar.parse_jobs({"jobs": { + "nightly": {"schedule": "*-*-* 02:00"}, "manual": {}}}) + units = ar.render_units(jobs, Path("/srv/retrobios"), "/usr/bin/python3") + self.assertEqual(sorted(units), ["retrobios-nightly.timer", "retrobios@.service"]) + service = units["retrobios@.service"] + self.assertIn("ExecStart=/usr/bin/python3 /srv/retrobios/scripts/autorun.py run %i\n", + service) + self.assertIn("WorkingDirectory=/srv/retrobios\n", service) + self.assertIn("SuccessExitStatus=75\n", service) + timer = units["retrobios-nightly.timer"] + self.assertIn("OnCalendar=*-*-* 02:00\n", timer) + self.assertIn("Persistent=true\n", timer) + self.assertIn("Unit=retrobios@nightly.service\n", timer) + pushed = ar.render_units(jobs, Path("/r"), "py", push=True)["retrobios@.service"] + self.assertIn("run %i --push\n", pushed) + + def test_units_are_written_and_enabled_only_when_asked(self): + units = ar.render_units(ar.load_jobs(), Path("/r"), "py") + calls = [] + timers = ar.install_timers(self.tmp, units, False, lambda a: calls.append(a) or 0) + self.assertEqual(calls, []) + self.assertEqual(sorted(p.name for p in self.tmp.iterdir()), sorted(units)) + ar.install_timers(self.tmp, units, True, lambda a: calls.append(a) or 0) + self.assertEqual(calls, [["daemon-reload"], ["enable", "--now", *timers]]) + + +class Cleanup(Origin): + HOURS = 72 + + def _worktree(self, name: str, age_hours: float) -> Path: + path = self.repo / "sessions" / name + run_git(self.repo, "worktree", "add", "--quiet", "-b", f"wt-{name}", str(path)) + stamp = time.time() - age_hours * 3600 + os.utime(path, (stamp, stamp)) + return path + + def _decisions(self, busy=()) -> dict[str, cd.Decision]: + return {d.path.name: d for d in cd.plan(self.repo, busy, self.HOURS, time.time())} + + def test_worktree_and_tmp_rules(self): + old_clean = self._worktree("old-clean", 100) + old_code = self._worktree("old-code", 100) + (old_code / "install" / "a.json").write_text("generated", encoding="utf-8") + (old_code / ".gitignore").write_text("edited", encoding="utf-8") + old_generated = self._worktree("old-generated", 100) + (old_generated / "database.json").write_text("generated", encoding="utf-8") + self._worktree("recent", 10) + busy = self._worktree("busy", 100) + tmp = self.repo / "tmp" + for name in ("acquire", "old", "fresh", "used"): + (tmp / name).mkdir(parents=True) + (tmp / "main.bundle").write_bytes(b"") + stamp = time.time() - 100 * 3600 + for name in ("acquire", "old", "used", "main.bundle"): + os.utime(tmp / name, (stamp, stamp)) + # The worktree mtimes moved when files were written into them. + for path in (old_clean, old_code, old_generated, busy): + os.utime(path, (stamp, stamp)) + + decisions = self._decisions(busy=[str(busy / "scripts"), str(tmp / "used" / "f")]) + actions = {name: d.action for name, d in decisions.items()} + self.assertEqual(actions, { + "old-clean": "remove", "old-code": "keep", "old-generated": "remove", + "recent": "keep", "busy": "keep", + "acquire": "keep", "main.bundle": "keep", "old": "remove", + "fresh": "keep", "used": "keep", + }) + self.assertIn("uncommitted code: 1 files", decisions["old-code"].reason) + + failures = cd.apply(self.repo, decisions.values()) + self.assertEqual(failures, []) + self.assertFalse(old_clean.exists()) + self.assertFalse(old_generated.exists()) + self.assertTrue(old_code.exists()) + self.assertFalse((tmp / "old").exists()) + self.assertTrue((tmp / "acquire").exists()) + branches = run_git(self.repo, "branch", "--format=%(refname:short)").split() + self.assertIn("wt-old-clean", branches) + self.assertIn("wt-old-generated", branches) + + def test_a_name_prefix_of_a_busy_path_is_not_in_use(self): + self.assertTrue(cd.in_use(Path("/r/tmp/a"), ["/r/tmp/a/x"])) + self.assertFalse(cd.in_use(Path("/r/tmp/a"), ["/r/tmp/a2/x"])) + + def test_the_automation_worktree_is_never_a_candidate(self): + self.worktree() + self.assertEqual(cd.session_worktrees(self.repo), []) + + def test_a_worktree_outside_the_checkout_is_never_a_candidate(self): + outside = self.tmp / "elsewhere" + run_git(self.repo, "worktree", "add", "--quiet", "-b", "away", str(outside)) + self.assertEqual(cd.session_worktrees(self.repo), []) + + def test_a_tmp_entry_holding_a_worktree_with_code_survives(self): + holder = self.repo / "tmp" / "holder" + holder.mkdir(parents=True) + work = holder / "wt" + run_git(self.repo, "worktree", "add", "--quiet", "-b", "held", str(work)) + (work / ".gitignore").write_text("edited", encoding="utf-8") + stamp = time.time() - 100 * 3600 + for path in (holder, work): + os.utime(path, (stamp, stamp)) + decisions = self._decisions() + self.assertEqual((decisions["wt"].action, decisions["holder"].action), ("keep", "keep")) + self.assertEqual(decisions["holder"].reason, "holds a worktree") + cd.apply(self.repo, decisions.values()) + self.assertEqual((work / ".gitignore").read_text(), "edited") + + def test_low_space_raises_the_alert_line(self): + line, low = cd.disk_line(self.tmp, alert_gb=10 ** 9) + self.assertTrue(low) + self.assertIn("ALERT: free space under", line) + self.assertFalse(cd.disk_line(self.tmp, alert_gb=0)[1]) + + +class AssetSync(Scratch): + GITIGNORE = "bios/A/big.bin\nbios/B/big.bin\nbios/C/solo.iso\n" + + def setUp(self) -> None: + super().setUp() + self.root = self.tmp + for rel, data in (("bios/A/big.bin", b"a"), ("bios/C/solo.iso", b"c")): + (self.root / rel).parent.mkdir(parents=True, exist_ok=True) + (self.root / rel).write_bytes(data) + self.db = {"files": { + sra.sha1_of(self.root / "bios/A/big.bin"): {"path": "bios/A/big.bin", "size": 1}, + "0" * 40: {"path": "bios/C/solo.iso", "size": 1}, + }} + + def test_only_a_file_whose_sha1_matches_twice_is_cleared(self): + findings = [("missing", "bios/A/big.bin", 1, None), ("size", "bios/C/solo.iso", 1, 5), + ("notes", "release description", 0, None)] + plans = sra.plan_uploads(self.root, self.db, self.GITIGNORE, findings) + self.assertEqual([(p.path, p.asset, p.verdict) for p in plans], [ + ("bios/A/big.bin", "A--big.bin", "upload"), + ("bios/C/solo.iso", "solo.iso", "refused"), + ]) + self.assertIn("not the database's", plans[1].reason) + + def test_two_reads_that_disagree_refuse_the_upload(self): + reads = iter(["1" * 40, "2" * 40]) + plans = sra.plan_uploads(self.root, self.db, self.GITIGNORE, + [("content", "bios/A/big.bin", 1, 1)], + hasher=lambda _p: next(reads)) + self.assertEqual(plans[0].verdict, "refused") + self.assertIn("two reads disagree", plans[0].reason) + + def test_an_upload_goes_under_its_asset_name_and_is_tried_twice(self): + plans = [sra.Upload("bios/A/big.bin", "A--big.bin", "missing", "upload"), + sra.Upload("bios/C/solo.iso", "solo.iso", "size", "refused", "x")] + seen = [] + + def upload(path: Path) -> int: + seen.append(path.name) + return 1 + self.assertEqual(sra.upload_all(self.root, plans, upload), ["bios/A/big.bin"]) + self.assertEqual(seen, ["A--big.bin", "A--big.bin"]) + self.assertFalse((self.root / sra.STAGING / "A--big.bin").exists()) + + +class BiosIntegrity(Scratch): + """A flipped bit keeps the size and the mtime, so git status shows a + clean tree. Only the rehash against database.json finds it.""" + + OLD = time.time() - 10 * 86400 + + def setUp(self) -> None: + super().setUp() + self.repo = self.tmp / "repo" + self.repo.mkdir() + run_git(self.repo, "init", "--quiet", "-b", "main") + files = { + "bios/Aristocrat/aristmk6.zip": bytes(range(256)) * 4, + "bios/Sony/scph5501.bin": b"\x55" * 512, + "bios/Sony/edited.bin": b"\x01" * 64, + "bios/Big/large.iso": b"\x99" * 2048, + } + (self.repo / ".gitignore").write_text("bios/Big/large.iso\n", encoding="utf-8") + self.db = {"files": {}} + for rel, data in files.items(): + path = self.repo / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(data) + # An old mtime keeps git from re-reading the file as racily clean. + os.utime(path, (self.OLD, self.OLD)) + sha1 = bib.sha1_of(path) + self.db["files"][sha1] = {"path": rel, "sha1": sha1} + run_git(self.repo, "add", ".gitignore", *[r for r in files if "Big" not in r]) + run_git(self.repo, "commit", "--quiet", "-m", "chore: start") + + def flip(self, rel: str, offset: int = 100) -> bytes: + """Flip one bit in place, size and mtime unchanged.""" + path = self.repo / rel + data = bytearray(path.read_bytes()) + original = bytes(data) + data[offset] ^= 0x10 + path.write_bytes(bytes(data)) + os.utime(path, (self.OLD, self.OLD)) + return original + + def test_the_flip_is_invisible_to_git_and_restored(self): + rel = "bios/Aristocrat/aristmk6.zip" + original = self.flip(rel) + self.assertEqual(run_git(self.repo, "status", "--porcelain"), "") + report = bib.check(self.repo, self.db, apply=False) + self.assertEqual([(f.path, f.verdict) for f in report.findings], [(rel, bib.FLAGGED)]) + self.assertNotEqual((self.repo / rel).read_bytes(), original) + + report = bib.check(self.repo, self.db, apply=True) + self.assertEqual([(f.path, f.verdict) for f in report.findings], [(rel, bib.RESTORED)]) + self.assertEqual((self.repo / rel).read_bytes(), original) + self.assertTrue(report.clean) + self.assertEqual(bib.check(self.repo, self.db, apply=True).findings, []) + + def test_a_flip_in_a_gitignored_file_is_flagged_never_fixed(self): + rel = "bios/Big/large.iso" + self.flip(rel) + flipped = (self.repo / rel).read_bytes() + report = bib.check(self.repo, self.db, apply=True) + self.assertEqual([(f.path, f.verdict) for f in report.findings], [(rel, bib.FLAGGED)]) + self.assertIn("gitignored", report.findings[0].reason) + self.assertEqual((self.repo / rel).read_bytes(), flipped) + self.assertFalse(report.clean) + + def test_a_working_change_git_sees_is_left_to_its_session(self): + rel = "bios/Sony/edited.bin" + (self.repo / rel).write_bytes(b"\x02" * 80) + report = bib.check(self.repo, self.db, apply=True) + self.assertEqual([(f.path, f.verdict) for f in report.findings], [(rel, bib.SKIPPED)]) + self.assertEqual((self.repo / rel).read_bytes(), b"\x02" * 80) + + def test_a_mismatch_the_second_read_does_not_confirm_is_a_misread(self): + reads = {} + expected = {e["path"]: e["sha1"] for e in self.db["files"].values()} + + def hasher(path: Path) -> str: + rel = path.relative_to(self.repo).as_posix() + reads[rel] = reads.get(rel, 0) + 1 + if rel == "bios/Sony/scph5501.bin" and reads[rel] == 1: + return "f" * 40 + return expected[rel] + report = bib.check(self.repo, self.db, apply=True, hasher=hasher) + self.assertEqual([(f.path, f.verdict) for f in report.findings], + [("bios/Sony/scph5501.bin", bib.MISREAD)]) + self.assertTrue(report.clean) + + def test_a_blob_that_is_not_the_database_content_is_not_written(self): + rel = "bios/Sony/scph5501.bin" + self.flip(rel) + db = json.loads(json.dumps(self.db)) + for entry in db["files"].values(): + if entry["path"] == rel: + entry["sha1"] = "0" * 40 + flipped = (self.repo / rel).read_bytes() + report = bib.check(self.repo, db, apply=True) + self.assertEqual(report.findings[0].verdict, bib.FLAGGED) + self.assertIn("other bytes", report.findings[0].reason) + self.assertEqual((self.repo / rel).read_bytes(), flipped) + + def test_the_runner_keeps_the_report_in_the_job_status(self): + rel = "bios/Aristocrat/aristmk6.zip" + self.flip(rel) + Path(self.db_path).write_text(json.dumps(self.db), encoding="utf-8") + script = str(REPO_ROOT / "scripts" / "bios_integrity.py") + ctx = ar.Context( + repo=self.repo, lock_path=self.tmp / "auto.lock", host=FakeHost(), + jobs=ar.parse_jobs({"jobs": {"bios-integrity": { + "worktree": False, "min_free_gb": 0, + "steps": [{"run": [PY, script, "--apply", "--json", "--db", self.db_path], + "stdout": "tmp/autorun/bios-integrity.json"}], + "report": "tmp/autorun/bios-integrity.json"}}})) + self.assertEqual(ar.run_job(ctx, "bios-integrity")[0], ar.OK) + report = self.state(ctx)["jobs"]["bios-integrity"]["report"] + self.assertEqual(report["counts"], {"restored": 1}) + self.assertEqual(report["findings"][0]["path"], rel) + self.assertIn("1 restored", ar.render_status(ar.status(ctx))) + + @property + def db_path(self) -> str: + return str(self.tmp / "database.json") + + +if __name__ == "__main__": + unittest.main() diff --git a/wiki/testing-guide.md b/wiki/testing-guide.md index 7e8989966..f7dfd49f0 100644 --- a/wiki/testing-guide.md +++ b/wiki/testing-guide.md @@ -58,6 +58,7 @@ Every module in `tests/`, by the area it guards. | Handheld launchers | a pty as the console, fake installer and firmware tools, stubbed PortMaster dialogs | `test_handheld_launchers`, `test_port` | | Running cores | parsers on fixed inputs, a libretro core built from C in the test when a compiler exists | `test_core_probe` | | Measurement records | a synthetic measurements repository built with git | `test_measurements` | +| Automation runner, disk cleanup, asset sync and bit-flip repair | fake jobs, git repositories with a bare origin and session worktrees, a committed file with one bit flipped, canned CI runs in place of `gh` | `test_autorun` | | Truth and native export | synthetic + cached originals | `test_export_counts`, `test_export_native_layer`, `test_export_report`, `test_exporters`, `test_truth_data_dir_readers`, `test_truth_diff`, `test_truth_layout` | | Scrapers, upstream and provenance | inline sources, synthetic | `test_batocera_md5_list`, `test_check_freshness`, `test_contributing_text`, `test_coverage_cache`, `test_data_dir_refs`, `test_dataclass_defaults`, `test_dedup`, `test_doc_pages`, `test_doc_surfaces`, `test_elec_scrapers`, `test_fbneo_parser`, `test_find_pinned_revision`, `test_hash_merge`, `test_launchers`, `test_launchers_emudeck`, `test_launchers_recalbox`, `test_launchers_retrobat`, `test_launchers_retrodeck`, `test_launchers_retropie`, `test_launchers_rocknix`, `test_mame_devices`, `test_mame_parser`, `test_mame_ref_audit`, `test_optional_dependencies`, `test_profile_sync`, `test_prose_style`, `test_provenance`, `test_refresh_data_dirs`, `test_refresh_failures`, `test_refresh_stale`, `test_retrodeck_scraper_errors`, `test_romset_recipes`, `test_schema_semantic_gate`, `test_scraper_contract`, `test_site_availability`, `test_site_content_data`, `test_site_copy`, `test_site_cores`, `test_site_emulator_page`, `test_site_exports`, `test_site_funnel`, `test_site_images`, `test_site_picker`, `test_site_platform_hashes`, `test_site_r2_components`, `test_site_r2_css`, `test_site_r2_emulators`, `test_site_r2_entry`, `test_site_r2_gate`, `test_site_r2_nav`, `test_site_r2_p0`, `test_site_r2_reference`, `test_site_r2_systems`, `test_site_r2_wiki`, `test_site_r3_collection`, `test_site_r3_components`, `test_site_r3_css`, `test_site_r3_emulators`, `test_site_r3_entry`, `test_site_r3_nav`, `test_site_r3_p0`, `test_site_r3_reference`, `test_site_r3_systems`, `test_site_r3_wiki`, `test_site_r4_components`, `test_site_r4_css`, `test_site_r4_emulators`, `test_site_r4_entry`, `test_site_r4_merge`, `test_site_r4_p0`, `test_site_r4_reference`, `test_site_r4_search`, `test_site_r4_systems`, `test_site_r4_wiki`, `test_site_r5_components`, `test_site_r5_css`, `test_site_r5_emulators`, `test_site_r5_entry`, `test_site_r5_reference`, `test_site_r5_search`, `test_site_r5_systems`, `test_site_r5_wiki`, `test_site_render`, `test_site_system_links`, `test_site_validation`, `test_site_weight`, `test_site_write`, `test_source_ref_targets`, `test_upstream`, `test_upstream_index`, `test_validate_pr`, `test_validate_pr_flow`, `test_validate_pr_hashes`, `test_workflow_paths` | diff --git a/wiki/tools.md b/wiki/tools.md index abff8b34c..f1cf2194b 100644 --- a/wiki/tools.md +++ b/wiki/tools.md @@ -706,6 +706,124 @@ python scripts/refresh_stale.py python scripts/refresh_stale.py --only platforms,targets --jobs 6 ``` +### autorun.py + +Runs the unattended jobs declared in `automation/jobs.yml`, one at a time. +The registry holds each job's commands, its gate, the artifacts computed +twice, the paths each commit may carry and the schedule. Changing any of +them is a reviewed commit. + +| Guard | Rule | +|-------|------| +| Lock | One host-wide `flock` on `~/.local/state/retrobios/auto.lock` (`RETROBIOS_AUTO_LOCK` moves it). A job that finds it held is `deferred`. | +| Load | Waits while the 1-minute load is above a quarter of the CPUs, up to `max_load_wait` seconds, then `deferred`. | +| Disk | Below the job's `min_free_gb` the job is `refused`. Under 100 GB free the `disk` alert is raised. | +| Retry | Exit 139 (or signal 11), `SystemError` or "unknown opcode" is run again once on an idle host. A second failure is a failure. | +| Double computation | Each `twice` entry recomputes an artifact the gate wrote and compares both, timestamps aside. A difference is recomputed once on an idle host, then the job fails with the `host-integrity` alert. | + +Jobs that commit run in `.cache/autorun/worktree`, on the branch `autorun`, +never in the shared checkout and its index. The first run creates it and +hard-links `.cache/large/`, `data/` and the gitignored `bios/` files from the +main checkout. Before each job it is fast-forwarded, or rebased when it holds +unpushed commits, onto `origin/main`. A job that finds anything uncommitted +there stops as `blocked` and cleans nothing. Commits name their files +(`git commit -- `), and their subjects are checked against the +commit-msg hook when the registry loads. A job that leaves a change no commit +rule takes fails. + +`--push` sends the batch: rebase on `origin/main` (a conflict keeps the batch +and pushes nothing), the gate again when `origin/main` moved, then a push that +never forces. It is off by default. A push waits while the previous automation +push has no green CI. A red run on a pushed automation commit, read with +`gh run list`, opens a circuit breaker: every committing job is `blocked` +until a green run on `main` follows. + +State is `.cache/autorun/state.json`, logs are +`.cache/autorun/logs//.log`, kept 30 days. A job's `report` key +names a JSON file one of its steps writes, and the runner keeps it in the +job's status. Exit codes: 0 for `ok`, +75 for `deferred`, 1 otherwise. + +| Job | Schedule | What it does | +|-----|----------|--------------| +| `nightly` | 02:00 | `check_freshness.py --json` saved to a file, `refresh_stale.py --findings` on its `data` and `catalogs` rows, the MAME and FBNeo hash scrapers, then `pipeline.py --offline` as the gate. `database.json` and `install/*.json` are computed a second time and compared. Commits the hash profiles, the recipes and the regenerated records | +| `assets` | 04:30 | `sync_release_assets.py`, as a dry run | +| `bios-integrity` | Sunday 08:30 | `bios_integrity.py --apply --json` in the main checkout, once the host is idle (up to four hours of waiting). Its report is kept in the job's status | +| `cleanup` | 05:30 | `cleanup_disk.py --apply --hours 72` in the main checkout | +| `disk-check` | hourly | `cleanup_disk.py --check`, without the lock | + +```bash +python scripts/autorun.py run nightly --dry-run # commands and the files each commit would take +python scripts/autorun.py run nightly +python scripts/autorun.py run nightly --push +python scripts/autorun.py status +python scripts/autorun.py status --json +python scripts/autorun.py install-timers # writes systemd user units, enables nothing +python scripts/autorun.py install-timers --enable +``` + +`install-timers` writes `retrobios@.service` and one `retrobios-.timer` +per scheduled job into `--unit-dir` (default `~/.config/systemd/user`). +`--enable` also runs `systemctl --user daemon-reload` and enables the timers. +`--with-push` writes units whose jobs run with `--push`. + +### cleanup_disk.py + +Frees the space idle session worktrees and `tmp/` hold. The candidates are +the linked worktrees placed inside the checkout, where sessions create +theirs. One is removed only when no process uses it, it is older than +`--hours`, and nothing tracked changed in it but generated files +(`database.json`, `install/*.json`, `exports/*.json`, `README.md`, +`mkdocs.yml`). One holding real changes is listed, never removed. Branches +are always kept. A worktree elsewhere on disk and the automation worktree are +never candidates. `tmp/` entries older than `--hours` that no process uses +are removed, except `tmp/acquire`, `*.bundle` and an entry that holds a +linked worktree. + +The last line gives free space, with an `ALERT` line under `--alert-gb` +(default 100), and the exit code is then 1. `--check` reports space only. + +```bash +python scripts/cleanup_disk.py # dry run +python scripts/cleanup_disk.py --apply --hours 72 +python scripts/cleanup_disk.py --check +``` + +### bios_integrity.py + +A flipped bit keeps a file's size and mtime, so `git status` never shows it. +This rehashes every file `database.json` names and compares the SHA-1. A +mismatch is read a second time: two reads that disagree are a host misread +and change nothing. A confirmed mismatch on a committed file that git sees as +unmodified is restored from its blob in HEAD with `--apply`, once the blob's +object id and its content SHA-1 both check out, then hashed again. A file git +lists as modified is a working change and is left alone. A gitignored large +file, an untracked one, or a HEAD blob holding other bytes than the database +records is flagged, never repaired. The exit code is 1 while anything is +flagged. Run it on an idle host, since a busy CPU here computes wrong hashes. + +```bash +python scripts/bios_integrity.py # report only +python scripts/bios_integrity.py --apply --json # restore committed flips +python scripts/bios_integrity.py --db database.json +``` + +### sync_release_assets.py + +Uploads the `large-files` assets that `check_release_assets.py` reports +missing, at another size or with another sha256. A file goes out only when +its SHA-1, read twice, gives the same value both times and equals the SHA-1 +`database.json` records for its path. The asset keeps the name +`largefiles.asset_names()` gives it, a failed upload is tried once more, and +the release is read again afterwards. The release description is never +edited. Dry run by default. The exit code is 1 while an asset stays out of +date or is refused, so the nightly `assets` job reports it. + +```bash +python scripts/sync_release_assets.py +python scripts/sync_release_assets.py --apply +``` + ### core_probe.py Runs a libretro core with no window, no sound and no input device, and @@ -845,6 +963,10 @@ it after a shallow fetch of the pinned commit. | `check_buildbot_system.py` | Detect stale data directories by comparing with buildbot | | `check_freshness.py` | One report on every upstream the repository transcribes (see above) | | `refresh_stale.py` | Run the refresher behind every stale row of that report (see above) | +| `autorun.py` | Run the unattended jobs of `automation/jobs.yml` under one lock, in their own worktree (see above) | +| `cleanup_disk.py` | Remove idle session worktrees and old `tmp/` entries, alert under 100 GB free (see above) | +| `bios_integrity.py` | Rehash every file the database names, restore a confirmed bit flip of a committed file from HEAD (see above) | +| `sync_release_assets.py` | Upload the `large-files` assets the release serves out of date, after a double SHA-1 check (see above) | | `migrate.py` | Migrate flat bios structure to Manufacturer/Console/ hierarchy | Further options, by script: @@ -865,7 +987,7 @@ Further options, by script: | `mame_ref_audit.py` | `--write` points each stale ref at the line declaring its set, `--offline`, `--cache-dir` | | `migrate.py` | `--dry-run`, `--include-branches` also reads BIOS files from every remote branch | | `provenance_report.py` | `--provenance-dir`, `--write-derivable PATH` (with `--derive`, and the pipeline writes `derivable.json`) | -| `refresh_stale.py` | `--freshness-arg FLAG` passes a flag through to `check_freshness.py`, repeatable | +| `refresh_stale.py` | `--freshness-arg FLAG` passes a flag through to `check_freshness.py`, repeatable. `--findings FILE` reads a saved `check_freshness.py --json` report instead of running it, kept to the `--only` areas | | `bundled_only.py` | `--strict` (exit 1 while a collected file only an emulator install provides remains, pipeline step 4d), `--json`, `--install-dir`. Lists every copy of such a file, the database keeping one path per content | | `release_record.py` | `--install-dir` (manifests the archives are held against), `--platforms-dir` (registry the full pack names are rebuilt from: any other pack in the directory is refused), `--output` | | `release_notes.py` | `--whats-new FILE` (required, appended as written), `--record`, `--platforms-dir`, `--output` |