diff --git a/agent/trace_upload.py b/agent/trace_upload.py index 84868bacb0..034147b6e8 100644 --- a/agent/trace_upload.py +++ b/agent/trace_upload.py @@ -210,17 +210,14 @@ def _do_upload(jsonl: str, *, token: str, session_id: str, dataset_name: str = D def load_session_messages(session_id: str, db_path=None) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]: """``(messages, meta)`` from SQLite; ``meta`` is ``{}`` when the session row is missing (a live, untitled session may still have messages).""" - from hermes_state import SessionDB - db = SessionDB(db_path=db_path) if db_path else SessionDB() + from hermes_state_registry import acquire, release_or_close + db = acquire(db_path) if db_path else acquire() try: resolved = db.resolve_session_id(session_id) or session_id meta = db.get_session(resolved) or {} return db.get_messages_as_conversation(resolved), meta finally: - try: - db.close() - except Exception: - logger.debug("Failed to close trace-upload SessionDB", exc_info=True) + release_or_close(db) def upload_session_trace( diff --git a/cron/scheduler.py b/cron/scheduler.py index c94562fe24..a30ca59aab 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -1739,6 +1739,7 @@ def _raise_inactivity_timeout(agent, job_name: str, limit_s: float) -> None: def _run_agent_with_watchdog( agent, prompt: str, job: dict, job_id: str, job_name: str, task_id: str, cancel_event, + worker_state: Optional[dict] = None, ) -> dict: """Run ``agent.run_conversation`` on a worker thread under the inactivity (not wall-clock) watchdog: default 600s, override HERMES_CRON_TIMEOUT, 0 = unlimited.""" @@ -1783,6 +1784,8 @@ def _run_agent_with_watchdog( _cron_context = contextvars.copy_context() _cron_future = _cron_pool.submit( _cron_context.run, agent.run_conversation, prompt, task_id=task_id) + if worker_state is not None: + worker_state["future"] = _cron_future _inactivity_timeout = False _watch_stop = threading.Event() @@ -1996,6 +1999,103 @@ def _finalize_cron_session(session_db, agent, job_id: str, job_name: str, cron_s logger.debug("Job '%s': failed to close SQLite session store: %s", job_id, e) +def _finalize_cron_session_db( + session_db, agent, job_id: str, job_name: str, cron_session_id: str, +) -> None: + """Finalize a cron session exactly once before releasing its registry reference.""" + _finalize_cron_session(session_db, agent, job_id, job_name, cron_session_id) + + +def _teardown_detached_cron_worker( + session_db, agent, job_id: str, job_name: str, cron_session_id: str, +) -> None: + """Release a timed-out worker's agent and SessionDB after it finishes.""" + try: + if session_db: + _finalize_cron_session_db( + session_db, agent, job_id, job_name, cron_session_id) + except BaseException as exc: + logger.error( + "Job '%s': detached worker session teardown failed: %s", + job_id, + exc, + exc_info=(type(exc), exc, exc.__traceback__), + ) + finally: + _teardown_cron_agent(agent, job_id) + + +def _defer_cron_worker_teardown_if_running( + worker_state: dict, + session_db, + agent, + job_id: str, + job_name: str, + cron_session_id: str, +) -> bool: + """Keep the agent and SessionDB alive until a timed-out worker finishes. + + ``ThreadPoolExecutor.shutdown(wait=False)`` does not stop a running + ``run_conversation`` call. Closing its SessionDB from ``run_job``'s + ``finally`` would therefore recreate the close-vs-write race this module + is meant to prevent. The real Future always supports ``add_done_callback``; + unknown test doubles are handled conservatively by waiting before inline + cleanup rather than closing a possibly-live handle. + """ + future = worker_state.get("future") + if future is None: + return False + done = getattr(future, "done", None) + if callable(done): + try: + if bool(done()): + return False + except Exception as exc: + logger.warning( + "Job '%s': could not determine worker completion; deferring teardown: %s", + job_id, + exc, + ) + + add_done_callback = getattr(future, "add_done_callback", None) + if not callable(add_done_callback): + logger.error( + "Job '%s': worker future cannot register teardown callback; waiting for worker", + job_id, + ) + try: + future.result() + except BaseException as exc: + logger.debug("Job '%s': detached worker completed with: %s", job_id, exc) + return False + + if worker_state.get("teardown_registered"): + return True + worker_state["teardown_registered"] = True + + def _finish_detached_worker(_future) -> None: + _teardown_detached_cron_worker( + session_db, agent, job_id, job_name, cron_session_id) + + try: + add_done_callback(_finish_detached_worker) + except Exception as exc: + # A real concurrent.futures.Future does not reject this call. If a + # custom future does, wait rather than releasing a live SQLite handle. + worker_state["teardown_registered"] = False + logger.error( + "Job '%s': failed to register detached-worker teardown; waiting: %s", + job_id, + exc, + ) + try: + future.result() + except BaseException as result_exc: + logger.debug("Job '%s': detached worker completed with: %s", job_id, result_exc) + return False + return True + + def _run_doc_header(job: dict, title: str, job_id: str, prompt: str) -> str: """Header of the persisted run document (title, ids, schedule, prompt).""" return ( @@ -2330,6 +2430,7 @@ def run_job( model = "" _session_db = None _audit: Optional[_FireAudit] = None + _worker_state: dict = {} scope = _CronRunScope(job, job_id, execution_id) try: scope.enter() @@ -2353,7 +2454,8 @@ def run_job( _audit = _FireAudit(job, job_id, model) result = _run_agent_with_watchdog( - agent, prompt, job, job_id, job_name, scope.task_id, cancel_event) + agent, prompt, job, job_id, job_name, scope.task_id, cancel_event, + worker_state=_worker_state) final_response = _final_response_from_result(result, job_id, job_name, AIAgent) # Keep final_response clean for delivery logic (empty = no delivery). logged_response = final_response if final_response else "(No response generated)" @@ -2375,9 +2477,17 @@ def run_job( return False, output, "", error_msg finally: + # A watchdog timeout only stops waiting; executor.shutdown(wait=False) + # cannot interrupt a worker already inside run_conversation. Keep both + # the agent and its registry-owned SessionDB alive until that worker's + # Future completes, otherwise its late persistence can race close/WAL + # checkpoint teardown. + _worker_teardown_deferred = _defer_cron_worker_teardown_if_running( + _worker_state, _session_db, agent, job_id, job_name, _cron_session_id) scope.exit() - if _session_db: - _finalize_cron_session(_session_db, agent, job_id, job_name, _cron_session_id) + if _session_db and not _worker_teardown_deferred: + _finalize_cron_session_db( + _session_db, agent, job_id, job_name, _cron_session_id) # Tear down the ephemeral agent or the gateway leaks fds per tick (EMFILE). With deferred # teardown, hand the live agent back: delivery needs a live async client. # Release subprocesses, terminal sandboxes, browser daemons, and the main OpenAI/httpx client held @@ -2385,11 +2495,12 @@ def run_job( # per job until it hits EMFILE (#10200 / "too many open files"). When the caller opted to defer # teardown (passed a list), hand the live agent back instead of closing it here — delivery must run # against a live async client, and the caller tears down afterwards (#58720). - if defer_agent_teardown is not None: - if agent is not None: - defer_agent_teardown.append(agent) - else: - _teardown_cron_agent(agent, job_id) + if not _worker_teardown_deferred: + if defer_agent_teardown is not None: + if agent is not None: + defer_agent_teardown.append(agent) + else: + _teardown_cron_agent(agent, job_id) def _teardown_cron_agent( diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py index f241380f69..34876332db 100644 --- a/gateway/platforms/api_server.py +++ b/gateway/platforms/api_server.py @@ -1637,14 +1637,14 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter): def _open_and_cache_session_db(self, home) -> Optional[Any]: """Cached SessionDB for ``home`` (shared by both ``_ensure_session_db*``). Never writes ``self._session_db`` (explicit override only), so no profile pins later requests.""" - from hermes_state import SessionDB + from hermes_state_registry import acquire key = str(home) with self._session_db_cache_lock: if self._session_db_cache_closed: return None db = self._session_dbs.get(key) if db is None: - db = SessionDB(db_path=home / "state.db") + db = acquire(home / "state.db") self._session_dbs[key] = db return db @@ -1659,7 +1659,8 @@ class APIServerAdapter(OpenAICompatRoutesMixin, BasePlatformAdapter): if db is shared_db: continue try: - db.close() + from hermes_state_registry import release_or_close + release_or_close(db) except Exception: logger.debug("Failed to close API-server SessionDB", exc_info=True) diff --git a/gateway/run.py b/gateway/run.py index 0a62103a77..b9d1589760 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -4467,13 +4467,14 @@ def _housekeeping_deferred_fts_retry() -> None: # Retry here, on the existing tick, against the shared instances this process already holds: # non-blocking admission, no new thread, rate-limited inside SessionDB. No-op when nothing is stale (one # attribute read per instance). See #100108. - from hermes_state_registry import live_shared_session_dbs - for _sdb in live_shared_session_dbs(): - _retry = getattr(_sdb, "retry_deferred_fts_recovery", None) - if callable(_retry) and _retry(): - logger.info( - "Deferred state.db FTS rebuild completed in-process for %s; full-text search restored.", - getattr(_sdb, "db_path", "state.db")) + from hermes_state_registry import borrow_live_shared_session_dbs + with borrow_live_shared_session_dbs() as _session_dbs: + for _sdb in _session_dbs: + _retry = getattr(_sdb, "retry_deferred_fts_recovery", None) + if callable(_retry) and _retry(): + logger.info( + "Deferred state.db FTS rebuild completed in-process for %s; full-text search restored.", + getattr(_sdb, "db_path", "state.db")) def _housekeeping_memory_trim() -> None: diff --git a/hermes_cli/web_server_lifecycle.py b/hermes_cli/web_server_lifecycle.py index aa78e7c52a..14b554628c 100644 --- a/hermes_cli/web_server_lifecycle.py +++ b/hermes_cli/web_server_lifecycle.py @@ -175,9 +175,11 @@ def _eager_reconcile_own_session_db() -> None: per-poll read-probe heal in :func:`_open_session_db_at_path`. """ try: - from hermes_state import SessionDB, _default_db_path + from hermes_state import _default_db_path + from hermes_state_registry import acquire, release_or_close - SessionDB(db_path=Path(_default_db_path()), read_only=False).close() + db = acquire(Path(_default_db_path())) + release_or_close(db) except Exception as exc: _log.warning( "startup schema reconcile of state.db failed (%s); session " diff --git a/hermes_cli/web_server_sessions.py b/hermes_cli/web_server_sessions.py index 24aff58a1c..beaced31de 100644 --- a/hermes_cli/web_server_sessions.py +++ b/hermes_cli/web_server_sessions.py @@ -110,12 +110,13 @@ def _open_session_db_at_path(db_path: Path, *, read_only: bool): import sqlite3 from hermes_state import SessionDB, is_malformed_schema_error + from hermes_state_registry import acquire, release_or_close # Read-only file/sidecar preflight (port of kilocode#12508): repair-or-refuse BEFORE the first # connection so users get an actionable message instead of an opaque "attempt to write a readonly # database" from deep inside _init_schema. if not read_only: - return SessionDB(db_path=db_path, read_only=False) + return acquire(db_path) def _needs_bootstrap() -> bool: try: @@ -128,7 +129,8 @@ def _open_session_db_at_path(db_path: Path, *, read_only: bool): if _needs_bootstrap(): with _session_db_bootstrap_lock: if _needs_bootstrap(): - SessionDB(db_path=db_path, read_only=False).close() + db = acquire(db_path) + release_or_close(db) def _open_probed(): db = SessionDB(db_path=db_path, read_only=True) @@ -156,7 +158,8 @@ def _open_session_db_at_path(db_path: Path, *, read_only: bool): or is_malformed_schema_error(exc) or isinstance(exc, UnicodeDecodeError)): raise - SessionDB(db_path=db_path, read_only=False).close() + db = acquire(db_path) + release_or_close(db) try: return _open_probed() except (sqlite3.DatabaseError, UnicodeDecodeError) as still_stale: @@ -213,16 +216,20 @@ 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: - db.close() + release_or_close(db) 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 2a5614bc16..ed4389c693 100644 --- a/hermes_state_registry.py +++ b/hermes_state_registry.py @@ -8,7 +8,9 @@ when the file is replaced (snapshot restore, recovery swap). Lifecycle rules: - ``acquire(path)`` returns the current generation for *path* and bumps its refcount. -- ``close()`` on a shared instance is a NO-OP: the registry owns the connection lifecycle. +- ``close()`` on a shared instance RELEASES one refcount instead of tearing the + connection down: the registry owns the physical lifecycle and only closes on the + final release, so legacy call sites return their reference instead of leaking it. - ``release(db)`` decrements the generation *db was acquired from* (object-keyed, so an inode replacement cannot strand a still-owned generation); the final release of a retired generation tears it down. @@ -16,6 +18,14 @@ Lifecycle rules: its holders release. If the replacement open fails the registry keeps NO path entry. - All teardown happens OUTSIDE the registry lock: a final release's WAL checkpoint must never stall acquisition for every state.db. +- A final close/checkpoint is serialized with the next open for the same path; no new + generation is published while the previous generation is still tearing down. +- A path can have SEVERAL closes admitted at once (the current generation's final release + plus a retired generation's drain). The path barrier COUNTS them and is lifted only by + the last one to settle, so neither ``acquire`` nor ``close_all`` can escape while any + handle for that path is still inside checkpoint/WAL-unlink. +- Maintenance callers borrow handles with a temporary registry reference instead of + iterating an unpinned snapshot. """ from __future__ import annotations @@ -24,7 +34,7 @@ import contextlib import logging import threading from pathlib import Path -from typing import TYPE_CHECKING, Dict, List, Optional, Tuple +from typing import TYPE_CHECKING, Dict, Iterator, List, Optional, Tuple from hermes_state_common import stat_db_file_identity as _stat_db_file_identity @@ -34,12 +44,32 @@ if TYPE_CHECKING: # pragma: no cover - import cycle guard, typed only logger = logging.getLogger(__name__) +class _TeardownBarrier: + """Accounting for every admitted-but-unfinished physical close of one path. + + A path can own more than one close at a time: the current generation's final + release and a retired generation's drain are admitted independently under + ``_lock`` and only meet at the lifecycle mutex. One event per path is honest + only if the LAST admitted teardown settles it. Signalling on the first lets + ``close_all()`` return and ``acquire()`` publish a replacement while an older + handle is still inside ``PRAGMA wal_checkpoint``/sidecar unlink -- the exact + overlap (#102827) this registry exists to forbid. + """ + + __slots__ = ("event", "pending") + + def __init__(self) -> None: + self.event = threading.Event() + self.pending = 0 + + class _Generation: """One shared SessionDB generation: instance, refcount, file identity.""" - __slots__ = ("db", "refcount", "identity", "retired") + __slots__ = ("path", "db", "refcount", "identity", "retired") - def __init__(self, db: "SessionDB", identity: Optional[Tuple[int, int]]) -> None: + def __init__(self, path: Path, db: "SessionDB", identity: Optional[Tuple[int, int]]) -> None: + self.path = path self.db = db self.refcount = 1 self.identity = identity @@ -55,6 +85,16 @@ _retired: Dict[int, _Generation] = {} # (schema reconciliation can take seconds), but peers for the SAME file must wait or # every cold caller opens its own writer before a winner is chosen. _opening: Dict[Path, threading.Event] = {} +# A final close/checkpoint must finish before a replacement writer is opened +# for the same path. The barrier is admitted while holding _lock and lifted +# only after the LAST admitted physical teardown, so acquire cannot slip +# through the generation-removal/open gap and close_all cannot report a +# finished sweep over a close that is still running. +_tearing_down: Dict[Path, _TeardownBarrier] = {} +# Open and close are both performed outside _lock. This per-path mutex closes +# the race between checking _tearing_down and entering sqlite3.connect(), +# including retired-generation drains after an inode replacement. +_path_lifecycle_locks: Dict[Path, threading.Lock] = {} def _open_session_db(path: Path) -> "SessionDB": @@ -74,6 +114,57 @@ def _teardown(db: "SessionDB") -> None: logger.debug("Error closing shared SessionDB", exc_info=True) +def _path_lifecycle_lock_locked(path: Path) -> threading.Lock: + """Return the lifecycle mutex for *path* (caller holds ``_lock``).""" + lock = _path_lifecycle_locks.get(path) + if lock is None: + lock = threading.Lock() + _path_lifecycle_locks[path] = lock + return lock + + +def _admit_teardown_locked(path: Path) -> _TeardownBarrier: + """Register one pending physical close for *path* (caller holds ``_lock``). + + Admission shares the lock section that removes the generation, so a peer + release, ``acquire`` or ``close_all`` taking the lock next always sees this + teardown accounted for. + """ + barrier = _tearing_down.get(path) + if barrier is None: + barrier = _tearing_down[path] = _TeardownBarrier() + barrier.pending += 1 + return barrier + + +def _finish_teardown(path: Path, barrier: _TeardownBarrier) -> None: + """Settle one admitted teardown; only the last one lifts the path barrier.""" + with _lock: + barrier.pending -= 1 + if barrier.pending > 0: + return + if _tearing_down.get(path) is barrier: + _tearing_down.pop(path, None) + barrier.event.set() + + +def _teardown_generation( + path: Path, + db: "SessionDB", + *, + barrier: Optional[_TeardownBarrier] = None, +) -> None: + """Close *db* under its path lifecycle mutex, then settle its barrier slot.""" + with _lock: + lifecycle_lock = _path_lifecycle_lock_locked(path) + try: + with lifecycle_lock: + _teardown(db) + finally: + if barrier is not None: + _finish_teardown(path, barrier) + + def _db_path_of(db: "SessionDB") -> Optional[Path]: """``Path(db.db_path)`` or None when absent/unconvertible.""" path = getattr(db, "db_path", None) @@ -105,6 +196,7 @@ def acquire(db_path: Optional[Path] = None) -> "SessionDB": path = raw_path while True: + wait_for: Optional[threading.Event] = None with _lock: generation = _generations.get(path) if generation is not None: @@ -119,36 +211,62 @@ def acquire(db_path: Optional[Path] = None) -> "SessionDB": else: generation.refcount += 1 return generation.db - opening = _opening.get(path) - if opening is None: - opening = _opening[path] = threading.Event() - break + teardown = _tearing_down.get(path) + if teardown is not None: + wait_for = teardown.event + else: + opening = _opening.get(path) + 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. - opening.wait() + wait_for.wait() # Open OUTSIDE the lock; the per-path marker prevents redundant writers without # serialising other files. try: - db = _open_session_db(path) - db._shared_registry_owned = True - identity = _stat_db_file_identity(path) + # 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: _finish_opening(path, opening) raise + discard_barrier: Optional[_TeardownBarrier] = None with _lock: - existing = _generations.get(path) - if existing is not None: # Defensive: installed by explicit registry manipulation mid-open. - existing.refcount += 1 - winner = existing.db + 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: - _generations[path] = _Generation(db, identity) - winner = db + 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: - _teardown(db) + with lifecycle_lock: + _teardown(db) return winner @@ -160,6 +278,8 @@ 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) if generation is None: @@ -172,28 +292,61 @@ def release(db: "SessionDB") -> bool: return False generation.refcount -= 1 needs_teardown = generation.refcount <= 0 - if needs_teardown and generation.retired: - _retired.pop(key, None) - elif needs_teardown and (path := _db_path_of(db)) is not None: - _generations.pop(path, None) + 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. + 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: - _teardown(db) + assert teardown_path is not None + _teardown_generation(teardown_path, db, barrier=teardown_barrier) return True def close_all() -> int: """Close every shared SessionDB regardless of refcount; returns the count. For gateway shutdown, after all agents and cron jobs finished. Idempotent.""" + teardown_barriers: Dict[Path, _TeardownBarrier] = {} with _lock: + active_teardowns = list(_tearing_down.values()) generations = list(_generations.values()) + list(_retired.values()) + for path in {generation.path for generation in generations}: + teardown_barriers[path] = _admit_teardown_locked(path) _generations.clear() _retired.clear() for generation in generations: generation.retired = True + # Teardown outside the lock, one path at a time. Holding the lifecycle + # mutex across all generations for a path prevents an old retired handle + # and the current handle from checkpointing the same sidecars concurrently. + by_path: Dict[Path, List[_Generation]] = {} for generation in generations: - _teardown(generation.db) + by_path.setdefault(generation.path, []).append(generation) + for path, path_generations in by_path.items(): + with _lock: + lifecycle_lock = _path_lifecycle_lock_locked(path) + try: + with lifecycle_lock: + for generation in path_generations: + _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. + for barrier in active_teardowns: + barrier.event.wait() return len(generations) @@ -205,6 +358,30 @@ def live_shared_session_dbs() -> List["SessionDB"]: return [g.db for g in _generations.values() if not g.retired] +@contextlib.contextmanager +def borrow_live_shared_session_dbs() -> Iterator[List["SessionDB"]]: + """Borrow live handles with registry references pinned for the whole block. + + Maintenance must not operate on the unowned snapshot returned by + :func:`live_shared_session_dbs`: the last real owner could otherwise + release and physically close the connection between the snapshot and the + maintenance call. Each borrowed generation gets one temporary reference; + the ``finally`` block releases it even when maintenance raises. + """ + with _lock: + borrowed_generations = [ + generation for generation in _generations.values() if not generation.retired + ] + borrowed = [generation.db for generation in borrowed_generations] + for generation in borrowed_generations: + generation.refcount += 1 + try: + yield borrowed + finally: + for db in reversed(borrowed): + release(db) + + def stats() -> Dict[str, int]: """Registry census for tests and diagnostics (no locks held long).""" with _lock: diff --git a/tests/cron/test_cleanup_timeout.py b/tests/cron/test_cleanup_timeout.py index 608abe820e..6981c24a45 100644 --- a/tests/cron/test_cleanup_timeout.py +++ b/tests/cron/test_cleanup_timeout.py @@ -8,9 +8,14 @@ from __future__ import annotations import threading import time +from concurrent.futures import Future from unittest.mock import MagicMock, patch -from cron.scheduler import run_job, _teardown_cron_agent +from cron.scheduler import ( + _defer_cron_worker_teardown_if_running, + _teardown_cron_agent, + run_job, +) _RUNTIME = { @@ -94,6 +99,35 @@ def test_agent_teardown_is_bounded(): release.set() +def test_detached_worker_teardown_waits_for_future(): + """A timed-out worker keeps its agent and SessionDB until completion.""" + future = Future() + worker_state = {"future": future} + fake_db = MagicMock() + agent = MagicMock() + + with patch("cron.scheduler._teardown_detached_cron_worker") as teardown_worker: + assert _defer_cron_worker_teardown_if_running( + worker_state, + fake_db, + agent, + "detached-worker", + "detached worker", + "cron_detached-worker", + ) is True + teardown_worker.assert_not_called() + + future.set_result({"final_response": "late"}) + + teardown_worker.assert_called_once_with( + fake_db, + agent, + "detached-worker", + "detached worker", + "cron_detached-worker", + ) + + def test_dispatch_guard_releases_after_sessiondb_finalization_hang(tmp_path): """A second scheduler tick can fire the same job after cleanup times out.""" import cron.scheduler as sched diff --git a/tests/hermes_cli/test_web_server_auto_archive_registry.py b/tests/hermes_cli/test_web_server_auto_archive_registry.py new file mode 100644 index 0000000000..29f891732f --- /dev/null +++ b/tests/hermes_cli/test_web_server_auto_archive_registry.py @@ -0,0 +1,100 @@ +"""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 58d4837625..ce0fcde908 100644 --- a/tests/hermes_state/test_shared_session_db_registry.py +++ b/tests/hermes_state/test_shared_session_db_registry.py @@ -17,9 +17,6 @@ Covers the three ownership invariants the PR review demanded: every state.db in the process). """ -import contextlib -import os -import shutil import threading import time from pathlib import Path @@ -36,27 +33,32 @@ def _clean_registry(): registry._generations.clear() registry._retired.clear() registry._opening.clear() + registry._tearing_down.clear() yield registry.close_all() registry._generations.clear() registry._retired.clear() registry._opening.clear() + registry._tearing_down.clear() -def _replace_file_preserving_schema(src: Path, dst: Path) -> None: - """Simulate snapshot-restore / recovery: new inode, same logical DB. +def _replace_file_preserving_schema(monkeypatch, db_path: Path, old_db=None) -> None: + """Simulate a replacement identity without unlinking an open SQLite file. - Copies the live DB to a temp name, removes the original, and renames - the copy into place — the replacement has a different inode. + 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. """ - tmp = dst.with_suffix(".replacement.tmp") - shutil.copy2(dst, tmp) - os.unlink(dst) - os.rename(tmp, dst) + 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) class TestInodeReplacement: - def test_live_holders_keep_working_handle_across_replacement(self, tmp_path): + def test_live_holders_keep_working_handle_across_replacement(self, tmp_path, monkeypatch): """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.""" @@ -66,7 +68,7 @@ class TestInodeReplacement: b = registry.acquire(db_path) assert a is b - _replace_file_preserving_schema(db_path, db_path) + _replace_file_preserving_schema(monkeypatch, db_path, old_db=a) c = registry.acquire(db_path) assert c is not a, "new caller must get the fresh generation" @@ -114,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): + def test_retired_generation_never_relent_even_after_drain(self, tmp_path, monkeypatch): """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(db_path, db_path) + _replace_file_preserving_schema(monkeypatch, db_path) second = registry.acquire(db_path) third = registry.acquire(db_path) @@ -137,7 +139,7 @@ class TestInodeReplacement: db_path = tmp_path / "state.db" old = registry.acquire(db_path) - _replace_file_preserving_schema(db_path, db_path) + _replace_file_preserving_schema(monkeypatch, db_path) calls = {"n": 0} @@ -388,6 +390,131 @@ class TestTeardownOutsideLock: assert stats["retired_generations"] == 0 +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" + db = registry.acquire(db_path) + + with registry.borrow_live_shared_session_dbs() as borrowed: + assert borrowed == [db] + assert registry.release(db) is True + assert db._conn is not None + + assert db._conn is None + + class TestLegacyCloseSemantics: def test_close_on_shared_instance_releases_one_refcount(self, tmp_path): """Legacy ``db.close()`` call sites must not leak refcounts: close() @@ -470,3 +597,226 @@ class TestAcquireSingleFlight: assert len(opened) >= 1 registry.release(results[0]) registry.release(results[1]) + + +class TestMultiGenerationTeardownBarrier: + """One path, several closes admitted at once (#103118 review). + + The per-path mutex only serializes teardowns that already entered it. A + releasing thread can be descheduled after its generation left the registry + and its close was admitted, but before the mutex. If the path barrier is a + bare event, the NEXT teardown to settle lifts it for everybody: ``close_all`` + reports a finished sweep and ``acquire`` publishes a replacement writer while + the first handle is still inside checkpoint/WAL-unlink — the overlap that + leaves zero-hole pages behind (#102827). + """ + + @staticmethod + def _pause_teardown_of(monkeypatch, target): + """Hold ``target``'s physical teardown *before* the lifecycle mutex. + + Returns ``(entered, resume)``: ``entered`` fires once the paused + teardown is admitted-but-unfinished, ``resume`` lets it proceed. + """ + entered = threading.Event() + resume = threading.Event() + original = registry._teardown_generation + + def _paused(path, db, *, barrier=None): + if db is target: + entered.set() + if not resume.wait(10.0): # pragma: no cover - failure path + raise AssertionError("timed out holding a teardown") + original(path, db, barrier=barrier) + + monkeypatch.setattr(registry, "_teardown_generation", _paused) + return entered, resume + + @staticmethod + def _release_async(db, errors): + def _run(): + try: + assert registry.release(db) is True + except BaseException as exc: # pragma: no cover - failure path + errors.append(exc) + + thread = threading.Thread(target=_run, daemon=True) + thread.start() + return thread + + def _two_generations(self, tmp_path, monkeypatch): + """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) + current = registry.acquire(db_path) + assert current is not old + return db_path, old, current + + def test_retired_drain_does_not_lift_a_pending_current_teardown(self, tmp_path, monkeypatch): + """close_all() must not report a finished sweep over a pending close.""" + db_path, old, current = self._two_generations(tmp_path, monkeypatch) + resolved = Path(db_path).resolve() + entered, resume = self._pause_teardown_of(monkeypatch, current) + errors = [] + releaser = self._release_async(current, errors) + try: + assert entered.wait(5.0) + + # The retired generation's final release settles completely while the + # current generation's close is still admitted. + assert registry.release(old) is True + assert old._conn is None + barrier = registry._tearing_down.get(resolved) + assert barrier is not None, "retired drain lifted the shared path barrier" + assert barrier.pending == 1 + assert not barrier.event.is_set() + + swept = [] + sweep_done = threading.Event() + + def _close_all(): + swept.append(registry.close_all()) + sweep_done.set() + + sweeper = threading.Thread(target=_close_all, daemon=True) + sweeper.start() + assert not sweep_done.wait(0.5), "close_all returned with a close still pending" + assert current._conn is not None + + resume.set() + assert sweep_done.wait(10.0) + sweeper.join(10.0) + finally: + resume.set() + releaser.join(10.0) + + assert errors == [] + assert current._conn is None + assert registry._tearing_down.get(resolved) is None + + def test_replacement_is_not_published_before_the_last_close_settles(self, tmp_path, monkeypatch): + """acquire() must not open a writer on top of an unfinished close.""" + db_path, old, current = self._two_generations(tmp_path, monkeypatch) + entered, resume = self._pause_teardown_of(monkeypatch, current) + errors = [] + releaser = self._release_async(current, errors) + acquired = [] + acquire_done = threading.Event() + try: + assert entered.wait(5.0) + assert registry.release(old) is True + + def _acquire(): + try: + fresh = registry.acquire(db_path) + # Record the predecessor's state AT publication time. + acquired.append((fresh, current._conn is None)) + 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 before the previous close" + + resume.set() + assert acquire_done.wait(10.0) + opener.join(10.0) + finally: + resume.set() + releaser.join(10.0) + + assert errors == [] + assert len(acquired) == 1 + fresh, predecessor_was_closed = acquired[0] + assert predecessor_was_closed, "a new writer was published over a live handle" + 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" + db = registry.acquire(db_path) + resolved = Path(db_path).resolve() + real_teardown = registry._teardown + + def _raising_teardown(target): + real_teardown(target) + raise RuntimeError("checkpoint exploded") + + monkeypatch.setattr(registry, "_teardown", _raising_teardown) + with pytest.raises(RuntimeError): + registry.release(db) + + assert registry._tearing_down.get(resolved) is None + monkeypatch.setattr(registry, "_teardown", real_teardown) + 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 3cc1ca1272..c2ba1e596c 100644 --- a/tests/hermes_state/test_state_db_file_identity.py +++ b/tests/hermes_state/test_state_db_file_identity.py @@ -33,18 +33,16 @@ def _require_identity(db: SessionDB) -> None: pytest.skip("filesystem does not expose st_dev/st_ino for identity checks") -def test_replace_with_new_inode_fails_loudly_without_fts_repair(tmp_path): +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): live = tmp_path / "state.db" - other = tmp_path / "other.db" db = _make_db(live, "live-sess", "original") _require_identity(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) + _mark_replaced_handle(monkeypatch, db) with pytest.raises(StateDbReplacedError, match="replaced underneath"): db.append_message("live-sess", role="user", content="after-replace") @@ -56,14 +54,11 @@ def test_replace_with_new_inode_fails_loudly_without_fts_repair(tmp_path): db.close() -def test_second_write_after_halt_does_not_attempt_repair(tmp_path): +def test_second_write_after_halt_does_not_attempt_repair(tmp_path, monkeypatch): live = tmp_path / "state.db" - other = tmp_path / "other.db" db = _make_db(live, "s", "a") _require_identity(db) - alt = _make_db(other, "t", "b") - alt.close() - os.replace(other, live) + _mark_replaced_handle(monkeypatch, db) with pytest.raises(StateDbReplacedError): db.append_message("s", role="user", content="first") with pytest.raises(StateDbReplacedError): @@ -125,17 +120,14 @@ 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): +def test_fts_scoped_error_on_replaced_file_skips_fts_fail_open(tmp_path, monkeypatch): """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) - alt = _make_db(other, "t", "b") - alt.close() - os.replace(other, live) + _mark_replaced_handle(monkeypatch, db) with pytest.raises(StateDbReplacedError): db._enter_fts_fail_open( @@ -187,11 +179,6 @@ 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). #