"""state.db repair, backup and writability preflight (split from hermes_state). Patchable helpers are looked up as module globals at call time, so tests patch ``hermes_state_repair.``. """ from __future__ import annotations import contextlib import datetime import hashlib import itertools import json import logging import os import re import shutil import sqlite3 import stat import time from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from hermes_constants import get_hermes_home from hermes_startup_watchdog import report_startup_progress from hermes_state_holders import read_only_db_uri from hermes_state_common import ( _acquire_db_flock, _clear_lock_holder_record, _describe_lock_holder, _read_lock_holder_record, is_advisory_lock_contention, ) # Log-record parity with the origin module (caplog tests pin "hermes_state"). logger = logging.getLogger("hermes_state") _REPAIR_LOCK_POLL_SECONDS = 0.1 # Snapshot copies are data transfer, not locking: bounded separately at 10 MiB/s (historical two-minute floor). _REPAIR_SNAPSHOT_MIN_THROUGHPUT_BYTES_PER_SECOND = 10 * 1024 * 1024 # ── Repair-loop bounding + dead-backup hygiene (#86747) ───────────────────── ``_claim_repair_attempt`` # above is an in-memory set: it bounds the loop only WITHIN one process. A corruption class the strategies # cannot heal (b-tree page damage) failed repair on EVERY process start, and each pass took a fresh ~900MB # forensic backup — 105 attempts / 89GB of identical dead copies in the reporting install. Two persistent # bounds fix the class: * a sidecar attempt ledger (``.repair-attempts.json``) that refuses further # surgery after ``_MAX_PERSISTENT_REPAIR_ATTEMPTS`` failures on the SAME damaged file (fingerprint = size + # a bounded content sample; any successful repair or replacement changes it and resets the count); * backup # dedupe + a retention cap in ``_backup_db_file`` — an identical damaged file is never copied twice, and # only the newest ``_MAX_MALFORMED_BACKUPS`` forensic copies are kept. _MAX_PERSISTENT_REPAIR_ATTEMPTS = 3 _MAX_MALFORMED_BACKUPS = 3 # Sidecars copied with a damaged DB and pruned with it. ``-journal``: DELETE mode (the NFS/SMB/FUSE/ZFS and # WAL-reset-bug fallback) leaves a hot journal whenever a transaction was open; without it the copy cannot roll back. _DB_SIDECAR_SUFFIXES = ("-wal", "-shm", "-journal") # Head/tail bytes sampled by ``_db_fingerprint``: changes on any real repair/truncation/restore, O(1) on a multi-GB file. _FINGERPRINT_SAMPLE_BYTES = 65536 # Header ranges that move on ordinary commits, not on repair — file change counter (24-27), version-valid-for # (92-95) — masked out of the sample. A malformed-SCHEMA DB still accepts writes and DELETE mode writes the main # file directly, so without the mask any live write re-keys the ledger and the repair budget resets to 1 # forever. The page-1 sqlite_master b-tree (repair identity) sits after byte 100. _FINGERPRINT_VOLATILE_HEADER_RANGES = ((24, 28), (92, 96)) # Headroom for the forensic backup (a full raw copy; a repair loop on a large state.db is a disk amplifier). # Proportional, not a flat multi-GB floor: a refused backup is a HARD STOP, and a big reserve would make repair # never run on small volumes. # The backup is a full raw copy of the damaged DB (plus its -wal/-shm sidecars), so a repair loop on a large # state.db is a disk amplifier: the reporting incident wrote ~98MB every ~10s until the volume was nearly # full, which would have taken down every agent on the host. Require the copy itself plus a small slice of # the volume, clamped to a modest floor. See #69603. _REPAIR_BACKUP_MIN_FREE_BYTES = 256 * 1024 * 1024 # 256 MiB absolute floor _REPAIR_BACKUP_FREE_FRACTION = 0.02 # plus 2% of the volume _FTS_TABLES = ("messages_fts", "messages_fts_trigram", "messages_fts_cjk") _MANUAL_RECOVER_HINT = ("Free disk space, then retry (or recover manually with " "`hermes sessions recover --source {db_path} --inspect-only` first).") def _sidecars(db_path: Path): """The three SQLite sidecar paths for *db_path* (present or not).""" return (db_path.with_name(db_path.name + suffix) for suffix in _DB_SIDECAR_SUFFIXES) def _read_offline(db_path: Path, what: str, reader) -> Optional[str]: """Run *reader()* under ``hermes_cli.sqlite_safe_read.offline_file_access``. ``close()`` on ANY raw descriptor cancels every POSIX advisory lock this process holds on the file, including a peer connection's RESERVED lock (``sqlite_safe_read`` rule 1), so a raw read is only safe with no live connection; ``None`` when that makes it unsafe or the file is unreadable. Scaffold/embed installs without hermes_cli have no tracked connections.""" try: from hermes_cli.sqlite_safe_read import LiveConnectionError, offline_file_access except ImportError: offline_file_access, LiveConnectionError = (lambda _p, **_k: contextlib.nullcontext()), OSError try: with offline_file_access(db_path, what=what): return reader() except (LiveConnectionError, OSError): return None def _claim_repair_attempt(db_path: Path) -> bool: """Claim the one-shot per-process repair attempt for *db_path*: True for the first caller, False afterwards (bounds the repair/reopen loop and stops concurrent callers racing surgery on one file).""" from hermes_state import _repair_attempt_lock, _repair_attempted_paths with _repair_attempt_lock: if str(db_path) in _repair_attempted_paths: return False _repair_attempted_paths.add(str(db_path)) return True def _open_lock_file(db_path: Path, suffix: str, what: str, tail: str): """Open ``.`` for locking; on failure warn and return None.""" lock_path = db_path.with_name(db_path.name + suffix) try: lock_path.parent.mkdir(parents=True, exist_ok=True) return lock_path, lock_path.open("a+b") except OSError as exc: logger.warning(f"Could not open state.db {what} lock %s (%s) — {tail}", lock_path, exc) return lock_path, None def _msvcrt_lock(handle, flag_name: str) -> None: import msvcrt handle.seek(0) msvcrt.locking(handle.fileno(), getattr(msvcrt, flag_name), 1) # type: ignore[attr-defined] def _try_lock_nonblocking(handle) -> None: """Take the advisory lock on *handle* without waiting (raises on contention).""" from hermes_state import _IS_WINDOWS if _IS_WINDOWS: _msvcrt_lock(handle, "LK_NBLCK") else: import fcntl fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) def _release_lock_handle(handle, *, clear_record: bool = False) -> None: """Drop the advisory lock on *handle* (best effort) and close it.""" from hermes_state import _IS_WINDOWS with contextlib.closing(handle), contextlib.suppress(OSError): # best-effort release; always close if _IS_WINDOWS: _msvcrt_lock(handle, "LK_UNLCK") else: import fcntl if clear_record: _clear_lock_holder_record(handle) fcntl.flock(handle.fileno(), fcntl.LOCK_UN) def _acquire_repair_lock_windows(lock_path: Path, handle, timeout: float): """Windows counterpart of ``_acquire_db_flock``: True / False (timed out) / None (non-contention error).""" deadline = time.monotonic() + timeout while True: try: _try_lock_nonblocking(handle) return True except (BlockingIOError, OSError) as exc: if not is_advisory_lock_contention(exc): logger.warning("Could not acquire state.db repair lock %s (%s) — skipping schema surgery on a " "non-contention error.", lock_path, exc) return None if time.monotonic() >= deadline: return False time.sleep(_REPAIR_LOCK_POLL_SECONDS) @contextlib.contextmanager def _cross_process_repair_lock(db_path: Path): """Serialize state.db schema surgery across processes. Yields True when this process holds the repair lock, False when the bounded acquire timed out or the lock file could not be opened; on False the caller must NOT do surgery (unlocked surgery IS the interleaving this prevents). ``flock``: the kernel drops it when the holder dies (a pidfile would wedge every future repair); a forked child inheriting the fd is the exception, so the acquire records pid + start time and breaks a provably dead holder's lock. Bounded because a live repairer can sit in ``VACUUM`` for minutes. An unopenable lock file (no space/inodes/descriptors) fails closed too: a sibling that opened ITS handle before the disk filled may be inside surgery. The acquire is still bounded because a *live* repairer can legitimately sit in ``VACUUM`` for minutes on a large DB, and an unbounded wait would hang the caller's open with no traceback (the failure shape of #36644). """ from hermes_state import _IS_WINDOWS, _REPAIR_LOCK_TIMEOUT_SECONDS lock_path, handle = _open_lock_file( db_path, ".repair.lock", "repair", "skipping schema surgery rather than running it without cross-process authority.") if handle is None: yield False return acquired = False try: if _IS_WINDOWS: acquired = _acquire_repair_lock_windows(lock_path, handle, _REPAIR_LOCK_TIMEOUT_SECONDS) else: acquired, handle = _acquire_db_flock(str(lock_path), handle, _REPAIR_LOCK_TIMEOUT_SECONDS, _REPAIR_LOCK_POLL_SECONDS, "state.db repair lock") if acquired is None: acquired = False # non-contention failure already logged with its errno elif not acquired: logger.warning("state.db repair lock %s held by another process for more than %.0fs — skipping schema " "surgery in this process to avoid racing the repairer. Recorded holder: %s.", lock_path, _REPAIR_LOCK_TIMEOUT_SECONDS, _describe_lock_holder(None if _IS_WINDOWS else _read_lock_holder_record(handle))) yield acquired finally: if acquired: _release_lock_handle(handle, clear_record=True) else: handle.close() def _try_acquire_auto_maintenance_lock(db_path: Path) -> Optional[Any]: """Non-blocking cross-process lock for one auto-maintenance pass (None = skip the pass: otherwise two startups both pass the interval check and the second prunes a row the first has only just closed recoverably).""" _lock_path, handle = _open_lock_file(db_path, ".auto-maintenance.lock", "auto-maintenance", "skipping automatic maintenance.") try: if handle is not None: _try_lock_nonblocking(handle) return handle except (BlockingIOError, OSError): handle.close() return None _release_auto_maintenance_lock = _release_lock_handle # release a _try_acquire_auto_maintenance_lock handle def _bump_schema_cookie(conn: sqlite3.Connection) -> None: """Increment the schema cookie after direct ``sqlite_master`` surgery. Ordinary DDL bumps it and peers compare it before running a prepared statement (how they discard a cached schema); ``writable_schema=ON`` edits do NOT, so live connections would keep firing triggers into ``messages_fts*`` shadow tables that no longer exist. Best-effort, never raises.""" try: current = conn.execute("PRAGMA schema_version").fetchone()[0] # Wrap within SQLite's 32-bit signed range; peers compare for equality. conn.execute(f"PRAGMA schema_version={(int(current) + 1) & 0x7FFFFFFF}") except (sqlite3.DatabaseError, TypeError, IndexError) as exc: logger.warning("Could not bump state.db schema cookie: %s", exc) def _mask_volatile_header(head: bytes) -> bytes: """Zero the commit-counter fields so ordinary writes don't re-key the ledger.""" if len(head) < 96: return head buf = bytearray(head) for start, end in _FINGERPRINT_VOLATILE_HEADER_RANGES: buf[start:end] = b"\x00" * (end - start) return bytes(buf) def _repair_backup_headroom_bytes(total_bytes: int) -> int: """Free space required *beyond* the copy itself, for a volume of *total_bytes*.""" return max(_REPAIR_BACKUP_MIN_FREE_BYTES, int(total_bytes * _REPAIR_BACKUP_FREE_FRACTION)) def _disk_budget(db_path: Path, refusal: str) -> "Tuple[Optional[str], int, int, int]": """``(error, bundle_bytes, free_bytes, headroom_bytes)`` for *db_path*'s volume (main file plus every PRESENT sidecar); *error* is set (and the sizes zero) on stat()/disk_usage() failure. Fails CLOSED: the nearly-full volume these guards exist for is exactly where they are most likely to fail.""" try: need = db_path.stat().st_size + sum(p.stat().st_size for p in _sidecars(db_path) if p.exists()) usage = shutil.disk_usage(db_path.parent) except OSError as exc: return (f"could not determine free space on {db_path.parent} ({exc}); refusing the {refusal} rather than risk " "filling the volume"), 0, 0, 0 return None, need, usage.free, _repair_backup_headroom_bytes(usage.total) def _repair_scratch_space_error(db_path: Path) -> Optional[str]: """Return an error unless snapshot, VACUUM and promotion can fit safely.""" error, snapshot_bytes, free, headroom = _disk_budget(db_path, "repair snapshot") if error is not None: return error # VACUUM on the staged DB may need up to 2x the database size (SQLite docs); the same reserve then covers # transactional promotion into the live DB. if free >= snapshot_bytes + (2 * snapshot_bytes) + headroom: return None return (f"only {free / 1e9:.2f}GB free on {db_path.parent}; the repair snapshot needs up to " f"{snapshot_bytes / 1e9:.2f}GB, VACUUM may need another {(2 * snapshot_bytes) / 1e9:.2f}GB, and " f"{headroom / 1e9:.2f}GB must remain as headroom. Free disk space, then retry.") def _backup_free_space_error(db_path: Path) -> Optional[str]: """Disk guard for the forensic copy: reason to refuse, or None. A full raw copy on a nearly-full volume (which a preceding repair loop may itself have caused) can finish off the disk and every process on the machine.""" hint = _MANUAL_RECOVER_HINT.format(db_path=db_path) error, need, free, headroom = _disk_budget(db_path, "forensic copy") if error is not None: return f"{error}. {hint}" if free - need >= headroom: return None return (f"only {free / 1e9:.2f}GB free on {db_path.parent}; copying the damaged DB needs {need / 1e9:.2f}GB and must " f"leave {headroom / 1e9:.2f}GB headroom. {hint}") def _repair_snapshot_timeout_seconds(source_path: Path) -> float: """Bound one SQLite snapshot by source size incl. sidecars (a WAL can hold committed rows not yet in the main file), so a healthy large-database copy is not cut off by the repair-lock timeout.""" from hermes_state import _REPAIR_LOCK_TIMEOUT_SECONDS source_bytes = 0 for candidate in (source_path, *_sidecars(source_path)): with contextlib.suppress(FileNotFoundError): # a sidecar may vanish mid-walk source_bytes += candidate.stat().st_size return max(_REPAIR_LOCK_TIMEOUT_SECONDS, source_bytes / _REPAIR_SNAPSHOT_MIN_THROUGHPUT_BYTES_PER_SECOND) def _repair_failure_consumes_attempt(exc: BaseException) -> bool: """Whether a pre-strategy SQLite failure proves deterministic corruption. Lock contention, timeouts, disk-full and I/O failures are environmental — a retry may succeed, so they must not burn the repair ledger. Only SQLite's corruption/image result codes prove deterministic damage.""" if not isinstance(exc, sqlite3.DatabaseError): return False if isinstance(error_code := getattr(exc, "sqlite_errorcode", None), int): # Extended result codes keep the primary code in the low byte. return (error_code & 0xFF) in (sqlite3.SQLITE_CORRUPT, sqlite3.SQLITE_NOTADB) # Older sqlite3 without result-code attributes: narrow message match only, never turning generic # "disk is full"/"readonly" into permanent failures. return any(m in str(exc).lower() for m in ("file is not a database", "database disk image is malformed")) def _repair_ledger_path(db_path: Path) -> Path: return db_path.with_name(db_path.name + ".repair-attempts.json") def _db_fingerprint(db_path: Path) -> "Optional[str]": """Cheap identity for a damaged DB file: size + a bounded content sample. EXCLUDES mtime: a malformed-schema DB still accepts writes, so live writers, checkpoints and the strategies move mtime between passes; keyed on mtime every pass looked like a NEW file and the attempt counter reset forever. Hashing a multi-GB file per open is the cost this ledger avoids, so sample the head/tail slices any real repair, truncation or restore must change. ``None`` = identity unavailable (:func:`_read_offline`; a live peer is expected here since this runs BEFORE ``_backup_db_file``'s live-connection guard). Callers MUST NOT substitute a differently-shaped key: the ledger compares for equality, so alternating shapes never match and the unbounded loop returns.""" def _sample() -> str: st = db_path.stat() with open(db_path, "rb") as fh: head = fh.read(_FINGERPRINT_SAMPLE_BYTES) fh.seek(max(0, st.st_size - _FINGERPRINT_SAMPLE_BYTES)) tail = fh.read(_FINGERPRINT_SAMPLE_BYTES) if st.st_size > _FINGERPRINT_SAMPLE_BYTES else b"" return f"{st.st_size}:{hashlib.sha256(_mask_volatile_header(head) + tail).hexdigest()[:32]}" return _read_offline(db_path, "fingerprint", _sample) def _backup_content_identity(db_path: Path) -> "Optional[str]": """Recovery-image identity for forensic-backup dedupe: whole file + sidecars. A DIFFERENT relation from :func:`_db_fingerprint` (never conflate them): the fingerprint answers "same repair epoch?" from head+tail only, and a live writer can commit into an *interior* page while preserving size and both 64 KiB slices — reusing a backup on that basis hands the operator a snapshot predating real user data. A forensic copy must claim byte identity, so this digests the ENTIRE main file plus every present sidecar (the WAL can hold uncheckpointed frames). ``None`` when a live connection makes the read unsafe (:func:`_read_offline`) — the caller then takes a fresh backup, never a false reuse. See #87409. """ def _digest() -> str: hasher = hashlib.sha256() members = [("main", db_path), *((sfx, p) for sfx, p in zip(_DB_SIDECAR_SUFFIXES, _sidecars(db_path)) if p.exists())] for label, path in members: # Length-delimit every member (main file included) so the concatenation is prefix-free; otherwise a # main-file tail could coincide with a main+sidecar split and dedupe two images together. hasher.update(f"\0{label}:{path.stat().st_size}\0".encode()) with open(path, "rb") as fh: for chunk in iter(lambda: fh.read(1024 * 1024), b""): hasher.update(chunk) return hasher.hexdigest() return _read_offline(db_path, "backup-identity", _digest) def _read_repair_ledger(db_path: Path) -> "Dict[str, Any]": with contextlib.suppress(OSError, ValueError): raw = json.loads(_repair_ledger_path(db_path).read_text(encoding="utf-8")) return raw if isinstance(raw, dict) else {} return {} def _persistent_repair_attempts_exhausted(db_path: Path) -> bool: """Whether *db_path* has already burned its cross-restart repair budget. True only when the ledger records ``_MAX_PERSISTENT_REPAIR_ATTEMPTS`` failures against the CURRENT fingerprint. Never raises; a missing/corrupt ledger or unstatable DB reads as "not exhausted" (the in-process claim and cross-process lock still bound one run). Fingerprint unavailable (live connection) -> compare the SIZE the ledger recorded, otherwise a peer connection hides an exhausted budget on every pass.""" ledger = _read_repair_ledger(db_path) recorded = ledger.get("fingerprint") fp = _db_fingerprint(db_path) try: # size is the one key component that needs no raw read same = recorded == fp if fp is not None else ( isinstance(recorded, str) and recorded.startswith(f"{db_path.stat().st_size}:")) except OSError: return False return same and int(ledger.get("failed_attempts", 0)) >= _MAX_PERSISTENT_REPAIR_ATTEMPTS def _persistent_repair_exhausted_error(db_path: Path) -> str: """The stable operator-facing diagnostic for an exhausted repair budget. The ``hermes`` commands carry the profile selector: a bare ``hermes`` follows ``active_profile`` (#105887).""" from hermes_constants import profile_cli_selector profile_arg = profile_cli_selector() return (f"automatic repair has already failed {_MAX_PERSISTENT_REPAIR_ATTEMPTS} times on this exact file — the " f"corruption is beyond the schema/FTS repair strategies (likely b-tree page damage). Manual recovery " f"required: restore a backup, or salvage with `hermes {profile_arg}sessions recover --source {db_path} " f"--inspect-only`, then (if it reports recoverable) `hermes {profile_arg}sessions recover --source {db_path} " f"--output recovered-state.db` (recovery snapshots the damaged file first, then runs the page-level " f"`.recover` lane on the copy; do NOT point a raw `sqlite3` shell at the live database). " f"Delete {_repair_ledger_path(db_path).name} to force another automatic attempt.") def _record_repair_outcome(db_path: Path, *, repaired: bool, fingerprint: "Optional[str]" = None) -> None: """Update the persistent attempt ledger after a repair pass. Never raises. Keys on the post-attempt fingerprint (what the NEXT exhaustion probe sees). If a live connection makes it unavailable, keep the recorded key and still increment — dropping the pass lets a peer reset the budget every time. Never write a differently shaped key.""" ledger_path = _repair_ledger_path(db_path) try: if repaired: ledger_path.unlink(missing_ok=True) return ledger = _read_repair_ledger(db_path) recorded = ledger.get("fingerprint") fp = fingerprint if fingerprint is not None else _db_fingerprint(db_path) if fp is None: if not isinstance(recorded, str): # No prior key to extend and no safe way to mint one; the in-process claim and cross-process lock # still bound this run. return fp = recorded attempts = int(ledger.get("failed_attempts", 0)) + 1 if recorded == fp else 1 stamp = datetime.datetime.now().isoformat(timespec="seconds") ledger_path.write_text(json.dumps({"fingerprint": fp, "failed_attempts": attempts, "last_attempt": stamp}), encoding="utf-8") except Exception as exc: # pragma: no cover - best effort logger.warning("Could not update state.db repair ledger: %s", exc) def _existing_malformed_backups(db_path: Path) -> "List[Path]": """Timestamped forensic backups of *db_path*, newest first.""" prefix = f"{db_path.name}.malformed-backup-" try: found = [p for p in db_path.parent.iterdir() if p.name.startswith(prefix) and not p.name.endswith(_DB_SIDECAR_SUFFIXES)] except OSError: return [] return sorted(found, key=lambda p: p.name, reverse=True) def _prune_malformed_backups(db_path: Path, keep: int = _MAX_MALFORMED_BACKUPS) -> None: """Delete all but the *keep* newest forensic backups (and sidecars).""" for stale in _existing_malformed_backups(db_path)[keep:]: for victim in (stale, *_sidecars(stale)): try: victim.unlink(missing_ok=True) except OSError as exc: # pragma: no cover - best effort logger.warning("Could not prune stale DB backup %s: %s", victim, exc) def _publish_backup_bundle(db_path: Path, staging: Path, backup_path: Path) -> None: """Copy DB + sidecars to *staging* names, then rename each into place. ORDER MATTERS: the main DB name is the bundle's commit marker (what ``_existing_malformed_backups`` counts), so sidecars publish FIRST and the main DB LAST — a failure partway never leaves a countable backup over a missing sidecar. On failure, staging files AND anything promoted are removed.""" main = (db_path, staging, backup_path) sidecars = [(sidecar, staging.with_name(staging.name + suffix), backup_path.with_name(backup_path.name + suffix)) for suffix, sidecar in zip(_DB_SIDECAR_SUFFIXES, _sidecars(db_path)) if sidecar.exists()] published: "List[Path]" = [] try: for src, staged, _dst in (main, *sidecars): shutil.copy2(src, staged) for _src, staged, dst in (*sidecars, main): os.replace(staged, dst) published.append(dst) except Exception: for victim in (*(staged for _s, staged, _d in (*sidecars, main)), *published): with contextlib.suppress(OSError): # best effort; a failure here is never the caller's error victim.unlink(missing_ok=True) raise def _backup_db_file(db_path: Path) -> "Tuple[Optional[Path], Optional[str]]": """Raw-copy a (possibly malformed) DB plus sidecars to a timestamped backup. Raw bytes on purpose: the DB won't open cleanly, so preserve them exactly for forensics. Returns ``(backup_path, None)`` or ``(None, reason)``; repair treats a refused backup as a HARD STOP because the bundle is the recovery path when every strategy fails. Refuses while a connection to this DB is live in the process: the raw read would ``close()`` a descriptor and cancel that connection's POSIX advisory locks (``hermes_cli.sqlite_safe_read``) — real case: one SessionDB enters repair while the gateway holds others. Dedupe: reuse the newest backup when byte-identical to the current recovery image (``_backup_content_identity`` — NOT mtime, NOT ``_db_fingerprint``); a repair loop once re-copied the same bytes on every restart. Staging names live OUTSIDE the ``.malformed-backup-`` prefix: inside it they count as a backup, sort NEWEST (prune kept partials, deleted intact copies) and dedupe could return one with no real forensic copy on disk. Refusal reasons (``_backup_free_space_error`` / ``_MANUAL_RECOVER_HINT``) point operators at the safe lane, `hermes sessions recover --source --inspect-only`, never at a raw sqlite3 shell on the live file. See #69603. """ with contextlib.suppress(ImportError): # scaffold/embed installs without hermes_cli track no connections from hermes_cli.sqlite_safe_read import has_live_connection if has_live_connection(db_path): reason = (f"a connection to {db_path} is still open in this process; raw-copying it would cancel that " "connection's POSIX advisory locks. Close all SessionDB handles first.") logger.error("Refusing to raw-copy %s for backup: %s", db_path, reason) return None, reason stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") backup_path = db_path.with_name(f"{db_path.name}.malformed-backup-{stamp}") for seq in itertools.count(1): # same-second collision must not overwrite the earlier forensic copy if not backup_path.exists(): break backup_path = db_path.with_name(f"{db_path.name}.malformed-backup-{stamp}_{seq}") try: # Sweep staging debris from an interrupted pass BEFORE the dedupe (it is byte-identical to the damaged # DB, so dedupe would hand it back as a backup). Also the old ``.incomplete`` spelling, which # prefix-matches as a backup, sorts NEWEST and would otherwise survive prune forever. for pattern in (f"{db_path.name}.backup-staging-*", f"{db_path.name}.malformed-backup-*.incomplete*"): for old in db_path.parent.glob(pattern): with contextlib.suppress(OSError): old.unlink(missing_ok=True) with contextlib.suppress(OSError): # Hash the source only when there is a candidate: hashing a multi-GB file before copying it is waste. if newest := _existing_malformed_backups(db_path)[:1]: src_id = _backup_content_identity(db_path) if src_id is not None and _backup_content_identity(newest[0]) == src_id: logger.info("Reusing existing forensic backup %s (identical to the damaged DB).", newest[0]) return newest[0], None if (reason := _backup_free_space_error(db_path)) is not None: logger.error("Refusing forensic backup of %s: %s", db_path, reason) return None, reason _publish_backup_bundle(db_path, db_path.with_name(f"{db_path.name}.backup-staging-{stamp}"), backup_path) # Retention cap (#86747): keep only the newest few forensic copies. _prune_malformed_backups(db_path) return backup_path, None except Exception as exc: # pragma: no cover - best effort logger.warning("Could not back up malformed DB %s: %s", db_path, exc) return None, f"backup copy failed: {exc}" def preflight_db_writability(db_path: Path, *, db_label: str = "state.db") -> None: """Refuse-or-repair read-only DB files BEFORE the first connection opens. A stray read-only ``state.db`` / ``-wal`` / ``-shm`` (sudo run, restored backup, copied dotfiles) otherwise surfaces as an opaque "attempt to write a readonly database" inside ``_init_schema``, and the obvious wrong "fix" (deleting the ``-wal``) loses committed transactions. ``chmod u+rw`` repair only inside the Hermes home tree (Hermes owns those files; ``chmod`` fails on files the user doesn't own, bounding the repair exactly); otherwise fail fast naming the file and command. Never deletes/truncates a WAL sidecar — once writable, the normal open checkpoints it. ``:memory:``/``file:`` skipped. Shared with ``kanban_db``. Port of Kilo-Org/kilocode#12508's startup preflight. This preflight: """ if str(db_path) == ":memory:" or str(db_path).startswith("file:"): return home: Optional[Path] = None with contextlib.suppress(Exception): # pragma: no cover - defensive home = Path(get_hermes_home()).resolve() # SQLite needs a writable directory in every journal mode (WAL/SHM sidecars, or the DELETE-mode journal). sidecars = (db_path.with_name(db_path.name + "-wal"), db_path.with_name(db_path.name + "-shm")) for p, is_dir in [(db_path.parent, True), *((p, False) for p in (db_path, *sidecars) if p.is_file())]: if (is_dir and not p.is_dir()) or os.access(p, os.R_OK | os.W_OK): continue x = "x" if is_dir else "" in_scope = False with contextlib.suppress(OSError, ValueError): in_scope = home is not None and p.resolve().is_relative_to(home) if in_scope: os.chmod(p, p.stat().st_mode | stat.S_IRUSR | stat.S_IWUSR | (stat.S_IXUSR if is_dir else 0)) if in_scope and os.access(p, os.R_OK | os.W_OK): logger.info("%s preflight: repaired read-only %s (chmod u+rw%s)", db_label, p, x) continue wal_note = (" Do NOT delete the -wal file — it contains committed data that " "will be merged into the database once it is writable." if p.name.endswith("-wal") else "") raise sqlite3.OperationalError( f"{db_label} is not writable: {'directory' if is_dir else 'file'} {p} is read-only for this user. Hermes " f"needs read-write access to open the database. Fix with: chmod u+rw{x} '{p}' (files owned by another " f"user may need sudo/chown).{wal_note}") def _connect_repair_durable(db_path: Path, *, timeout: float = 5.0) -> sqlite3.Connection: """``sqlite3.connect`` for the repair/probe paths, with macOS write barriers. These paths bypass ``SessionDB``/:func:`apply_wal_with_fallback`, so they inherited ``synchronous=NORMAL`` and no ``checkpoint_fullfsync`` — on Darwin an interrupted ``REINDEX``/``VACUUM``/``writable_schema`` rewrite leaves half-written b-tree pages. Autocommit (``isolation_level=None``): DDL and ``VACUUM`` are illegal inside an implicit transaction. Barriers are best-effort: on a malformed schema even ``PRAGMA synchronous=FULL`` raises, so whole-file rewrites call :func:`_reapply_durability_barriers` once the schema parses again. Tracked (:func:`hermes_cli.sqlite_safe_read.connect_tracked`) because repair connections hold the strongest locks in the process (``locking_mode=EXCLUSIVE``, ``BEGIN IMMEDIATE``); an untracked fd let the byte-level probes ``open()``/``close()`` the live file, which cancels every POSIX advisory lock this process holds on it (sqlite.org/howtocorrupt §2.2) and lets an external writer commit mid-repair (#63386). """ from hermes_cli.sqlite_safe_read import connect_tracked conn = connect_tracked(db_path, tracking_path=db_path, timeout=timeout, isolation_level=None) _reapply_durability_barriers(conn) return conn def _repair_conn(db_path: Path, *, timeout: float = 5.0): """A :func:`_connect_repair_durable` connection as a context manager, closed on exit.""" return contextlib.closing(_connect_repair_durable(db_path, timeout=timeout)) def _reapply_durability_barriers(conn: sqlite3.Connection) -> bool: """Best-effort (re)application of the macOS write barriers; True if accepted. Call before ``VACUUM``/``REINDEX`` once the schema parses: a connection opened on a malformed schema could not take them at open time. Never raises.""" from hermes_state_wal import _apply_macos_checkpoint_barrier, _enforce_macos_synchronous_full try: _apply_macos_checkpoint_barrier(conn) _enforce_macos_synchronous_full(conn) return True except Exception: # schema still unparseable, or anything else — pragmas cannot be set yet return False def apply_durability_barriers(conn: sqlite3.Connection) -> bool: """Durability barriers for guest users of ``state.db`` that must inherit its owner's journal mode. Also applies the configured ``database.synchronous`` level, a per-connection pragma that otherwise only rides on the journal-mode setup path guests must not run.""" from hermes_state_wal import _apply_synchronous_pragma ok = _reapply_durability_barriers(conn) with contextlib.suppress(Exception): from hermes_cli.config import cfg_get, load_config_readonly # local: avoids an import cycle if (raw_synchronous := cfg_get(load_config_readonly(), "database", "synchronous", default=None)) is not None: _apply_synchronous_pragma(conn, raw_synchronous, db_label="state.db (guest)") return ok def _close_unpinned(conn: sqlite3.Connection) -> None: """Leave EXCLUSIVE locking mode (so the file is never left pinned) and close.""" with contextlib.suppress(Exception): conn.execute("PRAGMA locking_mode=NORMAL") conn.close() def _open_exclusive(db_path: Path, begin: str) -> sqlite3.Connection: """Zero-timeout connection holding ``locking_mode=EXCLUSIVE`` after a rolled-back *begin*; closed (unpinned) and re-raised when exclusion cannot be taken.""" conn = _connect_repair_durable(db_path, timeout=0.0) try: for statement in ("PRAGMA locking_mode=EXCLUSIVE", begin, "ROLLBACK"): conn.execute(statement) except BaseException: _close_unpinned(conn) raise return conn @contextlib.contextmanager def _exclusive_repair_db_guard(db_path: Path): """Yield ``(conn, None)`` — one live connection that excludes writers for repair surgery — or ``(None, exc)`` when exclusion could not be taken. ``locking_mode=EXCLUSIVE`` retains file-level exclusion after the short ``BEGIN EXCLUSIVE`` is rolled back; the rollback is essential because ``Connection.backup`` uses this connection as *source* and later as the promotion *destination*, both transaction-free. It stays open across the snapshot -> strategies -> promotion window so no writer can commit a change promotion would overwrite. Existing readers make acquisition fail rather than being disturbed (fail closed). Timeout 0: the cross-process lock already serializes repairers, and a partial repair is less safe than "stop the gateway and retry".""" try: guard = _open_exclusive(db_path, "BEGIN EXCLUSIVE") except (sqlite3.Error, OSError) as exc: yield None, exc return try: yield guard, None finally: # Releasing the exclusive locks before close keeps a close-time checkpoint from being mistaken for a # repair write by callers that reopen immediately. _close_unpinned(guard) def _copy_database_snapshot(source_path: Path, destination_path: Path, *, source_connection: Optional[sqlite3.Connection] = None, destination_connection: Optional[sqlite3.Connection] = None) -> None: """Copy one complete SQLite snapshot without replacing either file inode: the online backup API folds committed WAL frames into the source snapshot and writes the destination in one transaction (rolled back if interrupted), so ``state.db`` is never swapped out from under handles that refer to it.""" # Deadline first: a sidecar vanishing mid-stat must not leak a just-opened descriptor. deadline_seconds = _repair_snapshot_timeout_seconds(source_path) deadline = time.monotonic() + deadline_seconds def _check_deadline(_status: int, _remaining: int, _total: int) -> None: if time.monotonic() >= deadline: raise TimeoutError(f"timed out copying SQLite repair snapshot after {deadline_seconds:.0f}s") with contextlib.ExitStack() as owned: # closes only connections opened here, destination first source = source_connection or owned.enter_context(_repair_conn(source_path)) if destination_connection is not None and destination_connection.in_transaction: # sqlite3_backup needs a transaction-free destination (the guard holds exclusion via locking_mode). raise sqlite3.ProgrammingError("SQLite repair backup destination has an active transaction") destination = destination_connection or owned.enter_context(_repair_conn(destination_path)) source.backup(destination, pages=256, progress=_check_deadline, sleep=_REPAIR_LOCK_POLL_SECONDS) def _schema_not_built(exc: BaseException) -> bool: """``no such table/column``: FTS5 / core tables not created yet (brand new file mid-init).""" return any(m in str(exc).lower() for m in ("no such table", "no such column")) # Hermes-owned FTS5 objects: the virtual tables and their shadow b-trees. Full-matched, so a # user-created lookalike (``archive_fts_data``) is not swept into the rebuildable set. _FTS_OBJECT_RE = re.compile( r"messages_fts(_trigram|_cjk)?(_data|_idx|_content|_docsize|_config|_segdir|_segments)?" ) _INTEGRITY_TREE_RE = re.compile(r"\bTree (\d+)\b") _INTEGRITY_MISSING_INDEX_RE = re.compile(r"missing from index (\S+)") def integrity_damage_is_structural(integrity_lines, master_rows) -> bool: """True when any damaged object named by ``PRAGMA integrity_check`` output lies outside the FTS shadow set: a ``Tree N`` id mapped through ``sqlite_master.rootpage``, an index in ``row N missing from index X``, or the file's own freelist. Unparseable lines are not counted (the FTS wording stays, which is incomplete rather than wrong). #88587: an FTS rebuild cannot repair a canonical b-tree, and the ``.malformed-backup`` it leaves behind is a snapshot of the same damage.""" name_by_rootpage = {int(rp): name for rp, _type, name in master_rows if rp} for line in integrity_lines: text = str(line) if text.startswith("Freelist"): return True tree = _INTEGRITY_TREE_RE.search(text) if tree: name = name_by_rootpage.get(int(tree.group(1)), "") if name and not _FTS_OBJECT_RE.fullmatch(name): return True missing = _INTEGRITY_MISSING_INDEX_RE.search(text) if missing and not _FTS_OBJECT_RE.fullmatch(missing.group(1)): return True return False def state_db_has_structural_damage(db_path: Path) -> bool: """Read-only ``integrity_check`` + ``sqlite_master`` rootpage map on a fresh connection; ``integrity_damage_is_structural`` over the result. A check that RAISES instead of reporting (a torn page under the walk) is structural too: no FTS-only fixture does that while ``messages``/``sessions`` read cleanly, and the FTS rebuild ladder cannot help. Cannot-open / locked stays False so the caller keeps the FTS path.""" try: conn = sqlite3.connect(read_only_db_uri(db_path), uri=True, timeout=1.0) except sqlite3.Error: return False try: master_rows = [tuple(r) for r in conn.execute( "SELECT rootpage, type, name FROM sqlite_master WHERE rootpage > 0").fetchall()] lines = [str(r[0]) for r in conn.execute("PRAGMA integrity_check").fetchall()] except sqlite3.OperationalError: return False except sqlite3.DatabaseError: return True finally: conn.close() return integrity_damage_is_structural( itertools.chain.from_iterable(line.splitlines() for line in lines), master_rows) def _db_opens_cleanly(db_path: Path) -> Optional[str]: """Probe a DB on a fresh connection. Returns None if healthy, else a reason. Runs the first statement that trips the malformed-schema parse (``PRAGMA journal_mode``), ``integrity_check``, a ``sessions`` read, FTS5 MATCH probes and a rolled-back ``messages`` write — so FTS5 index corruption (reads and ``integrity_check`` pass, every ``INSERT INTO messages`` fails through the FTS triggers) is reported as unhealthy. See #50502. """ from hermes_state import SessionDB, load_fts5_cjk_extension # ── Strategy 0.5: rebuild stale B-tree indexes (#63386) ── PRAGMA integrity_check can report "wrong # # of entries in index" when a B-tree index (e.g. idx_sessions_handoff_state) falls out of sync with its # base table. REINDEX rewrites the index b-tree from the canonical table rows using the existing index # definition, fixing the mismatch without touching data or FTS schema. conn = _connect_repair_durable(db_path) try: with contextlib.closing(conn): # Best-effort tokenizer load: messages_fts_cjk needs cjk_unicode61 before any statement can touch it; # tokenizer absence must never classify as corruption. load_fts5_cjk_extension(conn) conn.execute("PRAGMA journal_mode").fetchone() rows = conn.execute("PRAGMA integrity_check").fetchall() problems = [str(r[0]) for r in rows if r and str(r[0]).lower() != "ok"] if problems: return "; ".join(problems[:3]) conn.execute("SELECT COUNT(*) FROM sessions").fetchone() # FTS5 read probe: partial shadow-table corruption makes MATCH/snippet/rank raise while integrity_check # reports healthy. MATCH '""' (empty phrase) parses, scans zero rows and exercises the shadow tables; # FTS5 rejects MATCH ''. for fts_table in _FTS_TABLES: try: conn.execute(f"SELECT 1 FROM {fts_table} WHERE {fts_table} MATCH '\"\"' LIMIT 1").fetchone() except sqlite3.DatabaseError as exc: # Builds without fts5/trigram raise "no such module|tokenizer"; calling that corruption would # send the DB into repair, whose final fallback deletes messages_fts%. "no such table/column" = # not built yet. benign = SessionDB._is_fts5_unavailable_error(exc) or _schema_not_built(exc) if not (isinstance(exc, sqlite3.OperationalError) and benign): # This is the corruption class #66724 actually wants caught: partial shadow-table # damage where MATCH / snippet / rank queries raise DatabaseError("database disk # image is malformed") while reads of the FTS5 table itself parse fine. return f"fts5 read probe failed on {fts_table}: {exc}" # FTS write probe: drive a row through the messages_fts* triggers in a transaction that is always # rolled back. The trigger INSERT alone only buffers the row in FTS5's in-memory segment; the # ``flush`` command writes that segment to ``_idx``/``_data`` exactly as a committed append # would, so a stale ``_idx`` row at the next segid (IntegrityError "constraint failed", the #100227 # class: integrity_check and MATCH both clean, every real append fails) is hit here rather than # by the user's next message. probe_session_id = f"_hermes_fts_health_probe_{time.time_ns()}" try: conn.execute("BEGIN IMMEDIATE") conn.execute("INSERT INTO sessions (id, source, started_at) VALUES (?, ?, ?)", (probe_session_id, "_health_probe", time.time())) conn.execute("INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, ?, ?, ?)", (probe_session_id, "user", "_fts_health_probe", time.time())) for fts_table in _FTS_TABLES: try: conn.execute(f"INSERT INTO {fts_table}({fts_table}) VALUES('flush')") except sqlite3.OperationalError as exc: if not (SessionDB._is_fts5_unavailable_error(exc) or _schema_not_built(exc)): raise except sqlite3.DatabaseError as exc: # IntegrityError is a DatabaseError sibling of OperationalError, not a child: catching only the # latter let the trigram-segment collision report "healthy". # Missing messages/sessions tables = brand new file mid-init, not corruption. "no such tokenizer": # this process lacks the cjk extension the DB's index needs — capability gap; a tokenizer-less # SessionDB drops the triggers itself. if _schema_not_built(exc) or "no such tokenizer: cjk_unicode61" in str(exc).lower(): return None return f"fts5 write probe failed: {exc}" finally: with contextlib.suppress(sqlite3.Error): conn.execute("ROLLBACK") return None except sqlite3.DatabaseError as exc: return str(exc) def _live_writer_holds_db(db_path: Path) -> bool: """True when another process (or a connection outside this call) still holds ``db_path``. The foreign-holder scan (``hermes_state_holders``) is the authority: any other process with the DB or a WAL sidecar open, a deleted WAL generation, or an unknown/uninspectable holder fails CLOSED. The SQLite probe (``locking_mode=EXCLUSIVE`` + ``BEGIN IMMEDIATE``) is only an additional positive signal — it cannot see a ``journal_mode=DELETE`` reader and cannot run on a malformed file, which is why the scan comes first.""" import hermes_state_holders as _state_holders return _state_holders.live_writer_holds_db(db_path, connect_repair_durable=_connect_repair_durable) def _repair_skip(report: Dict[str, Any], verb: str, error: str, exc: Optional[BaseException] = None) -> Dict[str, Any]: """Record *error* on *report* and log it as ``state.db repair ``. An *exc* proving deterministic corruption consumes the persistent repair budget (private ``_repair_attempted`` marker, popped by the caller).""" if exc is not None and _repair_failure_consumes_attempt(exc): report["_repair_attempted"] = True report["error"] = error logger.error(f"state.db repair {verb}: %s", error) return report def repair_state_db_schema(db_path: Path, *, backup: bool = True) -> Dict[str, Any]: """Repair a state.db whose ``sqlite_master`` is malformed or whose FTS indexes reject writes. Two corruption classes: malformed schema / "duplicate object definition" (even ``PRAGMA`` fails), and FTS write-corruption (reads and ``integrity_check`` pass, writes fail through ``messages_fts*`` triggers). ``_REPAIR_STRATEGIES`` run least-destructive first on a complete snapshot; a success is copied back transactionally, so canonical rows are never modified by a failed attempt. A raw backup is taken first unless ``backup=False``. Serialised across processes (gateway, Desktop backend and CLI open the same file; concurrent ``writable_schema`` surgery is itself a corruption source). Returns ``{repaired, strategy, backup_path, error}``. See #50502. """ report: Dict[str, Any] = {"repaired": False, "strategy": None, "backup_path": None, "error": None} # Startup-watchdog lease: repair is I/O-bound (near-zero CPU), which the watchdog's CPU fallback would # misread as a parked deadlock. One lease (clamped to _MAX_LEASE_S=900) beats per-chunk renewal complexity. report_startup_progress(900.0, phase="state_db_repair") db_path = Path(db_path) if not db_path.exists(): report["error"] = f"{db_path} does not exist" return report # Cross-restart cap: the in-memory claim bounds one process, but unhealable b-tree damage used to re-run # surgery + a fresh backup on EVERY restart. # Cross-restart attempt cap (#86747): the in-memory claim bounds one process, but a corruption class the # strategies below cannot heal (b-tree page damage) previously re-ran the whole surgery — and took a # fresh multi-hundred-MB forensic backup — on EVERY restart, forever. After # _MAX_PERSISTENT_REPAIR_ATTEMPTS failures against the same damaged file, stop retrying and surface a # terminal, actionable error. if _persistent_repair_attempts_exhausted(db_path): return _repair_skip(report, "skipped", _persistent_repair_exhausted_error(db_path)) with _cross_process_repair_lock(db_path) as holding_lock: if not holding_lock: # Another process holds the lock (or the lock file was unopenable); it may have healed the file # already, so re-probe before failing. if _db_opens_cleanly(db_path) is None: report["repaired"], report["strategy"] = True, "repaired_by_other_process" else: report["error"] = ("could not obtain the state.db repair lock (held by another process, or the lock " "file was unopenable); skipped schema surgery to avoid racing a concurrent repairer") return report result = report # Recheck exhaustion after acquisition: a queued repairer can have recorded the final failure while # this process waited. if _persistent_repair_attempts_exhausted(db_path): _repair_skip(report, "skipped", _persistent_repair_exhausted_error(db_path)) # WAL-holder preflight: fail closed for active readers before a backup is taken. Not the race defence — the # exclusive guard in the locked routine excludes writers through promotion and sees DELETE-mode readers too. elif _live_writer_holds_db(db_path): _repair_skip(report, "skipped", "a live writer still holds state.db; skipped schema surgery to avoid tearing " "b-tree pages under a concurrent writer. Stop the gateway (hermes gateway stop) and retry.") else: # Probe journal mode BEFORE surgery: a rebuilt file comes back in the default (delete) mode and nothing # else records the flip. Unprobeable (damaged file) -> database.journal_mode is the restore target. # The open-time WAL-reset gate never sees this flip because it happens inside the repair path # (distinct from the open-time flip #89393 warns about). before_mode = _probe_journal_mode_for_repair(db_path) result = _repair_state_db_schema_locked(db_path, backup=backup, report=report, journal_mode_before=before_mode) if result.get("repaired"): result["journal_mode_before"] = before_mode # Environmental aborts (before a strategy mutates the snapshot) are retriable, not proof of exhaustion; # the private marker stays out of the public report. The ledger update stays under the cross-process # lock so two repairers cannot lose each other's updates; a queued loser must not record at all. if result.pop("_repair_attempted", False) or result.get("repaired"): _record_repair_outcome(db_path, repaired=bool(result.get("repaired"))) return result def _probe_journal_mode_for_repair(db_path: Path) -> Optional[str]: """Best-effort journal-mode probe: ``wal``/``delete``, or ``None`` when the file cannot be opened or probed (malformed header, concurrent opener's locks — both expected on the repair path); callers then fall back to ``database.journal_mode``.""" from hermes_state_wal import _on_disk_journal_mode try: with _repair_conn(db_path) as conn: return _on_disk_journal_mode(conn) except (sqlite3.Error, OSError): return None def _restore_journal_mode_after_repair(db_path: Path, before_mode: Optional[str], *, conn=None) -> None: """Re-apply the journal mode after schema surgery. ``conn`` must be the exclusive repair guard connection when called from the repair path: opening a fresh connection AFTER the guard released let a writer still holding the unlinked old ``-wal`` inode coexist with a brand-new ``state.db-wal`` — two generations of one store. The reopen is the hazard, not the mode. A rebuilt file comes back in the default (delete) mode; without this a corruption event silently moves a WAL store out of WAL (the open-time WAL-reset gate never sees a flip made inside repair). Routed through :func:`apply_wal_with_fallback`, not a direct pragma, so it inherits the WAL-reset gate (a vulnerable runtime deliberately keeps DELETE; the journal_mode-changed WARNING is expected there), the macOS-NFS silent-refusal handling and the WAL companions. ``before_mode`` is only for the log comparison; the target is ``database.journal_mode``. Best-effort: the repair already succeeded, so failures log at WARNING. See #89674. See #89393. The transactional promotion already leaves the destination in its pre-repair mode, so on that path this is mostly the WAL-companion re-assertion; the reopen is the hazard, not the mode. See #101064. """ from hermes_state_wal import apply_wal_with_fallback try: if conn is None: with _repair_conn(db_path) as owned: after = apply_wal_with_fallback(owned, db_label=db_path.name) else: after = apply_wal_with_fallback(conn, db_label=db_path.name) if before_mode and after != before_mode: logger.warning("state.db repair changed journal_mode %r -> %r (pre-surgery probe %r; restore resolved " "through apply_wal_with_fallback per database.journal_mode and the WAL-reset gate)", before_mode, after, before_mode) except (sqlite3.Error, OSError) as exc: logger.warning("state.db repair at %s: post-surgery journal-mode restore failed (%s); verify with PRAGMA " "journal_mode on the next open", db_path, exc) def _repair_state_db_schema_locked( db_path: Path, *, backup: bool, report: Dict[str, Any], journal_mode_before: Optional[str] = None, ) -> Dict[str, Any]: """Repair strategies for :func:`repair_state_db_schema`; caller holds the cross-process repair lock. Strategies run on a SCRATCH COPY, copied back through SQLite's transactional backup API only once proven to open cleanly, so a failed repair cannot modify or lose committed data (a WAL checkpoint of committed frames on guard release is not a repair mutation). WHY not in place: the final strategy ends in ``VACUUM``, which rebuilds the file from the schema SQLite can still parse — when the damage IS in the schema b-tree (the ``malformed database schema ()`` class) every table hanging off the unreadable part is silently dropped, the probe still reports malformed, and repair returned ``repaired=False`` having destroyed what it was asked to save. The pre-repair backup (#69603) does not close this: it is a forensic artefact that nothing reads back, so recovery still depends on a human noticing a ``.malformed-backup-*`` file and knowing what to do with it. Not mutating the original in the first place is the property that holds without a human in the loop. """ scratch = db_path.with_name(f"{db_path.name}.repair-scratch") if (cleanup_error := _unlink_db_triple(scratch)) is not None: return _repair_skip(report, "aborted", f"could not remove a stale repair snapshot before probing state.db: {cleanup_error}") # Re-probe under the lock: a process we queued behind may have just repaired the file; redoing surgery # would undo it (the repair/re-corrupt cascade). if _db_opens_cleanly(db_path) is None: report["repaired"], report["strategy"] = True, "already_healthy" return report if backup: bpath, backup_error = _backup_db_file(db_path) report["backup_path"] = str(bpath) if bpath else None if bpath is None: # HARD STOP: the forensic image is the recovery path when every strategy fails. return _repair_skip(report, "aborted", "pre-repair backup refused; aborting schema repair to avoid " f"mutating the only copy of the damaged DB: {backup_error}") # The forensic copy precedes this guard on purpose: its live-holder checks would be poisoned by our own # exclusive connection. Everything touching the repair image or live promotion happens under writer exclusion. with _exclusive_repair_db_guard(db_path) as (live_guard, guard_error): if live_guard is None: return _repair_skip(report, "skipped", "could not acquire exclusive state.db repair ownership; skipped " f"schema surgery to avoid overwriting a concurrent writer. Stop the gateway and retry: " f"{guard_error}", exc=guard_error) if (space_error := _repair_scratch_space_error(db_path)) is not None: return _repair_skip(report, "aborted", space_error) try: # Source = live_guard: it owns the exclusion, and a second connection could be blocked by our own # EXCLUSIVE lock on some SQLite builds. _copy_database_snapshot(db_path, scratch, source_connection=live_guard) except (OSError, sqlite3.Error, TimeoutError) as exc: _unlink_db_triple(scratch) return _repair_skip(report, "aborted", f"could not stage a complete SQLite repair snapshot of {db_path}: {exc}", exc=exc) try: # Private marker for the outer wrapper: a strategy failure consumes the persistent budget; a # promotion failure is classified separately. report["_repair_attempted"] = True _run_repair_strategies(scratch, report) if report.get("repaired"): # Never ``os.replace`` the live DB: Windows rejects replacement under open handles and POSIX would # leave those handles on the old inode. The guard keeps writer exclusion throughout. try: _copy_database_snapshot(scratch, db_path, destination_connection=live_guard) except (OSError, sqlite3.Error, TimeoutError) as exc: report.update(repaired=False, strategy=None, _repair_attempted=_repair_failure_consumes_attempt(exc)) report["error"] = f"repaired snapshot could not be promoted transactionally: {exc}" logger.error("state.db repair promotion failed: %s", exc) else: logger.warning("state.db repaired via '%s' and promoted transactionally: %s", report.get("strategy"), db_path) _restore_journal_mode_after_repair(db_path, journal_mode_before, conn=live_guard) if not report.get("repaired"): # Logged HERE, not in the strategies: they see the scratch copy, and the message a human acts on # must name a path that still exists. logger.error("state.db schema repair could not recover %s automatically (no committed canonical data " "was modified or lost; backup: %s); manual restore from backup may be required.", db_path, report["backup_path"]) return report finally: # Never leave a half-repaired file beside the DB to be mistaken for the real thing. if (cleanup_error := _unlink_db_triple(scratch)) is not None: logger.warning("Could not remove state.db repair snapshot after repair: %s", cleanup_error) def _unlink_db_triple(path: Path) -> Optional[str]: """Remove *path* and every SQLite sidecar; return any cleanup failure.""" from hermes_state import _IS_WINDOWS failures: List[str] = [] for victim in (path, *_sidecars(path)): for attempt in range(10): try: victim.unlink(missing_ok=True) except OSError as exc: # Windows may retain a just-closed SQLite handle for a few scheduler ticks; bounded retry. if isinstance(exc, PermissionError) and _IS_WINDOWS and attempt < 9: time.sleep(0.05) continue failures.append(f"{victim}: {exc}") break return "; ".join(failures) or None # ── Repair strategies, least destructive first (each mutates its connection's DB) ── def _edit_sqlite_master(conn: sqlite3.Connection, edit) -> None: """Run *edit* under ``writable_schema=ON``; bump the schema cookie when it reports a change so live peers discard their cached schema.""" conn.execute("PRAGMA writable_schema=ON") if edit(): _bump_schema_cookie(conn) conn.execute("PRAGMA writable_schema=OFF") conn.commit() def _strategy_rebuild_fts(conn: sqlite3.Connection) -> None: """FTS5 'rebuild' rewrites each index from the content table: the least- destructive fix for an index that rejects writes while reads work.""" from hermes_state import load_fts5_cjk_extension # The cjk index can only be rebuilt with its tokenizer loaded (best-effort). load_fts5_cjk_extension(conn) for table_name in _FTS_TABLES: with contextlib.suppress(sqlite3.OperationalError): # table absent (FTS disabled / trigram off / cjk absent) conn.execute(f"INSERT INTO {table_name}({table_name}) VALUES('rebuild')") def _strategy_reindex(conn: sqlite3.Connection) -> None: """integrity_check reports "wrong # of entries in index" when a B-tree index drifts from its base table; REINDEX rewrites it from canonical rows.""" # REINDEX rewrites every index b-tree; take the barriers now that the schema parses. _reapply_durability_barriers(conn) conn.execute("REINDEX") conn.commit() def _strategy_dedup_schema(conn: sqlite3.Connection) -> None: """De-duplicate sqlite_master (lowest rowid per type/name wins), keeping FTS.""" def _dedup() -> bool: dupes = conn.execute( "SELECT type, name, COUNT(*) AS c, MIN(rowid) AS keep FROM sqlite_master GROUP BY type, name HAVING c > 1" ).fetchall() for type_, name, _count, keep in dupes: conn.execute("DELETE FROM sqlite_master WHERE type IS ? AND name IS ? AND rowid <> ?", (type_, name, keep)) return bool(dupes) _edit_sqlite_master(conn, _dedup) def _strategy_drop_fts_vacuum(conn: sqlite3.Connection) -> None: """Drop all FTS schema and VACUUM; indexes rebuild on the next open. The destructive one, and why strategies run on a scratch copy: on a damaged schema b-tree VACUUM silently drops every table hanging off the unreadable part.""" _edit_sqlite_master(conn, lambda: conn.execute("DELETE FROM sqlite_master WHERE name LIKE 'messages_fts%'") or True) # The schema parses now, so the barriers can stick — VACUUM rewrites the whole file. _reapply_durability_barriers(conn) conn.execute("VACUUM") # (name, body, success log, failure log) in escalation order. failure log None = # final strategy: its failure lands in report["error"] instead of logged-and-skipped. _REPAIR_STRATEGIES = ( ("rebuild_fts", _strategy_rebuild_fts, "state.db FTS indexes rebuilt in place (schema preserved): %s", "state.db FTS in-place rebuild pass failed: %s"), ("reindex_btree", _strategy_reindex, "state.db B-tree indexes rebuilt via REINDEX: %s", "state.db REINDEX pass failed: %s"), ("dedup_schema", _strategy_dedup_schema, "state.db schema repaired by de-duplicating sqlite_master (FTS index preserved): %s", "state.db dedup repair pass failed: %s"), ("drop_fts_rebuild", _strategy_drop_fts_vacuum, "state.db schema repaired by dropping FTS schema; indexes will rebuild from messages on next open: %s", None), ) def _run_repair_strategies(db_path: Path, report: Dict[str, Any]) -> Dict[str, Any]: """Escalating repair attempts, applied to *db_path* IN PLACE — only ever a scratch copy nothing else holds open, never the user's database. The "could not recover" log lives in the caller so it names the user's database.""" for name, body, success_msg, failure_msg in _REPAIR_STRATEGIES: try: with _repair_conn(db_path) as conn: body(conn) reason = _db_opens_cleanly(db_path) except sqlite3.DatabaseError as exc: reason = str(exc) if failure_msg is not None: logger.warning(failure_msg, exc) if reason is None: report["repaired"], report["strategy"] = True, name logger.warning(success_msg, db_path) return report if failure_msg is None: report["error"] = reason return report