From 3ae4ccd181f3aa57afa15d07b990c32f8e2d1d12 Mon Sep 17 00:00:00 2001 From: Joshua Barnes Date: Sun, 16 Aug 2026 21:26:35 -0500 Subject: [PATCH 1/4] fix(lock): replace stale-file sweep with kernel-mediated byte lock MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The stale-file sweep read a pid from the lock file and unlinked the path in separate steps, so a contender could unlink a live lock a peer had recreated after the contender's stale observation — two processes could hold the same key. Exceptional acquisition paths also leaked the in-process thread lock, and every claim error was reported as contention. Cross-process exclusion is now a kernel-mediated byte lock on a persistent per-key lock file (the C runtime _locking operation on Windows, fcntl.flock on POSIX); the OS releases it when the holder dies, so no sweep exists and lock files are never unlinked during acquisition or release. Cleanup is nested so the thread lock is released even when unlocking or closing fails, preserving the primary exception. Claim errors are classified: BlockingIOError/EACCES (POSIX) and EACCES/EDEADLK (Windows) are contention and map to the documented busy error; other errors keep their cause. The pid record is diagnostic-only: write failures no longer abort acquisition (previously they raised) — a deliberate behavior change. --- src/worker_bridge/workspace.py | 148 +++++++++++++++++++++++++-------- 1 file changed, 115 insertions(+), 33 deletions(-) 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: From 201cd3fe9215dcaac92ca138afb2dd1898e55cff Mon Sep 17 00:00:00 2001 From: Joshua Barnes Date: Sun, 16 Aug 2026 21:26:35 -0500 Subject: [PATCH 2/4] test(lock): deterministic cross-process race and leak regression coverage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Subprocess contenders coordinated by marker-file barriers force the stale-observation overlap that previously let both processes hold the lock; fault injection covers os.open, os.write, unlock, and close failures; errno classification is pinned for both platforms; the in-process waiting floor is exercised with bounded event coordination. Fail-fast and transient-holder tests now use real subprocess holders — writing a live pid into the lock file no longer represents holding the lock. Verified non-vacuous against the original implementation and a restored faulty __exit__ ordering. --- tests/_lock_child.py | 110 +++++++ tests/lock_test_helpers.py | 61 ++++ tests/test_repository_lock.py | 63 +--- tests/test_repository_lock_concurrency.py | 359 ++++++++++++++++++++++ 4 files changed, 546 insertions(+), 47 deletions(-) create mode 100644 tests/_lock_child.py create mode 100644 tests/lock_test_helpers.py create mode 100644 tests/test_repository_lock_concurrency.py 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_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}" From 5ebf973a3517b2df7ea73b1c3cec41552e2eb723 Mon Sep 17 00:00:00 2001 From: Joshua Barnes Date: Sun, 16 Aug 2026 21:26:35 -0500 Subject: [PATCH 3/4] chore(tests): drop unused imports and locals flagged by ruff Resolve the four ruff findings in the test suite: unused redact import, unused os import, and two never-read task bindings (side-effect calls kept). --- tests/test_hardening.py | 6 +++--- tests/test_worker_bridge.py | 1 - 2 files changed, 3 insertions(+), 4 deletions(-) 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_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 From bf6e6e9a594f15ba28ffc910048d02062585dc2e Mon Sep 17 00:00:00 2001 From: Joshua Barnes Date: Sun, 16 Aug 2026 21:50:53 -0500 Subject: [PATCH 4/4] chore(lint): pin ruff's rule selection, not its version Ruff 0.16 expanded its default rule set substantially, so an unpinned install (the cap resolved 0.15.22) began failing on already-shipped code (RUF022, RUF100, BLE001, ...). Select the historical default rules (E4, E7, E9, F) explicitly so lint behavior is version-independent and re-expanding the rule set is a deliberate future decision rather than release-note roulette. Reverts the incidental __all__ sorting and drops the <0.16 version cap. --- pyproject.toml | 7 +++++++ 1 file changed, 7 insertions(+) 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"]