refactor(state): extract read-connection budgeting to hermes_state_readpool

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/<pid>/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

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/<pid>/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

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(readpool, "_fd_soft_limit", lambda: 256)
-    monkeypatch.setattr(hermes_state, "_open_fd_count", lambda: 250)
+    monkeypatch.setattr(readpool, "_open_fd_count", lambda: 250)
-    monkeypatch.setattr(hermes_state, "_fd_usage_cache", (0.0, None))
+    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(readpool, "_open_fd_count", lambda: 10)
-    monkeypatch.setattr(hermes_state, "_fd_usage_cache", (0.0, None))
+    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)
+    monkeypatch.setattr(readpool, "_fd_soft_limit", lambda: None)
-    assert hermes_state._fd_headroom_ok() is True
+    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(readpool, "_fd_soft_limit", lambda: 256)
-    monkeypatch.setattr(hermes_state, "_open_fd_count", lambda: -1)
+    monkeypatch.setattr(readpool, "_open_fd_count", lambda: -1)
-    monkeypatch.setattr(hermes_state, "_fd_usage_cache", (0.0, None))
+    monkeypatch.setattr(readpool, "_fd_usage_cache", (0.0, None))
-    assert hermes_state._fd_headroom_ok() is False
+    assert readpool._fd_headroom_ok() is False
 
 
 @pytest.mark.requires_wal