hermes_state: opt-in in-place self-heal for a lost WAL generation (fixes #109687)

hermes_state.py

@@ -494,6 +494,12 @@ def __init__(self, db_path: Path = None, read_only: bool = False):
         # Read-open failure backoff is a TIMESTAMP, not a sticky bool: the likeliest trigger
         # is transient EMFILE, and a permanent flag would demote every reader forever.
         self._read_open_failed_at = 0.0
+        # Read-pool epoch marker, created when the handle adopts a new WAL generation
+        # (the #109687 self-heal): a reader checked out mid-heal holds a descriptor
+        # into the orphaned generation, and returning it to the pool would serve stale
+        # reads forever. Once the set exists, only connections in it (minted or
+        # verified after the heal) are usable; a pre-heal checkout is closed instead.
+        self._fresh_read_conns: Optional[set] = None
         self._wal_active, self._write_count = False, 0
         # File identity of the opened state.db, compared on every write so an out-of-band
         # replace cannot limp through in-place surgery (inode: mv/new-file; application_id: cp).
@@ -741,6 +747,8 @@ def _get_read_conn(self) -> Optional[sqlite3.Connection]:
             apply_database_pragmas(conn, db_label="state.db")
             if self._fts_cjk_loaded:  # registers in the connection, not the file: ro is fine
                 load_fts5_cjk_extension(conn)
+            if self._fresh_read_conns is not None:
+                self._fresh_read_conns.add(conn)
         except BaseException as exc:
             # A half-open connection (open ok, extension load failed) is a live tracked descriptor,
             # the leak shape this pool exists to fix; a stranded permit would shrink the read
@@ -778,10 +786,18 @@ def _checkout_read_conn(self) -> Optional[sqlite3.Connection]:
         A pool hit costs no permit (the connection already holds one)."""
         if not self._wal_active or self.read_only:
             return None
-        try:
-            return self._read_pool.get_nowait()
-        except queue.Empty:
-            return self._get_read_conn()
+        while True:
+            try:
+                conn = self._read_pool.get_nowait()
+            except queue.Empty:
+                return self._get_read_conn()
+            fresh = self._fresh_read_conns
+            if fresh is not None and conn not in fresh:
+                # Minted before a WAL-generation self-heal adopted the current
+                # generation: its descriptors name the orphaned one. Never reuse.
+                self._close_read_conn(conn)
+                continue
+            return conn
 
     @contextmanager
     def _read_ctx(self) -> Iterator[sqlite3.Connection]:
@@ -799,6 +815,8 @@ def _read_ctx(self) -> Iterator[sqlite3.Connection]:
                     if not self._read_conns_closed:
                         try:
                             self._read_pool.put_nowait(conn)
+                            if self._fresh_read_conns is not None:
+                                self._fresh_read_conns.add(conn)
                             returned = True
                         except queue.Full:
                             pass
@@ -1102,6 +1120,89 @@ def _wal_generation_was_lost(self) -> bool:
             self._db_sidecar_identity = current_identity
         return False
 
+    def _wal_self_heal_enabled(self) -> bool:
+        """Whether the operator opted into automatic WAL-generation recovery.
+
+        Default OFF: losing the -wal/-shm generation under a live writer is a
+        data-integrity event, and the fail-closed halt (capture + sticky
+        refusal) is the conservative posture (#109687 keeps it for unflagged
+        installs). An operator who accepts the trade-off -- captured frames are
+        preserved for inspection in the retired-wal artifact, then this handle
+        drops its orphaned descriptors and reopens onto the current generation
+        -- can enable ``database.wal_self_heal: true`` so a short-lived CLI
+        reader no longer bricks the gateway until a human intervenes.
+        """
+        try:
+            from hermes_cli.config import load_config_readonly
+            database = (load_config_readonly() or {}).get("database", {})
+            raw = database.get("wal_self_heal", False) if isinstance(database, dict) else False
+            return bool(raw)
+        except Exception:
+            return False
+
+    def _try_heal_lost_wal_generation(self) -> bool:
+        """Recover this handle from a lost WAL/SHM generation in place (#109687).
+
+        Automates the exact remediation the capture machinery already defines:
+        (1) capture the retired frames durably (the unlinked WAL inode dies
+        with this process's last descriptor -- preserved frames outlive it),
+        (2) close this handle's descriptors with SQLite's close-time
+        checkpoint disabled so retired frames cannot be written over the newer
+        generation (Python < 3.12 cannot disable it, so no heal is attempted
+        there: the pin path stays), (3) reopen through
+        ``refuse_deleted_wal_generation`` exactly like a fresh process -- if
+        any OTHER process still holds an orphaned sidecar, the guard refuses
+        and the heal fails closed -- and (4) adopt the current generation's
+        identity. The sticky halt flags are cleared only after the reopen
+        succeeded. Caller holds ``self._lock``.
+        """
+        if not _close_time_checkpoint_configurable():
+            return False
+        conn = self._conn
+        if conn is None:
+            return False
+        try:
+            artifact = self._capture_retired_generation("self-heal")
+        except RetiredGenerationCaptureError:
+            return False  # capture retries at close(); the handle stays halted
+        # Close every descriptor this handle holds onto the orphaned generation
+        # BEFORE reopening: the refuse-guard scans /proc for deleted sidecars,
+        # and this process's own fds would trip it (self-pid is not exempt there).
+        if not self._disable_close_time_checkpoint():
+            return False  # cannot suppress the close-time checkpoint: closing here
+            # could write the retired frames over the newer generation; fail closed
+        self._conn = None
+        self._close_connection_quietly(conn)
+        while self._evict_one_idle_read_conn():
+            pass
+        # From here on only descriptors opened against the CURRENT generation may
+        # serve reads: a reader checked out mid-heal still holds an orphaned fd and
+        # must never re-enter the pool.
+        self._fresh_read_conns = set()
+        try:
+            refuse_deleted_wal_generation(self.db_path)
+        except DeletedWalGenerationError:
+            # Another process still holds an orphaned sidecar: this handle alone
+            # cannot clear the box. Fail closed with the canonical refusal.
+            logger.error(_DELETED_WAL_GENERATION_MSG)
+            raise
+        except Exception as exc:
+            raise DeletedWalGenerationError(
+                f"state.db WAL-generation self-heal for {self.db_path} failed at the "
+                f"pre-reopen guard: {exc}"
+            ) from exc
+        self._conn = self._open_writer_conn()
+        # Adopt the current generation and clear the sticky halt: writes resume.
+        self._record_db_file_identity()
+        self._db_wal_generation_lost = False
+        logger.warning(
+            "state.db %s lost its WAL/SHM generation under a live writer (#109687); the retired frames "
+            "are preserved at %s, this handle dropped its orphaned descriptors and reopened onto the "
+            "current generation — writes resumed without operator intervention (database.wal_self_heal).",
+            self.db_path, artifact,
+        )
+        return True
+
     def _halt_if_db_generation_changed(self) -> None:
         """Stop writes (logging once) when the file was replaced or its WAL/SHM generation
         is gone: never run in-file repair on a new generation, never keep committing on a
@@ -1124,6 +1225,8 @@ def _halt_if_db_generation_changed(self) -> None:
                     "Could not capture the retired WAL generation of %s at halt: %s. close() retries "
                     "the capture and refuses to settle without it.", self.db_path, exc,
                 )
+            if self._wal_self_heal_enabled() and self._try_heal_lost_wal_generation():
+                return  # healed: sticky flags cleared, current generation adopted
             logger.error(_DELETED_WAL_GENERATION_MSG)
             raise DeletedWalGenerationError(_DELETED_WAL_GENERATION_MSG)
 

tests/hermes_state/test_wal_generation_self_heal.py

@@ -0,0 +1,213 @@
+"""Automatic in-place recovery for a lost WAL/SHM generation (#109687).
+
+A short-lived reader's clean close unlinks the ``-wal``/``-shm`` sidecars
+while a long-lived gateway writer still holds them, leaving the writer
+committing into an unlinked inode. The fail-closed halt (#105670 capture +
+sticky refusal) is the default; these tests cover the opt-in
+``database.wal_self_heal`` path that lets a flagged writer recover without
+operator intervention: capture the retired frames, drop the orphaned
+descriptors (close-time checkpoint disabled), reopen through the refuse
+guard, adopt the current generation.
+
+The real unlink/adopt flow is Linux-only (Windows cannot unlink a held
+sidecar; the /proc deleted-fd scan is Linux-only) — same gate as the guard
+tests in test_deleted_wal_generation_guard.py. The gating, refusal and
+read-pool-epoch logic is exercised cross-platform by arming a faked loss.
+"""
+
+import contextlib
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+import hermes_state
+from hermes_state import DeletedWalGenerationError, SessionDB
+from tests.hermes_state._wal_generation_harness import (
+    integrity_ok_path, lose_sidecars, make_db, message_count,
+    pin_wal, require_wal, write_second_generation,
+)
+
+
+@pytest.fixture
+def force_wal(monkeypatch):
+    pin_wal(monkeypatch)
+
+
+@pytest.fixture
+def self_heal_enabled(monkeypatch):
+    """Opt the runtime into the self-heal without a config file."""
+    monkeypatch.setattr(SessionDB, "_wal_self_heal_enabled", lambda self: True)
+
+
+@pytest.fixture
+def armable_loss(monkeypatch):
+    """Lose the generation only once the test ARMS it (``db._armed_loss = True``),
+    so seeding writes through the same handle still succeed. Works on Windows,
+    where unlinking an open -wal raises PermissionError."""
+    monkeypatch.setattr(
+        SessionDB, "_wal_generation_was_lost",
+        lambda self: bool(getattr(self, "_armed_loss", False)),
+    )
+
+
+def test_flag_defaults_off(tmp_path, force_wal):
+    db = make_db(tmp_path / "state.db", "s", "x")
+    try:
+        assert db._wal_self_heal_enabled() is False
+    finally:
+        db.close()
+
+
+def test_heal_off_keeps_fail_closed_halt(tmp_path, force_wal, armable_loss, monkeypatch):
+    """Unflagged: the halt stays exactly as shipped — capture + sticky refusal.
+    The heal is not even attempted: the flag gates the call itself."""
+    attempted = []
+    monkeypatch.setattr(
+        SessionDB, "_try_heal_lost_wal_generation",
+        lambda self: attempted.append(1) or False,
+    )
+    db = make_db(tmp_path / "state.db", "s", "held")
+    try:
+        db._armed_loss = True
+        with pytest.raises(DeletedWalGenerationError):
+            db.append_message("s", role="user", content="post-loss")
+        assert attempted == []
+        assert db._db_wal_generation_lost is True
+    finally:
+        with contextlib.suppress(Exception):
+            db.close()
+
+
+def test_heal_attempted_then_sticky_when_it_cannot_heal(
+    tmp_path, force_wal, self_heal_enabled, armable_loss,
+):
+    """Flagged: the heal IS attempted; when it cannot complete (capture fails
+    here — this process holds no retired inode to preserve) the halt is exactly
+    the shipped sticky refusal."""
+    db = make_db(tmp_path / "state.db", "s", "held")
+    try:
+        db._armed_loss = True
+        with pytest.raises(DeletedWalGenerationError):
+            db.append_message("s", role="user", content="post-loss")
+        assert db._db_wal_generation_lost is True  # heal failed closed, sticky
+    finally:
+        with contextlib.suppress(Exception):
+            db.close()
+
+
+def test_heal_refuses_when_guard_detects_foreign_holders(
+    tmp_path, force_wal, self_heal_enabled, armable_loss, monkeypatch,
+):
+    """_try_heal_lost_wal_generation raises DeletedWalGenerationError when the
+    pre-reopen guard finds a foreign holder: the heal never mints a second WAL
+    while another live process holds the orphan. Capture is stubbed (the frames
+    are faked-preserved) so the flow reaches the guard."""
+    monkeypatch.setattr(
+        SessionDB, "_capture_retired_generation",
+        lambda self, trigger: Path(self.db_path).parent / "stubbed-retired-wal",
+    )
+    db = make_db(tmp_path / "state.db", "s", "held")
+    try:
+        db._armed_loss = True
+        # Patch the guard AFTER open: _connect_and_init calls it too.
+        def _fake_refuse(db_path):
+            raise DeletedWalGenerationError("foreign holder")
+        monkeypatch.setattr(hermes_state, "refuse_deleted_wal_generation", _fake_refuse)
+        with pytest.raises(DeletedWalGenerationError):
+            db.append_message("s", role="user", content="must not land")
+        assert db._db_wal_generation_lost is True  # sticky: still halted
+        assert db._conn is None  # the heal closed the orphaned writer descriptor
+    finally:
+        with contextlib.suppress(Exception):
+            db.close()
+
+
+def test_heal_adopts_current_generation_and_resumes(
+    tmp_path, force_wal, self_heal_enabled, armable_loss, monkeypatch,
+):
+    """The full in-place heal: capture (stubbed), close the orphaned
+    descriptors, reopen through the real guard (clean — no holders), adopt the
+    CURRENT generation, clear the sticky halt, resume writes."""
+    monkeypatch.setattr(
+        SessionDB, "_capture_retired_generation",
+        lambda self, trigger: Path(self.db_path).parent / "stubbed-retired-wal",
+    )
+    path = tmp_path / "state.db"
+    db = make_db(path, "s", "held")
+    try:
+        db._armed_loss = True
+        db.append_message("s", role="user", content="post-heal turn")
+        # Writes resumed: sticky flag cleared, identity re-adopted.
+        assert db._db_wal_generation_lost is False
+        assert db._db_sidecar_identity is not None
+        contents = [m["content"] for m in db.get_messages("s")]
+        assert "post-heal turn" in contents
+        # Subsequent writes keep landing (no one-shot).
+        db.append_message("s", role="user", content="post-heal turn 2")
+        assert "post-heal turn 2" in [m["content"] for m in db.get_messages("s")]
+    finally:
+        db.close()
+
+
+def test_pool_evicts_pre_heal_read_conns(tmp_path, force_wal, self_heal_enabled):
+    """A read connection minted BEFORE the heal never serves reads afterwards:
+    the epoch marker closes it at checkout instead of reusing the orphaned fd."""
+    path = tmp_path / "state.db"
+    db = make_db(path, "s", "held")
+    require_wal(db)
+    try:
+        # Mint a read conn under the pre-heal epoch, return it to the pool.
+        stale = db._checkout_read_conn()
+        assert stale is not None
+        with db._read_conns_lock:
+            db._read_pool.put_nowait(stale)
+            assert db._read_conns_closed is False
+        # Simulate the heal having adopted a new generation.
+        db._fresh_read_conns = set()
+        conn = db._checkout_read_conn()
+        assert conn is not None
+        assert conn is not stale  # stale evicted, a fresh one minted
+    finally:
+        db.close()
+
+
+@pytest.mark.skipif(
+    not sys.platform.startswith("linux"),
+    reason="real sidecar unlink + /proc deleted-fd scan is Linux-only",
+)
+def test_heal_recovers_writer_without_operator(tmp_path, force_wal, self_heal_enabled):
+    """Flagged, real flow: after the sidecars are unlinked and a second
+    generation exists on the path, the next write captures the retired frames,
+    drops the orphaned descriptors and resumes on the CURRENT generation."""
+    path = tmp_path / "state.db"
+    db = make_db(path, "s", "held")
+    wal = require_wal(db)
+    inode_before = wal.stat().st_ino
+    try:
+        lose_sidecars(path, rename=False)
+        # A separate (non-hermes) opener mints a fresh generation on the path,
+        # exactly like the field incident's short-lived CLI reader.
+        write_second_generation(path, 1)
+
+        # First write after the loss: the heal must run end to end.
+        db.append_message("s", role="user", content="post-heal turn")
+
+        # Writes resumed: sticky flag cleared, retired frames preserved.
+        assert db._db_wal_generation_lost is False
+        assert db._retired_generation_capture is not None
+        artifact = Path(db._retired_generation_capture)
+        assert artifact.exists()
+        wal_now = Path(os.fspath(path) + "-wal")
+        if wal_now.exists():
+            assert wal_now.stat().st_ino != inode_before
+        # The healed write is visible on the current generation.
+        contents = [m["content"] for m in db.get_messages("s")]
+        assert "post-heal turn" in contents
+        # And it survives a fresh process reading the file.
+        rows = message_count(path)
+        assert rows >= 2  # held + gen2 row + post-heal turn (>= the originals)
+        assert integrity_ok_path(path)
+    finally:
+        db.close()