mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:31:58 +00:00
数据事件触发按复核意见修正：途经驳回、去抖积压、source.changed 补写、按期次分窗口、补算缺省不认、示例出口字段同一套过滤
M1 途经驳回（编排者已定）：outbox.stage_run_finished 收可选的 rejected_via（缺省 None 即没途经驳回），事件内容带 rejected 和
rejected_via，说法接「（途经驳回）」，模板有 {{ event.rejected }}（true / false）。触发条件缺省只认没途经驳回的，trigger 加
include_rejected（界面「也认途经驳回的」勾选，说明照实写）。runner._finalize 把 _rejected_via(state) 传进来那一下留给编排者
在合并时接（runner 的 _rejected_via 在 W1-b 的分支上）；测试直接调 stage_run_finished 传参覆盖。
C1 去抖积压：以前一拍只走一个窗口，积压时后面的越等越久、全过了宽限，而且每个窗口各记一条跳过，原因还说调度没在跑。
现在 _drain 按发生时刻把等着的事件切成窗口，一拍里把到点的处理完（最多 100 个窗口、5000 条）；排在前面已过补发宽限的合成
一条 skipped_misfire（合并了几次事件、几个窗口、时间跨度，上传表格另写合掉的期次），派给任务时已经晚于宽限才说「调度未运行」，
否则照实说积压；宽限内的照常发起（照重叠规则跳过的照实记）。
C3 source.changed：这一处确实没法和变化记录同一个事务（一次探查、剖析的变化分好几批落库，一批只触发一次要等任务结束才知道），
改成说实话：models_triggers 的说明写清哪两种同事务、哪一种不是。trigger_events 加 dedupe_key（source.changed:<任务号>，唯一），
emit_source_changed 按它去重；调度的一拍（刚起的服务第一拍就补，之后每 10 分钟）由 backfill_source_changes 按任务号补写：2 天内
测到、还开着的要处理级变化，任务已结束却没有事件的补一条。分组在 Python 里做，不用各库写法不同的 JSON 取值。
C4 按期次分窗口：triggers.window_key 让上传表格按期次各自成窗口，9 月、10 月连着启用各有各的触发；同一期的重传照旧合并。
两个窗口结束在同一刻时后一个挪 1 微秒，fire_at 不撞唯一约束。
C5 应用补算：workflow.finished 缺省不认 app_backfill 发起的（升版补算连发好几条、各是某一期的历史输入），trigger 加
include_backfill（界面「也认应用补算的」），选了「只认应用补算」时本来就认；说法写「途经驳回、应用补算的不算」。
C6 示例出口字段：triggers.outputs_of 一处过滤「_」开头的键，真派发、预览、立即运行共用；示例事件也按 matches 跳过缺省不认的
途经驳回、应用补算，取合条件的最近一次。
测试：新加 test_schedule_events_review.py（11 项，先写、先看着失败；去掉 matches 的驳回判断、把每拍窗口数改回 1 回跑，对应用例
各自失败），test_schedule_events_api.py 加一项「也认」两项经接口存取和说法；check-schedules 加「上游工作流的请求体」一段、纯函数
几条、触发历史里合并跳过那一行的夹具。
验证：后端全量 14479 通过；tsc、check-copy；check-schedules 447 项、check-runs 327 项通过（拿锁、--stacks 1）。
This commit is contained in:
ThearchyHelios
@@ -439,8 +439,9 @@ async def _fire_outs(session: AsyncSession, fires: list[ScheduleFire]) -> list[d
|
||||
raw = f.inputs.get("event") if isinstance(f.inputs, dict) and f.schedule_id not in maintenance else None
|
||||
if not isinstance(raw, dict):
|
||||
return None
|
||||
# windows / periods 只在几个过了宽限的窗口合成一条跳过时有：合了几个窗口、合掉的期次（上传表格）
|
||||
return {key: raw.get(key) for key in ("event", "subject_id", "subject_name", "text", "merged", "first_at",
|
||||
"last_at", "depth", "sample", "real")}
|
||||
"last_at", "depth", "sample", "real", "windows", "periods")}
|
||||
|
||||
compared = {f.id: f.compare_artifact for f in fires if f.compare_artifact and f.run_id}
|
||||
compares = await auto_compare.briefs(list(compared.values()))
|
||||
|
||||
@@ -5,13 +5,20 @@
|
||||
|
||||
**为什么要一张「待派发事件」表，而不是在发生处直接发起运行。** 事件发生在别的事务里：启用上传表格的新版本、正式运行的
|
||||
收尾、记下数据源的变化。在那里直接发起运行，发起失败要不要回滚那次启用？重复发生、服务重启时发起到一半又算什么？说不清。
|
||||
改成在发生处、和那件事**同一个事务**里写一行 TriggerEvent（发生了就一定有这一行，没发生就一定没有），调度循环一拍取走、
|
||||
按各定时任务的触发条件派发：发起运行的规矩（重叠跳过、正式运行只从发布版本发起、触发记录唯一）全部沿用定时任务那一套。
|
||||
改成在发生处写一行 TriggerEvent，调度循环一拍取走、按各定时任务的触发条件派发：发起运行的规矩（重叠跳过、正式运行只从
|
||||
发布版本发起、触发记录唯一）全部沿用定时任务那一套。
|
||||
|
||||
什么时候和那件事同一个事务：table.version（切指针那一下）、workflow.finished（写终态那一下）是同一个事务——发生了就一定有
|
||||
这一行，没发生就一定没有。source.changed 不是：一次探查、剖析的变化分好几批落库，「这一批只触发一次」要等任务结束才知道，
|
||||
事件在任务结束后另开会话写（data/source_changes.notify_after_job）。中间进程死了、库被锁住写不进，变化记录在、事件没有：
|
||||
调度的一拍按任务号补写（scheduler/outbox.backfill_source_changes），dedupe_key（source.changed:<任务号>）唯一，补写和正常的
|
||||
那一次不会记成两条。
|
||||
调度关着（AGENTLAB_SCHEDULER）时没人取，事件留在表里，接口照实说有几条在等。第 2 波多进程时，取走靠的是条件 UPDATE，
|
||||
多个进程也只有一个取得到。
|
||||
|
||||
- TriggerEvent 一次事件：种类（table.version / workflow.finished / source.changed）、对象 id、给模板用的事件内容、触发链
|
||||
的深度和经过的运行、发生时刻。status：pending 还没派发；routed 已经按触发条件派给了 matched 个定时任务（0 个也算派完）。
|
||||
的深度和经过的运行、发生时刻、去重键（只有 source.changed 用）。status：pending 还没派发；routed 已经按触发条件派给了
|
||||
matched 个定时任务（0 个也算派完）。
|
||||
- TriggerDelivery 一个事件派给一个定时任务、等着去抖：同一个定时任务窗口内（第一条事件发生之后 debounce_s 秒）的几条
|
||||
合成一次触发，fire_id 是那次触发记录；status：pending 等着、fired 合进了某次触发（发起、跳过、未能发起都算）、dropped
|
||||
定时任务在等待期间被停用或改了触发条件，不再派发。(定时任务, 事件) 唯一：同一个事件不会派给同一个任务两次。
|
||||
@@ -61,6 +68,9 @@ class TriggerEvent(Base):
|
||||
routed_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None)
|
||||
#: 派给了几个定时任务
|
||||
matched: Mapped[int] = mapped_column(Integer, default=0)
|
||||
#: 去重键：同一件事只记一条。source.changed 是 source.changed:<任务号>（任务结束时记、没记上的补写，两边撞上唯一约束）；
|
||||
#: 别的种类为空（它们和那件事同一个事务，不会记两遍）
|
||||
dedupe_key: Mapped[str | None] = mapped_column(String(80), default=None, unique=True)
|
||||
created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
|
||||
|
||||
|
||||
|
||||
+215
-57
@@ -3,18 +3,24 @@
|
||||
|
||||
**写（发生处）。** 三处，各自只加一两行调用这里：
|
||||
- 上传表格按期累积的新一期启用（data/table_versions 的 publish_build / publish_snapshot，切指针的同一个事务）：stage_table_version；
|
||||
- 正式运行成功收尾（engine/runner 的 _finalize，写终态的同一个事务）：stage_run_finished；
|
||||
- 探查、剖析任务结束、这一批新记了「要处理」的变化（data/source_changes.notify_after_job）：emit_source_changed，一批一条。
|
||||
- 正式运行成功收尾（engine/runner 的 _finalize，写终态的同一个事务）：stage_run_finished；途经驳回的（W1-b 的驳回出口）带上
|
||||
rejected_via，缺省不认它的任务不派；
|
||||
- 探查、剖析任务结束、这一批新记了「要处理」的变化（data/source_changes.notify_after_job）：emit_source_changed，一个任务一条。
|
||||
这一处不和变化记录同一个事务（一批只触发一次要等任务结束才知道），中间丢了的由每一拍的 backfill_source_changes 按任务号补写，
|
||||
去重键唯一、不会记成两条。
|
||||
发生处只写一行 TriggerEvent，不查有没有任务在等、不发起运行：事务边界、重复与重启都在这边说清。
|
||||
|
||||
**取（调度循环的一拍，dispatch_due）。**
|
||||
1. 派：pending 的事件逐条按条件 UPDATE 认领成 routed（多个进程也只有一个认领得到），同一个事务里给每个合条件、启用着的
|
||||
事件触发任务写一行 TriggerDelivery。没人合条件的也认领掉（matched=0）。
|
||||
2. 发：每个有等待中的 delivery 的任务，窗口从最早那一条的发生时刻起算 debounce_s 秒；窗口过了就把落在窗口里的几条合成一次
|
||||
触发：内容取最新的那一条，fire_at 就是窗口的结束时刻（同一个窗口算出来的永远是同一个时刻，触发记录 (任务, fire_at, kind)
|
||||
唯一，重复派发、重启后补派都不会发起两次）。上一次的运行还没结束照 overlap=skip 记 skipped_overlap；触发链超过上限记未能
|
||||
发起；派发得太晚（调度停过、超出补发宽限）记 skipped_misfire。之后把这几条标 fired。任务停用了、改了触发条件，等着的标
|
||||
dropped，不再派发。
|
||||
2. 发：每个有等待中的 delivery 的任务，按发生时刻把它们切成去抖窗口：窗口从第一条的发生时刻起算 debounce_s 秒，落在里面的
|
||||
几条合成一次触发，内容取最新的那一条，fire_at 是窗口的结束时刻（同一个窗口算出来的永远是同一个时刻，触发记录
|
||||
(任务, fire_at, kind) 唯一，重复派发、重启后补派都不会发起两次）。上传表格按期次分窗口（triggers.window_key）：9 月、10 月
|
||||
连着启用各有各的触发，同一期的重传照旧合并。一拍里把到点的窗口处理完（最多 MAX_WINDOWS 个），不是一拍一个：以前积压时
|
||||
一拍只走一个窗口，后面的越等越久、最后全过了宽限。到点的窗口里已经过了补发宽限的（排在前面）合成**一条** skipped_misfire，
|
||||
写合并了几次事件、几个窗口、时间跨度——派给任务就已经晚了（调度停过）才说「调度未运行」，否则照实说积压；之后宽限内的
|
||||
照常发起：上一次的运行还没结束照 overlap=skip 记 skipped_overlap，触发链超过上限记未能发起。处理过的标 fired。任务停用了、
|
||||
改了触发条件，等着的标 dropped，不再派发。
|
||||
调度关着（AGENTLAB_SCHEDULER）时没有这一拍：事件留在表里，接口照实说有几条在等（pending_for）。
|
||||
|
||||
**留多久。** 派完的事件、合进触发的 delivery 留 RETAIN_DAYS 天给排查，之后由这一拍顺手清掉（每小时最多一次）。
|
||||
@@ -45,6 +51,12 @@ CLEAN_EVERY = timedelta(hours=1)
|
||||
CHAIN_KEEP = 10
|
||||
#: 一次触发里记几条合并进来的事件 id
|
||||
MERGED_KEEP = 20
|
||||
#: 一个任务一拍最多处理几个到点的去抖窗口、读几条等着的事件（积压特别多时分几拍处理完，不让一拍拖太久）
|
||||
MAX_WINDOWS = 100
|
||||
DRAIN_BATCH = 5000
|
||||
#: source.changed 往回补写几天内结束的任务；补写最多多久一次（刚起的服务第一拍就补）
|
||||
BACKFILL_DAYS = 2
|
||||
BACKFILL_EVERY = timedelta(minutes=10)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
@@ -76,9 +88,12 @@ async def stage_table_version(session: Any, *, source_id: str, source_name: str,
|
||||
period_end=period_end, activated_at=now, replaced=replaced, seq=seq)))
|
||||
|
||||
|
||||
def stage_run_finished(session: Any, run: Run) -> None:
|
||||
def stage_run_finished(session: Any, run: Run, rejected_via: Any = None) -> None:
|
||||
"""正式运行成功结束：往收尾的会话里加一条 workflow.finished（不提交，随终态一起提交）。探索运行、没挂在工作流上的不算。
|
||||
这次运行自己是事件触发的，事件带上它的触发深度和经过的运行，下游据此防环。"""
|
||||
这次运行自己是事件触发的，事件带上它的触发深度和经过的运行，下游据此防环。
|
||||
|
||||
rejected_via：这次运行途经了哪几个关卡的驳回（W1-b 的驳回出口：按驳回那条路走完，状态仍是 succeeded；runner 收尾时把
|
||||
_rejected_via(state) 交进来），缺省 None 是没途经驳回。照发，事件里带上它：缺省不认途经驳回的任务不派（triggers.matches）。"""
|
||||
if run.status != "succeeded" or run.run_class != "formal" or not run.workflow_id:
|
||||
return
|
||||
upstream = (run.origin or {}).get("trigger") if run.origin_kind == "schedule" and isinstance(run.origin, dict) else None
|
||||
@@ -87,29 +102,92 @@ def stage_run_finished(session: Any, run: Run) -> None:
|
||||
chain = [str(x) for x in (upstream.get("chain") or []) if x][-(CHAIN_KEEP - 1):] + [run.id]
|
||||
session.add(TriggerEvent(
|
||||
event="workflow.finished", subject_id=run.workflow_id, subject_name=(run.workflow_name or "")[:200],
|
||||
payload=triggers.run_payload(run), depth=depth, chain=chain, occurred_at=run.finished_at or utcnow()))
|
||||
payload=triggers.run_payload(run, rejected_via), depth=depth, chain=chain,
|
||||
occurred_at=run.finished_at or utcnow()))
|
||||
|
||||
|
||||
async def emit_source_changed(source_id: str, job_id: str, new: int) -> None:
|
||||
"""一次探查、剖析任务结束、新记了 new 条「要处理」的变化：记一条 source.changed（自己开会话提交）。同一个任务只记一条：
|
||||
「同一次探查 / 剖析产生的一批只触发一次」。出错只记日志：变化记录本身已经落库。"""
|
||||
def source_key(job_id: str) -> str:
|
||||
"""source.changed 的去重键：一个任务一条。"""
|
||||
return f"source.changed:{job_id}"
|
||||
|
||||
|
||||
async def emit_source_changed(source_id: str, job_id: str, new: int, *, occurred_at: datetime | None = None) -> bool:
|
||||
"""一次探查、剖析任务结束、新记了 new 条「要处理」的变化：记一条 source.changed（自己开会话提交），返回记没记下。同一个任务
|
||||
只记一条（去重键唯一）：「同一次探查 / 剖析产生的一批只触发一次」，补写（backfill_source_changes）和这里撞上也只有一条。
|
||||
出错只记日志：变化记录本身已经落库，没记上的由补写接着记。"""
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
|
||||
from app.data import source_changes
|
||||
from app.db.models import DataSource
|
||||
|
||||
try:
|
||||
async with SessionLocal() as session:
|
||||
if await session.scalar(select(TriggerEvent.id).where(TriggerEvent.dedupe_key == source_key(job_id))):
|
||||
return False
|
||||
source = await session.get(DataSource, source_id)
|
||||
got = await source_changes.counts(session, source_id)
|
||||
by = got.get("by_severity") or {}
|
||||
name = source.name if source is not None else ""
|
||||
session.add(TriggerEvent(
|
||||
event="source.changed", subject_id=source_id, subject_name=name[:200],
|
||||
event="source.changed", subject_id=source_id, subject_name=name[:200], dedupe_key=source_key(job_id),
|
||||
occurred_at=occurred_at or utcnow(),
|
||||
payload={"source": name, "source_id": source_id, "job_id": job_id, "counts": {
|
||||
"new": int(new), "action": int(by.get("action") or 0), "warning": int(by.get("warning") or 0),
|
||||
"info": int(by.get("info") or 0), "pending": int(got.get("pending") or 0)}}))
|
||||
await session.commit()
|
||||
try:
|
||||
await session.commit()
|
||||
except IntegrityError:
|
||||
await session.rollback() # 同一个任务刚被别处记上（补写和收尾撞在一起）
|
||||
return False
|
||||
return True
|
||||
except Exception: # noqa: BLE001
|
||||
logger.exception("任务 %s 结束后记数据源变化的待派发事件时出错", job_id)
|
||||
return False
|
||||
|
||||
|
||||
_backfilled_at: datetime | None = None
|
||||
|
||||
|
||||
async def backfill_source_changes(now: datetime | None = None) -> int:
|
||||
"""补写 source.changed：BACKFILL_DAYS 天内测到、还开着的「要处理」级变化，按任务号分组；任务已经结束（探查任务；剖析任务
|
||||
只算取值观察那几种，同 notify_after_job）、却没有那个任务的事件的，补一条（发生时刻取任务结束的时刻）。返回补了几条。
|
||||
在 Python 里按任务号分组，不用各库写法不同的 JSON 取值。"""
|
||||
from app.data.catalog_profile_jobs import OBSERVING_SCOPES
|
||||
from app.db.models import CatalogProfileJob
|
||||
from app.db.models_sources import IntrospectJob, SourceChange
|
||||
|
||||
now = now or utcnow()
|
||||
async with SessionLocal() as session:
|
||||
rows = (await session.execute(select(SourceChange.source_id, SourceChange.detected_by).where(
|
||||
SourceChange.severity == "action", SourceChange.status == "open",
|
||||
SourceChange.detected_at >= now - timedelta(days=BACKFILL_DAYS)))).all()
|
||||
jobs: dict[tuple[str, str, str], int] = {}
|
||||
for source_id, detected in rows:
|
||||
detected = detected if isinstance(detected, dict) else {}
|
||||
job_id, kind = detected.get("job_id"), detected.get("job_kind")
|
||||
if job_id and kind in ("introspect", "profile"):
|
||||
key = (str(kind), str(job_id), source_id)
|
||||
jobs[key] = jobs.get(key, 0) + 1
|
||||
if not jobs:
|
||||
return 0
|
||||
have = set((await session.execute(select(TriggerEvent.dedupe_key).where(
|
||||
TriggerEvent.dedupe_key.in_([source_key(j) for _, j, _ in jobs])))).scalars())
|
||||
todo: list[tuple[str, str, int, datetime | None]] = []
|
||||
for (kind, job_id, source_id), n in jobs.items():
|
||||
if source_key(job_id) in have:
|
||||
continue
|
||||
job = await session.get(IntrospectJob if kind == "introspect" else CatalogProfileJob, job_id)
|
||||
if job is None or job.status in ("queued", "running"):
|
||||
continue # 还在进行：结束时自己会记
|
||||
if kind == "profile" and job.scope not in OBSERVING_SCOPES:
|
||||
continue
|
||||
todo.append((source_id, job_id, n, job.finished_at))
|
||||
made = 0
|
||||
for source_id, job_id, n, finished in todo:
|
||||
if await emit_source_changed(source_id, job_id, n, occurred_at=finished):
|
||||
made += 1
|
||||
logger.info("补写了任务 %s 的数据源变化事件（它结束时没记上）", job_id)
|
||||
return made
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
@@ -170,8 +248,16 @@ _cleaned_at: datetime | None = None
|
||||
|
||||
|
||||
async def dispatch_due(now: datetime | None = None) -> list[ScheduleFire]:
|
||||
"""完整的一拍：派新来的事件、发去抖窗口到了的触发、顺手清理。返回这一拍记下的触发。一个任务出错不影响别的。"""
|
||||
"""完整的一拍：补写没记上的 source.changed（每 BACKFILL_EVERY 一次，刚起的服务第一拍就补）、派新来的事件、发去抖窗口到了的
|
||||
触发、顺手清理。返回这一拍记下的触发。一个任务出错不影响别的。"""
|
||||
global _backfilled_at
|
||||
now = now or utcnow()
|
||||
if _backfilled_at is None or now - _backfilled_at >= BACKFILL_EVERY:
|
||||
_backfilled_at = now
|
||||
try:
|
||||
await backfill_source_changes(now)
|
||||
except Exception: # noqa: BLE001 - 补写是兜底，出错不挡派发
|
||||
logger.exception("补写数据源变化的待派发事件时出错")
|
||||
await route(now)
|
||||
out = await fire_waiting(now)
|
||||
await _clean(now)
|
||||
@@ -213,72 +299,144 @@ async def route(now: datetime | None = None) -> int:
|
||||
|
||||
|
||||
async def fire_waiting(now: datetime | None = None) -> list[ScheduleFire]:
|
||||
"""去抖窗口到了的任务各合成一次触发。"""
|
||||
"""有等着的事件的任务，各把到点的去抖窗口处理完。"""
|
||||
now = now or utcnow()
|
||||
async with SessionLocal() as session:
|
||||
waiting = (await session.execute(
|
||||
select(TriggerDelivery.schedule_id, func.min(TriggerDelivery.occurred_at))
|
||||
.where(TriggerDelivery.status == "pending").group_by(TriggerDelivery.schedule_id))).all()
|
||||
waiting = list((await session.execute(
|
||||
select(TriggerDelivery.schedule_id).where(TriggerDelivery.status == "pending").distinct())).scalars())
|
||||
out: list[ScheduleFire] = []
|
||||
for schedule_id, first in waiting:
|
||||
for schedule_id in waiting:
|
||||
try:
|
||||
fire = await _fire_one(schedule_id, first, now)
|
||||
if fire is not None:
|
||||
out.append(fire)
|
||||
out.extend(await _drain(schedule_id, now))
|
||||
except Exception: # noqa: BLE001 - 一个任务出错不能拖住这一拍的其他任务
|
||||
logger.exception("事件触发的定时任务 %s 这一拍处理失败", schedule_id)
|
||||
return out
|
||||
|
||||
|
||||
async def _fire_one(schedule_id: str, first: datetime, now: datetime) -> ScheduleFire | None:
|
||||
@dataclass
|
||||
class _Window:
|
||||
"""一个去抖窗口：同一个分窗口的键（上传表格是期次）、落在里面的几条（delivery id, 事件, 派给任务的时刻）、结束的时刻。"""
|
||||
|
||||
key: str
|
||||
start: datetime
|
||||
due: datetime
|
||||
items: list[tuple[str, TriggerEvent, datetime]] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def events(self) -> list[TriggerEvent]:
|
||||
return [ev for _, ev, _ in self.items]
|
||||
|
||||
|
||||
def windows(items: list[tuple[str, TriggerEvent, datetime]], debounce_s: int) -> list[_Window]:
|
||||
"""按发生时刻切成去抖窗口（items 已按发生时刻排好）：每个分窗口的键各自一串，窗口从第一条起算 debounce_s 秒；按结束时刻
|
||||
排。两个窗口结束在同一刻（不同期次同一刻启用）时后一个往后挪 1 微秒：fire_at 不撞触发记录的唯一约束，同样的输入永远切出
|
||||
同样的窗口。"""
|
||||
from app.scheduler import service
|
||||
|
||||
span = timedelta(seconds=debounce_s)
|
||||
current: dict[str, _Window] = {}
|
||||
out: list[_Window] = []
|
||||
for did, ev, routed in items:
|
||||
at = service.as_utc(ev.occurred_at)
|
||||
key = triggers.window_key(ev.event, ev.payload)
|
||||
win = current.get(key)
|
||||
if win is None or at > win.due:
|
||||
win = current[key] = _Window(key, at, at + span)
|
||||
out.append(win)
|
||||
win.items.append((did, ev, routed))
|
||||
out.sort(key=lambda w: (w.due, w.key))
|
||||
for prev, cur in zip(out, out[1:]):
|
||||
if cur.due <= prev.due:
|
||||
cur.due = prev.due + timedelta(microseconds=1)
|
||||
return out
|
||||
|
||||
|
||||
async def _drain(schedule_id: str, now: datetime) -> list[ScheduleFire]:
|
||||
"""一个任务这一拍到点的窗口：已过补发宽限的合成一条跳过，其余照常发起。"""
|
||||
from app.scheduler import loop, service
|
||||
|
||||
async with SessionLocal() as session:
|
||||
schedule = await session.get(Schedule, schedule_id)
|
||||
if schedule is None:
|
||||
return None
|
||||
return []
|
||||
trigger, tz = schedule.trigger if isinstance(schedule.trigger, dict) else {}, schedule.timezone
|
||||
if not schedule.enabled or schedule.trigger_kind != "event":
|
||||
# 等着的时候被停用、改成了别的触发方式：停用是「别再触发了」，等着的不再派发
|
||||
await _close(session, _pending(schedule_id), "dropped")
|
||||
return None
|
||||
due = service.as_utc(first) + timedelta(seconds=triggers.debounce_of(trigger))
|
||||
if now < due:
|
||||
return None
|
||||
window = (await session.execute(
|
||||
select(TriggerDelivery.id, TriggerEvent).join(TriggerEvent, TriggerEvent.id == TriggerDelivery.event_id)
|
||||
.where(TriggerDelivery.schedule_id == schedule_id, TriggerDelivery.status == "pending",
|
||||
TriggerDelivery.occurred_at <= due)
|
||||
.order_by(TriggerDelivery.occurred_at, TriggerEvent.created_at))).all()
|
||||
return []
|
||||
rows = (await session.execute(
|
||||
select(TriggerDelivery.id, TriggerDelivery.created_at, TriggerEvent)
|
||||
.join(TriggerEvent, TriggerEvent.id == TriggerDelivery.event_id)
|
||||
.where(TriggerDelivery.schedule_id == schedule_id, TriggerDelivery.status == "pending")
|
||||
.order_by(TriggerDelivery.occurred_at, TriggerEvent.created_at).limit(DRAIN_BATCH))).all()
|
||||
# 等着的时候改了触发条件（换了对象、加了过滤）：已经不合条件的不再派发
|
||||
stale = [did for did, ev in window if not triggers.matches(trigger, ev.event, ev.subject_id, ev.payload)]
|
||||
stale = [did for did, _, ev in rows if not triggers.matches(trigger, ev.event, ev.subject_id, ev.payload)]
|
||||
if stale:
|
||||
await _close(session, TriggerDelivery.id.in_(stale), "dropped")
|
||||
window = [(did, ev) for did, ev in window if did not in set(stale)]
|
||||
if not window:
|
||||
return None
|
||||
live = [(did, ev, service.as_utc(routed)) for did, routed, ev in rows if did not in set(stale)]
|
||||
due = [w for w in windows(live, triggers.debounce_of(trigger)) if w.due <= now][:MAX_WINDOWS]
|
||||
if not due:
|
||||
return []
|
||||
grace = max(int(schedule.misfire_grace_s or 0), 0)
|
||||
events = [ev for _, ev in window]
|
||||
latest = events[-1]
|
||||
outputs = await _outputs_of(session, latest)
|
||||
fire_event = EventFire(
|
||||
limit = max(loop.ON_TIME_S, grace)
|
||||
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}
|
||||
out: list[ScheduleFire] = []
|
||||
if missed:
|
||||
fire = await _skip_missed(schedule_id, missed, tz, grace)
|
||||
if fire is not None:
|
||||
out.append(fire)
|
||||
await _close_window(missed, fire.id)
|
||||
for w in timely:
|
||||
fire = await service.fire_once(schedule_id, w.due, "event", event=_event_fire(w.events, tz, outputs[id(w)]))
|
||||
if fire is not None:
|
||||
out.append(fire)
|
||||
await _close_window([w], fire.id)
|
||||
return out
|
||||
|
||||
|
||||
def _event_fire(events: list[TriggerEvent], tz: str, outputs: dict[str, Any]) -> EventFire:
|
||||
"""几条合成一次的事件：内容取最新的那一条，深度取最深的那一条加 1。"""
|
||||
latest = events[-1]
|
||||
return EventFire(
|
||||
event=latest.event, subject_id=latest.subject_id, subject_name=latest.subject_name,
|
||||
values=triggers.values(latest.event, latest.payload, tz, outputs=outputs),
|
||||
text=triggers.event_text(latest.event, latest.subject_name, latest.payload),
|
||||
depth=max(int(ev.depth or 0) for ev in events) + 1, chain=[str(x) for x in latest.chain or []],
|
||||
event_ids=[ev.id for ev in events], first_at=_iso(events[0].occurred_at), last_at=_iso(latest.occurred_at))
|
||||
lateness = (now - due).total_seconds()
|
||||
if lateness > max(loop.ON_TIME_S, grace):
|
||||
reason = (f"事件发生在 {service.local_text(events[0].occurred_at, tz)}，服务当时没有派发（调度未运行或已停止），"
|
||||
f"超出补发宽限（{service.duration_text(grace)}），没有发起")
|
||||
fire = await service.record_fire(schedule_id, due, "event", "skipped_misfire", reason,
|
||||
{"event": fire_event.brief()})
|
||||
|
||||
|
||||
async def _skip_missed(schedule_id: str, missed: list[_Window], tz: str, grace: int) -> ScheduleFire | None:
|
||||
"""已过补发宽限的几个窗口合成一条 skipped_misfire（fire_at 取最后一个的结束时刻）：合并了几次事件、几个窗口、时间跨度、
|
||||
合掉的期次。派给任务的时刻离事件发生已经超过宽限的（调度停过、没在派）才说「调度未运行」，否则照实说积压。"""
|
||||
from app.scheduler import service
|
||||
|
||||
events = [ev for w in missed for ev in w.events]
|
||||
events.sort(key=lambda ev: service.as_utc(ev.occurred_at))
|
||||
late_routed = any((routed - service.as_utc(ev.occurred_at)).total_seconds() > grace
|
||||
for w in missed for _, ev, routed in w.items)
|
||||
first, last = service.local_text(events[0].occurred_at, tz), service.local_text(events[-1].occurred_at, tz)
|
||||
span = first if first == last else f"{first} 至 {last}"
|
||||
count = f"{len(events)} 次事件" + (f"（{len(missed)} 个去抖窗口）" if len(missed) > 1 else "")
|
||||
if late_routed:
|
||||
why = "服务当时没有派发（调度未运行或已停止）"
|
||||
else:
|
||||
fire = await service.fire_once(schedule_id, due, "event", event=fire_event)
|
||||
if fire is not None:
|
||||
async with SessionLocal() as session:
|
||||
await _close(session, TriggerDelivery.id.in_([did for did, _ in window]), "fired", fire_id=fire.id)
|
||||
return fire
|
||||
why = "事件积压，轮到派发时"
|
||||
reason = f"{span} 的 {count}：{why}，已超出补发宽限（{service.duration_text(grace)}），没有发起；积压的事件不补"
|
||||
brief = _event_fire(events, tz, {}).brief()
|
||||
brief["windows"] = len(missed)
|
||||
periods = [p for p in dict.fromkeys(str((ev.payload or {}).get("period") or "") for ev in events) if p]
|
||||
if events[-1].event == "table.version" and periods:
|
||||
brief["periods"] = periods
|
||||
return await service.record_fire(schedule_id, max(w.due for w in missed), "event", "skipped_misfire", reason,
|
||||
{"event": brief})
|
||||
|
||||
|
||||
async def _close_window(wins: list[_Window], fire_id: str) -> None:
|
||||
async with SessionLocal() as session:
|
||||
await _close(session, TriggerDelivery.id.in_([did for w in wins for did, _, _ in w.items]), "fired",
|
||||
fire_id=fire_id)
|
||||
|
||||
|
||||
def _pending(schedule_id: str) -> Any:
|
||||
@@ -292,13 +450,13 @@ async def _close(session: Any, cond: Any, status: str, *, fire_id: str | None =
|
||||
|
||||
|
||||
async def _outputs_of(session: Any, ev: TriggerEvent) -> dict[str, Any]:
|
||||
"""workflow.finished 的出口字段：读那次运行的成果（运行已经结束、不会再变）。读不到为空。"""
|
||||
"""workflow.finished 的出口字段：读那次运行的成果（运行已经结束、不会再变），过滤同预览、立即运行（triggers.outputs_of）。
|
||||
读不到为空。"""
|
||||
if ev.event != "workflow.finished":
|
||||
return {}
|
||||
run_id = (ev.payload or {}).get("run_id")
|
||||
run = await session.get(Run, run_id) if run_id else None
|
||||
output = run.output if run is not None and isinstance(run.output, dict) else {}
|
||||
return {k: v for k, v in output.items() if not str(k).startswith("_")}
|
||||
return triggers.outputs_of(run.output if run is not None else None)
|
||||
|
||||
|
||||
def _iso(moment: datetime | None) -> str | None:
|
||||
|
||||
@@ -8,11 +8,15 @@
|
||||
- table.version：一张按期累积的上传表格启用了新的一期（data/table_versions 启用那一下）。可选只认新增的一期（period=new）：
|
||||
重传、替换已有那一期（改错数）的不算；
|
||||
- workflow.finished：一个工作流的正式运行成功结束。可选只认某个应用（app_id）发起的、某种来源（origin_kind）的；
|
||||
下游的输入模板能引用这次运行的出口字段（{{ event.outputs.<字段> }}）；
|
||||
下游的输入模板能引用这次运行的出口字段（{{ event.outputs.<字段> }}）。两种成功结束缺省不认：途经驳回的（审批接了驳回出口、
|
||||
按驳回那条路走完，状态仍是 succeeded，事件里带 rejected_via）——下游会以为「A 成功出了结果」，其实那一期被驳回了，
|
||||
include_rejected 时才认、模板里有 {{ event.rejected }}；应用补算的（app_backfill，升版后按历次输入补算，一连发好几条、
|
||||
各是某一期的历史输入），include_backfill 或只认这种来源时才认；
|
||||
- source.changed：一个数据源出现「要处理」级的变化。按源去抖：同一次探查、剖析产生的一批只记一条事件（data/source_changes）。
|
||||
|
||||
trigger 的形状：{event, 对象（source_id / workflow_id / source_id）, 过滤（见上）, debounce_s（缺省 60）}。同一个定时任务
|
||||
在 debounce_s 内的几次事件合成一次触发（scheduler/outbox）。
|
||||
trigger 的形状：{event, 对象（source_id / workflow_id / source_id）, 过滤（period；app_id、origin_kind、include_rejected、
|
||||
include_backfill）, debounce_s（缺省 60）}。同一个定时任务在 debounce_s 内的几次事件合成一次触发，上传表格按期次分窗口
|
||||
（window_key，scheduler/outbox）。
|
||||
|
||||
**防环。** workflow.finished 能把工作流串起来：A 跑完触发 B、B 跑完又触发 A 就停不下来。保存时查串联成不成环（find_cycle，
|
||||
停用的任务也算：启用不该绕过这道检查）；运行时按 origin 里记的触发链深度兜底，超过 MAX_DEPTH 层不发起、照实记一次未能发起
|
||||
@@ -56,7 +60,10 @@ ORIGIN_LABEL = {
|
||||
}
|
||||
#: trigger 里认得的键（多出来的拒收：拼错的过滤条件悄悄不生效，比报错更糟）。subject_name 是保存时记下的对象名（接口不给、
|
||||
#: 带回来也不认，对象删掉之后说法里用它）
|
||||
_KEYS = {"event", "source_id", "workflow_id", "period", "app_id", "origin_kind", "debounce_s", "subject_name"}
|
||||
_KEYS = {"event", "source_id", "workflow_id", "period", "app_id", "origin_kind", "debounce_s", "subject_name",
|
||||
"include_rejected", "include_backfill"}
|
||||
#: 应用补算发起的运行的来源（engine/origin.py）
|
||||
BACKFILL_ORIGIN = "app_backfill"
|
||||
#: 示例事件里给不出真实值时写的占位说明
|
||||
EXAMPLE_NOTE = "示例值"
|
||||
|
||||
@@ -118,6 +125,18 @@ def normalize(value: Any) -> dict[str, Any]:
|
||||
labels = "、".join(ORIGIN_LABEL[o] for o in FORMAL_ORIGINS)
|
||||
raise TriggerError(f"只认的来源需要是{labels}之一")
|
||||
out[name] = text
|
||||
for name in ("include_rejected", "include_backfill"):
|
||||
raw = value.get(name)
|
||||
if raw in (None, False):
|
||||
continue
|
||||
if event != "workflow.finished":
|
||||
raise TriggerError("只有「上游工作流运行成功结束」能选也认途经驳回的、应用补算的")
|
||||
if raw is not True:
|
||||
raise TriggerError("也认途经驳回的、应用补算的需要是 true 或 false")
|
||||
# 只认补算的：补算本来就认，不再另记
|
||||
if name == "include_backfill" and out.get("origin_kind") == BACKFILL_ORIGIN:
|
||||
continue
|
||||
out[name] = True
|
||||
debounce = value.get("debounce_s", DEFAULT_DEBOUNCE_S)
|
||||
if isinstance(debounce, bool) or not isinstance(debounce, int) or not 0 <= debounce <= MAX_DEBOUNCE_S:
|
||||
raise TriggerError(f"去抖需要是 0 到 {MAX_DEBOUNCE_S} 之间的整数秒数（最长 1 天）", "debounce_s")
|
||||
@@ -149,6 +168,11 @@ def matches(trigger: dict[str, Any] | None, event: str, subject_id: str, payload
|
||||
return False
|
||||
if trigger.get("origin_kind") and payload.get("origin_kind") != trigger["origin_kind"]:
|
||||
return False
|
||||
if payload.get("rejected") and not trigger.get("include_rejected"):
|
||||
return False
|
||||
if payload.get("origin_kind") == BACKFILL_ORIGIN and not (
|
||||
trigger.get("include_backfill") or trigger.get("origin_kind") == BACKFILL_ORIGIN):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
@@ -282,6 +306,15 @@ def describe(trigger: dict[str, Any], subject_name: str | None, *, app_name: str
|
||||
notes.append(f"只认应用「{app_name}」发起的" if app_name else "只认指定的应用发起的")
|
||||
if trigger.get("origin_kind"):
|
||||
notes.append(f"只认来源为「{ORIGIN_LABEL.get(trigger['origin_kind'], trigger['origin_kind'])}」的")
|
||||
if event == "workflow.finished":
|
||||
# 缺省不认的两种成功结束，照实写出来
|
||||
kinds = [("途经驳回", bool(trigger.get("include_rejected")))]
|
||||
if trigger.get("origin_kind") != BACKFILL_ORIGIN:
|
||||
kinds.append(("应用补算", bool(trigger.get("include_backfill"))))
|
||||
excluded = "、".join(k for k, on in kinds if not on)
|
||||
included = "、".join(k for k, on in kinds if on)
|
||||
notes += [f"{included}的也算"] if included else []
|
||||
notes += [f"{excluded}的不算"] if excluded else []
|
||||
debounce = debounce_of(trigger)
|
||||
from app.scheduler.service import duration_text
|
||||
|
||||
@@ -300,7 +333,8 @@ def event_text(event: str, subject_name: str, payload: dict[str, Any] | None) ->
|
||||
return f"上传表格{name}{what}{f'（{period}）' if period else ''}"
|
||||
if event == "workflow.finished":
|
||||
run = str(payload.get("run_id") or "")
|
||||
return f"工作流{name}的正式运行成功结束{f'（#{run[:6]}）' if run else ''}"
|
||||
rejected = "（途经驳回）" if payload.get("rejected") else ""
|
||||
return f"工作流{name}的正式运行成功结束{f'（#{run[:6]}）' if run else ''}{rejected}"
|
||||
if event == "source.changed":
|
||||
n = (payload.get("counts") or {}).get("new")
|
||||
return f"数据源{name}出现要处理的变化{f'（新记 {n} 条）' if isinstance(n, int) and n else ''}"
|
||||
@@ -326,6 +360,7 @@ VARIABLES: dict[str, tuple[tuple[str, str, str], ...]] = {
|
||||
("workflow_id", "上游工作流的编号", "wf…"),
|
||||
("run_id", "上游那次运行的编号", "run…"),
|
||||
("finished_at", "上游运行结束的时间（带时区偏移的 ISO 时间）", "2026-10-08T09:30:00+08:00"),
|
||||
("rejected", "上游这次运行有没有途经驳回（true / false；缺省不认途经驳回的，勾了「也认」才会是 true）", "false"),
|
||||
("outputs", "上游运行的出口字段（写 event.outputs.<字段名>）", "{…}"),
|
||||
),
|
||||
"source.changed": (
|
||||
@@ -396,7 +431,7 @@ def values(event: str, payload: dict[str, Any] | None, tz: str, *, outputs: dict
|
||||
if event == "workflow.finished":
|
||||
return {"workflow": payload.get("workflow") or "", "workflow_id": payload.get("workflow_id") or "",
|
||||
"run_id": payload.get("run_id") or "", "finished_at": _zoned(payload.get("finished_at"), tz),
|
||||
"outputs": dict(outputs or {})}
|
||||
"rejected": "true" if payload.get("rejected") else "false", "outputs": dict(outputs or {})}
|
||||
if event == "source.changed":
|
||||
counts = payload.get("counts") if isinstance(payload.get("counts"), dict) else {}
|
||||
return {"source": payload.get("source") or "", "source_id": payload.get("source_id") or "",
|
||||
@@ -501,11 +536,34 @@ async def sample(session: Any, trigger: dict[str, Any]) -> Sample:
|
||||
query = query.where(Run.app_id == trigger["app_id"])
|
||||
if trigger.get("origin_kind"):
|
||||
query = query.where(Run.origin_kind == trigger["origin_kind"])
|
||||
run = (await session.execute(query.order_by(Run.finished_at.desc()).limit(1))).scalars().first() if subject else None
|
||||
if run is None:
|
||||
return Sample(event, subject, name, {"workflow": name or EXAMPLE_NOTE, "workflow_id": subject, "run_id": "",
|
||||
"finished_at": ""}, real=False)
|
||||
return Sample(event, subject, name, run_payload(run), outputs=dict(run.output or {}))
|
||||
candidates = list((await session.execute(query.order_by(Run.finished_at.desc()).limit(SAMPLE_SCAN)))
|
||||
.scalars()) if subject else []
|
||||
# 和真派发同一套条件（matches）：缺省不认的途经驳回、应用补算跳过，取合条件的最近一次
|
||||
for run in candidates:
|
||||
payload = run_payload(run, await _rejected_via_of(session, run.id))
|
||||
if matches(trigger, event, subject, payload):
|
||||
return Sample(event, subject, name, payload, outputs=outputs_of(run.output))
|
||||
return Sample(event, subject, name, {"workflow": name or EXAMPLE_NOTE, "workflow_id": subject, "run_id": "",
|
||||
"finished_at": "", "rejected": False}, real=False)
|
||||
|
||||
|
||||
#: 示例事件往回找几次成功的正式运行（找合条件的最近一次）
|
||||
SAMPLE_SCAN = 20
|
||||
|
||||
|
||||
async def _rejected_via_of(session: Any, run_id: str) -> list[Any] | None:
|
||||
"""这次运行途经的驳回：run.finished 事件里的 rejected_via（W1-b 的驳回出口写的），没有为 None。"""
|
||||
from app.db.models import RunEvent
|
||||
|
||||
data = (await session.execute(select(RunEvent.data).where(RunEvent.run_id == run_id, RunEvent.type == "run.finished")
|
||||
.order_by(RunEvent.seq.desc()).limit(1))).scalar()
|
||||
via = data.get("rejected_via") if isinstance(data, dict) else None
|
||||
return list(via) if isinstance(via, list) and via else None
|
||||
|
||||
|
||||
def outputs_of(output: Any) -> dict[str, Any]:
|
||||
"""给模板的出口字段（{{ event.outputs.* }}）：「_」开头的是系统写的（出具档位之类），不给。真派发、预览、立即运行共用。"""
|
||||
return {k: v for k, v in output.items() if not str(k).startswith("_")} if isinstance(output, dict) else {}
|
||||
|
||||
|
||||
def table_payload(name: str, *, snapshot_id: str, period_start: str | None, period_end: str | None,
|
||||
@@ -516,11 +574,21 @@ def table_payload(name: str, *, snapshot_id: str, period_start: str | None, peri
|
||||
"activated_at": _iso(activated_at), "replaced": bool(replaced), "seq": seq}
|
||||
|
||||
|
||||
def run_payload(run: Any) -> dict[str, Any]:
|
||||
"""workflow.finished 的事件内容：不含出口字段（派发时按 run_id 读成果）。"""
|
||||
def run_payload(run: Any, rejected_via: Any = None) -> dict[str, Any]:
|
||||
"""workflow.finished 的事件内容：不含出口字段（派发时按 run_id 读成果）。rejected_via 是途经了哪几个关卡的驳回（W1-b 的
|
||||
驳回出口；runner 收尾时交进来），没途经驳回为 None。"""
|
||||
via = (list(rejected_via) if isinstance(rejected_via, (list, tuple)) else [rejected_via]) if rejected_via else None
|
||||
return {"workflow": run.workflow_name or "", "workflow_id": run.workflow_id or "", "run_id": run.id,
|
||||
"finished_at": _iso(run.finished_at), "app_id": run.app_id, "origin_kind": run.origin_kind,
|
||||
"version": run.version}
|
||||
"version": run.version, "rejected": bool(via), "rejected_via": via}
|
||||
|
||||
|
||||
def window_key(event: str, payload: dict[str, Any] | None) -> str:
|
||||
"""去抖时按什么分窗口：上传表格按期次分（9 月、10 月连着启用，各有各的触发，不把 9 月合掉），别的不分。"""
|
||||
if event != "table.version":
|
||||
return ""
|
||||
payload = payload or {}
|
||||
return str(payload.get("period") or f"{payload.get('period_start') or ''}~{payload.get('period_end') or ''}")
|
||||
|
||||
|
||||
def _iso(moment: datetime | None) -> str:
|
||||
|
||||
@@ -369,3 +369,21 @@ async def test_trigger_options(client):
|
||||
assert flows[draft]["ok"] is False and "发布" in flows[draft]["reason"]
|
||||
assert "hook" in [o["kind"] for o in out["origins"]] and out["max_depth"] == 5
|
||||
assert out["default_debounce_s"] == 60
|
||||
|
||||
|
||||
async def test_include_rejected_and_backfill_through_the_api(client):
|
||||
"""复核 M1、C5：上游工作流的触发缺省不认途经驳回的、应用补算的，说法照实写；勾了「也认」的存下来、说法跟着变；
|
||||
别的事件带这两项拒收"""
|
||||
a, a_name = await workflow("日报")
|
||||
b, _ = await workflow("汇总", fields=[{"name": "rej"}])
|
||||
plain = (await client.post("/api/schedules", json=body(b, {"event": "workflow.finished", "workflow_id": a},
|
||||
input_template={"rej": "{{ event.rejected }}"}))).json()
|
||||
assert "途经驳回、应用补算的不算" in plain["trigger_text"] and "include_rejected" not in plain["trigger"]
|
||||
r = await client.patch(f"/api/schedules/{plain['id']}", json={
|
||||
"trigger": {"event": "workflow.finished", "workflow_id": a, "include_rejected": True}, "version": plain["version"]})
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["trigger"]["include_rejected"] is True and "途经驳回的也算；应用补算的不算" in r.json()["trigger_text"]
|
||||
src = await upload_table("门店销售")
|
||||
bad = await client.post("/api/schedules", json=body(b, {"event": "table.version", "source_id": src,
|
||||
"include_backfill": True}))
|
||||
assert bad.status_code == 422 and bad.json()["field"] == "trigger"
|
||||
|
||||
@@ -0,0 +1,277 @@
|
||||
"""数据事件触发复核意见的修正（docs/design/orchestration-round/00-总方案.md §4；scheduler/outbox.py、triggers.py）。
|
||||
|
||||
- M1 途经驳回：正式运行走驳回出口跑完（succeeded，run.finished 带 rejected_via）照发 workflow.finished，事件内容带 rejected_via、
|
||||
说法接「（途经驳回）」、模板有 {{ event.rejected }}；触发条件缺省只认没途经驳回的，include_rejected 时也认；
|
||||
- C1 去抖积压：一拍处理到没有到点的窗口为止；已过宽限的几个窗口合成一条跳过（合并了几条、时间跨度），「调度未运行」只在
|
||||
派发确实晚到（派给任务的时刻离事件发生超过宽限）时说，否则照实说积压；积压清完之后新来的事件照常触发；
|
||||
- C3 source.changed 按任务补写：同一个任务只记一条（dedupe），任务结束后没记上的（进程死了、库被锁）在调度的一拍按 job_id 补写；
|
||||
- C4 不同期次的 table.version 不合进同一个窗口：各期有各自的触发；同一期的重传照旧合并；
|
||||
- C5 应用补算（app_backfill）发的 workflow.finished 缺省不认，include_backfill 或只认补算时才认；
|
||||
- C6 预览、立即运行的示例事件和真派发同一套出口字段过滤（「_」开头的不给模板）。
|
||||
|
||||
工作流、表格、数据源都是虚构的。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import timedelta
|
||||
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.db.base import SessionLocal, utcnow
|
||||
from app.db.models import DataSource, Run, Schedule
|
||||
from app.db.models_sources import IntrospectJob, SourceChange
|
||||
from app.db.models_triggers import TriggerDelivery, TriggerEvent
|
||||
from app.scheduler import outbox, triggers
|
||||
from tests.test_schedule_events import ( # noqa: F401 （engine_up：每个测试清掉定时任务和事件）
|
||||
deliveries, engine_up, event_schedule, events, fires, formal_run, later, put_event, settle, workflow,
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# M1 途经驳回
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_rejected_via_is_carried_and_excluded_by_default():
|
||||
upstream = await workflow("门店日报")
|
||||
plain = await event_schedule(await workflow("区域汇总"),
|
||||
{"event": "workflow.finished", "workflow_id": upstream, "debounce_s": 0}, name="不认驳回")
|
||||
also = await event_schedule(await workflow("驳回复核", fields=[{"name": "rej"}]),
|
||||
{"event": "workflow.finished", "workflow_id": upstream, "debounce_s": 0,
|
||||
"include_rejected": True}, template={"rej": "{{ event.rejected }}"}, name="也认驳回")
|
||||
a = await formal_run(upstream)
|
||||
async with SessionLocal() as session:
|
||||
# runner 收尾时把这次运行途经的驳回交进来（W1-b 的 _rejected_via(state)）：这里直接调，覆盖那一下
|
||||
await session.execute(TriggerEvent.__table__.delete().where(TriggerEvent.subject_id == upstream))
|
||||
outbox.stage_run_finished(session, await session.get(Run, a.id),
|
||||
rejected_via=[{"node_id": "gate", "label": "主管审核"}])
|
||||
await session.commit()
|
||||
[ev] = await events("workflow.finished", upstream)
|
||||
assert ev.payload["rejected"] is True and ev.payload["rejected_via"] == [{"node_id": "gate", "label": "主管审核"}]
|
||||
assert triggers.event_text(ev.event, ev.subject_name, ev.payload).endswith("（途经驳回）")
|
||||
[fire] = await outbox.dispatch_due(later())
|
||||
assert fire.schedule_id == also and fire.status == "started" and fire.inputs["inputs"] == {"rej": "true"}
|
||||
assert fire.inputs["event"]["text"].endswith("（途经驳回）")
|
||||
assert await fires(plain) == [] and await deliveries(plain) == []
|
||||
await settle(fire.run_id)
|
||||
|
||||
|
||||
async def test_plain_runs_say_not_rejected():
|
||||
upstream = await workflow("门店日报")
|
||||
sid = await event_schedule(await workflow("区域汇总", fields=[{"name": "rej"}]),
|
||||
{"event": "workflow.finished", "workflow_id": upstream, "debounce_s": 0},
|
||||
template={"rej": "{{ event.rejected }}"})
|
||||
await formal_run(upstream) # runner 收尾没交 rejected_via：没途经驳回
|
||||
[ev] = await events("workflow.finished", upstream)
|
||||
assert ev.payload["rejected"] is False and ev.payload["rejected_via"] is None
|
||||
[fire] = await outbox.dispatch_due(later())
|
||||
assert fire.schedule_id == sid and fire.inputs["inputs"] == {"rej": "false"}
|
||||
assert "途经驳回" not in fire.inputs["event"]["text"]
|
||||
await settle(fire.run_id)
|
||||
|
||||
|
||||
def test_rejected_and_backfill_options_are_described():
|
||||
base = triggers.normalize({"event": "workflow.finished", "workflow_id": "wf"})
|
||||
assert "（途经驳回、应用补算的不算；去抖 1 分钟）" in triggers.describe(base, "门店日报")
|
||||
both = triggers.normalize({"event": "workflow.finished", "workflow_id": "wf", "include_rejected": True,
|
||||
"include_backfill": True})
|
||||
assert both["include_rejected"] is True and both["include_backfill"] is True
|
||||
assert "（途经驳回、应用补算的也算；去抖 1 分钟）" in triggers.describe(both, "门店日报")
|
||||
# 只认应用补算的：补算本来就要认，不再说「补算的不算」
|
||||
only = triggers.normalize({"event": "workflow.finished", "workflow_id": "wf", "origin_kind": "app_backfill"})
|
||||
text = triggers.describe(only, "门店日报")
|
||||
assert "只认来源为「应用补算」的" in text and "应用补算的不算" not in text and "途经驳回的不算" in text
|
||||
# 别的事件不收这两项
|
||||
for key in ("include_rejected", "include_backfill"):
|
||||
try:
|
||||
triggers.normalize({"event": "table.version", "source_id": "x", key: True})
|
||||
except triggers.TriggerError as e:
|
||||
assert "上游工作流运行成功结束" in str(e)
|
||||
else:
|
||||
raise AssertionError(key)
|
||||
assert "event.rejected" in triggers.names("workflow.finished")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# C5 应用补算
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_backfill_runs_are_excluded_unless_asked():
|
||||
payload = {"origin_kind": "app_backfill", "rejected": False}
|
||||
base = triggers.normalize({"event": "workflow.finished", "workflow_id": "wf"})
|
||||
assert not triggers.matches(base, "workflow.finished", "wf", payload)
|
||||
assert triggers.matches(base, "workflow.finished", "wf", {**payload, "origin_kind": "launcher"})
|
||||
assert triggers.matches({**base, "include_backfill": True}, "workflow.finished", "wf", payload)
|
||||
assert triggers.matches({**base, "origin_kind": "app_backfill"}, "workflow.finished", "wf", payload)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# C6 示例事件的出口字段
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_sample_outputs_use_the_same_filter_as_dispatch():
|
||||
upstream = await workflow("门店日报")
|
||||
run = await formal_run(upstream)
|
||||
async with SessionLocal() as session:
|
||||
await session.execute(Run.__table__.update().where(Run.id == run.id).values(
|
||||
output={"r": "日报完成", "_issuance": {"tier": "full"}}))
|
||||
await session.commit()
|
||||
got = await triggers.sample(session, {"event": "workflow.finished", "workflow_id": upstream})
|
||||
assert got.real and got.outputs == {"r": "日报完成"}
|
||||
assert triggers.outputs_of({"r": 1, "_x": 2}) == {"r": 1}
|
||||
|
||||
|
||||
async def test_sample_skips_rejected_and_backfill_runs_by_default():
|
||||
upstream = await workflow("门店日报")
|
||||
plain = await formal_run(upstream)
|
||||
backfill = await formal_run(upstream, origin_kind="app_backfill")
|
||||
async with SessionLocal() as session:
|
||||
got = await triggers.sample(session, {"event": "workflow.finished", "workflow_id": upstream})
|
||||
wide = await triggers.sample(session, {"event": "workflow.finished", "workflow_id": upstream,
|
||||
"include_backfill": True})
|
||||
assert got.payload["run_id"] == plain.id and wide.payload["run_id"] == backfill.id
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# C1 去抖积压
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def _backlog(upstream: str, sid: str, *, n: int, gap_s: int, routed_late: bool) -> tuple:
|
||||
"""n 条相隔 gap_s 秒的事件，派给了任务、还没处理。routed_late：派给任务的时刻是现在（调度停过）；否则发生后 1 秒就派了
|
||||
（调度在跑，只是没处理完）。"""
|
||||
t0 = utcnow().replace(microsecond=0) - timedelta(seconds=gap_s * (n - 1) + 61)
|
||||
ids = []
|
||||
async with SessionLocal() as session:
|
||||
for i in range(n):
|
||||
at = t0 + timedelta(seconds=gap_s * i)
|
||||
ev = TriggerEvent(event="workflow.finished", subject_id=upstream, subject_name="门店日报",
|
||||
payload={"workflow_id": upstream, "run_id": f"run-{i}", "rejected": False},
|
||||
occurred_at=at, created_at=at, status="routed", matched=1)
|
||||
session.add(ev)
|
||||
await session.flush()
|
||||
session.add(TriggerDelivery(schedule_id=sid, event_id=ev.id, occurred_at=at,
|
||||
created_at=utcnow() if routed_late else at + timedelta(seconds=1)))
|
||||
ids.append(ev.id)
|
||||
await session.commit()
|
||||
return t0, ids
|
||||
|
||||
|
||||
async def test_backlog_is_drained_in_one_tick_and_missed_windows_merge_into_one_skip():
|
||||
"""12 条相隔 5 分钟、去抖 60 秒、宽限 600 秒：一拍处理完。过了宽限的 10 个窗口合成一条跳过（合并 10 次、写时间跨度），
|
||||
宽限内的两个照常发起（第二个照重叠规则可能跳过）；积压清完以后新来的事件照常触发"""
|
||||
upstream = await workflow("门店日报")
|
||||
sid = await event_schedule(await workflow("区域汇总", fields=[{"name": "run"}]),
|
||||
{"event": "workflow.finished", "workflow_id": upstream, "debounce_s": 60}, grace=600,
|
||||
template={"run": "{{ event.run_id }}"})
|
||||
t0, ids = await _backlog(upstream, sid, n=12, gap_s=300, routed_late=True)
|
||||
done = await outbox.dispatch_due(utcnow())
|
||||
assert len(done) == 3, [(f.status, f.reason) for f in done]
|
||||
skip, *rest = done
|
||||
assert skip.status == "skipped_misfire" and skip.kind == "event"
|
||||
assert skip.inputs["event"]["merged"] == 10 and skip.inputs["event"]["windows"] == 10
|
||||
assert skip.inputs["event"]["event_ids"] == ids[:10]
|
||||
assert "服务当时没有派发（调度未运行或已停止）" in skip.reason and " 至 " in skip.reason and "10 次事件" in skip.reason
|
||||
assert [f.inputs["event"]["event_ids"] for f in rest] == [[ids[10]], [ids[11]]]
|
||||
assert rest[0].status == "started" and rest[1].status in ("started", "skipped_overlap")
|
||||
assert {d.status for d in await deliveries(sid)} == {"fired"}
|
||||
for f in rest:
|
||||
if f.run_id:
|
||||
await settle(f.run_id)
|
||||
# 积压清完：新来的一条照常触发
|
||||
new = await put_event("workflow.finished", upstream, {"workflow_id": upstream, "run_id": "run-new"}, utcnow())
|
||||
[fresh] = await outbox.dispatch_due(later(61))
|
||||
assert fresh.status == "started" and fresh.inputs["event"]["event_ids"] == [new]
|
||||
await settle(fresh.run_id)
|
||||
|
||||
|
||||
async def test_backlog_while_scheduler_ran_says_backlog_not_scheduler_off():
|
||||
upstream = await workflow("门店日报")
|
||||
sid = await event_schedule(await workflow("区域汇总"),
|
||||
{"event": "workflow.finished", "workflow_id": upstream, "debounce_s": 60}, grace=600)
|
||||
await _backlog(upstream, sid, n=4, gap_s=600, routed_late=False)
|
||||
skip, *rest = await outbox.dispatch_due(utcnow())
|
||||
assert skip.status == "skipped_misfire" and skip.inputs["event"]["merged"] == 3
|
||||
assert "调度未运行" not in skip.reason and "积压" in skip.reason, skip.reason
|
||||
for f in rest:
|
||||
if f.run_id:
|
||||
await settle(f.run_id)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# C4 不同期次各自一个窗口
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_different_periods_get_their_own_windows():
|
||||
table = uuid.uuid4().hex
|
||||
sid = await event_schedule(await workflow("月报", fields=[{"name": "month"}]),
|
||||
{"event": "table.version", "source_id": table, "debounce_s": 60},
|
||||
template={"month": "{{ event.period }}"})
|
||||
t0 = utcnow().replace(microsecond=0)
|
||||
payload = lambda p, s, e: {"table": "门店销售", "period": p, "period_start": s, "period_end": e, # noqa: E731
|
||||
"version": f"snap-{p}", "replaced": False}
|
||||
sep = await put_event("table.version", table, payload("2026-09", "2026-09-01", "2026-09-30"), t0, name="门店销售")
|
||||
oct_ = await put_event("table.version", table, payload("2026-10", "2026-10-01", "2026-10-31"),
|
||||
t0 + timedelta(seconds=30), name="门店销售")
|
||||
again = await put_event("table.version", table, {**payload("2026-09", "2026-09-01", "2026-09-30"), "replaced": True},
|
||||
t0 + timedelta(seconds=40), name="门店销售")
|
||||
[first] = await outbox.dispatch_due(t0 + timedelta(seconds=61))
|
||||
# 9 月一个窗口：9 月和 9 月的重传合在一起（内容取最新的那一条），10 月不合进来
|
||||
assert first.inputs["event"]["event_ids"] == [sep, again] and first.inputs["inputs"] == {"month": "2026-09"}
|
||||
await settle(first.run_id)
|
||||
[second] = await outbox.dispatch_due(t0 + timedelta(seconds=91))
|
||||
assert second.inputs["event"]["event_ids"] == [oct_] and second.inputs["inputs"] == {"month": "2026-10"}
|
||||
assert second.fire_at == t0 + timedelta(seconds=90) and second.status == "started"
|
||||
await settle(second.run_id)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# C3 source.changed 按任务补写
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def _source_with_changes(job_status: str) -> tuple[str, str]:
|
||||
async with SessionLocal() as session:
|
||||
source = DataSource(name=f"门店库-{uuid.uuid4().hex[:6]}", kind="sqlite", database=":memory:")
|
||||
session.add(source)
|
||||
await session.flush()
|
||||
job = IntrospectJob(source_id=source.id, status=job_status,
|
||||
finished_at=utcnow() if job_status != "running" else None)
|
||||
session.add(job)
|
||||
await session.flush()
|
||||
for i in range(2):
|
||||
session.add(SourceChange(source_id=source.id, source_name=source.name, table_name="orders",
|
||||
column_name=f"c{i}", kind="column_removed", severity="action",
|
||||
detected_by={"job_kind": "introspect", "job_id": job.id},
|
||||
fingerprint=uuid.uuid4().hex, status="open"))
|
||||
await session.commit()
|
||||
return source.id, job.id
|
||||
|
||||
|
||||
async def test_source_changed_is_backfilled_once_per_job():
|
||||
"""任务结束了、变化记下了，那条事件却没记上（进程在中间死了）：调度的一拍按 job_id 补写，只补一次；之后再走到
|
||||
notify_after_job 也不重复"""
|
||||
source_id, job_id = await _source_with_changes("succeeded")
|
||||
running_source, _ = await _source_with_changes("running")
|
||||
assert await events("source.changed", source_id) == []
|
||||
assert await outbox.backfill_source_changes(utcnow()) >= 1
|
||||
[ev] = await events("source.changed", source_id)
|
||||
assert ev.payload["job_id"] == job_id and ev.payload["counts"]["new"] == 2 and ev.dedupe_key == f"source.changed:{job_id}"
|
||||
# 还在进行的任务不补：它结束时自己会记
|
||||
assert await events("source.changed", running_source) == []
|
||||
await outbox.backfill_source_changes(utcnow())
|
||||
from app.data import source_changes
|
||||
|
||||
await source_changes.notify_after_job(source_id, job_id)
|
||||
assert len(await events("source.changed", source_id)) == 1
|
||||
|
||||
|
||||
async def test_dispatch_tick_runs_the_backfill():
|
||||
source_id, _ = await _source_with_changes("succeeded")
|
||||
outbox._backfilled_at = None # 刚起的服务：第一拍就补
|
||||
await outbox.dispatch_due(utcnow())
|
||||
assert len(await events("source.changed", source_id)) == 1
|
||||
@@ -55,6 +55,10 @@ export const TRIGGER_TEXT = {
|
||||
originLabel: '只认来源',
|
||||
originAny: '不限来源',
|
||||
filterHint: '只在上游那次正式运行由这个应用、这种来源发起时触发',
|
||||
includeRejected: '也认途经驳回的',
|
||||
includeRejectedHint: '上游某个审批被驳回、按驳回那条路走完时，运行仍算成功结束。不勾时这类运行不触发；勾了以后，输入模板里的 {{ event.rejected }} 是 true',
|
||||
includeBackfill: '也认应用补算的',
|
||||
includeBackfillHint: '应用升版后按历次输入补算，会连着结束好几次正式运行，每次是某一期的历史输入。不勾时补算的运行不触发',
|
||||
debounce: '去抖',
|
||||
debounceHint: '这段时间内的几次事件合成一次触发，按最新的那一次渲染输入；上一次运行还没结束时仍按重叠策略跳过',
|
||||
debounceOption: (span: string) => (span ? `${span}内合成一次` : '不去抖，每次事件都触发'),
|
||||
@@ -86,6 +90,9 @@ export const TRIGGER_TEXT = {
|
||||
merged: (n: number) => `合并了 ${n} 次事件`,
|
||||
depth: (n: number) => `触发链第 ${n} 层`,
|
||||
sample: '立即运行，按最近一次事件',
|
||||
/** 几个过了补发宽限的去抖窗口合成一条跳过 */
|
||||
windows: (n: number) => `${n} 个去抖窗口`,
|
||||
periods: (list: string) => `期次 ${list}`,
|
||||
sampleExample: '立即运行，事件为示例值',
|
||||
|
||||
// ---- 运行详情页头
|
||||
|
||||
@@ -199,6 +199,8 @@ function FireRow({ fire: f, schedule, now }: { fire: ScheduleFire; schedule: Sch
|
||||
function EventFire({ event }: { event: ScheduleFireEvent }) {
|
||||
const notes = [
|
||||
event.merged > 1 ? TT.merged(event.merged) : null,
|
||||
event.windows && event.windows > 1 ? TT.windows(event.windows) : null,
|
||||
event.periods && event.periods.length > 1 ? TT.periods(event.periods.join('、')) : null,
|
||||
event.depth > 1 ? TT.depth(event.depth) : null,
|
||||
event.sample ? (event.real ? TT.sample : TT.sampleExample) : null,
|
||||
].filter(Boolean)
|
||||
|
||||
@@ -6,7 +6,9 @@ import { errorMessage } from '../../lib/errors'
|
||||
import { timezoneOptions } from '../../lib/freq'
|
||||
import { TRIGGER_TEXT as TT } from '../../lib/termsTrigger'
|
||||
import type { ScheduleEventKind, ScheduleTriggerKind, ScheduleTriggerOptions } from '../../types'
|
||||
import { DEBOUNCE_OPTIONS, EVENT_KINDS, TRIGGER_KINDS, graceLabel, switchEvent, type TriggerForm } from './model'
|
||||
import {
|
||||
BACKFILL_ORIGIN, DEBOUNCE_OPTIONS, EVENT_KINDS, TRIGGER_KINDS, graceLabel, switchEvent, type TriggerForm,
|
||||
} from './model'
|
||||
import { isUnsupported } from './parts'
|
||||
|
||||
/**
|
||||
@@ -179,6 +181,21 @@ export function EventSection({ value, onChange, timezone, setTimezone, savedName
|
||||
</Field>
|
||||
</div>
|
||||
<p className="text-2xs leading-relaxed text-faint">{TT.filterHint}</p>
|
||||
{/* 缺省不认的两种成功结束：途经驳回的、应用补算的（服务端 triggers.matches），要认的勾上 */}
|
||||
<div className="space-y-1 pt-1">
|
||||
<label className="flex items-start gap-2 text-xs">
|
||||
<input type="checkbox" className="mt-0.5" checked={value.includeRejected} data-trigger-include-rejected=""
|
||||
onChange={(e) => set({ includeRejected: e.target.checked })} />
|
||||
<span>{TT.includeRejected}<span className="block text-2xs leading-relaxed text-faint">{TT.includeRejectedHint}</span></span>
|
||||
</label>
|
||||
{value.originKind !== BACKFILL_ORIGIN && (
|
||||
<label className="flex items-start gap-2 text-xs">
|
||||
<input type="checkbox" className="mt-0.5" checked={value.includeBackfill} data-trigger-include-backfill=""
|
||||
onChange={(e) => set({ includeBackfill: e.target.checked })} />
|
||||
<span>{TT.includeBackfill}<span className="block text-2xs leading-relaxed text-faint">{TT.includeBackfillHint}</span></span>
|
||||
</label>
|
||||
)}
|
||||
</div>
|
||||
<p className="text-2xs leading-relaxed text-faint" data-trigger-chain="">{TT.chainHint(options.data.max_depth)}</p>
|
||||
</div>
|
||||
)}
|
||||
|
||||
@@ -43,9 +43,15 @@ export interface TriggerForm {
|
||||
period: 'any' | 'new'
|
||||
appId: string
|
||||
originKind: string
|
||||
/** 也认途经驳回的、也认应用补算的（只给上游工作流，缺省都不认） */
|
||||
includeRejected: boolean
|
||||
includeBackfill: boolean
|
||||
debounceS: number
|
||||
}
|
||||
|
||||
/** 应用补算的来源：选了「只认应用补算」时，「也认应用补算的」不再适用 */
|
||||
export const BACKFILL_ORIGIN = 'app_backfill'
|
||||
|
||||
export const EVENT_KINDS: readonly ScheduleEventKind[] = ['table.version', 'workflow.finished', 'source.changed']
|
||||
export const TRIGGER_KINDS: readonly ScheduleTriggerKind[] = ['cron', 'event']
|
||||
/** 去抖的选项（秒）：不去抖、1 分钟（缺省）、5 分钟、15 分钟、1 小时 */
|
||||
@@ -60,6 +66,8 @@ export function triggerFormOf(t: ScheduleTrigger | null | undefined): TriggerFor
|
||||
period: t?.period === 'new' ? 'new' : 'any',
|
||||
appId: t?.app_id ?? '',
|
||||
originKind: t?.origin_kind ?? '',
|
||||
includeRejected: t?.include_rejected === true,
|
||||
includeBackfill: t?.include_backfill === true,
|
||||
debounceS: typeof t?.debounce_s === 'number' ? t.debounce_s : DEFAULT_DEBOUNCE_S,
|
||||
}
|
||||
}
|
||||
@@ -71,6 +79,8 @@ export function triggerBody(t: TriggerForm): ScheduleTrigger {
|
||||
out.workflow_id = t.subjectId
|
||||
if (t.appId) out.app_id = t.appId
|
||||
if (t.originKind) out.origin_kind = t.originKind
|
||||
if (t.includeRejected) out.include_rejected = true
|
||||
if (t.includeBackfill && t.originKind !== BACKFILL_ORIGIN) out.include_backfill = true
|
||||
} else {
|
||||
out.source_id = t.subjectId
|
||||
if (t.event === 'table.version' && t.period === 'new') out.period = 'new'
|
||||
@@ -81,7 +91,8 @@ export function triggerBody(t: TriggerForm): ScheduleTrigger {
|
||||
/** 换了事件种类：对象、过滤都不再适用，清掉；去抖留着 */
|
||||
export function switchEvent(t: TriggerForm, event: ScheduleEventKind): TriggerForm {
|
||||
if (t.event === event) return t
|
||||
return { event, subjectId: '', period: 'any', appId: '', originKind: '', debounceS: t.debounceS }
|
||||
return { event, subjectId: '', period: 'any', appId: '', originKind: '', includeRejected: false, includeBackfill: false,
|
||||
debounceS: t.debounceS }
|
||||
}
|
||||
|
||||
export interface ScheduleForm {
|
||||
|
||||
@@ -4069,6 +4069,12 @@ export interface ScheduleTrigger {
|
||||
/** 只认某个应用发起的、某种来源的运行，只给 workflow.finished */
|
||||
app_id?: string
|
||||
origin_kind?: string
|
||||
/**
|
||||
* 也认途经驳回的（上游走了某个审批的驳回出口、按驳回那条路跑完，状态仍是成功）、也认应用补算的；缺省都不认。只给
|
||||
* workflow.finished，不写即 false
|
||||
*/
|
||||
include_rejected?: boolean
|
||||
include_backfill?: boolean
|
||||
/** 去抖：同一个任务这么多秒内的几次事件合成一次 */
|
||||
debounce_s: number
|
||||
/** 保存时记下的对象名（对象删掉之后界面写它）；读出来才有，交上去不用带 */
|
||||
@@ -4108,6 +4114,9 @@ export interface ScheduleFireEvent {
|
||||
/** 立即运行时按示例事件发起的（不是真来了一次事件）；real 为假时是示例值 */
|
||||
sample: boolean
|
||||
real: boolean
|
||||
/** 几个过了补发宽限的去抖窗口合成一条跳过时才有：合了几个窗口、合掉的期次（上传表格）；老服务端不给 */
|
||||
windows?: number | null
|
||||
periods?: string[] | null
|
||||
}
|
||||
|
||||
export interface ScheduleFire {
|
||||
|
||||
@@ -326,12 +326,20 @@ await section('纯函数：事件触发的表单 ⇄ 请求体', async () => {
|
||||
body.trigger_kind === 'event' && JSON.stringify(body.trigger) === JSON.stringify({ event: 'table.version', debounce_s: 300,
|
||||
source_id: 'ds-sales', period: 'new' }) && body.cron === '' && body.name === '门店月报'
|
||||
&& JSON.stringify(body.input_template) === '{"date":"{{ event.period }}"}', JSON.stringify(body))
|
||||
const wf = m.triggerBody({ event: 'workflow.finished', subjectId: 'wf-daily', period: 'new', appId: 'app-board', originKind: '', debounceS: 0 })
|
||||
check('上游工作流：写 workflow_id，带应用、不带期次和空的来源', JSON.stringify(wf) === JSON.stringify({ event: 'workflow.finished',
|
||||
debounce_s: 0, workflow_id: 'wf-daily', app_id: 'app-board' }), JSON.stringify(wf))
|
||||
const switched = m.switchEvent(ev.trigger, 'source.changed')
|
||||
const wfForm = { event: 'workflow.finished', subjectId: 'wf-daily', period: 'new', appId: 'app-board', originKind: '',
|
||||
includeRejected: false, includeBackfill: false, debounceS: 0 }
|
||||
const wf = m.triggerBody(wfForm)
|
||||
check('上游工作流：写 workflow_id，带应用、不带期次和空的来源，缺省不带「也认」两项', JSON.stringify(wf) === JSON.stringify({
|
||||
event: 'workflow.finished', debounce_s: 0, workflow_id: 'wf-daily', app_id: 'app-board' }), JSON.stringify(wf))
|
||||
const wide = m.triggerBody({ ...wfForm, includeRejected: true, includeBackfill: true })
|
||||
check('勾了也认途经驳回的、应用补算的：各交 true', wide.include_rejected === true && wide.include_backfill === true, JSON.stringify(wide))
|
||||
const onlyBackfill = m.triggerBody({ ...wfForm, originKind: 'app_backfill', includeBackfill: true })
|
||||
check('只认应用补算的：不再另交「也认应用补算的」', onlyBackfill.origin_kind === 'app_backfill' && !('include_backfill' in onlyBackfill),
|
||||
JSON.stringify(onlyBackfill))
|
||||
check('表格的触发条件不带这两项', !('include_rejected' in m.triggerBody({ ...wfForm, event: 'table.version', includeRejected: true })))
|
||||
const switched = m.switchEvent({ ...ev.trigger, includeRejected: true }, 'source.changed')
|
||||
check('换事件种类：对象、过滤清掉，去抖留着', switched.subjectId === '' && switched.period === 'any' && switched.appId === ''
|
||||
&& switched.originKind === '' && switched.debounceS === 300 && m.switchEvent(ev.trigger, 'table.version') === ev.trigger)
|
||||
&& switched.originKind === '' && switched.includeRejected === false && switched.debounceS === 300 && m.switchEvent(ev.trigger, 'table.version') === ev.trigger)
|
||||
check('问题：事件触发没选对象时说要选什么，不查频率', m.formProblems({ ...ev, trigger: switched, freq: { ...m.DEFAULT_FREQ, preset: 'weekly',
|
||||
weekdays: [] } }, null).trigger === '请选择数据源' && !m.formProblems({ ...ev, freq: { ...m.DEFAULT_FREQ, preset: 'weekly', weekdays: [] } },
|
||||
null).freq)
|
||||
@@ -625,7 +633,8 @@ const TRIGGER_OPTIONS = {
|
||||
{ id: 'ds-shop', name: '门店库', ok: true, reason: null },
|
||||
] },
|
||||
],
|
||||
origins: [{ kind: 'launcher', label: '手动正式运行' }, { kind: 'schedule', label: '定时任务' }, { kind: 'hook', label: '入站钩子' }],
|
||||
origins: [{ kind: 'launcher', label: '手动正式运行' }, { kind: 'schedule', label: '定时任务' },
|
||||
{ kind: 'app_backfill', label: '应用补算' }, { kind: 'hook', label: '入站钩子' }],
|
||||
apps: [{ id: 'app-board', name: '门店看板' }],
|
||||
default_debounce_s: 60, max_debounce_s: 86400, max_depth: 5,
|
||||
}
|
||||
@@ -672,6 +681,10 @@ function eventFires() {
|
||||
event: { ...EVENT_BRIEF, merged: 0, sample: true, real: false } }),
|
||||
fire('fe3', { ...none, kind: 'event', status: 'skipped_overlap', fire_at: iso(NOW - 48 * H), created_at: iso(NOW - 48 * H),
|
||||
reason: '上一次运行 #a1b2c3 仍在进行中，本次触发已跳过', event: { ...EVENT_BRIEF, merged: 1, depth: 2 } }),
|
||||
// 调度停过：三期的事件都过了补发宽限，合成一条跳过（几个窗口、合掉的期次照实写）
|
||||
fire('fe4', { ...none, kind: 'event', status: 'skipped_misfire', fire_at: iso(NOW - 72 * H), created_at: iso(NOW - 70 * H),
|
||||
reason: '2026-10-14 09:00 至 2026-10-14 09:20 的 3 次事件（3 个去抖窗口）：服务当时没有派发（调度未运行或已停止），已超出补发宽限（1 小时），没有发起；积压的事件不补',
|
||||
event: { ...EVENT_BRIEF, merged: 3, windows: 3, periods: ['2026-07', '2026-08', '2026-09'] } }),
|
||||
]
|
||||
}
|
||||
/** 由事件触发的那次运行：来源里记着 trigger（服务端发起时写下的那一句） */
|
||||
@@ -2336,6 +2349,32 @@ await section('事件触发：新建的请求体', async () => {
|
||||
check('没有页面报错、原生对话框', !errors.length, errors.join(' ｜ '))
|
||||
})
|
||||
|
||||
await section('事件触发：上游工作流的请求体（途经驳回、应用补算）', async () => {
|
||||
const { page, sent, errors } = await open('/schedules/new')
|
||||
await page.locator('[data-schedule-form]').waitFor()
|
||||
await page.locator('[data-workflow-picker] input').click()
|
||||
await page.locator('[data-workflow-option="wf-weekly"]').click()
|
||||
await page.locator('[data-trigger-mode="event"]').click()
|
||||
await page.locator('[data-trigger-event="workflow.finished"]').click()
|
||||
await page.locator('[data-trigger-subject="workflow.finished"]').selectOption('wf-daily')
|
||||
const rej = page.locator('[data-trigger-include-rejected]')
|
||||
const back = page.locator('[data-trigger-include-backfill]')
|
||||
check('上游工作流：有「也认途经驳回的」「也认应用补算的」，缺省都不勾', await rej.count() === 1 && await back.count() === 1
|
||||
&& !(await rej.isChecked()) && !(await back.isChecked()))
|
||||
check('……说明照实写不勾时这类运行不触发', (await text(page, '[data-trigger-section]')).includes('不勾时这类运行不触发'))
|
||||
await page.locator('[data-trigger-origin]').selectOption('app_backfill')
|
||||
check('……只认应用补算时「也认应用补算的」不再出现', await back.count() === 0)
|
||||
await page.locator('[data-trigger-origin]').selectOption('')
|
||||
await rej.check()
|
||||
await page.locator('[data-input-field="date"] input').fill('{{ event.rejected }}')
|
||||
await page.locator('[data-schedule-save]').click()
|
||||
await until(() => writes(sent, /^POST \/schedules$/).length === 1)
|
||||
const body = writes(sent, /^POST \/schedules$/)[0]?.body
|
||||
check('POST 请求体：勾了的「也认途经驳回的」交 true，没勾的不交', JSON.stringify(body?.trigger) === JSON.stringify({
|
||||
event: 'workflow.finished', debounce_s: 60, workflow_id: 'wf-daily', include_rejected: true }), JSON.stringify(body?.trigger))
|
||||
check('没有页面报错', !errors.length, errors.join(' ｜ '))
|
||||
})
|
||||
|
||||
await section('事件触发：列表、详情与触发历史', async () => {
|
||||
const state = freshState({ schedules: [structuredClone(S1), structuredClone(S_EVENT), structuredClone(S_EVENT_OFF)], fires: eventFires() })
|
||||
const list = await open('/schedules', { state })
|
||||
@@ -2375,6 +2414,9 @@ await section('事件触发：列表、详情与触发历史', async () => {
|
||||
&& fe2.includes('立即运行，事件为示例值') && !fe2.includes('由事件触发')
|
||||
&& (await text(page, '[data-fire-id="fe2"]')).includes('还没有发生过这种事件'), fe2)
|
||||
check('……上一次没结束跳过的写第几层', (await text(page, '[data-fire-id="fe3"] [data-fire-event]')).includes('触发链第 2 层'))
|
||||
const fe4 = await text(page, '[data-fire-id="fe4"] [data-fire-event]')
|
||||
check('……过了宽限合成一条跳过的：写合并了几次、几个窗口、合掉的期次', fe4.includes('合并了 3 次事件') && fe4.includes('3 个去抖窗口')
|
||||
&& fe4.includes('期次 2026-07、2026-08、2026-09'), fe4)
|
||||
await shot(page, 'event-history-dark')
|
||||
// 390px 窄屏、浅色：事件那一节（三种事件的按钮、对象下拉里带原因的长名字）不撑出横向滚动
|
||||
const narrow = await open('/schedules/s-event', { state, width: 390, height: 844, theme: 'light' })
|
||||
|
||||
Reference in New Issue
Block a user