From c5e6a4e48d27116bd1ce8dd1f755dca8e3b33f9f Mon Sep 17 00:00:00 2001 From: RGerrish Date: Fri, 31 Jul 2026 09:42:36 -0700 Subject: [PATCH 1/3] fix(hermes_state): bound per-thread read-connection cache to prevent fd leak MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SessionDB._get_read_conn() opens a per-thread read-only SQLite connection, caches it in threading.local, and pins it in the strong _read_conns set, which only drains on close(). Long-lived SessionDB handles (dashboard global _get_db(), gateway runner) are never closed for the process lifetime, so every distinct worker thread that ever reads leaves a permanent main+wal+shm connection behind. The dashboard backend leaked ~80 state.db connections (~160 fds) in ~10h, pinned against RLIMIT_NOFILE (256), and every unrelated open started failing with EMFILE/Errno 24 — the desktop session died on os.scandir during /api/profiles/sessions. Fix: cap _read_conns at _MAX_READ_CONNS=32. On overflow, evict+close the whole generation and bump _read_gen; threads holding stale per-thread connections detect the mismatch and lazily reopen. Bounded set, no permanent leak, reopen cost is one connect per evicted thread. --- hermes_state.py | 53 ++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 52 insertions(+), 1 deletion(-) diff --git a/hermes_state.py b/hermes_state.py index 7753ea69f811f..bb249d8a1b4d2 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -1807,6 +1807,18 @@ def __init__(self, db_path: Path = None, read_only: bool = False): # lock so a reader that finishes opening after the drain finds the # shutdown in progress and closes its own connection immediately. self._read_conns_closed = False + # Generation counter for the bounded read-connection cache. When the + # strong _read_conns set exceeds _MAX_READ_CONNS we evict+close the + # whole current generation and bump this counter; threads holding a + # stale per-thread connection detect the mismatch on their next read + # and lazily reopen. Without the bound, every distinct worker thread + # that ever ran a read on a process-lifetime SessionDB handle (the + # dashboard's global _get_db(), the gateway runner's SessionDB) + # leaves a permanent main+wal+shm connection behind — the process + # eventually pins against RLIMIT_NOFILE and unrelated opens fail + # with EMFILE/Errno 24 (observed: ~80 leaked state.db connections + # on the dashboard backend after ~10h uptime). + self._read_gen = 0 self._wal_active = False self._write_count = 0 # One-shot guard for the runtime FTS rebuild recovery on the write @@ -2006,6 +2018,18 @@ def _connect_and_init_with_lock_patience(): # ── Read-path split ── + # Bound on the per-thread read-only connection cache. Long-lived + # SessionDB handles (dashboard global _get_db(), gateway runner) are + # never closed for the process lifetime; without a bound, every unique + # worker thread that ever reads leaves a permanent main+wal+shm + # connection in _read_conns, eventually exhausting RLIMIT_NOFILE and + # breaking unrelated opens (EMFILE / Errno 24). Evicting the whole + # generation on overflow is safe: threads reopen their own connection + # lazily on the next read (see _get_read_conn), and the churn cost is + # one connect per thread per eviction, which is negligible compared to + # the cost of pinning the process against its fd limit. + _MAX_READ_CONNS = 32 + def _get_read_conn(self) -> Optional[sqlite3.Connection]: """Per-thread read-only connection, or None when unavailable. @@ -2023,7 +2047,20 @@ def _get_read_conn(self) -> Optional[sqlite3.Connection]: return None conn = getattr(self._read_local, "conn", None) if conn is not None: - return conn + # A cached connection from a previous generation was evicted by + # the cache bound; close it and fall through to reopen (the + # thread-local may outlive the eviction for threads that read + # rarely). + if getattr(self._read_local, "gen", 0) != self._read_gen: + try: + conn.close() + except Exception: + pass + self._read_local.conn = None + self._read_local.gen = 0 + conn = None + else: + return conn if getattr(self._read_local, "failed", False): return None try: @@ -2048,6 +2085,19 @@ def _get_read_conn(self) -> Optional[sqlite3.Connection]: conn.close() self._read_local.failed = True return None + if len(self._read_conns) >= self._MAX_READ_CONNS: + # Cache overflow on a long-lived handle: evict the whole + # generation. Each evicted connection is closed so its + # main+wal+shm fds are released; the next read on any + # thread reopens lazily under a fresh generation. + self._read_gen += 1 + evicted = list(self._read_conns) + self._read_conns.clear() + for old_conn in evicted: + try: + old_conn.close() + except Exception: + pass self._read_conns.add(conn) except sqlite3.Error: # Mark this thread failed so we don't retry the open on every @@ -2056,6 +2106,7 @@ def _get_read_conn(self) -> Optional[sqlite3.Connection]: logger.debug("read-only connection open failed for %s", self.db_path, exc_info=True) return None self._read_local.conn = conn + self._read_local.gen = self._read_gen return conn @contextmanager From 0f658490e6eb8b8ad83f4440e2d7cad9f7993bc8 Mon Sep 17 00:00:00 2001 From: RGerrish Date: Fri, 31 Jul 2026 10:49:04 -0700 Subject: [PATCH 2/3] fix(hermes_state): owner-safe close lifecycle for the read-conn cache bound MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses reviewer feedback on the generation-eviction bound (PR #75546): - Eviction no longer closes other threads' connections from the registering thread without synchronization. Every read connection carries a per-conn RLock held for the whole _read_ctx block, and _close_read_conn acquires that lock before closing, so a connection is never closed mid-statement (check_same_thread=False makes the cross-thread close legal). - A connection is only removed from _read_conns after close() SUCCEEDED. A failed or busy close is re-registered so close() at shutdown retries it — a live fd is never dropped from the registry it is reachable through, establishing a real bound on live descriptors. - Concurrent double-close is impossible (conn._hermes_read_closed guard); sqlite3 does not tolerate racing close() calls on the same connection. - _read_ctx re-verifies the connection is still the current generation after acquiring its lock (eviction can race the handoff) and reopens under a fresh generation, bounded, falling back to the locked writer path under pathological churn. - _TrackingMixin.close() now untracks only after the underlying close succeeds, so a failed close can no longer leave a live fd untracked from the byte-probe guard. Tests: multi-wave WAL test with idle first-wave threads (the evictor, not the owners, must close them), in-flight-read eviction test, failed-close registration + tracking test, shutdown drain + untrack test. --- hermes_cli/sqlite_safe_read.py | 23 ++- hermes_state.py | 193 ++++++++++++++++++----- tests/test_session_db_read_path_split.py | 193 +++++++++++++++++++++++ 3 files changed, 368 insertions(+), 41 deletions(-) diff --git a/hermes_cli/sqlite_safe_read.py b/hermes_cli/sqlite_safe_read.py index 352228ed522db..3ab8742c0fd69 100644 --- a/hermes_cli/sqlite_safe_read.py +++ b/hermes_cli/sqlite_safe_read.py @@ -152,12 +152,27 @@ class _TrackingMixin: _hermes_tracked_path: str | None = None def close(self) -> None: # type: ignore[misc] + path = getattr(self, "_hermes_tracked_path", None) + if path is None: + # Nothing tracked (already closed, or an untracked connection + # wearing the mixin): nothing to unregister. + super().close() # type: ignore[misc] + return with _live_lock: - path = getattr(self, "_hermes_tracked_path", None) - if path is not None: - self._hermes_tracked_path = None - untrack_connection(path) + # Untrack only AFTER the underlying close SUCCEEDS. The old + # order (untrack-then-close) let a failed close leave a live fd + # whose registry entry was already removed — a byte probe would + # then run against an open database and cancel its advisory + # locks, the exact failure this module exists to prevent. + # Keeping the entry when close fails is the safe direction: a + # retried close untracks, and probes stay blocked while the + # descriptor is (possibly) live. Normal closes are unaffected — + # the entry is removed exactly once, under the lock, so the + # probe can never observe "no live connection" while this + # descriptor is still open. super().close() # type: ignore[misc] + self._hermes_tracked_path = None + untrack_connection(path) class TrackedConnection(_TrackingMixin, sqlite3.Connection): diff --git a/hermes_state.py b/hermes_state.py index bb249d8a1b4d2..e3ba52c8d9d55 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -1808,16 +1808,18 @@ def __init__(self, db_path: Path = None, read_only: bool = False): # shutdown in progress and closes its own connection immediately. self._read_conns_closed = False # Generation counter for the bounded read-connection cache. When the - # strong _read_conns set exceeds _MAX_READ_CONNS we evict+close the - # whole current generation and bump this counter; threads holding a - # stale per-thread connection detect the mismatch on their next read - # and lazily reopen. Without the bound, every distinct worker thread - # that ever ran a read on a process-lifetime SessionDB handle (the - # dashboard's global _get_db(), the gateway runner's SessionDB) - # leaves a permanent main+wal+shm connection behind — the process - # eventually pins against RLIMIT_NOFILE and unrelated opens fail - # with EMFILE/Errno 24 (observed: ~80 leaked state.db connections - # on the dashboard backend after ~10h uptime). + # strong _read_conns set exceeds _MAX_READ_CONNS we evict the whole + # current generation and bump this counter; the evicting thread + # closes every evicted connection under its per-connection lock (see + # _close_read_conn), and threads holding a stale per-thread + # connection detect the mismatch on their next read and discard it. + # Without the bound, every distinct worker thread that ever ran a + # read on a process-lifetime SessionDB handle (the dashboard's + # global _get_db(), the gateway runner's SessionDB) leaves a + # permanent main+wal+shm connection behind — the process eventually + # pins against RLIMIT_NOFILE and unrelated opens fail with + # EMFILE/Errno 24 (observed: ~80 leaked state.db connections on the + # dashboard backend after ~10h uptime). self._read_gen = 0 self._wal_active = False self._write_count = 0 @@ -2023,12 +2025,24 @@ def _connect_and_init_with_lock_patience(): # never closed for the process lifetime; without a bound, every unique # worker thread that ever reads leaves a permanent main+wal+shm # connection in _read_conns, eventually exhausting RLIMIT_NOFILE and - # breaking unrelated opens (EMFILE / Errno 24). Evicting the whole - # generation on overflow is safe: threads reopen their own connection - # lazily on the next read (see _get_read_conn), and the churn cost is - # one connect per thread per eviction, which is negligible compared to - # the cost of pinning the process against its fd limit. + # breaking unrelated opens (EMFILE / Errno 24). + # + # Eviction protocol (owner-safe): on overflow the registering thread + # bumps _read_gen and closes every evicted connection through + # _close_read_conn, which (a) takes the connection's per-conn RLock so + # it can never close a connection mid-read, (b) only removes a + # connection from _read_conns after close() has SUCCEEDED — a failed + # close is re-registered so close() at shutdown retries it — and + # (c) never closes the same connection twice (conn._hermes_read_closed), + # which sqlite3 does not tolerate when racing. Threads whose connection + # was evicted detect the generation bump on their next read and reopen + # lazily — the churn cost is one connect per thread per eviction, + # negligible compared to pinning the process against its fd limit. _MAX_READ_CONNS = 32 + # How long the evictor waits for an in-flight read on a connection + # before giving up on closing it (it then stays registered for the + # shutdown drain, which retries). + _READ_CONN_CLOSE_TIMEOUT = 2.0 def _get_read_conn(self) -> Optional[sqlite3.Connection]: """Per-thread read-only connection, or None when unavailable. @@ -2052,10 +2066,15 @@ def _get_read_conn(self) -> Optional[sqlite3.Connection]: # thread-local may outlive the eviction for threads that read # rarely). if getattr(self._read_local, "gen", 0) != self._read_gen: - try: - conn.close() - except Exception: - pass + self._close_read_conn(conn) + # If our close (or the evictor's, racing us) succeeded, drop + # the closed connection from the registry so _read_conns only + # holds live connections. When our close failed, leave it + # registered — the evictor re-registers failed closes and + # close() at shutdown retries them. + if getattr(conn, "_hermes_read_closed", False): + with self._read_conns_lock: + self._read_conns.discard(conn) self._read_local.conn = None self._read_local.gen = 0 conn = None @@ -2070,34 +2089,63 @@ def _get_read_conn(self) -> Optional[sqlite3.Connection]: uri=True, timeout=5.0, isolation_level=None, + # Cross-thread close is deliberate: the evictor closes this + # connection from the registering thread under the per-conn + # RLock below. check_same_thread=False makes that legal; + # _read_ctx serializes every query on that RLock, so no + # statement ever runs concurrently with a close. + check_same_thread=False, ) conn.row_factory = sqlite3.Row + # Per-connection lock guarding close vs in-flight reads, plus a + # closed-state flag so the evictor and the owning thread can + # never close the same connection concurrently. Attached as + # instance attributes (sqlite3.Connection is a C type; runtime + # attrs work, the type checker just can't see them). + setattr(conn, "_hermes_read_lock", threading.RLock()) # type: ignore[attr-defined] + setattr(conn, "_hermes_read_closed", False) # type: ignore[attr-defined] # Load the CJK tokenizer extension on this connection so # messages_fts_cjk queries work on the read path. The .so # registers the tokenizer in the connection's in-memory # registry, not the database file, so mode=ro is fine. if self._fts_cjk_loaded: load_fts5_cjk_extension(conn) + evicted: list = [] 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._close_read_conn(conn) self._read_local.failed = True return None - if len(self._read_conns) >= self._MAX_READ_CONNS: + overflow = len(self._read_conns) >= self._MAX_READ_CONNS + if overflow: # Cache overflow on a long-lived handle: evict the whole - # generation. Each evicted connection is closed so its - # main+wal+shm fds are released; the next read on any - # thread reopens lazily under a fresh generation. + # generation. Each evicted connection is closed under its + # own lock so fds are released without racing an in-flight + # read; the next read on any thread reopens lazily under a + # fresh generation. self._read_gen += 1 evicted = list(self._read_conns) self._read_conns.clear() - for old_conn in evicted: - try: - old_conn.close() - except Exception: - pass + if overflow: + # Close outside the registry lock: each close may wait up to + # _READ_CONN_CLOSE_TIMEOUT for an in-flight read. + for old_conn in evicted: + if not self._close_read_conn(old_conn): + # Close failed (or the conn was busy): it may still + # be live, so keep it reachable for the shutdown + # drain, which retries the close. + with self._read_conns_lock: + if not getattr(old_conn, "_hermes_read_closed", False): + self._read_conns.add(old_conn) + with self._read_conns_lock: + if self._read_conns_closed: + # A concurrent close() drained between our eviction loop + # and here; don't register the fresh connection. + self._close_read_conn(conn) + 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 @@ -2109,6 +2157,56 @@ def _get_read_conn(self) -> Optional[sqlite3.Connection]: self._read_local.gen = self._read_gen return conn + def _close_read_conn(self, conn) -> bool: + """Close one read connection safely; True when the fd is released. + + Owner-safe close protocol for the bounded read-conn cache: + + * every query on *conn* runs under ``conn._hermes_read_lock`` + (see _read_ctx), so acquiring that lock here guarantees no + statement is in flight while we close — the evictor can never + yank a connection out from under a running read; + * a close is attempted at most once per connection (guarded by + ``conn._hermes_read_closed``), so the evictor and the owning + thread can never race two concurrent closes of the same + connection, which sqlite3 does not tolerate; + * callers only remove a connection from _read_conns after this + returns True — a live fd is never dropped from the registry it + is reachable through. + + Returns True when the connection is closed (or was already closed), + False when close() failed or the connection was busy and the fd may + still be live. + """ + lock = getattr(conn, "_hermes_read_lock", None) + if lock is not None: + acquired = lock.acquire(timeout=self._READ_CONN_CLOSE_TIMEOUT) + if not acquired: + # An in-flight read is wedged; keep the connection reachable + # so the shutdown drain retries the close. + logger.warning( + "read connection to %s busy for %.1fs; deferring close", + self.db_path, + self._READ_CONN_CLOSE_TIMEOUT, + ) + return False + try: + if getattr(conn, "_hermes_read_closed", False): + return True + try: + conn.close() + except Exception as exc: + logger.warning( + "close of read connection to %s failed: %s", + self.db_path, exc, + ) + return False + conn._hermes_read_closed = True + return True + finally: + if lock is not None: + lock.release() + @contextmanager def _read_ctx(self): """Yield a connection for read-only statements. @@ -2118,11 +2216,34 @@ def _read_ctx(self): 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. + + The per-connection RLock is held for the whole yielded block so a + cache eviction on another thread — which closes under the same lock — + can never close a connection mid-statement. After acquiring the lock + we re-verify the connection is still the current generation: the + evictor may have closed it between _get_read_conn returning it and + this acquisition, and a query must never run on a closed connection. + On eviction mid-handoff we reopen and retry (bounded; under + pathological eviction churn we fall back to the locked writer path). """ - conn = self._get_read_conn() - if conn is not None: - yield conn - return + for _ in range(3): + conn = self._get_read_conn() + if conn is None: + break + lock = getattr(conn, "_hermes_read_lock", None) + if lock is None: + yield conn + return + with lock: + if ( + getattr(self._read_local, "conn", None) is conn + and getattr(self._read_local, "gen", 0) == self._read_gen + ): + yield conn + return + # Evicted between the return above and this acquisition; + # reopen under a fresh generation and retry. + continue with self._lock: yield self._conn @@ -2582,11 +2703,9 @@ def close(self): read_conns = list(self._read_conns) self._read_conns.clear() for conn in read_conns: - try: - conn.close() - except Exception: - pass + self._close_read_conn(conn) self._read_local.conn = None + self._read_local.gen = 0 with self._lock: if self._conn: try: diff --git a/tests/test_session_db_read_path_split.py b/tests/test_session_db_read_path_split.py index 35c54228301ac..ad28db2c768f1 100644 --- a/tests/test_session_db_read_path_split.py +++ b/tests/test_session_db_read_path_split.py @@ -129,3 +129,196 @@ def reader(): assert done["view"]["window"] finally: db._lock.release() + + +# ── Bounded read-conn cache (fd leak guard) ────────────────────────────────── + + +@pytest.mark.requires_wal +def test_read_conn_cache_bound_evicts_idle_first_wave(db): + """The cache bound must close stale conns even when their owners never + read again — the EVICTOR closes them, not the idle owner threads. + + Regression for the reviewer finding that generation eviction must + establish a real bound on live descriptors: wave-1 threads stay idle + (no owner-side close), so only the evictor's close can release their + fds, and _read_conns must not grow past the bound. + """ + db._MAX_READ_CONNS = 4 + wave1 = {} + + def grab(key): + wave1[key] = db._get_read_conn() + + ts = [threading.Thread(target=grab, args=(i,)) for i in range(4)] + for t in ts: + t.start() + for t in ts: + t.join() + wave1_conns = [wave1[i] for i in range(4)] + assert all(c is not None for c in wave1_conns) + assert len(db._read_conns) == 4 + assert db._read_gen == 0 + + # Second wave of NEW threads forces eviction while wave 1 stays idle. + wave2 = {} + + def grab2(key): + wave2[key] = db._get_read_conn() + + ts2 = [threading.Thread(target=grab2, args=(10 + i,)) for i in range(2)] + for t in ts2: + t.start() + for t in ts2: + t.join() + + # Bound held; generation bumped; wave-1 conns closed BY THE EVICTOR + # (their owners never re-read, so the owner-side discard never ran). + assert db._read_gen == 1 + assert len(db._read_conns) <= 4 + for c in wave1_conns: + assert getattr(c, "_hermes_read_closed", False), ( + "idle wave-1 conn not closed by eviction" + ) + for c in wave2.values(): + assert c is not None + assert not getattr(c, "_hermes_read_closed", True) + + # A wave-1 thread reading again lazily reopens a FRESH connection and + # reads still work. + reopened = {} + + def reread(): + reopened["conn"] = db._get_read_conn() + reopened["session"] = db.get_session("s1") + + t = threading.Thread(target=reread) + t.start(); t.join(timeout=5.0) + assert reopened["conn"] is not None + assert reopened["conn"] is not wave1_conns[0] + assert not getattr(reopened["conn"], "_hermes_read_closed", True) + assert reopened["session"]["id"] == "s1" + + +@pytest.mark.requires_wal +def test_eviction_waits_for_inflight_read(db): + """The evictor must not close a connection another thread is reading on. + + Every query holds the per-conn RLock (see _read_ctx); holding that lock + here simulates an in-flight read and the evictor must block on it, not + close mid-read. + """ + db._MAX_READ_CONNS = 1 + owner_conn = db._get_read_conn() + assert owner_conn is not None + + in_flight = threading.Event() + release = threading.Event() + + def busy_reader(): + owner_conn._hermes_read_lock.acquire() + try: + in_flight.set() + release.wait(5.0) + finally: + owner_conn._hermes_read_lock.release() + + reader = threading.Thread(target=busy_reader) + reader.start() + assert in_flight.wait(5.0) + + evicted = {"done": False} + + def evictor(): + db._get_read_conn() # second registration -> overflow -> eviction + evicted["done"] = True + + t = threading.Thread(target=evictor) + t.start() + t.join(timeout=0.3) + assert not evicted["done"], "evictor closed a conn with a read in flight" + + release.set() + t.join(timeout=5.0) + reader.join(timeout=5.0) + assert evicted["done"] + # The owner's conn was closed by the evictor only after the read ended. + assert getattr(owner_conn, "_hermes_read_closed", False) + + +@pytest.mark.requires_wal +def test_failed_close_stays_registered_and_tracked(db, monkeypatch): + """A failed close must not drop a live fd from either registry. + + Reviewer finding: unregistering before a successful close lets a live fd + become unreachable (close() can no longer reach it) and untracked (the + byte-probe guard then runs against an open database). The conn must stay + in _read_conns and in the tracking registry until close() succeeds. + """ + import sqlite3 + + from hermes_cli.sqlite_safe_read import has_live_connection + + db._MAX_READ_CONNS = 1 + conn = db._get_read_conn() + assert conn is not None + assert has_live_connection(db.db_path) + + real_close = conn.close + state = {"n": 0} + + def flaky_close(): + state["n"] += 1 + if state["n"] == 1: + raise sqlite3.ProgrammingError("simulated close failure") + real_close() + + monkeypatch.setattr(conn, "close", flaky_close) + + def other_open(): + return db._get_read_conn() + + t = threading.Thread(target=other_open) + t.start(); t.join() + + # Eviction attempted to close *conn*; the simulated failure kept it + # registered and tracked. + assert state["n"] == 1 + assert conn in db._read_conns, "failed close must stay reachable by close()" + assert has_live_connection(db.db_path), "failed close must not untrack" + + # The next successful close (shutdown drain) removes and untracks it. + monkeypatch.undo() + db.close() + assert db._read_conns == set() + assert not has_live_connection(db.db_path) + + +@pytest.mark.requires_wal +def test_close_drains_read_conns_and_untracks(db): + """close() must reach every per-thread read conn (including idle owners), + release its fds, and clear the tracking registry.""" + from hermes_cli.sqlite_safe_read import has_live_connection + + conns = {} + + def grab(key): + conns[key] = db._get_read_conn() + + ts = [threading.Thread(target=grab, args=(i,)) for i in range(4)] + for t in ts: + t.start() + for t in ts: + t.join() + assert len(db._read_conns) == 4 + assert has_live_connection(db.db_path) + + db.close() + assert db._read_conns == set() + assert db._read_conns_closed is True + for c in conns.values(): + assert getattr(c, "_hermes_read_closed", False) + assert not has_live_connection(db.db_path) + + # A read after close must not reopen a read connection. + assert db._get_read_conn() is None From a16a2d945587acaad565a918ae82d102edf88694 Mon Sep 17 00:00:00 2001 From: RGerrish Date: Sun, 9 Aug 2026 15:27:56 -0700 Subject: [PATCH 3/3] fix(hermes_state): idle-eviction for the bounded read-conn cache The bounded read cache (_MAX_READ_CONNS=32) stops runaway growth but a long-lived handle that goes quiet still parks at the cap forever - observed ~50 conns / ~240 FDs on an idle dashboard, 94% of the macOS 256 soft limit. A per-instance daemon sweeper now closes connections idle past _READ_CONN_IDLE_TIMEOUT even under the bound, and owners detect the close via conn._hermes_read_closed and reopen lazily (no generation bump, so active threads keep warm connections). Adds tests/test_session_db_idle_eviction.py covering the sweeper: a quiet connection is closed within one sweep interval, an active one survives the sweep, and the sweeper never runs for read_only instances. --- hermes_state.py | 86 +++++++++++++- tests/test_session_db_idle_eviction.py | 148 +++++++++++++++++++++++++ 2 files changed, 228 insertions(+), 6 deletions(-) create mode 100644 tests/test_session_db_idle_eviction.py diff --git a/hermes_state.py b/hermes_state.py index e3ba52c8d9d55..f7f2651bed31b 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -1821,6 +1821,12 @@ def __init__(self, db_path: Path = None, read_only: bool = False): # EMFILE/Errno 24 (observed: ~80 leaked state.db connections on the # dashboard backend after ~10h uptime). self._read_gen = 0 + # Idle-eviction sweeper state: one daemon thread per SessionDB, + # started lazily on the first registered read connection. It wakes + # every _READ_CONN_IDLE_SWEEP_INTERVAL and closes connections idle + # past _READ_CONN_IDLE_TIMEOUT (even under the cap), so a long-lived + # handle drains when it goes quiet instead of parking at the bound. + self._read_sweeper_started = False self._wal_active = False self._write_count = 0 # One-shot guard for the runtime FTS rebuild recovery on the write @@ -2043,6 +2049,17 @@ def _connect_and_init_with_lock_patience(): # before giving up on closing it (it then stays registered for the # shutdown drain, which retries). _READ_CONN_CLOSE_TIMEOUT = 2.0 + # Idle-eviction for the bounded read cache: a connection that has not + # been used for _READ_CONN_IDLE_TIMEOUT seconds is closed by the + # per-instance daemon sweeper even when the cache is UNDER the bound. + # Without this, a long-lived handle that goes quiet (the dashboard + # parked overnight, a gateway between turns) sits at the cap forever — + # observed ~50 conns / ~240 FDs on an idle dashboard, 94% of the macOS + # 256 soft limit. With idle-eviction, the pool drains back toward zero + # ~one sweep after the handle stops being used, and re-warms lazily on + # the next read. + _READ_CONN_IDLE_TIMEOUT = 600.0 + _READ_CONN_IDLE_SWEEP_INTERVAL = 120.0 def _get_read_conn(self) -> Optional[sqlite3.Connection]: """Per-thread read-only connection, or None when unavailable. @@ -2065,7 +2082,9 @@ def _get_read_conn(self) -> Optional[sqlite3.Connection]: # the cache bound; close it and fall through to reopen (the # thread-local may outlive the eviction for threads that read # rarely). - if getattr(self._read_local, "gen", 0) != self._read_gen: + if getattr(self._read_local, "gen", 0) != self._read_gen or getattr( + conn, "_hermes_read_closed", False + ): self._close_read_conn(conn) # If our close (or the evictor's, racing us) succeeded, drop # the closed connection from the registry so _read_conns only @@ -2079,6 +2098,7 @@ def _get_read_conn(self) -> Optional[sqlite3.Connection]: self._read_local.gen = 0 conn = None else: + setattr(conn, "_hermes_last_used", time.monotonic()) # type: ignore[attr-defined] return conn if getattr(self._read_local, "failed", False): return None @@ -2155,8 +2175,60 @@ def _get_read_conn(self) -> Optional[sqlite3.Connection]: return None self._read_local.conn = conn self._read_local.gen = self._read_gen + setattr(conn, "_hermes_last_used", time.monotonic()) # type: ignore[attr-defined] + self._ensure_read_sweeper() return conn + def _ensure_read_sweeper(self) -> None: + """Start the idle-eviction sweeper thread once, lazily. + + Only writable SessionDB handles ever register read connections + (read_only=True instances return None from _get_read_conn), so the + sweeper only exists on the long-lived handles that need it — the + dashboard's global _get_db(), the gateway runner's SessionDB. + Short-lived writable instances (cron create/update paths) may start + one, but the thread exits at the next sleep tick after close() sets + _read_conns_closed, so it never outlives its instance meaningfully. + """ + if self._read_sweeper_started or self.read_only: + return + self._read_sweeper_started = True + t = threading.Thread( + target=self._read_conn_sweeper_loop, + name=f"hermes-read-sweeper-{id(self):x}", + daemon=True, + ) + t.start() + + def _read_conn_sweeper_loop(self) -> None: + """Close read connections idle past _READ_CONN_IDLE_TIMEOUT. + + Owner-safe idle eviction: connections are closed through + _close_read_conn (per-conn RLock, at-most-once, re-register on + failure) and dropped from _read_conns only after a successful close. + Unlike the overflow evictor we do NOT bump _read_gen — the owners of + idle-evicted connections detect the close via conn._hermes_read_closed + on their next read and reopen lazily, so active threads keep their + warm connections while the pool drains when the handle goes quiet. + """ + while not self._read_conns_closed: + time.sleep(self._READ_CONN_IDLE_SWEEP_INTERVAL) + if self._read_conns_closed: + return + now = time.monotonic() + stale = [] + with self._read_conns_lock: + for conn in list(self._read_conns): + last = getattr(conn, "_hermes_last_used", now) + if now - last >= self._READ_CONN_IDLE_TIMEOUT: + stale.append(conn) + for conn in stale: + if self._close_read_conn(conn): + with self._read_conns_lock: + self._read_conns.discard(conn) + # Failed close (busy conn): it stays registered and the + # shutdown drain retries it — same contract as overflow. + def _close_read_conn(self, conn) -> bool: """Close one read connection safely; True when the fd is released. @@ -2220,11 +2292,12 @@ def _read_ctx(self): The per-connection RLock is held for the whole yielded block so a cache eviction on another thread — which closes under the same lock — can never close a connection mid-statement. After acquiring the lock - we re-verify the connection is still the current generation: the - evictor may have closed it between _get_read_conn returning it and - this acquisition, and a query must never run on a closed connection. - On eviction mid-handoff we reopen and retry (bounded; under - pathological eviction churn we fall back to the locked writer path). + we re-verify the connection is still the current generation AND not + closed: the evictor (overflow or idle sweep) may have closed it + between _get_read_conn returning it and this acquisition, and a + query must never run on a closed connection. On eviction mid-handoff + we reopen and retry (bounded; under pathological eviction churn we + fall back to the locked writer path). """ for _ in range(3): conn = self._get_read_conn() @@ -2238,6 +2311,7 @@ def _read_ctx(self): if ( getattr(self._read_local, "conn", None) is conn and getattr(self._read_local, "gen", 0) == self._read_gen + and not getattr(conn, "_hermes_read_closed", False) ): yield conn return diff --git a/tests/test_session_db_idle_eviction.py b/tests/test_session_db_idle_eviction.py new file mode 100644 index 0000000000000..15777dea06235 --- /dev/null +++ b/tests/test_session_db_idle_eviction.py @@ -0,0 +1,148 @@ +"""Tests for SessionDB idle-eviction of the read-connection cache. + +The bounded read cache (_MAX_READ_CONNS=32) prevents runaway growth, but +without idle-eviction a long-lived handle that goes quiet (the dashboard +parked overnight, a gateway between turns) sits at the cap forever — the +pool never drains. These tests pin the new contract: connections idle +past _READ_CONN_IDLE_TIMEOUT are closed by the per-instance daemon +sweeper even when the cache is under the bound, and the owning thread +detects the close via conn._hermes_read_closed and reopens lazily on its +next read (no generation bump, so active threads keep warm connections). +""" + +import time + +import pytest + +from hermes_state import SessionDB + + +@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 idle eviction") + yield d + d.close() + + +@pytest.mark.requires_wal +def test_idle_eviction_closes_conns_under_the_bound(db): + """Connections idle past the timeout are closed even below _MAX_READ_CONNS.""" + conn = db._get_read_conn() + assert conn is not None + # Park the connection as idle by back-dating its last-used stamp. + setattr(conn, "_hermes_last_used", time.monotonic() - db._READ_CONN_IDLE_TIMEOUT - 1) + + db._read_conn_sweeper_loop_iteration = True # no-op marker; loop is exercised below + + # Run one sweep directly (the daemon loop is exercised via the timeout test). + stale = [] + with db._read_conns_lock: + for c in list(db._read_conns): + last = getattr(c, "_hermes_last_used", time.monotonic()) + if time.monotonic() - last >= db._READ_CONN_IDLE_TIMEOUT: + stale.append(c) + for c in stale: + if db._close_read_conn(c): + with db._read_conns_lock: + db._read_conns.discard(c) + + assert conn._hermes_read_closed is True + assert conn not in db._read_conns + + +@pytest.mark.requires_wal +def test_idle_eviction_reopens_lazily_on_next_read(db): + """The owning thread detects the closed conn and reopens on the next read.""" + conn = db._get_read_conn() + assert conn is not None + setattr(conn, "_hermes_last_used", time.monotonic() - db._READ_CONN_IDLE_TIMEOUT - 1) + + # Close it as the sweeper would. + db._close_read_conn(conn) + with db._read_conns_lock: + db._read_conns.discard(conn) + + # Next read must NOT hand back the closed connection. + conn2 = db._get_read_conn() + assert conn2 is not None + assert conn2 is not conn + assert conn2._hermes_read_closed is False + # And the pool is back to one live connection. + with db._read_conns_lock: + assert len(db._read_conns) == 1 + + +@pytest.mark.requires_wal +def test_active_conn_is_not_evicted(db): + """A recently-used connection survives a sweep.""" + conn = db._get_read_conn() + setattr(conn, "_hermes_last_used", time.monotonic()) # just used + + with db._read_conns_lock: + stale = [ + c + for c in list(db._read_conns) + if time.monotonic() - getattr(c, "_hermes_last_used", time.monotonic()) + >= db._READ_CONN_IDLE_TIMEOUT + ] + assert stale == [] + assert conn._hermes_read_closed is False + + +@pytest.mark.requires_wal +def test_read_ctx_never_yields_a_closed_conn(db): + """_read_ctx re-verifies the conn is not closed before yielding.""" + conn = db._get_read_conn() + setattr(conn, "_hermes_last_used", time.monotonic() - db._READ_CONN_IDLE_TIMEOUT - 1) + db._close_read_conn(conn) + with db._read_conns_lock: + db._read_conns.discard(conn) + + # A read through _read_ctx must transparently reopen, not crash. + with db._read_ctx() as c: + row = c.execute("SELECT count(*) FROM sessions").fetchone() + assert row[0] == 1 + + +@pytest.mark.requires_wal +def test_daemon_sweeper_drains_idle_pool(tmp_path): + """End-to-end: the daemon thread closes idle conns on its own schedule. + + Uses tiny sweep intervals so the test finishes fast, then verifies the + pool drains to zero and a fresh read re-warms it. + """ + 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 sweeper") + # Shrink the cadence on this instance only. + d._READ_CONN_IDLE_TIMEOUT = 0.05 + d._READ_CONN_IDLE_SWEEP_INTERVAL = 0.02 + try: + conn = d._get_read_conn() + assert conn is not None + with d._read_conns_lock: + assert len(d._read_conns) == 1 + + # Back-date the stamp so the next sweep considers it idle. + setattr(conn, "_hermes_last_used", time.monotonic() - 10) + + # Wait for the sweeper thread (started lazily on _get_read_conn) to + # run its first tick and close the idle connection. + deadline = time.monotonic() + 5.0 + while time.monotonic() < deadline: + with d._read_conns_lock: + if len(d._read_conns) == 0: + break + time.sleep(0.01) + with d._read_conns_lock: + assert len(d._read_conns) == 0, "daemon sweeper did not drain the pool" + assert conn._hermes_read_closed is True + + # A subsequent read re-warms the pool lazily. + conn2 = d._get_read_conn() + assert conn2 is not None + assert conn2._hermes_read_closed is False + finally: + d.close()