feat(state-db): 看门狗 + 止血脚本
- 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
This commit is contained in:
小唯 A06
parent
a25489e105
commit
83ad0bcb1a
|
|
@ -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())
|
||||
|
|
@ -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())
|
||||
Loading…
Reference in New Issue