fix(cron): 修复固定间隔任务因 store 并发替换导致的重复执行
_on_timer 中 await _execute_job 让出控制权期间,前端轮询触发的 list_jobs 调用 _load_store 从磁盘重新加载覆盖 self._store, 已执行任务的状态被旧值回退,导致再次触发。 引入 _timer_active 标志位,在任务执行期间阻止并发 _load_store 替换 store。同时修复 store 为空时未重新 arm timer 的问题。 Made-with: Cursor
This commit is contained in:
@@ -80,6 +80,7 @@ class CronService:
|
|||||||
self._store: CronStore | None = None
|
self._store: CronStore | None = None
|
||||||
self._timer_task: asyncio.Task | None = None
|
self._timer_task: asyncio.Task | None = None
|
||||||
self._running = False
|
self._running = False
|
||||||
|
self._timer_active = False
|
||||||
self.max_sleep_ms = max_sleep_ms
|
self.max_sleep_ms = max_sleep_ms
|
||||||
|
|
||||||
def _load_jobs(self) -> tuple[list[CronJob], int]:
|
def _load_jobs(self) -> tuple[list[CronJob], int]:
|
||||||
@@ -171,7 +172,11 @@ class CronService:
|
|||||||
def _load_store(self) -> CronStore:
|
def _load_store(self) -> CronStore:
|
||||||
"""Load jobs from disk. Reloads automatically if file was modified externally.
|
"""Load jobs from disk. Reloads automatically if file was modified externally.
|
||||||
- Reload every time because it needs to merge operations on the jobs object from other instances.
|
- Reload every time because it needs to merge operations on the jobs object from other instances.
|
||||||
|
- During _on_timer execution, return the existing store to prevent concurrent
|
||||||
|
_load_store calls (e.g. from list_jobs polling) from replacing it mid-execution.
|
||||||
"""
|
"""
|
||||||
|
if self._timer_active and self._store:
|
||||||
|
return self._store
|
||||||
jobs, version = self._load_jobs()
|
jobs, version = self._load_jobs()
|
||||||
self._store = CronStore(version=version, jobs=jobs)
|
self._store = CronStore(version=version, jobs=jobs)
|
||||||
self._merge_action()
|
self._merge_action()
|
||||||
@@ -290,8 +295,11 @@ class CronService:
|
|||||||
"""Handle timer tick - run due jobs."""
|
"""Handle timer tick - run due jobs."""
|
||||||
self._load_store()
|
self._load_store()
|
||||||
if not self._store:
|
if not self._store:
|
||||||
|
self._arm_timer()
|
||||||
return
|
return
|
||||||
|
|
||||||
|
self._timer_active = True
|
||||||
|
try:
|
||||||
now = _now_ms()
|
now = _now_ms()
|
||||||
due_jobs = [
|
due_jobs = [
|
||||||
j for j in self._store.jobs
|
j for j in self._store.jobs
|
||||||
@@ -302,6 +310,8 @@ class CronService:
|
|||||||
await self._execute_job(job)
|
await self._execute_job(job)
|
||||||
|
|
||||||
self._save_store()
|
self._save_store()
|
||||||
|
finally:
|
||||||
|
self._timer_active = False
|
||||||
self._arm_timer()
|
self._arm_timer()
|
||||||
|
|
||||||
async def _execute_job(self, job: CronJob) -> None:
|
async def _execute_job(self, job: CronJob) -> None:
|
||||||
|
|||||||
Reference in New Issue
Block a user