RomanNum3ral
/
hermes-agent
mirror of https://github.com/NousResearch/hermes-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

state: bound PEAK read connections with a permit, not just pooled returns 

Review of this PR was right that maxsize=8 bounds the wrong thing. The
LifoQueue caps how many connections are RETURNED; _checkout_read_conn opened
unconditionally on a miss, so N readers arriving on a cold pool all missed, all
opened, and peaked at N. The surplus was closed on release, so nothing
accumulated 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 the
original wedge was still reachable. Measured on the previous commit: 64
concurrent readers held 64 live connections at once.

A connection now holds a permit for its whole lifetime -- acquired in
_get_read_conn() before the open, released in _close_read_conn() after the
close -- so open+checked-out is bounded together. A pool hit costs no permit
because the connection it hands back already holds one, which leaves
_get_read_conn() as the only place that can open. The acquire is non-blocking:
past the ceiling readers fall back to the locked writer connection rather than
queueing, since blocking would convert descriptor exhaustion into a stall,
which is the same outage with a different stack trace. Same burst now peaks at
8. BoundedSemaphore rather than Semaphore so an unpaired release raises instead
of silently widening the ceiling.

Two latent leaks in the same function, found while doing this:

  - a CJK extension load that failed after a successful open returned None
    without closing the connection, leaking a descriptor the tracking registry
    still counted -- the same leak shape one level down;
  - any non-sqlite3.Error between open and return stranded a permit
    permanently, which would ratchet the ceiling down to zero and silently
    demote every later read to the writer lock.

On the test: the existing one joins every worker before counting, so it
measures the pool at rest and structurally cannot observe peak -- which is why
this got through. The new one uses a barrier so all 64 workers hold their
connections until every worker has checked out, making the count taken at that
moment the actual simultaneous peak. Verified it fails against the previous
commit (64 checked out, 65 live) and passes at 8/9. Also covers the
writer-connection fallback, permit recovery after a failed open, and that
close() releases exactly the permits it drained.
This commit is contained in:
Yishova 2026-08-02 06:33:05 -04:00 committed by Teknium
parent 87aedbe7b6
commit 0472c31aa1
2 changed files with 275 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

122
hermes_state.py
View File

@ -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:

161
tests/test_session_db_read_conn_pool.py
View File

@ -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"