Require project entry files and reduce CI restore disk usage

.github/CODE_QUALITY.md

@@ -53,7 +53,9 @@ Current checks:
 - project `main_results[*].declaration` values include every `main_declarations` entry, so compact project cards and richer documentation metadata cannot drift
 - project `slug` and `entry_module` values are unique
 - every registered project `entry_module` is imported directly by `LeanPool.lean`
-- every top-level project module `LeanPool/Foo.lean`, except `LeanPool/Basic.lean`, is registered as an `entry_module`
+- every project has a top-level entry file `LeanPool/<Project>.lean`, including projects with nested registered entry modules
+- every nested registered entry module is reachable through its top-level entry file's imports
+- every top-level project module `LeanPool/Foo.lean`, except `LeanPool/Basic.lean`, belongs to the namespace of a registered `entry_module`
 - project entry modules and listed main declarations resolve in Lean
 - generated entry-point project cards match `LeanPool/projects.yml`
 - public declarations depend only on the allowed axiom set: `Classical.choice`, `propext`, and `Quot.sound`

python/lean_pool/ci_artifacts.py

@@ -9,20 +9,24 @@
 from __future__ import annotations
 
 import argparse
+import filecmp
 import gzip
 import hashlib
 import io
 import json
 import logging
 import os
 import shutil
+import stat
 import subprocess
 import tarfile
 import tempfile
 import time
 import zipfile
-from contextlib import contextmanager
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import AbstractContextManager, contextmanager
 from pathlib import Path, PurePosixPath
+from typing import BinaryIO
 from urllib.parse import urlencode
 
 LOGGER = logging.getLogger(__name__)
@@ -101,7 +105,33 @@ def pack_build(root: Path, destination: Path) -> None:
             archive.add(root / ".lake/build", arcname=".lake/build")
 
 
-def _extract_build(root: Path, archive: tarfile.TarFile, temporary: Path) -> bool:
+def _reuse_unchanged_file(
+    root: Path,
+    temporary: Path,
+    member: tarfile.TarInfo,
+    reused: dict[Path, os.stat_result],
+) -> None:
+    """Keep identical cached files without retaining a second full build copy."""
+    relative = Path(member.name)
+    cached = root / relative
+    staged = temporary / relative
+    if (
+        cached.is_symlink()
+        or not cached.is_file()
+        or not filecmp.cmp(cached, staged, shallow=False)
+    ):
+        return
+    reused[relative] = staged.stat()
+    staged.unlink()
+    staged.hardlink_to(cached)
+
+
+def _extract_build(
+    root: Path,
+    archive: tarfile.TarFile,
+    temporary: Path,
+    reused: dict[Path, os.stat_result],
+) -> bool:
     """Check the manifest first, then extract only ordinary build files."""
     manifest = archive.next()
     if manifest is None or manifest.name != MANIFEST_NAME or not manifest.isfile():
@@ -110,38 +140,53 @@ def _extract_build(root: Path, archive: tarfile.TarFile, temporary: Path) -> boo
         LOGGER.info("CI build has different sources or configuration; rebuilding")
         return False
     has_files = False
+    seen: set[PurePosixPath] = set()
     for member in archive:
         if member is manifest:
             continue
         path = PurePosixPath(member.name)
         if (
             path.parts[:2] != (".lake", "build")
             or ".." in path.parts
+            or path in seen
             or not (member.isfile() or member.isdir())
         ):
             raise ValueError(f"Unexpected build archive member: {member.name}")
+        seen.add(path)
         has_files |= member.isfile()
         archive.extract(member, temporary, filter="data")
+        if member.isfile():
+            _reuse_unchanged_file(root, temporary, member, reused)
     if not has_files:
         raise ValueError("Empty Lean build artifact")
     return True
 
 
 def restore_build(root: Path, archive_path: Path) -> bool:
-    """Validate and unpack once, then move the finished build into place."""
+    """Validate and unpack once, sharing unchanged files with the cached build."""
+    return _restore_archive(root, _compressed_stream(archive_path))
+
+
+def _restore_archive(root: Path, payload: AbstractContextManager[BinaryIO]) -> bool:
+    """Install staged output only after the entire compressed payload validates."""
     (root / ".lake").mkdir(exist_ok=True)
     with tempfile.TemporaryDirectory(
         prefix="lean-ci-build-", dir=root / ".lake"
     ) as name:
         temporary = Path(name)
-        with _compressed_stream(archive_path) as stream:
+        reused: dict[Path, os.stat_result] = {}
+        with payload as stream:
             with tarfile.open(fileobj=stream, mode="r|") as archive:
-                if not _extract_build(root, archive, temporary):
+                if not _extract_build(root, archive, temporary, reused):
                     return False
         destination = root / ".lake/build"
         if destination.exists():
             shutil.rmtree(destination)
         (temporary / ".lake/build").rename(destination)
+        for relative, metadata in reused.items():
+            path = root / relative
+            path.chmod(stat.S_IMODE(metadata.st_mode))
+            os.utime(path, ns=(metadata.st_atime_ns, metadata.st_mtime_ns))
     LOGGER.info("Restored Lean CI build for the current source tree")
     return True
 
@@ -178,9 +223,47 @@ def wait_for_artifact(repository: str, run: dict, wait_seconds: int) -> int | No
         time.sleep(min(20, max(0, deadline - time.monotonic())))
 
 
+def _pipe_payload(source: BinaryIO, destination: BinaryIO) -> None:
+    """Feed the ZIP entry into the decompressor without materializing another file."""
+    with destination:
+        shutil.copyfileobj(source, destination)
+
+
+@contextmanager
+def _artifact_payload(source: BinaryIO, name: str):
+    """Stream the compressed ZIP entry and verify decompression and ZIP integrity."""
+    if name.endswith(".gz"):
+        with gzip.GzipFile(fileobj=source) as stream:
+            yield stream
+            while stream.read(1024 * 1024):
+                pass
+        return
+    command = ["zstd", "--quiet", "--decompress", "--stdout"]
+    with subprocess.Popen(
+        command, stdin=subprocess.PIPE, stdout=subprocess.PIPE
+    ) as process:
+        with ThreadPoolExecutor(max_workers=1) as executor:
+            copying = executor.submit(_pipe_payload, source, process.stdin)
+            try:
+                yield process.stdout
+                while process.stdout.read(1024 * 1024):
+                    pass
+                copying.result()
+                if process.wait() != 0:
+                    raise subprocess.CalledProcessError(process.returncode, command)
+            finally:
+                process.stdout.close()
+                if process.poll() is None:
+                    process.terminate()
+                process.wait()
+
+
 def download_build(root: Path, repository: str, artifact: int) -> bool:
     """Download the archive and validate it before installing any build files."""
-    with tempfile.TemporaryDirectory(prefix="lean-ci-download-") as temporary:
+    (root / ".lake").mkdir(exist_ok=True)
+    with tempfile.TemporaryDirectory(
+        prefix="lean-ci-download-", dir=root / ".lake"
+    ) as temporary:
         directory = Path(temporary)
         zipped = directory / "artifact.zip"
         with zipped.open("wb") as output:
@@ -196,9 +279,7 @@ def download_build(root: Path, repository: str, artifact: int) -> bool:
                 raise ValueError("Unexpected files in Lean CI artifact")
             name = names[0]
             with archive.open(name) as source:
-                with (directory / name).open("wb") as destination:
-                    shutil.copyfileobj(source, destination)
-        return restore_build(root, directory / name)
+                return _restore_archive(root, _artifact_payload(source, name))
 
 
 def reuse_build(root: Path, repository: str, head: str, event: str, wait: int) -> bool:

python/lean_pool/quality.py

@@ -13,7 +13,7 @@
 
 import yaml
 
-from lean_pool.indexes import requires_project_roots, structure_errors
+from lean_pool.indexes import project_modules, requires_project_roots, structure_errors
 from lean_pool.validation_cache import ValidationCache, pool_units
 
 ALLOWED_AXIOMS = {"propext", "Quot.sound", "Classical.choice"}
@@ -874,6 +874,9 @@ def _check_projects(
     if errors:
         return errors
     errors.extend(_check_top_level_project_modules(root, path, projects))
+    if errors:
+        return errors
+    errors.extend(_check_project_wrapper_imports(root, projects))
     if errors:
         return errors
 
@@ -933,6 +936,16 @@ def _check_project_indexes(root: Path) -> list[_QualityError]:
     ]
 
 
+def _check_project_entry_files(root: Path) -> list[_QualityError]:
+    """Require the top-level entry file used by per-project build and extraction."""
+    return [
+        _QualityError(path, 1, "missing top-level project entry module")
+        for project in project_modules(root)
+        for path in [root / "LeanPool" / f"{project}.lean"]
+        if not path.is_file()
+    ]
+
+
 def _check_project_entry_imports(
     root: Path, projects: list[Any]
 ) -> list[_QualityError]:
@@ -970,13 +983,14 @@ def _check_project_entry_imports(
 def _check_top_level_project_modules(
     root: Path, path: Path, projects: list[Any]
 ) -> list[_QualityError]:
-    """Require every top-level LeanPool project module in `projects.yml`."""
+    """Require each top-level module to belong to a registered project namespace."""
     entry_modules = {
         project["entry_module"]
         for project in projects
         if isinstance(project, dict) and isinstance(project.get("entry_module"), str)
     }
-    missing = sorted(_top_level_project_modules(root) - entry_modules)
+    registered_roots = {".".join(module.split(".")[:2]) for module in entry_modules}
+    missing = sorted(_top_level_project_modules(root) - registered_roots)
     return [
         _QualityError(
             path, 1, f"top-level project module {module} missing from projects.yml"
@@ -985,6 +999,35 @@ def _check_top_level_project_modules(
     ]
 
 
+def _check_project_wrapper_imports(
+    root: Path, projects: list[Any]
+) -> list[_QualityError]:
+    """Require project extraction wrappers to expose their registered entry."""
+    entry_modules = {
+        project["entry_module"]
+        for project in projects
+        if isinstance(project, dict) and isinstance(project.get("entry_module"), str)
+    }
+    errors = []
+    for entry_module in sorted(entry_modules):
+        wrapper = ".".join(entry_module.split(".")[:2])
+        path = _module_to_path(root, wrapper)
+        if entry_module == wrapper or not path.is_file():
+            continue
+        if _module_to_path(root, entry_module) not in _reachable_leanpool_files(
+            root, wrapper
+        ):
+            errors.append(
+                _QualityError(
+                    path,
+                    1,
+                    f"registered project entry {entry_module} is not reachable "
+                    f"from {wrapper}; import the entry module",
+                )
+            )
+    return errors
+
+
 def _top_level_project_modules(root: Path) -> set[str]:
     """Return direct `LeanPool.Foo` modules that represent project entry points."""
     lean_pool = root / "LeanPool"
@@ -1404,6 +1447,7 @@ def run_checks(
     """Run all deterministic quality checks."""
     checks = [
         _check_project_indexes,
+        _check_project_entry_files,
         _check_reachability,
         _check_headers,
         _check_forbidden_lean_text,

python/tests/test_ci_artifacts.py

@@ -4,6 +4,8 @@
 
 import io
 import json
+import os
+import stat
 import subprocess
 import tarfile
 import zipfile
@@ -247,6 +249,57 @@ def test_invalid_member_after_valid_output_preserves_current_build(repository: P
     ).read_bytes() == b"compiled"
 
 
+def test_unchanged_files_share_disk_until_install(repository: Path, monkeypatch):
+    """Staging consumes space for changed files while preserving original metadata."""
+    archive = repository / "build.tar.gz"
+    output = repository / ".lake/build/lib/lean/LeanPool.olean"
+    output.chmod(0o755)
+    ci_artifacts.pack_build(repository, archive)
+    archived_time = output.stat().st_mtime_ns
+    output.chmod(0o600)
+    os.utime(output, ns=(1_000_000_000, 1_000_000_000))
+    original = ci_artifacts._extract_build
+
+    def extract(root, contents, temporary, reused):
+        result = original(root, contents, temporary, reused)
+        staged = temporary / output.relative_to(repository)
+        assert staged.samefile(output)
+        assert output.stat().st_mtime_ns == 1_000_000_000
+        assert stat.S_IMODE(output.stat().st_mode) == 0o600
+        assert output.stat().st_nlink == 2
+        return result
+
+    monkeypatch.setattr(ci_artifacts, "_extract_build", extract)
+    assert ci_artifacts.restore_build(repository, archive)
+    assert abs(output.stat().st_mtime_ns - archived_time) <= 1_000
+    assert stat.S_IMODE(output.stat().st_mode) == 0o755
+    assert output.stat().st_nlink == 1
+
+
+def test_duplicate_member_cannot_overwrite_shared_cache(repository: Path):
+    """Reject duplicate paths before extraction could write through a hard link."""
+    destination = repository / "build.tar.gz"
+    output = repository / ".lake/build/lib/lean/LeanPool.olean"
+    metadata = output.stat()
+    with tarfile.open(destination, "w:gz") as archive:
+        for name, payload in (
+            (
+                ci_artifacts.MANIFEST_NAME,
+                json.dumps(ci_artifacts.build_identity(repository)).encode(),
+            ),
+            (".lake/build/lib/lean/LeanPool.olean", b"compiled"),
+            (".lake/build/lib/lean/./LeanPool.olean", b"changed"),
+        ):
+            member = tarfile.TarInfo(name)
+            member.size = len(payload)
+            archive.addfile(member, io.BytesIO(payload))
+    with pytest.raises(ValueError, match="Unexpected build archive member"):
+        ci_artifacts.restore_build(repository, destination)
+    assert output.read_bytes() == b"compiled"
+    assert output.stat().st_mtime_ns == metadata.st_mtime_ns
+    assert output.stat().st_nlink == 1
+
+
 @pytest.mark.parametrize(
     "name", [ci_artifacts.ARCHIVE_NAME, ci_artifacts.LEGACY_ARCHIVE_NAME]
 )
@@ -271,5 +324,41 @@ def download(command, *, stdout, **kwargs):
         stdout.write(zipped.getvalue())
 
     monkeypatch.setattr(subprocess, "run", download)
+    original_restore = ci_artifacts._restore_archive
+
+    def restore(root, payload):
+        downloads = list((root / ".lake").glob("lean-ci-download-*"))
+        assert len(downloads) == 1
+        assert [path.name for path in downloads[0].iterdir()] == ["artifact.zip"]
+        return original_restore(root, payload)
+
+    monkeypatch.setattr(ci_artifacts, "_restore_archive", restore)
     assert ci_artifacts.download_build(repository, "owner/repo", 7)
     assert output.read_bytes() == b"compiled"
+    assert not list((repository / ".lake").glob("lean-ci-download-*"))
+
+
+@pytest.mark.parametrize(
+    "name", [ci_artifacts.ARCHIVE_NAME, ci_artifacts.LEGACY_ARCHIVE_NAME]
+)
+def test_download_rejects_truncated_payload(repository: Path, monkeypatch, name):
+    """Decompression must finish before streamed download replaces cached output."""
+    archive = repository / name
+    ci_artifacts.pack_build(repository, archive)
+    zipped = io.BytesIO()
+    with zipfile.ZipFile(zipped, "w") as bundle:
+        bundle.writestr(name, archive.read_bytes()[:-8])
+    output = repository / ".lake/build/lib/lean/LeanPool.olean"
+    output.write_bytes(b"old")
+    original_run = subprocess.run
+
+    def download(command, *, stdout, **kwargs):
+        if command[0] != "gh":
+            return original_run(command, stdout=stdout, **kwargs)
+        stdout.write(zipped.getvalue())
+
+    monkeypatch.setattr(subprocess, "run", download)
+    with pytest.raises((subprocess.SubprocessError, tarfile.TarError, EOFError)):
+        ci_artifacts.download_build(repository, "owner/repo", 7)
+    assert output.read_bytes() == b"old"
+    assert not list((repository / ".lake").glob("lean-ci-*"))