From ef6ce56cad1e2e156b50e6effe6c785a2517e5c2 Mon Sep 17 00:00:00 2001 From: Ben Barclay Date: Fri, 24 Jul 2026 13:37:04 +1000 Subject: [PATCH] fix(cron): reconcile external provider after a claimed direct run (#70479) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A direct run (cronjob action='run' / webhook-triggered manual fire) takes the job's fire_claim and advances next_run_at. When it races the external provider's scheduled fire for the same occurrence, Chronos loses the claim and — by design — does not re-arm (the winner owns the re-arm). But the direct-run winner never notified the provider, so the NAS one-shot for the consumed occurrence was left stale forever and the recurring job silently stopped firing. Observed in production: a managed 1-minute review job stalled for 20 hours because a GitHub-webhook direct run claimed the job 2s before the Chronos fire arrived; every subsequent occurrence was orphaned while /api/status stayed green. Fix: after a *claimed* direct execution completes (success or failure — next_run_at advances at claim time either way), call _notify_provider_jobs_changed_safe() so the active provider re-arms the post-run next_run_at. No-op for the built-in ticker; claim-lost direct runs still never notify (the winning scheduler owns the re-arm). --- tests/tools/test_cronjob_run_immediate.py | 44 ++++++++++++++++++++++- tools/cronjob_tools.py | 6 ++++ 2 files changed, 49 insertions(+), 1 deletion(-) diff --git a/tests/tools/test_cronjob_run_immediate.py b/tests/tools/test_cronjob_run_immediate.py index 9efa60e82cb1..6588222a4bdf 100644 --- a/tests/tools/test_cronjob_run_immediate.py +++ b/tests/tools/test_cronjob_run_immediate.py @@ -32,12 +32,53 @@ class TestCronjobRunExecutesImmediately: m_claim.assert_called_once_with("job-run-1") # at-most-once claim taken m_run.assert_called_once() # fired via the shared body + def test_run_reconciles_external_provider_after_claimed_execution(self): + """A direct run must re-arm Chronos after it advances next_run_at. + + Otherwise a scheduled Chronos fire that loses its claim to this direct + run is consumed without a successor one-shot, permanently stalling the + recurring job. + """ + order = [] + ran = {"id": "job-run-1", "last_status": "ok", "last_error": None} + with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \ + patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \ + patch("cron.scheduler.run_one_job", + side_effect=lambda *a, **kw: order.append("run") or True), \ + patch("tools.cronjob_tools.get_job", return_value=ran), \ + patch("tools.cronjob_tools._notify_provider_jobs_changed_safe", + side_effect=lambda: order.append("notify")) as m_notify: + out = json.loads(cronjob(action="run", job_id="job-run-1")) + + assert out["job"]["executed"] is True + m_notify.assert_called_once_with() + # Reconcile only AFTER the run persisted its final state (mark_job_run + # inside run_one_job), so the provider arms the post-run next_run_at. + assert order == ["run", "notify"] + + def test_run_reconciles_external_provider_even_when_claimed_run_fails(self): + """A claimed direct run advances next_run_at at claim time, so the + provider must be reconciled even when the execution itself fails.""" + failed = {"id": "job-run-1", "last_status": "error", "last_error": "provider 500"} + with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \ + patch("tools.cronjob_tools.claim_job_for_fire", return_value=True), \ + patch("cron.scheduler.run_one_job", side_effect=RuntimeError("boom")), \ + patch("tools.cronjob_tools.mark_job_run"), \ + patch("tools.cronjob_tools.get_job", return_value=failed), \ + patch("tools.cronjob_tools._notify_provider_jobs_changed_safe") as m_notify: + out = json.loads(cronjob(action="run", job_id="job-run-1")) + + assert out["job"]["executed"] is True + assert out["job"]["execution_success"] is False + m_notify.assert_called_once_with() + def test_run_skips_when_claim_lost(self): """If the scheduler already holds the fire claim, do NOT double-run.""" with patch("tools.cronjob_tools.resolve_job_ref", return_value=dict(_JOB)), \ patch("tools.cronjob_tools.claim_job_for_fire", return_value=False), \ patch("cron.scheduler.run_one_job") as m_run, \ - patch("tools.cronjob_tools.get_job", return_value=dict(_JOB)): + patch("tools.cronjob_tools.get_job", return_value=dict(_JOB)), \ + patch("tools.cronjob_tools._notify_provider_jobs_changed_safe") as m_notify: out = json.loads(cronjob(action="run", job_id="job-run-1")) assert out["success"] is True @@ -45,6 +86,7 @@ class TestCronjobRunExecutesImmediately: assert out["job"]["execution_success"] is False assert "execution_skipped" in out["job"] m_run.assert_not_called() # claim lost -> never fired + m_notify.assert_not_called() # the winning scheduler owns the re-arm def test_run_reports_failure_from_last_status(self): """A failed run is reported via the re-read job's last_status/last_error.""" diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index 430cd1dda401..a623d38c6f56 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -842,6 +842,12 @@ def cronjob( # _execute_job_now advances next_run_at and blocks a concurrent tick # from double-firing. exec_result = _execute_job_now(job) + # A claimed direct run advances next_run_at and may race the + # external one-shot for the same occurrence. If Chronos loses that + # claim, its consumed fire cannot re-arm itself; reconcile from the + # winning direct path after the run has persisted its final state. + if exec_result.get("claimed", False): + _notify_provider_jobs_changed_safe() # Re-read so the response reflects the post-run last_run_at/last_status. result = _format_job(get_job(job_id) or {"id": job_id}) result["executed"] = exec_result.get("claimed", False)