🐛 fix(seed): keep partial wheel images out of the cache

docs/changelog/+atomic-seed-image.bugfix.rst

@@ -0,0 +1,3 @@
+Build the seed wheel image in a staging folder and move it into place only once it is complete, so an extraction that
+fails part way, for example on a full disk, after a killed process or on a path over the Windows length limit, no longer
+leaves a partial image that later runs seed as a pip without ``METADATA``, ``RECORD`` or ``INSTALLER``.

src/virtualenv/seed/embed/via_app_data/pip_install/base.py

@@ -9,6 +9,7 @@
 import zipfile
 from abc import ABC, abstractmethod
 from configparser import ConfigParser
+from contextlib import suppress
 from itertools import chain
 from pathlib import Path
 from tempfile import mkdtemp
@@ -70,14 +71,39 @@ def install(self, version_info: tuple[int, ...]) -> None:
         LOGGER.debug("generated console scripts %s", " ".join(i.name for i in consoles))
 
     def build_image(self) -> None:
-        """Extract the seed wheel into the image directory and fix up its RECORD file.
+        """Extract the seed wheel into a staging directory, fix up its RECORD file, then move it into place.
 
         Each archive entry is validated before extraction so a tampered wheel cannot escape the image directory via an
-        absolute path or ``..`` traversal.
+        absolute path or ``..`` traversal. Later runs reuse any image directory they find, so it only appears once the
+        image is complete: a build that fails part way, for example on a full disk or a path that is too long, removes
+        its staging directory, and the next build removes one that a killed process left behind.
 
         :raises RuntimeError: if the wheel contains an entry that would land outside the image directory.
 
         """
+        image_dir: Final = self._image_dir
+        # beside the installer folder the staging path is exactly as long as the image path, so the Windows path length
+        # limits hit the build the same way they would hit the image
+        staging: Final = image_dir.parent.with_name(f"{image_dir.parent.name}-{image_dir.name}")
+        if staging.exists():
+            safe_delete(staging)
+        image_dir.parent.mkdir(parents=True, exist_ok=True)
+        staging.mkdir()
+        self._image_dir = staging
+        try:
+            self._populate_image()
+            # keep the short path Windows may have switched the staging directory to, the install reads through it
+            built: Final = self._image_dir.parent / image_dir.parent.name / image_dir.name
+            with suppress(FileNotFoundError):
+                built.rmdir()  # an empty image directory counts as missing, but blocks the rename on Windows
+            os.replace(self._image_dir, built)
+        except BaseException:
+            safe_delete(self._image_dir)  # the staging directory, through the short path if Windows switched to it
+            raise
+        self._image_dir = built
+        self.__dist_info = None  # it points into the staging directory
+
+    def _populate_image(self) -> None:
         # 1. first extract the wheel
         LOGGER.debug("build install image for %s to %s", self._wheel.name, self._image_dir)
         with zipfile.ZipFile(str(self._wheel)) as zip_ref:

src/virtualenv/seed/embed/via_app_data/pip_install/symlink.py

@@ -57,8 +57,8 @@ def _fix_records(self, extra_record_data: set[Path]) -> None:
         extra_record_data_str = self._records_text(sorted(extra_record_data, key=str))  # ty: ignore[invalid-argument-type]
         (self._dist_info / "RECORD").write_text(extra_record_data_str, encoding="utf-8")  # ty: ignore[unsupported-operator]
 
-    def build_image(self) -> None:
-        super().build_image()
+    def _populate_image(self) -> None:
+        super()._populate_image()
         # protect the image by making it read only
         set_tree(self._image_dir, S_IREAD | S_IRGRP | S_IROTH)
 

tests/unit/seed/embed/test_bootstrap_link_via_app_data.py

@@ -2,6 +2,7 @@
 
 import contextlib
 import csv
+import errno
 import hashlib
 import logging
 import os
@@ -539,6 +540,110 @@ def _run(name) -> None:
     return exceptions
 
 
+_DISK_FULL: Final = OSError(errno.ENOSPC, "No space left on device")
+
+
+def _fail_during_extraction(zip_ref: zipfile.ZipFile, target_dir: Path) -> None:
+    # stop short of the last entries, where the wheel keeps its dist-info files
+    for info in zip_ref.infolist()[:-5]:
+        zip_ref.extract(info, str(target_dir))
+    raise _DISK_FULL
+
+
+@pytest.mark.parametrize(
+    ("target", "failure", "linked"),
+    [
+        pytest.param("base._safe_extract_zip", _fail_during_extraction, False, id="copy-extract"),
+        pytest.param("base.PipInstall._generate_new_files", _DISK_FULL, False, id="copy-finish"),
+        pytest.param(
+            "base._safe_extract_zip",
+            _fail_during_extraction,
+            True,
+            id="symlink-extract",
+            marks=pytest.mark.skipif(not fs_supports_symlink(), reason="symlinks are unavailable"),
+        ),
+        pytest.param(
+            "base.PipInstall._generate_new_files",
+            _DISK_FULL,
+            True,
+            id="symlink-finish",
+            marks=pytest.mark.skipif(not fs_supports_symlink(), reason="symlinks are unavailable"),
+        ),
+        pytest.param(
+            "symlink.set_tree",
+            _DISK_FULL,
+            True,
+            id="symlink-read-only",
+            marks=pytest.mark.skipif(not fs_supports_symlink(), reason="symlinks are unavailable"),
+        ),
+    ],
+)
+def test_seed_image_build_failure_leaves_no_partial_image(
+    tmp_path: Path,
+    mocker: MockerFixture,
+    target: str,
+    failure: Callable[[zipfile.ZipFile, Path], None] | OSError,
+    linked: bool,
+) -> None:
+    cache: Final = tmp_path / "cache"
+    args: Final = [
+        str(tmp_path / "env"),
+        "--app-data",
+        str(cache),
+        "--no-periodic-update",
+        "--no-setuptools",
+        *(["--symlink-app-data"] if linked else []),
+    ]
+    patched: Final = mocker.patch(f"virtualenv.seed.embed.via_app_data.pip_install.{target}", side_effect=failure)
+    with pytest.raises(RuntimeError, match="No space left on device"):
+        cli_run(args, setup_logging=False)
+    mocker.stop(patched)
+    installer: Final = "SymlinkPipInstall" if linked else "CopyPipInstall"
+    images: Final = next(cache.glob("wheel/*/image/1"))
+    left: Final = sorted(i.name for i in [*images.iterdir(), *(images / installer).iterdir()] if i.is_dir())
+    purelib: Final = cli_run(args, setup_logging=False).creator.purelib
+    dist_info: Final = next(purelib.glob("pip-*.dist-info"))
+    assert (
+        left,
+        (dist_info / "INSTALLER").is_file(),
+        (dist_info / "RECORD").is_file(),
+        (purelib / f"{dist_info.stem}.virtualenv").is_file(),
+    ) == ([installer], True, True, True)
+
+
+@pytest.mark.parametrize(
+    ("leftover", "files", "kept"),
+    [
+        pytest.param("CopyPipInstall-{}", ("pip.py",), False, id="killed-build"),
+        pytest.param("CopyPipInstall/{}", (), True, id="empty-image"),
+    ],
+)
+def test_seed_image_build_replaces_leftovers(tmp_path: Path, leftover: str, files: tuple[str, ...], kept: bool) -> None:
+    cache: Final = tmp_path / "cache"
+    version: Final = PythonInfo.current_system().version_release_str
+    path: Final = cache / "wheel" / version / "image" / "1" / leftover.format(Path(BUNDLE_SUPPORT[version]["pip"]).stem)
+    path.mkdir(parents=True)
+    for name in files:
+        (path / name).write_text("", encoding="utf-8")
+    args: Final = [str(tmp_path / "env"), "--app-data", str(cache), "--no-periodic-update", "--no-setuptools"]
+    purelib: Final = cli_run([*args, "--pip", "embed"], setup_logging=False).creator.purelib
+    assert (path.exists(), (next(purelib.glob("pip-*.dist-info")) / "RECORD").is_file()) == (kept, True)
+
+
+def test_seed_image_build_direct_call(tmp_path: Path) -> None:
+    session: Final = cli_run([str(tmp_path / "env"), "--no-seed"], setup_logging=False)
+    wheel: Final = BUNDLE_FOLDER / BUNDLE_SUPPORT[session.interpreter.version_release_str]["pip"]
+    image: Final = tmp_path / "image" / "CopyPipInstall" / wheel.stem
+    installer: Final = CopyPipInstall(wheel, session.creator, image)
+    installer.build_image()
+    installer.install(session.interpreter.version_info)
+    assert (
+        sorted(i.name for i in image.parent.parent.iterdir()),
+        (next(image.glob("*.dist-info")) / "INSTALLER").is_file(),
+        session.creator.script("pip").exists(),
+    ) == (["CopyPipInstall"], True, True)
+
+
 def _write_zip_with_entry(zip_path: Path, entry_name: str) -> None:
     with zipfile.ZipFile(str(zip_path), "w") as archive:
         archive.writestr(entry_name, "payload")