剖析收尾照实说:没写进目录的检查项不再标「完成」,目录和表的剖析结果同一个事务提交

复核 R1。目录只在一张表结束时由 _record 写一次,进程被杀(启动时标中断)、任务整体出错(剖析到一半快照被改动)时,
正在剖析的那张表没走到这一步,可它已经做完的检查项仍记 done:检查清单上标着完成,findings 是空的,目录里那几条关系
还是推断。_close_tasks 只改 pending / running,于是把没写进去的结论说成了做完。

- _close_tasks:还没结束的表上 done 的检查项也记为跳过,原因同这张表(中断 cancelled、出错 error),说明是
  「{原因},查完了但结论没有写进数据目录」,算进 unfinished,重试时带上这张表;这张表的 error 写明「已查完的 N 项
  没有写进数据目录,剩下 M 项没查」。跳过的项按检查项列全:进行中的表只随时落了检查项,按规则不查的(遮罩……)以前
  收尾时不进跳过的项。优雅停止、关停不走这条路,正在剖析的那张表做完的项照常写进目录(error 说「照常写进了数据目录」),
  前端可以据此分开说。
- 目录和这张表的剖析结果改成同一个事务:catalog.write_entry 加 commit=False(只写进事务、冲突照样回滚),
  catalog_profile._record 写目录不提交,recorded(_save_table)也不提交,最后一起提交。以前是两次提交,进程在中间没了
  会出现目录写了、剖析记录却说没写的情形,上面那句「没写进目录」就不照实。
- 不改成每项做完就写目录:每写一次目录升一个版本,表结构快照的哈希跟着变,和人工编辑撞乐观锁的机会也成倍增加;
  设计文档定的是整张表结束时并一次。

测试:启动标中断那条改断言(查完的码值记 skipped / cancelled、error 带「已查完的 1 项」);新增剖析到一半快照被改动
(4 条关系查过了、全记 skipped / error、目录不动、重试带上 visits);新增记结果那一步出错时目录跟着回滚(换回两次提交
这条就挂)。
This commit is contained in:
ThearchyHelios
2026-10-15 19:11:28 +08:00
parent f375101c5e
commit 35c83d3001
4 changed files with 113 additions and 22 deletions
+7 -3
View File
@@ -421,9 +421,12 @@ async def read_catalog(session: AsyncSession, source_id: str) -> dict[str, Catal
async def write_entry(session: AsyncSession, source_id: str, table: str, notes: dict[str, Any], *,
if_version: int, actor: str | None) -> CatalogEntry:
if_version: int, actor: str | None, commit: bool = True) -> CatalogEntry:
"""整份写入一张表的目录(乐观锁),提交事务,返回写入后的目录。
commit=False 只写进当前事务、不提交,由调用方和它自己的写入一起提交:剖析一张表结束时,目录和这张表的剖析结果要么
都写进去、要么都没写(进程在两次提交之间没了,剖析结果会说「没写进目录」,目录里其实已经写了)。冲突时照样回滚。
- if_version 是写入方读到的版本,还没有目录时传 0;和库里的不一致抛 CatalogConflict,什么都不写。
比对和写入在同一条 UPDATE … WHERE version = ? 里完成:两个请求同时读到第 3 版,只有一个写得进去。
- 内容和库里的完全相同时不写、不升版本:冻结进表结构快照的是「版本 + 内容」,空写一次也升版本,
@@ -446,7 +449,7 @@ async def write_entry(session: AsyncSession, source_id: str, table: str, notes:
row = CatalogNote(source_id=source_id, table_name=table, notes=stored, version=1, updated_by=actor)
session.add(row)
try:
await session.commit()
await (session.commit() if commit else session.flush())
except IntegrityError as e:
# 唯一约束 (source_id, table_name):另一个请求抢先建了这张表的目录
await session.rollback()
@@ -461,7 +464,8 @@ async def write_entry(session: AsyncSession, source_id: str, table: str, notes:
if result.rowcount != 1:
await session.rollback()
raise CatalogConflict(table, current + 1)
await session.commit()
if commit:
await session.commit()
await session.refresh(row)
return _entry(row)
+12 -5
View File
@@ -811,7 +811,8 @@ class TableHooks:
- progress:一项检查(含开头的估行数)开始、结束时;result 里的 checks、current、queries、query_log 是这时的样子;
- query:每发出(或被守卫拒掉)一条查询之后;
- recorded:整张表结束,和写目录在同一次库操作里(session 是那一次的,shield 起来,停止时也不会被打断);
- recorded:整张表结束,和写目录在同一个事务里(session 是那一次的,shield 起来,停止时也不会被打断;不用自己提交,
写完由 _record 和目录一起提交);
- stop_text:停止时的说法(「剖析已停止」「服务重启,剖析中断」),停止之后才调。
"""
@@ -1617,7 +1618,8 @@ async def profile_table(cx: ProfileContext, table: str, *,
停下时剩下的项记 failed / slow,error 是止损的那一句。
- 停止(任务被 cancel):正在跑的那条查询由数据库按单条时限截止;已做完的项照常写进目录,没做的记 cancelled、保留旧
结论;写完再把停止传下去(CancelledError)。
- 整张表结束把结论并进目录(乐观锁,撞上别人刚改过重读重并,三次仍冲突记 failed),和 hooks.recorded 在同一次库操作里。
- 整张表结束把结论并进目录(乐观锁,撞上别人刚改过重读重并,三次仍冲突记 failed),和 hooks.recorded 在同一个事务里
提交。
db 是执行库操作的:db(fn, *args) 开一个会话调 fn(session, *args)。后台任务给它的 SerialWriter(一条一条、shield),
不给就各开一个会话。快照被改动(SnapshotTampered)原样抛出,不写目录:整个任务都该停。
@@ -1685,7 +1687,10 @@ def _settle_status(result: TableResult) -> None:
async def _record(session: AsyncSession, source_id: str, table: str, out: _TableOutcome | None, result: TableResult,
actor: str | None, recorded: Callable[[AsyncSession, TableResult], Awaitable[None]] | None) -> None:
"""整张表结束:把结论并进目录(out 为 None 的不写:表结构里没有、视图),定下表的状态,再交给 recorded(同一个会话)。
修改时间取这一次写入的时间。"""
修改时间取这一次写入的时间。
目录和 recorded 写的这张表的结果在同一个事务里提交:进程在中间没了,两样都没写进去,下次启动收尾时这张表做完的项
照实记「没写进目录」(catalog_profile_jobs._close_tasks),不会出现目录写了、剖析记录却说没写的情形。"""
if out is not None:
try:
await _write(session, source_id, table, out, result, at=catalog.now_iso(), actor=actor)
@@ -1699,11 +1704,12 @@ async def _record(session: AsyncSession, source_id: str, table: str, out: _Table
_settle_status(result)
if recorded is not None:
await recorded(session, result)
await session.commit()
async def _write(session: AsyncSession, source_id: str, table: str, out: _TableOutcome, result: TableResult, *,
at: str, actor: str | None) -> None:
"""并入并写入(乐观锁)。撞上别人刚改过就重读重并,最多三次;三次仍冲突这张表记 failed。"""
"""并入并写入(乐观锁)。撞上别人刚改过就重读重并,最多三次;三次仍冲突这张表记 failed。写进事务、不提交(见 _record)。"""
for _attempt in range(3):
entry = await catalog.read_entry(session, source_id, table)
version = entry.version if entry else 0
@@ -1714,7 +1720,8 @@ async def _write(session: AsyncSession, source_id: str, table: str, out: _TableO
if entry is None and not notes:
return # 什么也没得出,不建空目录
try:
written = await catalog.write_entry(session, source_id, table, notes, if_version=version, actor=actor)
written = await catalog.write_entry(session, source_id, table, notes, if_version=version, actor=actor,
commit=False)
except catalog.CatalogConflict:
continue
result.version = written.version
+41 -11
View File
@@ -12,7 +12,7 @@ data/catalog_job_runtime.py),逐表剖析(引擎在 data/catalog_profile.p
起草在前时 waiting_for 是 "draft"。建任务时顺手清掉超出保留数的旧任务(prune)。
- **在后台剖析**(manager.start):轮到了才开始。每张表开始前重读数据源——停用了、关掉了剖析就停下,剩下的表 cancelled、
error 写明;遮罩按最新的设置——再逐表调 catalog_profile.profile_table(它读的是那时目录里的最新内容:起草刚提的关系也
核对)。检查项、正在做哪一项、查询数、发过的查询随时落库;整张表的结果和写目录在同一次库操作里,停止时也不会被打断。
核对)。检查项、正在做哪一项、查询数、发过的查询随时落库;整张表的结果和写目录在同一个事务里,停止时也不会被打断。
- **任务级止损**(替代以前的整次总量上限):连续 3 条查询失败、最近 10 条里 7 条超时,整个任务停下(stop_reason、error),
正在做的那张表写入已做完的项,后面的表 cancelled、error 带同一句原因;最近 6 条里 4 条超时时 notice 有一句提醒,好转后
清掉。任务最后的状态按逐表算:有 failed、partial、cancelled 的表是 partial,否则 succeeded;任务整体出错(数据源没了、
@@ -21,7 +21,8 @@ data/catalog_job_runtime.py),逐表剖析(引擎在 data/catalog_profile.p
目录、没做的保留旧结论,后面的表 cancelled,任务 cancelled。
- **重试**(retry_tables + create_job):新任务,只含原任务里 failed、partial、cancelled 的表(按原来的顺序),范围 selected,
retry_of 记原任务;设置取数据源当前的。
- **服务重启**(startup):排队、进行中的标 interrupted,没结束的表 cancelled、error「服务重启,剖析中断」。不自动续跑:
- **服务重启**(startup):排队、进行中的标 interrupted,没结束的表 cancelled、error「服务重启,剖析中断」(进程没了时正在
剖析的那张表,查完的项也没写进目录,照实写明,见文末)。不自动续跑:
复制出来的数据目录启动时不能自己去查业务库,界面让人点「重试」。关停时(manager.shutdown)停掉进行中的,同样标中断。
启动时还一次性迁移剖析设置的旧键(catalog_profile.migrate_settings)。
- **保留**(prune):每个源保留最近 KEEP_JOBS 个结束了的剖析任务,更早的删掉;某张表最近一次剖析所在的任务不删,表详情的
@@ -30,7 +31,9 @@ data/catalog_job_runtime.py),逐表剖析(引擎在 data/catalog_profile.p
的「上次剖析」)、due_reason(这张表要不要剖析、为什么)。
没结束的表收尾时(停止、中断、整体出错)检查项里还排着的、正在做的记为跳过(原因 cancelled 或 error),不留「排队中」的
检查项在一张已经结束的表上。
检查项在一张已经结束的表上。目录只在一张表结束时和它的结果一起写(同一个事务):收尾时还没结束的表,查完的项结论也没
写进目录(进程被杀、快照被改动),同样记为跳过、写明「查完了但结论没有写进数据目录」,算没做完,重试时带上这张表。
停止、关停这类能收尾的停下不走这条路:正在剖析的那张表做完的项照常写进目录(catalog_profile.profile_table)。
"""
from __future__ import annotations
@@ -68,9 +71,17 @@ SOURCE_DISABLED = "数据源「{name}」已停用,剖析已停止"
PROFILE_TURNED_OFF = "数据源「{name}」已关闭数据剖析,剖析已停止"
#: 没结束的检查项在收尾时的说明({why} 是停止、中断、出错的原因)
_UNCHECKED = "{why},未检查"
#: 查完了、结论却没写进目录的检查项:这张表没走到整张表结束、写目录那一步(进程没了、任务整体出错)
_UNSAVED = "{why},查完了但结论没有写进数据目录"
_CHECK_OPEN = ("pending", "running")
def unsaved_error(why: str, done: int, rest: int) -> str:
"""没走到写目录那一步的表的整句原因:查完的几项没写进目录,剩下几项没查。"""
text = f"{why}:已查完的 {done} 项没有写进数据目录"
return text + f",剩下 {rest} 项没查" if rest else text
class ProfileBusy(Exception):
"""这个数据源已经有排队或进行中的剖析任务。message 给人看(写明进行到第几张)。"""
@@ -525,22 +536,42 @@ async def prune(session: AsyncSession, source_id: str, *, keep: int | None = Non
async def _close_tasks(session: AsyncSession, job_id: str, task_status: str, task_error: str | None,
*, why: str) -> None:
"""还没结束的表记 task_status(原因 task_error);它们的检查项里还排着的、正在做的记为跳过(停止、中断是 cancelled,
出错是 error,说明是「{why},未检查」),计进跳过的项和没做完的项。"""
出错是 error,说明是「{why},未检查」),计进跳过的项和没做完的项。
还没结束的表没走到整张表结束那一步:目录只在那一步和这张表的结果一起写(catalog_profile._record,同一个事务),
所以这里已经查完(done)的项结论也没写进目录——进程没了、任务整体出错(快照被改动……)都是这样。它们同样记为跳过、
原因同上,说明是「{why},查完了但结论没有写进数据目录」,算没做完;这张表的原因写明查完的几项没写进目录。不能留着
「完成」:检查清单上标着完成、目录里却没有,看的人会以为结论已经在目录里了。"""
now = utcnow()
reason = "cancelled" if task_status == "cancelled" else "error"
detail = _UNCHECKED.format(why=why)
detail, unsaved = _UNCHECKED.format(why=why), _UNSAVED.format(why=why)
tasks = (await session.execute(
select(CatalogProfileTask).where(CatalogProfileTask.job_id == job_id, CatalogProfileTask.status.in_(OPEN))
)).scalars()
for task in tasks:
checks = [dict(c) for c in task.checks or [] if isinstance(c, dict)]
skipped = list(task.skipped or [])
listed = {(s.get("kind"), s.get("target"), s.get("path")) for s in skipped if isinstance(s, dict)}
done = rest = 0
for check in checks:
if check.get("state") in _CHECK_OPEN:
state = check.get("state")
if state == "done":
done += 1
check.update(state="skipped", reason=reason, detail=unsaved)
elif state in _CHECK_OPEN:
rest += 1
check.update(state="skipped", reason=reason, detail=detail)
skipped.append(ProfileSkip(check.get("kind") or "", check.get("target") or "", reason, detail,
elif state != "skipped":
continue
# 跳过的项按检查项列全:进行中的表只随时落了检查项,按规则不查的(遮罩……)也要进跳过的项
key = (check.get("kind"), check.get("target"), check.get("path"))
if key not in listed:
listed.add(key)
skipped.append(ProfileSkip(check.get("kind") or "", check.get("target") or "",
check.get("reason") or "", check.get("detail") or "",
check.get("path")).to_dict())
task.status, task.error, task.finished_at, task.current = task_status, task_error, now, None
error = unsaved_error(task_error or why, done, rest) if done else task_error
task.status, task.error, task.finished_at, task.current = task_status, error, now, None
task.checks, task.skipped = checks or None, skipped or None
task.skipped_count = len(skipped)
task.unfinished = sum(1 for s in skipped if s.get("reason") in UNFINISHED_REASONS)
@@ -776,7 +807,7 @@ class _ProfileRun(runtime.JobRun):
await session.commit()
async def _save_table(self, session: AsyncSession, seq: int, result: TableResult) -> None:
"""一张表结束(和写目录在同一次库操作里):整张表的结果。"""
"""一张表结束:整张表的结果。和写目录在同一个事务里,由 catalog_profile._record 一起提交,这里不提交。"""
await session.execute(
self._task(CatalogProfileTask.seq == seq, CatalogProfileTask.status.in_(OPEN))
.values(status=result.status, checks=[dict(c) for c in result.checks] or None, current=None,
@@ -790,7 +821,6 @@ class _ProfileRun(runtime.JobRun):
if values := self._counters():
await session.execute(update(CatalogProfileJob).where(CatalogProfileJob.id == self.job_id).values(**values)
.execution_options(synchronize_session=False))
await session.commit()
async def _finish(self, session: AsyncSession) -> None:
"""剖析完:万一还有没收尾的表(进度回调写库失败之类)记失败;有 failed、partial、cancelled 的表是 partial,否则
@@ -820,4 +850,4 @@ __all__ = ["INTERRUPTED_ERROR", "KEEP_JOBS", "MAX_TABLES", "NOTHING_AT_ALL", "NO
"ProfileBusy", "ProfileFinished", "ProfileJobs", "ProfileNothing", "active_job", "busy_message", "cancel",
"create_job", "due_reason", "ensure_free", "job_brief", "job_dict", "job_out", "last_detail",
"last_profiles", "latest_job", "load", "load_task", "manager", "prune", "public_last", "retry_tables",
"startup", "task_detail", "task_row", "task_rows", "waiting_for"]
"startup", "task_detail", "task_row", "task_rows", "unsaved_error", "waiting_for"]
+53 -3
View File
@@ -304,13 +304,63 @@ async def test_startup_marks_unfinished_jobs_interrupted_and_does_not_resume(cli
assert got["status"] == "interrupted" and got["error"] == "服务重启,剖析中断" and got["finished_at"]
assert by["channels"]["status"] == "ok" and by["channels"]["error"] is None
assert by["visits"]["status"] == by["gates"]["status"] == "cancelled"
assert by["visits"]["error"] == by["gates"]["error"] == "服务重启,剖析中断"
assert by["gates"]["error"] == "服务重启,剖析中断"
# 进程没了,visits 没走到写目录那一步:查完的码值也没写进目录,不能还标「完成」
assert by["visits"]["error"] == "服务重启,剖析中断:已查完的 1 项没有写进数据目录,剩下 1 项没查"
visits = await _detail(client, sid, job_id, 1)
assert [c["state"] for c in visits["checks"]] == ["done", "skipped"]
assert visits["checks"][1]["reason"] == "cancelled" and visits["unfinished"] == 1
assert [(c["state"], c["reason"]) for c in visits["checks"]] == [("skipped", "cancelled"), ("skipped", "cancelled")]
assert visits["checks"][0]["detail"] == "服务重启,剖析中断,查完了但结论没有写进数据目录"
assert visits["checks"][1]["detail"] == "服务重启,剖析中断,未检查"
assert visits["unfinished"] == visits["skipped_count"] == 2 and visits["findings"] == []
assert not catalog_profile_jobs.manager.running(job_id), "不自动续跑"
async def test_a_job_failing_mid_table_does_not_claim_the_finished_checks(client, make_source, scenic_db,
monkeypatch):
"""visits 的 4 条关系核对完、取码值时快照被改动:任务整体失败,visits 没走到写目录那一步。核对完的关系不能还标「完成」:
记为跳过(error)、写明没写进目录,算没做完;目录里那几条关系原样不动;重试带上 visits。"""
from app.data.engine import SnapshotTampered
sid = await make_source(scenic_db, options=_on(), draft=["visits"])
async def tampered(*_args, **_kwargs):
raise SnapshotTampered("数据文件与登记的版本不一致,可能被修改过,已拒绝查询")
monkeypatch.setattr(catalog_profile, "_check_codes", tampered)
body = await _profile(client, sid, ["visits"])
visits = body["tables"][0]
assert body["status"] == "failed" and visits["status"] == "failed"
relations = [c for c in visits["checks"] if c["kind"] == "relation"]
assert len(relations) == 4 and all(c["queries"] > 0 for c in relations), "关系确实查过了"
assert all((c["state"], c["reason"]) == ("skipped", "error") for c in visits["checks"])
assert relations[0]["detail"] == "数据文件与登记的版本不一致,可能被修改过,已拒绝查询,查完了但结论没有写进数据目录"
assert visits["error"] == "数据文件与登记的版本不一致,可能被修改过,已拒绝查询:已查完的 4 项没有写进数据目录,剩下 2 项没查"
assert visits["unfinished"] == 6 and visits["findings"] == [] and visits["query_log"], "发过的查询照样留痕"
notes = (await _notes(client, sid, "visits"))["notes"]
assert _relation(notes, "gate_id")["status"] == "proposed" and _relation(notes, "gate_id")["source"] == "name"
async with SessionLocal() as session:
failed = await session.get(CatalogProfileJob, body["id"])
assert await catalog_profile_jobs.retry_tables(session, failed) == ["visits"]
async def test_the_catalog_and_the_table_result_commit_together(client, make_source, scenic_db, monkeypatch):
"""目录和这张表的剖析结果在同一个事务里:记结果那一步出错,目录也没写进去,收尾时照实说「没写进目录」。"""
sid = await make_source(scenic_db, options=_on(), draft=["visits"])
real = catalog_profile_jobs._ProfileRun._save_table
async def broken(self, session, seq, result):
await real(self, session, seq, result)
raise RuntimeError("disk I/O error")
monkeypatch.setattr(catalog_profile_jobs._ProfileRun, "_save_table", broken)
body = await _profile(client, sid, ["visits"])
visits = body["tables"][0]
assert body["status"] == "failed" and visits["status"] == "failed"
assert visits["error"].endswith("已查完的 6 项没有写进数据目录")
notes = (await _notes(client, sid, "visits"))["notes"]
assert _relation(notes, "gate_id")["status"] == "proposed", "目录跟着结果一起回滚,没写进去"
async def test_retry_takes_only_unfinished_tables(client, make_source, scenic_db, monkeypatch):
sid = await make_source(scenic_db, options=_on(table_max_queries=2), draft=["visits"])
first = await _profile(client, sid, ["visits", "channels", "nope"])