diff --git a/agent/trace_upload.py b/agent/trace_upload.py index 034147b6e8..31b3af5b84 100644 --- a/agent/trace_upload.py +++ b/agent/trace_upload.py @@ -211,7 +211,7 @@ def load_session_messages(session_id: str, db_path=None) -> Tuple[List[Dict[str, """``(messages, meta)`` from SQLite; ``meta`` is ``{}`` when the session row is missing (a live, untitled session may still have messages).""" from hermes_state_registry import acquire, release_or_close - db = acquire(db_path) if db_path else acquire() + db = acquire(db_path or None) try: resolved = db.resolve_session_id(session_id) or session_id meta = db.get_session(resolved) or {} diff --git a/hermes_cli/web_server_sessions.py b/hermes_cli/web_server_sessions.py index beaced31de..0a4464f799 100644 --- a/hermes_cli/web_server_sessions.py +++ b/hermes_cli/web_server_sessions.py @@ -216,20 +216,16 @@ def _maybe_auto_archive_for_profile(profile: Optional[str]) -> None: _last_auto_archive_check[key] = now from hermes_cli.config import load_config as _load_full_config - from hermes_state_registry import release_or_close cfg = (_load_full_config().get("sessions") or {}) if not cfg.get("auto_archive", False): return - # Bind the release helper BEFORE acquiring: the sweep's ``finally`` must - # never raise NameError over a held registry reference, or every eligible - # sweep leaks one and pins a retired generation open forever. db = _open_session_db_for_profile(profile, read_only=False) try: db.maybe_auto_archive( idle_days=float(cfg.get("auto_archive_days", 3)), min_interval_hours=int(cfg.get("min_interval_hours", 24))) finally: - release_or_close(db) + db.close() except Exception as exc: _log.debug("opportunistic auto-archive skipped: %s", exc) diff --git a/hermes_state_registry.py b/hermes_state_registry.py index ed4389c693..883e5da36d 100644 --- a/hermes_state_registry.py +++ b/hermes_state_registry.py @@ -219,55 +219,47 @@ def acquire(db_path: Optional[Path] = None) -> "SessionDB": if opening is None: opening = _opening[path] = threading.Event() lifecycle_lock = _path_lifecycle_lock_locked(path) - break - wait_for = opening - # Another caller is constructing this path; wait without holding the global - # lock. A failed opener signals too, so a waiter can retry. - wait_for.wait() + else: + wait_for = opening + if wait_for is not None: + # Another caller is constructing or closing this path; wait without holding the + # global lock. A failed opener signals too, so a waiter can retry. + wait_for.wait() + continue + + # Open OUTSIDE the registry lock; the per-path marker prevents redundant writers without + # serialising other files, the lifecycle mutex keeps the open off a same-path close. + try: + with lifecycle_lock: + db = _open_session_db(path) + db._shared_registry_owned = True + identity = _stat_db_file_identity(path) + except BaseException: + with _lock: + _finish_opening(path, opening) + raise - # Open OUTSIDE the lock; the per-path marker prevents redundant writers without - # serialising other files. - try: - # Serialize connection construction with a final close/checkpoint for - # this path, while keeping unrelated paths independent. - with lifecycle_lock: - db = _open_session_db(path) - db._shared_registry_owned = True - identity = _stat_db_file_identity(path) - except BaseException: with _lock: + teardown = _tearing_down.get(path) + if teardown is None: + existing = _generations.get(path) + if existing is not None: # Defensive: installed by explicit registry manipulation mid-open. + existing.refcount += 1 + winner = existing.db + else: + _generations[path] = _Generation(path, db, identity) + winner = db _finish_opening(path, opening) - raise - - discard_barrier: Optional[_TeardownBarrier] = None - with _lock: - teardown = _tearing_down.get(path) if teardown is not None: - # A shutdown or retired-generation final release began while this - # opener was constructing the handle. Do not publish a new - # generation into that teardown window; close this speculative - # connection and retry after the barrier. - discard_barrier = teardown - winner = None - else: - existing = _generations.get(path) - if existing is not None: # Defensive: installed by explicit registry manipulation mid-open. - existing.refcount += 1 - winner = existing.db - else: - _generations[path] = _Generation(path, db, identity) - winner = db - _finish_opening(path, opening) - if discard_barrier is not None: - with lifecycle_lock: - _teardown(db) - discard_barrier.event.wait() - return acquire(path) - assert winner is not None - if winner is not db: - with lifecycle_lock: - _teardown(db) - return winner + # A shutdown or retired-generation final release was admitted while this opener was + # constructing: never publish into that window — discard the speculative handle and + # go round again once the barrier lifts. + _teardown_generation(path, db) + teardown.event.wait() + continue + if winner is not db: + _teardown_generation(path, db) + return winner def release(db: "SessionDB") -> bool: @@ -278,7 +270,6 @@ def release(db: "SessionDB") -> bool: if db is None: return False key = id(db) - teardown_path: Optional[Path] = None teardown_barrier: Optional[_TeardownBarrier] = None with _lock: generation = _retired.get(key) @@ -293,22 +284,17 @@ def release(db: "SessionDB") -> bool: generation.refcount -= 1 needs_teardown = generation.refcount <= 0 if needs_teardown: - teardown_path = generation.path if generation.retired: _retired.pop(key, None) elif _generations.get(generation.path) is generation: _generations.pop(generation.path, None) - # Remove the lendable entry and admit this close in the SAME lock - # section, then keep the path blocked until checkpoint/close - # completes. A retired generation's drain is admitted too: it - # checkpoints and unlinks the same sidecars as the current one, so - # a replacement writer must not open on top of it. + # A retired generation's drain is admitted too: it checkpoints and unlinks the same + # sidecars as the current one, so a replacement writer must not open on top of it. teardown_barrier = _admit_teardown_locked(generation.path) # Teardown OUTSIDE the lock: stopping the token writer, WAL checkpoint and read-pool # drain must not block acquisition for every other state.db. if needs_teardown: - assert teardown_path is not None - _teardown_generation(teardown_path, db, barrier=teardown_barrier) + _teardown_generation(generation.path, db, barrier=teardown_barrier) return True @@ -340,11 +326,8 @@ def close_all() -> int: _teardown(generation.db) finally: _finish_teardown(path, teardown_barriers[path]) - # A concurrent final release may have removed its generation before this - # sweep took the registry lock. It still owns the physical close; wait for - # that barrier rather than returning while SQLite teardown is in flight. - # A barrier is lifted only once EVERY teardown admitted for its path has - # settled, so this cannot return over a close that is still running. + # A final release that removed its generation before this sweep took _lock still owns + # its physical close; wait for it rather than return over a running teardown. for barrier in active_teardowns: barrier.event.wait() return len(generations) diff --git a/tests/hermes_cli/test_web_server_auto_archive_registry.py b/tests/hermes_cli/test_web_server_auto_archive_registry.py deleted file mode 100644 index 29f891732f..0000000000 --- a/tests/hermes_cli/test_web_server_auto_archive_registry.py +++ /dev/null @@ -1,100 +0,0 @@ -"""The opportunistic auto-archive sweep must return its registry reference. - -``_maybe_auto_archive_for_profile`` borrows a shared writable ``SessionDB`` from -the registry and returns it in a ``finally``. Its whole body is wrapped in -``except Exception`` with a debug log, so a cleanup that raises is invisible: -the sweep still archives, the endpoint still answers, and only the refcount is -wrong. Every eligible sweep then leaks one reference, which pins the generation -open and defeats the physical teardown the registry owns (#102827 / #103118). - -Asserting on the archive result or on "the helper did not raise" cannot see -this. These regressions assert the refcount itself, on both the success and the -failure path, and that the last real holder can still tear the handle down. -""" - -from __future__ import annotations - -from pathlib import Path - -import pytest - -import hermes_state_registry as registry -import hermes_cli.web_server_sessions as sessions - - -@pytest.fixture(autouse=True) -def _reset_throttle(): - """The sweep is throttled per profile for 300s; tests must not inherit it.""" - sessions._last_auto_archive_check.clear() - yield - sessions._last_auto_archive_check.clear() - - -def _arm_sweep(monkeypatch, tmp_path: Path, *, enabled: bool = True) -> Path: - """Point one sweep at ``tmp_path/state.db`` with a config double.""" - db_path = tmp_path / "state.db" - config = { - "sessions": { - "auto_archive": enabled, - "auto_archive_days": 3, - "min_interval_hours": 0, - } - } - monkeypatch.setattr("hermes_cli.config.load_config", lambda *a, **kw: config) - monkeypatch.setattr( - sessions, "_open_session_db_for_profile", - lambda profile, *, read_only: registry.acquire(db_path)) - return db_path - - -def _refcount_for(db_path: Path) -> int: - generation = registry._generations.get(Path(db_path).resolve()) - return 0 if generation is None else generation.refcount - - -class TestAutoArchiveReleasesItsRegistryReference: - def test_successful_sweep_returns_the_borrowed_reference(self, tmp_path, monkeypatch): - db_path = _arm_sweep(monkeypatch, tmp_path) - holder = registry.acquire(db_path) - try: - before = _refcount_for(db_path) - sessions._maybe_auto_archive_for_profile(None) - assert _refcount_for(db_path) == before, "auto-archive leaked a registry reference" - finally: - assert registry.release(holder) is True - # The independent holder was the last one: teardown must now be reachable. - assert holder._conn is None - - def test_raising_sweep_still_returns_the_borrowed_reference(self, tmp_path, monkeypatch): - db_path = _arm_sweep(monkeypatch, tmp_path) - holder = registry.acquire(db_path) - - def _boom(self, **kwargs): - raise RuntimeError("archive failed") - - monkeypatch.setattr(type(holder), "maybe_auto_archive", _boom, raising=False) - try: - before = _refcount_for(db_path) - sessions._maybe_auto_archive_for_profile(None) - assert _refcount_for(db_path) == before, "failed sweep leaked a registry reference" - finally: - assert registry.release(holder) is True - assert holder._conn is None - - def test_repeated_eligible_sweeps_do_not_accumulate_references(self, tmp_path, monkeypatch): - db_path = _arm_sweep(monkeypatch, tmp_path) - holder = registry.acquire(db_path) - try: - before = _refcount_for(db_path) - for _ in range(3): - sessions._last_auto_archive_check.clear() - sessions._maybe_auto_archive_for_profile(None) - assert _refcount_for(db_path) == before - finally: - assert registry.release(holder) is True - assert holder._conn is None - - def test_disabled_auto_archive_never_acquires(self, tmp_path, monkeypatch): - db_path = _arm_sweep(monkeypatch, tmp_path, enabled=False) - sessions._maybe_auto_archive_for_profile(None) - assert registry._generations.get(Path(db_path).resolve()) is None diff --git a/tests/hermes_state/test_shared_session_db_registry.py b/tests/hermes_state/test_shared_session_db_registry.py index ce0fcde908..bc828b5f6a 100644 --- a/tests/hermes_state/test_shared_session_db_registry.py +++ b/tests/hermes_state/test_shared_session_db_registry.py @@ -17,6 +17,8 @@ Covers the three ownership invariants the PR review demanded: every state.db in the process). """ +import os +import shutil import threading import time from pathlib import Path @@ -34,6 +36,7 @@ def _clean_registry(): registry._retired.clear() registry._opening.clear() registry._tearing_down.clear() + registry._path_lifecycle_locks.clear() yield registry.close_all() registry._generations.clear() @@ -42,23 +45,20 @@ def _clean_registry(): registry._tearing_down.clear() -def _replace_file_preserving_schema(monkeypatch, db_path: Path, old_db=None) -> None: - """Simulate a replacement identity without unlinking an open SQLite file. +def _replace_file_preserving_schema(src: Path, dst: Path) -> None: + """Simulate snapshot-restore / recovery: new inode, same logical DB. - Windows correctly refuses to unlink a file with an open connection. The - registry contract is about observing a changed identity, so use its probe - seam here and, when requested, mark the old handle as replaced for the - SessionDB-level guard as well. + Copies the live DB to a temp name, removes the original, and renames + the copy into place — the replacement has a different inode. """ - original = registry._stat_db_file_identity(db_path) - replacement = (original[0] + 1, original[1]) if original is not None else (1, 1) - monkeypatch.setattr(registry, "_stat_db_file_identity", lambda _path: replacement) - if old_db is not None: - monkeypatch.setattr(old_db, "_db_file_was_replaced", lambda: True) + tmp = dst.with_suffix(".replacement.tmp") + shutil.copy2(dst, tmp) + os.unlink(dst) + os.rename(tmp, dst) class TestInodeReplacement: - def test_live_holders_keep_working_handle_across_replacement(self, tmp_path, monkeypatch): + def test_live_holders_keep_working_handle_across_replacement(self, tmp_path): """Two active refs → inode replacement → third caller gets NEW generation; the first two keep a working handle and their releases tear down only their own generation.""" @@ -68,7 +68,7 @@ class TestInodeReplacement: b = registry.acquire(db_path) assert a is b - _replace_file_preserving_schema(monkeypatch, db_path, old_db=a) + _replace_file_preserving_schema(db_path, db_path) c = registry.acquire(db_path) assert c is not a, "new caller must get the fresh generation" @@ -116,14 +116,14 @@ class TestInodeReplacement: assert stats["live_generations"] == 0 assert stats["retired_generations"] == 0 - def test_retired_generation_never_relent_even_after_drain(self, tmp_path, monkeypatch): + def test_retired_generation_never_relent_even_after_drain(self, tmp_path): """After replacement, repeated acquires all return the NEW generation — the retired one is never lent again, even while it still has live holders.""" db_path = tmp_path / "state.db" first = registry.acquire(db_path) - _replace_file_preserving_schema(monkeypatch, db_path) + _replace_file_preserving_schema(db_path, db_path) second = registry.acquire(db_path) third = registry.acquire(db_path) @@ -139,7 +139,7 @@ class TestInodeReplacement: db_path = tmp_path / "state.db" old = registry.acquire(db_path) - _replace_file_preserving_schema(monkeypatch, db_path) + _replace_file_preserving_schema(db_path, db_path) calls = {"n": 0} @@ -391,117 +391,6 @@ class TestTeardownOutsideLock: class TestLifecycleBarrier: - def test_acquire_waits_for_final_teardown(self, tmp_path, monkeypatch): - """A replacement writer cannot open while the old generation is closing.""" - db_path = tmp_path / "state.db" - db = registry.acquire(db_path) - teardown_started = threading.Event() - allow_teardown = threading.Event() - acquire_started = threading.Event() - acquire_finished = threading.Event() - release_finished = threading.Event() - acquired = [] - errors = [] - original_teardown = registry._teardown - - def _blocked_teardown(target): - teardown_started.set() - if not allow_teardown.wait(5.0): - raise AssertionError("timed out waiting to release teardown") - original_teardown(target) - - monkeypatch.setattr(registry, "_teardown", _blocked_teardown) - - def _release(): - try: - assert registry.release(db) is True - except BaseException as exc: # pragma: no cover - failure path - errors.append(exc) - finally: - release_finished.set() - - def _acquire(): - try: - acquire_started.set() - acquired.append(registry.acquire(db_path)) - except BaseException as exc: # pragma: no cover - failure path - errors.append(exc) - finally: - acquire_finished.set() - - release_thread = threading.Thread(target=_release) - release_thread.start() - assert teardown_started.wait(5.0) - - acquire_thread = threading.Thread(target=_acquire) - acquire_thread.start() - assert acquire_started.wait(2.0) - # The teardown gate is still closed, so this cannot have returned - # unless acquire bypassed the path barrier. - assert not acquire_finished.is_set() - - allow_teardown.set() - release_thread.join(10.0) - acquire_thread.join(10.0) - assert release_finished.is_set() - assert acquire_finished.is_set() - assert errors == [] - assert len(acquired) == 1 - assert acquired[0] is not db - assert registry.release(acquired[0]) is True - - def test_final_release_waits_for_inflight_write(self, tmp_path, monkeypatch): - """Physical close waits for a write holding the SessionDB lock.""" - db_path = tmp_path / "state.db" - db = registry.acquire(db_path) - write_started = threading.Event() - allow_write = threading.Event() - close_started = threading.Event() - release_finished = threading.Event() - errors = [] - original_teardown = registry._teardown - - def _observed_teardown(target): - close_started.set() - original_teardown(target) - - monkeypatch.setattr(registry, "_teardown", _observed_teardown) - - def _write(conn): - write_started.set() - if not allow_write.wait(5.0): - raise AssertionError("timed out waiting to finish write") - conn.execute("SELECT 1") - - def _run_write(): - try: - db._execute_write(_write) - except BaseException as exc: # pragma: no cover - failure path - errors.append(exc) - - def _release(): - try: - assert registry.release(db) is True - except BaseException as exc: # pragma: no cover - failure path - errors.append(exc) - finally: - release_finished.set() - - writer = threading.Thread(target=_run_write) - writer.start() - assert write_started.wait(5.0) - releaser = threading.Thread(target=_release) - releaser.start() - assert close_started.wait(5.0) - assert not release_finished.is_set() - - allow_write.set() - writer.join(10.0) - releaser.join(10.0) - assert errors == [] - assert release_finished.is_set() - assert db._conn is None - def test_maintenance_borrow_pins_connection_until_scope_exits(self, tmp_path): """Maintenance gets a temporary holder, not an unpinned snapshot.""" db_path = tmp_path / "state.db" @@ -648,7 +537,7 @@ class TestMultiGenerationTeardownBarrier: """Acquire a generation, replace the file identity, acquire its successor.""" db_path = tmp_path / "state.db" old = registry.acquire(db_path) - _replace_file_preserving_schema(monkeypatch, db_path, old) + _replace_file_preserving_schema(db_path, db_path) current = registry.acquire(db_path) assert current is not old return db_path, old, current @@ -735,51 +624,6 @@ class TestMultiGenerationTeardownBarrier: assert fresh is not current assert registry.release(fresh) is True - def test_current_release_does_not_lift_a_pending_retired_drain(self, tmp_path, monkeypatch): - """Converse ordering: the retired drain is the one still running.""" - db_path, old, current = self._two_generations(tmp_path, monkeypatch) - resolved = Path(db_path).resolve() - entered, resume = self._pause_teardown_of(monkeypatch, old) - errors = [] - releaser = self._release_async(old, errors) - acquire_done = threading.Event() - acquired = [] - try: - assert entered.wait(5.0) - - # The CURRENT generation's final release settles first. - assert registry.release(current) is True - assert current._conn is None - barrier = registry._tearing_down.get(resolved) - assert barrier is not None, "current release lifted the shared path barrier" - assert barrier.pending == 1 - assert not barrier.event.is_set() - - def _acquire(): - try: - acquired.append(registry.acquire(db_path)) - except BaseException as exc: # pragma: no cover - failure path - errors.append(exc) - finally: - acquire_done.set() - - opener = threading.Thread(target=_acquire, daemon=True) - opener.start() - assert not acquire_done.wait(0.5), "replacement published over a retired drain" - assert old._conn is not None - - resume.set() - assert acquire_done.wait(10.0) - opener.join(10.0) - finally: - resume.set() - releaser.join(10.0) - - assert errors == [] - assert old._conn is None - assert len(acquired) == 1 - assert registry.release(acquired[0]) is True - def test_failed_teardown_still_settles_the_barrier(self, tmp_path, monkeypatch): """A raising close must not strand the path barrier forever.""" db_path = tmp_path / "state.db" @@ -800,23 +644,3 @@ class TestMultiGenerationTeardownBarrier: fresh = registry.acquire(db_path) assert fresh is not db assert registry.release(fresh) is True - - def test_pending_teardown_does_not_block_an_unrelated_path(self, tmp_path, monkeypatch): - """Barrier accounting stays per-path; other state.db files keep moving.""" - blocked_path = tmp_path / "blocked" / "state.db" - blocked_path.parent.mkdir() - other_path = tmp_path / "other" / "state.db" - other_path.parent.mkdir() - blocked = registry.acquire(blocked_path) - entered, resume = self._pause_teardown_of(monkeypatch, blocked) - errors = [] - releaser = self._release_async(blocked, errors) - try: - assert entered.wait(5.0) - other = registry.acquire(other_path) - assert other is not blocked - assert registry.release(other) is True - finally: - resume.set() - releaser.join(10.0) - assert errors == [] diff --git a/tests/hermes_state/test_state_db_file_identity.py b/tests/hermes_state/test_state_db_file_identity.py index c2ba1e596c..3cc1ca1272 100644 --- a/tests/hermes_state/test_state_db_file_identity.py +++ b/tests/hermes_state/test_state_db_file_identity.py @@ -33,16 +33,18 @@ def _require_identity(db: SessionDB) -> None: pytest.skip("filesystem does not expose st_dev/st_ino for identity checks") -def _mark_replaced_handle(monkeypatch, db: SessionDB) -> None: - """Exercise the replacement guard without unlinking an open DB on Windows.""" - monkeypatch.setattr(db, "_db_file_was_replaced", lambda: True) - - -def test_replace_with_new_inode_fails_loudly_without_fts_repair(tmp_path, monkeypatch): +def test_replace_with_new_inode_fails_loudly_without_fts_repair(tmp_path): live = tmp_path / "state.db" + other = tmp_path / "other.db" db = _make_db(live, "live-sess", "original") _require_identity(db) - _mark_replaced_handle(monkeypatch, db) + alt = _make_db(other, "other-sess", "replacement") + alt.close() + + recorded = db._db_file_identity + assert recorded is not None + os.replace(other, live) + assert _stat_changed(live, recorded) with pytest.raises(StateDbReplacedError, match="replaced underneath"): db.append_message("live-sess", role="user", content="after-replace") @@ -54,11 +56,14 @@ def test_replace_with_new_inode_fails_loudly_without_fts_repair(tmp_path, monkey db.close() -def test_second_write_after_halt_does_not_attempt_repair(tmp_path, monkeypatch): +def test_second_write_after_halt_does_not_attempt_repair(tmp_path): live = tmp_path / "state.db" + other = tmp_path / "other.db" db = _make_db(live, "s", "a") _require_identity(db) - _mark_replaced_handle(monkeypatch, db) + alt = _make_db(other, "t", "b") + alt.close() + os.replace(other, live) with pytest.raises(StateDbReplacedError): db.append_message("s", role="user", content="first") with pytest.raises(StateDbReplacedError): @@ -120,14 +125,17 @@ def test_new_sessiondb_on_replaced_path_records_new_identity(tmp_path): reopened.close() -def test_fts_scoped_error_on_replaced_file_skips_fts_fail_open(tmp_path, monkeypatch): +def test_fts_scoped_error_on_replaced_file_skips_fts_fail_open(tmp_path): """Even FTS-provenance corruption must not authorize surgery on a replaced file. (A generic malformed error never reaches fail-open at all since the provenance classifier of #99652 rejects it earlier.)""" live = tmp_path / "state.db" + other = tmp_path / "other.db" db = _make_db(live, "s", "a") _require_identity(db) - _mark_replaced_handle(monkeypatch, db) + alt = _make_db(other, "t", "b") + alt.close() + os.replace(other, live) with pytest.raises(StateDbReplacedError): db._enter_fts_fail_open( @@ -179,6 +187,11 @@ def test_divert_session_transcript_jsonl_appends(tmp_path, monkeypatch): assert divert_session_transcript_jsonl("sess-jsonl", []) is None +def _stat_changed(path: Path, recorded) -> bool: + st = os.stat(path) + return (st.st_dev, st.st_ino) != recorded + + # --------------------------------------------------------------------------- # Lock safety of the identity probe itself (#100368 / howtocorrupt §2.2). #