diff --git a/backend/app/api/schedules.py b/backend/app/api/schedules.py index 35e8cca1..d7f5b0c3 100644 --- a/backend/app/api/schedules.py +++ b/backend/app/api/schedules.py @@ -491,6 +491,8 @@ async def create_schedule_row(session: AsyncSession, body: dict[str, Any], *, ac values = schedule_values(body) hook = values.pop("webhook") workflow = await _check_workflow(session, values["workflow_id"], values["run_class"], values["input_template"]) + # 目标的几列只经 scheduler/service 的一处封装写（工作流目标：target_id 同 workflow_id） + values |= service.target_columns("workflow", values["workflow_id"]) row = Schedule(**values, workflow_name=workflow.name, next_run_at=service.next_run(values["cron"], values["timezone"], values["enabled"]), created_by=actor, **_webhook_columns(hook)) @@ -528,7 +530,8 @@ async def update_schedule(schedule_id: str, request: Request, if "name" in body: values["name"] = _name(body["name"]) if "workflow_id" in body: - values["workflow_id"] = _workflow_id(body["workflow_id"]) + # 目标的几列一起改（scheduler/service.target_columns）：只改 workflow_id 的话 target_id 还指着原来的工作流 + values |= service.target_columns("workflow", _workflow_id(body["workflow_id"])) if "cron" in body: values["cron"] = _cron(body["cron"]) if "timezone" in body: diff --git a/backend/app/scheduler/service.py b/backend/app/scheduler/service.py index 14c25443..87d0ff18 100644 --- a/backend/app/scheduler/service.py +++ b/backend/app/scheduler/service.py @@ -3,6 +3,11 @@ 调度循环（loop.py）到点调 fire_once，接口的「立即运行」也调它：重叠跳过、渲染输入、记触发、发起运行这一套 只有一份。发起本身走 engine/launch.py 的 launch_run，和 HTTP 发起是同一套规则——正式运行只能从当前发布版本 发起，这一条不会在定时任务这边被绕过。 + +**定时的目标**（docs/design/next-round/00-总方案.md X4：定时只有一套）：一个定时任务指向一个目标——工作流、评测题集 +（第 3 波）或数据源维护（第 3 波）。认领、补跑、重叠跳过、触发记录、通知都只有这一套，fire_once 按目标种类分派。 +老表的 workflow_id 是非空约束，只加不改的迁移放宽不了：目标不是工作流时它存空串。这条规矩只在 target_of / +target_columns 这一处，别处读写目标都经它们，不直接碰这两列。 """ from __future__ import annotations @@ -64,6 +69,33 @@ INTERRUPTED_REASON = "服务在发起运行前停止，这次触发没有发起 GENERIC_REASON = "发起运行时出错，详细信息见服务日志" +#: 定时的目标种类（Schedule.target_kind），顺序即以后界面上的顺序。这一波 fire_once 只会发起 workflow +TARGET_KINDS = ("workflow", "eval_suite", "source_maintenance") +TARGET_LABEL = {"workflow": "工作流", "eval_suite": "评测题集", "source_maintenance": "数据源维护"} +#: 目标种类这个版本还发起不了（库里有别的版本建的任务、手改过库）：照实记「未能发起」，不当成工作流去跑 +TARGET_UNSUPPORTED = "这个定时任务的目标是「{kind}」，当前版本还不能按时发起这类目标，这次触发没有发起" + + +def target_of(schedule: Any) -> tuple[str, str]: + """一个定时任务的目标：(种类, id)。加列之前建的任务 target_kind 缺省 workflow、target_id 为空，目标就是 workflow_id。 + + workflow 时 id 以 workflow_id 为准（两列由 target_columns 一起写，对得上；老任务只有这一列）。""" + kind = getattr(schedule, "target_kind", None) or "workflow" + if kind == "workflow": + return kind, schedule.workflow_id or getattr(schedule, "target_id", None) or "" + return kind, getattr(schedule, "target_id", None) or "" + + +def target_columns(kind: str, target_id: str) -> dict[str, str]: + """写目标的几列：{target_kind, target_id, workflow_id}。workflow 时 target_id 同 workflow_id；别的种类 workflow_id 存空串 + （老表的非空约束）。种类不认识、id 为空抛 ValueError。""" + if kind not in TARGET_KINDS: + raise ValueError(f"不认识的定时目标种类：{kind}") + if not target_id: + raise ValueError("定时任务的目标 id 不能为空") + return {"target_kind": kind, "target_id": target_id, "workflow_id": target_id if kind == "workflow" else ""} + + class TemplateError(ValueError): """输入模板写错了。str() 是给人看的中文说明，指出是哪个输入。""" @@ -282,46 +314,61 @@ async def fire_once(schedule_id: str, fire_at: datetime, kind: str, *, actor: st """处理一次触发，返回记下的触发记录。同一 (任务, fire_at, kind) 已经有记录（两拍撞在一起、立即运行连点两下） 时不再发起，返回已有的那一条；任务不存在返回 None。 - 上一次运行还没结束就记 skipped_overlap；输入渲染不了、launch_run 拒绝或出错记 failed_to_start（拒绝写原话， - 其他异常写通用说明、详细信息进日志）；发起成功记 started 和运行 id。 + 按目标种类分派（target_of）：这一波只有工作流（_fire_workflow），别的种类记 failed_to_start、写明原因，不发起。 """ async with launch_lock(): async with SessionLocal() as session: schedule = await session.get(Schedule, schedule_id) if schedule is None: return None - name, workflow_id, run_class = schedule.name, schedule.workflow_id, schedule.run_class + target_kind, target_id = target_of(schedule) + name, run_class = schedule.name, schedule.run_class tz, cron_expr, template = schedule.timezone, schedule.cron, schedule.input_template - try: - inputs = render_inputs(check_template(template), template_values(fire_at, tz, cron_expr)) - except (TemplateError, cron.CronError, cron.TimezoneError) as e: - # 存进库之后才坏掉的（手改过库、时区库升级删了时区）：照实说是哪一项，不发起 - return await _record(schedule_id, fire_at, kind, "failed_to_start", f"无法生成这次运行的输入：{e}", {}) - busy = await _active_run(schedule_id) - if busy is not None: - return await _record(schedule_id, fire_at, kind, "skipped_overlap", overlap_reason(*busy), inputs) - fire = await _record(schedule_id, fire_at, kind, "failed_to_start", INTERRUPTED_REASON, inputs, fresh=True) - if fire is None: - return await _existing(schedule_id, fire_at, kind) - from app.engine.launch import LaunchRefused, launch_run + if target_kind != "workflow": + reason = TARGET_UNSUPPORTED.format(kind=TARGET_LABEL.get(target_kind, target_kind)) + return await _record(schedule_id, fire_at, kind, "failed_to_start", reason, {}) + return await _fire_workflow(schedule_id, fire_at, kind, actor=actor, name=name, workflow_id=target_id, + run_class=run_class, tz=tz, cron_expr=cron_expr, template=template) - _LAUNCHING.add(fire.id) + +async def _fire_workflow(schedule_id: str, fire_at: datetime, kind: str, *, actor: str | None, name: str, + workflow_id: str, run_class: str, tz: str, cron_expr: str, template: Any, + ) -> ScheduleFire | None: + """目标是工作流的一次触发（持着 launch_lock）。 + + 上一次运行还没结束就记 skipped_overlap；输入渲染不了、launch_run 拒绝或出错记 failed_to_start（拒绝写原话， + 其他异常写通用说明、详细信息进日志）；发起成功记 started 和运行 id。 + """ + try: + inputs = render_inputs(check_template(template), template_values(fire_at, tz, cron_expr)) + except (TemplateError, cron.CronError, cron.TimezoneError) as e: + # 存进库之后才坏掉的（手改过库、时区库升级删了时区）：照实说是哪一项，不发起 + return await _record(schedule_id, fire_at, kind, "failed_to_start", f"无法生成这次运行的输入：{e}", {}) + busy = await _active_run(schedule_id) + if busy is not None: + return await _record(schedule_id, fire_at, kind, "skipped_overlap", overlap_reason(*busy), inputs) + fire = await _record(schedule_id, fire_at, kind, "failed_to_start", INTERRUPTED_REASON, inputs, fresh=True) + if fire is None: + return await _existing(schedule_id, fire_at, kind) + from app.engine.launch import LaunchRefused, launch_run + + _LAUNCHING.add(fire.id) + try: try: - try: - async with SessionLocal() as session: - run = await launch_run(session, workflow_id=workflow_id, run_class=run_class, - input_payload=inputs, actor=actor or actor_label(name), - schedule_id=schedule_id, origin_kind="schedule", - origin={"schedule_id": schedule_id, "schedule": name, - "fire_id": fire.id, "fire_kind": kind}) - except LaunchRefused as e: - return await _finish(fire.id, status="failed_to_start", reason=e.message) - except Exception: # noqa: BLE001 - 发起失败不能让调度循环停下；详细信息进日志，记录上写通用说明 - logger.exception("定时任务「%s」（%s）发起运行失败", name, schedule_id) - return await _finish(fire.id, status="failed_to_start", reason=GENERIC_REASON) - return await _finish(fire.id, status="started", reason=None, run_id=run.id) - finally: - _LAUNCHING.discard(fire.id) + async with SessionLocal() as session: + run = await launch_run(session, workflow_id=workflow_id, run_class=run_class, + input_payload=inputs, actor=actor or actor_label(name), + schedule_id=schedule_id, origin_kind="schedule", + origin={"schedule_id": schedule_id, "schedule": name, + "fire_id": fire.id, "fire_kind": kind}) + except LaunchRefused as e: + return await _finish(fire.id, status="failed_to_start", reason=e.message) + except Exception: # noqa: BLE001 - 发起失败不能让调度循环停下；详细信息进日志，记录上写通用说明 + logger.exception("定时任务「%s」（%s）发起运行失败", name, schedule_id) + return await _finish(fire.id, status="failed_to_start", reason=GENERIC_REASON) + return await _finish(fire.id, status="started", reason=None, run_id=run.id) + finally: + _LAUNCHING.discard(fire.id) async def record_skip(schedule_id: str, fire_at: datetime, reason: str) -> ScheduleFire | None: diff --git a/backend/tests/test_schedule_target.py b/backend/tests/test_schedule_target.py new file mode 100644 index 00000000..ad717c8d --- /dev/null +++ b/backend/tests/test_schedule_target.py @@ -0,0 +1,135 @@ +"""定时任务的目标（docs/design/next-round/00-总方案.md X4：定时只有一套）。 + +骨架给 Schedule 加了 target_kind / target_id；workflow_id 是老表上的非空约束，只加不改的迁移放宽不了。规矩只在 +scheduler/service.py 的一处封装里：workflow 目标的 target_id 同 workflow_id；不是工作流的目标 workflow_id 存空串。 +这里守： + +- 存量任务（target_kind 是缺省的 workflow、target_id 为空）照旧按 workflow_id 发起，行为一字不变； +- 建任务、改工作流都经同一处封装写目标，两列对得上； +- fire_once 按目标种类分派。这一波只实现 workflow，别的种类照实记「未能发起」并写明原因，不发起运行、不悄悄当成 + 工作流去跑空串； +- 接口这一波不暴露新种类（接口返回的形状不变，测试见 test_schedules_api.py 的 SCHEDULE_KEYS）。 + +工作流、任务名都是虚构的。 +""" +from __future__ import annotations + +import uuid +from datetime import datetime, timezone + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import delete, func, select + +from app.db.base import SessionLocal +from app.db.models import Run, Schedule, ScheduleFire, Workflow, WorkflowVersion +from app.engine.runner import run_manager +from app.main import app +from app.scheduler import service + +UTC = timezone.utc +NINE = datetime(2026, 10, 5, 1, 0, tzinfo=UTC) + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + async with SessionLocal() as session: + await session.execute(delete(Schedule)) + await session.commit() + yield + await run_manager.shutdown() + + +def _node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +def _graph() -> dict: + return {"nodes": [_node("start", "input", fields=[{"name": "stat_date", "required": True}]), + _node("out", "output", fields=[{"name": "r", "value": "完成 {{ input.stat_date }}"}])], + "edges": [{"source": "start", "target": "out"}]} + + +async def _workflow() -> str: + async with SessionLocal() as session: + wf = Workflow(name=f"门店日报-{uuid.uuid4().hex[:6]}", graph=_graph(), status="published", published_version=1) + session.add(wf) + await session.flush() + session.add(WorkflowVersion(workflow_id=wf.id, version=1, graph=_graph())) + await session.commit() + return wf.id + + +def test_target_of_a_legacy_row_reads_the_workflow(): + """加列之前建的任务：target_kind 缺省 workflow、target_id 为空，目标就是 workflow_id。""" + row = Schedule(name="老任务", workflow_id="wf-legacy", cron="0 9 * * *") + assert service.target_of(row) == ("workflow", "wf-legacy") + row.target_kind = "workflow" + row.target_id = None + assert service.target_of(row) == ("workflow", "wf-legacy") + + +def test_target_columns_keep_the_two_columns_in_step(): + assert service.target_columns("workflow", "wf-1") == {"target_kind": "workflow", "target_id": "wf-1", + "workflow_id": "wf-1"} + # 非工作流目标：workflow_id 存空串（老表的非空约束），目标 id 只在 target_id + assert service.target_columns("eval_suite", "suite-1") == {"target_kind": "eval_suite", "target_id": "suite-1", + "workflow_id": ""} + row = Schedule(name="评测", cron="0 9 * * *", **service.target_columns("eval_suite", "suite-1")) + assert service.target_of(row) == ("eval_suite", "suite-1") + with pytest.raises(ValueError): + service.target_columns("nonsense", "x") + with pytest.raises(ValueError): + service.target_columns("workflow", "") + + +async def test_creating_and_retargeting_through_the_api_write_both_columns(): + wf_a, wf_b = await _workflow(), await _workflow() + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + made = await client.post("/api/schedules", json={ + "name": "每日早报", "workflow_id": wf_a, "cron": "0 9 * * *", "timezone": "Asia/Shanghai", + "input_template": {"stat_date": "{{ schedule.yesterday }}"}}) + assert made.status_code == 201, made.text + body = made.json() + assert "target_kind" not in body and "target_id" not in body, "这一波接口不暴露新种类" + async with SessionLocal() as session: + row = await session.get(Schedule, body["id"]) + assert (row.target_kind, row.target_id, row.workflow_id) == ("workflow", wf_a, wf_a) + moved = await client.patch(f"/api/schedules/{body['id']}", json={"workflow_id": wf_b, "version": body["version"]}) + assert moved.status_code == 200, moved.text + async with SessionLocal() as session: + row = await session.get(Schedule, body["id"]) + assert (row.target_kind, row.target_id, row.workflow_id) == ("workflow", wf_b, wf_b) + + +async def test_a_legacy_schedule_still_fires_its_workflow(): + wf = await _workflow() + async with SessionLocal() as session: + row = Schedule(name="老任务", workflow_id=wf, cron="0 9 * * *", timezone="Asia/Shanghai", run_class="formal", + input_template={"stat_date": "{{ schedule.yesterday }}"}, next_run_at=NINE) + session.add(row) + await session.commit() + schedule_id = row.id + fire = await service.fire_once(schedule_id, NINE, "manual") + assert fire is not None and fire.status == "started" and fire.run_id + async with SessionLocal() as session: + run = await session.get(Run, fire.run_id) + assert run.workflow_id == wf and run.input == {"stat_date": "2026-10-04"} + + +async def test_a_target_kind_this_wave_cannot_fire_is_recorded_not_run(): + async with SessionLocal() as session: + row = Schedule(name="题集回归", cron="0 9 * * *", timezone="Asia/Shanghai", + **service.target_columns("eval_suite", "suite-x"), next_run_at=NINE) + session.add(row) + await session.commit() + schedule_id = row.id + runs_before = await session.scalar(select(func.count()).select_from(Run)) + fire = await service.fire_once(schedule_id, NINE, "manual") + assert fire is not None and fire.status == "failed_to_start" and fire.run_id is None + assert fire.reason == service.TARGET_UNSUPPORTED.format(kind=service.TARGET_LABEL["eval_suite"]) + async with SessionLocal() as session: + assert await session.scalar(select(func.count()).select_from(Run)) == runs_before, "不能发起任何运行" + assert await session.scalar(select(func.count()).select_from(ScheduleFire) + .where(ScheduleFire.schedule_id == schedule_id)) == 1