diff --git a/hermes_cli/web_routers/sessions.py b/hermes_cli/web_routers/sessions.py index a4da40c3a1..657840ed62 100644 --- a/hermes_cli/web_routers/sessions.py +++ b/hermes_cli/web_routers/sessions.py @@ -31,7 +31,7 @@ from hermes_cli.web_models import ( SessionPrune, SessionRename, ) -from hermes_state import is_malformed_db_error +from hermes_state import is_malformed_db_error, is_transient_sqlite_error # Same logger the handlers used before extraction (identical logger object). _log = logging.getLogger("hermes_cli.web_server") @@ -197,6 +197,22 @@ def get_sessions( db.close() except HTTPException: raise + except sqlite3.OperationalError as exc: + _log.exception("GET /api/sessions failed") + # 503, not 500: the store is busy, not gone. The desktop keeps the + # sidebar it already has instead of reading a 500 as an authoritative + # empty list. Retrying the OPEN here is deliberately not done — the + # bounded retry lives in SessionDB's read-only constructor, so every + # read-only opener gets it, not just this route. + transient = is_transient_sqlite_error(exc) + raise HTTPException( + status_code=503 if transient else 500, + detail=( + "Session store is busy (disk I/O or lock). Retry; the list was not cleared." + if transient + else "Internal server error" + ), + ) from exc except Exception: _log.exception("GET /api/sessions failed") raise HTTPException(status_code=500, detail="Internal server error") diff --git a/hermes_state.py b/hermes_state.py index 21807841bc..2ca8b61fca 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -374,6 +374,20 @@ DEFAULT_DB_PATH = get_hermes_home() / "state.db" # query; short enough that transient fd pressure doesn't strand the read pool. _READ_OPEN_RETRY_SECONDS = 60.0 +# Transient SQLITE_IOERR retry budget for READ-ONLY opens (#100436). A WAL +# database being actively written (checkpoint, WAL reset/truncate, frame +# flush) can surface "disk I/O error" to a concurrent ``mode=ro`` reader in +# a millisecond-wide transition window: the read-only connection cannot +# perform the WAL recovery a read through a stale or mid-update -shm file +# needs, because recovery requires writing the -shm index, which mode=ro +# refuses. The window closes on its own (the writer finishes the transition), +# so a bounded number of short retries makes the open succeed instead of +# 500-ing the whole /api/sessions poll (or any other read-only opener). +# Deliberately NOT attempted on writable opens: a writer owns the +# transition, so an IOERR there means a real storage/fd problem. +_READ_ONLY_IOERR_RETRY_ATTEMPTS = 3 +_READ_ONLY_IOERR_RETRY_BACKOFF_S = 0.05 + # Hard ceiling on read-only connections ALIVE at once against one database # FILE — pooled idle ones and checked-out ones together, summed over every # SessionDB in this process that points at that file. See _PathReadBudget. @@ -2082,6 +2096,52 @@ def is_malformed_db_error(exc: BaseException) -> bool: return any(marker in str(exc).lower() for marker in _MALFORMED_DB_MARKERS) +# SQLITE_IOERR, matched as a plain substring so wrapped error strings still +# classify. Shared by the read-only open retry and the write-path BEGIN retry. +_DISK_IO_ERROR_MARKER = "disk i/o error" + +# Broader set for HTTP classification: a read that failed for one of these +# reasons found the store BUSY, not gone. Callers map it to 503 (retry, the +# list was not cleared) instead of 500. Corruption is deliberately absent — +# a malformed store must surface, not be retried into a timeout. +_TRANSIENT_SQLITE_MARKERS = ( + _DISK_IO_ERROR_MARKER, + "database is locked", + "database table is locked", + "busy", +) + + +def is_transient_sqlite_error(exc: BaseException) -> bool: + """True when a SQLite failure means "busy right now", not "damaged". + + One predicate so the read paths cannot drift apart on what counts as + recoverable: the read-only open retry, and the HTTP 503-vs-500 split on + the session-list endpoints, classify the same way. + """ + if not isinstance(exc, sqlite3.OperationalError): + return False + message = str(exc).lower() + return any(marker in message for marker in _TRANSIENT_SQLITE_MARKERS) + + +def _is_transient_read_only_ioerr(exc: sqlite3.OperationalError, *, attempt: int) -> bool: + """True when a read-only open should be retried rather than raised. + + A ``mode=ro`` connection cannot perform WAL recovery (recovery needs to + write the -shm index, which read-only mode refuses), so a concurrent WAL + checkpoint / reset / frame-flush can surface ``SQLITE_IOERR`` ("disk I/O + error") to a reader on an otherwise healthy database (#100436). The + transition is millisecond-scale, so a bounded number of short retries + clears it without changing classification for genuine storage failures — + a persistent IOERR still exhausts the budget and propagates. + """ + return ( + attempt < _READ_ONLY_IOERR_RETRY_ATTEMPTS + and _DISK_IO_ERROR_MARKER in str(exc).lower() + ) + + def is_malformed_schema_error(exc: BaseException) -> bool: """True only when SQLite explicitly reports malformed schema text. @@ -5123,46 +5183,67 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) # must already exist + be initialised (callers guard on # db_path.exists()); a SELECT against an empty file raises and # the caller degrades per-profile. - self._conn = _connect_tracked_db( - f"file:{self.db_path}?mode=ro", - tracking_path=self.db_path, - uri=True, - check_same_thread=False, - timeout=1.0, - isolation_level=None, - ) - self._conn.row_factory = sqlite3.Row - # FTS capability flags normally come from writable schema - # initialisation. Probe existing virtual tables with SELECTs - # only so read-only search keeps its FTS and trigram paths. - # Close the connection on ANY probe failure (e.g. malformed - # schema raises DatabaseError, not the OperationalError the - # probe handles). The constructor's outer finally also covers - # failures before this probe and BaseException paths, so a - # leaked tracked connection cannot block _backup_db_file's - # raw-copy for the rest of the process — the writable heal - # that follows would then repair WITHOUT its forensic backup. - try: - apply_database_pragmas(self._conn, db_label="state.db") - cursor = self._conn.cursor() - self._fts_enabled = ( - self._fts_table_probe(cursor, "messages_fts") is True - ) - if self._fts_enabled: - self._trigram_available = ( - self._fts_table_probe( - cursor, - "messages_fts_trigram", - ) - is True - ) - except BaseException: - conn, self._conn = self._conn, None + open_attempt = 0 + while True: try: - conn.close() - except Exception: - pass - raise + self._conn = _connect_tracked_db( + f"file:{self.db_path}?mode=ro", + tracking_path=self.db_path, + uri=True, + check_same_thread=False, + timeout=1.0, + isolation_level=None, + ) + self._conn.row_factory = sqlite3.Row + # FTS capability flags normally come from writable schema + # initialisation. Probe existing virtual tables with + # SELECTs only so read-only search keeps its FTS and + # trigram paths. Close the connection on ANY probe + # failure (e.g. malformed schema raises DatabaseError, + # not the OperationalError the probe handles). The + # constructor's outer finally also covers failures + # before this probe and BaseException paths, so a + # leaked tracked connection cannot block + # _backup_db_file's raw-copy for the rest of the + # process — the writable heal that follows would then + # repair WITHOUT its forensic backup. + try: + apply_database_pragmas(self._conn, db_label="state.db") + cursor = self._conn.cursor() + self._fts_enabled = ( + self._fts_table_probe(cursor, "messages_fts") + is True + ) + if self._fts_enabled: + self._trigram_available = ( + self._fts_table_probe( + cursor, + "messages_fts_trigram", + ) + is True + ) + except BaseException: + conn, self._conn = self._conn, None + try: + conn.close() + except Exception: + pass + raise + break + except sqlite3.OperationalError as ioerr: + # A WAL checkpoint / reset / frame-flush in flight on + # the writer side can surface SQLITE_IOERR to a + # concurrent mode=ro reader (it cannot perform the + # recovery the read needs — recovery writes the -shm + # index, which mode=ro refuses). The transition closes + # in milliseconds, so retry a bounded number of times + # before classifying the store as failed (#100436). + if not _is_transient_read_only_ioerr( + ioerr, attempt=open_attempt + ): + raise + open_attempt += 1 + time.sleep(_READ_ONLY_IOERR_RETRY_BACKOFF_S) self._record_db_file_identity() initialization_complete = True return @@ -5930,6 +6011,13 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) # Set on the first compression-busy collision so the short wait is # measured from then, not from the start of the write. compression_deadline: Optional[float] = None + # One retry for SQLITE_IOERR raised by BEGIN IMMEDIATE itself. The + # callback has not run at that point, so there is no durable effect + # to replay and the retry is exactly-once safe (#99502's contract). + # Once the callback starts, an IOERR leaves the write's settlement + # unknown and must propagate — this helper owns non-idempotent + # transcript/counter mutations, not just idempotent UPSERTs. + ioerr_begin_retried = False # Transient engine-level error observed on contended WAL appends # (dual gateway/agent writers; FTS5 trigram sync holds the write @@ -5943,6 +6031,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) while True: self._raise_if_db_replaced() + fn_started = False try: with self._lock: if self._conn is None: @@ -5951,6 +6040,7 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) self._reopen_after_close_locked(context="write") self._conn.execute("BEGIN IMMEDIATE") try: + fn_started = True result = fn(self._conn) self._conn.commit() except BaseException: @@ -6003,7 +6093,21 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) ) from exc if _is_no_more_rows(exc) and self._sleep_before_write_retry(deadline, patience_s): continue - # Non-lock error or patience exhausted — propagate. + if ( + _DISK_IO_ERROR_MARKER in err_msg + and not fn_started + and not ioerr_begin_retried + and self._sleep_before_write_retry(deadline, patience_s) + ): + # BEGIN IMMEDIATE itself hit a transient WAL-transition + # IOERR. Nothing has been mutated, so retrying on the SAME + # connection replays nothing. Never close()+reopen to + # "heal" it: close() cancels this process's POSIX locks on + # the file for every sibling connection (howtocorrupt §2.2). + ioerr_begin_retried = True + continue + # Non-lock error, the callback already ran (settlement is + # unknown — do not replay), or patience exhausted. raise except sqlite3.DatabaseError as exc: if _is_no_more_rows(exc) and self._sleep_before_write_retry(deadline, patience_s): diff --git a/tests/hermes_cli/test_session_list_reader_disposable.py b/tests/hermes_cli/test_session_list_reader_disposable.py new file mode 100644 index 0000000000..1109c1b5d8 --- /dev/null +++ b/tests/hermes_cli/test_session_list_reader_disposable.py @@ -0,0 +1,96 @@ +"""Read-only session-list opens must stay disposable. + +Two properties that a "keep the read-only handle for the process lifetime" +optimisation silently destroys. Both are asserted against the real +``_open_session_db_at_path`` read path the sidebar poll uses, because both +failures are invisible in a unit test that mocks the store. + +1. **The store on disk is the truth.** Recovering a corrupt ``state.db`` + is a file swap (``mv state.db state.db.corrupt-…; cp -a recovered.db + state.db``) performed while the backend is stopped, but a poll can also + race a restore. A reader pinned to the old inode keeps serving + pre-recovery rows forever, so the user "recovers" and still sees the + broken list. + +2. **Forensic backup must stay reachable.** ``offline_file_access`` refuses + raw byte access while ANY tracked connection is registered for the path, + because a raw ``close()`` would cancel this process's POSIX advisory locks + (howtocorrupt §2.2). ``_backup_db_file`` (the copy taken BEFORE a malformed + store is repaired) and ``_db_fingerprint`` (the repair-attempt ledger key) + both go through it. A never-closed list reader makes both fail for the rest + of the process, so a repair runs without its forensic backup and the + ledger degrades to a size-only key. +""" + +from __future__ import annotations + +import shutil + +from hermes_cli.sqlite_safe_read import LiveConnectionError, offline_file_access +from hermes_cli.web_server import _open_session_db_at_path +from hermes_state import SessionDB, _db_fingerprint + + +def _ids(db) -> list: + return [row["id"] for row in db.list_sessions_rich(limit=10, compact_rows=True)] + + +def test_poll_observes_a_replaced_state_db(tmp_path): + db_path = tmp_path / "state.db" + old = SessionDB(db_path=db_path) + old.create_session("before-recovery", source="cli") + old.close() + + first = _open_session_db_at_path(db_path, read_only=True) + try: + assert _ids(first) == ["before-recovery"] + finally: + first.close() + + # `hermes sessions recover` writes a clean database, which the operator + # then installs over the corrupt one. + recovered = tmp_path / "recovered-state.db" + rebuilt = SessionDB(db_path=recovered) + rebuilt.create_session("after-recovery", source="cli") + rebuilt.close() + + for suffix in ("-wal", "-shm"): + sidecar = db_path.with_name(db_path.name + suffix) + if sidecar.exists(): + sidecar.unlink() + db_path.unlink() + shutil.copy2(recovered, db_path) + + second = _open_session_db_at_path(db_path, read_only=True) + try: + assert _ids(second) == ["after-recovery"] + finally: + second.close() + + +def test_poll_leaves_forensic_backup_reachable(tmp_path): + db_path = tmp_path / "state.db" + writer = SessionDB(db_path=db_path) + writer.create_session("s1", source="cli") + writer.close() + + baseline = _db_fingerprint(db_path) + assert baseline is not None + + poll = _open_session_db_at_path(db_path, read_only=True) + try: + assert _ids(poll) == ["s1"] + finally: + poll.close() + + # The raw-copy path a malformed-store repair takes before it touches + # anything must still be permitted after the poll. + try: + with offline_file_access(db_path, what="forensic-backup"): + pass + except LiveConnectionError as exc: # pragma: no cover - failure detail + raise AssertionError( + f"a session-list poll left a tracked connection open: {exc}" + ) from exc + + assert _db_fingerprint(db_path) == baseline diff --git a/tests/hermes_cli/test_web_server.py b/tests/hermes_cli/test_web_server.py index e00fbd4031..9f7a6d810a 100644 --- a/tests/hermes_cli/test_web_server.py +++ b/tests/hermes_cli/test_web_server.py @@ -319,6 +319,29 @@ class TestWebServerEndpoints: monitor.close() writer.close() + def test_get_sessions_transient_ioerr_is_503(self, monkeypatch): + """Busy store, not a gone store: the desktop keeps the list it has.""" + import sqlite3 + + from hermes_cli import web_server + + def boom(*_args, **_kwargs): + raise sqlite3.OperationalError("disk I/O error") + + monkeypatch.setattr(web_server, "_open_session_db_for_profile", boom) + assert self.client.get("/api/sessions?limit=1&offset=0").status_code == 503 + + def test_get_sessions_non_transient_operational_error_is_500(self, monkeypatch): + import sqlite3 + + from hermes_cli import web_server + + def boom(*_args, **_kwargs): + raise sqlite3.OperationalError("no such table: sessions") + + monkeypatch.setattr(web_server, "_open_session_db_for_profile", boom) + assert self.client.get("/api/sessions?limit=1&offset=0").status_code == 500 + def test_get_status_loads_gateway_config_off_event_loop(self, monkeypatch): """Cold gateway config loading must not block the WebSocket loop. diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index 28480c1d15..54994c8b7b 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -281,6 +281,83 @@ class TestConnectionLifecycle: healed.close() assert list(tmp_path.glob("*malformed-backup*")) + def test_read_only_open_retries_transient_wal_ioerr(self, tmp_path, monkeypatch): + """A transient SQLITE_IOERR on a read-only open must retry, not raise. + + A ``mode=ro`` connection cannot perform WAL recovery (recovery would + need to write the -shm index, which read-only mode refuses), so a + concurrent checkpoint / WAL reset / frame-flush on the writer side can + surface "disk I/O error" to a reader on a perfectly healthy database + (#100436). The transition window is millisecond-scale; a bounded retry + must let the open succeed instead of 500-ing the /api/sessions poll + and every other read-only opener. + """ + import sqlite3 + + from hermes_cli.sqlite_safe_read import has_live_connection + + db_path = tmp_path / "state.db" + writable = SessionDB(db_path=db_path) + writable.create_session("wal-race", source="cli") + writable.close() + + real_connect = hermes_state._connect_tracked_db + attempts = [] + + def flaky_connect(*args, **kwargs): + attempts.append(kwargs.get("uri")) + if len(attempts) == 1: + # First open lands inside the writer's WAL transition window. + raise sqlite3.OperationalError("disk I/O error") + return real_connect(*args, **kwargs) + + monkeypatch.setattr(hermes_state, "_connect_tracked_db", flaky_connect) + # Keep the test fast: one backoff tick is enough; the retry budget + # itself is exercised by the attempt count below. + monkeypatch.setattr(hermes_state, "_READ_ONLY_IOERR_RETRY_BACKOFF_S", 0.0) + + read_only = SessionDB(db_path=db_path, read_only=True) + try: + assert read_only._fts_enabled is True + matches = read_only.search_messages("wal-race") + finally: + read_only.close() + + assert len(attempts) >= 2, "the transient IOERR must be retried" + assert has_live_connection(db_path) is False # no leaked connections + + def test_read_only_open_exhausts_retry_budget_for_persistent_ioerr( + self, tmp_path, monkeypatch + ): + """A persistent SQLITE_IOERR must exhaust the budget and raise. + + The retry exists to ride out a millisecond WAL transition — a + storage layer that keeps failing after the full budget is genuinely + broken and must surface the error (and not loop forever). + """ + import sqlite3 + + db_path = tmp_path / "state.db" + writable = SessionDB(db_path=db_path) + writable.create_session("broken-disk", source="cli") + writable.close() + + attempts = [] + + def bad_connect(*args, **kwargs): + attempts.append(1) + raise sqlite3.OperationalError("disk I/O error") + + monkeypatch.setattr(hermes_state, "_connect_tracked_db", bad_connect) + monkeypatch.setattr(hermes_state, "_READ_ONLY_IOERR_RETRY_BACKOFF_S", 0.0) + budget = hermes_state._READ_ONLY_IOERR_RETRY_ATTEMPTS + + with pytest.raises(sqlite3.OperationalError, match="disk I/O error"): + SessionDB(db_path=db_path, read_only=True) + + # budget + 1 = the initial attempt plus `budget` retries. + assert len(attempts) == budget + 1 + # ========================================================================= # Session lifecycle diff --git a/tests/test_state_db_notadb_fail_closed.py b/tests/test_state_db_notadb_fail_closed.py index fbdf376fef..6f5c787ea0 100644 --- a/tests/test_state_db_notadb_fail_closed.py +++ b/tests/test_state_db_notadb_fail_closed.py @@ -1,9 +1,10 @@ """Tests for fail-closed state.db NOTADB handling and journal-mode EIO retries. -Covers the two independently-valuable pieces salvaged from the state.db -hardening rollup: +Covers: * fail closed when a live write connection reports ``file is not a database``; +* the write-path SQLITE_IOERR retry boundary: admitted only when the callback + has provably not run, never by closing and replaying; * transient ``disk i/o error`` retry in ``_on_disk_journal_mode`` so a one-shot EIO doesn't push callers onto the fail-closed unknown-mode branch. """ @@ -49,6 +50,89 @@ class TestFailClosedAfterNotADb: db.close() +class TestWriteIoerrRetryBoundary: + """IOERR retry is admitted by EFFECT POSITION, not error spelling. + + ``_execute_write`` owns non-idempotent transcript/counter mutations, so + replaying its callback is only safe when the first attempt provably did + nothing. SQLite does not define ``SQLITE_IOERR`` as pre-effect-only (an + IOERR at fsync/commit may or may not have landed), so the admission gate + is "did the callback start", not "does the message say disk I/O". + """ + + def test_ioerr_on_begin_retries_because_the_callback_never_ran(self, tmp_path): + db = SessionDB(db_path=tmp_path / "state.db") + real_conn = db._conn + try: + + class _BeginIoerrOnce: + def __init__(self, conn): + self._real = conn + self.begins = 0 + + def execute(self, sql, *args, **kwargs): + if str(sql).strip().upper().startswith("BEGIN") and self.begins == 0: + self.begins += 1 + raise sqlite3.OperationalError("disk I/O error") + return self._real.execute(sql, *args, **kwargs) + + def __getattr__(self, name): + return getattr(self._real, name) + + proxy = _BeginIoerrOnce(real_conn) + db._conn = proxy + db.create_session(session_id="s1", source="cli", model="test") + assert proxy.begins == 1 + finally: + db._conn = real_conn + + rows = db.list_sessions_rich(limit=10, compact_rows=True) + assert [row["id"] for row in rows] == ["s1"] + db.close() + + def test_ioerr_after_the_callback_mutates_does_not_replay(self, tmp_path): + """Settlement is unknown once the callback has run — surface, don't rerun.""" + db = SessionDB(db_path=tmp_path / "state.db") + try: + calls = [] + + def mutate_then_fail(conn): + calls.append(1) + conn.execute( + "INSERT INTO sessions (id, started_at, source) VALUES (?, ?, ?)", + (f"row-{len(calls)}", 1.0, "cli"), + ) + raise sqlite3.OperationalError("disk I/O error") + + with pytest.raises(sqlite3.OperationalError, match="disk I/O error"): + db._execute_write(mutate_then_fail) + + assert calls == [1], "a started write must not be replayed" + assert db.list_sessions_rich(limit=10, compact_rows=True) == [] + finally: + db.close() + + def test_write_ioerr_never_closes_the_connection(self, tmp_path, monkeypatch): + """close() cancels this process's POSIX locks for every sibling fd.""" + db = SessionDB(db_path=tmp_path / "state.db") + try: + closed = [] + monkeypatch.setattr( + type(db._conn), "close", lambda self: closed.append(1), raising=False + ) + + def always_ioerr(conn): + raise sqlite3.OperationalError("disk I/O error") + + with pytest.raises(sqlite3.OperationalError): + db._execute_write(always_ioerr) + + assert closed == [] + assert db._conn is not None + finally: + db.close() + + class TestOnDiskJournalModeEioRetry: def _conn_raising_then(self, failures, result_rows): conn = MagicMock()