diff --git a/gateway/lifecycle_ledger.py b/gateway/lifecycle_ledger.py index 0950035939..28b04131af 100644 --- a/gateway/lifecycle_ledger.py +++ b/gateway/lifecycle_ledger.py @@ -41,6 +41,7 @@ from __future__ import annotations import json import logging import os +import sqlite3 import time from datetime import datetime, timezone from pathlib import Path @@ -50,6 +51,7 @@ logger = logging.getLogger(__name__) _LIFECYCLE_RELATIVE = ("state", "gateway.lifecycle.json") _EXIT_DIAG_RELATIVE = ("logs", "gateway-exit-diag.log") +_STATE_DB_RELATIVE = ("state.db",) # Heuristic OOM-suspicion thresholds applied to the last heartbeat's memory # sample. Deliberately conservative: this only annotates the report with a @@ -221,6 +223,44 @@ def detect_unclean_exit(home: Optional[Path] = None) -> Optional[Dict[str, Any]] return evidence +def check_state_db_integrity(home: Optional[Path] = None) -> str: + """Return ``"ok"``, ``"absent"``, or the first ``quick_check`` complaint. + + Called only after an unclean death, because that is when the store may + have been torn: a SIGKILL landing on a gateway mid-WAL-checkpoint can + leave half-written b-tree pages behind (see + ``_enforce_macos_synchronous_full`` in :mod:`hermes_state` — macOS + ``fsync`` guarantees neither data-on-platter nor write ordering). + + ``quick_check(1)`` stops at the first problem, so this costs ~2s on a + healthy 500MB store and less on a damaged one — cheap enough for a path + that runs at most once per unclean boot, far too expensive for every + boot. Corruption used to sit undetected for days (2026-08-26 → 08-30) + because nothing ever looked. + + The connection is opened normally, not read-only: a WAL store needs its + -shm sidecar for a read-only open. The PRAGMA itself writes nothing. + Never raises — this is forensics, not lifecycle. + """ + base = home if home is not None else _process_hermes_home() + path = base.joinpath(*_STATE_DB_RELATIVE) + if not path.exists(): + return "absent" + try: + # A WAL store cannot be opened read-only without its -shm sidecar, so + # open normally; quick_check itself writes nothing. + conn = sqlite3.connect(str(path)) + try: + row = conn.execute("PRAGMA quick_check(1)").fetchone() + finally: + conn.close() + except Exception as exc: # sqlite3.Error, OSError, anything + return f"check-failed: {exc}" + if not row or row[0] is None: + return "check-failed: no result" + return str(row[0]) + + def record_startup(home: Optional[Path] = None) -> Optional[Dict[str, Any]]: """Boot-time entry point: report any unclean previous exit, then claim the sentinel for the current life. @@ -233,6 +273,18 @@ def record_startup(home: Optional[Path] = None) -> Optional[Dict[str, Any]]: try: evidence = detect_unclean_exit(home) if evidence is not None: + # The death may have torn the store. This is the only moment + # we know to look, and looking is what turns a 3.5-day silent + # corruption into a startup warning. + verdict = check_state_db_integrity(home=home) + evidence["state_db_integrity"] = verdict + if verdict not in ("ok", "absent"): + logger.error( + "state.db FAILED integrity check after an unclean gateway " + "exit: %s — sessions may read as missing until it is " + "repaired. Run `hermes doctor`.", + verdict, + ) record = { "ts": datetime.now(timezone.utc).isoformat(), "tag": "gateway.previous_unclean_exit", diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 8d0fd26721..729f471ead 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -2363,23 +2363,79 @@ def _reap_unsupervised_gateway_orphans(extra_exclude: set | None = None) -> bool reaped = True # SIGTERM released the port in the field report but the orphan kept - # running until a follow-up SIGKILL — wait briefly, then force-kill - # any survivor so the replacement can bind the port cleanly. - deadline = time.monotonic() + 5.0 - survivors = list(orphans) - while survivors and time.monotonic() < deadline: - survivors = [p for p in survivors if _pid_exists(p)] - if survivors: - time.sleep(0.2) - for pid in survivors: - try: - os.kill(pid, getattr(signal, "SIGKILL", signal.SIGTERM)) - except (ProcessLookupError, PermissionError, OSError): - pass + # running until a follow-up SIGKILL — wait, then force-kill any survivor + # so the replacement can bind the port cleanly. + survivors = _await_gateway_exit(orphans, pid_exists=_pid_exists) + _force_kill_survivors(survivors) return reaped +# A retiring gateway runs a PASSIVE WAL checkpoint in ``SessionDB.close()``. +# On a large store with a WAL well past the 1000-page autocheckpoint threshold +# that does not finish in the 5s this used to allow, and the SIGKILL then +# landed mid-checkpoint — half-written b-tree pages, which is the 2026-08-31 +# ``state.db`` corruption. The outgoing gateway keeps serving while we wait, +# so a longer grace delays only the replacement's port bind, never traffic. +_ORPHAN_EXIT_GRACE_SECONDS = 30.0 +_ORPHAN_EXIT_POLL_SECONDS = 0.2 + + +def _await_gateway_exit( + pids, + *, + pid_exists, + sleep=None, + grace_s: float = _ORPHAN_EXIT_GRACE_SECONDS, + poll_s: float = _ORPHAN_EXIT_POLL_SECONDS, +): + """Wait up to *grace_s* for *pids* to exit; return those still alive. + + Polls rather than blocking so a process that exits early costs nothing. + ``pid_exists``/``sleep`` are injected so the wait is testable without + real processes. + """ + if sleep is None: + sleep = time.sleep + survivors = [p for p in pids] + for _ in range(max(1, int(grace_s / poll_s))): + survivors = [p for p in survivors if pid_exists(p)] + if not survivors: + break + sleep(poll_s) + else: + # Re-check after the LAST sleep. Without this a process that exits in + # the final interval is reported as a survivor and gets SIGKILL — + # usually a harmless ProcessLookupError, but a recycled PID would put + # that signal on an unrelated process. + survivors = [p for p in survivors if pid_exists(p)] + return survivors + + +def _force_kill_survivors(survivors, *, kill=None) -> None: + """SIGKILL processes that outlasted the grace period, loudly. + + A force-kill is the event that can tear the store, so it must leave a + trace: the 2026-08-31 incident had no record of which restart did it. + """ + if not survivors: + return + if kill is None: + kill = os.kill + for pid in survivors: + logger.warning( + "Gateway PID %s did not exit within %.0fs of SIGTERM — sending " + "SIGKILL. A kill during a WAL checkpoint can corrupt state.db; " + "the next start will run an integrity check.", + pid, + _ORPHAN_EXIT_GRACE_SECONDS, + ) + try: + kill(pid, getattr(signal, "SIGKILL", signal.SIGTERM)) + except (ProcessLookupError, PermissionError, OSError): + pass + + def stop_profile_gateway() -> bool: """Stop only the gateway for the current profile (HERMES_HOME-scoped). diff --git a/hermes_cli/web_routers/sessions.py b/hermes_cli/web_routers/sessions.py index 4dc7efb598..a4da40c3a1 100644 --- a/hermes_cli/web_routers/sessions.py +++ b/hermes_cli/web_routers/sessions.py @@ -15,6 +15,7 @@ the late-binding seam in :mod:`hermes_cli.web_deps` so tests that import asyncio # noqa: F401 — used by handlers import json import logging +import sqlite3 import time # noqa: F401 from typing import Any, Dict, List, Optional # noqa: F401 @@ -30,6 +31,7 @@ from hermes_cli.web_models import ( SessionPrune, SessionRename, ) +from hermes_state import is_malformed_db_error # Same logger the handlers used before extraction (identical logger object). _log = logging.getLogger("hermes_cli.web_server") @@ -51,6 +53,39 @@ _session_latest_descendant = late("_session_latest_descendant") _strip_session_list_rows = late("_strip_session_list_rows") +def _resolve_session_id(db, session_id: str) -> Optional[str]: + """Resolve *session_id*, distinguishing "absent" from "unreadable". + + A corrupt ``state.db`` does not raise on every read. The exact-match + lookup goes through the ``sessions`` primary-key index, so a damaged + index simply *misses* and returns None — indistinguishable from a + session that was never there. Only the prefix fallback scans the base + b-tree and raises ``database disk image is malformed``. + + Both outcomes used to end at ``404 Session not found``: one silently, one + as an unhandled 500 that the Desktop surfaced as "session unavailable". + Reporting a corrupt store as an empty one cost a day of looking at + session logic during the 2026-08-31 incident, so classify it here and say + what is actually wrong. + """ + try: + return db.resolve_session_id(session_id) + except sqlite3.DatabaseError as exc: + if not is_malformed_db_error(exc): + raise + _log.error( + "state.db is corrupt while resolving session %s: %s", session_id, exc + ) + raise HTTPException( + status_code=503, + detail=( + "Session store is corrupt (database disk image is malformed). " + "Sessions cannot be read until it is repaired — run " + "`hermes doctor` for diagnosis." + ), + ) from exc + + @list_router.get("/api/sessions") def get_sessions( # ``le=100`` caps the page size (idea from #39200): an unbounded limit @@ -562,7 +597,7 @@ async def get_session_stats(profile: Optional[str] = None): async def get_session_detail(session_id: str, profile: Optional[str] = None): db = _open_session_db_for_profile(profile, read_only=True) try: - sid = db.resolve_session_id(session_id) + sid = _resolve_session_id(db, session_id) session = db.get_session(sid) if sid else None if not session: raise HTTPException(status_code=404, detail="Session not found") @@ -622,7 +657,7 @@ async def get_session_messages( def _read(): db = _open_session_db_for_profile(profile, read_only=True) try: - sid = db.resolve_session_id(session_id) + sid = _resolve_session_id(db, session_id) if not sid: return None sid = db.resolve_resume_session_id(sid) @@ -697,7 +732,7 @@ async def delete_session_endpoint(session_id: str, profile: Optional[str] = None # leaves transient empty rows (reaped by empty-session hygiene) that # race the sidebar snapshot, which is exactly when this fired. Mirrors # the bulk-delete endpoint, which already treats ghost ids as success. - sid = db.resolve_session_id(session_id) + sid = _resolve_session_id(db, session_id) if not sid: return {"ok": True, "already_absent": True} db.delete_session(sid) @@ -767,7 +802,7 @@ async def rename_session_endpoint(session_id: str, body: SessionRename): """ db = _open_session_db_for_profile(body.profile, read_only=False) try: - sid = db.resolve_session_id(session_id) + sid = _resolve_session_id(db, session_id) if not sid: raise HTTPException(status_code=404, detail="Session not found") if ( @@ -815,7 +850,7 @@ async def export_session_endpoint(session_id: str, profile: Optional[str] = None def _prepare_export(): db = _open_session_db_for_profile(profile, read_only=True) try: - sid = db.resolve_session_id(session_id) + sid = _resolve_session_id(db, session_id) return (sid, db.get_session(sid)) if sid else None finally: db.close() diff --git a/tests/gateway/test_lifecycle_state_db_check.py b/tests/gateway/test_lifecycle_state_db_check.py new file mode 100644 index 0000000000..a1460ee6aa --- /dev/null +++ b/tests/gateway/test_lifecycle_state_db_check.py @@ -0,0 +1,124 @@ +"""An unclean gateway death must trigger a state.db integrity check. + +Regression for the 2026-08-31 incident. ``state.db`` was corrupt from +2026-08-26 evening (a SIGKILL landed on a gateway mid-WAL-checkpoint during a +``--replace`` restart storm), but nothing checked the file. The damage sat in +old, rarely-read session rows for 3.5 days until a Desktop read tripped over +it on 2026-08-30 17:15 and surfaced as "Session not found". + +``record_startup`` already detects the unclean exit and logs "SIGKILL / OOM / +VM death" — it just never looked at the database that death may have torn. +The check is gated on the unclean exit precisely because it costs ~2s on a +500MB store; a clean boot must not pay it. +""" +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path + +from gateway.lifecycle_ledger import ( + check_state_db_integrity, + get_lifecycle_sentinel_path, + record_startup, +) + +_DEAD_PID = 2 ** 22 + 12345 # beyond default pid_max; never alive + + +def _write_sentinel(home: Path, phase: str = "running") -> None: + path = get_lifecycle_sentinel_path(home) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps({ + "phase": phase, + "pid": _DEAD_PID, + "start_time": 1000.0, + "started_at": "2026-08-26T23:56:45+00:00", + }), + encoding="utf-8", + ) + + +def _make_state_db(home: Path, *, corrupt: bool) -> Path: + """Build a real SQLite file, optionally with a genuinely torn b-tree page.""" + path = home / "state.db" + conn = sqlite3.connect(path) + conn.execute("CREATE TABLE sessions (id INTEGER PRIMARY KEY, v TEXT)") + conn.executemany( + "INSERT INTO sessions (v) VALUES (?)", [(f"row-{i}" * 40,) for i in range(4000)] + ) + conn.commit() + conn.close() + if corrupt: + with open(path, "r+b") as handle: + handle.seek(4096 * 6) + handle.write(b"\xEF" * 4096) + return path + + +def _exit_diag_records(home: Path) -> list: + log = home / "logs" / "gateway-exit-diag.log" + if not log.exists(): + return [] + return [json.loads(line) for line in log.read_text().splitlines() if line.strip()] + + +# ── the checker itself ────────────────────────────────────────────────────── + + +def test_checker_passes_a_healthy_store(tmp_path: Path) -> None: + _make_state_db(tmp_path, corrupt=False) + assert check_state_db_integrity(home=tmp_path) == "ok" + + +def test_checker_reports_a_torn_btree_page(tmp_path: Path) -> None: + _make_state_db(tmp_path, corrupt=True) + verdict = check_state_db_integrity(home=tmp_path) + assert verdict != "ok" + assert "btreeInitPage" in verdict or "malformed" in verdict.lower() + + +def test_checker_tolerates_a_missing_store(tmp_path: Path) -> None: + assert check_state_db_integrity(home=tmp_path) == "absent" + + +# ── wiring into the unclean-exit path ─────────────────────────────────────── + + +def test_unclean_exit_records_the_corruption_verdict(tmp_path: Path) -> None: + _make_state_db(tmp_path, corrupt=True) + _write_sentinel(tmp_path) + + evidence = record_startup(home=tmp_path) + + assert evidence is not None + assert evidence["state_db_integrity"] != "ok" + record = _exit_diag_records(tmp_path)[0] + assert record["state_db_integrity"] != "ok" + + +def test_unclean_exit_on_a_healthy_store_records_ok(tmp_path: Path) -> None: + _make_state_db(tmp_path, corrupt=False) + _write_sentinel(tmp_path) + + evidence = record_startup(home=tmp_path) + + assert evidence is not None + assert evidence["state_db_integrity"] == "ok" + + +def test_clean_exit_does_not_pay_for_the_check(tmp_path: Path, monkeypatch) -> None: + """A clean boot must not scan the store — that is the whole cost gate.""" + _make_state_db(tmp_path, corrupt=True) + _write_sentinel(tmp_path, phase="exited") + + called = [] + import gateway.lifecycle_ledger as ledger + + monkeypatch.setattr( + ledger, "check_state_db_integrity", lambda **kw: called.append(1) or "ok" + ) + record_startup(home=tmp_path) + + assert not called, "integrity check ran on a clean boot" diff --git a/tests/gateway/test_orphan_exit_grace.py b/tests/gateway/test_orphan_exit_grace.py new file mode 100644 index 0000000000..5030e95758 --- /dev/null +++ b/tests/gateway/test_orphan_exit_grace.py @@ -0,0 +1,101 @@ +"""A retiring gateway must get long enough to finish its WAL checkpoint. + +Regression for the 2026-08-31 corruption. ``gateway run --replace`` reaped the +previous instance with SIGTERM, waited **5 seconds**, then SIGKILLed any +survivor. A gateway closing a 500MB WAL store runs a PASSIVE checkpoint in +``SessionDB.close()``; on a WAL that had grown to 3936 pages (4x the 1000-page +autocheckpoint threshold) that does not reliably finish in 5s. A SIGKILL +landing mid-checkpoint leaves half-written b-tree pages — macOS ``fsync`` +guarantees neither data-on-platter nor write ordering, which is exactly why +``hermes_state._enforce_macos_synchronous_full`` exists. + +The port-rebinding reason the 5s deadline was introduced still holds, so the +force-kill stays — it just must not fire on a process that is still shutting +down normally, and when it does fire it must say so. +""" +from __future__ import annotations + +import pytest + +from hermes_cli.gateway import ( + _ORPHAN_EXIT_GRACE_SECONDS, + _await_gateway_exit, +) + + +def _exits_after(polls: int): + """pid_exists() that reports the process gone after *polls* checks.""" + seen = {"n": 0} + + def pid_exists(_pid: int) -> bool: + seen["n"] += 1 + return seen["n"] <= polls + + return pid_exists + + +def test_a_slow_but_healthy_shutdown_is_not_force_killed() -> None: + """12s of checkpointing — well past the old 5s budget — must survive.""" + survivors = _await_gateway_exit( + [4242], + pid_exists=_exits_after(60), # 60 polls x 0.2s = 12s + sleep=lambda _s: None, + ) + assert survivors == [] + + +def test_a_truly_hung_process_is_still_reported_for_force_kill() -> None: + survivors = _await_gateway_exit( + [4242], + pid_exists=lambda _pid: True, + sleep=lambda _s: None, + ) + assert survivors == [4242] + + +def test_an_already_dead_process_returns_immediately() -> None: + slept = [] + survivors = _await_gateway_exit( + [4242], + pid_exists=lambda _pid: False, + sleep=lambda s: slept.append(s), + ) + assert survivors == [] + assert slept == [], "waited on a process that was already gone" + + +def test_the_grace_period_covers_a_large_wal_checkpoint() -> None: + """5s was the value that let the SIGKILL land mid-checkpoint.""" + assert _ORPHAN_EXIT_GRACE_SECONDS >= 20.0 + + +def test_force_kill_is_logged_so_the_next_incident_has_evidence(caplog) -> None: + import logging + + from hermes_cli import gateway as gw + + killed = [] + with caplog.at_level(logging.WARNING): + gw._force_kill_survivors([4242], kill=lambda pid, sig: killed.append(pid)) + + assert killed == [4242] + assert any("4242" in r.getMessage() for r in caplog.records), ( + "a force-kill left no trace in the log" + ) + + +def test_a_pid_that_vanishes_in_the_final_interval_is_not_killed() -> None: + """The last sleep must still be followed by a check. + + Without a re-check after the final sleep the process is reported as a + survivor and gets SIGKILL. That is normally a harmless + ProcessLookupError — but if the PID has already been recycled the signal + lands on an unrelated process. + """ + polls = int(_ORPHAN_EXIT_GRACE_SECONDS / 0.2) # every in-loop check says "alive" + survivors = _await_gateway_exit( + [4242], + pid_exists=_exits_after(polls), + sleep=lambda _s: None, + ) + assert survivors == [], "a process that exited during the last interval was killed" diff --git a/tests/test_session_detail_malformed_db.py b/tests/test_session_detail_malformed_db.py new file mode 100644 index 0000000000..16a3eb12a1 --- /dev/null +++ b/tests/test_session_detail_malformed_db.py @@ -0,0 +1,163 @@ +"""A corrupt state.db must not be reported as a missing session. + +Regression for the 2026-08-31 incident: ``state.db`` corruption (rowid out of +order in ``sessions``, wrong entry count in ``idx_messages_session_id``) made +``resolve_session_id`` raise ``sqlite3.DatabaseError: database disk image is +malformed``. The Desktop showed "Couldn't open this session - session +unavailable" and the main process logged ``404: {"detail":"Session not +found"}`` — a session that existed was reported as absent, which sent the +diagnosis in entirely the wrong direction for a day. + +A corrupt store is an unavailable store, not an empty one. +""" + +import sqlite3 + +import pytest +from fastapi import HTTPException + +from hermes_cli import web_server +from hermes_cli.web_routers import sessions as sessions_router + + +class _MalformedDB: + """SessionDB stand-in that fails the way a corrupt file really fails.""" + + def __init__(self): + self.closed = False + + def resolve_session_id(self, session_id): + raise sqlite3.DatabaseError("database disk image is malformed") + + def get_session(self, sid): # pragma: no cover - never reached + raise AssertionError("resolve_session_id should have raised first") + + def close(self): + self.closed = True + + +@pytest.fixture +def malformed_db(monkeypatch): + db = _MalformedDB() + monkeypatch.setattr( + web_server, "_open_session_db_for_profile", lambda profile, *, read_only: db + ) + return db + + +@pytest.mark.asyncio +async def test_corrupt_db_is_not_reported_as_missing_session(malformed_db): + with pytest.raises(HTTPException) as excinfo: + await sessions_router.get_session_detail("20260830_180820_744f05") + + assert excinfo.value.status_code != 404, ( + "corruption reported as 'Session not found' - this is the bug" + ) + assert excinfo.value.status_code == 503 + + +@pytest.mark.asyncio +async def test_corrupt_db_detail_names_the_real_cause(malformed_db): + with pytest.raises(HTTPException) as excinfo: + await sessions_router.get_session_detail("20260830_180820_744f05") + + detail = str(excinfo.value.detail).lower() + assert "corrupt" in detail or "malformed" in detail + + +@pytest.mark.asyncio +async def test_db_is_closed_even_when_corruption_raises(malformed_db): + with pytest.raises(HTTPException): + await sessions_router.get_session_detail("20260830_180820_744f05") + + assert malformed_db.closed, "connection leaked on the corruption path" + + +# ── The same lookup runs in four more handlers ────────────────────────────── +# +# delete is the dangerous one: an unresolvable id is treated as idempotent +# success ({"ok": True, "already_absent": True}), so a corrupt store made +# DELETE report that it had removed a session that is still on disk — and the +# Desktop drops the sidebar row on that answer. + + +@pytest.mark.asyncio +async def test_messages_endpoint_reports_corruption(malformed_db): + with pytest.raises(HTTPException) as excinfo: + await sessions_router.get_session_messages( + "20260830_180820_744f05", None, None, 0, None, False + ) + assert excinfo.value.status_code == 503 + + +@pytest.mark.asyncio +async def test_delete_does_not_claim_success_on_a_corrupt_store(malformed_db): + with pytest.raises(HTTPException) as excinfo: + await sessions_router.delete_session_endpoint("20260830_180820_744f05") + assert excinfo.value.status_code == 503 + + +@pytest.mark.asyncio +async def test_rename_endpoint_reports_corruption(malformed_db): + from hermes_cli.web_models import SessionRename + + with pytest.raises(HTTPException) as excinfo: + await sessions_router.rename_session_endpoint( + "20260830_180820_744f05", SessionRename(title="neu") + ) + assert excinfo.value.status_code == 503 + + +@pytest.mark.asyncio +async def test_export_endpoint_reports_corruption(malformed_db): + with pytest.raises(HTTPException) as excinfo: + await sessions_router.export_session_endpoint("20260830_180820_744f05") + assert excinfo.value.status_code == 503 + + +# ── A genuinely absent session must still be a 404, not a 503 ────────────── + + +class _EmptyDB: + def __init__(self): + self.closed = False + + def resolve_session_id(self, session_id): + return None + + def get_session(self, sid): + return None + + def close(self): + self.closed = True + + +@pytest.mark.asyncio +async def test_absent_session_is_still_404(monkeypatch): + monkeypatch.setattr( + web_server, + "_open_session_db_for_profile", + lambda profile, *, read_only: _EmptyDB(), + ) + with pytest.raises(HTTPException) as excinfo: + await sessions_router.get_session_detail("does_not_exist") + assert excinfo.value.status_code == 404 + + +# ── Unrelated DatabaseErrors must not be relabelled as corruption ────────── + + +class _OtherErrorDB(_EmptyDB): + def resolve_session_id(self, session_id): + raise sqlite3.DatabaseError("some unrelated database failure") + + +@pytest.mark.asyncio +async def test_non_corruption_database_error_is_not_swallowed(monkeypatch): + monkeypatch.setattr( + web_server, + "_open_session_db_for_profile", + lambda profile, *, read_only: _OtherErrorDB(), + ) + with pytest.raises(sqlite3.DatabaseError): + await sessions_router.get_session_detail("20260830_180820_744f05")