diff --git a/pyproject.toml b/pyproject.toml index 2ac5a08..3151bf4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -42,5 +42,12 @@ packages = ["src/worker_bridge"] line-length = 110 target-version = "py310" +[tool.ruff.lint] +# The historical default selection, pinned explicitly: ruff 0.16 expanded +# its defaults substantially and began failing on already-shipped code +# (RUF022, RUF100, BLE001, ...). Keep lint behavior version-independent; +# re-expanding the rule set should be a deliberate future decision. +select = ["E4", "E7", "E9", "F"] + [tool.pytest.ini_options] testpaths = ["tests"] diff --git a/src/worker_bridge/workspace.py b/src/worker_bridge/workspace.py index 0e056f6..942f81a 100644 --- a/src/worker_bridge/workspace.py +++ b/src/worker_bridge/workspace.py @@ -2,6 +2,7 @@ from __future__ import annotations +import errno import fnmatch import hashlib import os @@ -12,8 +13,6 @@ from pathlib import Path from typing import Any -import psutil - from worker_bridge.environ import get_home as get_hermes_home @@ -169,6 +168,54 @@ def _reftx_hook_present(repository: Path) -> bool: return (common / "hooks" / "reference-transaction").exists() +def _claim_lock_file(fd: int) -> None: + """Atomically claim the lock file's first byte, or raise OSError. + + Windows: ``msvcrt.locking`` — the C runtime's ``_locking`` operation, + held per descriptor and released by the operating system when the + owning process exits. POSIX: ``fcntl.flock`` — held per open file + description with the same death-release guarantee. Both turn "is the + holder still alive?" into a kernel question. + """ + if os.name == "nt": + import msvcrt + + os.lseek(fd, 0, os.SEEK_SET) + msvcrt.locking(fd, msvcrt.LK_NBLCK, 1) + else: + import fcntl + + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + + +def _is_lock_contention(exc: OSError) -> bool: + """True when a claim failure means another holder owns the lock. + + POSIX ``flock`` reports contention as EWOULDBLOCK/EAGAIN (raised as + BlockingIOError) or EACCES. The Windows C runtime's ``_locking`` + reports contention as EACCES/EDEADLK; EBADF, EINVAL, and other errors + are genuine failures whose cause must survive. Anything not classified + as contention propagates unchanged. + """ + if isinstance(exc, BlockingIOError): + return True + if os.name == "nt": + return exc.errno in {errno.EACCES, getattr(errno, "EDEADLK", -1)} + return exc.errno == errno.EACCES + + +def _release_lock_file(fd: int) -> None: + if os.name == "nt": + import msvcrt + + os.lseek(fd, 0, os.SEEK_SET) + msvcrt.locking(fd, msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(fd, fcntl.LOCK_UN) + + class RepositoryLock: """Process and cross-process lock used by direct-workspace mode. @@ -178,9 +225,25 @@ class RepositoryLock: not serialize on one global per-repo lock. Same operation on the same repository still excludes. - ``wait_seconds`` bounds how long acquisition waits for a live holder to - release before raising. The default (0) keeps the historical fail-fast - behavior direct mode relies on. + Cross-process exclusion is a kernel-mediated byte lock on a small + per-key lock file (``msvcrt.locking`` on Windows, ``fcntl.flock`` on + POSIX). The kernel releases it when the owning process dies, so a + crashed holder needs no stale-file sweep — reading a pid from the file + and unlinking it (the previous scheme) could not distinguish a dead + holder's file from one a peer had just legitimately recreated, and the + resulting unlink could delete a live lock. Lock files are therefore + never unlinked: they are permanent claim points, one per repository + (and operation), a few bytes each. + + ``wait_seconds`` bounds how long acquisition waits for a *cross-process* + holder to release before raising; the default (0) fails fast. In-process + contention on the same key waits up to ``max(30, wait_seconds)`` seconds + on the thread lock first — parallel worktree setup relies on that floor. + + Behavior note: the pid written to the lock file is diagnostic only. A + failure to record it is ignored instead of aborting acquisition — the + previous implementation raised from the pid write, so this is a + deliberate behavior change. """ def __init__( @@ -202,39 +265,58 @@ def __init__( def __enter__(self) -> "RepositoryLock": if not self._thread_lock.acquire(timeout=max(30.0, self._wait_seconds)): raise WorkspaceError("repository is busy") - self._path.parent.mkdir(parents=True, exist_ok=True) - deadline = time.monotonic() + self._wait_seconds - swept_stale = False - while True: - try: - self._fd = os.open(self._path, os.O_CREAT | os.O_EXCL | os.O_WRONLY) - os.write(self._fd, str(os.getpid()).encode()) - return self - except FileExistsError as exc: - stale = False + owned = False + fd: int | None = None + try: + self._path.parent.mkdir(parents=True, exist_ok=True) + deadline = time.monotonic() + self._wait_seconds + while True: + candidate = os.open(self._path, os.O_CREAT | os.O_RDWR, 0o644) try: - pid = int(self._path.read_text(encoding="ascii").strip()) - stale = not psutil.pid_exists(pid) - except (OSError, ValueError): - stale = True - if stale and not swept_stale: - swept_stale = True - self._path.unlink(missing_ok=True) - continue - if time.monotonic() < deadline: + _claim_lock_file(candidate) + except OSError as exc: + # Drop the claiming fd; a close failure here must not + # replace the claim's error, which carries the real cause. + try: + os.close(candidate) + except OSError: + pass + if not _is_lock_contention(exc): + raise + if time.monotonic() >= deadline: + raise WorkspaceError(f"repository lock exists: {self._path}") from exc time.sleep(0.2) - # The holder may die while we wait; allow another sweep. - swept_stale = False continue - self._thread_lock.release() - raise WorkspaceError(f"repository lock exists: {self._path}") from exc + fd = candidate + break + self._fd = fd + owned = True + try: + # Diagnostics only: the owning pid for humans inspecting the + # locks directory. Never gates ownership. + os.ftruncate(fd, 0) + os.write(fd, str(os.getpid()).encode()) + except OSError: + pass + return self + finally: + if not owned: + try: + if fd is not None: + os.close(fd) + finally: + self._thread_lock.release() def __exit__(self, *_exc: Any) -> None: - if self._fd is not None: - os.close(self._fd) - self._fd = None - self._path.unlink(missing_ok=True) - self._thread_lock.release() + try: + if self._fd is not None: + fd, self._fd = self._fd, None + try: + _release_lock_file(fd) + finally: + os.close(fd) + finally: + self._thread_lock.release() class WorkspaceManager: diff --git a/tests/_lock_child.py b/tests/_lock_child.py new file mode 100644 index 0000000..f658368 --- /dev/null +++ b/tests/_lock_child.py @@ -0,0 +1,110 @@ +"""Subprocess contender used by RepositoryLock concurrency tests. + +Driven by a JSON config file so the parent test can orchestrate real +cross-process contention deterministically: + + python _lock_child.py + +Config keys: + repo, operation, wait_seconds -- lock identity and wait budget + signaldir -- directory for markers/outcomes + id -- this child's marker suffix ("0"/"1") + +Optional deterministic seams (inert unless the implementation calls the +patched functions during acquisition): + barrier_pid_exists -- synchronize BOTH children at the stale-pid + observation point before psutil.pid_exists returns + wait_peer_created_then_unlink -- before unlinking the lock path, block + until the peer reports holding the lock (forces + the stale-sweep to run against a *live* lock) + +Outcome protocol: writes ``outcome-.json`` ({acquired, error}); when +acquired, also writes ``holding-`` and keeps the lock until a +``release-`` file appears (polling with hard caps so nothing hangs). +""" +from __future__ import annotations + +import json +import os +import sys +import time +from pathlib import Path + + +def _await_file(path: Path, timeout: float = 15.0) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if path.exists(): + return True + time.sleep(0.02) + return False + + +def _patch_pid_exists_barrier(signaldir: Path, child_id: str, peer_id: str) -> None: + """Both children meet here before agreeing the pid is dead.""" + import psutil + + mine = signaldir / f"saw-stale-{child_id}" + peers = signaldir / f"saw-stale-{peer_id}" + + def synchronized(pid: int) -> bool: + mine.write_text(str(pid), encoding="ascii") + if not _await_file(peers): + raise RuntimeError("pid_exists barrier timed out") + return False # both contenders agree: stale + + psutil.pid_exists = synchronized + + +def _patch_unlink_waits_for_peer(signaldir: Path, child_id: str, peer_id: str, lock_path: Path) -> None: + """Force the stale-sweep unlink to run after the peer recreated the lock.""" + real_unlink = os.unlink + + def ordered_unlink(path, *args, **kwargs): + try: + if Path(path).resolve() == lock_path.resolve(): + if not _await_file(signaldir / f"holding-{peer_id}"): + raise RuntimeError("peer-created barrier timed out") + except OSError: + pass # not the lock path, or already gone: behave like unlink + return real_unlink(path, *args, **kwargs) + + os.unlink = ordered_unlink + + +def main() -> int: + config = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8")) + signaldir = Path(config["signaldir"]) + child_id = str(config["id"]) + peer_id = str(config["peer"]) + outcome_path = signaldir / f"outcome-{child_id}.json" + + from worker_bridge.workspace import RepositoryLock + + lock = RepositoryLock( + config["repo"], operation=config["operation"], wait_seconds=config["wait_seconds"] + ) + + if config.get("barrier_pid_exists"): + _patch_pid_exists_barrier(signaldir, child_id, peer_id) + if config.get("wait_peer_created_then_unlink"): + _patch_unlink_waits_for_peer(signaldir, child_id, peer_id, lock._path) + + try: + with lock: + (signaldir / f"holding-{child_id}").write_text("", encoding="ascii") + outcome_path.write_text( + json.dumps({"id": child_id, "acquired": True, "error": None}), encoding="utf-8" + ) + _await_file(signaldir / f"release-{child_id}", timeout=30.0) + return 0 + except Exception as exc: # noqa: BLE001 - report every failure shape to the parent + outcome_path.write_text( + json.dumps({"id": child_id, "acquired": False, "error": f"{type(exc).__name__}: {exc}"}), + encoding="utf-8", + ) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/lock_test_helpers.py b/tests/lock_test_helpers.py new file mode 100644 index 0000000..835f95c --- /dev/null +++ b/tests/lock_test_helpers.py @@ -0,0 +1,61 @@ +"""Shared helpers for cross-process RepositoryLock tests. + +Spawns real contender subprocesses (``_lock_child.py``) coordinated through +marker files, so mutual exclusion is tested against actual OS-level +processes rather than simulated lock-file contents. +""" +from __future__ import annotations + +import json +import subprocess +import sys +import time +from pathlib import Path + +CHILD = Path(__file__).parent / "_lock_child.py" + + +def child_config(tmp: Path, *, repo: Path, operation: str, wait_seconds: float, + id: str, peer: str, **overrides) -> Path: # noqa: A002 - test fixture ids + config = { + "repo": str(repo), + "operation": operation, + "wait_seconds": wait_seconds, + "signaldir": str(tmp / "signals"), + "id": id, + "peer": peer, + } + config.update(overrides) + (tmp / "signals").mkdir(parents=True, exist_ok=True) + path = tmp / f"config-{id}.json" + path.write_text(json.dumps(config), encoding="utf-8") + return path + + +def spawn(config: Path) -> subprocess.Popen: + return subprocess.Popen( + [sys.executable, str(CHILD), str(config)], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + + +def await_file(path: Path, timeout: float = 30.0) -> str: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if path.exists(): + return path.read_text(encoding="utf-8") + time.sleep(0.02) + raise AssertionError(f"timed out waiting for {path.name}") + + +def outcome(tmp: Path, child_id: str) -> dict: + return json.loads(await_file(tmp / "signals" / f"outcome-{child_id}.json")) + + +def release_and_join(tmp: Path, children: list[subprocess.Popen]) -> None: + for child_id in ("0", "1"): + (tmp / "signals" / f"release-{child_id}").write_text("", encoding="ascii") + for proc in children: + proc.wait(timeout=30) diff --git a/tests/test_hardening.py b/tests/test_hardening.py index 9a4ad2e..9d5bbce 100644 --- a/tests/test_hardening.py +++ b/tests/test_hardening.py @@ -31,7 +31,7 @@ from worker_bridge.adapters.mock import MockWorkerAdapter from worker_bridge.orchestrator import WorkerBridge -from worker_bridge.redaction import redact, redact_text +from worker_bridge.redaction import redact_text from worker_bridge.registry import WorkerRegistry from worker_bridge.store import WorkerStore from worker_bridge.workspace import WorkspaceManager @@ -80,7 +80,7 @@ def _bridge(store: WorkerStore, adapter: MockWorkerAdapter, root: Path, **kwargs def test_recover_running_leaves_live_pid_untouched(tmp_path: Path, repository: Path): store = WorkerStore(tmp_path / "s.db") - task = store.create_task(_spec(repository, task_id="live")) + store.create_task(_spec(repository, task_id="live")) # A task owned by THIS (alive) process must survive a peer's recovery sweep. store.update_task("live", status="running", runtime={"pid": os.getpid()}) assert store.recover_running() == 0 @@ -99,7 +99,7 @@ def test_constructing_a_second_bridge_does_not_pause_live_task(tmp_path: Path, r store = WorkerStore(tmp_path / "s.db") adapter = MockWorkerAdapter() b1 = _bridge(store, adapter, tmp_path / "wt") - task = b1.create_task(_spec(repository, task_id="t")) + b1.create_task(_spec(repository, task_id="t")) store.update_task("t", status="running", runtime={"pid": os.getpid()}) # Any other `hermes worker ...` command / runner spawn builds a bridge: _bridge(WorkerStore(store.path), MockWorkerAdapter(), tmp_path / "wt2") diff --git a/tests/test_repository_lock.py b/tests/test_repository_lock.py index c97f6bb..aa4e42e 100644 --- a/tests/test_repository_lock.py +++ b/tests/test_repository_lock.py @@ -1,12 +1,15 @@ -"""RepositoryLock operation scoping, bounded wait, and worktree-setup serialization.""" +"""RepositoryLock operation scoping, stale recovery, and worktree-setup serialization. + +Fail-fast against a live holder and bounded waiting are covered by real +cross-process holder tests in ``test_repository_lock_concurrency.py`` — +simulating a holder by writing a pid into the lock file no longer +represents holding now that exclusion is a kernel-mediated byte lock. +""" from __future__ import annotations import asyncio -import os import subprocess -import threading -import time from pathlib import Path import pytest @@ -15,7 +18,7 @@ from worker_bridge.orchestrator import WorkerBridge from worker_bridge.registry import WorkerRegistry from worker_bridge.store import WorkerStore -from worker_bridge.workspace import RepositoryLock, WorkspaceError, WorkspaceManager +from worker_bridge.workspace import RepositoryLock, WorkspaceManager @pytest.fixture @@ -43,52 +46,18 @@ def test_operation_scoped_lock_does_not_exclude_plain_lock(repository: Path): pass -def test_same_operation_excludes_and_fail_fast_raises(repository: Path): - lock = RepositoryLock(repository, operation="worktree-setup") - with lock: - # Simulate a foreign live holder: the file exists and names a live - # pid, and we bypass the in-process thread lock by making a second - # lock object target the same file via a fresh key... the thread lock - # is shared per key, so use the lock FILE directly. - pass - # Foreign live holder: create the lock file with our own (live) pid. - holder = RepositoryLock(repository, operation="worktree-setup") - holder._path.parent.mkdir(parents=True, exist_ok=True) - holder._path.write_text(str(os.getpid()), encoding="ascii") - try: - with pytest.raises(WorkspaceError, match="repository lock exists"): - # thread lock is free (we never entered holder), so this exercises - # the file-lock fail-fast path against a live foreign pid. - with RepositoryLock(repository, operation="worktree-setup"): - pass - finally: - holder._path.unlink(missing_ok=True) - - -def test_wait_seconds_outlasts_a_transient_holder(repository: Path): - lock = RepositoryLock(repository, operation="worktree-setup", wait_seconds=10) - lock._path.parent.mkdir(parents=True, exist_ok=True) - lock._path.write_text(str(os.getpid()), encoding="ascii") # live foreign holder - - def release_soon() -> None: - time.sleep(0.5) - lock._path.unlink(missing_ok=True) - - thread = threading.Thread(target=release_soon) - thread.start() - started = time.monotonic() - try: - with lock: - waited = time.monotonic() - started - finally: - thread.join() - assert waited >= 0.4 # actually waited for the holder, didn't raise +# Fail-fast against a live foreign holder and bounded waiting under a +# transient holder moved to test_repository_lock_concurrency.py, where real +# subprocess holders exercise them (a lock file containing a live pid is no +# longer equivalent to holding the lock). -def test_stale_lock_from_dead_pid_is_swept(repository: Path): +def test_stale_lock_file_from_dead_pid_is_recoverable(repository: Path): lock = RepositoryLock(repository, operation="worktree-setup") lock._path.parent.mkdir(parents=True, exist_ok=True) - # A pid that cannot be alive (way beyond any real pid table on CI). + # A pid that cannot be alive (way beyond any real pid table on CI). With + # kernel-mediated exclusion the leftover record claims nothing: the file + # is only ever a claim point, so acquisition succeeds immediately. lock._path.write_text("999999999", encoding="ascii") with lock: pass # acquired despite the leftover file diff --git a/tests/test_repository_lock_concurrency.py b/tests/test_repository_lock_concurrency.py new file mode 100644 index 0000000..276b91c --- /dev/null +++ b/tests/test_repository_lock_concurrency.py @@ -0,0 +1,359 @@ +"""RepositoryLock concurrency regression tests (cross-process, deterministic). + +Reproductions for the two reported defects, plus the invariants they must +keep after the fix: + +1. Stale-sweep race (cross-process, forced interleaving): + two contenders both observe a planted stale lock; the losing contender's + sweep is forced to run *after* the winner recreated the lock, so a + check-then-unlink implementation deletes a live lock and both processes + end up inside the critical section. +2. Leaked in-process thread lock: exceptions between thread-lock acquisition + and cross-process ownership (os.open / os.write failures) used to escape + without releasing the thread lock. + +Determinism: all cross-process coordination uses marker files and barriers +injected at the exact code boundary under test — never sleep-and-hope. +""" + +from __future__ import annotations + +import errno +import os +import threading +import time +from pathlib import Path + +import pytest + +from worker_bridge import workspace +from worker_bridge.workspace import RepositoryLock, WorkspaceError + +from lock_test_helpers import await_file, child_config, outcome, release_and_join, spawn + +OPERATION = "concurrency-test" + + +@pytest.fixture +def repo(tmp_path: Path) -> Path: + repo = tmp_path / "repo" + repo.mkdir() + return repo + + +def _plant_stale_lock(repo: Path) -> None: + probe = RepositoryLock(repo, operation=OPERATION) + probe._path.parent.mkdir(parents=True, exist_ok=True) + probe._path.write_text("999999999", encoding="ascii") + + +# --- Finding 1: the stale-sweep race ----------------------------------------- + + +def test_two_contenders_never_both_hold_after_stale_sweep(tmp_path: Path, repo: Path) -> None: + # Both contenders synchronize at the stale-pid observation; contender 1's + # sweep is then forced to run only after contender 0 holds the lock, so a + # check-then-unlink implementation necessarily deletes a live lock. + _plant_stale_lock(repo) + child0 = spawn(child_config( + tmp_path, repo=repo, operation=OPERATION, wait_seconds=0, + id="0", peer="1", barrier_pid_exists=True)) + child1 = spawn(child_config( + tmp_path, repo=repo, operation=OPERATION, wait_seconds=0, + id="1", peer="0", barrier_pid_exists=True, wait_peer_created_then_unlink=True)) + try: + outcomes = [outcome(tmp_path, "0"), outcome(tmp_path, "1")] + finally: + release_and_join(tmp_path, [child0, child1]) + + acquired = [o["acquired"] for o in outcomes] + assert acquired.count(True) == 1, f"mutual exclusion violated, both contenders hold: {outcomes}" + loser = next(o for o in outcomes if not o["acquired"]) + assert "repository lock exists" in (loser["error"] or "") + holding = [p for p in (tmp_path / "signals").glob("holding-*") if p.exists()] + assert len(holding) == 1, "exactly one contender may enter the critical section" + + +def test_waiter_does_not_delete_replacement_lock(tmp_path: Path, repo: Path) -> None: + # Symmetric variant of the race: both contenders carry a wait budget, so + # the loser genuinely retries while the winner holds — and whichever + # child wins, the winner must stay undisturbed until released. The test + # must not assume which contender wins the (fair) initial claim. + _plant_stale_lock(repo) + child0 = spawn(child_config( + tmp_path, repo=repo, operation=OPERATION, wait_seconds=1, + id="0", peer="1", barrier_pid_exists=True)) + child1 = spawn(child_config( + tmp_path, repo=repo, operation=OPERATION, wait_seconds=1, + id="1", peer="0", barrier_pid_exists=True, wait_peer_created_then_unlink=True)) + try: + outcomes = [outcome(tmp_path, "0"), outcome(tmp_path, "1")] + finally: + release_and_join(tmp_path, [child0, child1]) + + acquired = [o for o in outcomes if o["acquired"]] + failed = [o for o in outcomes if not o["acquired"]] + assert len(acquired) == 1 and len(failed) == 1, outcomes + assert "repository lock exists" in (failed[0]["error"] or "") + # The winner survived the loser's failed attempts: it was still holding + # when released (its outcome was written from inside the critical section). + assert (tmp_path / "signals" / f"holding-{acquired[0]['id']}").exists() + + +# --- Finding 2: leaked in-process thread lock --------------------------------- + + +def _thread_lock_for(repo: Path, operation: str = "leak-test") -> threading.Lock: + return RepositoryLock(repo, operation=operation)._thread_lock + + +def test_failed_open_releases_thread_lock(repo: Path, monkeypatch) -> None: + def injected_open(*_args, **_kwargs): + raise OSError(13, "injected os.open failure") + + monkeypatch.setattr(workspace.os, "open", injected_open) + with pytest.raises(OSError, match="injected"): + with RepositoryLock(repo, operation="leak-test"): + pass + monkeypatch.undo() + + thread_lock = _thread_lock_for(repo) + assert not thread_lock.locked(), "thread lock leaked by a failed acquisition" + assert thread_lock.acquire(blocking=False), "same key must be immediately re-lockable" + thread_lock.release() + with RepositoryLock(repo, operation="leak-test"): + pass # full acquisition works right after the failure + + +def test_failed_write_does_not_break_acquisition_or_leak(repo: Path, monkeypatch) -> None: + def injected_write(*_args, **_kwargs): + raise OSError(5, "injected os.write failure") + + monkeypatch.setattr(workspace.os, "write", injected_write) + # Ownership must not depend on the diagnostic pid write: acquisition + # succeeds, the failure is contained, and nothing leaks. + with RepositoryLock(repo, operation="leak-test"): + pass + monkeypatch.undo() + + assert not _thread_lock_for(repo).locked(), "thread lock leaked despite successful ownership" + with RepositoryLock(repo, operation="leak-test"): + pass + + +# --- Preserved invariants ------------------------------------------------------ + + +def test_live_foreign_holder_blocks_fail_fast_and_survives(tmp_path: Path, repo: Path) -> None: + child = spawn(child_config( + tmp_path, repo=repo, operation=OPERATION, wait_seconds=5, id="0", peer="1")) + try: + await_file(tmp_path / "signals" / "holding-0") + with pytest.raises(WorkspaceError, match="repository lock exists"): + with RepositoryLock(repo, operation=OPERATION): + pass + assert (tmp_path / "signals" / "holding-0").exists(), "waiter disturbed a live holder" + finally: + release_and_join(tmp_path, [child]) + assert outcome(tmp_path, "0")["acquired"] is True + with RepositoryLock(repo, operation=OPERATION): + pass # lock is reusable immediately after the holder exits + + +def test_wait_seconds_outlasts_a_transient_real_holder(tmp_path: Path, repo: Path) -> None: + child = spawn(child_config( + tmp_path, repo=repo, operation=OPERATION, wait_seconds=5, id="0", peer="1")) + try: + await_file(tmp_path / "signals" / "holding-0") + started = time.monotonic() + timer = threading.Timer(0.4, lambda: (tmp_path / "signals" / "release-0").write_text("")) + timer.start() + with RepositoryLock(repo, operation=OPERATION, wait_seconds=10): + waited = time.monotonic() - started + timer.join() + finally: + release_and_join(tmp_path, [child]) + assert waited >= 0.3, "waiter did not actually wait for the live holder" + + +def test_exit_releases_after_exception_in_block(repo: Path) -> None: + lock = RepositoryLock(repo, operation="cleanup-test") + with pytest.raises(ValueError, match="boom"): + with lock: + raise ValueError("boom") + assert not lock._thread_lock.locked() + with RepositoryLock(repo, operation="cleanup-test"): + pass + + +def test_uncontended_acquisition_is_reusable(repo: Path) -> None: + lock = RepositoryLock(repo, operation="uncontended-test") + with lock: + assert lock._fd is not None + assert lock._fd is None + assert not lock._thread_lock.locked() + with RepositoryLock(repo, operation="uncontended-test"): + pass + + +def test_lock_records_owner_pid_diagnostic(repo: Path) -> None: + lock = RepositoryLock(repo, operation="diagnostic-test") + with lock: + pass + # The file persists after release (it is a claim point, not a marker), + # and on Windows the locked byte range is unreadable from other handles + # while held — so diagnostics are read after release, when free. + recorded = lock._path.read_text(encoding="ascii").strip() + assert recorded == str(os.getpid()), "lock file should carry the owning pid for diagnostics" + + +# --- Cleanup failures must not leak the thread lock --------------------------- + + +def test_exit_unlock_failure_still_releases_thread_lock(repo: Path, monkeypatch) -> None: + def failing_unlock(_fd: int) -> None: + raise OSError(errno.EIO, "injected unlock failure") + + monkeypatch.setattr(workspace, "_release_lock_file", failing_unlock) + lock = RepositoryLock(repo, operation="cleanup-test") + with pytest.raises(OSError, match="injected unlock"): + with lock: + pass + monkeypatch.undo() + + assert not lock._thread_lock.locked(), "unlock failure leaked the thread lock" + with RepositoryLock(repo, operation="cleanup-test"): + pass # key remains usable + + +def test_exit_close_failure_still_releases_thread_lock(repo: Path, monkeypatch) -> None: + real_close = workspace.os.close + leaked: list[int] = [] + + def failing_close(fd: int) -> None: + leaked.append(fd) + raise OSError(errno.EIO, "injected close failure") + + monkeypatch.setattr(workspace.os, "close", failing_close) + lock = RepositoryLock(repo, operation="cleanup-test") + with pytest.raises(OSError, match="injected close"): + with lock: + pass + monkeypatch.undo() + real_close(leaked[0]) # finish the cleanup the injection interrupted + + assert not lock._thread_lock.locked(), "close failure leaked the thread lock" + with RepositoryLock(repo, operation="cleanup-test"): + pass + + +def test_failed_acquisition_with_failing_cleanup_preserves_primary_error( + repo: Path, monkeypatch +) -> None: + # Partial initialization followed by cleanup failure: the claim error is + # the real cause and must survive; the thread lock must still be freed. + real_close = workspace.os.close + leaked: list[int] = [] + + def bad_claim(_fd: int) -> None: + raise OSError(errno.EBADF, "injected invalid descriptor") + + def failing_close(fd: int) -> None: + leaked.append(fd) + raise OSError(errno.EIO, "injected close failure") + + monkeypatch.setattr(workspace, "_claim_lock_file", bad_claim) + monkeypatch.setattr(workspace.os, "close", failing_close) + with pytest.raises(OSError) as caught: + with RepositoryLock(repo, operation="cleanup-test"): + pass + monkeypatch.undo() + real_close(leaked[0]) + + assert caught.value.errno == errno.EBADF, "primary claim error was replaced by cleanup noise" + assert not _thread_lock_for(repo, "cleanup-test").locked() + + +# --- Claim-error classification ------------------------------------------------- + + +def test_genuine_claim_failure_keeps_cause(repo: Path, monkeypatch) -> None: + def bad_claim(_fd: int) -> None: + raise OSError(errno.EBADF, "injected invalid descriptor") + + monkeypatch.setattr(workspace, "_claim_lock_file", bad_claim) + with pytest.raises(OSError) as caught: + with RepositoryLock(repo, operation="claim-test"): + pass + monkeypatch.undo() + + assert caught.value.errno == errno.EBADF, "genuine claim failure must not masquerade as busy" + assert not _thread_lock_for(repo, "claim-test").locked() + with RepositoryLock(repo, operation="claim-test"): + pass + + +def test_contention_claim_failure_maps_to_busy(repo: Path, monkeypatch) -> None: + def contended_claim(_fd: int) -> None: + raise BlockingIOError(errno.EAGAIN, "injected contention") + + monkeypatch.setattr(workspace, "_claim_lock_file", contended_claim) + with pytest.raises(WorkspaceError, match="repository lock exists"): + with RepositoryLock(repo, operation="claim-test", wait_seconds=0): + pass + monkeypatch.undo() + + assert not _thread_lock_for(repo, "claim-test").locked() + with RepositoryLock(repo, operation="claim-test"): + pass + + +def test_claim_error_classification_by_platform(monkeypatch) -> None: + # POSIX: EAGAIN (BlockingIOError) and EACCES are contention; EBADF/EINVAL + # are not. Windows: EACCES/EDEADLK are contention; EBADF/EINVAL are not. + monkeypatch.setattr(workspace.os, "name", "posix") + assert workspace._is_lock_contention(BlockingIOError(errno.EAGAIN, "x")) + assert workspace._is_lock_contention(OSError(errno.EACCES, "x")) + assert not workspace._is_lock_contention(OSError(errno.EBADF, "x")) + assert not workspace._is_lock_contention(OSError(errno.EINVAL, "x")) + monkeypatch.setattr(workspace.os, "name", "nt") + assert workspace._is_lock_contention(OSError(errno.EACCES, "x")) + assert not workspace._is_lock_contention(OSError(errno.EBADF, "x")) + assert not workspace._is_lock_contention(OSError(errno.EINVAL, "x")) + + +# --- Documented in-process waiting floor --------------------------------------- + + +def test_in_process_contention_waits_for_holder_then_acquires(repo: Path) -> None: + # Same key, same process: the second acquisition waits on the thread lock + # (documented >=30s budget) instead of failing fast, and succeeds once + # the holder releases — verified without waiting anywhere near 30s. + holding, release = threading.Event(), threading.Event() + result: dict[str, object] = {} + + def holder() -> None: + with RepositoryLock(repo, operation="thread-test"): + holding.set() + release.wait(10) + + def waiter() -> None: + try: + with RepositoryLock(repo, operation="thread-test"): + result["waited"] = time.monotonic() - result["started"] # type: ignore[operator] + except Exception as exc: # noqa: BLE001 - surface any failure shape + result["error"] = exc + + t_holder = threading.Thread(target=holder) + t_holder.start() + assert holding.wait(5), "holder never acquired" + result["started"] = time.monotonic() + t_waiter = threading.Thread(target=waiter) + t_waiter.start() + time.sleep(0.3) # let the waiter actually contend, not race the holder + release.set() + t_holder.join(10) + t_waiter.join(10) + + assert "error" not in result, f"in-process contention must wait, not fail: {result}" + assert result["waited"] >= 0.25, f"waiter did not wait for the holder: {result}" diff --git a/tests/test_worker_bridge.py b/tests/test_worker_bridge.py index cc68cdb..297a8eb 100644 --- a/tests/test_worker_bridge.py +++ b/tests/test_worker_bridge.py @@ -3,7 +3,6 @@ import argparse import asyncio import json -import os import subprocess import sys from pathlib import Path