diff --git a/src/iai_mcp/_filelock.py b/src/iai_mcp/_filelock.py index db1d11a..15f4830 100644 --- a/src/iai_mcp/_filelock.py +++ b/src/iai_mcp/_filelock.py @@ -35,6 +35,13 @@ LOCK_NB = 4 LOCK_UN = 8 + # msvcrt has no shared locks, so LOCK_SH is serviced as an exclusive + # byte-range lock (see the module docstring). Callers doing in-place lock + # *conversion* need to know this: a "shared" hold here is already exclusive + # and cannot be co-held by another process, so escalating a shared lock this + # fd already owns is a relabel, not a contended acquire. + SHARED_IS_EXCLUSIVE = True + _LOCK_BYTES = 2**30 # Poll interval when emulating POSIX's block-until-acquired behaviour. _BLOCK_POLL_SECONDS = 0.05 @@ -93,8 +100,18 @@ def flock(fd: int, operation: int) -> None: LOCK_NB = _fcntl.LOCK_NB LOCK_UN = _fcntl.LOCK_UN + # POSIX shared locks are genuinely shared: escalation must still contend. + SHARED_IS_EXCLUSIVE = False + def flock(fd: int, operation: int) -> None: _fcntl.flock(fd, operation) -__all__ = ["flock", "LOCK_EX", "LOCK_NB", "LOCK_SH", "LOCK_UN"] +__all__ = [ + "flock", + "LOCK_EX", + "LOCK_NB", + "LOCK_SH", + "LOCK_UN", + "SHARED_IS_EXCLUSIVE", +] diff --git a/src/iai_mcp/hippo/_db.py b/src/iai_mcp/hippo/_db.py index 98b34fc..d313aea 100644 --- a/src/iai_mcp/hippo/_db.py +++ b/src/iai_mcp/hippo/_db.py @@ -19,7 +19,14 @@ import numpy as np import pyarrow as pa -from iai_mcp._filelock import LOCK_EX, LOCK_NB, LOCK_SH, LOCK_UN, flock +from iai_mcp._filelock import ( + LOCK_EX, + LOCK_NB, + LOCK_SH, + LOCK_UN, + SHARED_IS_EXCLUSIVE, + flock, +) from iai_mcp.crypto import ( decrypt_field, encrypt_field, @@ -351,10 +358,22 @@ def downgrade_to_shared(self) -> None: if held is None: return base_fd, refcount = held + # Downgrade EXCLUSIVE -> SHARED in place. On POSIX fcntl.flock + # converts the held lock atomically and never blocks (SHARED is + # weaker than the EXCLUSIVE this fd already owns). On Windows msvcrt + # has no shared locks and no atomic conversion: LOCK_SH is serviced + # as the *same* exclusive byte-range lock this fd already holds, so a + # *blocking* flock(LOCK_SH) polls LK_NBLCK forever against our own + # range -- wedging the daemon while _PROCESS_LOCKS_GUARD is held, + # which the liveness watchdog then force-kills. Use the non-blocking + # form and treat "already held" (EWOULDBLOCK) as success: we are + # relabelling a lock this fd already owns, so no OS re-acquisition is + # needed on either platform. try: - flock(base_fd, LOCK_SH) - except OSError: - return + flock(base_fd, LOCK_SH | LOCK_NB) + except OSError as exc: + if exc.errno not in (errno.EAGAIN, errno.EWOULDBLOCK): + return del _PROCESS_LOCKS[self._lock_key] _PROCESS_LOCKS_SHARED[self._lock_key] = (base_fd, refcount) self._access_mode = AccessMode.SHARED @@ -391,7 +410,17 @@ def escalate_to_exclusive(self, intent_budget_ms: int = 4000) -> None: deadline = time.monotonic() + intent_budget_ms / 1000.0 acquired = False - while time.monotonic() < deadline: + # On Windows the shim has no shared locks: a "shared" hold is already an + # exclusive msvcrt byte-range lock that no other process can co-hold, so + # if this fd already owns it (held is not None) we effectively already + # have exclusive access. A real flock(LOCK_EX) here would self-conflict + # against our own range and spuriously time out into HippoLockHeldError, + # so relabel without touching the OS lock. On POSIX shared locks are + # genuinely shared (another reader may hold one), so escalation must + # still contend for LOCK_EX below. + if SHARED_IS_EXCLUSIVE and held is not None: + acquired = True + while not acquired and time.monotonic() < deadline: try: flock(base_fd, LOCK_EX | LOCK_NB) acquired = True diff --git a/tests/test_hippo_lock_cycle_no_wedge.py b/tests/test_hippo_lock_cycle_no_wedge.py new file mode 100644 index 0000000..22be201 --- /dev/null +++ b/tests/test_hippo_lock_cycle_no_wedge.py @@ -0,0 +1,144 @@ +"""Regression guard: the daemon's hippo lock cycle must never block/wedge. + +Before the fix, ``HippoDB.downgrade_to_shared()`` called a *blocking* +``flock(base_fd, LOCK_SH)`` -- the only lock site in ``hippo/_db.py`` without +``LOCK_NB``. On Windows the ``_filelock`` shim services ``LOCK_SH`` as an +exclusive msvcrt byte-range lock with no atomic EX->SH conversion, so a blocking +``LOCK_SH`` on a fd that already holds ``LOCK_EX`` polled ``LK_NBLCK`` forever +against the daemon's own range -- wedging every thread under +``_PROCESS_LOCKS_GUARD`` until the liveness watchdog force-killed the daemon +(the recurring ``pre_kill_forensic_dump`` at ``downgrade_to_shared``). +``escalate_to_exclusive()`` had the symmetric self-conflict, bounded by its +deadline so it *degraded* into ``HippoLockHeldError`` rather than wedging. + +These tests run the exact daemon sequence (EXCLUSIVE -> downgrade -> escalate -> +downgrade). Crucially they run on **all** platforms -- including Windows, where +the wedge lived and where ``test_hippo_concurrency.py`` is skipped, which is why +the bug went uncaught for weeks. + +Every lock call -- including ``close()`` teardown -- runs in a watchdog thread +with a join timeout. A reintroduced blocking flock therefore makes an assertion +FAIL within the timeout instead of hanging the suite (the wedged thread holds +``_PROCESS_LOCKS_GUARD``, so an un-bounded ``close()`` would otherwise deadlock). +""" +from __future__ import annotations + +import threading +import time +from pathlib import Path + +from iai_mcp.hippo import AccessMode, HippoLockHeldError +from iai_mcp.hippo._db import HippoDB + +# A healthy conversion is sub-millisecond; escalate's own intent budget is 4s. +# 15s is far below the wedge's "forever" yet generous enough never to flake on a +# slow CI box. +_TIMEOUT_S = 15.0 + + +def _run_bounded(fn): + """Run ``fn()`` in a daemon thread joined with a timeout. + + Returns ``(completed, elapsed_s, error)``. ``completed is False`` means the + call did not return within the timeout -- i.e. it wedged. This mirrors how + the daemon invokes these conversions (via ``asyncio.to_thread`` on a pool + thread), and guarantees a regression fails the test rather than hanging it. + """ + box: dict = {} + + def _target() -> None: + t0 = time.monotonic() + try: + fn() + except BaseException as exc: # noqa: BLE001 -- surfaced to the assertion + box["error"] = exc + finally: + box["elapsed"] = time.monotonic() - t0 + + t = threading.Thread(target=_target, daemon=True) + t.start() + t.join(_TIMEOUT_S) + return (not t.is_alive()), box.get("elapsed"), box.get("error") + + +def _make_exclusive_db(tmp_path: Path) -> HippoDB: + return HippoDB( + path=tmp_path, + crypto_key_provider=lambda: b"0" * 32, + access_mode=AccessMode.EXCLUSIVE, + ) + + +def test_downgrade_to_shared_does_not_wedge_holding_exclusive(tmp_path: Path) -> None: + db = _make_exclusive_db(tmp_path) + try: + assert db._access_mode is AccessMode.EXCLUSIVE + + completed, elapsed, error = _run_bounded(db.downgrade_to_shared) + + assert completed, ( + f"downgrade_to_shared() did not return within {_TIMEOUT_S}s -- " + "the blocking-flock wedge is back" + ) + assert error is None, f"downgrade_to_shared() raised: {error!r}" + assert db._access_mode is AccessMode.SHARED + assert elapsed is not None and elapsed < 2.0, ( + f"downgrade unexpectedly slow ({elapsed:.3f}s): a bounded but " + "contended flock retry loop may have crept in" + ) + finally: + _run_bounded(db.close) # bounded so a wedge cannot deadlock teardown + + +def test_full_lock_cycle_no_wedge(tmp_path: Path) -> None: + """EXCLUSIVE -> downgrade(SHARED) -> escalate(EXCLUSIVE) -> downgrade(SHARED).""" + db = _make_exclusive_db(tmp_path) + try: + steps = ( + ("downgrade", db.downgrade_to_shared, AccessMode.SHARED), + ("escalate", db.escalate_to_exclusive, AccessMode.EXCLUSIVE), + ("downgrade", db.downgrade_to_shared, AccessMode.SHARED), + ) + for name, fn, expected in steps: + completed, elapsed, error = _run_bounded(fn) + assert completed, ( + f"{name}() did not return within {_TIMEOUT_S}s -- lock cycle wedged" + ) + assert error is None, f"{name}() raised: {error!r}" + assert db._access_mode is expected, ( + f"after {name}(): expected {expected}, got {db._access_mode}" + ) + assert elapsed is not None and elapsed < 5.0, ( + f"{name}() unexpectedly slow: {elapsed:.3f}s" + ) + finally: + _run_bounded(db.close) + + +def test_escalate_relabels_when_already_holding_lock(tmp_path: Path) -> None: + """escalate_to_exclusive() must reclaim EXCLUSIVE from a lock this fd already + holds without timing out into HippoLockHeldError -- the Windows self-conflict + that used to make sleep consolidation silently run without the EXCLUSIVE lock. + """ + db = _make_exclusive_db(tmp_path) + try: + # Reach SHARED first (the daemon's state before a sleep cycle). Bounded so + # a broken downgrade fails here rather than hanging the suite. + completed, _, _ = _run_bounded(db.downgrade_to_shared) + assert completed and db._access_mode is AccessMode.SHARED, "setup downgrade wedged" + + completed, elapsed, error = _run_bounded(db.escalate_to_exclusive) + + assert completed, f"escalate_to_exclusive() wedged (> {_TIMEOUT_S}s)" + assert not isinstance(error, HippoLockHeldError), ( + "escalate_to_exclusive() spuriously raised HippoLockHeldError on a " + "lock this fd already holds (Windows self-conflict regression)" + ) + assert error is None, f"escalate_to_exclusive() raised: {error!r}" + assert db._access_mode is AccessMode.EXCLUSIVE + assert elapsed is not None and elapsed < 5.0, ( + f"escalate unexpectedly slow ({elapsed:.3f}s): the intent-budget " + "deadline may be firing on a self-conflict" + ) + finally: + _run_bounded(db.close)