diff --git a/hermes_state.py b/hermes_state.py index 4f7b6453461a..65c8430bac5e 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -20,6 +20,7 @@ import json import logging import os +import queue import random import re import sqlite3 @@ -235,6 +236,29 @@ def _delete_delegate_children(conn, parent_ids: List[str]) -> List[str]: DEFAULT_DB_PATH = get_hermes_home() / "state.db" +# How long SessionDB stops attempting read-only opens after one fails, before +# probing again. Long enough that a genuinely unreadable file isn't retried per +# query; short enough that transient fd pressure doesn't strand the read pool. +_READ_OPEN_RETRY_SECONDS = 60.0 + +# Hard ceiling on read-only connections ALIVE at once per SessionDB — pooled +# idle ones and checked-out ones together. +# +# Deliberately one constant for both the pool's maxsize and the permit count, +# because bounding only the pool bounds the wrong thing. A LifoQueue caps how +# many connections are *returned*; it says nothing about how many are *open*. +# With an open-on-miss checkout, N readers arriving on an empty pool all miss, +# all open, and peak at N — the surplus is closed on release, so nothing +# accumulates forever, but EMFILE is a peak-instant condition and the burst +# that empties the pool is exactly the burst that exhausts the fd table. +# +# So a connection holds a permit for its whole lifetime: acquired in +# _get_read_conn() before the open, released in _close_read_conn() after the +# close. Once permits are gone the read path degrades to the locked writer +# connection instead of opening more descriptors — slower under load, which is +# the correct trade against a process-wide wedge the supervisor cannot see. +_READ_POOL_MAX = 8 + # Import-time snapshot used by _default_db_path() to detect a deliberately # re-pointed DEFAULT_DB_PATH (tests monkeypatch the constant directly). _IMPORT_DEFAULT_DB_PATH = DEFAULT_DB_PATH @@ -1849,20 +1873,57 @@ def __init__(self, db_path: Path = None, read_only: bool = False): self.read_only = read_only self._lock = threading.Lock() - # Read-path split (WAL only): recall/browse queries run on per-thread - # read-only connections so they never queue behind writer flushes on - # self._lock. See _read_ctx(). - self._read_local = threading.local() - # Strong set of all live read connections across all threads. We - # hold a reference so short-lived reader threads' connections are - # not GC'd without close() — that would leak tracked fds in - # _live_connections. close() drains this set. - self._read_conns: "set[sqlite3.Connection]" = set() + # Read-path split (WAL only): recall/browse queries borrow a + # read-only connection from a bounded pool so they never queue + # behind writer flushes on self._lock. See _read_ctx(). + # + # The pool is BOUNDED because the previous per-thread + # (threading.local + strong set) scheme pinned one connection per + # (SessionDB x thread) for the life of the process. Starlette + # dispatches sync routes on anyio worker threads, so a SessionDB + # that is never closed accumulated a connection — and two fds, the + # database and its -wal — for every worker thread that ever read, + # until the process hit the 256 soft RLIMIT_NOFILE a service manager + # hands it and every request failed with EMFILE while the process + # stayed alive, so the supervisor's restart-on-exit never fired. + # Same bug class as the closing(...) fix in gateway/readiness.py + # (#69678 / #69567). + self._read_pool: "queue.LifoQueue[sqlite3.Connection]" = queue.LifoQueue( + maxsize=_READ_POOL_MAX + ) + # One permit per live read connection, held from before the open in + # _get_read_conn() until after the close in _close_read_conn(). This + # is what bounds PEAK descriptors; _read_pool alone bounds only the + # idle set. See _READ_POOL_MAX. Acquired non-blocking on purpose: a + # reader that cannot get a permit must degrade to the writer lock, not + # queue here — blocking would convert fd exhaustion into a stall, which + # is the same outage with a different stack trace. + self._read_permits = threading.BoundedSemaphore(_READ_POOL_MAX) + # Count of reads that found no permit and fell back to the locked + # writer connection. Not load-bearing; it is the only externally + # visible signal that the ceiling is actually being reached, so a + # too-small _READ_POOL_MAX is diagnosable from a running process + # instead of inferred from latency. + self._read_permit_exhausted = 0 self._read_conns_lock = threading.Lock() - # Set when close() begins. _get_read_conn checks this under the - # lock so a reader that finishes opening after the drain finds the - # shutdown in progress and closes its own connection immediately. + # Set when close() begins. _read_ctx checks this under the lock + # before returning a connection to the pool, so a reader still in + # flight during the drain closes its own connection instead of + # re-populating a pool nobody will drain again. self._read_conns_closed = False + # "read-only opens are failing against this file" backoff stamp. + # Instance-wide rather than per-thread: with a shared pool the open + # is no longer a per-thread event, and retrying a known-bad open on + # every query is a syscall storm for no benefit. The locked writer + # connection still serves reads while the backoff holds. + # Deliberately a TIMESTAMP, not a sticky bool: the likeliest trigger + # is transient fd pressure (EMFILE) — the very condition this pool + # exists to prevent — and a permanent flag would demote every reader + # on this instance to the writer lock for the life of the process. + # The gateway shares one SessionDB across every agent, so that turns + # a momentary blip into a permanent global convoy. Expires after + # _READ_OPEN_RETRY_SECONDS so the read path self-heals. + self._read_open_failed_at = 0.0 self._wal_active = False self._write_count = 0 # One-shot guard for the runtime FTS rebuild recovery on the write @@ -2063,7 +2124,10 @@ def _connect_and_init_with_lock_patience(): # ── Read-path split ── def _get_read_conn(self) -> Optional[sqlite3.Connection]: - """Per-thread read-only connection, or None when unavailable. + """Open a fresh read-only connection, or None when unavailable. + + Callers must return the connection to self._read_pool (see + _read_ctx); this opens, it does not track. Only used under WAL: WAL readers see a consistent snapshot and never block on (or get blocked by) the writer, so recall/browse queries can @@ -2077,16 +2141,45 @@ def _get_read_conn(self) -> Optional[sqlite3.Connection]: """ if not self._wal_active or self.read_only: return None - conn = getattr(self._read_local, "conn", None) - if conn is not None: - return conn - if getattr(self._read_local, "failed", False): + with self._read_conns_lock: + if self._read_conns_closed: + return None + if ( + self._read_open_failed_at + and time.monotonic() - self._read_open_failed_at + < _READ_OPEN_RETRY_SECONDS + ): + return None + # Take the descriptor permit BEFORE the open, so concurrent openers + # race for permits rather than for file descriptors. Non-blocking: + # losing the race means "use the writer connection", not "wait". + if not self._read_permits.acquire(blocking=False): + with self._read_conns_lock: + self._read_permit_exhausted += 1 + logger.debug( + "read pool at capacity (%d) for %s; serving this read from the " + "locked writer connection", + _READ_POOL_MAX, + self.db_path, + ) return None + # Bound before the try: the except handlers close it if the open + # half-succeeded, and an unbound name there would raise NameError over + # the top of the real failure. + conn = None try: conn = _connect_tracked_db( f"file:{self.db_path}?mode=ro", tracking_path=self.db_path, uri=True, + # Pooled connections are borrowed by whichever thread runs + # the next read, and sqlite3 otherwise refuses cross-thread + # use ("SQLite objects created in a thread can only be used + # in that same thread") — including on close(), which is how + # the old per-thread connections became unclosable and leaked + # their fds. Exclusive ownership is enforced by the pool + # checkout/return, not by sqlite3. Matches the writer opens. + check_same_thread=False, timeout=5.0, isolation_level=None, ) @@ -2097,36 +2190,134 @@ def _get_read_conn(self) -> Optional[sqlite3.Connection]: # registry, not the database file, so mode=ro is fine. if self._fts_cjk_loaded: load_fts5_cjk_extension(conn) - with self._read_conns_lock: - if self._read_conns_closed: - # close() already drained — don't register; close - # immediately so no tracked fd leaks. - conn.close() - self._read_local.failed = True - return None - self._read_conns.add(conn) except sqlite3.Error: - # Mark this thread failed so we don't retry the open on every - # query; the locked writer connection still serves reads. - self._read_local.failed = True + # A partially-constructed connection — _connect_tracked_db + # succeeded, the CJK extension load did not — must be closed here. + # Dropping it on the floor still open leaves a live descriptor the + # tracking registry still counts: the same leak shape this pool + # exists to fix, one level further down. + self._discard_partial_read_conn(conn) + # Back off from retrying the open on every query; the locked + # writer connection still serves reads until the stamp expires. + with self._read_conns_lock: + self._read_open_failed_at = time.monotonic() logger.debug("read-only connection open failed for %s", self.db_path, exc_info=True) + self._read_permits.release() return None - self._read_local.conn = conn + except BaseException: + # Anything else (a non-sqlite3 extension-load failure, MemoryError, + # KeyboardInterrupt landing between open and return) must not + # strand the permit: a stranded permit is not a transient error, it + # permanently shrinks the read path by one slot for the life of the + # process. + self._discard_partial_read_conn(conn) + self._read_permits.release() + raise return conn + def _discard_partial_read_conn(self, conn) -> None: + """Close a connection that failed between open and hand-off. + + Separate from _close_read_conn because that one releases a permit and + this runs on paths that release their own. + """ + if conn is None: + return + try: + conn.close() + except Exception as exc: + logger.warning( + "partially-opened read conn close failed for %s: %s", self.db_path, exc + ) + + def _close_read_conn(self, conn) -> None: + """Close a pooled read connection and release its descriptor permit. + + This was a bare ``except Exception: pass``, which silently swallowed + the sqlite3.ProgrammingError raised when close() ran on a thread + other than the one that opened the connection — the exact signature + of the fd leak this pool fixes. A close that fails leaks a tracked + fd, so it must not be invisible. + + The permit is released even when close() raises: the descriptor is + already lost at that point, and withholding the permit too would turn + one leaked fd into a permanently narrower read path — failing twice for + one fault. The warning is the signal that matters. + + Pairs with _get_read_conn(). Calling this on a connection that did not + come from there over-releases the BoundedSemaphore, which raises + ValueError rather than silently widening the ceiling. + """ + try: + conn.close() + except Exception as exc: + logger.warning("read-conn close failed for %s: %s", self.db_path, exc) + finally: + self._read_permits.release() + + def _checkout_read_conn(self) -> Optional[sqlite3.Connection]: + """Borrow a read connection from the pool, opening one on a miss. + + The single acquisition seam for the read path: the WAL/read_only gate, + the pool checkout and the open-on-miss all live here, so there is + exactly one place to exercise (and one place for a caller to bypass by + accident). Returns None when the read path is unavailable and the + caller must fall back to the locked writer connection. + + A pool hit costs no permit — the connection it hands back is already + holding one. Only the miss path can open, and only _get_read_conn() can + take a permit, so peak live connections is bounded by _READ_POOL_MAX no + matter how many threads miss simultaneously. + """ + if not self._wal_active or self.read_only: + return None + try: + return self._read_pool.get_nowait() + except queue.Empty: + return self._get_read_conn() + @contextmanager def _read_ctx(self): """Yield a connection for read-only statements. - WAL: a per-thread read-only connection with NO lock — recall queries - never convoy behind writer flushes (the gateway shares one SessionDB - across every agent, so this lock was a global choke point). - Non-WAL or read-conn failure: the shared writer connection under - self._lock, byte-for-byte the legacy behavior. + WAL: a read-only connection borrowed from a bounded pool with NO + lock — recall queries never convoy behind writer flushes (the + gateway shares one SessionDB across every agent, so this lock was a + global choke point). The connection is checked out for the duration + of the block, so no two threads ever touch it concurrently. + Non-WAL, read-conn failure, or _READ_POOL_MAX already reached: the + shared writer connection under self._lock, byte-for-byte the legacy + behavior. + + That last case is the deliberate degradation. Past the ceiling readers + convoy on the writer lock instead of opening descriptors — measurably + slower under a burst, and the alternative is EMFILE, which takes the + whole process down in a way a restart-on-exit supervisor cannot see. """ - conn = self._get_read_conn() + conn = self._checkout_read_conn() if conn is not None: - yield conn + try: + yield conn + finally: + returned = False + with self._read_conns_lock: + if not self._read_conns_closed: + try: + self._read_pool.put_nowait(conn) + returned = True + except queue.Full: + pass + if not returned: + # close() has already drained the pool, so this connection + # is surplus. Close it here — dropping it on the floor is + # what leaked the fd. + # + # queue.Full is now unreachable in practice (permits and + # maxsize are both _READ_POOL_MAX, so there can never be a + # ninth connection to return), but the branch stays: it is + # load-bearing if those two ever drift apart, and a leak is + # the failure mode it prevents. + self._close_read_conn(conn) return with self._lock: yield self._conn @@ -2637,23 +2828,18 @@ def close(self): # (instance, function), so this removes exactly our registration; # no-op when the writer never started. atexit.unregister(self._drain_token_queue_at_exit) - # Close all read-only connections across all threads. Per-thread - # connections live in threading.local() and would otherwise be GC'd - # without calling close(), leaking tracked fds in _live_connections. - # The strong set holds references so short-lived reader threads' - # connections survive until close() drains them. Setting the closed - # flag under the lock prevents a reader from registering a new - # connection after the drain. + # Drain the read-only connection pool. Setting the closed flag + # under the lock first means a reader still in flight closes its own + # connection on release instead of re-populating a pool that has + # already been drained. with self._read_conns_lock: self._read_conns_closed = True - read_conns = list(self._read_conns) - self._read_conns.clear() - for conn in read_conns: + while True: try: - conn.close() - except Exception: - pass - self._read_local.conn = None + conn = self._read_pool.get_nowait() + except queue.Empty: + break + self._close_read_conn(conn) with self._lock: if self._conn: try: diff --git a/tests/test_session_db_read_conn_pool.py b/tests/test_session_db_read_conn_pool.py new file mode 100644 index 000000000000..3d2dabcd3d00 --- /dev/null +++ b/tests/test_session_db_read_conn_pool.py @@ -0,0 +1,386 @@ +"""The SessionDB read path must not leak one connection per (SessionDB x thread). + +``_get_read_conn`` used to cache a read-only connection in ``threading.local()`` +and pin it in a strong set (``_read_conns``) that was only ever drained by +``close()``. Starlette dispatches sync routes on anyio worker threads, so a +SessionDB that is never closed -- the dashboard's module-global ``_db`` and +the per-session ``session_db`` handles -- gained a connection, and a file +descriptor, for every worker thread that ever served a read. In production +that walked into the 256 soft ``RLIMIT_NOFILE`` a service manager hands the +process, after which every request failed with ``OSError`` EMFILE while the +process stayed alive, so the supervisor's restart-on-exit never fired. + +Worse, those connections were opened WITHOUT ``check_same_thread=False`` (both +writer opens pass it), so ``close()`` on them raised ``ProgrammingError`` from +a different thread and the bare ``except Exception: pass`` hid it -- leaving +``hermes_cli.sqlite_safe_read``'s registry permanently over-counted as well. + +The contract pinned here: reads borrow from a BOUNDED pool, connections are +returned and reused, surplus connections are closed rather than dropped, and +``close()`` actually closes them from whatever thread it runs on. + +Bounded means bounded at PEAK, not merely at rest. Pooling returns behind a +``maxsize`` LifoQueue while opening unconditionally on a miss still lets a +burst of N simultaneous readers on a cold pool open N descriptors before +closing the surplus -- which is the exact shape of the production incident, +since the burst that exhausts the pool is the burst that exhausts the fd +table. Peak is held down by a permit acquired before the open and released +after the close; past the ceiling readers degrade to the locked writer +connection. Tests that join their workers before counting cannot see any of +this, so the peak assertions use a barrier. + +These assert on the pool/registry counts, never on ``lsof``: SQLite's unix VFS +parks a closed descriptor on a per-inode reuse list while any connection still +holds POSIX locks on that inode, so raw descriptor counts lag the real +connection count and make such assertions flaky. +""" + +import threading + +import pytest + +from hermes_state import SessionDB + + +def _live_count(path) -> int: + """Live-connection count the tracking registry holds for *path*.""" + import hermes_cli.sqlite_safe_read as mod + + with mod._live_lock: + return mod._live_connections.get(mod._key(path), 0) + + +@pytest.fixture() +def db(tmp_path): + d = SessionDB(db_path=tmp_path / "state.db") + d.create_session(session_id="s1", source="cli", model="m") + d.append_message("s1", role="user", content="hello graphiti world") + d.append_message("s1", role="assistant", content="the neo4j daemon is healthy") + yield d + d.close() + + +def _read(db): + db.get_session("s1") + db.search_messages("graphiti", limit=5) + db.get_messages("s1") + + +@pytest.mark.requires_wal +def test_read_pool_is_bounded_across_many_threads(db): + """150 short-lived reader threads must not pin 150 connections. + + NOTE: this measures the pool AT REST -- every worker is joined before the + count is taken, so by construction it cannot observe how many connections + were open simultaneously. It is a real assertion about accumulation and a + non-assertion about peak. See + test_peak_live_connections_bounded_under_simultaneous_burst for the peak. + """ + maxsize = db._read_pool.maxsize + assert maxsize > 0, "read pool must be bounded" + + for _ in range(6): + threads = [threading.Thread(target=_read, args=(db,)) for _ in range(25)] + for t in threads: + t.start() + for t in threads: + t.join() + assert db._read_pool.qsize() <= maxsize + + # The pre-fix code held 151 connections here (150 readers + main thread). + assert db._read_pool.qsize() <= maxsize + # +1 for the writer connection SessionDB always holds. + assert _live_count(db.db_path) <= maxsize + 1 + + +@pytest.mark.requires_wal +def test_read_conn_returned_to_pool_and_reused(db): + """Sequential reads on one thread reuse a pooled connection, not a new one.""" + with db._read_ctx() as conn: + first = conn + assert db._read_pool.qsize() >= 1, "connection was not returned to the pool" + with db._read_ctx() as conn: + assert conn is first, "pooled connection was not reused" + + +@pytest.mark.requires_wal +def test_pooled_conn_is_usable_from_another_thread(db): + """A pooled connection is handed between threads, so it must not be + bound to its creating thread (check_same_thread=False).""" + with db._read_ctx() as conn: + borrowed = conn + + errors = [] + + def use_it(): + try: + borrowed.execute("SELECT 1").fetchone() + except Exception as exc: # noqa: BLE001 + errors.append(exc) + + t = threading.Thread(target=use_it) + t.start() + t.join() + assert not errors, f"pooled connection unusable off-thread: {errors}" + + +@pytest.mark.requires_wal +def test_close_drains_pool_from_a_foreign_thread(tmp_path): + """close() must actually close pooled connections, including ones opened + on threads that have since exited -- the swallowed ProgrammingError.""" + d = SessionDB(db_path=tmp_path / "state2.db") + d.create_session(session_id="s1", source="cli", model="m") + + # Populate the pool from a worker thread, then let that thread die. + t = threading.Thread(target=lambda: d.get_session("s1")) + t.start() + t.join() + assert d._read_pool.qsize() >= 1 + + d.close() + assert d._read_pool.qsize() == 0 + # Registry back to zero proves the closes succeeded rather than raising + # ProgrammingError into a bare except. + assert _live_count(d.db_path) == 0 + + +@pytest.mark.requires_wal +def test_reader_after_close_does_not_repopulate_pool(db): + """A read racing close() must close its connection, not refill the pool.""" + db.close() + assert db._read_pool.qsize() == 0 + # A read arriving after the drain must not open-and-requeue a connection + # that nothing will ever close again. + with db._read_ctx(): + pass + assert db._read_pool.qsize() == 0 + + +def test_reads_are_still_correct_under_concurrency(db): + """Pooling must not corrupt results when threads share connections.""" + results = [] + errors = [] + + def reader(): + try: + results.append(db.get_session("s1")["id"]) + results.append(len(db.get_messages("s1"))) + except Exception as exc: # noqa: BLE001 + errors.append(exc) + + threads = [threading.Thread(target=reader) for _ in range(12)] + for t in threads: + t.start() + for t in threads: + t.join() + assert not errors, f"concurrent reads failed: {errors}" + assert results.count("s1") == 12 + assert results.count(2) == 12 + + +@pytest.mark.requires_wal +def test_read_open_failure_backs_off_but_recovers(db): + """A failed read-only open must not permanently demote the read path. + + The first version of this fix used a sticky instance-wide boolean + (``_read_open_failed``). Its likeliest trigger is transient fd pressure -- + EMFILE, the very condition this pool exists to prevent -- and because the + gateway shares ONE SessionDB across every agent, a single blip would have + convoyed every subsequent reader behind the writer lock for the life of + the process. The stamp must expire. + """ + import time as _time + + from hermes_state import _READ_OPEN_RETRY_SECONDS + + baseline = db._get_read_conn() + assert baseline is not None, "baseline read open should succeed" + db._close_read_conn(baseline) + + db._read_open_failed_at = _time.monotonic() + assert db._get_read_conn() is None, "should back off immediately after a failure" + + db._read_open_failed_at = _time.monotonic() - (_READ_OPEN_RETRY_SECONDS + 1) + recovered = db._get_read_conn() + assert recovered is not None, "read path must self-heal once the window expires" + db._close_read_conn(recovered) + + +@pytest.mark.requires_wal +def test_checkout_seam_is_the_single_acquisition_point(db): + """``_read_ctx`` must acquire via ``_checkout_read_conn`` and nothing else. + + If a future edit re-inlines the pool checkout into ``_read_ctx``, patching + ``_get_read_conn`` silently exercises nothing whenever the pool is warm -- + which is exactly how the writer-lock fallback test below would rot into a + no-op without failing. + """ + calls = [] + original = db._checkout_read_conn + + def _spy(): + calls.append(1) + return original() + + db._checkout_read_conn = _spy + try: + with db._read_ctx(): + pass + finally: + db._checkout_read_conn = original + assert calls, "_read_ctx must route acquisition through _checkout_read_conn" + + +def test_fallback_to_locked_writer_when_read_conn_unavailable(db, monkeypatch): + """With no read connection available, reads still work under self._lock. + + Patched at the acquisition SEAM rather than at ``_get_read_conn``: the + pool is consulted first, so a patched ``_get_read_conn`` is never reached + while the pool holds a connection and this test would pass while + exercising nothing. + """ + monkeypatch.setattr(db, "_checkout_read_conn", lambda: None) + assert db.get_session("s1")["id"] == "s1" + assert db.search_messages("graphiti", limit=5) + + +@pytest.mark.requires_wal +def test_peak_live_connections_bounded_under_simultaneous_burst(db): + """N readers checked out AT THE SAME INSTANT must not open N connections. + + This is the assertion the join-then-count test above cannot make. A + LifoQueue with a maxsize bounds how many connections are RETURNED, not how + many are OPEN: with an open-on-miss checkout, 64 readers arriving on a cold + pool opened 64 descriptors and only then closed 56 of them on release. + Bounded at rest, unbounded at peak -- and EMFILE is a peak-instant + condition, so the process could still wedge exactly as it did in + production. + + The barrier is the whole point: every worker holds its connection until all + of them have checked out, so the count below IS the simultaneous peak + rather than a sample of it. + """ + from hermes_state import _READ_POOL_MAX + + n = 64 + assert n > _READ_POOL_MAX, "burst must exceed the ceiling to test anything" + + ready = threading.Barrier(n + 1) + release = threading.Event() + checked_out = [] + fell_back = [] + lock = threading.Lock() + + def worker(): + conn = db._checkout_read_conn() + with lock: + (checked_out if conn is not None else fell_back).append(conn) + ready.wait(timeout=30) # everyone is now holding whatever they got + release.wait(timeout=30) + if conn is not None: + db._close_read_conn(conn) + + threads = [threading.Thread(target=worker) for _ in range(n)] + for t in threads: + t.start() + + ready.wait(timeout=30) + # ---- the instant every worker is simultaneously checked out ---- + peak_live = _live_count(db.db_path) + peak_checked_out = len(checked_out) + release.set() + for t in threads: + t.join(timeout=30) + + assert peak_checked_out <= _READ_POOL_MAX, ( + f"{peak_checked_out} connections checked out at once; the ceiling is " + f"{_READ_POOL_MAX}. Peak is unbounded -- the pool bounds returns, not opens." + ) + # +1 for the writer connection SessionDB always holds. + assert peak_live <= _READ_POOL_MAX + 1, ( + f"{peak_live} live connections at peak, ceiling is {_READ_POOL_MAX} (+1 writer)" + ) + assert fell_back, "with n > ceiling some readers must degrade to the writer path" + assert len(checked_out) + len(fell_back) == n, "every worker must be accounted for" + + +@pytest.mark.requires_wal +def test_exhausted_permits_fall_back_to_the_writer_connection(db): + """Past the ceiling the read path degrades, it does not fail or block. + + A reader that cannot get a permit must serve from the locked writer + connection. Blocking instead would convert descriptor exhaustion into a + stall -- the same outage with a different stack trace. + """ + from hermes_state import _READ_POOL_MAX + + held = [db._checkout_read_conn() for _ in range(_READ_POOL_MAX)] + assert all(c is not None for c in held), "the first _READ_POOL_MAX must succeed" + try: + assert db._checkout_read_conn() is None, "ceiling must refuse the next open" + with db._read_ctx() as conn: + assert conn is db._conn, "must fall back to the shared writer connection" + assert conn.execute("SELECT 1").fetchone()[0] == 1, "fallback must work" + finally: + for c in held: + db._close_read_conn(c) + + # Permits come back: the read path recovers once the burst drains. + recovered = db._checkout_read_conn() + assert recovered is not None, "permits must be released back after close" + db._close_read_conn(recovered) + + +@pytest.mark.requires_wal +def test_permits_are_not_stranded_by_a_failed_open(db, monkeypatch): + """A failed open must return its permit, or the ceiling ratchets to zero. + + A permit leaked per failure is not a transient error: it permanently + shrinks the read path, so a burst of transient open failures would silently + demote every later read to the writer lock for the life of the process. + """ + import sqlite3 as _sqlite3 + + import hermes_state as _hs + from hermes_state import _READ_POOL_MAX + + def boom(*a, **kw): + raise _sqlite3.OperationalError("simulated open failure") + + monkeypatch.setattr(_hs, "_connect_tracked_db", boom) + for _ in range(_READ_POOL_MAX * 3): + assert db._get_read_conn() is None + db._read_open_failed_at = 0.0 # defeat the backoff so every call opens + monkeypatch.undo() + + db._read_open_failed_at = 0.0 + held = [db._checkout_read_conn() for _ in range(_READ_POOL_MAX)] + try: + assert all(c is not None for c in held), ( + "permits were stranded by failed opens -- the ceiling ratcheted down" + ) + finally: + for c in held: + if c is not None: + db._close_read_conn(c) + + +@pytest.mark.requires_wal +def test_close_returns_every_permit(db): + """close() must release the permits its drained connections held.""" + from hermes_state import _READ_POOL_MAX + + held = [db._checkout_read_conn() for _ in range(_READ_POOL_MAX)] + for c in held: + db._read_pool.put_nowait(c) + assert db._read_pool.qsize() == _READ_POOL_MAX + + db.close() + assert db._read_pool.qsize() == 0 + assert _live_count(db.db_path) == 0 + # BoundedSemaphore raises on over-release, so draining exactly + # _READ_POOL_MAX permits proves close() released neither too few nor too + # many. + for _ in range(_READ_POOL_MAX): + assert db._read_permits.acquire(blocking=False), "close() stranded a permit" + assert not db._read_permits.acquire(blocking=False), "close() over-released" diff --git a/tests/test_session_db_read_path_split.py b/tests/test_session_db_read_path_split.py index 35c54228301a..fb173ca94268 100644 --- a/tests/test_session_db_read_path_split.py +++ b/tests/test_session_db_read_path_split.py @@ -1,11 +1,12 @@ -"""Tests for the SessionDB read-path split (per-thread read-only connections). +"""Tests for the SessionDB read-path split (pooled read-only connections). The gateway shares ONE SessionDB across every agent, so recall/browse reads used to queue behind writer flushes on self._lock — a measured production convoy (a 0.2s FTS query stretched to 112s while 6-8 concurrent turns flushed tool results). These tests pin the new contract: reads run on a -per-thread read-only connection under WAL, never touch self._lock, and fall -back to the legacy locked path when WAL or the read connection is missing. +read-only connection borrowed from a bounded pool under WAL, never touch +self._lock, and fall back to the legacy locked path when WAL or the read +connection is missing. """ import threading @@ -39,8 +40,19 @@ def grab(key): assert conns[1] is not conns[2] -def test_read_conn_reused_within_thread(db): - assert db._get_read_conn() is db._get_read_conn() +@pytest.mark.requires_wal +def test_read_conn_reused_via_pool(db): + """Reuse is now the pool's job, not a per-thread memo. + + The old contract (``_get_read_conn()`` returns the same object twice on one + thread) was the leak: that memo pinned one unclosable connection per + (SessionDB x thread) forever. ``_get_read_conn`` now always opens a fresh + connection and reuse happens via checkout/return, so assert on that. + """ + with db._read_ctx() as first: + assert first is not None + with db._read_ctx() as second: + assert second is first, "sequential readers must reuse the pooled conn" @pytest.mark.requires_wal