diff --git a/hermes_state.py b/hermes_state.py index 0be368c618e52..2bb6c58fd417f 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -339,6 +339,24 @@ DEFAULT_DB_PATH = get_hermes_home() / "state.db" # 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 @@ -2539,8 +2557,22 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) # Same bug class as the closing(...) fix in gateway/readiness.py # (#69678 / #69567). self._read_pool: "queue.LifoQueue[sqlite3.Connection]" = queue.LifoQueue( - maxsize=8 + 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. _read_ctx checks this under the lock # before returning a connection to the pool, so a reader still in @@ -2817,6 +2849,23 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) < _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", @@ -2842,27 +2891,69 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) if self._fts_cjk_loaded: load_fts5_cjk_extension(conn) except sqlite3.Error: + # 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 + 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, reporting failures. + """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. @@ -2872,6 +2963,11 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) 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 @@ -2889,8 +2985,14 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) 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 or read-conn failure: the shared writer connection under - self._lock, byte-for-byte the legacy behavior. + 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._checkout_read_conn() if conn is not None: @@ -2906,9 +3008,15 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) except queue.Full: pass if not returned: - # More concurrent readers than maxsize, or close() has - # already drained: this connection is surplus. Close it - # here — dropping it on the floor is what leaked the fd. + # 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: diff --git a/tests/test_session_db_read_conn_pool.py b/tests/test_session_db_read_conn_pool.py index 176c4d2419ef6..3d2dabcd3d009 100644 --- a/tests/test_session_db_read_conn_pool.py +++ b/tests/test_session_db_read_conn_pool.py @@ -19,6 +19,16 @@ 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 @@ -58,7 +68,14 @@ def _read(db): @pytest.mark.requires_wal def test_read_pool_is_bounded_across_many_threads(db): - """150 short-lived reader threads must not pin 150 connections.""" + """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" @@ -225,3 +242,145 @@ def test_fallback_to_locked_writer_when_read_conn_unavailable(db, monkeypatch): 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"