ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

fix(state): cap read connections per PROCESS and yield when the fd table is tight

Browse Source 
Follow-up to the per-file budget in the previous commit, which closed the
scaling axis it measured and left three others open.

* A per-file ceiling still lets the cost grow with the PROFILE count: a
  multiplexed gateway serves N profiles from one process and each has its own
  state.db, so `_READ_POOL_MAX` bounded each file while the process total went
  unbounded — the per-instance bug one level out. `_READ_POOL_PROCESS_MAX`
  (three files' worth) now bounds the process, and a miss reclaims an idle
  connection from ANY path before degrading: a profile quiet for an hour must
  not hold descriptors the profile being served right now needs.

* Hermes's SQLite descriptors are only ever a share of the fd table. In #98573
  the ~20 state.db handles were not the whole 256 — they were the share that
  pushed httpx sockets and terminal subprocess pipes over, and the EMFILE
  surfaced in tools/terminal_tool.py rather than here. New read connections are
  now refused when the process is within `_FD_HEADROOM_RESERVE` of its soft
  RLIMIT_NOFILE, measured from /proc/self/fd or /dev/fd and cached briefly. The
  guard fails OPEN where it cannot measure (Windows has neither the fd
  directory nor RLIMIT_NOFILE, and a CRT limit in the thousands) and CLOSED on
  evidence — including a probe that could not get a descriptor of its own.
  `_read_open_denied_fd_headroom` makes it diagnosable from a running process.

* Writer connections cannot be rationed the way read connections can: a
  SessionDB without one cannot write. Their only real bound is not opening
  redundant handles, so a process that accumulates more than
  `_HANDLES_PER_PATH_WARN` handles on one file now says so once, and the next
  duplicate is visible before it is an incident instead of inferred from an
  lsof after one.

`_READ_POOL_MAX` itself is deliberately unchanged at 8. Retuning that constant
is #98585's subject; with a process ceiling above it and the headroom guard
in front of it, the value is no longer the binding constraint.

Refs #98573

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
joaomarcos
2026-08-30 14:06:35 -03:00
committed by Teknium
parent e8c41568ca
commit 0e9b57d310
2 changed files with 366 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_state.py
+201 -11
View File
@@ -386,6 +386,131 @@ _READ_OPEN_RETRY_SECONDS = 60.0
# the correct trade against a process-wide wedge the supervisor cannot see.
_READ_POOL_MAX = 8
# Hard ceiling on read-only connections ALIVE at once in this PROCESS, across
# every state.db it has open.
#
# _READ_POOL_MAX bounds one file. A multiplexed gateway serves N profiles from
# one process and each profile has its OWN state.db, so a per-file ceiling
# still lets the descriptor cost grow with the profile count — the same shape
# as the per-instance bug, one level out (#98573).
#
# Three profiles' worth. Past it, readers on the (N+1)th file degrade to their
# writer connection instead of opening descriptors, which is the same trade
# _READ_POOL_MAX makes and for the same reason: a slow read path is
# recoverable, a process-wide EMFILE is not.
_READ_POOL_PROCESS_MAX = 24
# Warn when one process accumulates more than this many SessionDB handles on a
# single file. Not a limit — writer connections cannot be rationed the way read
# connections can — a diagnostic for the duplicate-handle class of bug.
_HANDLES_PER_PATH_WARN = 4
# Descriptors kept in reserve for everything that is NOT this module: httpx
# sockets, terminal subprocess pipes, log files.
#
# The ceilings above bound Hermes's SQLite descriptors, which is only ever part
# of the fd table. The #98573 report is exactly that case: ~20 state.db
# descriptors were not the whole 256, they were the share that pushed httpx and
# terminal pipes over, and the EMFILE surfaced in tools/terminal_tool.py rather
# than here. So the read pool also yields when the PROCESS is close to its
# limit, whatever is consuming it.
_FD_HEADROOM_RESERVE = 64
# The fd count is a directory listing; cache it briefly so a burst of reads
# does not turn one syscall per query. Stale by at most this long, which can
# let through at most the ceiling's worth of opens — already bounded above.
_FD_USAGE_CACHE_SECONDS = 0.25
_process_read_permits = threading.BoundedSemaphore(_READ_POOL_PROCESS_MAX)
# Count of read opens refused because the process was low on descriptors. The
# only externally visible signal that the guard is firing; guarded by
# _read_budgets_lock.
_read_open_denied_fd_headroom = 0
_fd_usage_lock = threading.Lock()
_fd_usage_cache: "tuple[float, Optional[int]]" = (0.0, None)
def _open_fd_count() -> Optional[int]:
"""Descriptors open in THIS process, or None when it cannot be measured.
``/proc/self/fd`` on Linux, ``/dev/fd`` on macOS and the BSDs. Windows has
neither, and no RLIMIT_NOFILE to compare against, so the guard is inert
there — which is correct: the CRT limit is thousands of handles, not 256.
"""
for fd_dir in ("/proc/self/fd", "/dev/fd"):
try:
return len(os.listdir(fd_dir))
except OSError as exc:
if exc.errno in (errno.EMFILE, errno.ENFILE):
# The probe itself could not get a descriptor. That IS the
# answer: there is no headroom.
return -1
continue
return None
def _fd_soft_limit() -> Optional[int]:
"""The process's soft RLIMIT_NOFILE, or None when there is no usable one."""
try:
import resource
except ImportError:
return None
try:
soft, _hard = resource.getrlimit(resource.RLIMIT_NOFILE)
except (OSError, ValueError):
return None
if soft in (resource.RLIM_INFINITY, -1):
return None
return int(soft)
def _fd_headroom_ok() -> bool:
"""Whether the process can spare a descriptor for a new read connection.
Fails OPEN when the platform cannot be measured (Windows, no fd directory,
unlimited RLIMIT_NOFILE): an unmeasurable platform is not a tight one, and
refusing every read there would be a self-inflicted convoy. Fails CLOSED
only on evidence — a measured shortfall, or a probe that could not get a
descriptor of its own.
"""
soft = _fd_soft_limit()
if soft is None:
return True
global _fd_usage_cache
now = time.monotonic()
with _fd_usage_lock:
stamp, cached = _fd_usage_cache
fresh = cached is not None and (now - stamp) < _FD_USAGE_CACHE_SECONDS
if not fresh:
cached = _open_fd_count()
with _fd_usage_lock:
_fd_usage_cache = (now, cached)
if cached is None:
return True
if cached < 0:
return False
return (soft - cached) > _FD_HEADROOM_RESERVE
def _reclaim_idle_read_conn_anywhere() -> bool:
"""Close one idle read connection on ANY path in this process.
The process ceiling is shared across files, so the connection that has to
go to make room may belong to a different database entirely — a profile
that has been quiet for an hour should not hold descriptors the profile
being served right now needs.
"""
with _read_budgets_lock:
budgets = list(_read_budgets.values())
for budget in budgets:
if budget.reclaim_idle():
return True
return False
class _PathReadBudget:
"""The read-connection permits for ONE database file, shared process-wide.
@@ -417,27 +542,88 @@ class _PathReadBudget:
# peers' budget object; __del__ still runs close() and returns the
# permits.
self._members: "weakref.WeakSet[SessionDB]" = weakref.WeakSet()
self._duplicate_handles_warned = False
def register(self, db: "SessionDB") -> None:
with self._lock:
self._members.add(db)
handles = len(self._members)
warn = (
handles > _HANDLES_PER_PATH_WARN
and not self._duplicate_handles_warned
)
if warn:
self._duplicate_handles_warned = True
if warn:
# The read connections are capped; the WRITER connection each
# handle holds is not, and cannot be — a SessionDB without one
# cannot write. The only real bound on writers is not opening
# redundant handles in the first place (which is what
# GatewayRunner borrowing SessionStore's handle does, #98573), so
# the next duplicate should be visible before it becomes an
# incident rather than inferred from an lsof after one.
logger.warning(
"%d live SessionDB handles on %s in this process; each holds "
"its own writer connection (read connections are capped at %d "
"for the file). A long-lived process should share one handle "
"per path.",
handles,
db.db_path,
_READ_POOL_MAX,
)
def acquire(self, requester: "SessionDB") -> bool:
"""Take a permit, reclaiming an idle peer connection if need be."""
if self.permits.acquire(blocking=False):
"""Take a permit for a new read connection, or refuse.
Three gates, broadest first: the process's descriptor headroom, the
process-wide read ceiling, then this file's ceiling. Refusing means
the caller serves the read from the locked writer connection — slower,
never an error.
"""
if not _fd_headroom_ok():
global _read_open_denied_fd_headroom
with _read_budgets_lock:
_read_open_denied_fd_headroom += 1
return False
if not self._acquire_process_permit():
return False
if self._acquire_path_permit(requester):
return True
if not self._reclaim_idle(requester):
_process_read_permits.release()
return False
def release(self) -> None:
"""Return one connection's permits. Pairs with a successful acquire()."""
self.permits.release()
_process_read_permits.release()
def _acquire_process_permit(self) -> bool:
if _process_read_permits.acquire(blocking=False):
return True
if not _reclaim_idle_read_conn_anywhere():
return False
# Another thread may take the freed permit first; that is a legitimate
# loss, and the caller degrades to the writer lock rather than looping.
return _process_read_permits.acquire(blocking=False)
def _acquire_path_permit(self, requester: "SessionDB") -> bool:
if self.permits.acquire(blocking=False):
return True
if not self.reclaim_idle(exclude=requester):
return False
return self.permits.acquire(blocking=False)
def _reclaim_idle(self, requester: "SessionDB") -> bool:
"""Close one idle pooled connection held by a peer. True if one went."""
def reclaim_idle(self, exclude: "Optional[SessionDB]" = None) -> bool:
"""Close one idle pooled connection held by a member. True if one went.
Closing it runs release(), which returns both the path permit and the
process permit, so this is the single reclaim primitive both ceilings
use.
"""
with self._lock:
peers = [db for db in self._members if db is not requester]
for peer in peers:
if peer._evict_one_idle_read_conn():
members = [db for db in self._members if db is not exclude]
for member in members:
if member._evict_one_idle_read_conn():
return True
return False
@@ -5029,7 +5215,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
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()
self._read_budget.release()
return None
except BaseException:
# Anything else (a non-sqlite3 extension-load failure, MemoryError,
@@ -5038,7 +5224,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
# permanently shrinks the read path by one slot for the life of the
# process.
self._discard_partial_read_conn(conn)
self._read_permits.release()
self._read_budget.release()
raise
return conn
@@ -5097,7 +5283,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
except Exception as exc:
logger.warning("read-conn close failed for %s: %s", self.db_path, exc)
finally:
self._read_permits.release()
self._read_budget.release()
def _checkout_read_conn(self) -> Optional[sqlite3.Connection]:
"""Borrow a read connection from the pool, opening one on a miss.
@@ -13769,6 +13955,10 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
# Export and cleanup
# =========================================================================
def is_explicit_fork_child(self, session_id: str) -> bool:
session = self.get_session(session_id)
return bool(session and self._is_explicit_fork_child_row(session))
def _is_explicit_fork_child_row(self, session: Dict[str, Any]) -> bool:
"""True when ``session`` is a branch, delegate, or tool child of its parent.
tests/test_session_db_read_conn_pool.py
+165
View File
@@ -35,6 +35,7 @@ holds POSIX locks on that inode, so raw descriptor counts lag the real
connection count and make such assertions flaky.
"""
import queue
import threading
import pytest
@@ -481,3 +482,167 @@ def test_idle_permits_are_reclaimed_from_a_peer_instance(db):
second._close_read_conn(conn)
finally:
second.close()
# ── The process ceiling, and the descriptors this module does not own ──
#
# _READ_POOL_MAX bounds ONE file. A multiplexed gateway serves N profiles from
# one process and each has its own state.db, so a per-file ceiling still lets
# the cost grow with the profile count -- the per-instance bug one level out.
# And Hermes's SQLite descriptors are only ever a share of the fd table: the
# #98573 report is a process where ~20 state.db handles were the share that
# pushed httpx sockets and terminal subprocess pipes past 256, and the EMFILE
# surfaced in tools/terminal_tool.py, not here.
@pytest.mark.requires_wal
def test_peak_is_bounded_across_many_database_files(tmp_path):
"""Read connections must be capped for the PROCESS, not just per file."""
import hermes_state
from hermes_state import SessionDB, _READ_POOL_MAX, _READ_POOL_PROCESS_MAX
n_files = (_READ_POOL_PROCESS_MAX // _READ_POOL_MAX) + 2
dbs = []
try:
for i in range(n_files):
d = SessionDB(db_path=tmp_path / f"p{i}" / "state.db")
d.create_session(session_id="s1", source="cli", model="m")
d.append_message("s1", role="user", content="graphiti")
dbs.append(d)
# Fill every pool, one file at a time.
held = []
for d in dbs:
for _ in range(_READ_POOL_MAX):
conn = d._checkout_read_conn()
if conn is None:
break
held.append((d, conn))
assert len(held) <= _READ_POOL_PROCESS_MAX, (
f"{len(held)} read connections open across {n_files} files; the "
f"process ceiling is {_READ_POOL_PROCESS_MAX}. Per-file bounds "
f"alone let the descriptor cost grow with the profile count."
)
assert len(held) > _READ_POOL_MAX, (
"the process ceiling must be wider than one file's, or a "
"multiplexed gateway serves every profile from the writer lock"
)
total_live = sum(_live_count(d.db_path) for d in dbs)
assert total_live <= _READ_POOL_PROCESS_MAX + n_files, (
f"{total_live} live connections process-wide (ceiling "
f"{_READ_POOL_PROCESS_MAX} read + {n_files} writers)"
)
for d, conn in held:
d._close_read_conn(conn)
finally:
for d in dbs:
d.close()
assert hermes_state._process_read_permits.acquire(blocking=False), (
"close() stranded a process permit"
)
hermes_state._process_read_permits.release()
@pytest.mark.requires_wal
def test_idle_connections_are_reclaimed_across_database_files(tmp_path):
"""A quiet profile's idle connections must not starve the busy one."""
from hermes_state import SessionDB, _READ_POOL_MAX, _READ_POOL_PROCESS_MAX
quiet = []
try:
# Saturate the process ceiling with IDLE connections on other files.
for i in range(_READ_POOL_PROCESS_MAX // _READ_POOL_MAX):
d = SessionDB(db_path=tmp_path / f"q{i}" / "state.db")
quiet.append(d)
conns = [d._checkout_read_conn() for _ in range(_READ_POOL_MAX)]
assert all(c is not None for c in conns)
for c in conns:
d._read_pool.put_nowait(c)
busy = SessionDB(db_path=tmp_path / "busy" / "state.db")
quiet.append(busy)
conn = busy._checkout_read_conn()
assert conn is not None, (
"idle connections on quiet profiles starved the profile being "
"served; its reads silently fell back to the writer lock"
)
busy._close_read_conn(conn)
finally:
for d in quiet:
d.close()
@pytest.mark.requires_wal
def test_no_read_connection_is_opened_without_descriptor_headroom(db, monkeypatch):
"""Low on fds, the read path yields to the rest of the process.
The descriptors this module rations are shared with httpx sockets and
subprocess pipes, and EMFILE lands on whoever asks next -- which in the
report was terminal_tool, not SQLite.
"""
import hermes_state
# Drain the pool so the next read must OPEN rather than reuse.
while True:
try:
db._close_read_conn(db._read_pool.get_nowait())
except queue.Empty:
break
monkeypatch.setattr(hermes_state, "_fd_soft_limit", lambda: 256)
monkeypatch.setattr(hermes_state, "_open_fd_count", lambda: 250)
monkeypatch.setattr(hermes_state, "_fd_usage_cache", (0.0, None))
assert db._get_read_conn() is None, "a read connection was opened with 6 fds left"
# The read still has to work -- degradation, not failure.
assert db.get_session("s1") is not None
assert hermes_state._read_open_denied_fd_headroom > 0, (
"the guard fired without leaving a trace to diagnose it from"
)
monkeypatch.setattr(hermes_state, "_open_fd_count", lambda: 10)
monkeypatch.setattr(hermes_state, "_fd_usage_cache", (0.0, None))
conn = db._get_read_conn()
assert conn is not None, "headroom returned but the read path stayed degraded"
db._close_read_conn(conn)
def test_fd_headroom_guard_fails_open_where_it_cannot_measure(monkeypatch):
"""No RLIMIT_NOFILE (Windows) means unmeasurable, not tight."""
import hermes_state
monkeypatch.setattr(hermes_state, "_fd_soft_limit", lambda: None)
assert hermes_state._fd_headroom_ok() is True
# A probe that could not get a descriptor of its own is evidence, not
# absence of evidence.
monkeypatch.setattr(hermes_state, "_fd_soft_limit", lambda: 256)
monkeypatch.setattr(hermes_state, "_open_fd_count", lambda: -1)
monkeypatch.setattr(hermes_state, "_fd_usage_cache", (0.0, None))
assert hermes_state._fd_headroom_ok() is False
@pytest.mark.requires_wal
def test_duplicate_handles_on_one_path_are_reported(db, caplog):
"""Writer connections cannot be capped, so duplicates must be visible."""
import logging
from hermes_state import SessionDB, _HANDLES_PER_PATH_WARN
extra = []
try:
with caplog.at_level(logging.WARNING, logger="hermes_state"):
for _ in range(_HANDLES_PER_PATH_WARN):
extra.append(SessionDB(db_path=db.db_path))
assert any(
"live SessionDB handles on" in r.getMessage()
for r in caplog.records
), (
f"{_HANDLES_PER_PATH_WARN + 1} handles on one file went unreported; "
f"each holds a writer connection nothing bounds"
)
finally:
for d in extra:
d.close()