diff --git a/backend/app/data/catalog.py b/backend/app/data/catalog.py index 9c674cb4..6ca00287 100644 --- a/backend/app/data/catalog.py +++ b/backend/app/data/catalog.py @@ -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) diff --git a/backend/app/data/catalog_profile.py b/backend/app/data/catalog_profile.py index 76a03c4e..0636cd70 100644 --- a/backend/app/data/catalog_profile.py +++ b/backend/app/data/catalog_profile.py @@ -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 diff --git a/backend/app/data/catalog_profile_jobs.py b/backend/app/data/catalog_profile_jobs.py index 78b9511e..3e2735d1 100644 --- a/backend/app/data/catalog_profile_jobs.py +++ b/backend/app/data/catalog_profile_jobs.py @@ -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"] diff --git a/backend/tests/test_catalog_profile_jobs.py b/backend/tests/test_catalog_profile_jobs.py index 8fb45a28..dd0bd2f4 100644 --- a/backend/tests/test_catalog_profile_jobs.py +++ b/backend/tests/test_catalog_profile_jobs.py @@ -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"])