376 lines
19 KiB
Python
Executable File
376 lines
19 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（最小动作）
|
||
- ⚠️ 铁律：只能在 gateway 停止时运行（ExecStartPre 或手动停机后）！
|
||
运行中 gateway 也是 state.db 写进程 → auto_restore 文件替换会造成双写进程 → 损坏
|
||
（对应 upstream P1 #103362：gateway 在线执行维护命令 → WAL 突破 → SQLite 损坏）
|
||
只读检查 hermes doctor 可在线，但任何修复/restore 必须停机
|
||
- 出错不删原文件，全部走 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": []}
|
||
|
||
# 体检连接（2026-09-06 修正）：非空 WAL = gateway 非干净退出/有未 checkpoint 帧 →
|
||
# 必须按 WAL 一致性读（mode=ro），否则 immutable 读过期主文件假报 FTS malformed →
|
||
# 误触发自动恢复覆盖（历史"每次重启都损坏"的放大器）。
|
||
# immutable 只用于"WAL 空 + gateway 停"的干净场景（最快且零副作用）。
|
||
wal_nonempty = WAL_PATH.exists() and WAL_PATH.stat().st_size > 0
|
||
if wal_nonempty:
|
||
health_conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True, timeout=10)
|
||
else:
|
||
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:
|
||
# 🔴 2026-09-06 双保险：自动恢复前先拷副本(db+wal)普通连接复检（重放 WAL）。
|
||
# 只有副本也真损坏才覆盖——杜绝 immutable/WAL 假阳性把健康库打成旧快照
|
||
# （历史"每次重启都损坏/丢消息"的放大器之一）。
|
||
import shutil as _sh2, tempfile as _tf
|
||
_tmpd = _tf.mkdtemp(prefix="statedb-confirm-")
|
||
_cp = Path(_tmpd) / "state.db"
|
||
_sh2.copy2(DB_PATH, _cp)
|
||
if WAL_PATH.exists() and WAL_PATH.stat().st_size > 0:
|
||
_sh2.copy2(WAL_PATH, Path(_tmpd) / "state.db-wal")
|
||
try:
|
||
_cc = sqlite3.connect(str(_cp), timeout=10)
|
||
_intc = _cc.execute("PRAGMA integrity_check").fetchone()[0]
|
||
_cc.close()
|
||
except sqlite3.Error as _e:
|
||
_intc = f"copy_check_error={_e}"
|
||
_sh2.rmtree(_tmpd, ignore_errors=True)
|
||
if _intc == "ok":
|
||
log(" ✅ 副本复检健康（WAL 一致性读）→ 跳过自动恢复（体检假阳性防护）")
|
||
report["warnings"].append("health_check_false_positive_skipped_restore")
|
||
else:
|
||
log(f" ⚠️ 副本复检确认损坏: {_intc} → 执行自动恢复")
|
||
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()
|
||
# 2026-09-08 根治: 不限定进程名——任何持有 state.db 的进程都拦
|
||
# （残留 gateway/外部脚本可能叫 python3 等，历史只匹配 "hermes" 漏检）
|
||
if len(parts) >= 2:
|
||
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}")
|
||
if (not dry_run) and report.get("journal_mode") != "wal":
|
||
# apply（ExecStartPre 门禁）：DB 非 WAL 且被其他进程持有 → 拒绝启动，
|
||
# 避免 delete 模式放行 gateway 重演转换风暴；systemd 重试直至持有者退出。
|
||
report["warnings"].append("db_held_non_wal_refuse_start")
|
||
log(" 🔴 apply 且 DB 非 WAL、另有进程持有 → 硬失败（返回 1，等待持有者退出后重试）")
|
||
return 1
|
||
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
|
||
|
||
# === 阶段 2.5：apply 硬失败判定（2026-09-06 模式15）===
|
||
# ExecStartPre 门禁语义：损坏且自动恢复未能修复 → 拒绝启动（防在坏库上重复损坏循环）。
|
||
# fatal 只在此设/在 [3/5] WAL 强制失败与 [4/5] 验证失败时置位；良性警告不算。
|
||
fatal = False
|
||
if not dry_run and (not ok or integrity != "ok"):
|
||
final_int = report.get("integrity_after_restore", report.get("integrity_check"))
|
||
if final_int != "ok":
|
||
fatal = True
|
||
report["warnings"].append(f"integrity_unrepaired={final_int}")
|
||
log(f" 🔴 完整性异常且自动恢复未能修复 → 硬失败（拒绝在坏库上启动）")
|
||
|
||
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")
|
||
|
||
# 关键修复 C：清理 0 字节 WAL 残留（如果有）——必须在切 WAL 前做，
|
||
# 否则 SQLite 可能把残留文件当作活动 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}")
|
||
|
||
# 关键修复 E（模式15 根治）：显式切 journal_mode=WAL。
|
||
# 快照/自动恢复链路产出 journal_mode=delete 库（.backup / cp 产物默认 delete 头，
|
||
# 见 t_91014257）；若不切回 WAL，gateway 启动多连接并发抢 delete→WAL 转换 →
|
||
# disk I/O error 风暴（#55305/#71498）→ FTS5 影子表(Tree 60/64) 撕裂。
|
||
# 失败 = 硬失败（fatal），ExecStartPre 拒绝启动。
|
||
_jm_before = report.get("journal_mode")
|
||
ok, jm_set = safe_run(fix_conn, "PRAGMA journal_mode=WAL")
|
||
report["journal_mode_after_set"] = jm_set
|
||
if ok and jm_set == "wal":
|
||
report["actions"].append("journal_mode=WAL")
|
||
log(f" ✅ journal_mode 强制 = WAL（原 {_jm_before}）")
|
||
else:
|
||
fatal = True
|
||
report["warnings"].append(f"journal_mode_force_wal_failed={jm_set}")
|
||
log(f" 🔴 journal_mode 强制 WAL 失败: {jm_set} → 硬失败")
|
||
|
||
# 关键修复 B：journal_size_limit 限定 WAL 大小（须在 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 撑爆）")
|
||
|
||
# 关键修复 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")
|
||
ok, jm2 = safe_run(verify_conn, "PRAGMA journal_mode")
|
||
report["busy_timeout_after"] = bt2
|
||
report["journal_size_limit_after"] = jsl
|
||
report["journal_mode_after"] = jm2
|
||
log(f" busy_timeout = {bt2}ms ← 连接级属性，由 hermes 每次连接自行设置")
|
||
log(f" journal_size_limit = {jsl} ← 应为 67108864")
|
||
log(f" journal_mode = {jm2} ← 应为 wal（模式15：delete 库绝不放行 gateway 启动）")
|
||
if jm2 != "wal":
|
||
report["warnings"].append(f"journal_mode_not_wal={jm2}")
|
||
if not dry_run:
|
||
fatal = True
|
||
log(" 🔴 验证失败：journal_mode != wal → 硬失败（ExecStartPre 拒绝启动）")
|
||
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}")
|
||
|
||
report["fatal"] = fatal
|
||
log(f"\n=== 完成（{len(report['warnings'])} 个警告，{len(report['actions'])} 个动作）===")
|
||
if report["warnings"]:
|
||
for w in report["warnings"]:
|
||
log(f" ⚠️ {w}")
|
||
if dry_run:
|
||
# 只检模式：有警告返回 1（诊断语义，供人工/巡检读取）
|
||
return 0 if not report["warnings"] else 1
|
||
# apply（ExecStartPre 门禁）模式：仅 fatal 拦启动；良性警告（fts_mismatch 等由
|
||
# hermes 运行期自愈）不拦，避免启动前误杀可用库。
|
||
return 1 if fatal else 0
|
||
|
||
|
||
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="强制只体检（兼容旧用法）")
|
||
p.add_argument("--db", default=None, help="指定目标 DB 路径（默认 ~/.hermes/state.db；模拟验证用副本）")
|
||
args = p.parse_args()
|
||
if args.db:
|
||
global DB_PATH, WAL_PATH, SHM_PATH
|
||
DB_PATH = Path(args.db)
|
||
WAL_PATH = DB_PATH.with_suffix(DB_PATH.suffix + "-wal")
|
||
SHM_PATH = DB_PATH.with_suffix(DB_PATH.suffix + "-shm")
|
||
log(f"目标 DB 覆盖: {DB_PATH}")
|
||
# 🔴 安全默认反转（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()) |