定时任务遗留问题:失败通知不再许诺能继续运行、30 天前的触发停在审批上照样推送、调度循环测试不再依赖整分边界

- 失败通知的「处理」行改成中性的「请在 AgentLab 的运行记录中查看原因」:服务重启时还在排队、从没开始执行的运行
  没有断点,「修正后可继续运行」是错的;开始执行过的也可能续不了(执行第一个节点前就失败、系统升级后断点不兼容),
  能不能继续以记录页为准。没有开始时间的失败另写一行原因(没有可继续的进度),通用格式的 reason 字段同写;任务
  还在排期上时补「下一次触发会照常发起」,和挂起(之后的触发都跳过)说清区别
- 通知扫描的 30 天窗口加两种例外:停在审批、有待审批的运行不受限制(创建 30 天以后才停到审批上的以前收不到
  「等待审批」);运行记录 30 天内有过变化的照扫(等了一个多月的审批批完、跑完,以前收不到「运行结束」)。早就
  结束、多日没有变化的旧触发仍不扫
- test_loop_ticks_until_stopped 改成只看「发起了」和停下前后的条数差:建任务和第一拍之间跨过整分时会先记一条
  错过的跳过,以前按总条数断言偶发失败。加一组把下一次放在上一个整分的用例,必然走到跨整分这一支
This commit is contained in:
ThearchyHelios
2026-10-05 18:39:10 +08:00
parent 14dcbe19bb
commit d62f921ed4
4 changed files with 243 additions and 16 deletions
+31 -6
View File
@@ -9,8 +9,8 @@ Webhook 地址本身就是凭证(企业微信的 key、钉钉的 access_token
一个过滤器把推送期间的那几行换成主机名。
通知里只有任务名、状态、出具档位、缺口条数、运行链接、封存哈希(界面上叫清单哈希)前 12 位和起止时间,失败时
另写失败节点的名字。不放报告里的具体数字,也不放失败时的原始报错——报错里可能带着查询到的数据,Webhook 推去的群
不一定有权看。
另写失败节点的名字和去哪里查看(不许诺能继续运行)。不放报告里的具体数字,也不放失败时的原始报错——报错里可能带着
查询到的数据,Webhook 推去的群不一定有权看。
三个平台出错时都回 HTTP 200、错误码放在响应体里,所以送没送达要看响应体:按地址的主机认平台,不看选的格式——
格式选错了(飞书的地址选了通用格式)正是最需要认出来的情形。群机器人的地址只认自己平台的格式,保存时就拦下
@@ -52,7 +52,8 @@ EVENT_LABEL = {"finished": "运行结束", "failed": "运行失败", "awaiting_a
TIMEOUT_S = 10.0
#: 失败后再试 3 次,每次之前等这么久(秒)。4xx(429 除外)不重试:地址或参数错了,重试也一样
RETRY_DELAYS: tuple[float, ...] = (1.0, 2.0, 4.0)
#: 只看这么多天内的触发:更早的运行状态不会再变,不用每一拍都扫
#: 只看这么多天内的触发:更早结束的运行状态不会再变,不用每一拍都扫。例外见 notify_due:停在审批上、有待审批的不受
#: 这个限制,运行记录这些天内有过变化的也照扫——审批可以等上一个多月,批完接着执行、停到下一个审批点或跑完都要推
SCAN_DAYS = 30
#: 运行结束后等封存(清单哈希)最多等多久。runner 先写终态再封存,两步之间被这一拍撞上就等下一拍
SEAL_WAIT_S = 60
@@ -65,6 +66,14 @@ _STATUS_LABEL = {"succeeded": "已完成", "failed": "失败", "cancelled": "已
#: 挂起的运行要人去处理:不处理,这个任务之后的触发都按「上一次还没结束」跳过
SUSPENDED_REASON = "运行停在断点上、没有待审批,通常是服务重启所致"
SUSPENDED_ACTION = "请在 AgentLab 的「记录」中继续运行或放弃这次运行;它结束之前,这个定时任务之后的触发都会跳过"
#: 失败的运行怎么处理:不许诺「修正后可继续运行」。没开始执行的没有断点;开始执行过的也可能没有(执行第一个节点前就
#: 失败)或已不兼容(系统升级),能不能继续以记录页的提示为准
FAILED_ACTION = "请在 AgentLab 的运行记录中查看原因"
#: 从没开始执行就失败的(服务重启时还在排队,被改成失败、没有开始时间):没有可继续的进度。不照抄运行记录上的报错——
#: 通知一律不放原始报错,这一句是按「没有开始时间」认出来的,自己写
NOT_STARTED_REASON = "运行在开始执行前就已中止(通常是服务重启时仍在排队),没有可继续的进度"
#: 失败的运行不挡之后的触发(挂起的才挡):任务还在排期上时补这一句,免得人以为这个定时任务就此停了
NEXT_AS_USUAL = "下一次触发会照常发起"
# --------------------------------------------------------------------------
@@ -217,6 +226,7 @@ class Notice:
failed_to_start。没有的项为 None。
suspended:运行停在断点上、没有待审批(通常是服务重启挂起的),按 failed 事件推。error_node:失败节点的名字。
next_fire:任务启用着、往后还会触发(失败通知据此补「下一次触发会照常发起」)。
fmt、signed 只给测试消息用:说明里只说这条消息证明得了的事。"""
event: str
@@ -238,9 +248,15 @@ class Notice:
pending_approvals: int | None = None
suspended: bool = False
error_node: str | None = None
next_fire: bool = False
fmt: str | None = None
signed: bool = False
@property
def never_started(self) -> bool:
"""失败的运行从没开始执行(没有开始时间):服务重启时还在排队、被改成失败的就是这种,没有断点可续。"""
return self.status == "failed" and self.run_id is not None and self.started_at is None
def _local(moment: datetime | None, tz: str, *, seconds: bool = True) -> str | None:
if moment is None:
@@ -282,7 +298,9 @@ def message(notice: Notice) -> tuple[str, list[tuple[str, str]], str | None]:
# 不写原始报错(可能带着查询到的数据),写失败节点的名字和去哪里处理
if notice.error_node:
lines.append(("失败节点", f"「{notice.error_node}」"))
lines.append(("处理", "请在 AgentLab 的运行记录中查看报错,修正后可继续运行"))
if notice.never_started:
lines.append(("原因", NOT_STARTED_REASON))
lines.append(("处理", f"{FAILED_ACTION};{NEXT_AS_USUAL}" if notice.next_fire else FAILED_ACTION))
if notice.fire_at is not None:
when = f"{_local(notice.fire_at, notice.timezone, seconds=False)}({notice.timezone})"
# 立即运行不写「计划触发」:群里的人会以为报告按时出了
@@ -375,7 +393,8 @@ def _generic(notice: Notice, now: datetime) -> dict[str, Any]:
"started_at": _iso(notice.started_at),
"finished_at": _iso(notice.finished_at),
"reason": (notice.reason if notice.status == "failed_to_start"
else f"{SUSPENDED_REASON}。{SUSPENDED_ACTION}" if notice.suspended else None),
else f"{SUSPENDED_REASON}。{SUSPENDED_ACTION}" if notice.suspended
else NOT_STARTED_REASON if notice.never_started else None),
"error_node": notice.error_node,
"sent_at": _iso(now),
}
@@ -566,13 +585,18 @@ async def notify_due(now: datetime | None = None) -> int:
"""
now = now or utcnow()
jobs: list[_Job] = []
cutoff = now - timedelta(days=SCAN_DAYS)
# 扫描窗口按触发记录的创建时间算,两种例外:运行记录窗口内有过变化(运行状态一变 updated_at 就跟着变),以及
# 停在审批上、有待审批的——以前只看触发记录,创建 30 天以后才停到审批上的运行收不到「等待审批」
waiting = select(Approval.id).where(Approval.run_id == Run.id, Approval.status == "pending").exists()
# 和发起共用一把锁:发起时先记一条「未能发起」再发起运行、成功了才改成 started,扫描不能撞在这中间
async with service.launch_lock():
async with SessionLocal() as session:
rows = (await session.execute(
select(ScheduleFire, Run)
.outerjoin(Run, Run.id == ScheduleFire.run_id)
.where(ScheduleFire.created_at >= now - timedelta(days=SCAN_DAYS), or_(
.where(or_(ScheduleFire.created_at >= cutoff, Run.updated_at >= cutoff,
and_(Run.status == "interrupted", waiting)), or_(
and_(ScheduleFire.status == "failed_to_start", ScheduleFire.notified_status.is_(None)),
and_(ScheduleFire.status == "started", Run.status.in_(_WATCHED),
or_(ScheduleFire.notified_status.is_(None), ScheduleFire.notified_status != Run.status)),
@@ -664,6 +688,7 @@ async def _notice(session: Any, schedule: Schedule, fire: ScheduleFire, run: Run
pending_approvals=pending or None,
suspended=run is not None and run.status == "interrupted" and not pending,
error_node=_failed_node(run) if run is not None and run.status == "failed" else None,
next_fire=bool(schedule.enabled and schedule.next_run_at is not None),
)
+22 -9
View File
@@ -369,9 +369,20 @@ async def test_start_does_nothing_when_switched_off():
await scheduler.stop()
async def test_loop_ticks_until_stopped():
# 下一次放在整分上(和 cron 对齐):放在「现在 - 1 秒」的话,这一秒里恰好跨过整分时会多记一条错过的跳过
async def fire_count(schedule_id: str) -> int:
async with SessionLocal() as session:
return await session.scalar(select(func.count()).select_from(ScheduleFire)
.where(ScheduleFire.schedule_id == schedule_id))
@pytest.mark.parametrize("crossed", [False, True], ids=["on_the_minute", "minute_crossed"])
async def test_loop_ticks_until_stopped(crossed):
"""循环按拍执行、stop 之后不再有拍。只看「发起了」和停下前后的条数差,不看总条数:建任务和第一拍之间恰好跨过
整分时,第一拍会先记一条错过的跳过、再发起最近这一次——以前按总条数断言,跨整分时偶发失败。crossed 把下一次
放在上一个整分,必然走到这一种"""
due = cron.last_due("* * * * *", "UTC", datetime.now(UTC))
if crossed:
due -= timedelta(minutes=1)
sid = await schedule(await workflow(), next_run_at=due, cron="* * * * *")
on = Settings(_env_file=None, scheduler=True)
assert await scheduler.start(interval=0.05, config=on) is True
@@ -382,18 +393,20 @@ async def test_loop_ticks_until_stopped():
await asyncio.sleep(0.02)
await scheduler.stop()
assert scheduler.running() is False
# 只认发起的那一条:建任务和第一拍之间跨过整分时,前面还会有一条错过的跳过,那不是这里要测的
[fire] = [f for f in await fires(sid) if f.status == "started"]
await settle(fire.run_id)
# 停下之后不再有拍:把下一次拨回过去也不会再发起
started = [f for f in await fires(sid) if f.status == "started"]
assert started
if crossed:
assert [f.status for f in await fires(sid)][0] == "skipped_misfire"
for fire in started:
await settle(fire.run_id)
# 停下之后不再有拍:把下一次拨回过去,过一会儿条数一条不多
before = await fire_count(sid)
async with SessionLocal() as session:
await session.execute(Schedule.__table__.update().where(Schedule.id == sid)
.values(next_run_at=datetime.now(UTC) - timedelta(seconds=1)))
await session.commit()
await asyncio.sleep(0.2)
async with SessionLocal() as session:
assert await session.scalar(select(func.count()).select_from(ScheduleFire)
.where(ScheduleFire.schedule_id == sid)) == 1
assert await fire_count(sid) == before
async def test_start_twice_keeps_one_loop():
@@ -0,0 +1,188 @@
"""定时任务通知(app/scheduler/webhook.py)的两条规则:
- 失败通知的「处理」行不许诺能继续运行:服务重启时还在排队、从没开始执行的运行(被改成失败、没有开始时间)没有
断点,「修正后可继续运行」是错的。按有没有开始执行分开写——没开始的另写一行原因(没有可继续的进度),处理一律是
中性的「请在运行记录中查看原因」;任务还在排期上时补一句「下一次触发会照常发起」:失败的运行不挡之后的触发,
和挂起的那种(之后的触发都会跳过)说清区别
- 扫描窗口(SCAN_DAYS)之外的触发:停在审批、有待审批的运行不受窗口限制——创建 30 天以后才停到审批上的照样推送
「等待审批」;运行记录在窗口内有变化的(等了一个多月的审批批完、跑完)照样推送结束。早就结束、记录多日没有变化
的旧触发仍不扫
用 httpx.MockTransport 接住请求,不连外网。任务名、地址、数字都是虚构的。
"""
from __future__ import annotations
import json
from datetime import datetime, timedelta, timezone
import httpx
import pytest
from sqlalchemy import delete
from app.core.crypto import encrypt
from app.db.base import SessionLocal
from app.db.models import Approval, Run, RunEvent, Schedule, ScheduleFire, Setting
from app.scheduler import webhook
UTC = timezone.utc
GENERIC = "https://hooks.example.com/agentlab/inbox-3e9d"
NOT_STARTED = "运行在开始执行前就已中止(通常是服务重启时仍在排队),没有可继续的进度"
class Recorder:
def __init__(self) -> None:
self.requests: list[httpx.Request] = []
def __call__(self, request: httpx.Request) -> httpx.Response:
self.requests.append(request)
return httpx.Response(200, json={})
def bodies(self) -> list[dict]:
return [json.loads(r.content) for r in self.requests]
@pytest.fixture
def recorder(monkeypatch):
rec = Recorder()
monkeypatch.setattr(webhook, "_transport", httpx.MockTransport(rec))
monkeypatch.setattr(webhook, "RETRY_DELAYS", (0, 0, 0))
return rec
@pytest.fixture(autouse=True)
async def clean():
async with SessionLocal() as session:
await session.execute(delete(Schedule))
await session.execute(delete(Setting).where(Setting.key == "notify"))
await session.commit()
yield
def notice(**over) -> webhook.Notice:
base = dict(event="failed", schedule_id="s1", schedule_name="仓库日报", timezone="Asia/Shanghai",
fire_id="f1", fire_at=datetime(2026, 10, 5, 1, 0, tzinfo=UTC), fire_kind="scheduled",
run_id="r9876543210fedcba", status="failed", run_url=None, manifest_hash=None,
started_at=datetime(2026, 10, 5, 1, 0, 2, tzinfo=UTC), finished_at=None)
base.update(over)
return webhook.Notice(**base)
async def make(*, run_status: str, started: bool = True, fire_age: timedelta = timedelta(0),
run_age: timedelta = timedelta(0), enabled: bool = True, sealed: bool = True) -> tuple[str, str, str]:
"""一个配了通用 Webhook 的任务、一次触发和它发起的运行(不真的执行,状态直接写库)。fire_age、run_age 是触发
记录、运行记录最后一次更新离现在多久;started=False 是从没开始执行的(服务重启时还在排队)。"""
now = datetime.now(UTC)
async with SessionLocal() as session:
sched = Schedule(name="仓库日报", workflow_id="wf-demo", cron="0 9 * * *", timezone="Asia/Shanghai",
enabled=enabled, next_run_at=now + timedelta(hours=3) if enabled else None,
webhook_format="generic", webhook_url=encrypt(GENERIC),
webhook_events=["finished", "failed", "awaiting_approval"])
session.add(sched)
await session.flush()
ended = run_status in ("succeeded", "failed", "cancelled") and started
touched = now - run_age
run = Run(workflow_id=None, workflow_name="仓库日报", status=run_status, schedule_id=sched.id,
run_class="formal", started_at=touched - timedelta(minutes=1) if started else None,
finished_at=touched if ended else None, manifest_hash=("d00d" * 16) if sealed and ended else None,
created_at=now - fire_age, updated_at=touched,
error="服务重启时这次运行仍在排队,没有可恢复的进度,请重新发起运行" if not started else None)
session.add(run)
await session.flush()
fire = ScheduleFire(schedule_id=sched.id, fire_at=(now - fire_age).replace(microsecond=0), kind="scheduled",
status="started", run_id=run.id, created_at=now - fire_age)
session.add(fire)
await session.commit()
return sched.id, fire.id, run.id
async def notified(fire_id: str) -> str | None:
async with SessionLocal() as session:
return (await session.get(ScheduleFire, fire_id)).notified_status
# --------------------------------------------------------------------------
# 失败通知的「处理」行
# --------------------------------------------------------------------------
def test_a_run_that_never_started_is_not_promised_a_continue():
"""服务重启时还在排队的运行没有断点:不写「修正后可继续运行」,另写一行原因说没有可继续的进度"""
_, lines, _ = webhook.message(notice(started_at=None, next_fire=True))
shown = dict(lines)
assert shown["原因"] == NOT_STARTED
assert shown["处理"] == "请在 AgentLab 的运行记录中查看原因;下一次触发会照常发起"
assert not any("继续运行" in value for _, value in lines), lines
def test_a_failed_run_gets_a_neutral_action_without_promising_a_continue():
"""开始执行过的运行也不许诺能继续:断点可能没有(执行第一个节点前就失败)、可能已不兼容(系统升级),能不能继续
以记录页为准。失败节点照写,不另写原因(原始报错可能带着查询到的数据)"""
_, lines, _ = webhook.message(notice(error_node="汇总库存", next_fire=True))
shown = dict(lines)
assert shown["失败节点"] == "「汇总库存」" and "原因" not in shown
assert shown["处理"] == "请在 AgentLab 的运行记录中查看原因;下一次触发会照常发起"
assert "修正后可继续运行" not in webhook._text(notice(error_node="汇总库存"))
def test_next_fire_is_only_promised_while_the_schedule_is_on():
"""任务已暂停(或往后不再触发)时不说「下一次触发会照常发起」"""
_, lines, _ = webhook.message(notice(next_fire=False))
assert dict(lines)["处理"] == "请在 AgentLab 的运行记录中查看原因"
async def test_restart_failure_notice_end_to_end(recorder):
"""服务重启时还在排队、被改成失败的运行:不会封存,等过封存时限后推一次失败,原因、处理按上面写;通用格式的
reason 字段也写这句原因。任务停用时不补「下一次触发会照常发起」"""
_, on_fire, _ = await make(run_status="failed", started=False, sealed=False)
_, off_fire, _ = await make(run_status="failed", started=False, sealed=False, enabled=False)
later = datetime.now(UTC) + timedelta(seconds=webhook.SEAL_WAIT_S + 5)
assert await webhook.notify_due(later) == 2
by_fire = {b["fire_id"]: b for b in recorder.bodies()}
on, off = by_fire[on_fire], by_fire[off_fire]
assert (on["event"], on["status"], on["reason"]) == ("failed", "failed", NOT_STARTED)
assert "处理:请在 AgentLab 的运行记录中查看原因;下一次触发会照常发起" in on["text"]
assert "处理:请在 AgentLab 的运行记录中查看原因\n" in off["text"] + "\n" and "照常发起" not in off["text"]
for body in (on, off):
assert "继续运行" not in body["text"], body["text"]
# 运行记录上的原始报错不进通知
assert "请重新发起运行" not in body["text"]
# --------------------------------------------------------------------------
# 扫描窗口之外
# --------------------------------------------------------------------------
async def test_old_fire_stopped_at_an_approval_is_still_pushed(recorder):
"""触发是 40 天前的,运行刚停到审批上(或者停下时服务没在推送):有待审批的运行不受 30 天的窗口限制。
运行记录本身也很久没有变化,照样推「等待审批」"""
_, fire_id, run_id = await make(run_status="interrupted", fire_age=timedelta(days=40), run_age=timedelta(days=35))
async with SessionLocal() as session:
session.add(RunEvent(run_id=run_id, seq=1, type="run.started", ts=1.0, data={}))
session.add(Approval(run_id=run_id, node_id="gate", title="审核日报", status="pending"))
await session.commit()
assert await webhook.notify_due() == 1
[body] = recorder.bodies()
assert (body["event"], body["pending_approvals"]) == ("awaiting_approval", 1)
assert await notified(fire_id) == "interrupted:approval:1"
assert await webhook.notify_due() == 0
async def test_old_fire_whose_run_just_finished_is_pushed(recorder):
"""等了一个多月的审批批完、刚跑完:运行记录在窗口内有变化,推「运行结束」"""
_, fire_id, _ = await make(run_status="succeeded", fire_age=timedelta(days=45))
assert await webhook.notify_due() == 1
assert recorder.bodies()[0]["event"] == "finished" and await notified(fire_id) == "succeeded"
async def test_old_fire_whose_run_ended_long_ago_is_not_scanned(recorder):
"""早就结束、多日没有变化的旧触发仍不扫:窗口本来就是为了不每一拍把全部历史翻一遍"""
_, fire_id, _ = await make(run_status="succeeded", fire_age=timedelta(days=45), run_age=timedelta(days=40))
assert await webhook.notify_due() == 0
assert recorder.requests == [] and await notified(fire_id) is None
async def test_old_suspended_run_without_approval_is_not_rescanned_forever(recorder):
"""挂起(没有待审批)、多日没有变化的旧触发不在例外里:例外只给停在审批上的"""
_, fire_id, _ = await make(run_status="interrupted", fire_age=timedelta(days=40), run_age=timedelta(days=40))
assert await webhook.notify_due() == 0 and await notified(fire_id) is None
+2 -1
View File
@@ -178,7 +178,8 @@ def test_failed_run_text_points_to_the_node_without_repeating_itself():
error_node="查询门店销售"))
shown = dict(lines)
assert title.endswith("运行失败") and shown["状态"] == "失败"
assert shown["失败节点"] == "「查询门店销售」" and "继续运行" in shown["处理"]
# 处理只说去哪里看,不许诺能继续运行(没有断点的运行续不了,见 test_scheduler_notice_rules.py)
assert shown["失败节点"] == "「查询门店销售」" and shown["处理"] == "请在 AgentLab 的运行记录中查看原因"
# 不再写「原因:运行失败,详情请查看运行记录」这种同义反复
assert "原因" not in shown