mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
refactor(pm): path-based leftover sweep under uv's lock; trim tests to two invariants
The descriptor-anchored sweep defended against a process that swaps paths inside PM's own uv cache mid-sweep. That cache belongs to the operator's account, which SECURITY.md §2 treats as one trust envelope: anything able to swap those paths can already delete them directly. The sweep keeps the parts that matter for correctness: it holds uv's cache lock (a concurrent sync never loses an archive it is reading), refuses a linked archive-v0, skips linked buckets, and deletes only entries whose files prove a torn write. shutil.rmtree never follows links inside a tree on Linux/macOS. Tests keep the real-uv recovery case (red on main with the live error) and the not-corruption classification table; the live CLI A/B covers doctor and doctor --repair.
This commit is contained in:
Teknium committed
1 parent
9eaec9ba5d
commit
5aa8df742c
2 files changed
+16
-327
No files matched your search
+16
-102
@@ -6,7 +6,7 @@ machine stops mid-install (a hosted restart, a power cut), the rename can
|
||||
survive while the files read back as NUL bytes. uv then reuses that entry on
|
||||
every later sync and fails at install time with "The wheel is invalid", and
|
||||
retrying cannot help. These helpers recognise that failure and name the
|
||||
packages to purge. Deletion always goes through ``uv cache clean`` (the caller),
|
||||
packages to purge. Deletion goes through ``uv cache clean`` (the caller),
|
||||
which takes uv's cache lock, so a concurrent sync never loses files mid-read;
|
||||
the leftover-archive sweep takes that same lock itself.
|
||||
|
||||
@@ -19,7 +19,6 @@ import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import shutil
|
||||
import sys
|
||||
|
||||
from pm.filesystem import is_junction, lock_fd
|
||||
|
||||
@@ -75,13 +74,6 @@ _HEAD = 65536
|
||||
_LOCK_WAIT_SECONDS = 10.0
|
||||
_WHEEL_VERSION = re.compile(rb"(?im)^wheel-version[ \t]*:")
|
||||
_FILES = (("WHEEL", _WHEEL_VERSION), ("METADATA", None), ("RECORD", None))
|
||||
_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0)
|
||||
_DIRECTORY = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | _NOFOLLOW
|
||||
# The sweep works only through directory descriptors (rmtree's dir_fd is 3.11+;
|
||||
# the bootstrap runner may be older). Elsewhere it is skipped: the leftovers are
|
||||
# harmless, and on Windows uv's pointer files let `uv cache clean` take them anyway.
|
||||
_FD_SWEEP = (os.name == "posix" and sys.version_info >= (3, 11) and shutil.rmtree.avoids_symlink_attacks
|
||||
and os.open in os.supports_dir_fd and os.scandir in os.supports_fd)
|
||||
|
||||
|
||||
def _stream_damaged(stream, header: re.Pattern | None) -> bool:
|
||||
@@ -122,25 +114,6 @@ def _damaged(path: Path, header: re.Pattern | None = None) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def _damaged_at(directory: int, name: str, header: re.Pattern | None) -> bool:
|
||||
"""_damaged for *name* inside an open directory descriptor; links are not followed."""
|
||||
try:
|
||||
fd = os.open(name, os.O_RDONLY | _NOFOLLOW, dir_fd=directory)
|
||||
except FileNotFoundError:
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
try:
|
||||
# closefd=False: fdopen itself can fail (a directory in the file's place),
|
||||
# and the descriptor must be closed either way.
|
||||
with os.fdopen(fd, "rb", closefd=False) as stream:
|
||||
return _stream_damaged(stream, header)
|
||||
except OSError:
|
||||
return False
|
||||
finally:
|
||||
os.close(fd)
|
||||
|
||||
|
||||
def _dist_name(info: str) -> str | None:
|
||||
match = _DIST_INFO.fullmatch(info)
|
||||
return _valid(match.group("name")) if match else None
|
||||
@@ -191,29 +164,7 @@ def corrupt_cached_packages(cache: Path) -> set[str]:
|
||||
return set(corrupt_cache_entries(cache))
|
||||
|
||||
|
||||
def _damaged_packages_at(bucket: int) -> set[str]:
|
||||
"""corrupt_cache_entries for one bucket, read only through descriptors."""
|
||||
with os.scandir(bucket) as entries:
|
||||
infos = [entry.name for entry in entries
|
||||
if entry.name.endswith(".dist-info") and entry.is_dir(follow_symlinks=False)]
|
||||
found = set()
|
||||
for info in infos:
|
||||
name = _dist_name(info)
|
||||
if not name:
|
||||
continue
|
||||
try:
|
||||
info_fd = os.open(info, _DIRECTORY, dir_fd=bucket)
|
||||
except OSError:
|
||||
continue
|
||||
try:
|
||||
if any(_damaged_at(info_fd, file, header) for file, header in _FILES):
|
||||
found.add(name)
|
||||
finally:
|
||||
os.close(info_fd)
|
||||
return found
|
||||
|
||||
|
||||
def remove_corrupt_entries(cache: Path, names: set[str]) -> bool:
|
||||
def remove_corrupt_entries(cache: Path, names: set[str]) -> None:
|
||||
"""Delete the corrupt archive entries of *names* that ``uv cache clean`` left behind.
|
||||
|
||||
Where uv links a wheel to its archive by symlink (Linux, macOS), clean
|
||||
@@ -221,56 +172,19 @@ def remove_corrupt_entries(cache: Path, names: set[str]) -> bool:
|
||||
``hermes pm doctor`` would keep reporting. ``uv cache prune`` is not an
|
||||
option: PM seeds this cache from a bundle copy whose links are real
|
||||
directories, and prune treats their archives as unreachable, so the next
|
||||
install fails.
|
||||
install fails. On Windows uv's pointer files let clean take the archive.
|
||||
|
||||
The cache directory is opened once, and ``archive-v0``, the lock, every
|
||||
entry and every file are then reached only relative to that descriptor
|
||||
without following links, so renaming or replacing a path above or inside
|
||||
the cache mid-sweep cannot point a delete anywhere else. The sweep holds
|
||||
uv's own cache lock, so no uv process runs while it deletes, and it only
|
||||
takes entries whose files prove a torn write. Returns False when the
|
||||
sweep did not run (lock busy for _LOCK_WAIT_SECONDS, no descriptor
|
||||
support); the leftovers are harmless, and a later sweep or ``hermes pm
|
||||
doctor --repair`` takes them.
|
||||
Holds uv's own cache lock so no uv process is reading an entry while it is
|
||||
deleted. If uv keeps the lock busy past _LOCK_WAIT_SECONDS the leftovers
|
||||
stay; they are harmless and ``hermes pm doctor --repair`` takes them.
|
||||
"""
|
||||
if not names:
|
||||
return True
|
||||
if not _FD_SWEEP:
|
||||
return False
|
||||
try:
|
||||
cache_fd = os.open(cache, os.O_RDONLY | os.O_DIRECTORY)
|
||||
except FileNotFoundError:
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
archive = lock = None
|
||||
try:
|
||||
try:
|
||||
archive = os.open("archive-v0", _DIRECTORY, dir_fd=cache_fd)
|
||||
except FileNotFoundError:
|
||||
return True
|
||||
lock = os.open(".lock", os.O_RDWR | os.O_CREAT | _NOFOLLOW, 0o666, dir_fd=cache_fd)
|
||||
if not lock_fd(lock, wait=True, timeout=_LOCK_WAIT_SECONDS):
|
||||
return False
|
||||
with os.scandir(archive) as entries:
|
||||
buckets = [entry.name for entry in entries if entry.is_dir(follow_symlinks=False)]
|
||||
for bucket in buckets:
|
||||
try:
|
||||
bucket_fd = os.open(bucket, _DIRECTORY, dir_fd=archive)
|
||||
except OSError:
|
||||
continue
|
||||
try:
|
||||
doomed = bool(_damaged_packages_at(bucket_fd) & names)
|
||||
except OSError:
|
||||
doomed = False
|
||||
finally:
|
||||
os.close(bucket_fd)
|
||||
if doomed:
|
||||
shutil.rmtree(bucket, dir_fd=archive, ignore_errors=True)
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
finally:
|
||||
for fd in (lock, archive, cache_fd):
|
||||
if fd is not None:
|
||||
os.close(fd)
|
||||
archive = _archive_root(cache)
|
||||
if not names or os.name == "nt" or archive is None:
|
||||
return
|
||||
with open(cache / ".lock", "a+b") as lock:
|
||||
if not lock_fd(lock.fileno(), wait=True, timeout=_LOCK_WAIT_SECONDS):
|
||||
return
|
||||
for package, buckets in corrupt_cache_entries(cache).items():
|
||||
if package in names:
|
||||
for bucket in buckets:
|
||||
shutil.rmtree(archive / bucket, ignore_errors=True)
|
||||
@@ -20,34 +20,9 @@ import sys
|
||||
import pytest
|
||||
|
||||
from pm.environment import PythonEnvironment
|
||||
from pm.package import InstallError
|
||||
from pm.uv_cache_health import corrupt_cached_packages, failed_cached_packages
|
||||
from tests.pm._fixtures import _run, _wheel
|
||||
|
||||
LIVE_FAILURE = """\
|
||||
Building hermes-agent @ file:///opt/data/installs/351f/environments/g/workspace
|
||||
Built hermes-agent @ file:///opt/data/installs/351f/environments/g/workspace
|
||||
Prepared 1 package in 2.82s
|
||||
error: Failed to install: lxml-6.1.3-cp314-cp314-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl (lxml==6.1.3)
|
||||
cause: The wheel is invalid: Invalid Wheel-Version in `WHEEL` file: None
|
||||
"""
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("output", "expected"), [
|
||||
(LIVE_FAILURE, {"lxml"}),
|
||||
("error: Failed to install: Foo_Bar-1.0-py3-none-any.whl (foo-bar==1.0)\n"
|
||||
" cause: `RECORD` file is invalid\n cause: CSV deserialize error", {"foo-bar"}),
|
||||
("error: Failed to install: alpha-1.0-py3-none-any.whl (alpha==1.0)\n"
|
||||
" cause: failed to read directory `/c/archive-v0/waZDcAKYSZH8lwFY`: No such file or directory (os error 2)",
|
||||
{"alpha"}),
|
||||
("error: Failed to install: alpha-1.0-py3-none-any.whl (alpha==1.0)\n"
|
||||
" cause: failed to read directory `C:\\Users\\a\\cache\\uv\\archive-v0\\waZD`: "
|
||||
"The system cannot find the path specified. (os error 3)", {"alpha"}),
|
||||
])
|
||||
def test_cache_corruption_names_the_package(output, expected):
|
||||
assert failed_cached_packages(output) == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize("output", [
|
||||
"error: No solution found when resolving dependencies:\n Because alpha==2 was not found ...",
|
||||
"error: Failed to fetch: `https://pypi.org/simple/alpha/`\n Caused by: Request failed after 3 retries",
|
||||
@@ -143,203 +118,3 @@ def test_sync_recovers_from_nul_filled_cache_entries(project, tmp_path, monkeypa
|
||||
assert sum(c[0] == "sync" for c in calls) == 2
|
||||
assert corrupt_cached_packages(cache) == set()
|
||||
assert "alpha-dep, beta-dep is corrupt" in notes.getvalue()
|
||||
|
||||
|
||||
def test_a_wheel_broken_at_the_source_fails_once_with_a_clear_message(project, tmp_path, monkeypatch):
|
||||
"""Purging cannot fix a source that serves a broken wheel: one retry, then a named error."""
|
||||
source, uv, env, wheels = project
|
||||
cache = tmp_path / "cache"
|
||||
(wheels / "alpha_dep-1.0-py3-none-any.whl").unlink()
|
||||
_wheel(wheels, "alpha_dep", extra={"alpha_dep-1.0.dist-info/WHEEL": "\0" * 40})
|
||||
# The lock's hash would reject the new bytes before install; forget it so uv takes the wheel.
|
||||
(source / "uv.lock").unlink()
|
||||
_environment(uv, tmp_path, "relock", env).lock(source, timeout=120)
|
||||
shutil.rmtree(cache)
|
||||
calls = _record_uv_calls(monkeypatch)
|
||||
|
||||
broken = _environment(uv, tmp_path, "broken", env)
|
||||
broken.create()
|
||||
with pytest.raises(InstallError) as caught:
|
||||
broken.sync(source, timeout=120)
|
||||
|
||||
message = str(caught.value)
|
||||
assert "alpha-dep" in message and str(cache) in message and "still invalid" in message
|
||||
assert "broken wheel" in caught.value.remedy
|
||||
assert [c[:2] for c in calls].count(["cache", "clean"]) == 1
|
||||
assert sum(c[0] == "sync" for c in calls) == 2
|
||||
|
||||
|
||||
def test_a_failure_that_is_not_cache_corruption_is_not_retried(project, tmp_path, monkeypatch):
|
||||
"""A missing distribution (offline, nothing cached) fails once and never touches the cache."""
|
||||
source, uv, env, wheels = project
|
||||
cache = tmp_path / "cache"
|
||||
(wheels / "alpha_dep-1.0-py3-none-any.whl").unlink()
|
||||
shutil.rmtree(cache)
|
||||
calls = _record_uv_calls(monkeypatch)
|
||||
|
||||
missing = _environment(uv, tmp_path, "missing", env)
|
||||
missing.create()
|
||||
with pytest.raises(InstallError):
|
||||
missing.sync(source, timeout=120)
|
||||
|
||||
assert not any(c[:1] == ["cache"] for c in calls)
|
||||
assert sum(c[0] == "sync" for c in calls) == 1
|
||||
|
||||
|
||||
def test_doctor_reports_then_repairs_corrupt_cache_entries(project, tmp_path, monkeypatch, capsys):
|
||||
from pm import cli
|
||||
import pm.packages
|
||||
|
||||
_, uv, env, _ = project
|
||||
cache = tmp_path / "cache"
|
||||
_nul_fill(_cached(cache, "alpha_dep") / "METADATA")
|
||||
monkeypatch.setattr(pm.packages, "uv_cache_path", lambda: cache)
|
||||
monkeypatch.setattr("pm._uv._toolchain", lambda **kwargs: (uv, Path(sys.executable)))
|
||||
monkeypatch.setattr("pm.environment._base_environment", lambda base=None: dict(env))
|
||||
|
||||
assert cli._doctor_uv_cache(repair=False) == 1
|
||||
assert "alpha-dep" in capsys.readouterr().out
|
||||
assert corrupt_cached_packages(cache) == {"alpha-dep"} # read-only without --repair
|
||||
|
||||
assert cli._doctor_uv_cache(repair=True) == 0
|
||||
assert "removed corrupt entries for alpha-dep" in capsys.readouterr().out
|
||||
assert corrupt_cached_packages(cache) == set()
|
||||
assert cli._doctor_uv_cache(repair=False) == 0
|
||||
|
||||
|
||||
def _bucket(cache: Path, name: str = "alpha_dep", files=None) -> Path:
|
||||
"""An archive entry by hand: ``files`` maps dist-info file name -> bytes."""
|
||||
info = cache / "archive-v0" / "bucket" / f"{name}-1.0.dist-info"
|
||||
info.mkdir(parents=True)
|
||||
for file, body in (files or {"WHEEL": b"\0" * 20, "METADATA": b"Name: x\n", "RECORD": b"a,,\n"}).items():
|
||||
(info / file).write_bytes(body)
|
||||
return info.parent
|
||||
|
||||
|
||||
@pytest.mark.parametrize("files", [
|
||||
{"WHEEL": b"wheel-version: 1.0\n", "METADATA": b"metadata-version: 2.1\nname: alpha-dep\n", "RECORD": b"a,,\n"},
|
||||
{"WHEEL": b"Wheel-Version: 1.0\n", "METADATA": b"Metadata-Version: 2.1\n" + b"X: y\n" * 20000 + b"Name: a\n",
|
||||
"RECORD": b"a,,\n"},
|
||||
{"WHEEL": b"X: y\n" * 20000 + b"Wheel-Version: 1.0\n", "METADATA": b"Name: a\n", "RECORD": b"a,,\n"},
|
||||
])
|
||||
def test_valid_entries_uv_accepts_are_never_flagged(tmp_path, files):
|
||||
"""Header names are case-insensitive and METADATA may be long; only torn writes count."""
|
||||
_bucket(tmp_path, files=files)
|
||||
assert corrupt_cached_packages(tmp_path) == set()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("damage", ["WHEEL", "METADATA", "RECORD", "missing-RECORD", "WHEEL-without-version",
|
||||
"RECORD-tail"])
|
||||
def test_torn_writes_are_flagged(tmp_path, damage):
|
||||
files = {"WHEEL": b"Wheel-Version: 1.0\n", "METADATA": b"Name: alpha-dep\n", "RECORD": b"a,,\n"}
|
||||
if damage == "RECORD-tail": # the first pages reached the disk, a later one did not
|
||||
files["RECORD"] = b"a/b.py,,\n" * 20000 + b"\0" * 4096
|
||||
elif damage == "missing-RECORD":
|
||||
del files["RECORD"]
|
||||
elif damage == "WHEEL-without-version":
|
||||
files["WHEEL"] = b"Root-Is-Purelib: true\n"
|
||||
else:
|
||||
files[damage] = b"\0" * 152
|
||||
_bucket(tmp_path, files=files)
|
||||
assert corrupt_cached_packages(tmp_path) == {"alpha-dep"}
|
||||
|
||||
|
||||
@pytest.mark.platforms("linux", "macos")
|
||||
@pytest.mark.parametrize("link", ["archive", "bucket", "dist-info"])
|
||||
def test_sweep_never_deletes_through_a_symlink(tmp_path, link):
|
||||
from pm.uv_cache_health import remove_corrupt_entries
|
||||
|
||||
cache = tmp_path / "cache"
|
||||
outside = tmp_path / "outside"
|
||||
victim = _bucket(outside) # a corrupt-looking tree outside the cache
|
||||
(victim / "user-data").write_text("must survive", encoding="utf-8")
|
||||
archive = cache / "archive-v0"
|
||||
if link == "archive":
|
||||
cache.mkdir()
|
||||
archive.symlink_to(outside / "archive-v0", target_is_directory=True)
|
||||
elif link == "bucket":
|
||||
archive.mkdir(parents=True)
|
||||
(archive / "bucket").symlink_to(victim, target_is_directory=True)
|
||||
else:
|
||||
(archive / "bucket").mkdir(parents=True)
|
||||
info = next(victim.glob("*.dist-info"))
|
||||
(archive / "bucket" / info.name).symlink_to(info, target_is_directory=True)
|
||||
|
||||
assert corrupt_cached_packages(cache) == set()
|
||||
remove_corrupt_entries(cache, {"alpha-dep"})
|
||||
assert (victim / "user-data").is_file() and next(victim.glob("*.dist-info")).is_dir()
|
||||
|
||||
|
||||
@pytest.mark.platforms("linux", "macos")
|
||||
def test_sweep_stays_in_the_cache_it_opened_when_the_path_is_swapped(tmp_path, monkeypatch):
|
||||
"""Replacing the cache path with a link mid-sweep must not redirect the delete."""
|
||||
import pm.uv_cache_health as health
|
||||
|
||||
cache = tmp_path / "cache"
|
||||
torn = _bucket(cache)
|
||||
outside = tmp_path / "outside"
|
||||
victim = _bucket(outside)
|
||||
(victim / "user-data").write_text("must survive", encoding="utf-8")
|
||||
moved = tmp_path / "cache-original"
|
||||
real_lock = health.lock_fd
|
||||
|
||||
def swap_then_lock(fd, **kwargs):
|
||||
cache.rename(moved)
|
||||
cache.symlink_to(outside, target_is_directory=True)
|
||||
return real_lock(fd, **kwargs)
|
||||
|
||||
monkeypatch.setattr(health, "lock_fd", swap_then_lock)
|
||||
assert health.remove_corrupt_entries(cache, {"alpha-dep"}) is True
|
||||
assert (victim / "user-data").is_file()
|
||||
assert not (moved / "archive-v0" / torn.name).exists()
|
||||
|
||||
|
||||
@pytest.mark.platforms("linux", "macos")
|
||||
def test_sweep_waits_for_uv_and_leaves_entries_while_uv_holds_the_cache(tmp_path, monkeypatch):
|
||||
"""uv holds a shared flock on <cache>/.lock while it runs; the sweep must not delete under it."""
|
||||
import fcntl
|
||||
|
||||
import pm.uv_cache_health as health
|
||||
|
||||
monkeypatch.setattr(health, "_LOCK_WAIT_SECONDS", 0.2)
|
||||
bucket = _bucket(tmp_path)
|
||||
with open(tmp_path / ".lock", "w", encoding="utf-8") as held:
|
||||
fcntl.flock(held.fileno(), fcntl.LOCK_SH)
|
||||
assert health.remove_corrupt_entries(tmp_path, {"alpha-dep"}) is False
|
||||
assert bucket.is_dir()
|
||||
assert health.remove_corrupt_entries(tmp_path, {"alpha-dep"}) is True
|
||||
assert not bucket.exists()
|
||||
|
||||
|
||||
def test_a_failed_purge_keeps_the_original_install_error(project, tmp_path, monkeypatch):
|
||||
source, uv, env, _ = project
|
||||
_nul_fill(_cached(tmp_path / "cache", "alpha_dep") / "WHEEL")
|
||||
real = PythonEnvironment._run
|
||||
|
||||
def refuse_clean(self, args, *, cwd, timeout):
|
||||
if args[:2] == ["cache", "clean"]:
|
||||
return subprocess.CompletedProcess(args, 2, "", "error: cache is read-only")
|
||||
return real(self, args, cwd=cwd, timeout=timeout)
|
||||
|
||||
monkeypatch.setattr(PythonEnvironment, "_run", refuse_clean)
|
||||
second = _environment(uv, tmp_path, "second", env)
|
||||
second.create()
|
||||
with pytest.raises(InstallError) as caught:
|
||||
second.sync(source, timeout=120)
|
||||
assert "cache is read-only" in str(caught.value)
|
||||
assert "The wheel is invalid" in str(caught.value)
|
||||
|
||||
|
||||
@pytest.mark.platforms("linux")
|
||||
@pytest.mark.parametrize("odd", ["WHEEL", "METADATA", "RECORD"])
|
||||
def test_sweep_closes_every_descriptor_when_a_file_is_a_directory(tmp_path, odd):
|
||||
from pm.uv_cache_health import remove_corrupt_entries
|
||||
|
||||
files = {"WHEEL": b"Wheel-Version: 1.0\n", "METADATA": b"Name: alpha-dep\n", "RECORD": b"a,,\n"}
|
||||
del files[odd]
|
||||
info = next(_bucket(tmp_path, files=files).glob("*.dist-info"))
|
||||
(info / odd).mkdir()
|
||||
before = len(os.listdir("/proc/self/fd"))
|
||||
for _ in range(20):
|
||||
remove_corrupt_entries(tmp_path, {"alpha-dep"})
|
||||
assert len(os.listdir("/proc/self/fd")) == before
|
||||
Reference in new issue
Block a user