From 06cc983b86516344f7a6bc486ae8e0838ebaab09 Mon Sep 17 00:00:00 2001 From: isheng Date: Mon, 6 Jul 2026 15:02:37 +0800 Subject: [PATCH] fix(cron): prevent double-execution of one-shot jobs across concurrent schedulers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- cron/jobs.py | 17 +++++++++++++++++ tests/cron/test_jobs.py | 4 +++- 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/cron/jobs.py b/cron/jobs.py index 62633ff2d73..22d1b945b54 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -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: diff --git a/tests/cron/test_jobs.py b/tests/cron/test_jobs.py index 67f3e772f18..19080adf3c4 100644 --- a/tests/cron/test_jobs.py +++ b/tests/cron/test_jobs.py @@ -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)