fix: 归档脚本微审修复(fsync/文件名防覆盖/VACUUM容错/导出删除对账)

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
浅念
2026-10-14 09:34:19 +08:00
parent 3a3292c239
commit 78755d491b
2 changed files with 90 additions and 24 deletions
+54 -17
View File
@@ -14,11 +14,19 @@ messages → conversations, 单事务), 删了行才 VACUUM。导出/删除集
'+00:00' 后缀——同族字符串字典序比较对两者皆单调(差异位在 '.'/'+', '+00:00' 后缀——同族字符串字典序比较对两者皆单调(差异位在 '.'/'+',
仅整秒边界有微差, 天级截止无实际影响); 仅整秒边界有微差, 天级截止无实际影响);
- 导出先落盘后删除: 文件写失败 → 数据一字节不动(多一份孤儿导出无害, - 导出先落盘后删除: 文件写失败 → 数据一字节不动(多一份孤儿导出无害,
反之是不可逆丢失); 目录懒建(无可归档/失败时不产生任何文件); 反之是不可逆丢失); 落盘 = 临时文件 fsync 后原子 replace(崩溃级: 断电
窗口内要么旧文件在、要么新文件完整, "导出先于删除"不击穿); 目录懒建
(无可归档/失败时不产生任何文件);
- 导出/删除行数对账: 两边计数差一行即非零退出(删除已提交, 不一致必须
人工响, 不静默放过);
- --dry-run 只列不动: 库零变化、不导出、不 VACUUM; - --dry-run 只列不动: 库零变化、不导出、不 VACUUM;
- VACUUM 被占用锁(SQLITE_BUSY)不判归档失败: 导出/删除已安全完成, 提示
稍后手动执行;
- archives 目录 = tenants 目录的兄弟 data/archives(同 registry - archives 目录 = tenants 目录的兄弟 data/archives(同 registry
uploads_root 的派生纪律: 生产 <repo>/tenants → <repo>/data/archives, uploads_root 的派生纪律: 生产 <repo>/tenants → <repo>/data/archives,
测试注入 tmp tenants → tmp 内隔离; data/ 在 .gitignore 内天然覆盖)。 测试注入 tmp tenants → tmp 内隔离; data/ 在 .gitignore 内天然覆盖);
导出文件名带时分秒(<日期>-<时分秒>-<N>d.json), 同日同参重跑不覆盖
唯一副本。
用法(Windows 须 PYTHONUTF8=1, 输出含中文): 用法(Windows 须 PYTHONUTF8=1, 输出含中文):
PYTHONUTF8=1 python scripts/archive.py --tenant <slug> \ PYTHONUTF8=1 python scripts/archive.py --tenant <slug> \
@@ -29,6 +37,7 @@ from __future__ import annotations
import argparse import argparse
import asyncio import asyncio
import json import json
import os
import sqlite3 import sqlite3
import sys import sys
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
@@ -139,12 +148,15 @@ async def _export_document(conn, cutoff: str) -> dict:
def _write_export(path: Path, doc: dict) -> None: def _write_export(path: Path, doc: dict) -> None:
"""先落盘后删库的'落盘'端: 目录懒建 + 原子替换(临时文件 + replace, """先落盘后删库的'落盘'端: 目录懒建 + 临时文件写完 flush+fsync 再
中断不留半份导出)。""" 原子 replace——进程级升级到崩溃级(断电窗口内要么旧文件在、要么新文件
完整落盘, '导出先于删除'的承诺不被断电击穿)。"""
path.parent.mkdir(parents=True, exist_ok=True) path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(".json.tmp") tmp = path.with_suffix(".json.tmp")
tmp.write_text( with open(tmp, "w", encoding="utf-8") as f:
json.dumps(doc, ensure_ascii=False, indent=2), encoding="utf-8") json.dump(doc, f, ensure_ascii=False, indent=2)
f.flush()
os.fsync(f.fileno())
tmp.replace(path) tmp.replace(path)
@@ -191,7 +203,8 @@ async def run(args) -> None:
return return
export_path = archives_dir / args.tenant / ( export_path = archives_dir / args.tenant / (
f"{datetime.now(timezone.utc).strftime('%Y%m%d')}-{days}d.json") f"{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}"
f"-{days}d.json")
deleted: dict[str, int] = {} deleted: dict[str, int] = {}
if not old_ids: if not old_ids:
print("归档会话: 0 条") print("归档会话: 0 条")
@@ -206,7 +219,11 @@ async def run(args) -> None:
"days": days, "cutoff": cutoff, "days": days, "cutoff": cutoff,
"conversation_count": len(old_ids)} "conversation_count": len(old_ids)}
doc.update(await _export_document(conn, cutoff)) doc.update(await _export_document(conn, cutoff))
_write_export(export_path, doc) try:
_write_export(export_path, doc)
except OSError as exc:
raise SystemExit(
f"导出写盘失败({exc}): 中止, 库零变化(未删除)") from exc
# ---- 清库(倒 FK, 单事务; 子查询同源截止串) ------------------------ # ---- 清库(倒 FK, 单事务; 子查询同源截止串) ------------------------
for table in DELETE_ORDER: for table in DELETE_ORDER:
@@ -218,20 +235,40 @@ async def run(args) -> None:
deleted[table] = cur.rowcount deleted[table] = cur.rowcount
await conn.commit() await conn.commit()
# ---- VACUUM(删了行才跑; 已 commit, 事务外) ------------------------ # ---- 对账: 导出行数 vs 删除行数(删除已提交, 差一行必须人工响) ------
pages_before = await _scalar(conn, "PRAGMA page_count") mismatched = []
await conn.execute("VACUUM") for table in DELETE_ORDER:
pages_after = await _scalar(conn, "PRAGMA page_count") exported = (len(doc[table]) if table == "conversations"
else sum(len(rows) for rows in doc[table].values()))
# ---- 报告 ---------------------------------------------------------- if exported != deleted[table]:
mismatched.append(f"{table}(导出={exported} "
f"删除={deleted[table]})")
if mismatched:
raise SystemExit("导出/删除对账不一致: "
+ ", ".join(mismatched)
+ "; 导出文件已落盘, 请人工核查")
print(f"导出: {export_path}")
print(f"归档会话: {len(old_ids)} 条; 各表删除行数(倒 FK 序):") print(f"归档会话: {len(old_ids)} 条; 各表删除行数(倒 FK 序):")
for table in DELETE_ORDER: for table in DELETE_ORDER:
print(f" {table}: {deleted[table]}") print(f" {table}: {deleted[table]}")
print(f"导出: {export_path}") print("导出/删除对账: 一致")
# ---- VACUUM(删了行才跑; 已 commit, 事务外; 占用锁不判归档失败) ----
try:
pages_before = await _scalar(conn, "PRAGMA page_count")
await conn.execute("VACUUM")
pages_after = await _scalar(conn, "PRAGMA page_count")
print(f"VACUUM: 已执行(page_count {pages_before} "
f"→ {pages_after})")
print("归档结论: 导出已落盘, 删除已提交, VACUUM 完成")
except sqlite3.OperationalError as exc:
print(f"VACUUM: 被占用锁跳过({exc})——归档已完成, "
f"请稍后手动执行 VACUUM")
print("归档结论: 导出已落盘, 删除已提交")
# ---- 报告收尾: 库文件前后大小 --------------------------------------
size_after = tenant_db.stat().st_size size_after = tenant_db.stat().st_size
print(f"库文件: {size_before} → {size_after} 字节") print(f"库文件: {size_before} → {size_after} 字节")
print(f"VACUUM: 已执行(page_count {pages_before} → {pages_after})")
print("归档结论: 导出已落盘, 删除已提交, VACUUM 完成")
def main() -> None: def main() -> None:
+36 -7
View File
@@ -6,7 +6,9 @@
2. --dry-run 只列不动: 库零变化、不导出、不 VACUUM; 2. --dry-run 只列不动: 库零变化、不导出、不 VACUUM;
3. 清理后 VACUUM → page_count 减少(PRAGMA page_count 断言, 比文件大小稳); 3. 清理后 VACUUM → page_count 减少(PRAGMA page_count 断言, 比文件大小稳);
4. 导出 JSON 含被删行的完整字段(五表按会话分组); 4. 导出 JSON 含被删行的完整字段(五表按会话分组);
5. 门禁同 import_legacy: 租户须已 bootstrap(platform.db 有 slug)。 5. 门禁同 import_legacy: 租户须已 bootstrap(platform.db 有 slug);
6. 导出写盘失败 → 显性中止, 库零变化(导出先于删除的底线);
导出/删除行数对账一致(文件名带时分秒, 同日重跑不覆盖)。
测试方式(照 test_import_legacy.py): import + sys.argv 注入, 非 subprocess; 测试方式(照 test_import_legacy.py): import + sys.argv 注入, 非 subprocess;
夹具全在 tmp(archives 目录取 tenants 目录兄弟 data/ 派生, 同 registry 夹具全在 tmp(archives 目录取 tenants 目录兄弟 data/ 派生, 同 registry
@@ -17,6 +19,7 @@ uploads_root 的布局纪律, tmp 内自动隔离)。时间形态双覆盖: ORM
import hashlib import hashlib
import importlib.util import importlib.util
import json import json
import re
import sqlite3 import sqlite3
import sys import sys
from datetime import datetime, timedelta, timezone from datetime import datetime, timedelta, timezone
@@ -266,9 +269,10 @@ class Env:
def archives_dir(self) -> Path: def archives_dir(self) -> Path:
return self.tmp_path / "data" / "archives" return self.tmp_path / "data" / "archives"
def export_path(self, days: int) -> Path: def export_files(self, days: int) -> list[Path]:
date = datetime.now(timezone.utc).strftime("%Y%m%d") """<日期>-<时分秒>-<N>d.json(文件名带时分秒防同日重跑覆盖,
return self.archives_dir / "jiancai" / f"{date}-{days}d.json" 测试用 glob 取, 免跨秒竞态)。"""
return sorted(self.archives_dir.glob(f"jiancai/*-{days}d.json"))
# ---- 1. 归档: 旧会话五表清空, 近期不动 ------------------------------------------ # ---- 1. 归档: 旧会话五表清空, 近期不动 ------------------------------------------
@@ -285,7 +289,11 @@ def test_archive_old_conversations(tmp_path, monkeypatch, capsys):
assert _ids(env.tenant_db, table, CONV_ID_COL[table]) == KEPT[table], \ assert _ids(env.tenant_db, table, CONV_ID_COL[table]) == KEPT[table], \
table table
assert {OLD1, OLD2} & _ids(env.tenant_db, "conversations") == set() assert {OLD1, OLD2} & _ids(env.tenant_db, "conversations") == set()
assert env.export_path(365).is_file() # 导出落 data/archives/<tenant>/ exports = env.export_files(365)
assert len(exports) == 1 # 导出落 data/archives/<tenant>/
assert re.match(r"^\d{8}-\d{6}-365d\.json$", exports[0].name), \
"文件名须为 <日期>-<时分秒>-<N>d.json(同日重跑不覆盖)"
assert "导出/删除对账: 一致" in out
assert "导出: " in out assert "导出: " in out
@@ -354,7 +362,7 @@ def test_export_complete(tmp_path, monkeypatch, capsys):
run_archive(monkeypatch, *env.archive_args()) run_archive(monkeypatch, *env.archive_args())
capsys.readouterr() capsys.readouterr()
path = env.export_path(365) path = env.export_files(365)[0]
doc = json.loads(path.read_text(encoding="utf-8")) doc = json.loads(path.read_text(encoding="utf-8"))
assert doc["tenant"] == "jiancai" assert doc["tenant"] == "jiancai"
@@ -436,6 +444,27 @@ def test_no_old_conversations_noop(tmp_path, monkeypatch, capsys):
assert "归档会话: 0 条" in out assert "归档会话: 0 条" in out
assert "VACUUM: 未执行" in out assert "VACUUM: 未执行" in out
assert not env.export_path(365).exists() assert env.export_files(365) == []
assert not env.archives_dir.exists() assert not env.archives_dir.exists()
assert _sha256(env.tenant_db) == sha # 库文件字节级零变化 assert _sha256(env.tenant_db) == sha # 库文件字节级零变化
# ---- 导出写失败: 库零变化 -------------------------------------------------------
def test_export_write_failure_leaves_db_untouched(tmp_path, monkeypatch, capsys):
"""导出写盘失败(磁盘满/权限)→ 显性中止, 库零变化: 五表行级快照不变,
不产生导出文件, 删除从未发生。'导出先于删除'的底线用例。"""
env = Env(tmp_path, monkeypatch)
_seed(env.tenant_db)
snapshot = {t: _full_rows(env.tenant_db, t) for t in FIVE_TABLES}
def boom(path, doc):
raise OSError(28, "No space left on device")
monkeypatch.setattr(archive, "_write_export", boom)
with pytest.raises(SystemExit, match="导出写盘失败"):
run_archive(monkeypatch, *env.archive_args())
assert {t: _full_rows(env.tenant_db, t) for t in FIVE_TABLES} == snapshot
assert env.export_files(365) == []
assert not (env.archives_dir / "jiancai").exists()