diff --git a/backend/app/api/schedules.py b/backend/app/api/schedules.py index fe66843e..f924e9db 100644 --- a/backend/app/api/schedules.py +++ b/backend/app/api/schedules.py @@ -516,9 +516,9 @@ async def _schedule_outs(session: AsyncSession, rows: list[Schedule]) -> list[di from app import scheduler on = scheduler.scheduler_enabled() - for r in rows: - if _event_kind(r): - states[r.id] = {**await outbox.pending_for(session, r), "scheduler_on": on} + # 一次查齐（outbox.pending_for_many），不再每行各查一次 + waiting = await outbox.pending_for_many(session, [r for r in rows if _event_kind(r)]) + states = {sid: {**state, "scheduler_on": on} for sid, state in waiting.items()} def target(r: Schedule) -> dict[str, Any]: kind, target_id = service.target_of(r) @@ -592,36 +592,70 @@ async def list_variables(tz: str | None = Query(default=None, alias="timezone"), return service.variables(tz=zone, event=event) +#: 选对象时一页最多给几个（按名字排；更多的按关键字查） +TRIGGER_OPTIONS_LIMIT = 100 + + @router.get("/trigger-options") -async def trigger_options(session: AsyncSession = Depends(get_session)) -> dict[str, Any]: +async def trigger_options(event: str | None = None, q: str | None = None, selected: str | None = None, + limit: int = Query(default=TRIGGER_OPTIONS_LIMIT, ge=1, le=500), + session: AsyncSession = Depends(get_session)) -> dict[str, Any]: """新建、修改事件触发任务时能选的：每种事件的对象（能不能选、为什么不能，由这里判，界面不自己判）、「只认某种来源」的 来源、「只认某个应用」的应用、去抖的缺省值、触发链的上限。 - {events: [{event, label, subject_label, objects: [{id, name, ok, reason}]}], origins: [{kind, label}], + 对象按名字排、一页最多 limit 个，more 说后面还有没有：数据源、工作流成百上千时不一次全给，界面按关键字（q，名字里含它） + 再查。event 给了只查这一种事件的对象；selected 是当前选中的那一个（修改时），总在里面，不被分页挤掉、不显示成「已不存在」。 + + {events: [{event, label, subject_label, objects: [{id, name, ok, reason}], more}], origins: [{kind, label}], apps: [{id, name}], default_debounce_s, max_debounce_s, max_depth}""" from app.db.models import App, DataSource - sources = list((await session.execute(select(DataSource).order_by(DataSource.name))).scalars()) - flows = list((await session.execute(select(Workflow).where(Workflow.is_template.is_(False)) - .order_by(Workflow.name))).scalars()) - tables = [] - for src in sources: - if (src.origin or "manual") != "upload": - continue - ok = await triggers.accumulating(session, src) - tables.append({"id": src.id, "name": src.name, "ok": ok, - "reason": None if ok else "每次上传替换整张表，没有「新的一期」：要用这个事件请在导入设置里改为按期累积"}) - objects = { - "table.version": tables, - "workflow.finished": [{"id": w.id, "name": w.name, "ok": bool(w.published_version), - "reason": None if w.published_version else "还没有发布版本：只有正式运行成功结束才算事件"} - for w in flows], - "source.changed": [{"id": src.id, "name": src.name, "ok": True, "reason": None} for src in sources], - } + if event is not None and event not in triggers.EVENTS: + raise _invalid(f"不认识的事件种类：{event}", "event") + wanted = [event] if event else list(triggers.EVENTS) + words = (q or "").strip() + + async def page(model: Any, *conds: Any) -> tuple[list[Any], bool]: + query = select(model).where(*conds) + if words: + query = query.where(model.name.contains(words, autoescape=True)) + rows = list((await session.execute(query.order_by(model.name, model.id).limit(limit + 1))).scalars()) + more = len(rows) > limit + rows = rows[:limit] + if selected and all(r.id != selected for r in rows): + if (picked := (await session.execute(select(model).where(model.id == selected, *conds))).scalar_one_or_none()): + rows.append(picked) + return rows, more + + async def tables() -> tuple[list[dict[str, Any]], bool]: + # 只有上传表格有「新的一期」：按来源筛在库里做（老行没写来源的是手工接的源） + rows, more = await page(DataSource, DataSource.origin == "upload") + out = [] + for src in rows: + ok = await triggers.accumulating(session, src) + out.append({"id": src.id, "name": src.name, "ok": ok, + "reason": None if ok else "每次上传替换整张表，没有「新的一期」：要用这个事件请在导入设置里改为按期累积"}) + return out, more + + async def flows() -> tuple[list[dict[str, Any]], bool]: + rows, more = await page(Workflow, Workflow.is_template.is_(False)) + return [{"id": w.id, "name": w.name, "ok": bool(w.published_version), + "reason": None if w.published_version else "还没有发布版本：只有正式运行成功结束才算事件"} + for w in rows], more + + async def sources() -> tuple[list[dict[str, Any]], bool]: + rows, more = await page(DataSource) + return [{"id": src.id, "name": src.name, "ok": True, "reason": None} for src in rows], more + + makers = {"table.version": tables, "workflow.finished": flows, "source.changed": sources} + entries = [] + for e in wanted: + objects, more = await makers[e]() + entries.append({"event": e, "label": triggers.EVENT_LABEL[e], "subject_label": triggers.SUBJECT_LABEL[e], + "objects": objects, "more": more}) apps = (await session.execute(select(App.id, App.name).where(App.status != "archived").order_by(App.name))).all() return { - "events": [{"event": e, "label": triggers.EVENT_LABEL[e], "subject_label": triggers.SUBJECT_LABEL[e], - "objects": objects[e]} for e in triggers.EVENTS], + "events": entries, "origins": [{"kind": k, "label": triggers.ORIGIN_LABEL[k]} for k in triggers.FORMAL_ORIGINS], "apps": [{"id": i, "name": n} for i, n in apps], "default_debounce_s": triggers.DEFAULT_DEBOUNCE_S, "max_debounce_s": triggers.MAX_DEBOUNCE_S, diff --git a/backend/app/scheduler/outbox.py b/backend/app/scheduler/outbox.py index 366051a5..722147e0 100644 --- a/backend/app/scheduler/outbox.py +++ b/backend/app/scheduler/outbox.py @@ -273,12 +273,10 @@ async def route(now: datetime | None = None) -> int: .limit(ROUTE_BATCH))).scalars()) if not events: return 0 - rows = (await session.execute(select(Schedule.id, Schedule.trigger, Schedule.target_kind).where( - Schedule.trigger_kind == "event", Schedule.enabled.is_(True)))).all() - listeners = [(sid, trigger) for sid, trigger, kind in rows if (kind or "workflow") == "workflow"] + listening = await listeners(session, events) routed = 0 for ev in events: - matched = [sid for sid, trigger in listeners if triggers.matches(trigger, ev.event, ev.subject_id, ev.payload)] + matched = [sid for sid, trigger in listening if triggers.matches(trigger, ev.event, ev.subject_id, ev.payload)] try: async with SessionLocal() as session: # 条件 UPDATE 认领：两拍撞在一起（以后多开了进程）只有一拍改得动这一行 @@ -298,6 +296,28 @@ async def route(now: datetime | None = None) -> int: return routed +def _listening_to(pairs: Any) -> Any: + """触发条件监听着这几对（事件种类, 对象）的事件触发任务的 WHERE 条件：按触发条件里的事件种类和对象筛（SQLAlchemy 的通用 + JSON 访问，各库都能用），不把全部事件触发任务读出来。pairs 是 {事件种类: {对象 id…}}。""" + from sqlalchemy import and_, false, or_ + + conds = [and_(Schedule.trigger["event"].as_string() == kind, + Schedule.trigger[triggers.SUBJECT_KEY[kind]].as_string().in_(sorted(subjects))) + for kind, subjects in pairs.items() if kind in triggers.SUBJECT_KEY and subjects] + return or_(*conds) if conds else false() + + +async def listeners(session: Any, events: list[TriggerEvent]) -> list[tuple[str, dict[str, Any]]]: + """这几条事件可能派给的任务：启用着的、指向工作流的事件触发任务里，触发条件的事件种类和对象对得上的 [(任务 id, 触发条件)]。 + 过滤条件（应用、来源、期次……）由 triggers.matches 逐条再判。""" + pairs: dict[str, set[str]] = {} + for ev in events: + pairs.setdefault(ev.event, set()).add(ev.subject_id) + rows = (await session.execute(select(Schedule.id, Schedule.trigger, Schedule.target_kind).where( + Schedule.trigger_kind == "event", Schedule.enabled.is_(True), _listening_to(pairs)))).all() + return [(sid, trigger) for sid, trigger, kind in rows if (kind or "workflow") == "workflow"] + + async def fire_waiting(now: datetime | None = None) -> list[ScheduleFire]: """有等着的事件的任务，各把到点的去抖窗口处理完。""" now = now or utcnow() @@ -382,6 +402,7 @@ async def _drain(schedule_id: str, now: datetime) -> list[ScheduleFire]: missed = [w for w in due if (now - w.due).total_seconds() > limit] timely = [w for w in due if w not in missed] outputs = {id(w): await _outputs_of(session, w.events[-1]) for w in timely} + await _clear_of_earlier_fires(schedule_id, timely) out: list[ScheduleFire] = [] if missed: fire = await _skip_missed(schedule_id, missed, tz, grace) @@ -396,6 +417,32 @@ async def _drain(schedule_id: str, now: datetime) -> list[ScheduleFire]: return out +async def _clear_of_earlier_fires(schedule_id: str, wins: list[_Window]) -> None: + """要发起的窗口的结束时刻（fire_at）和这个任务已经有的另一次事件触发撞在同一刻（别的期次的窗口早一拍发了，结束时刻相同） + 时往后挪 1 微秒，直到不撞：service.fire_once 按 (任务, fire_at, kind) 认「已经处理过」，撞上了就交回已有的那一条，这个 + 窗口的几条事件会被记到别人的触发上、不再发起。已有的那次触发里有这个窗口的事件（两拍撞在一起处理了同一个窗口）的不挪： + 那就是同一次，交回已有的那一条是对的。""" + from app.scheduler import service + + if not wins: + return + lo, hi = min(w.due for w in wins), max(w.due for w in wins) + timedelta(seconds=1) + async with SessionLocal() as session: + rows = (await session.execute(select(ScheduleFire.fire_at, ScheduleFire.inputs).where( + ScheduleFire.schedule_id == schedule_id, ScheduleFire.kind == "event", + ScheduleFire.fire_at >= lo, ScheduleFire.fire_at <= hi))).all() + taken: dict[datetime, set[str]] = {} + for at, inputs in rows: + brief = (inputs or {}).get("event") if isinstance(inputs, dict) else None + ids = brief.get("event_ids") if isinstance(brief, dict) else None + taken.setdefault(service.as_utc(at), set()).update(str(i) for i in ids or []) + for w in sorted(wins, key=lambda w: w.due): + mine = {ev.id for ev in w.events} + while w.due in taken and not taken[w.due] & mine: + w.due += timedelta(microseconds=1) + taken.setdefault(w.due, set()).update(mine) + + def _event_fire(events: list[TriggerEvent], tz: str, outputs: dict[str, Any]) -> EventFire: """几条合成一次的事件：内容取最新的那一条，深度取最深的那一条加 1。""" latest = events[-1] @@ -491,17 +538,57 @@ async def _clean(now: datetime) -> None: async def pending_for(session: Any, schedule: Schedule) -> dict[str, Any]: - """一个事件触发任务现在有几条事件在等：events 还没派发的（调度关着时一直留在这里）、waiting 已经派给它、在去抖窗口里的； - due_at 是窗口结束、会合成一次触发的时刻（没有在等的为空）。""" - trigger = schedule.trigger if isinstance(schedule.trigger, dict) else {} - subject, event = triggers.subject_of(trigger), trigger.get("event") or "" - rows = (await session.execute(select(TriggerEvent.payload).where( - TriggerEvent.status == "pending", TriggerEvent.event == event, TriggerEvent.subject_id == subject) - .limit(ROUTE_BATCH))).scalars() if subject else [] - unrouted = sum(1 for payload in rows if triggers.matches(trigger, event, subject, payload)) - waiting, first = (await session.execute(select(func.count(), func.min(TriggerDelivery.occurred_at)).where( - TriggerDelivery.schedule_id == schedule.id, TriggerDelivery.status == "pending"))).one() + """一个事件触发任务现在有几条事件在等（见 pending_for_many）。""" + return (await pending_for_many(session, [schedule]))[schedule.id] + + +async def pending_for_many(session: Any, schedules: list[Schedule]) -> dict[str, dict[str, Any]]: + """几个事件触发任务各有几条事件在等，一次查齐（定时任务列表每行都要，不再每行各查一次）：任务 id → {events, waiting, due_at}。 + + events 是还没派发的（调度关着时一直留在这里）；waiting 是已经派给它、在去抖窗口里的；due_at 是最早那个窗口结束、会合成 + 一次触发的时刻（没有在等的为空）。等着的时候任务改了条件、已经不合条件的（下一拍会标 dropped、不再派发）不算：窗口按剩下的 + 最早一条起算，和派发时（_drain）切的窗口一样。待派发数按任务分组一条 SQL 算；窗口要看事件内容，等着的那几条一次读出来。""" from app.scheduler import service - due = service.as_utc(first) + timedelta(seconds=triggers.debounce_of(trigger)) if first is not None else None - return {"events": unrouted, "waiting": int(waiting or 0), "due_at": due.isoformat() if due else None} + rows = [s for s in schedules if s is not None] + out = {s.id: {"events": 0, "waiting": 0, "due_at": None} for s in rows} + if not rows: + return out + trig = {s.id: s.trigger if isinstance(s.trigger, dict) else {} for s in rows} + ids = list(out) + counts = dict((await session.execute(select(TriggerDelivery.schedule_id, func.count()).where( + TriggerDelivery.schedule_id.in_(ids), TriggerDelivery.status == "pending") + .group_by(TriggerDelivery.schedule_id))).all()) + waiting = (await session.execute( + select(TriggerDelivery.schedule_id, TriggerDelivery.id, TriggerDelivery.created_at, TriggerEvent) + .join(TriggerEvent, TriggerEvent.id == TriggerDelivery.event_id) + .where(TriggerDelivery.schedule_id.in_([i for i in ids if counts.get(i)]), TriggerDelivery.status == "pending") + .order_by(TriggerDelivery.occurred_at, TriggerEvent.created_at).limit(DRAIN_BATCH))).all() if counts else [] + by: dict[str, list[tuple[str, TriggerEvent, datetime]]] = {} + stale: dict[str, int] = {} + for sid, did, routed, ev in waiting: + if triggers.matches(trig[sid], ev.event, ev.subject_id, ev.payload): + by.setdefault(sid, []).append((did, ev, service.as_utc(routed))) + else: + stale[sid] = stale.get(sid, 0) + 1 + for sid in ids: + out[sid]["waiting"] = max(int(counts.get(sid) or 0) - stale.get(sid, 0), 0) + wins = windows(by.get(sid, []), triggers.debounce_of(trig[sid])) + if wins: + out[sid]["due_at"] = min(w.due for w in wins).isoformat() + pairs: dict[str, set[str]] = {} + for sid in ids: + if subject := triggers.subject_of(trig[sid]): + pairs.setdefault(str(trig[sid].get("event") or ""), set()).add(subject) + if pairs: + from sqlalchemy import and_, or_ + + cond = or_(*[and_(TriggerEvent.event == kind, TriggerEvent.subject_id.in_(sorted(subs))) + for kind, subs in pairs.items()]) + unrouted = (await session.execute(select(TriggerEvent.event, TriggerEvent.subject_id, TriggerEvent.payload) + .where(TriggerEvent.status == "pending", cond).limit(ROUTE_BATCH))).all() + for kind, subject, payload in unrouted: + for sid in ids: + if triggers.matches(trig[sid], kind, subject, payload): + out[sid]["events"] += 1 + return out diff --git a/backend/app/scheduler/triggers.py b/backend/app/scheduler/triggers.py index 1a667cf3..00ad81a5 100644 --- a/backend/app/scheduler/triggers.py +++ b/backend/app/scheduler/triggers.py @@ -248,29 +248,40 @@ async def find_cycle(session: Any, *, schedule_id: str | None, trigger: dict[str start, target = subject_of(trigger), target_workflow_id if start == target: return [start, start] - edges: dict[str, set[str]] = {} - rows = (await session.execute(select(Schedule.id, Schedule.trigger, Schedule.workflow_id, Schedule.target_kind) - .where(Schedule.trigger_kind == "event"))).all() - for sid, other, workflow_id, kind in rows: - if sid == schedule_id or (kind or "workflow") != "workflow" or not isinstance(other, dict): - continue - if other.get("event") == "workflow.finished" and subject_of(other) and workflow_id: - edges.setdefault(subject_of(other), set()).add(workflow_id) - # 从 B 出发找回 A 的路（广度优先，取最短的一条说给人听） - parent: dict[str, str | None] = {target: None} - queue = [target] - while queue: - node = queue.pop(0) - for nxt in sorted(edges.get(node, ())): - if nxt in parent: + + async def edges_from(nodes: list[str]) -> dict[str, set[str]]: + """这几个工作流跑完会触发哪些工作流：只读「上游工作流运行成功结束」、对象是它们的那几个任务（按触发条件里的事件种类和 + 对象筛，SQLAlchemy 的通用 JSON 访问），不把全部事件触发任务读出来。""" + rows = (await session.execute(select(Schedule.id, Schedule.trigger, Schedule.workflow_id, Schedule.target_kind) + .where(Schedule.trigger_kind == "event", + Schedule.trigger["event"].as_string() == "workflow.finished", + Schedule.trigger["workflow_id"].as_string().in_(nodes)))).all() + out: dict[str, set[str]] = {} + for sid, other, workflow_id, kind in rows: + if sid == schedule_id or (kind or "workflow") != "workflow" or not isinstance(other, dict): continue - parent[nxt] = node - if nxt == start: - back = [start] # A ← … ← B，倒着记 - while back[-1] != target: - back.append(parent[back[-1]] or target) - return [start, *reversed(back)] - queue.append(nxt) + if other.get("event") == "workflow.finished" and subject_of(other) and workflow_id: + out.setdefault(subject_of(other), set()).add(workflow_id) + return out + + # 从 B 出发找回 A 的路（广度优先，一层查一次，取最短的一条说给人听） + parent: dict[str, str | None] = {target: None} + frontier = [target] + while frontier: + edges = await edges_from(frontier) + nxt_level: list[str] = [] + for node in frontier: + for nxt in sorted(edges.get(node, ())): + if nxt in parent: + continue + parent[nxt] = node + if nxt == start: + back = [start] # A ← … ← B，倒着记 + while back[-1] != target: + back.append(parent[back[-1]] or target) + return [start, *reversed(back)] + nxt_level.append(nxt) + frontier = nxt_level return None diff --git a/backend/tests/test_schedule_events_scale.py b/backend/tests/test_schedule_events_scale.py new file mode 100644 index 00000000..547701a0 --- /dev/null +++ b/backend/tests/test_schedule_events_scale.py @@ -0,0 +1,171 @@ +"""事件触发的规模（scheduler/outbox.py、scheduler/triggers.py、api/schedules.py）： + +- 定时任务列表里每个事件触发任务「有几条在等」一次查齐（outbox.pending_for_many：一条 SQL 按任务分组算待派发数），不再 + 每行各查一次； +- 每一拍派发只读监听着这几种事件、这几个对象的任务（按触发条件里的事件种类、对象筛，用 SQLAlchemy 的通用 JSON 访问）， + 不再把全部事件触发任务读出来；保存时查环只顺着「上游工作流运行成功结束」那几条边往下找； +- 选对象（trigger-options）按事件、关键字查、分页：返回 more 说还有没有，当前选中的那一个总在里面； +- 去抖窗口最早那一条因为任务改了条件被标 dropped 时，「什么时候合成触发」按剩下的最早一条算；一个窗口的结束时刻和这个任务 + 已经有的另一次触发撞在同一刻（别的期次、早一拍发的）时往后挪，那几条事件照常发起，不并进已有的触发。 +工作流、表格、数据源都是虚构的。 +""" +from __future__ import annotations + +from contextlib import contextmanager +from datetime import timedelta + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import event as sa_event +from sqlalchemy import select + +from app.db.base import SessionLocal, engine, utcnow +from app.db.models import Schedule, ScheduleFire, Workflow +from app.db.models_triggers import TriggerDelivery +from app.scheduler import outbox +from tests.test_schedule_events import ( # noqa: F401 - engine_up 是本模块也要的 autouse 夹具 + deliveries, engine_up, event_schedule, fires, put_event, settle, workflow, +) + + +@contextmanager +def counted(): + """数这一段里发了几条 SQL。""" + seen: list[str] = [] + + def hook(conn, cursor, statement, *_a): # noqa: ANN001 + seen.append(statement) + + target = engine.sync_engine + sa_event.listen(target, "before_cursor_execute", hook) + try: + yield seen + finally: + sa_event.remove(target, "before_cursor_execute", hook) + + +@pytest.fixture +async def client(): + from app.main import app + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +async def _many(n: int) -> tuple[str, list[str]]: + upstream = await workflow("门店日报") + downstream = await workflow("区域汇总") + ids = [await event_schedule(downstream, {"event": "workflow.finished", "workflow_id": upstream, "debounce_s": 600}, + name=f"汇总{i}") for i in range(n)] + return upstream, ids + + +async def test_waiting_counts_for_many_schedules_in_one_go(): + upstream, ids = await _many(3) + t0 = utcnow().replace(microsecond=0) + for i in range(2): + await put_event("workflow.finished", upstream, {"workflow_id": upstream, "run_id": f"r{i}"}, t0 + timedelta(seconds=i)) + await outbox.route(t0 + timedelta(seconds=5)) + await put_event("workflow.finished", upstream, {"workflow_id": upstream, "run_id": "r9"}, t0 + timedelta(seconds=9)) + async with SessionLocal() as session: + rows = list((await session.execute(select(Schedule).where(Schedule.id.in_(ids)))).scalars()) + one = {r.id: await outbox.pending_for(session, r) for r in rows} + with counted() as sql: + many = await outbox.pending_for_many(session, rows) + assert many == one + assert all(v == {"events": 1, "waiting": 2, "due_at": (t0 + timedelta(seconds=600)).isoformat()} for v in many.values()) + assert len(sql) <= 3, sql + + +async def test_the_schedule_list_does_not_query_once_per_row(client): + await _many(2) + with counted() as few: + assert (await client.get("/api/schedules")).status_code == 200 + await _many(5) + with counted() as more: + assert (await client.get("/api/schedules")).status_code == 200 + assert len(more) <= len(few), (len(few), len(more)) + + +async def test_a_tick_reads_only_the_schedules_listening_to_the_pending_events(): + upstream, ids = await _many(2) + other = await workflow("别的日报") + elsewhere = await event_schedule(await workflow("别的汇总"), {"event": "workflow.finished", "workflow_id": other}) + await put_event("workflow.finished", upstream, {"workflow_id": upstream, "run_id": "r1"}, utcnow()) + async with SessionLocal() as session: + from app.db.models_triggers import TriggerEvent + + pending = list((await session.execute(select(TriggerEvent).where(TriggerEvent.status == "pending"))).scalars()) + listeners = await outbox.listeners(session, pending) + assert sorted(sid for sid, _ in listeners) == sorted(ids) and elsewhere not in {sid for sid, _ in listeners} + + +async def test_trigger_options_search_and_page(client): + names = [f"检索样例{c}-{i}" for c in "甲乙" for i in range(3)] + async with SessionLocal() as session: + rows = [Workflow(name=n, graph={"nodes": [], "edges": []}) for n in names] + session.add_all(rows) + await session.commit() + chosen = next(r.id for r in rows if r.name == "检索样例乙-2") + r = await client.get("/api/schedules/trigger-options", params={"event": "workflow.finished", "q": "检索样例甲", + "limit": 2}) + assert r.status_code == 200, r.text + [entry] = r.json()["events"] + assert entry["event"] == "workflow.finished" and entry["more"] is True + assert [o["name"] for o in entry["objects"]] == ["检索样例甲-0", "检索样例甲-1"] + r = await client.get("/api/schedules/trigger-options", params={"event": "workflow.finished", "q": "检索样例甲", + "limit": 2, "selected": chosen}) + [entry] = r.json()["events"] + assert chosen in [o["id"] for o in entry["objects"]], "当前选中的那一个总在里面（不被分页挤掉、不显示成已不存在）" + # 不带参数：照旧给全部种类，每种最多一页 + body = (await client.get("/api/schedules/trigger-options")).json() + assert {e["event"] for e in body["events"]} == {"table.version", "workflow.finished", "source.changed"} + assert all("more" in e for e in body["events"]) + + +async def test_a_dropped_earliest_event_no_longer_sets_the_window_end(): + """窗口最早那一条因为任务改了条件被标 dropped：什么时候合成触发按剩下的最早一条算（列表上的 due_at、真发起的 fire_at 一致）。""" + upstream = await workflow("门店日报") + downstream = await workflow("区域汇总") + sid = await event_schedule(downstream, {"event": "workflow.finished", "workflow_id": upstream, "debounce_s": 60}) + t0 = utcnow().replace(microsecond=0) + await put_event("workflow.finished", upstream, {"workflow_id": upstream, "run_id": "r1", "origin_kind": "launcher"}, t0) + e2 = await put_event("workflow.finished", upstream, {"workflow_id": upstream, "run_id": "r2", + "origin_kind": "app_entry"}, t0 + timedelta(seconds=30)) + await outbox.route(t0 + timedelta(seconds=31)) + async with SessionLocal() as session: + await session.execute(Schedule.__table__.update().where(Schedule.id == sid).values( + trigger={"event": "workflow.finished", "workflow_id": upstream, "debounce_s": 60, "origin_kind": "app_entry"})) + await session.commit() + state = await outbox.pending_for(session, await session.get(Schedule, sid)) + assert state["due_at"] == (t0 + timedelta(seconds=90)).isoformat(), state + assert await outbox.dispatch_due(t0 + timedelta(seconds=61)) == [] + [fire] = await outbox.dispatch_due(t0 + timedelta(seconds=91)) + assert fire.fire_at == t0 + timedelta(seconds=90) and fire.inputs["event"]["event_ids"] == [e2] + await settle(fire.run_id) + + +async def test_a_window_ending_where_another_fire_already_is_still_fires(): + """一个窗口的结束时刻和这个任务已有的另一次触发撞在同一刻（别的期次早一拍发了）：往后挪，这几条照常发起， + 不并进已有的那次触发、不被它吞掉。""" + source = "src-虚构" + downstream = await workflow("月报") + sid = await event_schedule(downstream, {"event": "table.version", "source_id": source, "debounce_s": 60}) + t0 = utcnow().replace(microsecond=0) + aug = await put_event("table.version", source, {"period": "2026-08", "table": "日报"}, t0) + [first] = await outbox.dispatch_due(t0 + timedelta(seconds=61)) + assert first.fire_at == t0 + timedelta(seconds=60) and first.inputs["event"]["event_ids"] == [aug] + await settle(first.run_id) + # 另一期的事件发生在同一刻，晚到了一拍（比如补写的）：窗口结束同样是 t0+60 + sep = await put_event("table.version", source, {"period": "2026-09", "table": "日报"}, t0) + [second] = await outbox.dispatch_due(t0 + timedelta(seconds=62)) + assert second.id != first.id and second.inputs["event"]["event_ids"] == [sep] + assert second.fire_at > first.fire_at + states = {d.event_id: (d.status, d.fire_id) for d in await deliveries(sid)} + assert states[sep] == ("fired", second.id) + await settle(second.run_id) + async with SessionLocal() as session: + assert len(list((await session.execute(select(ScheduleFire).where(ScheduleFire.schedule_id == sid))).scalars())) == 2 + assert len(await fires(sid)) == 2 + assert [d.status for d in await deliveries(sid)] == ["fired", "fired"] + assert TriggerDelivery is not None diff --git a/frontend/src/api/schedules.ts b/frontend/src/api/schedules.ts index d92e3015..98faad1f 100644 --- a/frontend/src/api/schedules.ts +++ b/frontend/src/api/schedules.ts @@ -36,8 +36,12 @@ export const schedules = { */ eventVariables: (event: ScheduleEventKind, opts?: RequestOptions) => get(`/schedules/variables${qs({ trigger_kind: 'event', event })}`, opts), - /** 新建、修改事件触发任务时能选的对象（能不能选、为什么由服务端判）、来源、应用。老服务端没有，回 404 */ - triggerOptions: (opts?: RequestOptions) => get('/schedules/trigger-options', opts), + /** + * 新建、修改事件触发任务时能选的对象（能不能选、为什么由服务端判）、来源、应用。老服务端没有，回 404。对象一页有上限 + * （more 为真时没给完）：按事件种类、名字里的关键字再查；selected 是当前选中的那一个，总在结果里 + */ + triggerOptions: (find?: { event?: ScheduleEventKind; q?: string; selected?: string }, opts?: RequestOptions) => + get(`/schedules/trigger-options${qs(find)}`, opts), /** 立即运行一次：记一条手动触发，上一次还没结束时照样按重叠策略跳过 */ run: (id: string) => post(`${at(id)}/run`), /** 触发历史，新的在前。before 是上一页给的 next_before */ diff --git a/frontend/src/lib/termsTrigger.ts b/frontend/src/lib/termsTrigger.ts index 2a94ad32..caeff647 100644 --- a/frontend/src/lib/termsTrigger.ts +++ b/frontend/src/lib/termsTrigger.ts @@ -43,6 +43,10 @@ export const TRIGGER_TEXT = { /** 不能选的对象：名字后面接服务端给的原因 */ subjectNotOk: (name: string, reason: string) => `${name}（${reason}）`, subjectGone: (name: string) => `${name}（已不存在）`, + /** 对象太多、一页没列完：按名字查找 */ + subjectSearch: (label: string) => `按名称查找${label}`, + subjectSearchHint: '只列出了一部分，输入名称中的字查找', + subjectSearchNone: '没有名称中含这些字的', optionsLoading: '正在读取可选的对象…', optionsFailed: '无法读取可选的对象', optionsUnsupported: '服务端暂不支持事件触发，请升级服务端后再使用', diff --git a/frontend/src/pages/schedules/TriggerEditor.tsx b/frontend/src/pages/schedules/TriggerEditor.tsx index d36f131b..b307e704 100644 --- a/frontend/src/pages/schedules/TriggerEditor.tsx +++ b/frontend/src/pages/schedules/TriggerEditor.tsx @@ -73,6 +73,28 @@ function useTriggerOptions(): OptionsState { return state } +/** + * 对象太多、一页没列完（服务端的 more）时按名字查：输入停一下再查，结果只换这一种事件的对象；当前选中的那一个总带上 + * （selected），不被分页挤掉、不显示成已不存在 + */ +function useObjectSearch(kind: ScheduleEventKind, selected: string, enabled: boolean) { + const [q, setQ] = useState('') + const [found, setFound] = useState<{ key: string; objects: ScheduleTriggerOptions['events'][number]['objects'] } | null>(null) + const words = q.trim() + const key = `${kind}\u0000${words}` + useEffect(() => { + if (!enabled || !words) return + const controller = new AbortController() + const timer = setTimeout(() => { + api.schedules.triggerOptions({ event: kind, q: words, selected: selected || undefined }, { signal: controller.signal }) + .then((data) => setFound({ key, objects: data.events.find((e) => e.event === kind)?.objects ?? [] })) + .catch(() => {}) + }, 300) + return () => { clearTimeout(timer); controller.abort() } + }, [enabled, kind, words, selected, key]) + return { q, setQ, objects: words && found?.key === key ? found.objects : null } +} + export function EventSection({ value, onChange, timezone, setTimezone, savedName, error, debounceError }: { value: TriggerForm onChange: (t: TriggerForm) => void @@ -91,7 +113,9 @@ export function EventSection({ value, onChange, timezone, setTimezone, savedName const kind = value.event const label = TT.subject[kind] const entry = options.status === 'ok' ? options.data.events.find((e) => e.event === kind) : undefined - const objects = entry?.objects ?? [] + const paged = !!entry?.more + const search = useObjectSearch(kind, value.subjectId, paged) + const objects = search.objects ?? entry?.objects ?? [] const listed = objects.some((o) => o.id === value.subjectId) let picker @@ -112,7 +136,10 @@ export function EventSection({ value, onChange, timezone, setTimezone, savedName aria-describedby={error ? 'schedule-trigger-subject-error' : undefined} onChange={(e) => set({ subjectId: e.target.value })}> - {value.subjectId && !listed && } + {/* 一页没列完时不在列表里不等于不存在：写保存时记下的名字，不标「已不存在」 */} + {value.subjectId && !listed && ( + + )} {objects.map((o) => (