From 68003e74c587b5b246cea197d2bf766b1f974eb7 Mon Sep 17 00:00:00 2001 From: Vasily Ilin Date: Thu, 15 Oct 2026 22:09:42 -0700 Subject: [PATCH 1/3] Repair weekly profile dependencies and check publication health --- python/tests/test_weekly_profile.py | 220 ++++++++++++++++++ scripts/weekly-profile/README.md | 23 ++ scripts/weekly-profile/check-health.py | 69 ++++++ .../lean-pool-source-profile-health.service | 10 + .../lean-pool-source-profile-health.timer | 10 + scripts/weekly-profile/run.py | 100 +++++++- 6 files changed, 425 insertions(+), 7 deletions(-) create mode 100644 scripts/weekly-profile/check-health.py create mode 100644 scripts/weekly-profile/lean-pool-source-profile-health.service create mode 100644 scripts/weekly-profile/lean-pool-source-profile-health.timer diff --git a/python/tests/test_weekly_profile.py b/python/tests/test_weekly_profile.py index f5c1c72eaf..54c2180e94 100644 --- a/python/tests/test_weekly_profile.py +++ b/python/tests/test_weekly_profile.py @@ -4,6 +4,7 @@ import json import os import subprocess +from datetime import UTC, datetime, timedelta from functools import partial from pathlib import Path from types import SimpleNamespace @@ -14,6 +15,11 @@ SPECIFICATION = importlib.util.spec_from_file_location("weekly_profile", SCRIPT) weekly_profile = importlib.util.module_from_spec(SPECIFICATION) SPECIFICATION.loader.exec_module(weekly_profile) +HEALTH_SPECIFICATION = importlib.util.spec_from_file_location( + "profile_health", SCRIPT.with_name("check-health.py") +) +profile_health = importlib.util.module_from_spec(HEALTH_SPECIFICATION) +HEALTH_SPECIFICATION.loader.exec_module(profile_health) def test_failed_capture_keeps_pinned_state_and_cannot_publish(monkeypatch, tmp_path): @@ -616,3 +622,217 @@ def test_low_disk_publish_keeps_pinned_checkout_guard(monkeypatch, completion_ru with pytest.raises(RuntimeError, match="pinned repository"): weekly_profile.checked_run_state(root) assert weekly_profile.read_json(root / "active.json") == state + + +@pytest.fixture +def damaged_capture(tmp_path): + """Keep a successful recording beside a failed file with missing imports.""" + run = tmp_path / "runs/test" + recording = run / "recording" + recording.mkdir(parents=True) + source = "LeanPool/Example/Failed.lean" + weekly_profile.write_json( + recording / "inventory.json", + {"projects": [{"files": [{"path": source}]}]}, + ) + weekly_profile.write_json( + recording / "errors.json", [{"stage": "capture", "path": source}] + ) + artifacts = tmp_path / "repository/.lake/packages/mathlib/.lake/build/lib/lean" + artifacts.mkdir(parents=True) + setup = tmp_path / "repository/.lake/build/ir/LeanPool/Example/Failed.setup.json" + setup.parent.mkdir(parents=True) + successful = recording / "retained.leanprofile.json" + successful.write_bytes(b"expensive successful capture") + return { + "root": tmp_path, + "run": run, + "recording": recording, + "setup": setup, + "artifacts": artifacts, + "successful": successful, + } + + +@pytest.mark.parametrize("suffix", [".olean", ".olean.server", ".olean.private", ".ir"]) +def test_capture_restores_missing_artifacts_despite_existing_trace( + monkeypatch, damaged_capture, suffix +): + """A trace and profiler marker cannot hide a removed import artifact.""" + damaged = damaged_capture + missing = damaged["artifacts"] / f"Dependency{suffix}" + missing.with_suffix(".trace").write_text("cache considers this decompressed") + weekly_profile.write_json( + damaged["setup"], {"importArts": {"Mathlib.Dependency": [[str(missing)]]}} + ) + calls = [] + + def command(*arguments, directory, capture=False): + calls.append(arguments) + if arguments[1:] == ("exe", "cache", "get!"): + missing.write_bytes(b"restored") + if arguments[0] == "node": + assert missing.is_file() + assert any("+LeanPool.Example.Failed:olean" in call for call in calls) + return "" + + monkeypatch.setattr(weekly_profile, "command", command) + weekly_profile.capture_with_recovery(damaged["root"], damaged["run"]) + assert len(calls) == 3 + assert calls[-1][0] == "node" + assert damaged["successful"].read_bytes() == b"expensive successful capture" + + +def test_capture_builds_dependencies_outside_mathlib_cache( + monkeypatch, damaged_capture +): + """A dependency absent from the cache is built explicitly before capture.""" + damaged = damaged_capture + missing = damaged["artifacts"] / "Extension.olean" + weekly_profile.write_json( + damaged["setup"], {"importArts": {"OtherPackage.Extension": [[str(missing)]]}} + ) + calls = [] + + def command(*arguments, directory, capture=False): + calls.append(arguments) + if "+OtherPackage.Extension:olean" in arguments: + missing.write_bytes(b"built from pinned source") + if arguments[0] == "node": + assert missing.exists() + return "" + + monkeypatch.setattr(weekly_profile, "command", command) + weekly_profile.capture_with_recovery(damaged["root"], damaged["run"]) + assert len(calls) == 4 + + +def test_new_capture_failure_repairs_once_and_reuses_recording( + monkeypatch, damaged_capture +): + """Recover in the same invocation, but never loop on a deterministic failure.""" + damaged = damaged_capture + errors = damaged["recording"] / "errors.json" + saved_errors = errors.read_text() + errors.unlink() + calls = [] + + def command(*arguments, directory, capture=False): + calls.append(arguments) + if arguments[0] == "node": + errors.write_text(saved_errors) + raise subprocess.CalledProcessError(1, arguments) + return "" + + monkeypatch.setattr(weekly_profile, "command", command) + with pytest.raises(subprocess.CalledProcessError): + weekly_profile.capture_with_recovery(damaged["root"], damaged["run"]) + assert [call[0] for call in calls] == ["node", "/data/.elan/bin/lake", "node"] + assert calls[0] == calls[-1] + assert damaged["successful"].exists() + + +def test_interrupted_dependency_repair_resumes_without_losing_captures( + monkeypatch, damaged_capture +): + """A failed cache restore stays retryable without erasing recorded proofs.""" + damaged = damaged_capture + missing = damaged["artifacts"] / "Dependency.olean" + weekly_profile.write_json( + damaged["setup"], {"importArts": {"Mathlib.Dependency": [[str(missing)]]}} + ) + restores = 0 + captures = [] + + def command(*arguments, directory, capture=False): + nonlocal restores + if arguments[1:] == ("exe", "cache", "get!"): + restores += 1 + if restores == 1: + raise subprocess.CalledProcessError(1, arguments) + missing.write_bytes(b"restored on retry") + if arguments[0] == "node": + captures.append(arguments) + return "" + + monkeypatch.setattr(weekly_profile, "command", command) + with pytest.raises(subprocess.CalledProcessError): + weekly_profile.capture_with_recovery(damaged["root"], damaged["run"]) + assert captures == [] + weekly_profile.capture_with_recovery(damaged["root"], damaged["run"]) + assert restores == 2 + assert len(captures) == 1 + assert damaged["successful"].read_bytes() == b"expensive successful capture" + + +def test_dependency_repair_verifies_artifacts_after_successful_build( + monkeypatch, damaged_capture +): + """A stale build trace returning success must not bypass the artifact check.""" + damaged = damaged_capture + missing = damaged["artifacts"] / "Dependency.olean" + weekly_profile.write_json( + damaged["setup"], {"importArts": {"Mathlib.Dependency": [[str(missing)]]}} + ) + calls = [] + monkeypatch.setattr( + weekly_profile, "command", lambda *args, **kwargs: calls.append(args) + ) + with pytest.raises(RuntimeError, match="remain missing"): + weekly_profile.capture_with_recovery(damaged["root"], damaged["run"]) + assert all(call[0] != "node" for call in calls) + + +def test_repair_rejects_failure_outside_inventory(damaged_capture): + """Malformed failure records cannot select arbitrary files for rebuilding.""" + recording = damaged_capture["recording"] + weekly_profile.write_json( + recording / "errors.json", [{"stage": "capture", "path": "../../other.lean"}] + ) + with pytest.raises(RuntimeError, match="outside the inventory"): + weekly_profile.repair_capture(damaged_capture["root"], recording) + + +def test_hourly_retry_activity_does_not_reset_progress_health(tmp_path): + """Fresh heartbeat timestamps with unchanged success counts are still stuck.""" + now = datetime(2026, 10, 15, tzinfo=UTC) + weekly_profile.write_json( + tmp_path / "active.json", {"runId": "test", "phase": "capture"} + ) + progress = tmp_path / "runs/test/recording/progress.json" + progress.parent.mkdir(parents=True) + weekly_profile.write_json( + progress, {"successful": 25409, "updatedAt": now.isoformat()} + ) + profile_health.check_progress(tmp_path, now) + later = now + timedelta(hours=73) + weekly_profile.write_json(progress, {"successful": 0}) + with pytest.raises(RuntimeError, match="No useful profiling progress"): + profile_health.check_progress(tmp_path, later) + weekly_profile.write_json( + progress, {"successful": 25409, "updatedAt": later.isoformat()} + ) + with pytest.raises(RuntimeError, match="No useful profiling progress"): + profile_health.check_progress(tmp_path, later) + weekly_profile.write_json(progress, {"successful": 25410}) + profile_health.check_progress(tmp_path, later) + + +@pytest.mark.parametrize( + "age,healthy", [(timedelta(days=9), True), (timedelta(days=11), False)] +) +def test_health_checks_actual_publication_freshness( + monkeypatch, tmp_path, age, healthy +): + """A responsive website serving an old snapshot must report unhealthy.""" + now = datetime(2026, 10, 15, tzinfo=UTC) + manifest = tmp_path / "manifest.json" + weekly_profile.write_json(manifest, {"completedAt": (now - age).isoformat()}) + monkeypatch.setattr( + profile_health, "urlopen", lambda *args, **kwargs: manifest.open() + ) + if healthy: + profile_health.check_health(tmp_path, now) + else: + with pytest.raises(RuntimeError, match="profile is stale"): + profile_health.check_health(tmp_path, now) diff --git a/scripts/weekly-profile/README.md b/scripts/weekly-profile/README.md index 790e8b0246..efb537a81f 100644 --- a/scripts/weekly-profile/README.md +++ b/scripts/weekly-profile/README.md @@ -15,6 +15,22 @@ 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. +Capture failures trigger one dependency repair per service attempt. The runner +checks the failed files' native Lake setup, force-restores the pinned Mathlib +cache when imported artifacts are missing (ordinary cache restoration can trust +stale trace files), and incrementally builds remaining imports and failed files +with Lake. It verifies the artifacts before retrying capture. Successful +recordings and source/profiler pins are preserved, including if repair itself is +interrupted. A repeated proof or profiler error fails the attempt rather than +looping inside it; systemd retains the hourly retry policy. + +An independent hourly health check fails if the public recording is over ten +days old or an unfinished local run has made no useful progress for 72 hours. +It tracks the highest captured-file count per run/stage, so restarting a failed +capture or rewriting its progress timestamp cannot hide a stall. The health +service's failure and journal are available to host monitoring; the timer itself +does not send notifications. + 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 @@ -36,11 +52,14 @@ From the LeanPool checkout: 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 755 scripts/weekly-profile/check-health.py /data/lean-pool-profile/bin/check-health.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 enable --now lean-pool-source-profile-health.timer +systemctl --user start --no-block lean-pool-source-profile-health.service systemctl --user start --no-block lean-pool-source-profile.service ``` @@ -56,6 +75,10 @@ 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 +# Independent freshness/progress check, also suitable for external monitoring: +python3 /data/lean-pool-profile/bin/check-health.py +systemctl --user status lean-pool-source-profile-health.service +journalctl --user -u lean-pool-source-profile-health.service -n 20 # Once capture starts, the current recording also has progress.json and errors.json. systemctl --user start --no-block lean-pool-source-profile.service ``` diff --git a/scripts/weekly-profile/check-health.py b/scripts/weekly-profile/check-health.py new file mode 100644 index 0000000000..fcf821b32d --- /dev/null +++ b/scripts/weekly-profile/check-health.py @@ -0,0 +1,69 @@ +#!/usr/bin/env python3 +"""Check publication age and useful capture progress independently of retries.""" + +import json +import logging +import os +from datetime import UTC, datetime, timedelta +from pathlib import Path +from urllib.request import urlopen + +LOGGER = logging.getLogger(__name__) +MANIFEST_URL = "https://vilin97.github.io/lean-source-profiler/pool/manifest.json" + + +def check_progress(root: Path, now: datetime) -> None: + """Repeated failed attempts must not count as useful recording progress.""" + active = root / "active.json" + if not active.exists(): + return + state = json.loads(active.read_text()) + progress = root / "runs" / state["runId"] / "recording/progress.json" + successful = ( + json.loads(progress.read_text())["successful"] if progress.exists() else 0 + ) + fingerprint = [state["runId"], state["phase"]] + checkpoint = root / "health.json" + saved = json.loads(checkpoint.read_text()) if checkpoint.exists() else {} + if saved.get("progress") != fingerprint or successful > saved.get("successful", 0): + saved = { + "progress": fingerprint, + "successful": successful, + "lastProgressAt": now.isoformat(), + } + temporary = checkpoint.with_suffix(".tmp") + temporary.write_text(json.dumps(saved, indent=2) + "\n") + temporary.replace(checkpoint) + if state["phase"] != "complete" and now - datetime.fromisoformat( + saved["lastProgressAt"] + ) > timedelta(hours=72): + raise RuntimeError(f"No useful profiling progress for 72 hours: {fingerprint}") + + +def check_health(root: Path, now: datetime) -> None: + """Require a reachable recent public dataset and an advancing local run.""" + check_progress(root, now) + with urlopen(MANIFEST_URL, timeout=20) as response: + manifest = json.load(response) + completed = datetime.fromisoformat(manifest["completedAt"]) + if now - completed > timedelta(days=10): + raise RuntimeError( + f"Public source profile is stale: completed {completed.isoformat()}" + ) + LOGGER.info("Public source profile completed %s; progress is healthy.", completed) + + +def main() -> int: + """Expose a nonzero health signal for systemd and external monitoring.""" + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") + root = Path(os.environ.get("LEAN_POOL_PROFILE_ROOT", "/data/lean-pool-profile")) + try: + check_health(root, datetime.now(UTC)) + except Exception: + LOGGER.exception("Weekly source profiling is unhealthy.") + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/weekly-profile/lean-pool-source-profile-health.service b/scripts/weekly-profile/lean-pool-source-profile-health.service new file mode 100644 index 0000000000..9979c0ac8c --- /dev/null +++ b/scripts/weekly-profile/lean-pool-source-profile-health.service @@ -0,0 +1,10 @@ +[Unit] +Description=Check LeanPool profile freshness and useful progress +Wants=network-online.target +After=network-online.target + +[Service] +Type=oneshot +ExecStart=/usr/bin/python3 /data/lean-pool-profile/bin/check-health.py +EnvironmentFile=-/data/lean-pool-profile/environment +TimeoutStartSec=60s diff --git a/scripts/weekly-profile/lean-pool-source-profile-health.timer b/scripts/weekly-profile/lean-pool-source-profile-health.timer new file mode 100644 index 0000000000..edf2c7c39f --- /dev/null +++ b/scripts/weekly-profile/lean-pool-source-profile-health.timer @@ -0,0 +1,10 @@ +[Unit] +Description=Hourly LeanPool source profile health check + +[Timer] +OnCalendar=*-*-* *:17:00 UTC +Persistent=true +Unit=lean-pool-source-profile-health.service + +[Install] +WantedBy=timers.target diff --git a/scripts/weekly-profile/run.py b/scripts/weekly-profile/run.py index 16c5b06221..e6fce9cda9 100644 --- a/scripts/weekly-profile/run.py +++ b/scripts/weekly-profile/run.py @@ -241,6 +241,98 @@ def checked_run_state(root: Path) -> dict: return run_state(root) +def failed_capture_sources(recording: Path) -> list[str]: + """Recover only source paths belonging to the pinned recording inventory.""" + errors = recording / "errors.json" + if not errors.exists(): + return [] + inventory = read_json(recording / "inventory.json") + sources = { + entry["path"] for project in inventory["projects"] for entry in project["files"] + } + failed = set() + for error in read_json(errors): + if error["stage"] != "capture": + continue + source = error["path"] + if source not in sources or ".." in Path(source).parts: + raise RuntimeError(f"Capture failure is outside the inventory: {source}") + if not source.startswith("LeanPool/") or not source.endswith(".lean"): + raise RuntimeError(f"Invalid capture source: {source}") + failed.add(source) + return sorted(failed) + + +def missing_capture_imports(repository: Path, sources: list[str]) -> set[str]: + """Check every artifact Lake resolved, including server, private and IR files.""" + missing = set() + for source in sources: + setup = repository / ".lake/build/ir" / Path(source).with_suffix(".setup.json") + if not setup.exists(): + continue + for module, groups in read_json(setup)["importArts"].items(): + if any( + not (repository / artifact).is_file() + for group in groups + for artifact in group + ): + missing.add(module) + return missing + + +def build_capture_modules(repository: Path, modules: set[str]) -> None: + """Let Lake repair the dependency graph instead of trusting profiler markers.""" + lake = os.environ.get("LAKE", "/data/.elan/bin/lake") + targets = [f"+{module}:olean" for module in sorted(modules)] + for offset in range(0, len(targets), 128): + command(lake, "build", *targets[offset : offset + 128], directory=repository) + + +def repair_capture(root: Path, recording: Path) -> bool: + """Repair failed files without changing pins or discarding successful captures.""" + sources = failed_capture_sources(recording) + if not sources: + return False + require_profile_space(root, "capture") + repository = root / "repository" + missing = missing_capture_imports(repository, sources) + LOGGER.warning( + "Repairing %d failed captures (%d missing imports).", len(sources), len(missing) + ) + if missing: + lake = os.environ.get("LAKE", "/data/.elan/bin/lake") + # Ordinary `cache get` trusts trace files even if an artifact was removed. + command(lake, "exe", "cache", "get!", directory=repository) + require_profile_space(root, "capture") + build_capture_modules(repository, missing_capture_imports(repository, sources)) + modules = {source.removesuffix(".lean").replace("/", ".") for source in sources} + build_capture_modules(repository, modules) + remaining = missing_capture_imports(repository, sources) + if remaining: + raise RuntimeError( + f"Capture imports remain missing after repair: {sorted(remaining)}" + ) + return True + + +def capture_with_recovery(root: Path, run: Path) -> None: + """Allow one dependency repair per attempt, including after an interrupted retry.""" + recording = run / "recording" + repaired = repair_capture(root, recording) + arguments = ( + "node", + "scripts/capture-whole-pool.cjs", + str(root / "repository"), + str(recording), + ) + try: + command(*arguments, directory=root / "profiler") + except subprocess.CalledProcessError: + if repaired or not repair_capture(root, recording): + raise + command(*arguments, directory=root / "profiler") + + def execute_profile_stage(root: Path, state: dict) -> None: """Run one nonpublication stage after its disk-space admission.""" phase = state["phase"] @@ -253,13 +345,7 @@ def execute_profile_stage(root: Path, state: dict) -> None: command("npm", "run", "build", directory=profiler) command(lake, "exe", "cache", "get", directory=repository) elif phase == "capture": - command( - "node", - "scripts/capture-whole-pool.cjs", - str(repository), - str(run / "recording"), - directory=profiler, - ) + capture_with_recovery(root, run) elif phase == "export": command( "python3", From e78aea9e7cb8b9443177b3905088f7e0fa05cc47 Mon Sep 17 00:00:00 2001 From: Vasily Ilin Date: Thu, 15 Oct 2026 22:24:43 -0700 Subject: [PATCH 2/3] Allow pinned profiler rollouts in new weekly recordings --- python/tests/test_weekly_profile.py | 21 +++++++++++++++++++++ scripts/weekly-profile/README.md | 8 ++++++++ scripts/weekly-profile/run.py | 14 ++++++++++++-- 3 files changed, 41 insertions(+), 2 deletions(-) diff --git a/python/tests/test_weekly_profile.py b/python/tests/test_weekly_profile.py index 54c2180e94..6cc0d5a491 100644 --- a/python/tests/test_weekly_profile.py +++ b/python/tests/test_weekly_profile.py @@ -93,6 +93,27 @@ def test_resume_refuses_changed_private_checkout(monkeypatch, tmp_path): weekly_profile.run_state(tmp_path) +def test_new_recording_can_pin_a_profiler_rollout(monkeypatch, tmp_path): + """A corrected profiler starts a distinct run without rewriting old captures.""" + monkeypatch.setenv("LEAN_POOL_PROFILE_POOL_REF", "validated-pool-commit") + monkeypatch.setenv("LEAN_POOL_PROFILE_PROFILER_REF", "corrected-profiler-commit") + calls = [] + + def checkout(root, name, url, revision="main"): + calls.append((name, revision)) + return revision + + monkeypatch.setattr(weekly_profile, "checkout", checkout) + state = weekly_profile.run_state(tmp_path) + assert calls == [ + ("repository", "validated-pool-commit"), + ("profiler", "corrected-profiler-commit"), + ] + assert state["commit"] == "validated-pool-commit" + assert state["profilerCommit"] == "corrected-profiler-commit" + assert weekly_profile.read_json(tmp_path / "active.json") == state + + 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" diff --git a/scripts/weekly-profile/README.md b/scripts/weekly-profile/README.md index efb537a81f..dbec4088c4 100644 --- a/scripts/weekly-profile/README.md +++ b/scripts/weekly-profile/README.md @@ -68,6 +68,14 @@ 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. +`LEAN_POOL_PROFILE_POOL_REF` and `LEAN_POOL_PROFILE_PROFILER_REF` optionally +select the Git revisions used for **new** recordings (both default to `main`). +An unfinished recording always keeps its original pins. To deploy a changed +profiler immediately, stop the service, preserve the old `active.json` alongside +its recording, set the new profiler ref, and start a new recording. Recordings +from different profiler revisions must not be combined. Clear a temporary ref +override after its fix lands on main to resume automatic weekly updates. + ## Monitor or run manually ```bash diff --git a/scripts/weekly-profile/run.py b/scripts/weekly-profile/run.py index e6fce9cda9..e8a2ce4743 100644 --- a/scripts/weekly-profile/run.py +++ b/scripts/weekly-profile/run.py @@ -140,8 +140,18 @@ def run_state(root: Path) -> dict: 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) + commit = checkout( + root, + "repository", + POOL_URL, + os.environ.get("LEAN_POOL_PROFILE_POOL_REF", "main"), + ) + profiler_commit = checkout( + root, + "profiler", + PROFILER_URL, + os.environ.get("LEAN_POOL_PROFILE_PROFILER_REF", "main"), + ) stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") state = { "runId": f"{stamp}-{commit[:12]}", From 40bf4a9af93e70e3666a9b740ef04b2c1d44c4c9 Mon Sep 17 00:00:00 2001 From: Vasily Ilin Date: Thu, 15 Oct 2026 22:30:33 -0700 Subject: [PATCH 3/3] Move weekly profiling operations to lean-pool-automation --- README.md | 4 +- python/tests/test_weekly_profile.py | 859 ------------------ scripts/weekly-profile/README.md | 98 -- scripts/weekly-profile/check-health.py | 69 -- .../lean-pool-source-profile-health.service | 10 - .../lean-pool-source-profile-health.timer | 10 - .../lean-pool-source-profile-resume.service | 10 - .../lean-pool-source-profile.service | 22 - .../lean-pool-source-profile.timer | 10 - scripts/weekly-profile/run.py | 425 --------- 10 files changed, 2 insertions(+), 1515 deletions(-) delete mode 100644 python/tests/test_weekly_profile.py delete mode 100644 scripts/weekly-profile/README.md delete mode 100644 scripts/weekly-profile/check-health.py delete mode 100644 scripts/weekly-profile/lean-pool-source-profile-health.service delete mode 100644 scripts/weekly-profile/lean-pool-source-profile-health.timer delete mode 100644 scripts/weekly-profile/lean-pool-source-profile-resume.service delete mode 100644 scripts/weekly-profile/lean-pool-source-profile.service delete mode 100644 scripts/weekly-profile/lean-pool-source-profile.timer delete mode 100644 scripts/weekly-profile/run.py diff --git a/README.md b/README.md index 83595b0ffd..cb0b9f07a0 100644 --- a/README.md +++ b/README.md @@ -19,8 +19,8 @@ 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; +from projects and files down to declarations and source lines. A weekly job +on our Azure VM refreshes the profile; the badge shows the date of the latest completed recording. diff --git a/python/tests/test_weekly_profile.py b/python/tests/test_weekly_profile.py deleted file mode 100644 index 6cc0d5a491..0000000000 --- a/python/tests/test_weekly_profile.py +++ /dev/null @@ -1,859 +0,0 @@ -"""Regression checks for interrupted weekly profiling and dataset publication.""" - -import importlib.util -import json -import os -import subprocess -from datetime import UTC, datetime, timedelta -from functools import partial -from pathlib import Path -from types import SimpleNamespace - -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) -HEALTH_SPECIFICATION = importlib.util.spec_from_file_location( - "profile_health", SCRIPT.with_name("check-health.py") -) -profile_health = importlib.util.module_from_spec(HEALTH_SPECIFICATION) -HEALTH_SPECIFICATION.loader.exec_module(profile_health) - - -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_new_recording_can_pin_a_profiler_rollout(monkeypatch, tmp_path): - """A corrected profiler starts a distinct run without rewriting old captures.""" - monkeypatch.setenv("LEAN_POOL_PROFILE_POOL_REF", "validated-pool-commit") - monkeypatch.setenv("LEAN_POOL_PROFILE_PROFILER_REF", "corrected-profiler-commit") - calls = [] - - def checkout(root, name, url, revision="main"): - calls.append((name, revision)) - return revision - - monkeypatch.setattr(weekly_profile, "checkout", checkout) - state = weekly_profile.run_state(tmp_path) - assert calls == [ - ("repository", "validated-pool-commit"), - ("profiler", "corrected-profiler-commit"), - ] - assert state["commit"] == "validated-pool-commit" - assert state["profilerCommit"] == "corrected-profiler-commit" - assert weekly_profile.read_json(tmp_path / "active.json") == state - - -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", - ] - - -@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) - - -@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 - - -@pytest.fixture -def damaged_capture(tmp_path): - """Keep a successful recording beside a failed file with missing imports.""" - run = tmp_path / "runs/test" - recording = run / "recording" - recording.mkdir(parents=True) - source = "LeanPool/Example/Failed.lean" - weekly_profile.write_json( - recording / "inventory.json", - {"projects": [{"files": [{"path": source}]}]}, - ) - weekly_profile.write_json( - recording / "errors.json", [{"stage": "capture", "path": source}] - ) - artifacts = tmp_path / "repository/.lake/packages/mathlib/.lake/build/lib/lean" - artifacts.mkdir(parents=True) - setup = tmp_path / "repository/.lake/build/ir/LeanPool/Example/Failed.setup.json" - setup.parent.mkdir(parents=True) - successful = recording / "retained.leanprofile.json" - successful.write_bytes(b"expensive successful capture") - return { - "root": tmp_path, - "run": run, - "recording": recording, - "setup": setup, - "artifacts": artifacts, - "successful": successful, - } - - -@pytest.mark.parametrize("suffix", [".olean", ".olean.server", ".olean.private", ".ir"]) -def test_capture_restores_missing_artifacts_despite_existing_trace( - monkeypatch, damaged_capture, suffix -): - """A trace and profiler marker cannot hide a removed import artifact.""" - damaged = damaged_capture - missing = damaged["artifacts"] / f"Dependency{suffix}" - missing.with_suffix(".trace").write_text("cache considers this decompressed") - weekly_profile.write_json( - damaged["setup"], {"importArts": {"Mathlib.Dependency": [[str(missing)]]}} - ) - calls = [] - - def command(*arguments, directory, capture=False): - calls.append(arguments) - if arguments[1:] == ("exe", "cache", "get!"): - missing.write_bytes(b"restored") - if arguments[0] == "node": - assert missing.is_file() - assert any("+LeanPool.Example.Failed:olean" in call for call in calls) - return "" - - monkeypatch.setattr(weekly_profile, "command", command) - weekly_profile.capture_with_recovery(damaged["root"], damaged["run"]) - assert len(calls) == 3 - assert calls[-1][0] == "node" - assert damaged["successful"].read_bytes() == b"expensive successful capture" - - -def test_capture_builds_dependencies_outside_mathlib_cache( - monkeypatch, damaged_capture -): - """A dependency absent from the cache is built explicitly before capture.""" - damaged = damaged_capture - missing = damaged["artifacts"] / "Extension.olean" - weekly_profile.write_json( - damaged["setup"], {"importArts": {"OtherPackage.Extension": [[str(missing)]]}} - ) - calls = [] - - def command(*arguments, directory, capture=False): - calls.append(arguments) - if "+OtherPackage.Extension:olean" in arguments: - missing.write_bytes(b"built from pinned source") - if arguments[0] == "node": - assert missing.exists() - return "" - - monkeypatch.setattr(weekly_profile, "command", command) - weekly_profile.capture_with_recovery(damaged["root"], damaged["run"]) - assert len(calls) == 4 - - -def test_new_capture_failure_repairs_once_and_reuses_recording( - monkeypatch, damaged_capture -): - """Recover in the same invocation, but never loop on a deterministic failure.""" - damaged = damaged_capture - errors = damaged["recording"] / "errors.json" - saved_errors = errors.read_text() - errors.unlink() - calls = [] - - def command(*arguments, directory, capture=False): - calls.append(arguments) - if arguments[0] == "node": - errors.write_text(saved_errors) - raise subprocess.CalledProcessError(1, arguments) - return "" - - monkeypatch.setattr(weekly_profile, "command", command) - with pytest.raises(subprocess.CalledProcessError): - weekly_profile.capture_with_recovery(damaged["root"], damaged["run"]) - assert [call[0] for call in calls] == ["node", "/data/.elan/bin/lake", "node"] - assert calls[0] == calls[-1] - assert damaged["successful"].exists() - - -def test_interrupted_dependency_repair_resumes_without_losing_captures( - monkeypatch, damaged_capture -): - """A failed cache restore stays retryable without erasing recorded proofs.""" - damaged = damaged_capture - missing = damaged["artifacts"] / "Dependency.olean" - weekly_profile.write_json( - damaged["setup"], {"importArts": {"Mathlib.Dependency": [[str(missing)]]}} - ) - restores = 0 - captures = [] - - def command(*arguments, directory, capture=False): - nonlocal restores - if arguments[1:] == ("exe", "cache", "get!"): - restores += 1 - if restores == 1: - raise subprocess.CalledProcessError(1, arguments) - missing.write_bytes(b"restored on retry") - if arguments[0] == "node": - captures.append(arguments) - return "" - - monkeypatch.setattr(weekly_profile, "command", command) - with pytest.raises(subprocess.CalledProcessError): - weekly_profile.capture_with_recovery(damaged["root"], damaged["run"]) - assert captures == [] - weekly_profile.capture_with_recovery(damaged["root"], damaged["run"]) - assert restores == 2 - assert len(captures) == 1 - assert damaged["successful"].read_bytes() == b"expensive successful capture" - - -def test_dependency_repair_verifies_artifacts_after_successful_build( - monkeypatch, damaged_capture -): - """A stale build trace returning success must not bypass the artifact check.""" - damaged = damaged_capture - missing = damaged["artifacts"] / "Dependency.olean" - weekly_profile.write_json( - damaged["setup"], {"importArts": {"Mathlib.Dependency": [[str(missing)]]}} - ) - calls = [] - monkeypatch.setattr( - weekly_profile, "command", lambda *args, **kwargs: calls.append(args) - ) - with pytest.raises(RuntimeError, match="remain missing"): - weekly_profile.capture_with_recovery(damaged["root"], damaged["run"]) - assert all(call[0] != "node" for call in calls) - - -def test_repair_rejects_failure_outside_inventory(damaged_capture): - """Malformed failure records cannot select arbitrary files for rebuilding.""" - recording = damaged_capture["recording"] - weekly_profile.write_json( - recording / "errors.json", [{"stage": "capture", "path": "../../other.lean"}] - ) - with pytest.raises(RuntimeError, match="outside the inventory"): - weekly_profile.repair_capture(damaged_capture["root"], recording) - - -def test_hourly_retry_activity_does_not_reset_progress_health(tmp_path): - """Fresh heartbeat timestamps with unchanged success counts are still stuck.""" - now = datetime(2026, 10, 15, tzinfo=UTC) - weekly_profile.write_json( - tmp_path / "active.json", {"runId": "test", "phase": "capture"} - ) - progress = tmp_path / "runs/test/recording/progress.json" - progress.parent.mkdir(parents=True) - weekly_profile.write_json( - progress, {"successful": 25409, "updatedAt": now.isoformat()} - ) - profile_health.check_progress(tmp_path, now) - later = now + timedelta(hours=73) - weekly_profile.write_json(progress, {"successful": 0}) - with pytest.raises(RuntimeError, match="No useful profiling progress"): - profile_health.check_progress(tmp_path, later) - weekly_profile.write_json( - progress, {"successful": 25409, "updatedAt": later.isoformat()} - ) - with pytest.raises(RuntimeError, match="No useful profiling progress"): - profile_health.check_progress(tmp_path, later) - weekly_profile.write_json(progress, {"successful": 25410}) - profile_health.check_progress(tmp_path, later) - - -@pytest.mark.parametrize( - "age,healthy", [(timedelta(days=9), True), (timedelta(days=11), False)] -) -def test_health_checks_actual_publication_freshness( - monkeypatch, tmp_path, age, healthy -): - """A responsive website serving an old snapshot must report unhealthy.""" - now = datetime(2026, 10, 15, tzinfo=UTC) - manifest = tmp_path / "manifest.json" - weekly_profile.write_json(manifest, {"completedAt": (now - age).isoformat()}) - monkeypatch.setattr( - profile_health, "urlopen", lambda *args, **kwargs: manifest.open() - ) - if healthy: - profile_health.check_health(tmp_path, now) - else: - with pytest.raises(RuntimeError, match="profile is stale"): - profile_health.check_health(tmp_path, now) diff --git a/scripts/weekly-profile/README.md b/scripts/weekly-profile/README.md deleted file mode 100644 index dbec4088c4..0000000000 --- a/scripts/weekly-profile/README.md +++ /dev/null @@ -1,98 +0,0 @@ -# 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. - -Capture failures trigger one dependency repair per service attempt. The runner -checks the failed files' native Lake setup, force-restores the pinned Mathlib -cache when imported artifacts are missing (ordinary cache restoration can trust -stale trace files), and incrementally builds remaining imports and failed files -with Lake. It verifies the artifacts before retrying capture. Successful -recordings and source/profiler pins are preserved, including if repair itself is -interrupted. A repeated proof or profiler error fails the attempt rather than -looping inside it; systemd retains the hourly retry policy. - -An independent hourly health check fails if the public recording is over ten -days old or an unfinished local run has made no useful progress for 72 hours. -It tracks the highest captured-file count per run/stage, so restarting a failed -capture or rewriting its progress timestamp cannot hide a stall. The health -service's failure and journal are available to host monitoring; the timer itself -does not send notifications. - -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 755 scripts/weekly-profile/check-health.py /data/lean-pool-profile/bin/check-health.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 enable --now lean-pool-source-profile-health.timer -systemctl --user start --no-block lean-pool-source-profile-health.service -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. - -`LEAN_POOL_PROFILE_POOL_REF` and `LEAN_POOL_PROFILE_PROFILER_REF` optionally -select the Git revisions used for **new** recordings (both default to `main`). -An unfinished recording always keeps its original pins. To deploy a changed -profiler immediately, stop the service, preserve the old `active.json` alongside -its recording, set the new profiler ref, and start a new recording. Recordings -from different profiler revisions must not be combined. Clear a temporary ref -override after its fix lands on main to resume automatic weekly updates. - -## 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 -# Independent freshness/progress check, also suitable for external monitoring: -python3 /data/lean-pool-profile/bin/check-health.py -systemctl --user status lean-pool-source-profile-health.service -journalctl --user -u lean-pool-source-profile-health.service -n 20 -# 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/check-health.py b/scripts/weekly-profile/check-health.py deleted file mode 100644 index fcf821b32d..0000000000 --- a/scripts/weekly-profile/check-health.py +++ /dev/null @@ -1,69 +0,0 @@ -#!/usr/bin/env python3 -"""Check publication age and useful capture progress independently of retries.""" - -import json -import logging -import os -from datetime import UTC, datetime, timedelta -from pathlib import Path -from urllib.request import urlopen - -LOGGER = logging.getLogger(__name__) -MANIFEST_URL = "https://vilin97.github.io/lean-source-profiler/pool/manifest.json" - - -def check_progress(root: Path, now: datetime) -> None: - """Repeated failed attempts must not count as useful recording progress.""" - active = root / "active.json" - if not active.exists(): - return - state = json.loads(active.read_text()) - progress = root / "runs" / state["runId"] / "recording/progress.json" - successful = ( - json.loads(progress.read_text())["successful"] if progress.exists() else 0 - ) - fingerprint = [state["runId"], state["phase"]] - checkpoint = root / "health.json" - saved = json.loads(checkpoint.read_text()) if checkpoint.exists() else {} - if saved.get("progress") != fingerprint or successful > saved.get("successful", 0): - saved = { - "progress": fingerprint, - "successful": successful, - "lastProgressAt": now.isoformat(), - } - temporary = checkpoint.with_suffix(".tmp") - temporary.write_text(json.dumps(saved, indent=2) + "\n") - temporary.replace(checkpoint) - if state["phase"] != "complete" and now - datetime.fromisoformat( - saved["lastProgressAt"] - ) > timedelta(hours=72): - raise RuntimeError(f"No useful profiling progress for 72 hours: {fingerprint}") - - -def check_health(root: Path, now: datetime) -> None: - """Require a reachable recent public dataset and an advancing local run.""" - check_progress(root, now) - with urlopen(MANIFEST_URL, timeout=20) as response: - manifest = json.load(response) - completed = datetime.fromisoformat(manifest["completedAt"]) - if now - completed > timedelta(days=10): - raise RuntimeError( - f"Public source profile is stale: completed {completed.isoformat()}" - ) - LOGGER.info("Public source profile completed %s; progress is healthy.", completed) - - -def main() -> int: - """Expose a nonzero health signal for systemd and external monitoring.""" - logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") - root = Path(os.environ.get("LEAN_POOL_PROFILE_ROOT", "/data/lean-pool-profile")) - try: - check_health(root, datetime.now(UTC)) - except Exception: - LOGGER.exception("Weekly source profiling is unhealthy.") - return 1 - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/weekly-profile/lean-pool-source-profile-health.service b/scripts/weekly-profile/lean-pool-source-profile-health.service deleted file mode 100644 index 9979c0ac8c..0000000000 --- a/scripts/weekly-profile/lean-pool-source-profile-health.service +++ /dev/null @@ -1,10 +0,0 @@ -[Unit] -Description=Check LeanPool profile freshness and useful progress -Wants=network-online.target -After=network-online.target - -[Service] -Type=oneshot -ExecStart=/usr/bin/python3 /data/lean-pool-profile/bin/check-health.py -EnvironmentFile=-/data/lean-pool-profile/environment -TimeoutStartSec=60s diff --git a/scripts/weekly-profile/lean-pool-source-profile-health.timer b/scripts/weekly-profile/lean-pool-source-profile-health.timer deleted file mode 100644 index edf2c7c39f..0000000000 --- a/scripts/weekly-profile/lean-pool-source-profile-health.timer +++ /dev/null @@ -1,10 +0,0 @@ -[Unit] -Description=Hourly LeanPool source profile health check - -[Timer] -OnCalendar=*-*-* *:17:00 UTC -Persistent=true -Unit=lean-pool-source-profile-health.service - -[Install] -WantedBy=timers.target diff --git a/scripts/weekly-profile/lean-pool-source-profile-resume.service b/scripts/weekly-profile/lean-pool-source-profile-resume.service deleted file mode 100644 index 0e1820b2c5..0000000000 --- a/scripts/weekly-profile/lean-pool-source-profile-resume.service +++ /dev/null @@ -1,10 +0,0 @@ -[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 deleted file mode 100644 index 3c071ab15d..0000000000 --- a/scripts/weekly-profile/lean-pool-source-profile.service +++ /dev/null @@ -1,22 +0,0 @@ -[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 deleted file mode 100644 index 87d60038ea..0000000000 --- a/scripts/weekly-profile/lean-pool-source-profile.timer +++ /dev/null @@ -1,10 +0,0 @@ -[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 deleted file mode 100644 index e8a2ce4743..0000000000 --- a/scripts/weekly-profile/run.py +++ /dev/null @@ -1,425 +0,0 @@ -#!/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_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", - "--depth", - "1", - "--single-branch", - url, - 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) - 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, - os.environ.get("LEAN_POOL_PROFILE_POOL_REF", "main"), - ) - profiler_commit = checkout( - root, - "profiler", - PROFILER_URL, - os.environ.get("LEAN_POOL_PROFILE_PROFILER_REF", "main"), - ) - 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 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 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 failed_capture_sources(recording: Path) -> list[str]: - """Recover only source paths belonging to the pinned recording inventory.""" - errors = recording / "errors.json" - if not errors.exists(): - return [] - inventory = read_json(recording / "inventory.json") - sources = { - entry["path"] for project in inventory["projects"] for entry in project["files"] - } - failed = set() - for error in read_json(errors): - if error["stage"] != "capture": - continue - source = error["path"] - if source not in sources or ".." in Path(source).parts: - raise RuntimeError(f"Capture failure is outside the inventory: {source}") - if not source.startswith("LeanPool/") or not source.endswith(".lean"): - raise RuntimeError(f"Invalid capture source: {source}") - failed.add(source) - return sorted(failed) - - -def missing_capture_imports(repository: Path, sources: list[str]) -> set[str]: - """Check every artifact Lake resolved, including server, private and IR files.""" - missing = set() - for source in sources: - setup = repository / ".lake/build/ir" / Path(source).with_suffix(".setup.json") - if not setup.exists(): - continue - for module, groups in read_json(setup)["importArts"].items(): - if any( - not (repository / artifact).is_file() - for group in groups - for artifact in group - ): - missing.add(module) - return missing - - -def build_capture_modules(repository: Path, modules: set[str]) -> None: - """Let Lake repair the dependency graph instead of trusting profiler markers.""" - lake = os.environ.get("LAKE", "/data/.elan/bin/lake") - targets = [f"+{module}:olean" for module in sorted(modules)] - for offset in range(0, len(targets), 128): - command(lake, "build", *targets[offset : offset + 128], directory=repository) - - -def repair_capture(root: Path, recording: Path) -> bool: - """Repair failed files without changing pins or discarding successful captures.""" - sources = failed_capture_sources(recording) - if not sources: - return False - require_profile_space(root, "capture") - repository = root / "repository" - missing = missing_capture_imports(repository, sources) - LOGGER.warning( - "Repairing %d failed captures (%d missing imports).", len(sources), len(missing) - ) - if missing: - lake = os.environ.get("LAKE", "/data/.elan/bin/lake") - # Ordinary `cache get` trusts trace files even if an artifact was removed. - command(lake, "exe", "cache", "get!", directory=repository) - require_profile_space(root, "capture") - build_capture_modules(repository, missing_capture_imports(repository, sources)) - modules = {source.removesuffix(".lean").replace("/", ".") for source in sources} - build_capture_modules(repository, modules) - remaining = missing_capture_imports(repository, sources) - if remaining: - raise RuntimeError( - f"Capture imports remain missing after repair: {sorted(remaining)}" - ) - return True - - -def capture_with_recovery(root: Path, run: Path) -> None: - """Allow one dependency repair per attempt, including after an interrupted retry.""" - recording = run / "recording" - repaired = repair_capture(root, recording) - arguments = ( - "node", - "scripts/capture-whole-pool.cjs", - str(root / "repository"), - str(recording), - ) - try: - command(*arguments, directory=root / "profiler") - except subprocess.CalledProcessError: - if repaired or not repair_capture(root, recording): - raise - command(*arguments, directory=root / "profiler") - - -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) - elif phase == "capture": - capture_with_recovery(root, run) - elif phase == "export": - command( - "python3", - "scripts/export-whole-pool.py", - str(run / "recording"), - 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) - write_json(run / "complete.json", {**state, "phase": "complete"}) - advance("complete") - (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 - temporary = root / "tmp" - temporary.mkdir(exist_ok=True) - os.environ["TMPDIR"] = str(temporary) - (root / "pending").touch() - execute(root, checked_run_state(root)) - - -if __name__ == "__main__": - main()