From 576accd92bdd5a1cced03e5bc7110f1ea81ff24c Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Sat, 12 Sep 2026 19:59:01 -0700 Subject: [PATCH] refactor(sqlite): one open_db/transaction layer for every small store; plugin DBs use the WAL fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Twelve modules each carried their own sqlite3.connect + PRAGMA + `with conn:` stack. The #69567 fd-leak fix (a `with conn:` commits but never closes, so each call leaked a connection and its WAL/SHM fds until GC) was pasted as code plus docstring into six of them and hosted_room_policy_checkpoint never received it; plugins/plugin_storage.plugin_db was the only production caller issuing a raw `PRAGMA journal_mode=WAL`, bypassing the network-FS fallback, the WAL-reset-bug gate and the never-live-downgrade invariant that hermes_state_wal.apply_wal_with_fallback carries. hermes_cli/sqlite_util.py (already home to add_column_if_missing/write_txn, imported by cron, gateway and hermes_cli alike) gains `open_db(path, *, db_label, busy_timeout_ms, wal, foreign_keys, synchronous_full, row_factory, check_same_thread, wal_lock_retries, initialize)` and `transaction(conn, immediate=)`; cron/ledger.py is deleted and hosted_rooms_common's open_sqlite/connect/transaction become 1-3 line forwarders. Migrated: agent/verification_evidence, cron/{executions,incidents,notepad, delivery_queue}, gateway/{delivery_ledger,hosted_room_policy_checkpoint, hosted_rooms_common (-> hosted_rooms, hosted_room_driver)}, hermes_cli/ projects_db, tools/async_delegation, plugins/plugin_storage. Behavior changes (each module keeps its effective PRAGMA set otherwise): - hosted_room_policy_checkpoint: connection now closed after every use and on init failure (was leaked per call), busy_timeout PRAGMA set explicitly. - projects_db: gains busy_timeout=5000 (was the sqlite3 default 5 s connect timeout with no PRAGMA); explicit and observable. - delivery_ledger / async_delegation: busy_timeout PRAGMA now mirrors the 10 s connect timeout they already had. - plugin_storage.plugin_db: WAL through apply_wal_with_fallback (DELETE on network filesystems / WAL-reset-vulnerable builds instead of raw WAL); busy_timeout=5000. - cron/incidents._redact_error: redact_sensitive_text(force=True) — the error text is persisted to disk. - delivery_ledger's private duplicate-column guard and the unguarded `ALTER TABLE ADD COLUMN` sites (shared_metrics, api_server_run_idempotency, holographic store, kanban model_override) go through add_column_if_missing. - hermes_state.py::_scrub_surrogates: dead byte-copy of hermes_state_messages._scrub_surrogates (0 callers) deleted. --- agent/verification_evidence.py | 38 +---- cron/delivery_queue.py | 98 ++++++------- cron/executions.py | 12 +- cron/incidents.py | 12 +- cron/ledger.py | 47 ------ cron/notepad.py | 10 +- gateway/delivery_ledger.py | 43 ++---- gateway/hosted_room_policy_checkpoint.py | 25 ++-- gateway/hosted_rooms_common.py | 62 +++----- .../platforms/api_server_run_idempotency.py | 4 +- hermes_cli/kanban_db_connect.py | 5 +- hermes_cli/observability/shared_metrics.py | 6 +- hermes_cli/projects_db.py | 31 ++-- hermes_cli/sqlite_util.py | 80 ++++++++++- hermes_state.py | 6 - plugins/memory/holographic/store.py | 3 +- plugins/plugin_storage.py | 10 +- tests/cron/test_upgrade_module_skew.py | 2 +- .../hermes_cli/test_sqlite_util_canonical.py | 136 ++++++++++++++++++ tools/async_delegation.py | 37 ++--- 20 files changed, 365 insertions(+), 302 deletions(-) delete mode 100644 cron/ledger.py create mode 100644 tests/hermes_cli/test_sqlite_util_canonical.py diff --git a/agent/verification_evidence.py b/agent/verification_evidence.py index 69e7f2c1aa..b376c29017 100644 --- a/agent/verification_evidence.py +++ b/agent/verification_evidence.py @@ -10,11 +10,10 @@ import shlex import sqlite3 import tempfile import threading -from contextlib import contextmanager from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path -from typing import Any, Iterator, Optional +from typing import Any, Optional from hermes_constants import get_hermes_home @@ -120,40 +119,15 @@ def _ledger_enabled() -> bool: def _connect() -> sqlite3.Connection: - from hermes_state_wal import apply_wal_with_fallback + from hermes_cli.sqlite_util import open_db - path = _db_path() - path.parent.mkdir(parents=True, exist_ok=True) - conn = sqlite3.connect(path) - conn.row_factory = sqlite3.Row - try: - apply_wal_with_fallback(conn, db_label="verification_evidence.db") - conn.execute("PRAGMA busy_timeout=5000") - _ensure_schema(conn) - except Exception: - # A PRAGMA/DDL failure after connect() must not leak the open connection. - conn.close() - raise - return conn + return open_db(_db_path(), db_label="verification_evidence.db", initialize=_ensure_schema) -@contextmanager -def _transaction() -> Iterator[sqlite3.Connection]: - """Open a connection, commit/rollback on exit, and ALWAYS close it. +def _transaction(): + from hermes_cli.sqlite_util import transaction - ``sqlite3.Connection`` as a context manager only commits/rolls back; without - the close, each call leaks a connection (and WAL/SHM fds) until GC runs. - - Using ``with _connect()`` alone therefore leaks a connection — and its WAL/SHM file descriptors — on - every call, deferring the close to the garbage collector, which over a long-running process can exhaust - ``RLIMIT_NOFILE`` (the cron-ledger sibling of this bug was #69567 / PR #69594). - """ - conn = _connect() - try: - with conn: - yield conn - finally: - conn.close() + return transaction(_connect()) def _ensure_schema(conn: sqlite3.Connection) -> None: diff --git a/cron/delivery_queue.py b/cron/delivery_queue.py index 612b626047..1ddb6e6cad 100644 --- a/cron/delivery_queue.py +++ b/cron/delivery_queue.py @@ -22,7 +22,7 @@ from typing import Any, Callable, Iterator, Optional from agent.redact import redact_sensitive_text from cron.executions import _owner_is_live, _process_start_time -from hermes_cli.sqlite_util import add_column_if_missing +from hermes_cli.sqlite_util import add_column_if_missing, open_db, transaction from hermes_constants import get_hermes_home from hermes_time import now as _hermes_now @@ -77,58 +77,54 @@ def _path() -> Path: return DELIVERY_DB or (get_hermes_home().resolve() / "cron" / "deliveries.db") +def _initialize_schema(conn: sqlite3.Connection) -> None: + conn.execute( + """CREATE TABLE IF NOT EXISTS deliveries ( + execution_id TEXT PRIMARY KEY, + job_json TEXT NOT NULL, + content TEXT NOT NULL, + for_failure INTEGER NOT NULL DEFAULT 0, + status TEXT NOT NULL CHECK(status IN + ('pending','delivering','delivered','failed','unknown')), + owner_process_id TEXT, + owner_pid INTEGER, + owner_started_at INTEGER, + created_at TEXT NOT NULL, + finished_at TEXT, + error TEXT + )""" + ) + conn.execute( + """CREATE TABLE IF NOT EXISTS delivery_tombstones ( + execution_id TEXT PRIMARY KEY, + terminal_status TEXT NOT NULL CHECK(terminal_status IN + ('delivered','failed','unknown')), + finished_at TEXT + )""" + ) + add_column_if_missing( + conn, "deliveries", "for_failure", + "for_failure INTEGER NOT NULL DEFAULT 0", + ) + + +def _connect() -> sqlite3.Connection: + path = _path() + conn = open_db(path, db_label="cron/deliveries.db", synchronous_full=True, initialize=_initialize_schema) + try: + path.chmod(0o600) + except OSError: + pass + return conn + + @contextmanager def _transaction() -> Iterator[sqlite3.Connection]: - with _lock: - path = _path() - path.parent.mkdir(parents=True, exist_ok=True) - conn = sqlite3.connect(path, timeout=5) - try: - path.chmod(0o600) - except OSError: - pass - conn.row_factory = sqlite3.Row - try: - from hermes_state_wal import apply_wal_with_fallback - - conn.execute("PRAGMA busy_timeout=5000") - apply_wal_with_fallback(conn, db_label="cron/deliveries.db") - conn.execute("PRAGMA synchronous=FULL") - conn.execute( - """CREATE TABLE IF NOT EXISTS deliveries ( - execution_id TEXT PRIMARY KEY, - job_json TEXT NOT NULL, - content TEXT NOT NULL, - for_failure INTEGER NOT NULL DEFAULT 0, - status TEXT NOT NULL CHECK(status IN - ('pending','delivering','delivered','failed','unknown')), - owner_process_id TEXT, - owner_pid INTEGER, - owner_started_at INTEGER, - created_at TEXT NOT NULL, - finished_at TEXT, - error TEXT - )""" - ) - conn.execute( - """CREATE TABLE IF NOT EXISTS delivery_tombstones ( - execution_id TEXT PRIMARY KEY, - terminal_status TEXT NOT NULL CHECK(terminal_status IN - ('delivered','failed','unknown')), - finished_at TEXT - )""" - ) - add_column_if_missing( - conn, "deliveries", "for_failure", - "for_failure INTEGER NOT NULL DEFAULT 0", - ) - # Pruning is done explicitly by the paths that create terminal - # rows (_finish / recover_abandoned / _terminalize_wait_timeout); - # read-only polls must not pay for a full-table UPDATE + COUNT. - with conn: - yield conn - finally: - conn.close() + # Pruning is done explicitly by the paths that create terminal + # rows (_finish / recover_abandoned / _terminalize_wait_timeout); + # read-only polls must not pay for a full-table UPDATE + COUNT. + with _lock, transaction(_connect()) as conn: + yield conn def enqueue( diff --git a/cron/executions.py b/cron/executions.py index e10a36c733..292b8b22df 100644 --- a/cron/executions.py +++ b/cron/executions.py @@ -16,7 +16,8 @@ from contextlib import contextmanager from pathlib import Path from typing import Any, Dict, Iterator, List, Optional -from cron.ledger import ledger_transaction, open_ledger, prepare_ledger +from cron.jobs import _ensure_cron_dir +from hermes_cli.sqlite_util import add_column_if_missing, open_db, transaction from hermes_constants import get_hermes_home from hermes_time import now as _hermes_now @@ -34,11 +35,12 @@ _PROCESS_ID = uuid.uuid4().hex # --- executions ledger -------------------------------------------------------------------------- def _connect() -> sqlite3.Connection: - return open_ledger(EXECUTIONS_FILE or (get_hermes_home().resolve() / "cron" / "executions.db")) + path = EXECUTIONS_FILE or (get_hermes_home().resolve() / "cron" / "executions.db") + _ensure_cron_dir(path.parent) + return open_db(path, db_label="cron/executions.db", synchronous_full=True, initialize=_initialize_schema) def _initialize_schema(conn: sqlite3.Connection) -> None: - prepare_ledger(conn, db_label="cron/executions.db") conn.execute( """CREATE TABLE IF NOT EXISTS executions ( id TEXT PRIMARY KEY, @@ -57,8 +59,6 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: error TEXT )""" ) - from hermes_cli.sqlite_util import add_column_if_missing - add_column_if_missing( conn, "executions", "handoff_pending", "handoff_pending INTEGER NOT NULL DEFAULT 0", @@ -84,7 +84,7 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: @contextmanager def _transaction() -> Iterator[sqlite3.Connection]: - with ledger_transaction(_lock, _connect, _initialize_schema) as conn: + with _lock, transaction(_connect()) as conn: yield conn diff --git a/cron/incidents.py b/cron/incidents.py index 02a9ae4396..44f9e0c1dc 100644 --- a/cron/incidents.py +++ b/cron/incidents.py @@ -19,7 +19,8 @@ from pathlib import Path from typing import Any, Dict, Iterator, List, Optional from cron import executions as _executions -from cron.ledger import ledger_transaction, open_ledger, prepare_ledger +from cron.jobs import _ensure_cron_dir +from hermes_cli.sqlite_util import open_db, transaction from hermes_constants import get_hermes_home from hermes_time import now as _hermes_now @@ -53,11 +54,12 @@ def _db_path() -> Path: def _connect() -> sqlite3.Connection: - return open_ledger(_db_path()) + path = _db_path() + _ensure_cron_dir(path.parent) + return open_db(path, db_label="cron/executions.db", synchronous_full=True, initialize=_initialize_schema) def _initialize_schema(conn: sqlite3.Connection) -> None: - prepare_ledger(conn, db_label="cron/executions.db") conn.execute( """CREATE TABLE IF NOT EXISTS cron_incidents ( id TEXT PRIMARY KEY, @@ -85,7 +87,7 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: @contextmanager def _transaction() -> Iterator[sqlite3.Connection]: - with ledger_transaction(_lock, _connect, _initialize_schema) as conn: + with _lock, transaction(_connect()) as conn: yield conn @@ -100,7 +102,7 @@ def _redact_error(error: str) -> str: try: from agent.redact import redact_sensitive_text - text = redact_sensitive_text(text) + text = redact_sensitive_text(text, force=True) # persisted to disk: always scrub except Exception: pass return text[:MAX_ERROR_CHARS] diff --git a/cron/ledger.py b/cron/ledger.py deleted file mode 100644 index 984da83cfc..0000000000 --- a/cron/ledger.py +++ /dev/null @@ -1,47 +0,0 @@ -"""SQLite connection and transaction helpers shared by cron ledgers.""" - -from __future__ import annotations - -import sqlite3 -import threading -from contextlib import contextmanager -from pathlib import Path -from typing import Callable, Iterator - - -def open_ledger(path: Path) -> sqlite3.Connection: - """Open a profile-local ledger DB, creating its cron directory securely.""" - from cron.jobs import _ensure_cron_dir - - _ensure_cron_dir(path.parent) - return sqlite3.connect(path, timeout=5) - - -def prepare_ledger( - conn: sqlite3.Connection, *, db_label: str, synchronous_full: bool = True -) -> None: - """Configure row access, busy timeout, WAL, and optional full synchronization.""" - from hermes_state_wal import apply_wal_with_fallback - - conn.row_factory = sqlite3.Row - conn.execute("PRAGMA busy_timeout=5000") - apply_wal_with_fallback(conn, db_label=db_label) - if synchronous_full: - conn.execute("PRAGMA synchronous=FULL") - - -@contextmanager -def ledger_transaction( - lock: threading.RLock, - connect: Callable[[], sqlite3.Connection], - initialize_schema: Callable[[sqlite3.Connection], None], -) -> Iterator[sqlite3.Connection]: - """Initialize, transact on, and always close one ledger connection.""" - with lock: - conn = connect() - try: - initialize_schema(conn) - with conn: - yield conn - finally: - conn.close() diff --git a/cron/notepad.py b/cron/notepad.py index ee740203b2..d7b2092b21 100644 --- a/cron/notepad.py +++ b/cron/notepad.py @@ -15,7 +15,8 @@ from contextlib import contextmanager from pathlib import Path from typing import Any, Dict, Iterator, List, Optional -from cron.ledger import ledger_transaction, open_ledger, prepare_ledger +from cron.jobs import _ensure_cron_dir +from hermes_cli.sqlite_util import open_db, transaction from hermes_constants import get_hermes_home from hermes_time import now as _hermes_now @@ -35,11 +36,12 @@ def _current_notepad_file() -> Path: def _connect() -> sqlite3.Connection: - return open_ledger(_current_notepad_file()) + path = _current_notepad_file() + _ensure_cron_dir(path.parent) + return open_db(path, db_label="cron/notepad.db", initialize=_initialize_schema) def _initialize_schema(conn: sqlite3.Connection) -> None: - prepare_ledger(conn, db_label="cron/notepad.db", synchronous_full=False) conn.execute( """CREATE TABLE IF NOT EXISTS cron_notepad ( job_id TEXT NOT NULL, @@ -53,7 +55,7 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: @contextmanager def _transaction() -> Iterator[sqlite3.Connection]: - with ledger_transaction(_lock, _connect, _initialize_schema) as conn: + with _lock, transaction(_connect()) as conn: yield conn diff --git a/gateway/delivery_ledger.py b/gateway/delivery_ledger.py index af3352754a..ae37d3662c 100644 --- a/gateway/delivery_ledger.py +++ b/gateway/delivery_ledger.py @@ -19,9 +19,9 @@ import re import sqlite3 import threading import time -from contextlib import closing, contextmanager -from typing import Any, Dict, Iterator, List, Optional +from typing import Any, Dict, List, Optional +from hermes_cli.sqlite_util import add_column_if_missing from hermes_constants import get_hermes_home logger = logging.getLogger(__name__) @@ -143,20 +143,15 @@ def _db_path(): def _connect() -> sqlite3.Connection: - path = _db_path() - path.parent.mkdir(parents=True, exist_ok=True) - conn = sqlite3.connect(path, timeout=10) - try: - _initialize_schema(conn) - except Exception: - conn.close() # a PRAGMA/DDL failure after connect() must not leak the connection - raise - return conn + from hermes_cli.sqlite_util import open_db + + # Shared state.db: SessionDB owns the durable PRAGMA set; this opener keeps the plain-tuple rows + # and the 10 s busy timeout it always had. + return open_db(_db_path(), db_label="state.db (delivery_ledger)", busy_timeout_ms=10_000, + row_factory=None, initialize=_initialize_schema) def _initialize_schema(conn: sqlite3.Connection) -> None: - from hermes_state_wal import apply_wal_with_fallback - apply_wal_with_fallback(conn, db_label="state.db (delivery_ledger)") conn.execute( """CREATE TABLE IF NOT EXISTS delivery_obligations ( obligation_id TEXT PRIMARY KEY, @@ -176,27 +171,13 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: )""" ) if "adapter_profile" not in {row[1] for row in conn.execute("PRAGMA table_info(delivery_obligations)")}: - try: - conn.execute("ALTER TABLE delivery_obligations ADD COLUMN adapter_profile TEXT") - except sqlite3.OperationalError as exc: - # Concurrent first-use connections can both observe the old schema. - if "duplicate column" not in str(exc).lower(): - raise + add_column_if_missing(conn, "delivery_obligations", "adapter_profile", "adapter_profile TEXT") -@contextmanager -def _transaction() -> Iterator[sqlite3.Connection]: - """Open a connection, commit/rollback on exit, and ALWAYS close it: ``sqlite3.Connection`` as a - context manager only commits/rolls back, so ``with _connect()`` alone leaks a connection (and its - WAL/SHM fds) per call — ``record_obligation`` runs on every final response; exhausts RLIMIT_NOFILE. +def _transaction(): + from hermes_cli.sqlite_util import transaction - On a long-running gateway that exhausts ``RLIMIT_NOFILE`` (the cron-ledger sibling of this bug was - #69567 / PR #69594). ``record_obligation`` runs on every outbound final response, so this ledger is the - highest-frequency leaker. - """ - conn = _connect() - with closing(conn), conn: - yield conn + return transaction(_connect()) def _start_time(pid: int) -> Optional[int]: diff --git a/gateway/hosted_room_policy_checkpoint.py b/gateway/hosted_room_policy_checkpoint.py index f4e9adb822..ae3efbe512 100644 --- a/gateway/hosted_room_policy_checkpoint.py +++ b/gateway/hosted_room_policy_checkpoint.py @@ -15,6 +15,7 @@ from typing import Any, Callable, Mapping from gateway import hosted_rooms from gateway.hosted_rooms_common import DbPath, compact_json, fenced_update +from hermes_cli.sqlite_util import open_db, transaction MAX_ACTIVE_POLICY_EVENTS = 64 @@ -105,17 +106,15 @@ class HostedRoomPolicyCheckpoint: """Incrementally index room policy without compacting visible history.""" def __init__(self, db_path: DbPath) -> None: self.db_path = Path(db_path) - with self._connect() as conn: + with self._transaction() as conn: for ddl in _SCHEMA_DDL: conn.execute(ddl) def _connect(self) -> sqlite3.Connection: - from hermes_state_wal import apply_wal_with_fallback - self.db_path.parent.mkdir(parents=True, exist_ok=True) - conn = sqlite3.connect(self.db_path, timeout=10) - conn.row_factory = sqlite3.Row - apply_wal_with_fallback(conn, db_label="shared-state.db (room policy checkpoint)") - return conn + return open_db(self.db_path, db_label="shared-state.db (room policy checkpoint)", busy_timeout_ms=10_000) + + def _transaction(self): + return transaction(self._connect()) @staticmethod def _store_active_event( @@ -278,7 +277,7 @@ class HostedRoomPolicyCheckpoint: def sync(self, *, room_id: str, latest_seq: int) -> int: """Materialize each unseen event exactly once by durable cursor.""" - with self._connect() as conn: + with self._transaction() as conn: conn.execute("BEGIN IMMEDIATE") cursor = self._ensure_cursor_and_transcript(conn, room_id) if cursor > latest_seq: @@ -290,7 +289,7 @@ class HostedRoomPolicyCheckpoint: next_cursor = int(page.get("cursor") or cursor) if not rows or next_cursor <= cursor: raise RuntimeError("hosted room policy cursor did not advance") - with self._connect() as conn: + with self._transaction() as conn: conn.execute("BEGIN IMMEDIATE") _require_room(conn, room_id) for event in rows: @@ -305,7 +304,7 @@ class HostedRoomPolicyCheckpoint: def snapshot(self, *, room_id: str, latest_seq: int) -> PolicySnapshot: """Return only the oldest active discussion and its watermark set.""" through_seq = self.sync(room_id=room_id, latest_seq=latest_seq) - with self._connect() as conn: + with self._transaction() as conn: cursor = conn.execute( "SELECT stopped_through_seq FROM hosted_room_policy_cursors WHERE room_id=?", (room_id,)).fetchone() stopped_through_seq = int(cursor["stopped_through_seq"]) @@ -335,12 +334,12 @@ class HostedRoomPolicyCheckpoint: ("""SELECT 1 FROM hosted_room_policy_publications WHERE room_id=? AND task_id=? AND kind IN ('turn.settled', 'turn.failed', 'turn.cancelled')""", (room_id, task_id))) - with self._connect() as conn: + with self._transaction() as conn: return conn.execute(sql, params).fetchone() is not None def events_for_task(self, *, room_id: str, source_event_seq: int) -> list[dict[str, Any]]: """Load one bounded discussion projection for terminal reconstruction.""" - with self._connect() as conn: + with self._transaction() as conn: row = conn.execute( f"SELECT {_ROOM_EVENT_COLUMNS} FROM hosted_room_events WHERE room_id=? AND seq=?", (room_id, source_event_seq)).fetchone() @@ -359,7 +358,7 @@ class HostedRoomPolicyCheckpoint: def compact_completed(self, *, room_id: str) -> None: """Drop any completed projections left by an interrupted sync.""" - with self._connect() as conn: + with self._transaction() as conn: for row in conn.execute( "SELECT discussion_event_id FROM hosted_room_policy_threads WHERE room_id=? AND completed=1", (room_id,) ).fetchall(): diff --git a/gateway/hosted_rooms_common.py b/gateway/hosted_rooms_common.py index 85c863b31f..2ac7feb627 100644 --- a/gateway/hosted_rooms_common.py +++ b/gateway/hosted_rooms_common.py @@ -12,10 +12,12 @@ import json import re import sqlite3 import time -from contextlib import contextmanager from pathlib import Path from typing import Any, Callable, Iterator, Mapping +from hermes_cli.sqlite_util import open_db +from hermes_cli.sqlite_util import transaction as _transaction + IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]*$") DbPath = Path | str @@ -95,11 +97,9 @@ def clock(now: float | None) -> float: def open_sqlite(path: DbPath, *, timeout: float = 10) -> sqlite3.Connection: - """Row-factory connection with foreign keys on; no journal or schema work.""" - conn = sqlite3.connect(path, timeout=timeout) - conn.row_factory = sqlite3.Row - conn.execute("PRAGMA foreign_keys=ON") - return conn + """Row-factory connection with foreign keys on; no journal or schema work (steady-state readers).""" + return open_db(path, db_label="shared-state.db", busy_timeout_ms=int(timeout * 1000), wal=False, + foreign_keys=True) def connect( @@ -109,34 +109,19 @@ def connect( Multiple profile gateways share this database, so every draft-schema transition is serialized in SQLite itself: a crash rolls back the whole DDL/data migration and - another process can safely retry it. Only the transient "database is locked" class - from the journal-mode pragma is retried (it may ignore the busy timeout while another - first opener initializes the DB, especially on Windows). + another process can safely retry it. """ - from hermes_state_wal import apply_wal_with_fallback - path = Path(db_path) - path.parent.mkdir(parents=True, exist_ok=True) - conn = sqlite3.connect(path, timeout=10) - conn.row_factory = sqlite3.Row - try: - for attempt in range(lock_retries): - try: - apply_wal_with_fallback(conn, db_label=db_label) - break - except sqlite3.OperationalError as exc: - if str(exc).lower() != "database is locked" or attempt + 1 == lock_retries: - raise - time.sleep(0.01 * (2**attempt)) - conn.execute("PRAGMA foreign_keys=ON") + def _initialize(conn: sqlite3.Connection) -> None: if not ready(conn): - conn.execute("BEGIN IMMEDIATE") - initialize(conn) - conn.commit() - except Exception: - conn.rollback() - conn.close() - raise - return conn + try: + conn.execute("BEGIN IMMEDIATE") + initialize(conn) + conn.commit() + except Exception: + conn.rollback() + raise + return open_db(db_path, db_label=db_label, busy_timeout_ms=10_000, foreign_keys=True, + wal_lock_retries=lock_retries, initialize=_initialize) def fenced_update(conn: sqlite3.Connection, sql: str, params: tuple, error: Exception) -> None: @@ -154,19 +139,8 @@ def table_columns(conn: sqlite3.Connection, table: str) -> frozenset[str]: return frozenset(row[1] for row in conn.execute(f"PRAGMA table_info({table})")) -@contextmanager def transaction( connect: Callable[[DbPath], sqlite3.Connection], db_path: DbPath, *, immediate: bool ) -> Iterator[sqlite3.Connection]: """Open via ``connect``, optionally ``BEGIN IMMEDIATE``, commit on success, always close.""" - conn = connect(db_path) - try: - if immediate: - conn.execute("BEGIN IMMEDIATE") - yield conn - conn.commit() - except Exception: - conn.rollback() - raise - finally: - conn.close() + return _transaction(connect(db_path), immediate=immediate) diff --git a/gateway/platforms/api_server_run_idempotency.py b/gateway/platforms/api_server_run_idempotency.py index 22e9f580f1..0820896aa9 100644 --- a/gateway/platforms/api_server_run_idempotency.py +++ b/gateway/platforms/api_server_run_idempotency.py @@ -10,6 +10,8 @@ from contextlib import contextmanager from pathlib import Path from typing import Any, Dict +from hermes_cli.sqlite_util import add_column_if_missing + # Keep the extracted store's log records on the API server logger. logger = logging.getLogger("gateway.platforms.api_server") @@ -101,7 +103,7 @@ class RunIdempotencyStore: columns = {str(row[1]) for row in self._conn.execute("PRAGMA table_info(run_idempotency)")} for column, ddl in _MIGRATIONS.items(): if column not in columns: - self._conn.execute(f"ALTER TABLE run_idempotency ADD COLUMN {column} {ddl}") + add_column_if_missing(self._conn, "run_idempotency", column, f"{column} {ddl}") self._conn.execute( "CREATE UNIQUE INDEX IF NOT EXISTS run_idempotency_run_id ON run_idempotency(run_id)") self._conn.commit() diff --git a/hermes_cli/kanban_db_connect.py b/hermes_cli/kanban_db_connect.py index 6db4994cb7..e6debc811b 100644 --- a/hermes_cli/kanban_db_connect.py +++ b/hermes_cli/kanban_db_connect.py @@ -861,10 +861,7 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None: conn.execute(copy_sql) for name, ddl in _LATER_TASK_COLUMNS: if name not in cols: - if name == "model_override": - conn.execute("ALTER TABLE tasks ADD COLUMN model_override TEXT") - else: - _add_column_if_missing(conn, "tasks", name, ddl) + _add_column_if_missing(conn, "tasks", name, ddl) # Indexes over additive ``tasks`` columns must be created AFTER the columns # exist: ``executescript`` parses each statement against the live schema, diff --git a/hermes_cli/observability/shared_metrics.py b/hermes_cli/observability/shared_metrics.py index 63c4d7141a..cc8a51d851 100644 --- a/hermes_cli/observability/shared_metrics.py +++ b/hermes_cli/observability/shared_metrics.py @@ -12,7 +12,7 @@ from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any -from hermes_cli.sqlite_util import write_txn +from hermes_cli.sqlite_util import add_column_if_missing, write_txn from hermes_constants import get_hermes_home from utils import atomic_json_write @@ -348,9 +348,7 @@ class SharedMetricsStore: } for column, declaration in _SEND_COLUMNS: if column not in existing: - connection.execute( - f"ALTER TABLE package_outbox ADD COLUMN {column} {declaration}" - ) + add_column_if_missing(connection, "package_outbox", column, f"{column} {declaration}") for statement in _CREATE_CONSENT_TABLES_SQL: connection.execute(statement) connection.execute( diff --git a/hermes_cli/projects_db.py b/hermes_cli/projects_db.py index 8feb802fc2..02ac787140 100644 --- a/hermes_cli/projects_db.py +++ b/hermes_cli/projects_db.py @@ -16,7 +16,7 @@ from dataclasses import dataclass, field from pathlib import Path from typing import Iterable, List, Optional -from hermes_cli.sqlite_util import add_column_if_missing as _add_column_if_missing, write_txn +from hermes_cli.sqlite_util import add_column_if_missing as _add_column_if_missing, open_db, write_txn from hermes_constants import get_hermes_home @@ -118,26 +118,19 @@ def connect(db_path: Optional[Path] = None) -> sqlite3.Connection: idempotent (``CREATE TABLE IF NOT EXISTS`` + additive migrations) and cached per-path per-process. """ path = db_path if db_path is not None else projects_db_path() - path.parent.mkdir(parents=True, exist_ok=True) resolved = str(path.resolve()) - conn = sqlite3.connect(str(path)) - try: - conn.row_factory = sqlite3.Row - from hermes_state_wal import apply_wal_with_fallback - apply_wal_with_fallback(conn, db_label="projects.db") - conn.execute("PRAGMA foreign_keys=ON") - if resolved not in _INITIALIZED_PATHS: - conn.executescript(SCHEMA_SQL) - cols = {row["name"] for row in conn.execute("PRAGMA table_info(projects)")} - for col in _OPTIONAL_PROJECT_COLUMNS: - if col not in cols: - _add_column_if_missing(conn, "projects", col, f"{col} TEXT") - _INITIALIZED_PATHS.add(resolved) - except Exception: - conn.close() - raise - return conn + def _initialize(conn: sqlite3.Connection) -> None: + if resolved in _INITIALIZED_PATHS: + return + conn.executescript(SCHEMA_SQL) + cols = {row["name"] for row in conn.execute("PRAGMA table_info(projects)")} + for col in _OPTIONAL_PROJECT_COLUMNS: + if col not in cols: + _add_column_if_missing(conn, "projects", col, f"{col} TEXT") + _INITIALIZED_PATHS.add(resolved) + + return open_db(path, db_label="projects.db", foreign_keys=True, initialize=_initialize) @contextlib.contextmanager diff --git a/hermes_cli/sqlite_util.py b/hermes_cli/sqlite_util.py index e1ed4bc09f..952a49b9a7 100644 --- a/hermes_cli/sqlite_util.py +++ b/hermes_cli/sqlite_util.py @@ -1,9 +1,82 @@ -"""Shared SQLite primitives for the small per-profile / board stores.""" +"""Shared SQLite primitives for the small per-profile / board stores. + +``open_db`` is the one connect + PRAGMA stack; ``transaction`` is the one commit-and-ALWAYS-close +shape. Every hand-rolled ``_connect``/``_transaction`` pair used to re-carry the #69567 fix (a +``with conn:`` only commits — it never closes, so each call leaked a connection and its WAL/SHM fds +until GC, exhausting ``RLIMIT_NOFILE`` on long-running gateways) and at least one copy missed it. +""" from __future__ import annotations import contextlib import sqlite3 +import time +from pathlib import Path +from typing import Callable, Iterator + + +def open_db( + path: Path | str, + *, + db_label: str, + busy_timeout_ms: int = 5000, + wal: bool = True, + foreign_keys: bool = False, + synchronous_full: bool = False, + row_factory=sqlite3.Row, + check_same_thread: bool = True, + wal_lock_retries: int = 1, + initialize: Callable[[sqlite3.Connection], None] | None = None, +) -> sqlite3.Connection: + """Open ``path`` (parent created), apply the PRAGMA set, run ``initialize``; closed if anything raises. + + ``busy_timeout_ms`` is the single busy knob: it is passed as ``connect(timeout=)`` AND set as the + explicit PRAGMA so it is observable. ``wal=True`` goes through ``apply_wal_with_fallback`` — the + only journal-mode setter that carries the WAL-reset-bug gate, the network-FS silent-refusal + fallback and the never-live-downgrade invariant; a raw ``PRAGMA journal_mode=WAL`` bypasses all + three. Only the transient ``database is locked`` from that pragma is retried (``wal_lock_retries``): + a first opener initializing a shared DB can make it ignore the busy timeout, notably on Windows. + """ + from hermes_state_wal import apply_wal_with_fallback + + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + # Resolved at call time: fd-leak tests patch ``sqlite3.connect`` through the caller's module. + conn = sqlite3.connect(path, timeout=busy_timeout_ms / 1000, check_same_thread=check_same_thread) + try: + conn.row_factory = row_factory + conn.execute(f"PRAGMA busy_timeout={int(busy_timeout_ms)}") + if wal: + for attempt in range(wal_lock_retries): + try: + apply_wal_with_fallback(conn, db_label=db_label) + break + except sqlite3.OperationalError as exc: + if str(exc).lower() != "database is locked" or attempt + 1 == wal_lock_retries: + raise + time.sleep(0.01 * (2**attempt)) + if foreign_keys: + conn.execute("PRAGMA foreign_keys=ON") + if synchronous_full: + conn.execute("PRAGMA synchronous=FULL") + if initialize is not None: + initialize(conn) + except BaseException: + conn.close() + raise + return conn + + +@contextlib.contextmanager +def transaction(conn: sqlite3.Connection, *, immediate: bool = False) -> Iterator[sqlite3.Connection]: + """Commit on success, roll back on error, and ALWAYS close ``conn`` (see the module docstring).""" + try: + if immediate: + conn.execute("BEGIN IMMEDIATE") + with conn: + yield conn + finally: + conn.close() def add_column_if_missing(conn: sqlite3.Connection, table: str, column: str, ddl: str) -> bool: @@ -24,8 +97,9 @@ def add_column_if_missing(conn: sqlite3.Connection, table: str, column: str, ddl @contextlib.contextmanager def write_txn(conn: sqlite3.Connection): - """An IMMEDIATE write transaction. The explicit ROLLBACK is guarded so a SQLite auto-rollback - (no transaction left under EIO / contention / corruption) cannot shadow the original error.""" + """An IMMEDIATE write transaction on a long-lived connection (stays open). The explicit ROLLBACK is + guarded so a SQLite auto-rollback (no transaction left under EIO / contention / corruption) cannot + shadow the original error.""" conn.execute("BEGIN IMMEDIATE") try: yield conn diff --git a/hermes_state.py b/hermes_state.py index 31cd1521e3..c69ea1e1a1 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -24,7 +24,6 @@ from collections import deque from contextlib import contextmanager from pathlib import Path -from agent.message_sanitization import _sanitize_surrogates from hermes_constants import get_hermes_home, mkdir_under_hermes_home from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple, TypeVar, cast @@ -144,11 +143,6 @@ def _compression_lock_holder_process_is_dead(holder: str) -> bool: return False -def _scrub_surrogates(value: Any) -> Any: - """Replace lone surrogates in text (sqlite3 raises UnicodeEncodeError, aborting the whole write).""" - return _sanitize_surrogates(value) if isinstance(value, str) else value - - # Billing buckets that aren't a routable provider identity: a session that persisted only # one of these (never ran /model) falls back to the config default. Shared by # session_gateway_runtime and tui_gateway.server so they cannot drift. diff --git a/plugins/memory/holographic/store.py b/plugins/memory/holographic/store.py index 600e991591..a913cf2d39 100644 --- a/plugins/memory/holographic/store.py +++ b/plugins/memory/holographic/store.py @@ -129,7 +129,8 @@ class MemoryStore: apply_wal_with_fallback(self._conn, db_label="memory_store.db (holographic)") self._conn.executescript(_SCHEMA) if "hrr_vector" not in {row[1] for row in self._conn.execute("PRAGMA table_info(facts)").fetchall()}: - self._conn.execute("ALTER TABLE facts ADD COLUMN hrr_vector BLOB") + from hermes_cli.sqlite_util import add_column_if_missing + add_column_if_missing(self._conn, "facts", "hrr_vector", "hrr_vector BLOB") self._conn.commit() def _one(self, sql: str, params=()): diff --git a/plugins/plugin_storage.py b/plugins/plugin_storage.py index 6be5a8e279..d1ace9f41c 100644 --- a/plugins/plugin_storage.py +++ b/plugins/plugin_storage.py @@ -38,7 +38,9 @@ def plugin_db(name: str, filename: str = "data.db") -> sqlite3.Connection: ``check_same_thread=False`` for the threaded FastAPI/tool env — caller owns transactions.""" if Path(filename).name != filename or not filename: raise ValueError(f"invalid plugin db filename: {filename!r}") - conn = sqlite3.connect(plugin_data_dir(name) / filename, check_same_thread=False) - conn.execute("PRAGMA journal_mode=WAL") - conn.execute("PRAGMA foreign_keys=ON") - return conn + from hermes_cli.sqlite_util import open_db + + # WAL via the shared fallback helper: network filesystems degrade to DELETE and WAL-reset-bug + # builds never enable it, instead of every plugin DB bypassing those rules with a raw PRAGMA. + return open_db(plugin_data_dir(name) / filename, db_label=f"plugin-data/{name}/{filename}", + foreign_keys=True, row_factory=None, check_same_thread=False) diff --git a/tests/cron/test_upgrade_module_skew.py b/tests/cron/test_upgrade_module_skew.py index 6e3828cc0e..64b144603c 100644 --- a/tests/cron/test_upgrade_module_skew.py +++ b/tests/cron/test_upgrade_module_skew.py @@ -13,7 +13,7 @@ def test_lazy_cron_stores_do_not_require_new_symbols_on_cached_executions_module import sys import cron.executions as executions -for name in ("ledger_transaction", "open_ledger", "prepare_ledger"): +for name in ("open_db", "transaction", "add_column_if_missing", "_ensure_cron_dir"): delattr(executions, name) sys.modules.pop("cron.incidents", None) sys.modules.pop("cron.notepad", None) diff --git a/tests/hermes_cli/test_sqlite_util_canonical.py b/tests/hermes_cli/test_sqlite_util_canonical.py new file mode 100644 index 0000000000..030cd60a91 --- /dev/null +++ b/tests/hermes_cli/test_sqlite_util_canonical.py @@ -0,0 +1,136 @@ +"""Invariants for the canonical SQLite connect/transaction layer (``hermes_cli/sqlite_util.py``). + +Every small store used to carry its own connect + PRAGMA + ``with conn:`` stack, so the #69567 fd-leak +fix and the WAL fallback rules had to be re-pasted per module (and at least one copy missed each). +These tests pin the contract: one opener, one closer, and every store routed through them. +""" + +from __future__ import annotations + +import sqlite3 + +import pytest + +from hermes_cli import sqlite_util + + +def test_transaction_closes_the_connection_even_when_the_body_raises(tmp_path): + db = tmp_path / "t.db" + conn = sqlite_util.open_db(db, db_label="t.db", wal=False) + conn.execute("CREATE TABLE t (x)") + conn.commit() + + with pytest.raises(RuntimeError): + with sqlite_util.transaction(conn) as c: + c.execute("INSERT INTO t VALUES (1)") + raise RuntimeError("mid-transaction") + + # Rolled back AND closed: a closed connection refuses every statement. + with pytest.raises(sqlite3.ProgrammingError): + conn.execute("SELECT 1") + with sqlite3.connect(db) as check: + assert check.execute("SELECT count(*) FROM t").fetchone()[0] == 0 + + # The success path closes too. + conn = sqlite_util.open_db(db, db_label="t.db", wal=False) + with sqlite_util.transaction(conn) as c: + c.execute("INSERT INTO t VALUES (2)") + with pytest.raises(sqlite3.ProgrammingError): + conn.execute("SELECT 1") + + +def test_open_db_closes_the_half_open_connection_when_initialize_raises(monkeypatch, tmp_path): + opened = [] + real_connect = sqlite3.connect + + def tracking_connect(*args, **kwargs): + conn = real_connect(*args, **kwargs) + opened.append(conn) + return conn + + monkeypatch.setattr(sqlite_util.sqlite3, "connect", tracking_connect) + + def broken(conn): + raise sqlite3.OperationalError("boom") + + with pytest.raises(sqlite3.OperationalError): + sqlite_util.open_db(tmp_path / "x.db", db_label="x.db", wal=False, initialize=broken) + assert len(opened) == 1 + with pytest.raises(sqlite3.ProgrammingError): + opened[0].execute("SELECT 1") + + +# Every store that opens its own SQLite file (path -> module attribute holding the opener). +_STORE_OPENERS = ( + ("agent.verification_evidence", "_connect"), + ("cron.executions", "_connect"), + ("cron.incidents", "_connect"), + ("cron.notepad", "_connect"), + ("cron.delivery_queue", "_connect"), + ("gateway.delivery_ledger", "_connect"), + ("tools.async_delegation", "_connect"), + ("hermes_cli.projects_db", "connect"), + ("gateway.hosted_rooms_common", "open_sqlite"), +) + + +@pytest.mark.parametrize("module_name, attr", _STORE_OPENERS) +def test_every_store_opens_through_the_canonical_open_db(monkeypatch, tmp_path, module_name, attr): + import importlib + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + module = importlib.import_module(module_name) + for name in ("EXECUTIONS_FILE", "NOTEPAD_FILE", "DELIVERY_DB"): + if hasattr(module, name): + monkeypatch.setattr(module, name, tmp_path / f"{name.lower()}.db") + if module_name == "cron.incidents": + monkeypatch.setattr(module._executions, "EXECUTIONS_FILE", tmp_path / "executions.db") + if module_name == "agent.verification_evidence": + monkeypatch.setattr(module, "_db_path", lambda: tmp_path / "ve.db") + if module_name in ("gateway.delivery_ledger", "tools.async_delegation"): + monkeypatch.setattr(module, "_db_path", lambda: tmp_path / "state.db") + + calls = [] + real_open_db = sqlite_util.open_db + + def spy(path, **kwargs): + calls.append(kwargs) + return real_open_db(path, **kwargs) + + # Patch where production reads: a module-level ``from sqlite_util import open_db`` binds its own name. + monkeypatch.setattr(sqlite_util, "open_db", spy) + if getattr(module, "open_db", None) is real_open_db: + monkeypatch.setattr(module, "open_db", spy) + opener = getattr(module, attr) + args = (tmp_path / "opened.db",) if module_name == "gateway.hosted_rooms_common" else () + if module_name == "hermes_cli.projects_db": + args = (tmp_path / "projects.db",) + conn = opener(*args) + try: + assert conn.execute("PRAGMA busy_timeout").fetchone()[0] > 0 + finally: + conn.close() + assert len(calls) == 1 and calls[0]["db_label"], module_name + + +def test_plugin_db_wal_goes_through_the_shared_fallback(monkeypatch, tmp_path): + """A raw ``PRAGMA journal_mode=WAL`` bypasses the network-FS fallback and the WAL-reset-bug gate; + plugin databases must obey the same rules as every core store.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + import hermes_state_wal + from plugins import plugin_storage + + seen = [] + real = hermes_state_wal.apply_wal_with_fallback + + def spy(conn, **kwargs): + seen.append(kwargs["db_label"]) + return real(conn, **kwargs) + + monkeypatch.setattr(hermes_state_wal, "apply_wal_with_fallback", spy) + conn = plugin_storage.plugin_db("board") + try: + assert conn.execute("PRAGMA foreign_keys").fetchone()[0] == 1 + finally: + conn.close() + assert seen == ["plugin-data/board/data.db"] diff --git a/tools/async_delegation.py b/tools/async_delegation.py index df2a9ffec6..0069015e7e 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -17,8 +17,7 @@ import threading import time import uuid from concurrent.futures import ThreadPoolExecutor -from contextlib import contextmanager -from typing import Any, Callable, Dict, Iterator, List, Optional +from typing import Any, Callable, Dict, List, Optional from hermes_constants import get_hermes_home from tools.daemon_pool import DaemonThreadPoolExecutor @@ -83,19 +82,18 @@ def _db_path(): def _connect() -> sqlite3.Connection: - path = _db_path() - path.parent.mkdir(parents=True, exist_ok=True) + from hermes_cli.sqlite_util import open_db # Same state.db as hermes_state.SessionDB -- reuse its owner-only (0600) # hardening so this writer doesn't create/leave the file (and its WAL # sidecars) at the process umask. See hermes_state._secure_state_db_files. from hermes_state import _secure_state_db_files + + path = _db_path() + path.parent.mkdir(parents=True, exist_ok=True) _secure_state_db_files(path, create_main=True) - conn = sqlite3.connect(path, timeout=10) - try: - _initialize_schema(conn) - except Exception: - conn.close() # don't leak the connection on PRAGMA/DDL failure - raise + # wal=False: SessionDB owns state.db's journal mode (_initialize_schema applies the barriers). + conn = open_db(path, db_label="state.db (async_delegation)", busy_timeout_ms=10_000, + wal=False, row_factory=None, initialize=_initialize_schema) _secure_state_db_files(path) return conn @@ -116,23 +114,10 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: reconcile_state_schema(conn) -@contextmanager -def _transaction() -> Iterator[sqlite3.Connection]: - """Open a connection, commit/rollback on exit, and ALWAYS close it (``with conn:`` - alone leaks the connection and WAL/SHM fds until GC). +def _transaction(): + from hermes_cli.sqlite_util import transaction - ``sqlite3.Connection.__enter__``/``__exit__`` only commit or roll back the transaction; they do not - close the connection. Using ``with _connect()`` alone therefore leaks a connection — and its WAL/SHM - file descriptors — on every durable dispatch, completion, and delivery-claim, deferring the close to the - garbage collector. On a long-running gateway that exhausts ``RLIMIT_NOFILE`` (the cron-ledger sibling of - this bug was #69567 / PR #69594). - """ - conn = _connect() - try: - with conn: - yield conn - finally: - conn.close() + return transaction(_connect()) def _capture_routing_origin() -> Dict[str, Any]: