fix(state-db): report corruption instead of "session not found", detect it early

Re-applied onto 3aee29089 after `hermes update` reset main to origin/main.

1. web_routers/sessions.py: _resolve_session_id() classifies malformed-DB
   errors via the existing is_malformed_db_error() and raises 503 at all five
   call sites. delete_session_endpoint was the worst — an unresolvable id
   counted as idempotent success, so DELETE reported it had removed a session
   that was still on disk.
2. gateway/lifecycle_ledger.py: check_state_db_integrity() runs PRAGMA
   quick_check(1) on the unclean-exit path only (~2s on 500MB) and records the
   verdict into gateway-exit-diag.log. The 2026-08-31 corruption sat undetected
   for 3.5 days because nothing ever looked.
3. hermes_cli/gateway.py: `gateway run --replace` gave the outgoing gateway 5s
   before SIGKILL; SessionDB.close() runs a PASSIVE WAL checkpoint that does
   not finish in 5s on a WAL 4x past the autocheckpoint threshold, and a kill
   mid-checkpoint tears b-tree pages. Grace raised to 30s via a testable
   _await_gateway_exit() that also re-checks after the final sleep (a PID
   exiting in the last interval must not be SIGKILLed — PID-reuse hazard).

NOT added: wal_checkpoint(TRUNCATE) at shutdown — removed upstream in #45383
because a TRUNCATE reset races the live writer and tears b-tree pages.

Adversarial review: Codex gpt-5.6-sol, 9.0/10 across three groups, no must-fix.
This commit is contained in:
Nio Thomas
2026-08-31 11:58:58 +02:00
committed by Teknium
parent 037a73fa23
commit ba7743b076
6 changed files with 549 additions and 18 deletions
+52
View File
@@ -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",
+69 -13
View File
@@ -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).
+40 -5
View File
@@ -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()
@@ -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"
+101
View File
@@ -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"
+163
View File
@@ -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")