288 lines
13 KiB
Python
Executable File
288 lines
13 KiB
Python
Executable File
#!/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": []}
|
||
|
||
# 用 immutable 只读连接快速体检（完全跳过 wal/lock，gateway 运行时零风险）
|
||
# 🔴 2026-09-04 教训：gateway 运行中普通只读连接（mode=ro 非 immutable）也会触发 wal 访问
|
||
health_conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro&immutable=1", 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")
|
||
|
||
# === 阶段 1.5：损坏自动恢复（2026-09-03 新增 — 防止 gateway 启动失败导致 agent 失联）===
|
||
# ExecStartPre 场景（gateway 尚未启动）若检测到真损坏 → 从最近健康快照恢复
|
||
# 前提：无 gateway 进程持有 DB（手动运行时保护）
|
||
if (not ok or integrity != "ok") and not dry_run:
|
||
import subprocess as _sp
|
||
_gw = _sp.run(["pgrep", "-f", "hermes_cli.main gateway run"],
|
||
capture_output=True, text=True)
|
||
if _gw.returncode == 0:
|
||
log(" ⚠️ gateway 正在运行，跳过自动恢复（避免覆盖活动 DB）")
|
||
else:
|
||
SNAP_DIR = Path.home() / ".hermes" / "backups" / "state-db-snap"
|
||
if SNAP_DIR.exists():
|
||
snaps = sorted(SNAP_DIR.glob("state-*.db"))
|
||
if snaps:
|
||
latest = snaps[-1]
|
||
try:
|
||
_sc = sqlite3.connect(f"file:{latest}?mode=ro", uri=True, timeout=10)
|
||
_sok = _sc.execute("PRAGMA integrity_check").fetchone()[0]
|
||
_sc.close()
|
||
except sqlite3.Error as _e:
|
||
_sok = None
|
||
log(f" ⚠️ 快照 {latest.name} 读取失败: {_e}")
|
||
if _sok == "ok":
|
||
import shutil as _sh
|
||
_ts = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
|
||
_bad = DB_PATH.with_name(f"state.db.bad-auto-{_ts}")
|
||
_sh.copy2(DB_PATH, _bad)
|
||
_sh.copy2(latest, DB_PATH)
|
||
for _p in (WAL_PATH, SHM_PATH):
|
||
if _p.exists():
|
||
try:
|
||
_p.unlink()
|
||
except OSError:
|
||
pass
|
||
log(f" 🔄 自动恢复：DB 损坏 → 从 {latest.name} 恢复（坏库备份 {_bad.name}）")
|
||
report["actions"].append(f"auto_restore_from={latest.name}")
|
||
# 重新体检
|
||
try:
|
||
_hc = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True, timeout=10)
|
||
_ok2, _int2 = safe_run(_hc, "PRAGMA integrity_check")
|
||
report["integrity_after_restore"] = _int2
|
||
log(f" 恢复后 integrity_check = {_int2}")
|
||
_hc.close()
|
||
except sqlite3.Error as _e:
|
||
log(f" ⚠️ 恢复后体检失败: {_e}")
|
||
else:
|
||
log(f" ⚠️ 最近快照 {latest.name} 也不健康（{_sok}），跳过自动恢复")
|
||
else:
|
||
log(" ⚠️ 无可用快照（backups/state-db-snap 为空），跳过自动恢复")
|
||
else:
|
||
log(" ⚠️ 无快照目录 backups/state-db-snap，跳过自动恢复")
|
||
|
||
# === 阶段 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 止血脚本（安全默认：只体检 dry-run，--apply 才执行修改）")
|
||
p.add_argument("--apply", action="store_true", help="执行实际修改（默认只体检不修改）")
|
||
p.add_argument("--dry-run", action="store_true", help="强制只体检（兼容旧用法）")
|
||
args = p.parse_args()
|
||
# 🔴 安全默认反转（2026-09-04）：无参数跑 = 只体检。显式 --apply（且未 --dry-run）才执行修改/自动恢复
|
||
dry_run = (not args.apply) or args.dry_run
|
||
log(f"模式: {'DRY-RUN 只体检（--apply 才执行修改）' if dry_run else 'APPLY 执行修改'}")
|
||
return stabilize(dry_run=dry_run)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main()) |