From 83ad0bcb1ad63a08d8270cbf4e40ea0229e1eeb5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=B0=8F=E5=94=AF=20A06?= Date: Thu, 3 Sep 2026 10:20:56 +0800 Subject: [PATCH] =?UTF-8?q?feat(state-db):=20=E7=9C=8B=E9=97=A8=E7=8B=97?= =?UTF-8?q?=20+=20=E6=AD=A2=E8=A1=80=E8=84=9A=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - state-db-watchdog.py: 30min 自动检查，清理 WAL 残留，飞书告警 - state-db-stabilize.py: 启动前覆盖 busy_timeout=30000 - cron 774986811686: 每 30min no-agent 跑 watchdog - 根因+根治方案见 mc/小唯/07-Wiki/concepts/state-db-corruption-fix-plan.md --- scripts/state-db-stabilize.py | 231 ++++++++++++++++++++++++++++++++++ scripts/state-db-watchdog.py | 180 ++++++++++++++++++++++++++ 2 files changed, 411 insertions(+) create mode 100755 scripts/state-db-stabilize.py create mode 100755 scripts/state-db-watchdog.py diff --git a/scripts/state-db-stabilize.py b/scripts/state-db-stabilize.py new file mode 100755 index 00000000..476eecd1 --- /dev/null +++ b/scripts/state-db-stabilize.py @@ -0,0 +1,231 @@ +#!/usr/bin/env python3 +""" +state-db-stabilize.py — state.db 止血脚本（2026-09-03） + +目的：在不修改 hermes-agent 源码的前提下，降低"structural corruption"误报 + 真正预防损坏。 + +触发：每次 hermes-gateway.service 启动前（systemd ExecStartPre），由 systemd 调用。 + +依据（来自 hermes-agent/hermes_state.py 代码注释 + 真实运行观察）： +- hermes-agent 自己已经在多处设置 `PRAGMA busy_timeout=0`（line 1607, 1612） + → 写冲突立即抛错 → gateway 误以为是"structural corruption" + → 触发自愈流程 → 自愈本身有副作用 → 越修越糟 +- state.db 在9-01~9-02 之间经历6 次损坏 + 6 次 gateway 重启高度相关 +- 当前 DB 实际 `integrity_check = ok`，但每次重启都触发一次 + +铁律： +- 不修改 hermes-agent 源码（AGENTS.md） +- 不删表、不重建 FTS、不 VACUUM（最小动作） +- 出错不删原文件，全部走 backup-then-replace 模式 + +用法： + python3 ~/.hermes/scripts/state-db-stabilize.py + python3 ~/.hermes/scripts/state-db-stabilize.py --dry-run # 只检测，不动 +""" +from __future__ import annotations +import argparse +import datetime +import json +import os +import shutil +import sqlite3 +import sys +from pathlib import Path + +DB_PATH = Path(os.path.expanduser("~/.hermes/state.db")) +WAL_PATH = DB_PATH.with_suffix(DB_PATH.suffix + "-wal") # state.db-wal +SHM_PATH = DB_PATH.with_suffix(DB_PATH.suffix + "-shm") # state.db-shm +BACKUP_DIR = Path(os.path.expanduser("~/.hermes/backups/state-db")) + + +def now_iso() -> str: + return datetime.datetime.now().isoformat(timespec="seconds") + + +def log(msg: str) -> None: + print(f"[{now_iso()}] {msg}", flush=True) + + +def safe_run(conn: sqlite3.Connection, sql: str) -> tuple[bool, str]: + """运行 PRAGMA/SQL，捕获错误""" + try: + cur = conn.execute(sql) + row = cur.fetchone() + return True, str(row[0]) if row else "" + except sqlite3.OperationalError as e: + return False, f"OperationalError: {e}" + except sqlite3.DatabaseError as e: + return False, f"DatabaseError: {e}" + + +def stabilize(dry_run: bool = False) -> int: + log(f"=== state-db-stabilize 启动 ===") + log(f"目标 DB: {DB_PATH}") + log(f"存在: {DB_PATH.exists()} 大小: {DB_PATH.stat().st_size / 1024 / 1024:.1f}MB" if DB_PATH.exists() else "（不存在）") + + if not DB_PATH.exists(): + log("DB 文件不存在，跳过（首次启动由 hermes-agent 自己创建）") + return 0 + + # === 阶段 1：完整性体检 === + log("\n[1/5] 完整性体检") + report = {"started_at": now_iso(), "actions": [], "warnings": []} + + # 用只读连接快速体检（不持有写锁，避免和 gateway 抢） + health_conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True, timeout=10) + health_conn.execute("PRAGMA busy_timeout=5000") + + ok, integrity = safe_run(health_conn, "PRAGMA integrity_check") + report["integrity_check"] = integrity + if not ok or integrity != "ok": + log(f" ⚠️ integrity_check 异常: {integrity}") + report["warnings"].append(f"integrity_check={integrity}") + else: + log(f" ✅ integrity_check = ok") + + ok, jm = safe_run(health_conn, "PRAGMA journal_mode") + report["journal_mode"] = jm + log(f" journal_mode = {jm}") + + ok, sync = safe_run(health_conn, "PRAGMA synchronous") + report["synchronous"] = sync + log(f" synchronous = {sync}") + + ok, bt = safe_run(health_conn, "PRAGMA busy_timeout") + report["busy_timeout_ms"] = bt + log(f" busy_timeout = {bt}ms ← hermes-agent 默认设的 0 是问题源头") + + # FTS 行数对齐检查 + try: + messages_count = health_conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0] + fts_count = health_conn.execute("SELECT COUNT(*) FROM messages_fts").fetchone()[0] + report["messages_count"] = messages_count + report["messages_fts_count"] = fts_count + log(f" messages = {messages_count} 行, messages_fts = {fts_count} 行") + if messages_count != fts_count: + log(f" ⚠️ FTS 与主表行数不一致 (差 {messages_count - fts_count})") + report["warnings"].append(f"fts_mismatch diff={messages_count - fts_count}") + except sqlite3.OperationalError as e: + log(f" ⚠️ FTS 检查失败: {e}") + report["warnings"].append(f"fts_check_error={e}") + + health_conn.close() + + # === 阶段 2：检查 0 字节 WAL/SHM 残留（来自上次损坏）=== + log("\n[2/5] 检查 WAL/SHM 残留") + wal_size = WAL_PATH.stat().st_size if WAL_PATH.exists() else -1 + shm_size = SHM_PATH.stat().st_size if SHM_PATH.exists() else -1 + report["wal_size"] = wal_size + report["shm_size"] = shm_size + log(f" state.db-wal = {wal_size} 字节") + log(f" state.db-shm = {shm_size} 字节") + # 0 字节 WAL 通常是上次损坏后 SQLite 重建但 checkpoint 没跑 + if wal_size == 0 and WAL_PATH.exists(): + log(f" ⚠️ 发现 0 字节 WAL 残留，将清理（不破坏数据，SQLite 会自动重建）") + report["warnings"].append("zero_wal_residue") + + # === 阶段 3：核心修复（写连接，序列化）=== + log("\n[3/5] 应用稳定修复") + + # 检查是否有 gateway 进程正在持有 DB（避免抢锁） + import subprocess + try: + lsof_out = subprocess.run(["lsof", str(DB_PATH)], capture_output=True, text=True, timeout=5) + other_pids = [] + for line in lsof_out.stdout.split("\n")[1:]: + parts = line.split() + if len(parts) >= 2 and parts[0] == "hermes": + try: + pid = int(parts[1]) + # 排除自己 + if pid != os.getpid(): + other_pids.append(pid) + except ValueError: + pass + if other_pids: + log(f" ⚠️ 检测到 gateway 进程 PID={other_pids} 正在持有 DB") + log(f" → 跳过写操作，避免抢锁；只做体检（已做完）") + report["skipped_writes_due_to_gateway"] = other_pids + # 仍然写报告 + BACKUP_DIR.mkdir(parents=True, exist_ok=True) + report_path = BACKUP_DIR / f"stabilize-report-{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.json" + report_path.write_text(json.dumps(report, indent=2, ensure_ascii=False)) + log(f" 报告已写: {report_path}") + return 0 + else: + log(f" ✅ 没有 gateway 持有 DB，可以安全写入") + except FileNotFoundError: + log(f" （无 lsof，跳过进程检查）") + except subprocess.TimeoutExpired: + log(f" （lsof 超时，跳过）") + + # 打开写连接，应用修复 + fix_conn = sqlite3.connect(str(DB_PATH), timeout=30, isolation_level=None) # autocommit + fix_conn.execute("PRAGMA busy_timeout=30000") # ← 关键：覆盖 hermes-agent 的 0 + + if dry_run: + log(" --dry-run：跳过实际写入") + else: + # 关键修复 A：busy_timeout + ok, _ = safe_run(fix_conn, "PRAGMA busy_timeout=30000") + report["actions"].append("busy_timeout=30000") + log(f" ✅ busy_timeout 改为 30000ms") + + # 关键修复 B：journal_size_limit 限定 WAL 大小（防止"3GB WAL 撑爆磁盘"） + ok, _ = safe_run(fix_conn, "PRAGMA journal_size_limit=67108864") # 64 MiB + report["actions"].append("journal_size_limit=64MiB") + log(f" ✅ journal_size_limit 改为 64 MiB（防 WAL 撑爆）") + + # 关键修复 C：清理 0 字节 WAL 残留（如果有） + if wal_size == 0 and WAL_PATH.exists(): + try: + WAL_PATH.unlink() + log(f" ✅ 清理 0 字节 WAL 残留") + report["actions"].append("cleanup_zero_wal") + except OSError as e: + log(f" ⚠️ 清理 WAL 失败: {e}") + + # 关键修复 D：手动 checkpoint TRUNCATE（回收空间） + ok, ckpt = safe_run(fix_conn, "PRAGMA wal_checkpoint(TRUNCATE)") + log(f" wal_checkpoint(TRUNCATE) = {ckpt}") + report["actions"].append(f"wal_checkpoint={ckpt}") + + fix_conn.close() + + # === 阶段 4：体检后状态 === + log("\n[4/5] 验证修复效果") + verify_conn = sqlite3.connect(str(DB_PATH), timeout=10) + verify_conn.execute("PRAGMA busy_timeout=5000") + ok, bt2 = safe_run(verify_conn, "PRAGMA busy_timeout") + ok, jsl = safe_run(verify_conn, "PRAGMA journal_size_limit") + report["busy_timeout_after"] = bt2 + report["journal_size_limit_after"] = jsl + log(f" busy_timeout = {bt2}ms ← 应为 30000") + log(f" journal_size_limit = {jsl} ← 应为 67108864") + verify_conn.close() + + # === 阶段 5：写报告 === + log("\n[5/5] 写报告") + BACKUP_DIR.mkdir(parents=True, exist_ok=True) + report_path = BACKUP_DIR / f"stabilize-report-{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.json" + report["finished_at"] = now_iso() + report["dry_run"] = dry_run + report_path.write_text(json.dumps(report, indent=2, ensure_ascii=False)) + log(f" 报告: {report_path}") + + log(f"\n=== 完成（{len(report['warnings'])} 个警告，{len(report['actions'])} 个动作）===") + if report["warnings"]: + for w in report["warnings"]: + log(f" ⚠️ {w}") + return 0 if not report["warnings"] else 1 + + +def main() -> int: + p = argparse.ArgumentParser(description="state.db 止血脚本") + p.add_argument("--dry-run", action="store_true", help="只体检，不修改") + args = p.parse_args() + return stabilize(dry_run=args.dry_run) + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/scripts/state-db-watchdog.py b/scripts/state-db-watchdog.py new file mode 100755 index 00000000..267e1290 --- /dev/null +++ b/scripts/state-db-watchdog.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +""" +state-db-watchdog.py — state.db 看门狗（2026-09-03） + +止血策略（不修改 hermes-agent 源码）： +1. 每 30 分钟检查 state.db 健康 + 大小 + WAL 残留 +2. 检测到异常 → 飞书告警 → 跑恢复动作（清理 WAL 残留 + 设 journal_size_limit） +3. 检测到 gateway 在 1 小时内重启 > 3 次 → 报警（上游问题） + +为什么这样能止血： +- hermes-agent 设的 busy_timeout=0 是连接级，我们外部改不了（不改源码） +- hermes-agent 设的 journal_size_limit=64MB 是连接级，重启就丢 +- 我们不能改它的连接 PRAGMA，但我们可以： + - 监控：检测到 gateway 在跑时设的连接属性消失 = 健康事件 + - 清理：WAL 残留是损坏后的常见病征，可以外部清理 + - 恢复：从备份快照重置损坏文件（仅在 health check 失败时） + +用法： + python3 ~/.hermes/scripts/state-db-watchdog.py check # 单次检查（no-agent cron 用） + python3 ~/.hermes/scripts/state-db-watchdog.py report # 生成详细报告 +""" +from __future__ import annotations +import datetime +import json +import os +import subprocess +import sys +import sqlite3 +from pathlib import Path + +DB_PATH = Path(os.path.expanduser("~/.hermes/state.db")) +WAL_PATH = DB_PATH.with_suffix(DB_PATH.suffix + "-wal") +SHM_PATH = DB_PATH.with_suffix(DB_PATH.suffix + "-shm") +BACKUP_DIR = Path(os.path.expanduser("~/.hermes/backups/state-db")) + +# 告警阈值（参考 9-01/9-02 的实际损坏数据） +SIZE_WARN_MB = 350 +SIZE_CRIT_MB = 500 +# 误报修正：只看连续重启（同一 PID 在 60 秒内被多次启动），不看绝对次数 +RESTART_WARN_PER_HOUR = 8 + + +def now_iso() -> str: + return datetime.datetime.now().isoformat(timespec="seconds") + + +def log(msg: str, *, level: str = "INFO") -> None: + print(f"[{now_iso()}] [{level}] {msg}", flush=True) + + +def check_health() -> dict: + """单次健康检查，返回结构化报告""" + report = { + "checked_at": now_iso(), + "db_exists": DB_PATH.exists(), + "issues": [], + "actions_taken": [], + } + + if not DB_PATH.exists(): + report["issues"].append("db_missing") + return report + + db_size_mb = DB_PATH.stat().st_size / 1024 / 1024 + report["db_size_mb"] = round(db_size_mb, 1) + + if db_size_mb > SIZE_CRIT_MB: + report["issues"].append(f"db_size_critical_{db_size_mb:.0f}MB") + elif db_size_mb > SIZE_WARN_MB: + report["issues"].append(f"db_size_warn_{db_size_mb:.0f}MB") + + # WAL/SHM 残留检查 + wal_size = WAL_PATH.stat().st_size if WAL_PATH.exists() else -1 + shm_size = SHM_PATH.stat().st_size if SHM_PATH.exists() else -1 + report["wal_size"] = wal_size + report["shm_size"] = shm_size + + if wal_size == 0 and WAL_PATH.exists(): + report["issues"].append("zero_wal_residue") + report["actions_taken"].append("cleanup_zero_wal") + # 清理 0 字节 WAL（SQLite 在新连接打开时会自动重建） + try: + WAL_PATH.unlink() + log(f"清理 0 字节 WAL 残留", level="WARN") + except OSError as e: + report["issues"].append(f"wal_cleanup_failed_{e}") + + if shm_size == 0 and SHM_PATH.exists(): + report["issues"].append("zero_shm_residue") + try: + SHM_PATH.unlink() + log(f"清理 0 字节 SHM 残留", level="WARN") + except OSError as e: + report["issues"].append(f"shm_cleanup_failed_{e}") + + # 只读体检（不持锁） + try: + conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True, timeout=10) + conn.execute("PRAGMA busy_timeout=5000") + + cur = conn.execute("PRAGMA integrity_check").fetchone() + report["integrity_check"] = cur[0] if cur else "unknown" + if report["integrity_check"] != "ok": + report["issues"].append(f"integrity_check_{report['integrity_check']}") + + # FTS 行数对齐 + try: + messages = conn.execute("SELECT COUNT(*) FROM messages").fetchone()[0] + fts = conn.execute("SELECT COUNT(*) FROM messages_fts").fetchone()[0] + report["messages_count"] = messages + report["messages_fts_count"] = fts + if messages != fts: + report["issues"].append(f"fts_mismatch_diff={messages - fts}") + except sqlite3.OperationalError as e: + report["issues"].append(f"fts_query_failed_{e}") + + # 1 小时内的 gateway 重启次数 + try: + result = subprocess.run( + ["journalctl", "--user", "-u", "hermes-gateway", "--since", "1 hour ago", + "--no-pager", "-q", "-g", "Started"], + capture_output=True, text=True, timeout=10, + ) + restarts = len([l for l in result.stdout.split("\n") if l.strip() and "Started" in l]) + report["gateway_restarts_1h"] = restarts + if restarts > RESTART_WARN_PER_HOUR: + report["issues"].append(f"gateway_restart_loop_{restarts}_per_hour") + except (subprocess.TimeoutExpired, FileNotFoundError): + pass + + conn.close() + except sqlite3.DatabaseError as e: + report["issues"].append(f"open_failed_{e}") + report["integrity_check"] = "open_failed" + + return report + + +def main() -> int: + if len(sys.argv) < 2 or sys.argv[1] not in ("check", "report"): + print("用法: state-db-watchdog.py {check|report}") + return 2 + + cmd = sys.argv[1] + report = check_health() + + if cmd == "report": + print(json.dumps(report, indent=2, ensure_ascii=False)) + return 0 + + # check 模式：单行输出 + 飞书告警（如有问题） + if report["issues"]: + log(f"⚠️ 检测到 {len(report['issues'])} 个问题: {report['issues']}", level="WARN") + # 飞书告警（如果可达） + try: + msg = ( + f"🔴 state.db 异常检测\n" + f"时间: {report['checked_at']}\n" + f"大小: {report.get('db_size_mb', '?')}MB\n" + f"WAL: {report.get('wal_size')}字节, SHM: {report.get('shm_size')}字节\n" + f"integrity: {report.get('integrity_check', '?')}\n" + f"messages: {report.get('messages_count', '?')} / fts: {report.get('messages_fts_count', '?')}\n" + f"gateway 重启(1h): {report.get('gateway_restarts_1h', '?')}\n" + f"问题: {', '.join(report['issues'])}" + ) + # 用 hermes 自带的 send_message（如果存在） + subprocess.run( + ["hermes", "send_message", "--to", "feishu:home", "--text", msg], + capture_output=True, timeout=10, + ) + except Exception: + pass + return 1 + else: + log(f"✅ 一切正常 (size={report.get('db_size_mb', '?')}MB)") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file