fix(cron): prevent double-execution of one-shot jobs across concurrent schedulers

When two scheduler processes (gateway + desktop) run concurrently,
both could pick up the same one-shot job from get_due_jobs() because
its next_run_at was not advanced before execution started — only
recurring jobs were advanced (L3446).  This caused duplicate deliveries
and wasted token spend (#59229).

Now _get_due_jobs_locked advances a one-shot's next_run_at by 60s
before returning it as due, persisted immediately under the same
file lock.  mark_job_run re-anchors next_run_at on completion, so a
tick death between advance and execution only delays the job by one
tick window — it is never lost.

Closes #59229
This commit is contained in:
isheng 2026-07-06 15:02:37 +08:00 committed by Teknium
parent f3af7930c2
commit 06cc983b86
2 changed files with 20 additions and 1 deletions

View file

@ -1703,6 +1703,23 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
break
continue
# Before returning a one-shot job as due, advance its
# next_run_at past the next scheduler tick so other processes
# (gateway + desktop running concurrent schedulers) don't
# double-execute it (#59229). The advance is persisted
# immediately under the same lock that get_due_jobs holds.
# mark_job_run re-anchors next_run_at on completion (success
# or failure), so a tick death between here and execution
# only delays the job by a single tick window — not lost.
if kind == "once":
claimed_next = (now + timedelta(seconds=60)).isoformat()
job["next_run_at"] = claimed_next
for rj in raw_jobs:
if rj["id"] == job["id"]:
rj["next_run_at"] = claimed_next
needs_save = True
break
due.append(job)
if needs_save:

View file

@ -927,7 +927,9 @@ class TestGetDueJobs:
due = get_due_jobs()
assert [job["id"] for job in due] == ["oneshot-recover"]
assert get_job("oneshot-recover")["next_run_at"] == run_at
# next_run_at is now advanced past the tick window to prevent
# cross-process double-execution (#59229).
assert get_job("oneshot-recover")["next_run_at"] == (now + timedelta(seconds=60)).isoformat()
def test_broken_stale_one_shot_without_next_run_is_not_recovered(self, tmp_cron_dir, monkeypatch):
now = datetime(2026, 3, 18, 4, 30, 0, tzinfo=timezone.utc)