b8f99bfc43
Reviewer P2 (kshitijk4poor): tests patch hermes_cli.models.get_cached_nous_inference_base_url
but models_pricing.pricing_cache_scope read its own module global, so the patch never reached
the call and the test passed on the default-endpoint fallback. Same seam-erosion class audited
across /tmp/rf/patch_traps.json (777 candidates) with an AST reachability check + a per-test
call-count probe (facade vs defining module) against BASE 63279301bcb; three seams actually
bypassed their patch on HEAD but not on BASE:
- hermes_cli.models.get_cached_nous_inference_base_url <- models_pricing.pricing_cache_scope
- hermes_state._connect_repair_durable <- hermes_state_repair._open_exclusive/_repair_conn/_db_opens_cleanly
- tools.approval._command_detection_variants <- approval_detection.detect_{dangerous,hardline}_command
Each now looks the name up through its facade at call time (the pattern hermes_state_repair
already used in live_writer_holds_db), restoring BASE's patchability.
1091 lines
62 KiB
Python
1091 lines
62 KiB
Python
"""state.db repair, backup and writability preflight (split from hermes_state).
|
|
|
|
Every name is re-imported into ``hermes_state``; intra-module calls to patchable helpers go through a lazy
|
|
``from hermes_state import ...`` at call time so monkeypatches there still intercept.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import datetime
|
|
import hashlib
|
|
import itertools
|
|
import json
|
|
import logging
|
|
import os
|
|
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_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 (``<db>.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 ``<db>.<suffix>`` 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, _REPAIR_SNAPSHOT_MIN_THROUGHPUT_BYTES_PER_SECOND
|
|
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."""
|
|
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 sessions recover --source {db_path} "
|
|
f"--inspect-only`, then (if it reports recoverable) `hermes 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 <db> --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."""
|
|
conn = sqlite3.connect(str(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 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 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"))
|
|
|
|
|
|
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.
|
|
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()))
|
|
conn.execute("ROLLBACK")
|
|
except sqlite3.OperationalError as exc:
|
|
with contextlib.suppress(sqlite3.Error):
|
|
conn.execute("ROLLBACK")
|
|
# 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 str(exc)
|
|
return None
|
|
except sqlite3.DatabaseError as exc:
|
|
return str(exc)
|
|
|
|
|
|
def _live_writer_holds_db(db_path: Path) -> bool:
|
|
"""True when a connection outside this call still holds ``db_path`` open.
|
|
|
|
Asks SQLite for what a repair needs and a live holder cannot grant: ``locking_mode=EXCLUSIVE`` then
|
|
``BEGIN IMMEDIATE`` — in WAL mode that needs exclusive WAL-index locks, so any other open connection fails
|
|
it with SQLITE_BUSY; neither statement parses the schema, so it works on malformed DBs. Fails **open**
|
|
(False) on anything but a positive busy/locked signal: refusing to repair a DB nobody holds would strand
|
|
the self-heal path. In ``journal_mode=DELETE`` a held reader takes only SHARED and this returns False;
|
|
repair is then serialised only by the cross-process repairer lock. Before probing, the foreign-holder scan
|
|
(``hermes_state_holders``) fails closed on deleted-WAL-generation, uninspectable, or unknown holders."""
|
|
import hermes_state_holders as _state_holders
|
|
from hermes_state import _connect_repair_durable as _connect # call-time lookup: tests patch hermes_state.<name>
|
|
return _state_holders.live_writer_holds_db(db_path, connect_repair_durable=_connect)
|
|
|
|
|
|
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 <verb>``. 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.
|
|
"""
|
|
from hermes_state import (_cross_process_repair_lock, _db_opens_cleanly, _live_writer_holds_db,
|
|
_persistent_repair_attempts_exhausted, _probe_journal_mode_for_repair,
|
|
_record_repair_outcome, _repair_state_db_schema_locked)
|
|
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 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 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.
|
|
"""
|
|
from hermes_state import (_backup_db_file, _copy_database_snapshot, _db_opens_cleanly, _exclusive_repair_db_guard,
|
|
_repair_scratch_space_error, _restore_journal_mode_after_repair, _run_repair_strategies,
|
|
_unlink_db_triple)
|
|
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."""
|
|
from hermes_state import _db_opens_cleanly
|
|
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
|