From 66d7a39ea68141b27c2839c48e4c995d60717ccb Mon Sep 17 00:00:00 2001 From: zhouou6 Date: Wed, 12 Aug 2026 16:22:07 -0700 Subject: [PATCH] fix(state): self-heal 'file is not a database' write connections + retry transient EIO on journal-mode probe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Salvaged remainder of PR #82280 (state.db hardening rollup): - Runtime connection corruption: a sibling process replacing/truncating the backing file breaks the live write connection — every subsequent write raises 'file is not a database' and the gateway wedges permanently (messages pile up in memory). Add a bounded one-shot reconnect on the write path: close the broken connection, reopen the DB file (re-running WAL activation + schema reconciliation), retry the failed write once. - _on_disk_journal_mode: retry transient 'disk i/o error' (virtualized block devices) a few times before returning None, so a one-shot EIO doesn't push callers onto the fail-closed unknown-mode branch. The rollup's write-lock machinery, checkpoint-strategy changes, and repair serialization are intentionally NOT included — superseded by PRs #84277 and #69609, or wrong-direction per the POSIX lock-cancellation findings (#71724 lineage). --- hermes_state.py | 136 +++++++++++++++++++++++-- tests/test_state_db_notadb_selfheal.py | 125 +++++++++++++++++++++++ 2 files changed, 250 insertions(+), 11 deletions(-) create mode 100644 tests/test_state_db_notadb_selfheal.py diff --git a/hermes_state.py b/hermes_state.py index e7f8e7aff0..0b7a8810f5 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -730,20 +730,40 @@ def _on_disk_journal_mode(conn: sqlite3.Connection) -> Optional[str]: Returns the mode string (e.g. ``"wal"``, ``"delete"``), or ``None`` if the value cannot be determined (new DB, or PRAGMA read failed). + + A PRAGMA read can fail transiently with ``disk i/o error`` on + virtualized block devices (XFS on cloud hosts). Treating that as + "mode unknown" pushes callers onto their fail-closed unknown-mode + branch even though the on-disk mode is perfectly readable a few + milliseconds later. Retry the read a few times before giving up: + transient EIO clears, deterministic unsupported-filesystem errors do + not. ``None`` is still returned on final failure so the caller's + existing "unknown → refuse to downgrade" logic applies. """ - try: - row = conn.execute("PRAGMA journal_mode").fetchone() - except sqlite3.OperationalError: - return None - if row is None: - return None - mode = row[0] - if isinstance(mode, bytes): # defensive: sqlite3 occasionally returns bytes + last_exc: Optional[Exception] = None + for _ in range(4): try: - mode = mode.decode("ascii") - except UnicodeDecodeError: + row = conn.execute("PRAGMA journal_mode").fetchone() + except sqlite3.OperationalError as exc: + last_exc = exc + if "disk i/o error" not in str(exc).lower(): + return None + time.sleep(0.05) + continue + if row is None: return None - return str(mode).strip().lower() if mode is not None else None + mode = row[0] + if isinstance(mode, bytes): # defensive: sqlite3 occasionally returns bytes + try: + mode = mode.decode("ascii") + except UnicodeDecodeError: + return None + return str(mode).strip().lower() if mode is not None else None + if last_exc is not None: + logger.debug( + "_on_disk_journal_mode: retries exhausted on disk read (%s)", last_exc + ) + return None def _apply_wal_size_limit(conn: sqlite3.Connection) -> None: @@ -1411,6 +1431,21 @@ def is_malformed_db_error(exc: BaseException) -> bool: return any(marker in str(exc).lower() for marker in _MALFORMED_SCHEMA_MARKERS) +def _is_not_a_database_error(exc: BaseException) -> bool: + """True if *exc* is SQLite's 'file is not a database' error. + + Raised when a connection's backing file is not a SQLite database — the + runtime connection-corruption class: a sibling process (forked curator + agent, external repair pass) replaced/truncated the file out from under + the live connection. The file on disk may be perfectly healthy; the + CONNECTION is broken. Distinct from the malformed-schema class: the fix + is a reconnect, not schema surgery. + """ + if not isinstance(exc, sqlite3.DatabaseError): + return False + return "file is not a database" in str(exc).lower() + + # Markers that mean the host filesystem cannot accept another write. Kept as # plain substrings so OSError, sqlite3.OperationalError, and wrapped RPC # error strings all match the same helper. @@ -2824,6 +2859,13 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) # in place at most once per SessionDB instance so a genuinely # unrecoverable database can't put writers into a rebuild loop. self._fts_runtime_rebuild_attempted = False + # One-shot guard for the runtime connection-reopen recovery on the + # write path. A connection whose backing file was replaced/truncated + # by a sibling process surfaces as "file is not a database" on every + # write; we close and reopen the connection at most once per + # SessionDB instance so a genuinely unrecoverable database can't put + # writers into a reconnect loop. + self._notadb_reconnect_attempted = False # One-shot guard for the usermerge-floor config write on the # incremental FTS merge cadence (see _merge_fts_incrementally). self._fts_usermerge_floor_applied = False @@ -3593,6 +3635,20 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) except sqlite3.DatabaseError as exc: if _is_no_more_rows(exc) and self._sleep_before_write_retry(deadline, patience_s): continue + # Runtime connection-corruption self-heal: a connection whose + # backing file was replaced/truncated by a sibling process + # (e.g. a forked curator agent inheriting and closing the + # write fd, or an external repair pass) surfaces as "file is + # not a database" on EVERY subsequent write. Without a + # reconnect branch the gateway wedges permanently: every + # transcript/routing write raises, messages stay in memory, + # and swap grows without bound until the process is killed. + # Close the broken connection, reopen the DB file, and retry + # the write once. + if _is_not_a_database_error(exc): + if not self._reconnect_after_notadb(): + raise + continue # Corrupt FTS shadow tables make every write raise the # malformed/corrupt error class through the FTS sync triggers # while the canonical messages table is intact. Recover here, @@ -3643,6 +3699,64 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) time.sleep(min(jitter, max(deadline - now, 0.001))) return True + def _reconnect_after_notadb(self) -> bool: + """Close the corrupted write connection and reopen state.db. + + Returns True when the connection was successfully replaced and the + failed write should be retried. Mirrors the constructor's + ``_connect_and_init`` so WAL/schema reconciliation runs on the fresh + connection. Never raises — logs and returns False on failure so the + original error propagates. + + One-shot per instance: a genuinely unrecoverable database must not + put writers into a reconnect loop that pins CPU on every write. + """ + if self._notadb_reconnect_attempted: + return False + self._notadb_reconnect_attempted = True + logger.warning( + "state.db connection reported 'file is not a database' — closing " + "and reopening the connection to self-heal (one-shot)." + ) + try: + with self._lock: + if self._conn is not None: + try: + self._conn.close() + except Exception: + pass + self._conn = None + new_conn = _connect_tracked_db( + str(self.db_path), + tracking_path=self.db_path, + check_same_thread=False, + timeout=1.0, + isolation_level=None, + ) + new_conn.row_factory = sqlite3.Row + # Publish BEFORE schema init: _init_schema/_reconcile_columns + # operate on self._conn, not on the local variable. + self._conn = new_conn + self._wal_active = ( + apply_wal_with_fallback(new_conn, db_label="state.db") + == "wal" + ) + apply_database_pragmas(new_conn, db_label="state.db") + new_conn.execute("PRAGMA foreign_keys=ON") + self._fts_cjk_loaded = load_fts5_cjk_extension(new_conn) + self._init_schema() + except Exception as exc: + logger.error( + "state.db reconnect after 'file is not a database' failed (%s); " + "the database may need the full offline repair path.", + exc, + ) + return False + logger.warning( + "state.db connection reopened successfully; retrying the failed write." + ) + return True + @staticmethod def _is_fts_write_corruption_error(exc: sqlite3.DatabaseError) -> bool: """True for the error class a corrupt FTS index raises on writes. diff --git a/tests/test_state_db_notadb_selfheal.py b/tests/test_state_db_notadb_selfheal.py new file mode 100644 index 0000000000..b30027044a --- /dev/null +++ b/tests/test_state_db_notadb_selfheal.py @@ -0,0 +1,125 @@ +"""Tests for the state.db runtime connection self-heal (PR #82280 remainder). + +Covers the two independently-valuable pieces salvaged from the state.db +hardening rollup: + +* one-shot reconnect when a live write connection reports + ``file is not a database`` (backing file replaced/truncated by a sibling + process — the connection is broken, the on-disk file may be healthy); +* 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. +""" + +import sqlite3 +from unittest.mock import MagicMock + +import pytest + +from hermes_state import SessionDB, _is_not_a_database_error, _on_disk_journal_mode + + +class _NotADbOnce: + """Connection proxy that raises 'file is not a database' on execute.""" + + def __init__(self, real_conn): + self._real = real_conn + + def execute(self, *args, **kwargs): + raise sqlite3.DatabaseError("file is not a database") + + def __getattr__(self, name): + return getattr(self._real, name) + + +class TestIsNotADatabaseError: + def test_matches_sqlite_message(self): + assert _is_not_a_database_error( + sqlite3.DatabaseError("file is not a database") + ) + + def test_rejects_other_database_errors(self): + assert not _is_not_a_database_error( + sqlite3.DatabaseError("database disk image is malformed") + ) + + def test_rejects_non_sqlite_exceptions(self): + assert not _is_not_a_database_error(ValueError("file is not a database")) + + +class TestReconnectAfterNotADb: + def test_write_self_heals_when_connection_breaks(self, tmp_path): + """A broken connection over a healthy file reconnects and retries.""" + db = SessionDB(db_path=tmp_path / "state.db") + try: + db.create_session(session_id="s1", source="cli", model="test") + # Simulate the runtime corruption class: the connection starts + # raising 'file is not a database' while the on-disk file is + # perfectly healthy (sibling replaced/truncated the old inode). + db._conn = _NotADbOnce(db._conn) + + db.create_session(session_id="s2", source="cli", model="test") + + assert db._notadb_reconnect_attempted is True + assert db.get_session("s2") is not None + # The pre-existing row survived (same on-disk file). + assert db.get_session("s1") is not None + finally: + db.close() + + def test_reconnect_is_one_shot(self, tmp_path): + """A second 'file is not a database' propagates instead of looping.""" + db = SessionDB(db_path=tmp_path / "state.db") + try: + db._notadb_reconnect_attempted = True + db._conn = _NotADbOnce(db._conn) + with pytest.raises(sqlite3.DatabaseError, match="not a database"): + db.create_session(session_id="s3", source="cli", model="test") + finally: + db._conn = None + db.close() + + def test_failed_reconnect_returns_false_and_original_error_propagates( + self, tmp_path, monkeypatch + ): + """If the reopen itself fails, the original write error surfaces.""" + db = SessionDB(db_path=tmp_path / "state.db") + try: + monkeypatch.setattr( + "hermes_state._connect_tracked_db", + MagicMock(side_effect=sqlite3.DatabaseError("file is not a database")), + ) + db._conn = _NotADbOnce(db._conn) + with pytest.raises(sqlite3.DatabaseError, match="not a database"): + db.create_session(session_id="s4", source="cli", model="test") + assert db._notadb_reconnect_attempted is True + finally: + db._conn = None + db.close() + + +class TestOnDiskJournalModeEioRetry: + def _conn_raising_then(self, failures, result_rows): + conn = MagicMock() + cursor = MagicMock() + cursor.fetchone.return_value = result_rows + conn.execute.side_effect = list(failures) + [cursor] + return conn + + def test_transient_eio_clears_on_retry(self): + conn = self._conn_raising_then( + [sqlite3.OperationalError("disk i/o error")] * 2, ("wal",) + ) + assert _on_disk_journal_mode(conn) == "wal" + + def test_persistent_eio_returns_none(self): + conn = MagicMock() + conn.execute.side_effect = sqlite3.OperationalError("disk i/o error") + assert _on_disk_journal_mode(conn) is None + # Bounded: retried a handful of times, not forever. + assert conn.execute.call_count == 4 + + def test_non_eio_operational_error_fails_fast(self): + conn = MagicMock() + conn.execute.side_effect = sqlite3.OperationalError("database is locked") + assert _on_disk_journal_mode(conn) is None + assert conn.execute.call_count == 1