From df62898d22381ba2ca6a84ac109cfb7a9d7ed088 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Fri, 2 Oct 2026 21:57:41 +0800 Subject: [PATCH] =?UTF-8?q?fix(automation):=20=E8=B0=83=E5=BA=A6=E5=BF=83?= =?UTF-8?q?=E8=B7=B3=E4=B8=8E=E4=BD=9C=E4=B8=9A=E5=91=A8=E6=9C=9F=E8=A7=A3?= =?UTF-8?q?=E8=80=A6=EF=BC=8C=E4=BF=AE=E6=8E=89=E7=B3=BB=E7=BB=9F=E6=80=A7?= =?UTF-8?q?=E8=B7=B3=E8=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 线上现象：同一页面上「CK 自动巡检」的最近运行是 21:45:34，而「下次运行」显示 21:50:34 且已经过去了 —— 巡检明明只要 19 秒，却总是等在那里。 根因是把心跳周期直接当成了作业周期：TICK_SECONDS=300 既是「循环多久醒一次」， 又被写进 next_run_at = 作业结束 + 300s。两者同频会系统性跳轮： 心跳 t=0 ck_health 19 秒跑完 → next = 5:19 心跳 t=5:00 差 19 秒没到点 → 整轮空转，什么都不做 心跳 t=10:00 才真正跑到 于是作业实际是每 10 分钟一轮，而不是设计的 5 分钟；只要作业耗时不是刚好 0， 就会稳定地"差一点点"被跳过。两个作业共享同一个心跳时，还会互相把对方的到期 时刻推后。 改动： - 拆成两个常量：TICK_SECONDS=30（心跳，只做「有没有到点的作业」检查，几乎不 耗资源）、JOB_INTERVAL_SECONDS=300（作业周期，跑完一轮后隔多久再跑）。 - next_run_at 改用 JOB_INTERVAL_SECONDS。 - /api/automation/jobs 的 scheduler 同时暴露两个值；启动日志一并打印。 - 页面：「单账号重检间隔」加 tooltip 说明它不决定调度频率（它只控制同一行多久 重检/重试），避免和调度周期混淆。 回归测试（3 条）：心跳必须比作业周期密至少 4 倍；周期未到时连跳多次心跳都不跑； 把 next_run_at 拨到过去后下一次心跳必须跑到。 全量 304 通过。 --- tests/test_automation_ck_health.py | 83 +++++++++++++++++++- web/backend/services/automation/scheduler.py | 19 ++++- web/frontend/src/api/types.ts | 3 + web/frontend/src/pages/AutomationPage.tsx | 1 + 4 files changed, 99 insertions(+), 7 deletions(-) diff --git a/tests/test_automation_ck_health.py b/tests/test_automation_ck_health.py index f5dce36..2241a2c 100644 --- a/tests/test_automation_ck_health.py +++ b/tests/test_automation_ck_health.py @@ -298,11 +298,11 @@ class TestScheduler(_Base): self.session.query(AutomationJob).filter(AutomationJob.key == "probe").one() ) assert job.last_status == "success" - # 下一轮在一个 tick 之后（作业每个 tick 都跑），与 interval_seconds 无关 - from web.backend.services.automation.scheduler import TICK_SECONDS + # 下一轮在一个作业周期之后，与 interval_seconds（每行多久重检）无关 + from web.backend.services.automation.scheduler import JOB_INTERVAL_SECONDS assert _aware(job.next_run_at) - _aware(job.last_run_at) == timedelta( - seconds=TICK_SECONDS + seconds=JOB_INTERVAL_SECONDS ) run = self.session.query(AutomationRun).one() assert (run.scanned, run.valid, run.invalid) == (3, 2, 1) @@ -400,7 +400,13 @@ class TestAutomationApi(_Base): def test_list_jobs_exposes_scheduler_and_job_state(self): api = self._api() payload = api.list_jobs(db=self.session, current=self.admin) - assert payload["scheduler"]["tick_seconds"] == 300 + # 心跳（轮询间隔）与作业周期是两个概念，接口都要暴露出来： + # 心跳密、只做「有没有到点」的检查；作业周期决定同一个作业隔多久跑下一轮。 + assert ( + payload["scheduler"]["tick_seconds"] + < payload["scheduler"]["job_interval_seconds"] + ) + assert payload["scheduler"]["job_interval_seconds"] == 300 assert payload["scheduler"]["running"] is False keys = [item["key"] for item in payload["jobs"]] assert ck_health.JOB_KEY in keys @@ -615,3 +621,72 @@ class TestOrphanRunCleanup(_Base): assert stale.finished_at is not None assert stale.message == "服务重启中断" assert done.status == "success" # 已完成的不受影响 + + +class TestHeartbeatDecoupledFromInterval(_Base): + """回归：心跳频率不能直接当作业周期用。 + + 曾经 TICK_SECONDS 既是心跳又是作业周期（next_run_at = 结束 + 300s，心跳也是 + 300s）。两者同频会系统性跳轮：作业 19 秒跑完 → next 在 5 分钟后，而下一次 + 心跳正好在 5 分钟整，常常差几十秒没到点，于是整轮空转 —— 实际变成每 10 分钟 + 才跑一次。线上表现就是「检测只要 19 秒，却总是排队等很久」。 + """ + + def test_heartbeat_is_much_shorter_than_job_interval(self): + from web.backend.services.automation.scheduler import ( + JOB_INTERVAL_SECONDS, + TICK_SECONDS, + ) + + assert TICK_SECONDS < JOB_INTERVAL_SECONDS, ( + "心跳必须比作业周期密，否则会出现「差几十秒没到点→整轮空转」的跳轮" + ) + # 至少密 4 倍：作业耗时波动时不会积压成整轮跳过 + assert JOB_INTERVAL_SECONDS / TICK_SECONDS >= 4 + + def test_job_runs_again_once_its_own_interval_elapsed(self): + """心跳多跳几次后，到点的作业必须被跑到。""" + + job = self._register_probe() + run_due_jobs(self.session) + first = self.session.query(AutomationRun).count() + + # 周期未到：无论心跳跳多少次都不跑 + for _ in range(3): + run_due_jobs(self.session) + assert self.session.query(AutomationRun).count() == first + + # 把 next_run_at 拨到过去（等价于「周期已到」），下一次心跳就该跑 + job.next_run_at = datetime.now(UTC) - timedelta(seconds=1) + self.session.commit() + run_due_jobs(self.session) + assert self.session.query(AutomationRun).count() == first + 1 + + def _register_probe(self): + from web.backend.services.automation.scheduler import ( + JobSpec, + register_job, + reset_registry, + ) + + reset_registry() + register_job( + JobSpec( + key="probe", + title="探针", + interval_seconds=6 * 3600, + batch_size=10, + concurrency=1, + run=lambda db, job: {"scanned": 1, "valid": 1, "invalid": 0}, + ) + ) + from web.backend.services.automation import ensure_jobs + + ensure_jobs(self.session) + job = ( + self.session.query(AutomationJob).filter(AutomationJob.key == "probe").one() + ) + job.enabled = True + job.next_run_at = datetime.now(UTC) - timedelta(seconds=1) + self.session.commit() + return job diff --git a/web/backend/services/automation/scheduler.py b/web/backend/services/automation/scheduler.py index ca9326d..05174ea 100644 --- a/web/backend/services/automation/scheduler.py +++ b/web/backend/services/automation/scheduler.py @@ -35,7 +35,17 @@ from ...models import AutomationJob, AutomationRun # 分开的原因：全库 12,881 条的 CK 若想「6 小时内都被查一遍」，需要约 36 条/分钟的 # 吞吐；按 5 分钟一轮、单轮 200 条算 = 40 条/分钟，刚好够。若把作业周期也设成 6h， # 就变成 200 条/6h = 800 条/天，全库要 16 天才能过一轮，等于没巡检。 -TICK_SECONDS = 300 +# 心跳间隔：只用来「看一眼有没有到点的作业」，几乎不耗资源，所以设得比作业 +# 周期密得多。作业周期由各自的 next_run_at（= 自己结束时刻 + JOB_INTERVAL）决定。 +# +# 这里曾经把心跳周期直接当作业周期用（心跳 300s + next_run_at = 结束 + 300s）， +# 两者同频会导致系统性跳轮：作业 19 秒跑完 → next = +5 分钟，而下一次心跳正好 +# 在 5 分钟整，常常差几十秒没到点，于是整轮空转，实际变成每 10 分钟才跑一次。 +# 表现就是「检测明明只要 19 秒，却要排队等很久」。 +TICK_SECONDS = 30 + +# 作业自身的运行周期：跑完一轮后隔多久再跑下一轮。 +JOB_INTERVAL_SECONDS = 300 TRUTHY = {"1", "true", "yes", "on"} @@ -199,7 +209,7 @@ def _run_one(db: Session, job: AutomationJob, spec: JobSpec) -> dict[str, Any]: job.last_run_at = finished # 下一轮固定在一个 tick 之后；interval_seconds 是「同一行多久重检」， # 由作业自己在选行时用，不参与调度。 - job.next_run_at = finished + timedelta(seconds=TICK_SECONDS) + job.next_run_at = finished + timedelta(seconds=JOB_INTERVAL_SECONDS) job.last_status = status job.last_message = message job.updated_at = finished @@ -246,7 +256,9 @@ def _loop() -> None: finally: db.close() - logger.info("自动化调度器已启动，tick={}s", TICK_SECONDS) + logger.info( + "自动化调度器已启动：心跳 {}s，作业周期 {}s", TICK_SECONDS, JOB_INTERVAL_SECONDS + ) while not _stop_event.wait(TICK_SECONDS): db = SessionLocal() try: @@ -291,5 +303,6 @@ def scheduler_status() -> dict[str, Any]: "enabled": automation_enabled(), "running": _thread is not None and _thread.is_alive(), "tick_seconds": TICK_SECONDS, + "job_interval_seconds": JOB_INTERVAL_SECONDS, "registered": sorted(_REGISTRY), } diff --git a/web/frontend/src/api/types.ts b/web/frontend/src/api/types.ts index 8c6bb9d..8436654 100644 --- a/web/frontend/src/api/types.ts +++ b/web/frontend/src/api/types.ts @@ -877,7 +877,10 @@ export interface AutomationJobItem { export interface AutomationSchedulerStatus { enabled: boolean; running: boolean; + /** 调度心跳（轮询间隔），比作业周期密，只做「有没有到点」的检查。 */ tick_seconds: number; + /** 作业周期：同一个作业跑完一轮后隔多久再跑下一轮。 */ + job_interval_seconds: number; registered: string[]; } diff --git a/web/frontend/src/pages/AutomationPage.tsx b/web/frontend/src/pages/AutomationPage.tsx index 7224cb3..b5eba65 100644 --- a/web/frontend/src/pages/AutomationPage.tsx +++ b/web/frontend/src/pages/AutomationPage.tsx @@ -312,6 +312,7 @@ export default function AutomationPage() {