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() {