From 79665335cf0e9a202cc51d897b3ef0307c2a9508 Mon Sep 17 00:00:00 2001 From: Vasily Ilin Date: Sat, 10 Oct 2026 07:39:59 +0000 Subject: [PATCH 1/3] Add weekly Azure VM source profiling and link the public explorer --- README.md | 6 + python/tests/test_weekly_profile.py | 151 +++++++++++ scripts/weekly-profile/README.md | 67 +++++ .../lean-pool-source-profile-resume.service | 10 + .../lean-pool-source-profile.service | 22 ++ .../lean-pool-source-profile.timer | 10 + scripts/weekly-profile/run.py | 245 ++++++++++++++++++ 7 files changed, 511 insertions(+) create mode 100644 python/tests/test_weekly_profile.py create mode 100644 scripts/weekly-profile/README.md create mode 100644 scripts/weekly-profile/lean-pool-source-profile-resume.service create mode 100644 scripts/weekly-profile/lean-pool-source-profile.service create mode 100644 scripts/weekly-profile/lean-pool-source-profile.timer create mode 100644 scripts/weekly-profile/run.py diff --git a/README.md b/README.md index df5c9de442..8c41f5e6e6 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,7 @@ [![Lean Action CI](https://github.com/LeanPool/lean-pool/actions/workflows/lean_action_ci.yml/badge.svg)](https://github.com/LeanPool/lean-pool/actions/workflows/lean_action_ci.yml) [![Documentation](https://img.shields.io/badge/docs-online-blue)](https://leanpool.github.io/lean-pool/) [![Exposition](https://img.shields.io/badge/exposition-online-8a4fff)](https://leanpool.github.io/lean-pool/exposition/) +[![Source profile](https://img.shields.io/endpoint?url=https%3A%2F%2Fvilin97.github.io%2Flean-source-profiler%2Fpool%2Fbadge.json)](https://vilin97.github.io/lean-source-profiler/pool/) [![Zulip](https://img.shields.io/badge/Zulip-Lean_Pool-6492FE?logo=zulip&logoColor=white)](https://leanprover.zulipchat.com/#narrow/channel/619231-Lean-Pool) [![Semantic Search](https://img.shields.io/badge/semantic_search-Octo-2f80ed)](https://octo.axiomatic-ai.com/search?scopes=repo%3ALeanPool%2Flean-pool) [![License](https://img.shields.io/github/license/LeanPool/lean-pool)](LICENSE) @@ -17,6 +18,11 @@ Lean Pool sits between [`mathlib`](https://github.com/leanprover-community/mathl Semantic search is also available via the [API](https://search.octo.axiomatic-ai.com/api/search). +Explore compile costs in the [LeanPool source profile](https://vilin97.github.io/lean-source-profiler/pool/), +from projects and files down to declarations and source lines. The +[weekly profiling job](scripts/weekly-profile/README.md) runs on our Azure VM; +the badge shows the date of the latest completed recording. + **295** formalization projects ยท **10,939,627** lines of Lean diff --git a/python/tests/test_weekly_profile.py b/python/tests/test_weekly_profile.py new file mode 100644 index 0000000000..b7e782b68d --- /dev/null +++ b/python/tests/test_weekly_profile.py @@ -0,0 +1,151 @@ +"""Regression checks for interrupted weekly profiling and dataset publication.""" + +import importlib.util +import json +import subprocess +from pathlib import Path + +import pytest + +SCRIPT = Path(__file__).resolve().parents[2] / "scripts/weekly-profile/run.py" +SPECIFICATION = importlib.util.spec_from_file_location("weekly_profile", SCRIPT) +weekly_profile = importlib.util.module_from_spec(SPECIFICATION) +SPECIFICATION.loader.exec_module(weekly_profile) + + +def test_failed_capture_keeps_pinned_state_and_cannot_publish(monkeypatch, tmp_path): + """Retries preserve completed preparation and never publish partial captures.""" + state = { + "runId": "test-run", + "commit": "pool-commit", + "profilerCommit": "profiler-commit", + "phase": "prepare", + } + commands = [] + + def command(*arguments, directory, capture=False): + commands.append(arguments) + if arguments[0] == "node": + raise subprocess.CalledProcessError(1, arguments) + return "" + + monkeypatch.setattr(weekly_profile, "command", command) + with pytest.raises(subprocess.CalledProcessError): + weekly_profile.execute(tmp_path, state) + checkpoint = json.loads((tmp_path / "active.json").read_text()) + assert checkpoint["phase"] == "capture" + assert checkpoint["commit"] == "pool-commit" + commands.clear() + with pytest.raises(subprocess.CalledProcessError): + weekly_profile.execute(tmp_path, checkpoint) + assert len(commands) == 1 + assert commands[0][0] == "node" + + +def test_publication_requires_audit_for_the_pinned_commit(tmp_path): + """A passing audit for another source revision cannot replace the snapshot.""" + export = tmp_path / "export" + export.mkdir() + (export / "coverage.json").write_text('{"result":"PASS"}') + (export / "manifest.json").write_text('{"commit":"other-commit"}') + with pytest.raises(RuntimeError, match="pinned commit"): + weekly_profile.publish(tmp_path, {"commit": "expected-commit"}) + + +def test_replace_dataset_preserves_viewer_and_removes_obsolete_assets(tmp_path): + """Weekly publication must retain UI changes and remove stale file IDs.""" + export, destination = tmp_path / "export", tmp_path / "destination" + for directory in (export / "data", destination / "data"): + directory.mkdir(parents=True) + (export / "data/new.json.gz").write_bytes(b"new") + (destination / "data/old.json.gz").write_bytes(b"old") + (destination / "app.js").write_text("current viewer") + (destination / "index.html").write_text("current toolbar") + (destination / "projects.yml").write_text("old registry") + weekly_profile.replace_generated(export, destination) + assert (destination / "app.js").read_text() == "current viewer" + assert (destination / "index.html").read_text() == "current toolbar" + assert (destination / "data/new.json.gz").read_bytes() == b"new" + assert not (destination / "data/old.json.gz").exists() + assert not (destination / "projects.yml").exists() + + +def test_resume_refuses_changed_private_checkout(monkeypatch, tmp_path): + """An interrupted run cannot silently move its profiling baseline.""" + state = { + "runId": "test-run", + "commit": "pinned-pool-commit", + "profilerCommit": "pinned-profiler-commit", + "phase": "capture", + } + (tmp_path / "active.json").write_text(json.dumps(state)) + monkeypatch.setattr(weekly_profile, "command", lambda *args, **kwargs: "new-commit") + with pytest.raises(RuntimeError, match="pinned repository"): + weekly_profile.run_state(tmp_path) + + +def test_publication_retries_racing_push_and_retains_new_viewer(monkeypatch, tmp_path): + """Publish a dataset against a real Git remote with a concurrent UI change.""" + remote, initial, run = tmp_path / "remote", tmp_path / "initial", tmp_path / "run" + monkeypatch.setenv("GIT_AUTHOR_NAME", "Profile Test") + monkeypatch.setenv("GIT_AUTHOR_EMAIL", "profile@example.invalid") + monkeypatch.setenv("GIT_COMMITTER_NAME", "Profile Test") + monkeypatch.setenv("GIT_COMMITTER_EMAIL", "profile@example.invalid") + original = weekly_profile.command + original( + "git", + "init", + "--bare", + "--initial-branch=main", + str(remote), + directory=tmp_path, + ) + original("git", "clone", str(remote), str(initial), directory=tmp_path) + pool = initial / "docs/pool" + pool.mkdir(parents=True) + (pool / "manifest.json").write_text('{"completedAt":"2026-10-01T00:00:00Z"}') + (pool / "app.js").write_text("old viewer") + original("git", "add", ".", directory=initial) + original("git", "commit", "-m", "initial", directory=initial) + original("git", "push", "origin", "main", directory=initial) + export = run / "export" + export.mkdir(parents=True) + (export / "coverage.json").write_text('{"result":"PASS"}') + manifest = {"commit": "pool-commit", "completedAt": "2026-10-10T00:00:00Z"} + (export / "manifest.json").write_text(json.dumps(manifest)) + monkeypatch.setattr(weekly_profile, "PROFILER_URL", str(remote)) + raced = False + + def command(*arguments, directory, capture=False): + nonlocal raced + if arguments[:2] == ("git", "push") and not raced: + raced = True + (pool / "app.js").write_text("concurrent viewer update") + original("git", "add", ".", directory=initial) + original("git", "commit", "-m", "viewer update", directory=initial) + original("git", "push", "origin", "main", directory=initial) + return original(*arguments, directory=directory, capture=capture) + + monkeypatch.setattr(weekly_profile, "command", command) + state = {"commit": "pool-commit"} + publication = weekly_profile.publish(run, state) + assert raced + original("git", "pull", "--ff-only", directory=initial) + assert (pool / "app.js").read_text() == "concurrent viewer update" + assert json.loads((pool / "manifest.json").read_text()) == manifest + assert weekly_profile.publish(run, state) == publication + + +def test_retention_preserves_unfinished_runs_and_two_completed_runs(tmp_path): + """Long failed recordings must survive bounded successful-run cleanup.""" + for name in ("01", "02", "03", "unfinished"): + run = tmp_path / "runs" / name + run.mkdir(parents=True) + if name != "unfinished": + (run / "complete.json").write_text("{}") + weekly_profile.prune_recordings(tmp_path) + assert sorted(path.name for path in (tmp_path / "runs").iterdir()) == [ + "02", + "03", + "unfinished", + ] diff --git a/scripts/weekly-profile/README.md b/scripts/weekly-profile/README.md new file mode 100644 index 0000000000..790e8b0246 --- /dev/null +++ b/scripts/weekly-profile/README.md @@ -0,0 +1,67 @@ +# Weekly whole-pool profiling + +The Azure VM captures LeanPool every **Monday at 02:00 UTC**, using +[lean-source-profiler](https://github.com/Vilin97/lean-source-profiler)'s compact +source capture. The audited result replaces the +[public explorer](https://vilin97.github.io/lean-source-profiler/pool/) and its +date badge. A manual run uses the same service and publication path. + +The service has no runtime timeout: a recording can take more than 24 hours. +It continues after logout, retries failures after an hour, and resumes an +unfinished run after reboot. A job lock prevents overlapping recordings. +Each run pins the current LeanPool and profiler commits; retries keep those +commits and reuse successfully captured files. Native Lake builds include every +source module, including files outside project entry imports. Sources, native +module setups, raw recordings and exported summaries are audited before +publication. Failed coverage leaves the previous public snapshot intact. + +All checkouts, recordings, temporary files and npm downloads live under +`/data/lean-pool-profile`. Existing elan and user cache directories also resolve +onto `/data`. The runner refuses to start with less than 30 GiB free and retains +the two most recent successful recordings plus unfinished runs. Builds/captures +use the repository's native options. Source timings include instrumentation and +VM scheduling overhead; isolated file totals are not a parallel build duration. + +## Install on this VM + +The VM needs elan/Lake, Node.js 20+, npm, Python 3, Git and authenticated GitHub +push access to `Vilin97/lean-source-profiler`. The user service manager must have +lingering enabled (`loginctl show-user vasil -p Linger`). Publishing uses GitHub +Pages from that repository's `main:docs` directory and ordinary pushes; a racing +push is retried from current main while preserving unrelated viewer changes. + +From the LeanPool checkout: + +```bash +cx doctor +install -d /data/lean-pool-profile/bin /data/lean-pool-profile/tmp +install -m 755 scripts/weekly-profile/run.py /data/lean-pool-profile/bin/run.py +install -m 644 scripts/weekly-profile/*.service scripts/weekly-profile/*.timer \ + ~/.config/systemd/user/ +systemctl --user daemon-reload +systemctl --user enable lean-pool-source-profile-resume.service +systemctl --user enable --now lean-pool-source-profile.timer +systemctl --user start --no-block lean-pool-source-profile.service +``` + +The runner owns private checkouts under its data directory and never switches +the shared development checkout. Optional environment overrides belong in +`/data/lean-pool-profile/environment`; changing capture scope or configuration +requires finishing the current run first. + +## Monitor or run manually + +```bash +systemctl --user status lean-pool-source-profile.service +systemctl --user list-timers lean-pool-source-profile.timer +journalctl --user -u lean-pool-source-profile.service -f +cat /data/lean-pool-profile/active.json +# Once capture starts, the current recording also has progress.json and errors.json. +systemctl --user start --no-block lean-pool-source-profile.service +``` + +`active.json` names the run under `runs/` and its current stage. Each recording +contains its pinned `inventory.json`, raw captures, source summaries, build +receipts, progress and final coverage. Successful publication adds +`complete.json` and records the publication commit. The dataset manifest and +coverage audit are public at `/pool/manifest.json` and `/pool/coverage.json`. diff --git a/scripts/weekly-profile/lean-pool-source-profile-resume.service b/scripts/weekly-profile/lean-pool-source-profile-resume.service new file mode 100644 index 0000000000..0e1820b2c5 --- /dev/null +++ b/scripts/weekly-profile/lean-pool-source-profile-resume.service @@ -0,0 +1,10 @@ +[Unit] +Description=Resume an interrupted LeanPool source profile after VM reboot +ConditionPathExists=/data/lean-pool-profile/pending + +[Service] +Type=oneshot +ExecStart=/usr/bin/systemctl --user start --no-block lean-pool-source-profile.service + +[Install] +WantedBy=default.target diff --git a/scripts/weekly-profile/lean-pool-source-profile.service b/scripts/weekly-profile/lean-pool-source-profile.service new file mode 100644 index 0000000000..3c071ab15d --- /dev/null +++ b/scripts/weekly-profile/lean-pool-source-profile.service @@ -0,0 +1,22 @@ +[Unit] +Description=Capture, audit and publish the whole LeanPool source profile +Wants=network-online.target +After=network-online.target +StartLimitIntervalSec=0 + +[Service] +Type=oneshot +WorkingDirectory=/data/lean-pool-profile +Environment=PATH=/data/.elan/bin:/home/vasil/.local/bin:/usr/local/bin:/usr/bin:/bin +Environment=TMPDIR=/data/lean-pool-profile/tmp +Environment=LEAN_NUM_THREADS=4 +EnvironmentFile=-/data/lean-pool-profile/environment +ExecStart=/usr/bin/python3 /data/lean-pool-profile/bin/run.py +TimeoutStartSec=infinity +TimeoutStopSec=30s +KillMode=control-group +Restart=on-failure +RestartSec=1h +Nice=10 +CPUWeight=20 +IOWeight=20 diff --git a/scripts/weekly-profile/lean-pool-source-profile.timer b/scripts/weekly-profile/lean-pool-source-profile.timer new file mode 100644 index 0000000000..87d60038ea --- /dev/null +++ b/scripts/weekly-profile/lean-pool-source-profile.timer @@ -0,0 +1,10 @@ +[Unit] +Description=Weekly whole LeanPool source profiling on the Azure VM + +[Timer] +OnCalendar=Mon *-*-* 02:00:00 UTC +Persistent=true +Unit=lean-pool-source-profile.service + +[Install] +WantedBy=timers.target diff --git a/scripts/weekly-profile/run.py b/scripts/weekly-profile/run.py new file mode 100644 index 0000000000..5bba20258e --- /dev/null +++ b/scripts/weekly-profile/run.py @@ -0,0 +1,245 @@ +#!/usr/bin/env python3 +"""Capture, audit and publish the weekly LeanPool source profile on the VM.""" + +import fcntl +import json +import logging +import os +import shutil +import subprocess +from datetime import UTC, datetime +from pathlib import Path + +LOGGER = logging.getLogger(__name__) +POOL_URL = "https://github.com/LeanPool/lean-pool.git" +PROFILER_URL = "https://github.com/Vilin97/lean-source-profiler.git" +GENERATED = ( + "data", + "manifest.json", + "coverage.json", + "badge.json", + "LEANPOOL-LICENSE", + "LEANPOOL-NOTICE", + "projects", + "projects.yml", +) + + +def command(*arguments: str, directory: Path, capture: bool = False) -> str: + """Run a command without a shell, retaining its normal diagnostic output.""" + result = subprocess.run( + arguments, + cwd=directory, + check=True, + text=True, + stdout=subprocess.PIPE if capture else None, + ) + return result.stdout.strip() if capture else "" + + +def read_json(path: Path) -> dict: + """Read a saved run checkpoint.""" + return json.loads(path.read_text()) + + +def write_json(path: Path, value: dict) -> None: + """Atomically replace a checkpoint on the persistent disk.""" + temporary = path.with_suffix(".tmp") + temporary.write_text(json.dumps(value, indent=2) + "\n") + temporary.replace(path) + + +def checkout(root: Path, name: str, url: str, revision: str = "main") -> str: + """Prepare a private checkout and return its pinned commit.""" + repository = root / name + if not repository.exists(): + command( + "git", + "clone", + "--depth", + "1", + "--single-branch", + url, + str(repository), + directory=root, + ) + command("git", "diff", "--exit-code", "HEAD", directory=repository) + command("git", "fetch", "--depth", "1", "origin", revision, directory=repository) + command("git", "checkout", "--detach", "FETCH_HEAD", directory=repository) + return command("git", "rev-parse", "HEAD", directory=repository, capture=True) + + +def run_state(root: Path) -> dict: + """Resume an unfinished run or pin current main for a new weekly run.""" + checkpoint = root / "active.json" + if checkpoint.exists(): + state = read_json(checkpoint) + if state["phase"] != "complete": + LOGGER.info("Resuming %s at %s", state["runId"], state["phase"]) + for name, key in (("repository", "commit"), ("profiler", "profilerCommit")): + actual = command( + "git", "rev-parse", "HEAD", directory=root / name, capture=True + ) + if actual != state[key]: + raise RuntimeError(f"The pinned {name} checkout changed.") + return state + commit = checkout(root, "repository", POOL_URL) + profiler_commit = checkout(root, "profiler", PROFILER_URL) + stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + state = { + "runId": f"{stamp}-{commit[:12]}", + "commit": commit, + "profilerCommit": profiler_commit, + "phase": "prepare", + } + (root / "runs" / state["runId"]).mkdir(parents=True) + write_json(checkpoint, state) + return state + + +def replace_generated(export: Path, destination: Path) -> None: + """Replace the dataset while keeping the publisher's current viewer assets.""" + for name in GENERATED: + target, source = destination / name, export / name + if target.is_dir(): + shutil.rmtree(target) + elif target.exists(): + target.unlink() + if source.is_dir(): + shutil.copytree(source, target) + elif source.exists(): + shutil.copyfile(source, target) + + +def publish(run: Path, state: dict) -> str: + """Publish with a normal Git push, retrying races without overwriting main.""" + export = run / "export" + coverage, manifest = ( + read_json(export / "coverage.json"), + read_json(export / "manifest.json"), + ) + if coverage["result"] != "PASS" or manifest["commit"] != state["commit"]: + raise RuntimeError( + "Publication requires audited coverage for this pinned commit." + ) + for attempt in range(1, 4): + publisher = run / f"publish-{attempt}" + if publisher.exists(): + shutil.rmtree(publisher) + command( + "git", + "clone", + "--depth", + "1", + "--single-branch", + PROFILER_URL, + str(publisher), + directory=run, + ) + previous = read_json(publisher / "docs/pool/manifest.json") + if previous["completedAt"] >= manifest["completedAt"]: + LOGGER.info("This recording is already published or superseded.") + return command( + "git", "rev-parse", "HEAD", directory=publisher, capture=True + ) + replace_generated(export, publisher / "docs/pool") + command("git", "add", "--", "docs/pool", directory=publisher) + command( + "git", + "commit", + "-m", + f"Update LeanPool source profile ({state['commit'][:12]})", + directory=publisher, + ) + try: + command("git", "push", "origin", "HEAD:main", directory=publisher) + except subprocess.CalledProcessError: + if attempt == 3: + raise + LOGGER.info("Retrying publication from current main.") + continue + return command("git", "rev-parse", "HEAD", directory=publisher, capture=True) + raise RuntimeError("Publication did not complete.") + + +def prune_recordings(root: Path, keep: int = 2) -> None: + """Retain the two most recent successful recordings and all unfinished runs.""" + completed = sorted(path.parent for path in (root / "runs").glob("*/complete.json")) + for run in completed[:-keep]: + LOGGER.info("Removing retained recording %s", run.name) + shutil.rmtree(run) + + +def execute(root: Path, state: dict) -> None: + """Advance persisted stages; failures retry the same recording and commits.""" + repository, profiler = root / "repository", root / "profiler" + run = root / "runs" / state["runId"] + lake = os.environ.get("LAKE", "/data/.elan/bin/lake") + + def advance(phase: str) -> None: + state["phase"] = phase + write_json(root / "active.json", state) + LOGGER.info("%s: %s", state["runId"], phase) + + if state["phase"] == "prepare": + command("npm", "ci", "--cache", str(root / "npm-cache"), directory=profiler) + command("npm", "run", "build", directory=profiler) + command(lake, "exe", "cache", "get", directory=repository) + advance("capture") + if state["phase"] == "capture": + command( + "node", + "scripts/capture-whole-pool.cjs", + str(repository), + str(run / "recording"), + directory=profiler, + ) + advance("export") + if state["phase"] == "export": + command( + "python3", + "scripts/export-whole-pool.py", + str(run / "recording"), + str(run / "export"), + directory=profiler, + ) + advance("publish") + if state["phase"] == "publish": + state["publicationCommit"] = publish(run, state) + advance("complete") + write_json(run / "complete.json", state) + (root / "pending").unlink(missing_ok=True) + try: + prune_recordings(root) + except OSError: + LOGGER.exception("Could not prune completed recordings.") + + +def main() -> None: + """Run once under a job lock; systemd owns scheduling and failure retries.""" + logging.basicConfig( + level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" + ) + root = Path( + os.environ.get("LEAN_POOL_PROFILE_ROOT", "/data/lean-pool-profile") + ).resolve() + if not root.is_relative_to(Path("/data")): + raise RuntimeError("All profiling storage must live under /data.") + root.mkdir(parents=True, exist_ok=True) + with (root / "job.lock").open("a") as lock: + try: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + LOGGER.info("A whole-pool profiling run is already active.") + return + if shutil.disk_usage(root).free < 30 * 1024**3: + raise RuntimeError("Profiling requires at least 30 GiB free on /data.") + temporary = root / "tmp" + temporary.mkdir(exist_ok=True) + os.environ["TMPDIR"] = str(temporary) + (root / "pending").touch() + execute(root, run_state(root)) + + +if __name__ == "__main__": + main() From 10b9e6780fe58343b7d5e83d0448e1319bd2d90d Mon Sep 17 00:00:00 2001 From: Vasily Ilin Date: Sat, 10 Oct 2026 09:33:08 +0000 Subject: [PATCH 2/3] Fix interrupted weekly source-profile finalization --- python/tests/test_weekly_profile.py | 167 ++++++++++++++++++++++++++++ scripts/weekly-profile/run.py | 2 +- 2 files changed, 168 insertions(+), 1 deletion(-) diff --git a/python/tests/test_weekly_profile.py b/python/tests/test_weekly_profile.py index b7e782b68d..c7c3af91e2 100644 --- a/python/tests/test_weekly_profile.py +++ b/python/tests/test_weekly_profile.py @@ -149,3 +149,170 @@ def test_retention_preserves_unfinished_runs_and_two_completed_runs(tmp_path): "03", "unfinished", ] + + +@pytest.fixture +def completion_run(tmp_path): + """Prepare pinned state and retained recordings for finalization recovery.""" + state = { + "runId": "03-test-run", + "commit": "pinned-pool-commit", + "profilerCommit": "pinned-profiler-commit", + "phase": "prepare", + } + run = tmp_path / "runs" / state["runId"] + run.mkdir(parents=True) + (tmp_path / "pending").touch() + for name in ("01", "02", "unfinished"): + previous = tmp_path / "runs" / name + previous.mkdir() + if name != "unfinished": + (previous / "complete.json").write_text("{}") + weekly_profile.write_json(tmp_path / "active.json", state) + return {"root": tmp_path, "state": state, "run": run} + + +@pytest.fixture +def completion_operations(monkeypatch, completion_run): + """Record mocked commands and publications without invoking external work.""" + root = completion_run["root"] + state = completion_run["state"] + run = completion_run["run"] + commands, publications = [], [] + monkeypatch.setenv("LAKE", "test-lake") + + def command(*arguments, directory, capture=False): + commands.append(arguments) + if arguments[:3] == ("git", "rev-parse", "HEAD"): + assert capture + assert directory in (root / "repository", root / "profiler") + return { + "repository": state["commit"], + "profiler": state["profilerCommit"], + }[directory.name] + assert not capture + name = "repository" if arguments[0] == "test-lake" else "profiler" + assert directory == root / name + return "" + + def publish(recording, checkpoint): + assert recording == run + assert checkpoint["runId"] == state["runId"] + assert checkpoint["commit"] == "pinned-pool-commit" + assert checkpoint["profilerCommit"] == "pinned-profiler-commit" + assert checkpoint["phase"] == "publish" + publications.append(dict(checkpoint)) + return "publication-commit" + + monkeypatch.setattr(weekly_profile, "command", command) + monkeypatch.setattr(weekly_profile, "publish", publish) + return commands, publications + + +@pytest.fixture +def completion_marker_failure(monkeypatch, completion_run): + """Install a single failure before or after atomic marker publication.""" + write_json = weekly_profile.write_json + marker = completion_run["run"] / "complete.json" + interrupted = False + + def install(marker_persisted): + def interrupted_write(path, value): + nonlocal interrupted + if path == marker and not interrupted: + interrupted = True + if marker_persisted: + write_json(path, value) + raise OSError("interrupted completion marker write") + write_json(path, value) + + monkeypatch.setattr(weekly_profile, "write_json", interrupted_write) + + return install + + +def assert_interrupted_completion(completion_run, marker_persisted): + """Keep the same pinned publish checkpoint until finalization can finish.""" + root, state = completion_run["root"], completion_run["state"] + marker = completion_run["run"] / "complete.json" + checkpoint = json.loads((root / "active.json").read_text()) + assert checkpoint["phase"] == state["phase"] == "publish" + assert checkpoint["runId"] == state["runId"] + assert checkpoint["commit"] == "pinned-pool-commit" + assert checkpoint["profilerCommit"] == "pinned-profiler-commit" + assert marker.exists() is marker_persisted + if marker_persisted: + assert json.loads(marker.read_text()) == { + **checkpoint, + "phase": "complete", + "publicationCommit": "publication-commit", + } + assert (root / "pending").exists() + assert (root / "runs/01").exists() + return checkpoint + + +def assert_completed_profile_commands(completion_run, commands): + """Prepare, capture and export exactly once before the publish-only retry.""" + root, run = completion_run["root"], completion_run["run"] + assert commands == [ + ("npm", "ci", "--cache", str(root / "npm-cache")), + ("npm", "run", "build"), + ("test-lake", "exe", "cache", "get"), + ( + "node", + "scripts/capture-whole-pool.cjs", + str(root / "repository"), + str(run / "recording"), + ), + ( + "python3", + "scripts/export-whole-pool.py", + str(run / "recording"), + str(run / "export"), + ), + ] + + +def assert_completed_recording(completion_run, checkpoint, publications): + """Finish both checkpoints, remove pending and retain exactly two runs.""" + root, run = completion_run["root"], completion_run["run"] + assert len(publications) == 2 + complete = json.loads((run / "complete.json").read_text()) + assert complete == json.loads((root / "active.json").read_text()) + assert complete == { + **checkpoint, + "phase": "complete", + "publicationCommit": "publication-commit", + } + assert not (root / "pending").exists() + assert sorted(path.name for path in (root / "runs").iterdir()) == [ + "02", + "03-test-run", + "unfinished", + ] + + +@pytest.mark.parametrize("marker_persisted", [False, True]) +def test_completion_marker_failure_resumes_same_run( + completion_run, completion_operations, completion_marker_failure, marker_persisted +): + """A finalization interruption must not strand a captured recording.""" + root, state = completion_run["root"], completion_run["state"] + commands, publications = completion_operations + completion_marker_failure(marker_persisted) + with pytest.raises(OSError, match="completion marker"): + weekly_profile.execute(root, state) + checkpoint = assert_interrupted_completion(completion_run, marker_persisted) + completed_commands = list(commands) + assert_completed_profile_commands(completion_run, completed_commands) + + resumed = weekly_profile.run_state(root) + assert resumed == checkpoint + weekly_profile.execute(root, resumed) + assert commands[:5] == completed_commands + assert commands[5:] == [ + ("git", "rev-parse", "HEAD"), + ("git", "rev-parse", "HEAD"), + ] + assert_completed_recording(completion_run, checkpoint, publications) diff --git a/scripts/weekly-profile/run.py b/scripts/weekly-profile/run.py index 5bba20258e..c843e31dc0 100644 --- a/scripts/weekly-profile/run.py +++ b/scripts/weekly-profile/run.py @@ -206,8 +206,8 @@ def advance(phase: str) -> None: advance("publish") if state["phase"] == "publish": state["publicationCommit"] = publish(run, state) + write_json(run / "complete.json", {**state, "phase": "complete"}) advance("complete") - write_json(run / "complete.json", state) (root / "pending").unlink(missing_ok=True) try: prune_recordings(root) From 26a3070c32d0b9297c566e355d7bba05959962e0 Mon Sep 17 00:00:00 2001 From: Vasily Ilin Date: Sat, 10 Oct 2026 07:14:29 -0700 Subject: [PATCH 3/3] fix: recover interrupted weekly profiling and resume publication --- python/tests/test_weekly_profile.py | 300 ++++++++++++++++++++++++++++ scripts/weekly-profile/run.py | 128 ++++++++++-- 2 files changed, 406 insertions(+), 22 deletions(-) diff --git a/python/tests/test_weekly_profile.py b/python/tests/test_weekly_profile.py index c7c3af91e2..f5c1c72eaf 100644 --- a/python/tests/test_weekly_profile.py +++ b/python/tests/test_weekly_profile.py @@ -2,8 +2,11 @@ import importlib.util import json +import os import subprocess +from functools import partial from pathlib import Path +from types import SimpleNamespace import pytest @@ -316,3 +319,300 @@ def test_completion_marker_failure_resumes_same_run( ("git", "rev-parse", "HEAD"), ] assert_completed_recording(completion_run, checkpoint, publications) + + +@pytest.fixture(autouse=True) +def profile_disk_space(monkeypatch): + """Keep the disk admission deterministic without affecting its policy.""" + monkeypatch.setattr( + weekly_profile.shutil, + "disk_usage", + lambda path: SimpleNamespace(free=31 * 1024**3), + ) + + +def mocked_checkout_command(recording, *arguments, directory, capture=False): + """Model clone interruption and local Git checks without running Git.""" + recording["commands"].append((arguments, directory, capture)) + if arguments[:2] == ("git", "clone"): + destination = Path(arguments[-1]) + destination.mkdir() + (destination / "tracked").write_text("cloned content") + recording["clones"] += 1 + if recording["interrupt"] and recording["clones"] == 1: + raise subprocess.CalledProcessError(1, arguments) + (destination / "head").write_text("pinned-commit") + elif arguments[:2] == ("git", "rev-parse"): + if not (directory / "head").exists(): + raise subprocess.CalledProcessError(1, arguments) + return (directory / "head").read_text() + elif arguments[:2] == ("git", "diff") and directory.name in recording["dirty"]: + raise subprocess.CalledProcessError(1, arguments) + elif arguments[:3] == ("git", "remote", "get-url"): + return recording["url"] + return "" + + +@pytest.fixture +def checkout_commands(monkeypatch): + """Collect local checkout operations and inject a single partial clone.""" + recording = { + "commands": [], + "clones": 0, + "interrupt": False, + "dirty": set(), + "url": "https://example.invalid/pool.git", + } + monkeypatch.setattr( + weekly_profile, "command", partial(mocked_checkout_command, recording) + ) + return recording + + +def test_interrupted_clone_retries_owned_staging(checkout_commands, tmp_path): + """A failed partial clone never becomes the destination and can resume.""" + checkout_commands["interrupt"] = True + url = checkout_commands["url"] + temporary = tmp_path / ".repository.clone" + with pytest.raises(subprocess.CalledProcessError): + weekly_profile.checkout(tmp_path, "repository", url) + assert not (tmp_path / "repository").exists() + assert (temporary / "tracked").read_text() == "cloned content" + assert (tmp_path / ".repository.clone-owner.json").exists() + + assert weekly_profile.checkout(tmp_path, "repository", url) == "pinned-commit" + assert checkout_commands["clones"] == 2 + assert (tmp_path / "repository/tracked").read_text() == "cloned content" + assert not temporary.exists() + assert not (tmp_path / ".repository.clone-owner.json").exists() + assert all( + arguments[-1] == str(temporary) + for arguments, _, _ in checkout_commands["commands"] + if arguments[:2] == ("git", "clone") + ) + + +def test_old_incomplete_checkout_is_preserved(checkout_commands, tmp_path): + """Retry an old direct-clone interruption without deleting its contents.""" + repository = tmp_path / "repository" + repository.mkdir() + (repository / "partial").write_text("retained content") + assert ( + weekly_profile.checkout(tmp_path, "repository", checkout_commands["url"]) + == "pinned-commit" + ) + preserved = tmp_path / "repository.incomplete/partial" + assert preserved.read_text() == "retained content" + assert (repository / "head").read_text() == "pinned-commit" + + +@pytest.mark.parametrize("name", ["repository", ".repository.clone"]) +def test_dirty_checkout_or_staging_is_preserved(checkout_commands, tmp_path, name): + """A usable dirty checkout must never be reset, replaced or quarantined.""" + repository = tmp_path / name + repository.mkdir() + (repository / "head").write_text("existing-commit") + (repository / "tracked").write_text("local change") + checkout_commands["dirty"].add(name) + if name.startswith("."): + weekly_profile.write_json( + tmp_path / ".repository.clone-owner.json", + { + "repository": "repository", + "url": checkout_commands["url"], + "temporary": str(repository), + }, + ) + with pytest.raises(subprocess.CalledProcessError): + weekly_profile.checkout(tmp_path, "repository", checkout_commands["url"]) + assert (repository / "head").read_text() == "existing-commit" + assert (repository / "tracked").read_text() == "local change" + assert checkout_commands["clones"] == 0 + assert not (tmp_path / "repository.incomplete").exists() + + +def test_unowned_clone_staging_is_preserved(checkout_commands, tmp_path): + """An unrelated staging directory is not permission to delete its files.""" + temporary = tmp_path / ".repository.clone" + temporary.mkdir() + (temporary / "partial").write_text("unowned content") + with pytest.raises(RuntimeError, match="Unowned clone staging"): + weekly_profile.checkout(tmp_path, "repository", checkout_commands["url"]) + assert (temporary / "partial").read_text() == "unowned content" + assert checkout_commands["clones"] == 0 + + +@pytest.fixture +def profile_main_environment(monkeypatch, completion_run): + """Keep main's environment changes local and allow only this test root.""" + root = completion_run["root"].resolve() + monkeypatch.setenv("LEAN_POOL_PROFILE_ROOT", str(root)) + monkeypatch.setenv("TMPDIR", str(root / "previous-tmp")) + if not root.is_relative_to(Path("/data")): + original = Path.is_relative_to + + def test_root_is_relative_to(path, *other): + if path == root and other == (Path("/data"),): + return True + return original(path, *other) + + monkeypatch.setattr(Path, "is_relative_to", test_root_is_relative_to) + return root + + +def test_low_disk_publish_resumes_and_finishes( + monkeypatch, completion_run, completion_operations, profile_main_environment +): + """Real main finalizes a saved publication below the profiling watermark.""" + root, state = completion_run["root"], completion_run["state"] + assert root == profile_main_environment + state["phase"] = "publish" + weekly_profile.write_json(root / "active.json", state) + monkeypatch.setattr( + weekly_profile.shutil, "disk_usage", lambda path: SimpleNamespace(free=1) + ) + commands, publications = completion_operations + weekly_profile.main() + assert commands == [("git", "rev-parse", "HEAD")] * 2 + assert publications == [state] + complete = weekly_profile.read_json(completion_run["run"] / "complete.json") + assert complete == weekly_profile.read_json(root / "active.json") + assert complete == { + **state, + "phase": "complete", + "publicationCommit": "publication-commit", + } + assert (root / "job.lock").exists() + assert (root / "tmp").is_dir() + assert os.environ["TMPDIR"] == str(root / "tmp") + assert not (root / "pending").exists() + assert sorted(path.name for path in (root / "runs").iterdir()) == [ + "02", + "03-test-run", + "unfinished", + ] + + +@pytest.mark.parametrize("phase", [None, "prepare", "capture", "export", "complete"]) +def test_low_disk_refuses_nonpublication_state(monkeypatch, tmp_path, phase): + """No new checkout or nonpublication profiling starts below the watermark.""" + if phase is not None: + weekly_profile.write_json( + tmp_path / "active.json", + { + "phase": phase, + "runId": "test-run", + "commit": "pool", + "profilerCommit": "profiler", + }, + ) + calls = [] + monkeypatch.setattr( + weekly_profile, "command", lambda *args, **kwargs: calls.append(args) + ) + monkeypatch.setattr( + weekly_profile.shutil, "disk_usage", lambda path: SimpleNamespace(free=1) + ) + with pytest.raises(RuntimeError, match="30 GiB"): + weekly_profile.checked_run_state(tmp_path) + assert calls == [] + + +@pytest.mark.parametrize("phase", ["prepare", "capture", "export"]) +def test_low_disk_refuses_each_profiling_stage(monkeypatch, tmp_path, phase): + """Direct execution and later stage transitions retain the profiling guard.""" + calls = [] + monkeypatch.setattr( + weekly_profile, "command", lambda *args, **kwargs: calls.append(args) + ) + monkeypatch.setattr( + weekly_profile.shutil, "disk_usage", lambda path: SimpleNamespace(free=1) + ) + state = { + "phase": phase, + "runId": "test-run", + "commit": "pool", + "profilerCommit": "profiler", + } + with pytest.raises(RuntimeError, match="30 GiB"): + weekly_profile.execute(tmp_path, state) + assert calls == [] + assert state["phase"] == phase + + +def test_completed_staging_promotes_without_recloning(checkout_commands, tmp_path): + """A crash after cloning can resume the verified staging checkout.""" + temporary = tmp_path / ".repository.clone" + temporary.mkdir() + (temporary / "head").write_text("staged-commit") + (temporary / "tracked").write_text("staged content") + weekly_profile.write_json( + tmp_path / ".repository.clone-owner.json", + { + "repository": "repository", + "url": checkout_commands["url"], + "temporary": str(temporary), + }, + ) + assert ( + weekly_profile.checkout(tmp_path, "repository", checkout_commands["url"]) + == "staged-commit" + ) + assert checkout_commands["clones"] == 0 + assert (tmp_path / "repository/tracked").read_text() == "staged content" + assert not temporary.exists() + + +def test_disk_drop_stops_before_capture(monkeypatch, completion_run): + """Completed preparation is retained when the next stage loses admission.""" + root, state = completion_run["root"], completion_run["state"] + commands, space = [], {"free": 31 * 1024**3} + monkeypatch.setattr( + weekly_profile.shutil, + "disk_usage", + lambda path: SimpleNamespace(free=space["free"]), + ) + + def command(*arguments, directory, capture=False): + commands.append(arguments) + if arguments[1:4] == ("exe", "cache", "get"): + space["free"] = 1 + return "" + + monkeypatch.setattr(weekly_profile, "command", command) + with pytest.raises(RuntimeError, match="30 GiB"): + weekly_profile.execute(root, state) + assert len(commands) == 3 + assert all(arguments[0] not in ("node", "python3") for arguments in commands) + assert state["phase"] == "capture" + assert weekly_profile.read_json(root / "active.json")["phase"] == "capture" + + +def test_existing_incomplete_backup_is_not_overwritten(checkout_commands, tmp_path): + """An ambiguous old checkout requires review without destroying either copy.""" + repository, preserved = tmp_path / "repository", tmp_path / "repository.incomplete" + repository.mkdir() + preserved.mkdir() + (repository / "partial").write_text("current partial clone") + (preserved / "retained").write_text("previous retained content") + with pytest.raises(RuntimeError, match="already preserved"): + weekly_profile.checkout(tmp_path, "repository", checkout_commands["url"]) + assert (repository / "partial").read_text() == "current partial clone" + assert (preserved / "retained").read_text() == "previous retained content" + assert checkout_commands["clones"] == 0 + + +def test_low_disk_publish_keeps_pinned_checkout_guard(monkeypatch, completion_run): + """The publication exception does not admit a changed pinned checkout.""" + root, state = completion_run["root"], completion_run["state"] + state["phase"] = "publish" + weekly_profile.write_json(root / "active.json", state) + monkeypatch.setattr( + weekly_profile.shutil, "disk_usage", lambda path: SimpleNamespace(free=1) + ) + monkeypatch.setattr( + weekly_profile, "command", lambda *args, **kwargs: "changed-commit" + ) + with pytest.raises(RuntimeError, match="pinned repository"): + weekly_profile.checked_run_state(root) + assert weekly_profile.read_json(root / "active.json") == state diff --git a/scripts/weekly-profile/run.py b/scripts/weekly-profile/run.py index c843e31dc0..16c5b06221 100644 --- a/scripts/weekly-profile/run.py +++ b/scripts/weekly-profile/run.py @@ -49,10 +49,49 @@ def write_json(path: Path, value: dict) -> None: temporary.replace(path) -def checkout(root: Path, name: str, url: str, revision: str = "main") -> str: - """Prepare a private checkout and return its pinned commit.""" - repository = root / name - if not repository.exists(): +def checkout_has_head(repository: Path) -> bool: + """Check whether a private checkout has a usable committed revision.""" + try: + command( + "git", "rev-parse", "--verify", "HEAD", directory=repository, capture=True + ) + except subprocess.CalledProcessError: + return False + return True + + +def preserve_incomplete_checkout(repository: Path) -> None: + """Keep an old incomplete destination before retrying its clone.""" + if repository.is_symlink() or (repository.exists() and not repository.is_dir()): + raise RuntimeError( + f"The private checkout path is not a directory: {repository}" + ) + if repository.exists() and not checkout_has_head(repository): + preserved = repository.with_name(repository.name + ".incomplete") + if preserved.exists() or preserved.is_symlink(): + raise RuntimeError( + f"An incomplete checkout is already preserved: {preserved}" + ) + repository.rename(preserved) + + +def clone_checkout(root: Path, name: str, url: str) -> None: + """Promote a clean clone atomically, resetting only owned partial staging.""" + repository, temporary = root / name, root / f".{name}.clone" + ownership = root / f".{name}.clone-owner.json" + expected = {"repository": name, "url": url, "temporary": str(temporary)} + if ownership.is_symlink() or temporary.is_symlink(): + raise RuntimeError("Clone staging must not use symbolic links.") + if ownership.exists(): + if read_json(ownership) != expected: + raise RuntimeError("Clone staging belongs to a different checkout.") + elif temporary.exists(): + raise RuntimeError("Unowned clone staging must be preserved.") + else: + write_json(ownership, expected) + if temporary.exists() and not checkout_has_head(temporary): + shutil.rmtree(temporary) + if not temporary.exists(): command( "git", "clone", @@ -60,9 +99,27 @@ def checkout(root: Path, name: str, url: str, revision: str = "main") -> str: "1", "--single-branch", url, - str(repository), + str(temporary), directory=root, ) + command("git", "diff", "--exit-code", "HEAD", directory=temporary) + origin = command( + "git", "remote", "get-url", "origin", directory=temporary, capture=True + ) + if origin != url: + raise RuntimeError("Clone staging has a different origin.") + if repository.exists() or repository.is_symlink(): + raise RuntimeError("The checkout destination appeared during cloning.") + temporary.rename(repository) + ownership.unlink() + + +def checkout(root: Path, name: str, url: str, revision: str = "main") -> str: + """Prepare a private checkout and return its pinned commit.""" + repository = root / name + preserve_incomplete_checkout(repository) + if not repository.exists(): + clone_checkout(root, name, url) command("git", "diff", "--exit-code", "HEAD", directory=repository) command("git", "fetch", "--depth", "1", "origin", revision, directory=repository) command("git", "checkout", "--detach", "FETCH_HEAD", directory=repository) @@ -170,23 +227,32 @@ def prune_recordings(root: Path, keep: int = 2) -> None: shutil.rmtree(run) -def execute(root: Path, state: dict) -> None: - """Advance persisted stages; failures retry the same recording and commits.""" - repository, profiler = root / "repository", root / "profiler" - run = root / "runs" / state["runId"] - lake = os.environ.get("LAKE", "/data/.elan/bin/lake") +def require_profile_space(root: Path, phase: str) -> None: + """Keep the profiling disk guard while allowing saved publication recovery.""" + if phase != "publish" and shutil.disk_usage(root).free < 30 * 1024**3: + raise RuntimeError("Profiling requires at least 30 GiB free on /data.") - def advance(phase: str) -> None: - state["phase"] = phase - write_json(root / "active.json", state) - LOGGER.info("%s: %s", state["runId"], phase) - if state["phase"] == "prepare": +def checked_run_state(root: Path) -> dict: + """Check space before a fresh checkout or a persisted profiling stage.""" + checkpoint = root / "active.json" + phase = read_json(checkpoint)["phase"] if checkpoint.exists() else "prepare" + require_profile_space(root, phase) + return run_state(root) + + +def execute_profile_stage(root: Path, state: dict) -> None: + """Run one nonpublication stage after its disk-space admission.""" + phase = state["phase"] + require_profile_space(root, phase) + repository, profiler = root / "repository", root / "profiler" + run = root / "runs" / state["runId"] + if phase == "prepare": + lake = os.environ.get("LAKE", "/data/.elan/bin/lake") command("npm", "ci", "--cache", str(root / "npm-cache"), directory=profiler) command("npm", "run", "build", directory=profiler) command(lake, "exe", "cache", "get", directory=repository) - advance("capture") - if state["phase"] == "capture": + elif phase == "capture": command( "node", "scripts/capture-whole-pool.cjs", @@ -194,8 +260,7 @@ def advance(phase: str) -> None: str(run / "recording"), directory=profiler, ) - advance("export") - if state["phase"] == "export": + elif phase == "export": command( "python3", "scripts/export-whole-pool.py", @@ -203,6 +268,27 @@ def advance(phase: str) -> None: str(run / "export"), directory=profiler, ) + else: + raise RuntimeError(f"Not a profiling stage: {phase}") + + +def execute(root: Path, state: dict) -> None: + """Advance persisted stages; failures retry the same recording and commits.""" + run = root / "runs" / state["runId"] + + def advance(phase: str) -> None: + state["phase"] = phase + write_json(root / "active.json", state) + LOGGER.info("%s: %s", state["runId"], phase) + + if state["phase"] == "prepare": + execute_profile_stage(root, state) + advance("capture") + if state["phase"] == "capture": + execute_profile_stage(root, state) + advance("export") + if state["phase"] == "export": + execute_profile_stage(root, state) advance("publish") if state["phase"] == "publish": state["publicationCommit"] = publish(run, state) @@ -232,13 +318,11 @@ def main() -> None: except BlockingIOError: LOGGER.info("A whole-pool profiling run is already active.") return - if shutil.disk_usage(root).free < 30 * 1024**3: - raise RuntimeError("Profiling requires at least 30 GiB free on /data.") temporary = root / "tmp" temporary.mkdir(exist_ok=True) os.environ["TMPDIR"] = str(temporary) (root / "pending").touch() - execute(root, run_state(root)) + execute(root, checked_run_state(root)) if __name__ == "__main__":