From 1e58862bf08ae6cb059c57bfdd9b9a0302d105bc Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Thu, 15 Oct 2026 18:10:50 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8A=BD=E5=87=BA=E6=95=B0=E6=8D=AE=E7=9B=AE?= =?UTF-8?q?=E5=BD=95=E5=90=8E=E5=8F=B0=E4=BB=BB=E5=8A=A1=E7=9A=84=E5=85=B1?= =?UTF-8?q?=E7=94=A8=E8=BF=90=E8=A1=8C=E6=97=B6=EF=BC=8C=E8=B5=B7=E8=8D=89?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E6=94=B9=E7=94=A8=E5=AE=83=EF=BC=9A=E4=B8=BA?= =?UTF-8?q?=E5=89=96=E6=9E=90=E6=94=B9=E6=88=90=E9=80=90=E8=A1=A8=E7=9A=84?= =?UTF-8?q?=E5=90=8E=E5=8F=B0=E4=BB=BB=E5=8A=A1=E5=81=9A=E5=87=86=E5=A4=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit data/catalog_job_runtime.py 从 catalog_jobs.py 抽出进程内任务的执行与停止（JobManager：start / stop(grace) / wait / running / shutdown；JobRun 把进队、等轮到自己、执行、按停止原因收尾、出队写一次）、同一任务的库写入一条一条来并 shield（SerialWriter）、启动时标中断（interrupt_on_startup、close_job）、状态词常量，以及每个源一条先来先到的队（SourceLanes：任务阻塞地占，同步写操作不阻塞地试占）。剖析和起草是同一类问题（关页面就没了、不能停、不留痕），生命周期各写一份迟早一边修了另一边没修；共用的是生命周期代码，表各建各的。 起草的行为不变（test_catalog_draft_jobs.py 一条不改全过），只是：ensure_idle 拆成「同种任务互斥」（ensure_draft_free，建起草任务、重试用，剖析任务不挡起草，起草任务排在它后面）和「同步写操作要求源空闲」（ensure_idle，同步起草、导入用，剖析任务也算）；同步起草、导入目录执行期间占着这个源的队，期间新建的任务排在后面；任务和清单顶上的进度带 waiting_for（起草排在剖析后面时为 profile）；排队中被停止的任务连同在队里等着的后台执行一起停掉、出队。 db/models.py 加 CatalogProfileJob、CatalogProfileTask 两张表（create_all 建，同源排队或进行中的部分唯一索引照起草的写法）；catalog_profile_jobs.py 先放剖析任务的忙碌判定，下一步补全。 --- backend/app/api/catalog.py | 24 +- backend/app/data/catalog_job_runtime.py | 400 +++++++++++++++++++++++ backend/app/data/catalog_jobs.py | 317 +++++++++--------- backend/app/data/catalog_profile_jobs.py | 37 +++ backend/app/db/models.py | 90 +++++ 5 files changed, 689 insertions(+), 179 deletions(-) create mode 100644 backend/app/data/catalog_job_runtime.py create mode 100644 backend/app/data/catalog_profile_jobs.py diff --git a/backend/app/api/catalog.py b/backend/app/api/catalog.py index 2d7da5b9..7a306d48 100644 --- a/backend/app/api/catalog.py +++ b/backend/app/api/catalog.py @@ -42,7 +42,8 @@ from app.api.coded import ( SNAPSHOT_TAMPERED, CodedHTTPException, ) from app.api.runs import actor_of -from app.data import catalog, catalog_impact, catalog_jobs, catalog_profile, introspect, table_versions +from app.data import catalog, catalog_impact, catalog_jobs, catalog_profile, catalog_profile_jobs, introspect +from app.data import table_versions from app.data import catalog_assistant, catalog_review, catalog_transfer from app.data.engine import SnapshotTampered from app.db.base import get_session @@ -317,13 +318,14 @@ async def import_catalog(source_id: str, payload: CatalogImportIn, x_actor: str 409 catalog_draft_busy；文件格式不对 422；没有表结构 409。""" row, source = await _import_source(session, source_id) try: - await catalog_jobs.ensure_idle(session, row) + async with catalog_jobs.import_write(session, row): + report = await catalog_transfer.run_import(session, row, source, payload.content, + include_guidance=payload.include_guidance, + actor=actor_of(x_actor)) except catalog_jobs.DraftBusy as e: raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e - try: - report = await catalog_transfer.run_import(session, row, source, payload.content, - include_guidance=payload.include_guidance, - actor=actor_of(x_actor)) + except catalog_profile_jobs.ProfileBusy as e: + raise CodedHTTPException(409, str(e), PROFILE_BUSY) from e except catalog_transfer.TransferInvalid as e: raise HTTPException(422, str(e)) from e return report.as_dict() @@ -398,7 +400,7 @@ async def create_draft_job(source_id: str, payload: DraftJobIn, x_actor: str | N raise HTTPException(422, str(e)) from e row, source = await _resolved(session, source_id) try: - await catalog_jobs.ensure_idle(session, row) + await catalog_jobs.ensure_draft_free(session, row) except catalog_jobs.DraftBusy as e: raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e cache = source.schema_cache or {} @@ -412,7 +414,7 @@ async def create_draft_job(source_id: str, payload: DraftJobIn, x_actor: str | N except catalog_jobs.DraftBusy as e: raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e catalog_jobs.manager.start(job.id) - return {"job": catalog_jobs.job_dict(job, tasks)} + return {"job": catalog_jobs.job_dict(job, tasks, waiting_for=await catalog_jobs.waiting_for(session, job))} @router.get("/{source_id}/catalog/draft-jobs/active") @@ -466,7 +468,7 @@ async def retry_draft_job(source_id: str, job_id: str, x_actor: str | None = Hea _, old = await _job_or_404(session, source_id, job_id) row, source = await _resolved(session, source_id) try: - await catalog_jobs.ensure_idle(session, row) + await catalog_jobs.ensure_draft_free(session, row) except catalog_jobs.DraftBusy as e: raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e try: @@ -484,7 +486,7 @@ async def retry_draft_job(source_id: str, job_id: str, x_actor: str | None = Hea except catalog_jobs.DraftBusy as e: raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e catalog_jobs.manager.start(job.id) - return {"job": catalog_jobs.job_dict(job, tasks)} + return {"job": catalog_jobs.job_dict(job, tasks, waiting_for=await catalog_jobs.waiting_for(session, job))} @router.put("/{source_id}/catalog/{table}") @@ -529,6 +531,8 @@ async def draft_catalog(source_id: str, payload: CatalogDraftIn, guidance=catalog_jobs.guidance_of(row)) except catalog_jobs.DraftBusy as e: raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e + except catalog_profile_jobs.ProfileBusy as e: + raise CodedHTTPException(409, str(e), PROFILE_BUSY) from e rows = [{"table_name": r.table, "added": r.added, "updated": r.updated, "removed": r.removed, "version": r.version, "error": r.error, "model_error": r.model_error, "model_items": r.model_items} for r in report.tables] diff --git a/backend/app/data/catalog_job_runtime.py b/backend/app/data/catalog_job_runtime.py new file mode 100644 index 00000000..970159a4 --- /dev/null +++ b/backend/app/data/catalog_job_runtime.py @@ -0,0 +1,400 @@ +"""数据目录后台任务的共用运行时：起草（data/catalog_jobs.py）和剖析（data/catalog_profile_jobs.py）各自建表、各自的步骤， +生命周期是同一套。 + +**为什么抽出来。** 剖析以前是一次同步长请求，和起草改成后台任务之前是同一类问题：关掉页面结果就没了、不能停、不留痕。 +两边要的是同一样东西——进程内跑、能停、重启标中断、写库不被停止打断、同源排队——各写一份，迟早一边修了另一边没修。 +共用的是生命周期代码，不是表：两种任务的进度完全不同（起草按段、块、模型用量，剖析按检查项、查询），合成一张 +「kind + JSON」的通用任务表要丢掉起草那些强类型列，还要迁移刚稳定的起草数据。 + +- **状态词**（ACTIVE、FINISHED、TASK_STATUSES、TASK_DONE、RETRYABLE、OPEN）：任务 queued → running → succeeded / partial + / failed / cancelled / interrupted；逐表 pending → running → ok / partial / failed / skipped / cancelled。两种任务、 + 接口、界面都用这一套词。 +- **SerialWriter**：同一个任务的库写入一条一条来，shield 起来——停止任务时不会打断写到一半的库操作；收尾前 settle 等它们 + 写完。 +- **JobRun / JobManager**：进程内一个 asyncio 任务一件事（照 app/scheduler 的写法）：start / stop(grace) / wait / running + / shutdown。JobRun.run 把「进队、等轮到自己、执行、按停止的原因收尾、出队」写一次，各种任务只写执行和收尾的那几步。 +- **SourceLanes**：每个数据源一条先来先到的队。起草、剖析都打业务库或模型、都写同一个源的目录：同源串行（先建的先跑）， + 不同源并行。任务阻塞地占：建出来是 queued，轮到了才 running。同步起草、导入目录这类同步写操作不阻塞地试占：占不到 + 说明源上正有别的事，接口回 409；占到了，执行期间新建的任务排在它后面。 +- **close_job / interrupt_on_startup**：收尾（还没结束的表记成某个状态，任务记成某个状态），和服务启动时的整理：排队、 + 进行中的标 interrupted，没结束的表记 cancelled。**不自动续跑**：复制出来的数据目录（检查脚本、验收副本）启动时不能 + 自己去调模型、查业务库，界面让人点「重试」接着做。 + +进程内登记、进程内的队：服务端是单进程部署，多进程部署时要换成库里的锁。 +""" +from __future__ import annotations + +import asyncio +import contextlib +import logging +from collections.abc import AsyncIterator, Awaitable, Callable, Iterator +from typing import Any + +from sqlalchemy import select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.base import SessionLocal, utcnow + +logger = logging.getLogger(__name__) + +#: 任务还没结束的状态（同一个源同一种任务只能有一个）、结束了的状态 +ACTIVE = ("queued", "running") +FINISHED = ("succeeded", "partial", "failed", "cancelled", "interrupted") +#: 逐表的状态，其中算「已结束」的、重试时要带上的、还没结束的 +TASK_STATUSES = ("pending", "running", "ok", "partial", "failed", "skipped", "cancelled") +TASK_DONE = ("ok", "partial", "failed", "skipped", "cancelled") +RETRYABLE = ("failed", "partial", "cancelled") +OPEN = ("pending", "running") + +#: 停止进行中的任务时最多等它收尾多久（秒）；关停服务时另给（shutdown 的 grace） +STOP_GRACE_S = 10.0 +SHUTDOWN_GRACE_S = 3.0 + + +class JobFailed(Exception): + """任务整体出错（数据源没了、快照没了、模型用不了……）：message 就是任务的 error。""" + + +# ========================================================================== +# 库写入 +# ========================================================================== + + +class SerialWriter: + """一个任务的库写入：一条一条来，每条开一个会话；shield 起来，停止任务时不会打断写到一半的库操作。""" + + def __init__(self) -> None: + self._writes: set[asyncio.Task[Any]] = set() + self._lock = asyncio.Lock() + + async def __call__(self, fn: Callable[..., Awaitable[Any]], *args: Any) -> Any: + async def go() -> Any: + async with self._lock: + async with SessionLocal() as session: + return await fn(session, *args) + + task = asyncio.ensure_future(go()) + self._writes.add(task) + task.add_done_callback(self._writes.discard) + return await asyncio.shield(task) + + async def settle(self) -> None: + """等手上的库写入都写完（停止、出错收尾之前）。""" + while self._writes: + await asyncio.gather(*list(self._writes), return_exceptions=True) + + +# ========================================================================== +# 每个源一条队 +# ========================================================================== + + +class LaneTicket: + """在一个源的队里的一个位置。kind 是占着它的是什么：draft 起草任务、profile 剖析任务、sync_draft 同步起草、 + import 导入目录；job_id 是任务的（同步写操作没有）。轮到它时 granted 置位。""" + + def __init__(self, lanes: SourceLanes, source_id: str, kind: str, job_id: str | None, + loop: asyncio.AbstractEventLoop) -> None: + self._lanes = lanes + self.source_id = source_id + self.kind = kind + self.job_id = job_id + self.loop = loop + self.granted = asyncio.Event() + self.released = False + + async def wait(self) -> None: + """等轮到自己。""" + await self.granted.wait() + + def release(self) -> None: + """出队（轮到了、没轮到都行，重复调用无害）：排在后面的第一个接着开始。""" + self._lanes._release(self) + + +class LaneBusy(Exception): + """不阻塞地试占时这个源的队里已经有人：holder 是排在最前面（正在做）的那一个。""" + + def __init__(self, holder: LaneTicket) -> None: + super().__init__(holder.kind) + self.holder = holder + + +class SourceLanes: + """每个数据源一条先来先到的队。""" + + def __init__(self) -> None: + self._queues: dict[str, list[LaneTicket]] = {} + + def _queue(self, source_id: str) -> list[LaneTicket]: + """这个源的队。别的事件循环留下的（测试每个用例一个循环）去掉：那个循环里没人会再出队了。""" + queue = self._queues.get(source_id) or [] + loop = asyncio.get_running_loop() + alive = [t for t in queue if t.loop is loop and not t.released] + if alive: + self._queues[source_id] = alive + alive[0].granted.set() + else: + self._queues.pop(source_id, None) + return alive + + def enter(self, source_id: str, kind: str, job_id: str | None = None) -> LaneTicket: + """排进队尾，马上返回（不等）。排在最前面的立刻 granted；调用方 await ticket.wait() 等轮到自己。""" + queue = self._queue(source_id) + ticket = LaneTicket(self, source_id, kind, job_id, asyncio.get_running_loop()) + queue.append(ticket) + self._queues[source_id] = queue + if len(queue) == 1: + ticket.granted.set() + return ticket + + def try_enter(self, source_id: str, kind: str) -> LaneTicket: + """队空着才占（马上就轮到）；有人在队里抛 LaneBusy。""" + queue = self._queue(source_id) + if queue: + raise LaneBusy(queue[0]) + return self.enter(source_id, kind) + + def holders(self, source_id: str) -> list[LaneTicket]: + """队里的各位，排在最前面（正在做）的在前。""" + return list(self._queue(source_id)) + + def ahead(self, source_id: str, job_id: str) -> list[str] | None: + """排在这个任务前面的各位是什么（kind，正在做的在前）；这个任务不在队里（刚建、后台还没排进来）返回 None。""" + queue = self._queue(source_id) + for i, ticket in enumerate(queue): + if ticket.job_id == job_id: + return [t.kind for t in queue[:i]] + return None + + def _release(self, ticket: LaneTicket) -> None: + if ticket.released: + return + ticket.released = True + queue = self._queues.get(ticket.source_id) + if not queue or ticket not in queue: + return + head = queue[0] is ticket + queue.remove(ticket) + if not queue: + self._queues.pop(ticket.source_id, None) + elif head: + queue[0].granted.set() + + @contextlib.asynccontextmanager + async def hold(self, source_id: str, kind: str, job_id: str | None = None) -> AsyncIterator[LaneTicket]: + """阻塞地占：排队，轮到了再进去，出来时出队。""" + ticket = self.enter(source_id, kind, job_id) + try: + await ticket.wait() + yield ticket + finally: + ticket.release() + + @contextlib.contextmanager + def try_hold(self, source_id: str, kind: str) -> Iterator[LaneTicket]: + """不阻塞地试占（同步起草、导入目录）：占不到抛 LaneBusy；占到了，执行期间新建的任务排在后面。""" + ticket = self.try_enter(source_id, kind) + try: + yield ticket + finally: + ticket.release() + + +lanes = SourceLanes() + + +# ========================================================================== +# 收尾、启动时的整理 +# ========================================================================== + + +async def close_job(session: AsyncSession, job_model: Any, task_model: Any, job_id: str, job_status: str, + job_error: str | None, task_status: str, task_error: str | None) -> None: + """收尾：还没结束的表记 task_status（原因 task_error），任务（还没结束的话）记 job_status。提交。""" + now = utcnow() + await session.execute( + update(task_model).where(task_model.job_id == job_id, task_model.status.in_(OPEN)) + .values(status=task_status, error=task_error, finished_at=now).execution_options(synchronize_session=False)) + await session.execute( + update(job_model).where(job_model.id == job_id, job_model.status.in_(ACTIVE)) + .values(status=job_status, error=job_error, finished_at=now).execution_options(synchronize_session=False)) + await session.commit() + + +async def cancel_queued(session: AsyncSession, job_model: Any, task_model: Any, job_id: str) -> bool: + """停止一个排队中的任务：任务和没结束的表直接记 cancelled。已经被后台认领（开始了）返回 False，交给调用方按进行中 + 的停。""" + now = utcnow() + claimed = await session.execute( + update(job_model).where(job_model.id == job_id, job_model.status == "queued") + .values(status="cancelled", finished_at=now).execution_options(synchronize_session=False)) + if claimed.rowcount != 1: + await session.rollback() + return False + await session.execute( + update(task_model).where(task_model.job_id == job_id, task_model.status.in_(OPEN)) + .values(status="cancelled", finished_at=now).execution_options(synchronize_session=False)) + await session.commit() + return True + + +async def claim(session: AsyncSession, job_model: Any, job_id: str) -> bool: + """认领：排队中 → 进行中。开始之前就被停止了返回 False。""" + result = await session.execute( + update(job_model).where(job_model.id == job_id, job_model.status == "queued") + .values(status="running", started_at=utcnow()).execution_options(synchronize_session=False)) + await session.commit() + return result.rowcount == 1 + + +async def interrupt_on_startup(job_model: Any, close: Callable[[AsyncSession, str], Awaitable[None]], *, + running: Callable[[str], bool]) -> int: + """服务启动时：排队、进行中、而且不在这个进程里跑的任务逐个 close（各自记成 interrupted、没结束的表记 cancelled）。 + 不自动续跑。返回整理了几个。""" + async with SessionLocal() as session: + ids = [job_id for job_id in (await session.execute( + select(job_model.id).where(job_model.status.in_(ACTIVE)))).scalars() if not running(job_id)] + for job_id in ids: + await close(session, job_id) + return len(ids) + + +# ========================================================================== +# 在后台执行 +# ========================================================================== + + +class JobRun: + """一个任务在后台的一次执行。子类给 kind（进队时的身份）、model（任务表，进队前读源和状态），写 _run（执行）和 + _stopped / _failed / _crashed（停止、整体出错、意外出错时的收尾）。""" + + kind = "" + model: Any = None + + def __init__(self, job_id: str) -> None: + self.job_id = job_id + self.task: asyncio.Task[None] | None = None + #: 为什么停：user 使用者点了停止，shutdown 服务关停（标中断） + self.stop_reason: str | None = None + self.writer = SerialWriter() + self.ticket: LaneTicket | None = None + + async def _write(self, fn: Callable[..., Awaitable[Any]], *args: Any) -> Any: + """一条库写入（SerialWriter：一条一条来，停止时不被打断）。""" + return await self.writer(fn, *args) + + async def _settle(self) -> None: + await self.writer.settle() + + async def run(self) -> None: + try: + try: + await self._queue() + await self._run() + except asyncio.CancelledError: + current = asyncio.current_task() + if current is not None: + current.uncancel() + await self._stopped() + except JobFailed as e: + await self._failed(str(e)) + except Exception as e: # noqa: BLE001 - 任务整体出错：照实记下，不留在「进行中」 + await self._crashed(e) + finally: + # 收尾写完才出队：排在后面的任务开始时，这个任务已经记成结束了 + if self.ticket is not None: + self.ticket.release() + + async def _queue(self) -> None: + """进这个源的队，等轮到自己。任务不在了、已经不是排队中（开始之前就被停止了）的不进队：_run 认领不到，直接结束。""" + async with SessionLocal() as session: + job = await session.get(self.model, self.job_id) + source_id = job.source_id if job is not None and job.status == "queued" else None + if source_id is None: + return + self.ticket = lanes.enter(source_id, self.kind, self.job_id) + await self.ticket.wait() + + async def _run(self) -> None: + raise NotImplementedError + + async def _stopped(self) -> None: + raise NotImplementedError + + async def _failed(self, reason: str) -> None: + raise NotImplementedError + + async def _crashed(self, e: Exception) -> None: + raise NotImplementedError + + +class JobManager: + """进程内的一种后台任务：一个 asyncio 任务一件事，停的时候给宽限。factory 按任务编号造一次执行（JobRun）；label 是 + 日志里的叫法（「数据目录起草任务」），name 是 asyncio 任务名的前缀。""" + + def __init__(self, factory: Callable[[str], JobRun], *, label: str, name: str) -> None: + self._runs: dict[str, JobRun] = {} + self._factory = factory + self.label = label + self.name = name + + def _alive(self, job_id: str) -> JobRun | None: + run = self._runs.get(job_id) + if run is None or run.task is None or run.task.done(): + return None + try: + if run.task.get_loop() is not asyncio.get_running_loop(): + self._runs.pop(job_id, None) # 别的事件循环留下的（测试每个用例一个循环） + return None + except RuntimeError: + return None + return run + + def start(self, job_id: str) -> None: + """开始在后台执行这个（排队中的）任务：先进这个源的队，轮到了才开始。""" + run = self._factory(job_id) + run.task = asyncio.get_running_loop().create_task(run.run(), name=f"{self.name}-{job_id}") + self._runs[job_id] = run + + def forget(_: asyncio.Task[None]) -> None: + if self._runs.get(job_id) is run: + self._runs.pop(job_id, None) + + run.task.add_done_callback(forget) + + def running(self, job_id: str) -> bool: + return self._alive(job_id) is not None + + async def stop(self, job_id: str, reason: str, *, grace: float = STOP_GRACE_S) -> bool: + """停掉这个进程里在跑（或在排队等）的任务，等它收尾（最多 grace 秒）。不在这个进程里返回 False。""" + run = self._alive(job_id) + if run is None: + return False + if run.stop_reason is None: + run.stop_reason = reason + run.task.cancel() # type: ignore[union-attr] + await asyncio.wait([run.task], timeout=grace) # type: ignore[list-item] + return True + + async def wait(self, job_id: str, timeout: float | None = None) -> None: + """等这个任务做完（测试、调试用）。不在跑直接返回。""" + run = self._alive(job_id) + if run is not None: + await asyncio.wait([run.task], timeout=timeout) # type: ignore[list-item] + + async def shutdown(self, *, grace: float = SHUTDOWN_GRACE_S) -> None: + """关停：停掉进行中、排队等着的任务（标中断），最多等 grace 秒收尾。""" + runs = [run for job_id in list(self._runs) if (run := self._alive(job_id)) is not None] + for run in runs: + if run.stop_reason is None: + run.stop_reason = "shutdown" + run.task.cancel() # type: ignore[union-attr] + if runs: + await asyncio.wait([run.task for run in runs], timeout=grace) # type: ignore[misc] + logger.info("已停止 %d 个%s", len(runs), self.label) + + +__all__ = ["ACTIVE", "FINISHED", "JobFailed", "JobManager", "JobRun", "LaneBusy", "LaneTicket", "OPEN", "RETRYABLE", + "SHUTDOWN_GRACE_S", "STOP_GRACE_S", "SerialWriter", "SourceLanes", "TASK_DONE", "TASK_STATUSES", "cancel_queued", + "claim", "close_job", "interrupt_on_startup", "lanes"] diff --git a/backend/app/data/catalog_jobs.py b/backend/app/data/catalog_jobs.py index d63c0313..4ed382c0 100644 --- a/backend/app/data/catalog_jobs.py +++ b/backend/app/data/catalog_jobs.py @@ -13,15 +13,20 @@ 接口读的是库里的状态。 - **停止**（cancel）：排队的直接 cancelled；进行中的停掉正在进行的模型调用，没开始和没写完的表记 cancelled，已经写进 目录的（含宽表已经写出来的段）不回滚，任务最后是 cancelled。写过段的表 model_error 说清写进了几段、剩几段没起草。 -- **和同步起草互斥**（sync_draft）：同步起草（POST /catalog/draft）进行期间，同一个源建任务、重试、导入（ensure_idle） - 都是 DraftBusy；有排队或进行中的任务时同步起草也是。进程内登记（单进程部署）。 +- **同种任务互斥**（ensure_draft_free）：同一个源同时只有一个排队或进行中的起草任务，同步起草（POST /catalog/draft） + 进行期间也算在起草，建任务、重试都是 DraftBusy。剖析任务不挡起草：同源有剖析任务时起草任务照建，排在它后面 + （waiting_for = "profile"），剖析结束后自动开始（每个源一条队，见 data/catalog_job_runtime.py）。 +- **同步写操作要求源空闲**（ensure_idle、sync_write）：同步起草、导入目录要的是这个源上没有别的事——有排队或进行中的 + 起草任务（DraftBusy）、剖析任务（catalog_profile_jobs.ProfileBusy）都不开始；开始了就占着这个源的队，执行期间 + 新建的任务排在它后面。进程内登记（单进程部署）。 - **重试**（retry_tables + create_job）：新任务，只含原任务里 failed、partial、cancelled 的表；用不用模型、本次说明沿用， 起草要求取数据源当前的。 - **服务重启**：启动时（startup）把排队、进行中的任务标成 interrupted，没结束的表记 cancelled、写明「服务重启，起草 中断」。不自动续跑：复制出来的数据目录（检查、验收）启动时不能自己去调模型；界面让人点「重试」续上。关停时 （manager.shutdown）停掉进行中的任务，同样标成 interrupted。 -任务的库写入一律经 _JobRun._write：一条一条来，shield 起来——停止任务时不会打断写到一半的库操作，收尾前等它们写完。 +任务的生命周期（进程内执行、停止、同源排队、库写入一条一条来并 shield、启动时标中断、状态词）在 +data/catalog_job_runtime.py，剖析任务用的是同一套；这里只写起草自己的步骤。 **起草要求**存在数据源 options 的 catalog_draft_guidance 里（上传表格的源存在它自己的 DataSource 行上），一段文字， 去首尾空白、最多 GUIDANCE_MAX 字，空串等于没有。只经 PUT /catalog/guidance 改：数据源接口不显示它、整份回写 options @@ -40,20 +45,15 @@ from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from app.data import catalog, catalog_draft +from app.data import catalog_job_runtime as runtime +from app.data.catalog_job_runtime import ACTIVE, FINISHED, RETRYABLE, STOP_GRACE_S, TASK_DONE, TASK_STATUSES from app.data.engine import CATALOG_DRAFT_GUIDANCE_OPTION from app.db.base import SessionLocal, new_id, utcnow -from app.db.models import CatalogDraftJob, CatalogDraftTask, DataSource +from app.db.models import CatalogDraftJob, CatalogDraftTask, CatalogProfileJob, DataSource logger = logging.getLogger(__name__) -#: 任务还没结束的状态（同一个源只能有一个）、结束了的状态 -ACTIVE = ("queued", "running") -FINISHED = ("succeeded", "partial", "failed", "cancelled", "interrupted") -#: 逐表的状态，以及其中算「已结束」的、重试时要带上的 -TASK_STATUSES = ("pending", "running", "ok", "partial", "failed", "skipped", "cancelled") -TASK_DONE = ("ok", "partial", "failed", "skipped", "cancelled") -RETRYABLE = ("failed", "partial", "cancelled") -_OPEN = ("pending", "running") +_OPEN = runtime.OPEN #: 一个任务最多几张表；起草要求、本次说明最多几个字 MAX_TABLES = 500 @@ -63,9 +63,6 @@ NOTE_MAX = 1000 LAST_DRAFT_JOBS = 100 #: 逐表最多记几个模型留空的列名 BLANK_SAMPLE = 5 -#: 停止进行中的任务时最多等它收尾多久（秒）；关停服务时另给（shutdown 的 grace） -STOP_GRACE_S = 10.0 - INTERRUPTED_ERROR = "服务重启，起草中断" #: 停止、中断时写过段的表（宽表每写出一段就并入目录）：写进了几段、剩几段没起草 STOPPED_WRITTEN = "已停止：已写进目录 {n} 段，剩下 {m} 段没起草" @@ -93,7 +90,7 @@ class TextTooLong(ValueError): """起草要求、本次说明超长。message 给人看。""" -class _JobFailed(Exception): +class _JobFailed(runtime.JobFailed): """任务整体出错（数据源没了、快照没了、模型用不了）：message 就是任务的 error。""" @@ -200,13 +197,15 @@ def _counts(statuses: list[str]) -> dict[str, int]: return counts -def job_dict(job: CatalogDraftJob, tasks: list[CatalogDraftTask]) -> dict[str, Any]: - """接口返回的任务（CatalogDraftJob）：逐表的状态按起草顺序；计数、合计都按逐表的算。""" +def job_dict(job: CatalogDraftJob, tasks: list[CatalogDraftTask], *, waiting_for: str | None = None) -> dict[str, Any]: + """接口返回的任务（CatalogDraftJob）：逐表的状态按起草顺序；计数、合计都按逐表的算。waiting_for：排队中、排在同源的 + 剖析任务后面时是 "profile"（见 waiting_for），否则 null。""" counts = _counts([t.status for t in tasks]) return { "id": job.id, "source_id": job.source_id, "status": job.status, + "waiting_for": waiting_for, "use_model": bool(job.use_model), "model": job.model, "guidance": job.guidance, @@ -235,12 +234,32 @@ def job_dict(job: CatalogDraftJob, tasks: list[CatalogDraftTask]) -> dict[str, A } +async def waiting_for(session: AsyncSession, job: CatalogDraftJob) -> str | None: + """排队中的起草任务在等什么：排在同源的剖析任务后面时是 "profile"，否则 None（进行中、结束了的都是 None）。 + + 先看这个源的队（进程内，排在它前面的是谁）；任务刚建、后台还没排进队时看库：同源有比它先建的、排队或进行中的剖析 + 任务。""" + if job.status != "queued": + return None + ahead = runtime.lanes.ahead(job.source_id, job.id) + if ahead is not None: + return "profile" if "profile" in ahead else None + earlier = (await session.execute( + select(CatalogProfileJob.id).where(CatalogProfileJob.source_id == job.source_id, + CatalogProfileJob.status.in_(ACTIVE), + CatalogProfileJob.created_at <= job.created_at).limit(1) + )).first() + return "profile" if earlier is not None else None + + async def job_out(session: AsyncSession, job: CatalogDraftJob | None) -> dict[str, Any] | None: - return None if job is None else job_dict(job, await tasks_of(session, job.id)) + if job is None: + return None + return job_dict(job, await tasks_of(session, job.id), waiting_for=await waiting_for(session, job)) async def job_brief(session: AsyncSession, job: CatalogDraftJob | None) -> dict[str, Any] | None: - """目录清单顶上的进度：{id, status, total, done}。""" + """目录清单顶上的进度：{id, status, total, done, waiting_for}。""" if job is None: return None rows = (await session.execute( @@ -249,7 +268,7 @@ async def job_brief(session: AsyncSession, job: CatalogDraftJob | None) -> dict[ )).all() counts = {status: n for status, n in rows} return {"id": job.id, "status": job.status, "total": sum(counts.values()), - "done": sum(counts.get(s, 0) for s in TASK_DONE)} + "done": sum(counts.get(s, 0) for s in TASK_DONE), "waiting_for": await waiting_for(session, job)} async def last_drafts(session: AsyncSession, source_id: str) -> dict[str, dict[str, Any]]: @@ -302,6 +321,10 @@ def sync_busy_message(row: Any) -> str: return f"数据源「{getattr(row, 'name', '')}」正在起草数据目录（同步起草），请等这次起草结束再开始新的起草" +def import_busy_message(row: Any) -> str: + return f"数据源「{getattr(row, 'name', '')}」正在导入数据目录，请等导入完成后再试" + + #: 正在同步起草（POST /catalog/draft）的数据源。进程内登记：服务端是单进程部署，多进程部署时要换成库里的锁 _SYNC: set[str] = set() @@ -311,30 +334,82 @@ def sync_drafting(source_id: str) -> bool: return source_id in _SYNC -async def ensure_idle(session: AsyncSession, row: Any) -> None: - """这个源没有在起草：没有排队或进行中的任务，也没有在同步起草。有就抛 DraftBusy。建任务、重试、导入目录之前都调它。""" +async def ensure_draft_free(session: AsyncSession, row: Any) -> None: + """同种任务互斥：这个源没有在起草——没有排队或进行中的起草任务，也没有在同步起草。有就抛 DraftBusy。建起草任务、 + 重试之前调它。剖析任务不算：同源有剖析任务时起草任务照建，排在它后面。""" if row.id in _SYNC: raise DraftBusy(sync_busy_message(row)) if await active_job(session, row.id) is not None: raise DraftBusy(busy_message(row)) +async def ensure_idle(session: AsyncSession, row: Any) -> None: + """同步写操作（导入目录、同步起草）要求源空闲：没有在起草（DraftBusy，同 ensure_draft_free），也没有排队或进行中的 + 剖析任务（catalog_profile_jobs.ProfileBusy，说法里写明进行到第几张）。""" + from app.data import catalog_profile_jobs + + await ensure_draft_free(session, row) + await catalog_profile_jobs.ensure_free(session, row) + + +def _lane_busy(row: Any, holder: runtime.LaneTicket) -> Exception: + """库里查过没有任务、这个源的队却被占着（同步起草、导入目录正在做，或者一个任务刚好在收尾）：按占着的是谁说。""" + if holder.kind == "profile": + from app.data import catalog_profile_jobs + + return catalog_profile_jobs.ProfileBusy(catalog_profile_jobs.busy_message(row, None)) + if holder.kind == "import": + return DraftBusy(import_busy_message(row)) + return DraftBusy(sync_busy_message(row) if holder.kind == "sync_draft" else busy_message(row)) + + +@contextlib.asynccontextmanager +async def _sync_write(row: Any, kind: str, check: Callable[[], Awaitable[None]]) -> AsyncIterator[None]: + """同步写操作期间占着这个源的队（不阻塞地试占）：先查库（check：源上有排队或进行中的任务抛 DraftBusy / ProfileBusy）， + 再占队——别的同步写操作正在做时占不到，同样抛忙，什么都不做。占到了，执行期间新建的任务排在它后面，做完才开始。""" + await check() + try: + ticket = runtime.lanes.try_enter(row.id, kind) + except runtime.LaneBusy as e: + raise _lane_busy(row, e.holder) from e + try: + yield + finally: + ticket.release() + + +async def _ensure_idle_but_sync(session: AsyncSession, row: Any) -> None: + """同 ensure_idle，只是不看 _SYNC：同步起草自己刚登记在里面。""" + from app.data import catalog_profile_jobs + + if await active_job(session, row.id) is not None: + raise DraftBusy(busy_message(row)) + await catalog_profile_jobs.ensure_free(session, row) + + @contextlib.asynccontextmanager async def sync_draft(session: AsyncSession, row: Any) -> AsyncIterator[None]: - """同步起草期间登记这个源：同一个源建任务、重试、导入、再开一次同步起草都抛 DraftBusy；有排队或进行中的任务时 - 这里就抛 DraftBusy。先登记、再查库：查库要让出事件循环，先登记才不会和同时建的任务互相看不见（建任务那边写库之后 - 也再看一次，见 create_job）。""" + """同步起草期间登记这个源、占着它的队：同一个源建起草任务、重试、导入、再开一次同步起草都抛 DraftBusy，期间新建的 + 剖析任务排在它后面；有排队或进行中的起草任务时这里抛 DraftBusy，有剖析任务时抛 ProfileBusy。先登记、再查库：查库要 + 让出事件循环，先登记才不会和同时建的任务互相看不见（建任务那边写库之后也再看一次，见 create_job）。""" if row.id in _SYNC: raise DraftBusy(sync_busy_message(row)) _SYNC.add(row.id) try: - if await active_job(session, row.id) is not None: - raise DraftBusy(busy_message(row)) - yield + async with _sync_write(row, "sync_draft", lambda: _ensure_idle_but_sync(session, row)): + yield finally: _SYNC.discard(row.id) +@contextlib.asynccontextmanager +async def import_write(session: AsyncSession, row: Any) -> AsyncIterator[None]: + """导入目录期间占着这个源的队：源上有排队或进行中的起草、剖析任务（或者正在同步起草、导入）时抛 DraftBusy / + ProfileBusy；导入期间新建的任务排在它后面。""" + async with _sync_write(row, "import", lambda: ensure_idle(session, row)): + yield + + async def check_model(session: AsyncSession) -> None: """起草模型能不能用：不能用抛 catalog.CatalogModelUnavailable（只建模型对象，不发请求）。""" await catalog.resolve_draft_model(session) @@ -344,8 +419,9 @@ async def create_job(session: AsyncSession, row: Any, source: Any, *, tables: li note: str | None, actor: str | None, retry_of: str | None = None) -> tuple[CatalogDraftJob, list[CatalogDraftTask]]: """建一个排队的任务（还没开始，调用方接着 manager.start）。row 是数据源记录（起草要求取它的），source 是读表结构 - 用的源（上传源是绑定当前快照的视图）。有排队或进行中的任务抛 DraftBusy。""" - await ensure_idle(session, row) + 用的源（上传源是绑定当前快照的视图）。有排队或进行中的起草任务（或正在同步起草）抛 DraftBusy；同源有剖析任务时照建， + 排在它后面。""" + await ensure_draft_free(session, row) cache = getattr(source, "schema_cache", None) or {} all_tables = cache.get("tables") or {} wanted = list(tables) if tables else await catalog.default_tables(session, source) @@ -395,22 +471,15 @@ async def retry_tables(session: AsyncSession, job: CatalogDraftJob) -> list[str] async def cancel(session: AsyncSession, job: CatalogDraftJob) -> None: - """停止一个任务。排队的直接 cancelled；进行中的交给 manager 停（停掉模型调用、没写完的表记 cancelled）。 - 已经结束的抛 DraftFinished。""" + """停止一个任务。排队的直接 cancelled（在队里等着的后台执行跟着停掉、出队）；进行中的交给 manager 停（停掉模型调用、 + 没写完的表记 cancelled）。已经结束的抛 DraftFinished。""" if job.status in FINISHED: raise DraftFinished("这次起草已经结束了，不需要停止") if job.status == "queued": - now = utcnow() - claimed = await session.execute( - update(CatalogDraftJob).where(CatalogDraftJob.id == job.id, CatalogDraftJob.status == "queued") - .values(status="cancelled", finished_at=now).execution_options(synchronize_session=False)) - if claimed.rowcount == 1: - await session.execute( - update(CatalogDraftTask).where(CatalogDraftTask.job_id == job.id, CatalogDraftTask.status.in_(_OPEN)) - .values(status="cancelled", finished_at=now).execution_options(synchronize_session=False)) - await session.commit() + if await runtime.cancel_queued(session, CatalogDraftJob, CatalogDraftTask, job.id): + await manager.stop(job.id, "user", grace=STOP_GRACE_S) return - await session.rollback() # 刚被后台认领、开始了：按进行中的停 + # 刚被后台认领、开始了：按进行中的停 if await manager.stop(job.id, "user", grace=STOP_GRACE_S): return # 库里说在进行，这个进程里却没有在跑它（记录对不上）：直接记停止 @@ -434,26 +503,19 @@ async def _close(session: AsyncSession, job_id: str, job_status: str, job_error: .values(status=task_status, error=task_error, finished_at=now, model_error=written.format(n=n, m=max(0, int(total or n) - int(n)))) .execution_options(synchronize_session=False)) - await session.execute( - update(CatalogDraftTask).where(CatalogDraftTask.job_id == job_id, CatalogDraftTask.status.in_(_OPEN)) - .values(status=task_status, error=task_error, finished_at=now).execution_options(synchronize_session=False)) - await session.execute( - update(CatalogDraftJob).where(CatalogDraftJob.id == job_id, CatalogDraftJob.status.in_(ACTIVE)) - .values(status=job_status, error=job_error, finished_at=now).execution_options(synchronize_session=False)) - await session.commit() + await runtime.close_job(session, CatalogDraftJob, CatalogDraftTask, job_id, job_status, job_error, task_status, + task_error) async def startup() -> int: """服务启动时：排队、进行中的任务标成 interrupted，没结束的表记 cancelled、写明服务重启。不自动续跑。返回标了几个。""" - async with SessionLocal() as session: - ids = [job_id for job_id in (await session.execute( - select(CatalogDraftJob.id).where(CatalogDraftJob.status.in_(ACTIVE)))).scalars() - if not manager.running(job_id)] - for job_id in ids: - await _close(session, job_id, "interrupted", INTERRUPTED_ERROR, "cancelled", INTERRUPTED_ERROR) - if ids: - logger.warning("服务重启：%d 个数据目录起草任务没有做完，已标为中断，可在数据目录页重试", len(ids)) - return len(ids) + async def interrupt(session: AsyncSession, job_id: str) -> None: + await _close(session, job_id, "interrupted", INTERRUPTED_ERROR, "cancelled", INTERRUPTED_ERROR) + + n = await runtime.interrupt_on_startup(CatalogDraftJob, interrupt, running=manager.running) + if n: + logger.warning("服务重启：%d 个数据目录起草任务没有做完，已标为中断，可在数据目录页重试", n) + return n # ========================================================================== @@ -461,60 +523,35 @@ async def startup() -> int: # ========================================================================== -class _JobRun: - """一个任务在后台的一次执行。""" +class _JobRun(runtime.JobRun): + """一个起草任务在后台的一次执行（进队、停止、收尾的框架在 runtime.JobRun）。""" + + kind = "draft" + model = CatalogDraftJob def __init__(self, job_id: str) -> None: - self.job_id = job_id - self.task: asyncio.Task[None] | None = None - #: 为什么停：user 使用者点了停止，shutdown 服务关停（标中断） - self.stop_reason: str | None = None - self._writes: set[asyncio.Task[Any]] = set() - self._lock = asyncio.Lock() + super().__init__(job_id) self._started: set[str] = set() self.source: Any = None self.actor: str | None = None - # ---------------------------------------------------------------- 库写入 + # ---------------------------------------------------------------- 停止、出错时收尾 - async def _write(self, fn: Callable[..., Awaitable[Any]], *args: Any) -> Any: - """一条库写入：同一个任务的写入一条一条来；shield 起来，停止任务时不会打断写到一半的库操作。""" - async def go() -> Any: - async with self._lock: - async with SessionLocal() as session: - return await fn(session, *args) + async def _stopped(self) -> None: + if self.stop_reason == "user": + await self._end("cancelled", None, "cancelled", None) + else: + await self._end("interrupted", INTERRUPTED_ERROR, "cancelled", INTERRUPTED_ERROR) - task = asyncio.ensure_future(go()) - self._writes.add(task) - task.add_done_callback(self._writes.discard) - return await asyncio.shield(task) + async def _failed(self, reason: str) -> None: + await self._end("failed", reason, "failed", reason) - async def _settle(self) -> None: - """等手上的库写入都写完（停止、出错收尾之前）。""" - while self._writes: - await asyncio.gather(*list(self._writes), return_exceptions=True) + async def _crashed(self, e: Exception) -> None: + from app.core.errors import describe_exception, first_line - # ---------------------------------------------------------------- 主流程 - - async def run(self) -> None: - try: - await self._run() - except asyncio.CancelledError: - current = asyncio.current_task() - if current is not None: - current.uncancel() - if self.stop_reason == "user": - await self._end("cancelled", None, "cancelled", None) - else: - await self._end("interrupted", INTERRUPTED_ERROR, "cancelled", INTERRUPTED_ERROR) - except _JobFailed as e: - await self._end("failed", str(e), "failed", str(e)) - except Exception as e: # noqa: BLE001 - 任务整体出错：照实记下，不留在「起草中」 - from app.core.errors import describe_exception, first_line - - logger.exception("数据目录起草任务 %s 出错", self.job_id) - reason = JOB_ERROR.format(line=first_line(e) or describe_exception(e)) - await self._end("failed", reason, "failed", reason) + logger.exception("数据目录起草任务 %s 出错", self.job_id) + reason = JOB_ERROR.format(line=first_line(e) or describe_exception(e)) + await self._end("failed", reason, "failed", reason) async def _end(self, job_status: str, job_error: str | None, task_status: str, task_error: str | None) -> None: """停止、出错时收尾：等手上的库写入写完，再把没结束的表和任务记下来。收尾本身写不进去只记日志（下次启动时 @@ -603,11 +640,7 @@ class _JobRun: .execution_options(synchronize_session=False) async def _claim(self, session: AsyncSession) -> bool: - result = await session.execute( - update(CatalogDraftJob).where(CatalogDraftJob.id == self.job_id, CatalogDraftJob.status == "queued") - .values(status="running", started_at=utcnow()).execution_options(synchronize_session=False)) - await session.commit() - return result.rowcount == 1 + return await runtime.claim(session, CatalogDraftJob, self.job_id) async def _set_model(self, session: AsyncSession, model_id: str) -> None: await session.execute(update(CatalogDraftJob).where(CatalogDraftJob.id == self.job_id) @@ -746,66 +779,11 @@ async def write_part(session: AsyncSession, source: Any, table: str, notes: dict return None -class DraftJobs: - """进程内的起草任务（照 app/scheduler 的写法：一个 asyncio 任务一件事，停的时候给宽限）。""" +class DraftJobs(runtime.JobManager): + """进程内的起草任务（start / stop / wait / running / shutdown 见 runtime.JobManager）。""" def __init__(self) -> None: - self._runs: dict[str, _JobRun] = {} - - def _alive(self, job_id: str) -> _JobRun | None: - run = self._runs.get(job_id) - if run is None or run.task is None or run.task.done(): - return None - try: - if run.task.get_loop() is not asyncio.get_running_loop(): - self._runs.pop(job_id, None) # 别的事件循环留下的（测试每个用例一个循环） - return None - except RuntimeError: - return None - return run - - def start(self, job_id: str) -> None: - """开始在后台起草这个（排队中的）任务。""" - run = _JobRun(job_id) - run.task = asyncio.get_running_loop().create_task(run.run(), name=f"catalog-draft-{job_id}") - self._runs[job_id] = run - - def forget(_: asyncio.Task[None]) -> None: - if self._runs.get(job_id) is run: - self._runs.pop(job_id, None) - - run.task.add_done_callback(forget) - - def running(self, job_id: str) -> bool: - return self._alive(job_id) is not None - - async def stop(self, job_id: str, reason: str, *, grace: float = STOP_GRACE_S) -> bool: - """停掉这个进程里在跑的任务，等它收尾（最多 grace 秒）。不在这个进程里跑返回 False。""" - run = self._alive(job_id) - if run is None: - return False - if run.stop_reason is None: - run.stop_reason = reason - run.task.cancel() # type: ignore[union-attr] - await asyncio.wait([run.task], timeout=grace) # type: ignore[list-item] - return True - - async def wait(self, job_id: str, timeout: float | None = None) -> None: - """等这个任务做完（测试、调试用）。不在跑直接返回。""" - run = self._alive(job_id) - if run is not None: - await asyncio.wait([run.task], timeout=timeout) # type: ignore[list-item] - - async def shutdown(self, *, grace: float = 3.0) -> None: - """关停：停掉进行中的任务（标中断），最多等 grace 秒收尾。""" - runs = [run for job_id in list(self._runs) if (run := self._alive(job_id)) is not None] - for run in runs: - if run.stop_reason is None: - run.stop_reason = "shutdown" - run.task.cancel() # type: ignore[union-attr] - if runs: - await asyncio.wait([run.task for run in runs], timeout=grace) # type: ignore[misc] - logger.info("已停止 %d 个数据目录起草任务", len(runs)) + super().__init__(_JobRun, label="数据目录起草任务", name="catalog-draft") manager = DraftJobs() @@ -814,6 +792,7 @@ manager = DraftJobs() __all__ = ["ACTIVE", "BLANK_SAMPLE", "DraftBusy", "DraftFinished", "DraftJobs", "FINISHED", "GUIDANCE_MAX", "INTERRUPTED_ERROR", "INTERRUPTED_WRITTEN", "MAX_TABLES", "NOTE_MAX", "NothingToRetry", "RETRYABLE", "STOPPED_WRITTEN", "TASK_DONE", "TASK_STATUSES", "TextTooLong", - "active_job", "cancel", "check_model", "clean_text", "create_job", "ensure_idle", "guidance_of", "job_brief", - "job_dict", "job_out", "last_drafts", "latest_job", "load", "manager", "retry_tables", "save_guidance", - "startup", "sync_busy_message", "sync_draft", "sync_drafting", "tasks_of", "write_part"] + "active_job", "cancel", "check_model", "clean_text", "create_job", "ensure_draft_free", "ensure_idle", + "guidance_of", "import_busy_message", "import_write", "job_brief", "job_dict", "job_out", "last_drafts", + "latest_job", "load", "manager", "retry_tables", "save_guidance", "startup", "sync_busy_message", "sync_draft", + "sync_drafting", "tasks_of", "waiting_for", "write_part"] diff --git a/backend/app/data/catalog_profile_jobs.py b/backend/app/data/catalog_profile_jobs.py new file mode 100644 index 00000000..6a97c3a7 --- /dev/null +++ b/backend/app/data/catalog_profile_jobs.py @@ -0,0 +1,37 @@ +"""数据目录的后台剖析任务（草稿，下一步补全）。""" +from __future__ import annotations + +from typing import Any + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.data.catalog_job_runtime import ACTIVE, TASK_DONE +from app.db.models import CatalogProfileJob, CatalogProfileTask + + +class ProfileBusy(Exception): + """这个数据源已经有排队或进行中的剖析任务。message 给人看。""" + + +async def active_job(session: AsyncSession, source_id: str) -> CatalogProfileJob | None: + return (await session.execute( + select(CatalogProfileJob).where(CatalogProfileJob.source_id == source_id, + CatalogProfileJob.status.in_(ACTIVE)) + .order_by(CatalogProfileJob.created_at.desc()).limit(1) + )).scalar_one_or_none() + + +def busy_message(row: Any, progress: tuple[int, int] | None) -> str: + name = getattr(row, "name", "") or "" + return f"数据源「{name}」正在剖析，请等这次剖析结束（或先停止它）再开始新的剖析" + + +async def ensure_free(session: AsyncSession, row: Any) -> None: + job = await active_job(session, row.id) + if job is not None: + rows = (await session.execute( + select(CatalogProfileTask.status, func.count()).where(CatalogProfileTask.job_id == job.id) + .group_by(CatalogProfileTask.status))).all() + counts = dict(rows) + raise ProfileBusy(busy_message(row, (sum(counts.get(s, 0) for s in TASK_DONE), sum(counts.values())))) diff --git a/backend/app/db/models.py b/backend/app/db/models.py index 21349730..7e3ad023 100644 --- a/backend/app/db/models.py +++ b/backend/app/db/models.py @@ -435,6 +435,96 @@ class CatalogDraftTask(Base): finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) +class CatalogProfileJob(Base, TimestampMixin): + """数据目录的一次后台剖析任务（data/catalog_profile_jobs.py）。逐表的结果在 catalog_profile_tasks。 + + 以前剖析是一次同步长请求：预算按整次剖析算（选 50 张表只做得完两三张）、不能停、关掉页面报告就没了，对业务库查了什么 + 也无从查。现在建任务后在后台逐表剖析，每张表各有自己的查询次数和时长，进度、发过的查询（字面量去掉）随时落库。生命周期 + 和起草任务是同一套（data/catalog_job_runtime.py），表各建各的：进度字段完全不同。 + + 同一个数据源同时只有一个排队或进行中的剖析任务：接口先查一遍，部分唯一索引兜底。和同源的起草任务按建立的先后排队。 + 服务重启时还没结束的标成 interrupted，不自动续跑。挂外键、级联删除：任务是数据源的附属记录。 + """ + + __tablename__ = "catalog_profile_jobs" + __table_args__ = ( + Index("ix_catalog_profile_jobs_source_created", "source_id", "created_at"), + Index("ux_catalog_profile_jobs_active", "source_id", unique=True, + sqlite_where=text("status IN ('queued', 'running')")), + ) + + id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id) + source_id: Mapped[str] = mapped_column(ForeignKey("data_sources.id", ondelete="CASCADE")) + # queued | running | succeeded | partial | failed | cancelled | interrupted + status: Mapped[str] = mapped_column(String(20), default="queued") + # selected 指定的表 | needed 需要剖析的表 | all 全部表 + scope: Mapped[str] = mapped_column(String(20), default="selected") + # 建任务时的剖析设置快照（catalog_profile.ProfileSettings.to_dict） + settings: Mapped[dict[str, Any]] = mapped_column(default=dict) + # 署名（开了登录是认证过的用户名，否则请求头 X-Actor 的自报） + created_by: Mapped[str | None] = mapped_column(String(100), default=None) + started_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) + finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) + # 整个任务发出的查询数、其中超时的、失败的（守卫拒掉的没发出去，不算） + queries: Mapped[int] = mapped_column(Integer, default=0) + timeouts: Mapped[int] = mapped_column(Integer, default=0) + errors: Mapped[int] = mapped_column(Integer, default=0) + # 任务级止损停下的原因：failed 连续失败、slow 超时比例；没有为 NULL + stop_reason: Mapped[str | None] = mapped_column(String(20), default=None) + # 提醒：最近的查询多数超时时的一句话，好转后清掉 + notice: Mapped[str | None] = mapped_column(Text, default=None) + # 任务整体出错、止损、数据源中途停用或关掉剖析时的整句原因 + error: Mapped[str | None] = mapped_column(Text, default=None) + # 重试建的任务记下是重试哪一个 + retry_of: Mapped[str | None] = mapped_column(String(32), default=None) + + +class CatalogProfileTask(Base): + """剖析任务里的一张表。按 seq（剖析顺序）排。状态：pending 排队、running 剖析中、ok 每项都有结论、partial 有项因为 + 运行原因没查成、failed 写入一直冲突或整张表出错、skipped 视图 / 表结构里没有 / 空表 / 没有可剖析的项、cancelled + （停止、止损或服务重启）。 + + 检查项、发现、跳过的项、日期范围、发过的查询都是 JSON：条数受每张表的查询上限约束，大小有界。发过的查询里的 SQL + 字面量都换成了「?」（sql_semantics.normalize_sql）：抽样到的键值不进 AgentLab 自己的库。""" + + __tablename__ = "catalog_profile_tasks" + __table_args__ = (UniqueConstraint("job_id", "seq"),) + + id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id) + job_id: Mapped[str] = mapped_column(ForeignKey("catalog_profile_jobs.id", ondelete="CASCADE"), index=True) + seq: Mapped[int] = mapped_column(Integer) + # 表结构里的键；表结构里没有的（skipped）是请求里的原样名字 + table_name: Mapped[str] = mapped_column(String(255)) + status: Mapped[str] = mapped_column(String(20), default="pending") + # 检查项：[{kind, target, path, state, reason, detail, queries}]。建任务时按当时的目录列一遍，开始剖析这张表时按最新的 + # 目录重列 + checks: Mapped[list[Any] | None] = mapped_column(default=None) + queries: Mapped[int] = mapped_column(Integer, default=0) + elapsed_ms: Mapped[int] = mapped_column(Integer, default=0) + # 进行中时正在做哪一项的整句（「正在核对 orders.customer_id → customers.id」） + current: Mapped[str | None] = mapped_column(Text, default=None) + # 行数估算 {rows, method, at_least} + row_estimate: Mapped[dict[str, Any] | None] = mapped_column(default=None) + findings: Mapped[list[Any] | None] = mapped_column(default=None) + skipped: Mapped[list[Any] | None] = mapped_column(default=None) + date_ranges: Mapped[list[Any] | None] = mapped_column(default=None) + # 发过的查询：[{at, kind, target, sql, ms, rows, outcome, detail}] + query_log: Mapped[list[Any] | None] = mapped_column(default=None) + # 发现的合计 {relations_checked, relations_verified, codes, codes_pending, business_dates}；跳过的项数、其中因为运行 + # 原因没查成的（表清单的「上次剖析」读这几列，不必读整份发现） + summary: Mapped[dict[str, Any] | None] = mapped_column(default=None) + skipped_count: Mapped[int] = mapped_column(Integer, default=0) + unfinished: Mapped[int] = mapped_column(Integer, default=0) + added: Mapped[int] = mapped_column(Integer, default=0) + updated: Mapped[int] = mapped_column(Integer, default=0) + removed: Mapped[int] = mapped_column(Integer, default=0) + version: Mapped[int] = mapped_column(Integer, default=0) + # 整张表的原因：表结构里没有、视图、空表、没有可剖析的项、写入一直冲突、停止 / 中断 / 止损 + error: Mapped[str | None] = mapped_column(Text, default=None) + started_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) + finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) + + class Setting(Base, TimestampMixin): """键值形式的用户设置，前端 Settings 页直接读写。"""