fix(scheduler): preserve dispatched plugin one-shot jobs (#6935)

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

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"))
         )

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)