diff --git a/hermes_state.py b/hermes_state.py index 728c609299..0e89b7dc49 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -7,7 +7,6 @@ splits sessions via parent_session_id chains; sessions are source-tagged import asyncio import atexit -import errno import hashlib import json import logging @@ -20,7 +19,6 @@ import sys import threading import time import uuid -import weakref from collections import deque from contextlib import contextmanager from pathlib import Path @@ -60,6 +58,10 @@ from hermes_state_guard import ( # noqa: F401 (re-exported; tests patch hermes _has_pytest_ancestor, _in_test_context, _is_production_state_db, _process_looks_like_pytest, _real_platform_state_root, _running_under_pytest, _set_last_init_error, get_last_init_error, ) +from hermes_state_readpool import ( # noqa: F401 (re-exported; tests import from hermes_state) + _HANDLES_PER_PATH_WARN, _READ_POOL_MAX, _READ_POOL_PROCESS_MAX, _PathReadBudget, + _proc_fd_targets, _process_read_permits, _read_budget_for, +) from hermes_state_portability import SessionPortabilityMixin from hermes_state_telegram import SessionTelegramTopicsMixin from hermes_state_schema import SessionSchemaMixin @@ -382,213 +384,6 @@ def _is_transient_read_only_ioerr(exc: sqlite3.OperationalError, *, attempt: int return attempt < _READ_ONLY_IOERR_RETRY_ATTEMPTS and _DISK_IO_ERROR_MARKER in str(exc).lower() -# Ceiling on read-only connections ALIVE at once against one database FILE -# (idle pooled + checked out, summed over every SessionDB on that file). One -# constant for both the pool maxsize and the permit count: a LifoQueue only caps -# how many are *returned*; with open-on-miss, N readers hitting an empty pool -# all open and peak at N, and EMFILE is a peak-instant condition. So a -# connection holds a permit for its whole lifetime (_get_read_conn -> -# _close_read_conn); once permits are gone reads degrade to the locked writer -# connection — slower, but not a process-wide wedge the supervisor can't see. -_READ_POOL_MAX = 8 - -# Ceiling on read-only connections ALIVE in this PROCESS across every state.db -# (a multiplexed gateway opens one per profile, so a per-file cap still scales -# with profile count). Three profiles' worth; past it readers degrade to the -# writer connection for the same reason as _READ_POOL_MAX. -_READ_POOL_PROCESS_MAX = 24 - -# Warn past this many SessionDB handles on one file in one process. Diagnostic -# only: writer connections cannot be rationed the way read connections can. -_HANDLES_PER_PATH_WARN = 4 - -# Descriptors kept in reserve for everything that is NOT this module (httpx -# sockets, terminal pipes, log files): SQLite's share is only part of the fd -# table, and the EMFILE it pushes over surfaces elsewhere (terminal_tool). -_FD_HEADROOM_RESERVE = 64 - -# The fd count is a directory listing; cache it briefly so a read burst isn't a -# syscall per query. Staleness lets through at most the ceiling's worth of opens. -_FD_USAGE_CACHE_SECONDS = 0.25 - -_process_read_permits = threading.BoundedSemaphore(_READ_POOL_PROCESS_MAX) - -# Read opens refused for low descriptor headroom — the only visible signal the -# guard fires. 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 _proc_fd_targets(pid: int) -> Iterator[str]: - """readlink() of every entry in /proc//fd (unreadable links skipped). - Raises OSError when the fd directory itself cannot be listed.""" - fd_dir = f"/proc/{pid}/fd" - for fd in os.listdir(fd_dir): - try: - yield os.readlink(f"{fd_dir}/{fd}") - except OSError: - continue - - -def _open_fd_count() -> Optional[int]: - """Open descriptors in THIS process; None when unmeasurable (Windows: no fd - dir and no RLIMIT_NOFILE, correctly inert — its limit is thousands); -1 when - the probe itself hit EMFILE/ENFILE (that IS the answer: no headroom).""" - 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): - return -1 - 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: - """Can the process spare a descriptor for a new read connection? - Fails OPEN when unmeasurable (refusing every read there would be a - self-inflicted convoy); fails CLOSED only on evidence (measured shortfall, - or a probe that couldn't get a descriptor itself).""" - 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 - return cached >= 0 and (soft - cached) > _FD_HEADROOM_RESERVE - - -def _reclaim_idle_read_conn_anywhere() -> bool: - """Close one idle read connection on ANY path: the process ceiling is shared - across files, so a quiet profile must not hold descriptors a busy one needs.""" - with _read_budgets_lock: - budgets = list(_read_budgets.values()) - return any(budget.reclaim_idle() for budget in budgets) - - -class _PathReadBudget: - """Read-connection permits for ONE database file, shared process-wide: - per-instance semaphores let N SessionDBs on one file peak at N x (1 + MAX) - and walk into EMFILE. An idle pooled connection keeps its permit, so a - permit miss first reclaims an IDLE connection from a peer on the same path - (idle descriptors are transferable, in-use ones are not).""" - - def __init__(self) -> None: - self.permits = threading.BoundedSemaphore(_READ_POOL_MAX) - self._lock = threading.Lock() - # Weak: a SessionDB dropped without close() must not pin peers' budget. - 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: - # Writer connections cannot be capped (a SessionDB without one cannot - # write); the only bound is not opening redundant handles. Make the - # next duplicate visible before it becomes an incident. - 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 for a new read connection, or refuse (caller then reads - via the locked writer connection — slower, never an error). Gates, - broadest first: fd headroom, process-wide ceiling, this file's ceiling.""" - 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 - _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: - # Another thread may take a 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) or ( - _reclaim_idle_read_conn_anywhere() and _process_read_permits.acquire(blocking=False) - ) - - def _acquire_path_permit(self, requester: "SessionDB") -> bool: - return self.permits.acquire(blocking=False) or ( - self.reclaim_idle(exclude=requester) and self.permits.acquire(blocking=False) - ) - - def reclaim_idle(self, exclude: "Optional[SessionDB]" = None) -> bool: - """Close one idle pooled connection held by a member; True if one went. - Its release() returns both permits, so both ceilings reclaim through here.""" - with self._lock: - members = [db for db in self._members if db is not exclude] - return any(member._evict_one_idle_read_conn() for member in members) - - -# canonical db path -> permits for that file. Weak values: the budget lives as -# long as some SessionDB on the path holds it, so tmp_path churn can't grow this. -_read_budgets: "weakref.WeakValueDictionary[str, _PathReadBudget]" = (weakref.WeakValueDictionary()) -_read_budgets_lock = threading.Lock() - - -def _read_budget_key(db_path) -> str: - """Canonicalise a db path so two spellings share one budget.""" - try: - return str(Path(db_path).resolve()) - except OSError: - return str(db_path) - - -def _read_budget_for(db_path) -> _PathReadBudget: - key = _read_budget_key(db_path) - with _read_budgets_lock: - budget = _read_budgets.get(key) - if budget is None: - budget = _PathReadBudget() - _read_budgets[key] = budget - return budget - - # Import-time snapshot so _default_db_path() can detect a deliberately # re-pointed DEFAULT_DB_PATH (tests monkeypatch the constant directly). _IMPORT_DEFAULT_DB_PATH = DEFAULT_DB_PATH diff --git a/hermes_state_readpool.py b/hermes_state_readpool.py new file mode 100644 index 0000000000..b74158fe29 --- /dev/null +++ b/hermes_state_readpool.py @@ -0,0 +1,225 @@ +"""Read-connection budgeting for SessionDB's WAL read path: per-file and +process-wide permit ceilings plus a descriptor-headroom gate, so N handles on +one state.db cannot walk the process into EMFILE. Idle pooled connections keep +their permit and are reclaimable across peers on the same path.""" + +import errno +import logging +import os +import threading +import time +import weakref +from pathlib import Path +from typing import TYPE_CHECKING, Iterator, Optional + +if TYPE_CHECKING: # pragma: no cover + from hermes_state import SessionDB + +# caplog tests pin the "hermes_state" logger name. +logger = logging.getLogger("hermes_state") + +# Ceiling on read-only connections ALIVE at once against one database FILE +# (idle pooled + checked out, summed over every SessionDB on that file). One +# constant for both the pool maxsize and the permit count: a LifoQueue only caps +# how many are *returned*; with open-on-miss, N readers hitting an empty pool +# all open and peak at N, and EMFILE is a peak-instant condition. So a +# connection holds a permit for its whole lifetime (_get_read_conn -> +# _close_read_conn); once permits are gone reads degrade to the locked writer +# connection — slower, but not a process-wide wedge the supervisor can't see. +_READ_POOL_MAX = 8 + +# Ceiling on read-only connections ALIVE in this PROCESS across every state.db +# (a multiplexed gateway opens one per profile, so a per-file cap still scales +# with profile count). Three profiles' worth; past it readers degrade to the +# writer connection for the same reason as _READ_POOL_MAX. +_READ_POOL_PROCESS_MAX = 24 + +# Warn past this many SessionDB handles on one file in one process. Diagnostic +# only: writer connections cannot be rationed the way read connections can. +_HANDLES_PER_PATH_WARN = 4 + +# Descriptors kept in reserve for everything that is NOT this module (httpx +# sockets, terminal pipes, log files): SQLite's share is only part of the fd +# table, and the EMFILE it pushes over surfaces elsewhere (terminal_tool). +_FD_HEADROOM_RESERVE = 64 + +# The fd count is a directory listing; cache it briefly so a read burst isn't a +# syscall per query. Staleness lets through at most the ceiling's worth of opens. +_FD_USAGE_CACHE_SECONDS = 0.25 + +_process_read_permits = threading.BoundedSemaphore(_READ_POOL_PROCESS_MAX) + +# Read opens refused for low descriptor headroom — the only visible signal the +# guard fires. 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 _proc_fd_targets(pid: int) -> Iterator[str]: + """readlink() of every entry in /proc//fd (unreadable links skipped). + Raises OSError when the fd directory itself cannot be listed.""" + fd_dir = f"/proc/{pid}/fd" + for fd in os.listdir(fd_dir): + try: + yield os.readlink(f"{fd_dir}/{fd}") + except OSError: + continue + + +def _open_fd_count() -> Optional[int]: + """Open descriptors in THIS process; None when unmeasurable (Windows: no fd + dir and no RLIMIT_NOFILE, correctly inert — its limit is thousands); -1 when + the probe itself hit EMFILE/ENFILE (that IS the answer: no headroom).""" + 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): + return -1 + 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: + """Can the process spare a descriptor for a new read connection? + Fails OPEN when unmeasurable (refusing every read there would be a + self-inflicted convoy); fails CLOSED only on evidence (measured shortfall, + or a probe that couldn't get a descriptor itself).""" + 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 + return cached >= 0 and (soft - cached) > _FD_HEADROOM_RESERVE + + +def _reclaim_idle_read_conn_anywhere() -> bool: + """Close one idle read connection on ANY path: the process ceiling is shared + across files, so a quiet profile must not hold descriptors a busy one needs.""" + with _read_budgets_lock: + budgets = list(_read_budgets.values()) + return any(budget.reclaim_idle() for budget in budgets) + + +class _PathReadBudget: + """Read-connection permits for ONE database file, shared process-wide: + per-instance semaphores let N SessionDBs on one file peak at N x (1 + MAX) + and walk into EMFILE. An idle pooled connection keeps its permit, so a + permit miss first reclaims an IDLE connection from a peer on the same path + (idle descriptors are transferable, in-use ones are not).""" + + def __init__(self) -> None: + self.permits = threading.BoundedSemaphore(_READ_POOL_MAX) + self._lock = threading.Lock() + # Weak: a SessionDB dropped without close() must not pin peers' budget. + 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: + # Writer connections cannot be capped (a SessionDB without one cannot + # write); the only bound is not opening redundant handles. Make the + # next duplicate visible before it becomes an incident. + 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 for a new read connection, or refuse (caller then reads + via the locked writer connection — slower, never an error). Gates, + broadest first: fd headroom, process-wide ceiling, this file's ceiling.""" + 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 + _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: + # Another thread may take a 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) or ( + _reclaim_idle_read_conn_anywhere() and _process_read_permits.acquire(blocking=False) + ) + + def _acquire_path_permit(self, requester: "SessionDB") -> bool: + return self.permits.acquire(blocking=False) or ( + self.reclaim_idle(exclude=requester) and self.permits.acquire(blocking=False) + ) + + def reclaim_idle(self, exclude: "Optional[SessionDB]" = None) -> bool: + """Close one idle pooled connection held by a member; True if one went. + Its release() returns both permits, so both ceilings reclaim through here.""" + with self._lock: + members = [db for db in self._members if db is not exclude] + return any(member._evict_one_idle_read_conn() for member in members) + + +# canonical db path -> permits for that file. Weak values: the budget lives as +# long as some SessionDB on the path holds it, so tmp_path churn can't grow this. +_read_budgets: "weakref.WeakValueDictionary[str, _PathReadBudget]" = (weakref.WeakValueDictionary()) +_read_budgets_lock = threading.Lock() + + +def _read_budget_key(db_path) -> str: + """Canonicalise a db path so two spellings share one budget.""" + try: + return str(Path(db_path).resolve()) + except OSError: + return str(db_path) + + +def _read_budget_for(db_path) -> _PathReadBudget: + key = _read_budget_key(db_path) + with _read_budgets_lock: + budget = _read_budgets.get(key) + if budget is None: + budget = _PathReadBudget() + _read_budgets[key] = budget + return budget diff --git a/tests/test_session_db_read_conn_pool.py b/tests/test_session_db_read_conn_pool.py index c25f1e2c04..3afb88f6e1 100644 --- a/tests/test_session_db_read_conn_pool.py +++ b/tests/test_session_db_read_conn_pool.py @@ -582,7 +582,7 @@ def test_no_read_connection_is_opened_without_descriptor_headroom(db, monkeypatc subprocess pipes, and EMFILE lands on whoever asks next -- which in the report was terminal_tool, not SQLite. """ - import hermes_state + import hermes_state_readpool as readpool # Drain the pool so the next read must OPEN rather than reuse. while True: @@ -591,19 +591,19 @@ def test_no_read_connection_is_opened_without_descriptor_headroom(db, monkeypatc 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)) + monkeypatch.setattr(readpool, "_fd_soft_limit", lambda: 256) + monkeypatch.setattr(readpool, "_open_fd_count", lambda: 250) + monkeypatch.setattr(readpool, "_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, ( + assert readpool._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)) + monkeypatch.setattr(readpool, "_open_fd_count", lambda: 10) + monkeypatch.setattr(readpool, "_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) @@ -611,17 +611,17 @@ def test_no_read_connection_is_opened_without_descriptor_headroom(db, monkeypatc def test_fd_headroom_guard_fails_open_where_it_cannot_measure(monkeypatch): """No RLIMIT_NOFILE (Windows) means unmeasurable, not tight.""" - import hermes_state + import hermes_state_readpool as readpool - monkeypatch.setattr(hermes_state, "_fd_soft_limit", lambda: None) - assert hermes_state._fd_headroom_ok() is True + monkeypatch.setattr(readpool, "_fd_soft_limit", lambda: None) + assert readpool._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 + monkeypatch.setattr(readpool, "_fd_soft_limit", lambda: 256) + monkeypatch.setattr(readpool, "_open_fd_count", lambda: -1) + monkeypatch.setattr(readpool, "_fd_usage_cache", (0.0, None)) + assert readpool._fd_headroom_ok() is False @pytest.mark.requires_wal