From b3581e02df1951eeafdcc5c6fb874a411f33697e Mon Sep 17 00:00:00 2001 From: jxxghp Date: Fri, 9 Oct 2026 21:39:33 +0800 Subject: [PATCH] fix(scheduler): preserve dispatched plugin one-shot jobs (#6935) --- app/scheduler/execution.py | 7 +++- app/scheduler/oncejob.py | 7 ++-- tests/test_plugin_once_job.py | 63 ++++++++++++++++++++++++++++++++++- 3 files changed, 73 insertions(+), 4 deletions(-) diff --git a/app/scheduler/execution.py b/app/scheduler/execution.py index 002a609ff..c659a40e0 100644 --- a/app/scheduler/execution.py +++ b/app/scheduler/execution.py @@ -51,7 +51,10 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase): def _prepare_job(self, job_id: str) -> Optional[dict[str, Any]]: """ - 准备定时任务 + 在锁内检查任务准入并原子取得本次执行所有权。 + + APScheduler 到期后会先从 job store 删除 DateTrigger，再进入本方法；插件一次性任务 + 在这里通过准入后即可退出“已派发、待开始”状态，避免配置保存时的服务重建误删它。 """ started_at = self._format_time() with self._lock: @@ -83,6 +86,8 @@ class SchedulerExecutionOwner(_SchedulerOwnerBase): if not self._registry.claim_generation(job_id, generation): JobExecutionState.finish(job, started_at, None) return None + if job.get("once"): + job["_once_pending"] = False job["_metric_started_at"] = time.perf_counter() progress = ProgressHelper(self._get_progress_key(job_id)) progress.start() diff --git a/app/scheduler/oncejob.py b/app/scheduler/oncejob.py index ea7a2c1fe..953d1d6f7 100644 --- a/app/scheduler/oncejob.py +++ b/app/scheduler/oncejob.py @@ -70,6 +70,8 @@ class SchedulerPluginOnceJobOwner(_SchedulerOwnerBase): # 插件未运行时读不到名称，用插件 ID 兜底，避免仪表盘和日志显示 None provider_name=plugin_manager.get_plugin_attr(pid, "plugin_name") or pid, once=True, + # APScheduler 先移除到期 DateTrigger，再调用 start；准入前仍属于待执行任务。 + _once_pending=True, # func 是插件传入的任意可调用对象，只有绑定方法才带 __self__； # 记录注册实例，供服务重建时丢弃已被替换实例的遗留任务。 _plugin_instance=getattr(func, "__self__", None), @@ -106,9 +108,10 @@ class SchedulerPluginOnceJobOwner(_SchedulerOwnerBase): pass def _is_plugin_once_job_pending(self, runtime_id: str, job: dict[str, Any]) -> bool: - """判断一次性任务是否仍待执行或正在执行；已执行完的状态可以回收。""" + """判断一次性任务是否待派发、待执行或正在执行；已完成的状态可以回收。""" return ( - self._scheduler.get_job(runtime_id) is not None + bool(job.get("_once_pending")) + or self._scheduler.get_job(runtime_id) is not None or self._is_job_active(runtime_id) or bool(job.get("running")) ) diff --git a/tests/test_plugin_once_job.py b/tests/test_plugin_once_job.py index 6928e3a69..a1b55bbfd 100644 --- a/tests/test_plugin_once_job.py +++ b/tests/test_plugin_once_job.py @@ -12,16 +12,19 @@ from types import SimpleNamespace import pytest from apscheduler.schedulers.background import BackgroundScheduler +import app.scheduler.execution as execution_module import app.scheduler.oncejob as oncejob_module import app.scheduler.reconcile as reconcile_module from app.runtime.log import current_plugin_instance_id +from app.scheduler.execution import SchedulerExecutionOwner from app.scheduler.oncejob import SchedulerPluginOnceJobOwner from app.scheduler.reconcile import SchedulerReconcileOwner +from app.scheduler.registry import ExecutionRegistry PID = "DemoOncePlugin" -class _Owner(SchedulerReconcileOwner, SchedulerPluginOnceJobOwner): +class _Owner(SchedulerExecutionOwner, SchedulerReconcileOwner, SchedulerPluginOnceJobOwner): """按 Scheduler Facade 的组合方式拼装对账与一次性任务两个 owner。""" @@ -68,10 +71,16 @@ def owner(monkeypatch): owner._assign_job_generation = lambda _job_id, job: job.setdefault("_generation", 0) owner._accepting_submissions = lambda: True owner._is_job_active = lambda _job_id: False + owner._registry = ExecutionRegistry(owner._lock) + owner._lifecycle_state = "running" + owner._format_time = lambda: "now" + owner._get_progress_key = lambda job_id: job_id def _start(job_id: str) -> bool: # 真实 start 的准入、进度与收尾由既有调度测试覆盖，这里只执行已登记的函数。 job = owner._jobs[job_id] + if job.get("once"): + job["_once_pending"] = False job["func"](**job["kwargs"]) return True @@ -137,6 +146,58 @@ def test_update_plugin_job_keeps_once_jobs_registered_by_current_instance(owner) assert owner._scheduler.get_job(f"{PID}_periodic") is not None +def test_update_plugin_job_keeps_once_job_after_scheduler_dispatch(owner) -> None: + """APScheduler 已消费触发器但尚未进入执行准入时，服务重建仍保留一次性任务。""" + job_id = _once_id("dispatching") + assert owner.add_plugin_once_job( + PID, + "dispatching", + owner.plugin.run_once, + "正在派发的一次性任务", + delay_seconds=60, + ) + + # DateTrigger 到期后 APScheduler 会先从 job store 移除任务，再调用 start。 + owner._scheduler.remove_job(job_id) + owner.update_plugin_job(PID) + + assert job_id in owner._jobs + assert owner.start(job_id) + assert owner.plugin.done.wait(5) + assert len(owner.plugin.calls) == 1 + + +def test_prepare_job_clears_once_pending_state_after_admission(owner, monkeypatch) -> None: + """一次性任务进入执行准入后不再以待派发状态阻止后续回收。""" + + class _ProgressStub: + """隔离一次性任务准入测试的进度缓存。""" + + def __init__(self, _key: str) -> None: + """接收进度键但不访问缓存后端。""" + + def start(self) -> None: + """记录进度开始。""" + + def update(self, **_kwargs) -> None: + """忽略本用例不关心的进度快照。""" + + monkeypatch.setattr(execution_module, "ProgressHelper", _ProgressStub) + job_id = _once_id("admitted") + assert owner.add_plugin_once_job( + PID, + "admitted", + owner.plugin.run_once, + "已准入的一次性任务", + delay_seconds=60, + ) + + job = owner._prepare_job(job_id) + + assert job is owner._jobs[job_id] + assert job["_once_pending"] is False + + def test_update_plugin_job_drops_once_jobs_of_replaced_instance(owner) -> None: """插件重载换成新实例后，旧实例追加的任务不得再回调旧实例。""" owner.add_plugin_once_job(PID, "stale", owner.plugin.run_once, "旧实例任务", delay_seconds=60)