From 7ecc822e1165f5f4d274075a40066a8ab04214d0 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 8 Jul 2026 05:13:18 -0700 Subject: [PATCH] fix(cron): stop the ticker from stalling forever on a wedged jobs lock (#60703) (#60855) Three fixes for the silent post-restart ticker stall: 1. _jobs_lock() bounds its cross-process flock: LOCK_NB polled against a 30s deadline instead of an unbounded LOCK_EX taken while holding the process-wide RLock. On timeout it logs at ERROR and degrades to in-process-only locking (the existing fallback path), so a sibling process wedged while holding .jobs.lock can no longer freeze every cron function - including the ticker's get_due_jobs() and thus the heartbeat - forever with zero logging. 2. fire_claim/run_claim freshness checks are bounded on both sides (0 <= age < ttl): a claim stamped in the future (clock/TZ skew across a restart) was previously fresh forever, making the job permanently unfireable and every manual run report 'already being fired'. 3. _execute_job_now distinguishes paused/disabled/missing jobs from a genuinely held claim instead of mislabeling them all as 'already being fired'. --- cron/jobs.py | 61 ++++++++- tests/cron/test_ticker_stall_60703.py | 186 ++++++++++++++++++++++++++ tools/cronjob_tools.py | 16 ++- 3 files changed, 257 insertions(+), 6 deletions(-) create mode 100644 tests/cron/test_ticker_stall_60703.py diff --git a/cron/jobs.py b/cron/jobs.py index 7ba8fd5abec..76cfc60cbf5 100644 --- a/cron/jobs.py +++ b/cron/jobs.py @@ -84,6 +84,15 @@ TICKER_INTERVAL_SECONDS = 60 # concurrent mark_job_run / advance_next_run calls can clobber each other. _jobs_file_lock = threading.RLock() _jobs_lock_state = threading.local() + +# Upper bound on waiting for the cross-process .jobs.lock flock (#60703). +# Every cron function in the process funnels through _jobs_lock(), and the +# flock is taken while holding the process-wide RLock — so an unbounded wait +# on a lock held by a wedged sibling process silently freezes the ticker +# heartbeat and every job forever. 30s is orders of magnitude above any +# legitimate critical section (field updates only) while keeping the ticker's +# worst-case stall well under one status-alarm threshold. +_JOBS_LOCK_TIMEOUT_SECONDS = 30.0 OUTPUT_DIR = CRON_DIR / "output" ONESHOT_GRACE_SECONDS = 120 @@ -177,7 +186,42 @@ def _jobs_lock(): lock_fd = open(_jobs_lock_file(), "a+", encoding="utf-8") lock_fd.seek(0) if fcntl is not None: - fcntl.flock(lock_fd, fcntl.LOCK_EX) + # Bounded acquisition (#60703): a plain blocking + # fcntl.flock(LOCK_EX) here has NO timeout, and it is + # taken while holding the process-wide _jobs_file_lock + # RLock above. If another process wedges while holding + # .jobs.lock (e.g. an old gateway draining through a + # restart), a single blocked acquirer freezes EVERY cron + # function in this process — including the ticker's + # get_due_jobs() — silently and forever: the heartbeat + # file stops updating and all jobs stop firing with no + # error logged. Poll LOCK_NB against a deadline instead; + # on timeout, log loudly and fall through to the same + # in-process-only degraded mode used when locking is + # unavailable. A briefly-torn cross-process write is + # strictly better than a permanently dead scheduler. + _deadline = time.monotonic() + _JOBS_LOCK_TIMEOUT_SECONDS + while True: + try: + fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + break + except (OSError, IOError): + if time.monotonic() >= _deadline: + logger.error( + "Timed out after %.0fs waiting for the cron " + "jobs lock (%s) — another process is holding " + "it. Proceeding with in-process locking only " + "so the scheduler stays alive (#60703).", + _JOBS_LOCK_TIMEOUT_SECONDS, + _jobs_lock_file(), + ) + try: + lock_fd.close() + except OSError: + pass + lock_fd = None + break + time.sleep(0.1) elif msvcrt is not None: getattr(msvcrt, "locking")(lock_fd.fileno(), getattr(msvcrt, "LK_LOCK"), 1) except (OSError, IOError) as e: @@ -1565,7 +1609,14 @@ def claim_job_for_fire(job_id: str, *, claim_ttl_seconds: int = 300) -> bool: if existing: try: claimed_at = _ensure_aware(datetime.fromisoformat(existing["at"])) - if (now - claimed_at).total_seconds() < claim_ttl_seconds: + # Bounded on BOTH sides (#60703): a claim stamped in the + # future (clock/TZ skew across a restart, or a corrupted + # timestamp) would otherwise have a negative age and stay + # "fresh" forever — the job becomes permanently unfireable + # and every manual `cron run` reports "already being + # fired". Treat future-dated claims as stale/overwritable. + _age = (now - claimed_at).total_seconds() + if 0 <= _age < claim_ttl_seconds: return False # someone holds a fresh claim except Exception: pass # malformed claim → overwrite @@ -1626,7 +1677,11 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]: claimed_at = _ensure_aware( datetime.fromisoformat(existing_claim["at"]) ) - if (now - claimed_at).total_seconds() < _run_claim_ttl: + # 0 <= age: a future-dated claim (clock/TZ skew across a + # restart) must be treated as stale, not eternally fresh, + # or the one-shot is skipped forever (#60703). + _age = (now - claimed_at).total_seconds() + if 0 <= _age < _run_claim_ttl: continue # a fresh claim is held by an in-flight run except (KeyError, ValueError, TypeError): pass # malformed claim → fall through and (re)claim diff --git a/tests/cron/test_ticker_stall_60703.py b/tests/cron/test_ticker_stall_60703.py new file mode 100644 index 00000000000..b4d4cca9520 --- /dev/null +++ b/tests/cron/test_ticker_stall_60703.py @@ -0,0 +1,186 @@ +"""Regression tests for #60703 — cron ticker silently stalls after gateway restart. + +Three fixes under test: + +1. ``_jobs_lock()`` bounds its cross-process flock: when another process holds + ``.jobs.lock`` indefinitely, acquisition times out, logs at ERROR, and falls + through to in-process-only locking — instead of blocking the calling thread + (and, transitively, the cron ticker heartbeat) forever. + +2. Claim freshness checks are bounded on both sides (``0 <= age < ttl``): a + ``fire_claim``/``run_claim`` stamped in the FUTURE (clock/TZ skew across a + restart) is treated as stale/overwritable, not eternally fresh. + +3. ``_execute_job_now`` no longer mislabels paused/disabled/missing jobs as + "already being fired". +""" + +import json +import os +import threading +import time +from datetime import timedelta +from pathlib import Path + +import pytest + +import cron.jobs as jobs_mod +from cron.jobs import ( + _jobs_lock, + claim_job_for_fire, + create_job, + get_due_jobs, + get_job, + load_jobs, + save_jobs, +) + + +try: + import fcntl +except ImportError: # pragma: no cover - non-POSIX + fcntl = None + + +pytestmark = pytest.mark.skipif(fcntl is None, reason="flock semantics are POSIX-only") + + +def _hold_jobs_flock(path: Path, release: threading.Event, held: threading.Event): + """Hold an exclusive flock on *path* from a separate fd until released. + + flock locks are per-open-file-description, so a second open() in the SAME + process contends exactly like another process would. + """ + fd = open(path, "a+", encoding="utf-8") + try: + fcntl.flock(fd, fcntl.LOCK_EX) + held.set() + release.wait(timeout=30) + finally: + try: + fcntl.flock(fd, fcntl.LOCK_UN) + except OSError: + pass + fd.close() + + +class TestBoundedJobsLock: + def test_lock_acquisition_times_out_and_degrades(self, monkeypatch, caplog): + """A foreign holder of .jobs.lock must NOT block _jobs_lock forever.""" + jobs_mod.ensure_dirs() + lock_path = jobs_mod._jobs_lock_file() + lock_path.touch() + + monkeypatch.setattr(jobs_mod, "_JOBS_LOCK_TIMEOUT_SECONDS", 1.0) + + release = threading.Event() + held = threading.Event() + holder = threading.Thread( + target=_hold_jobs_flock, args=(lock_path, release, held), daemon=True + ) + holder.start() + assert held.wait(timeout=10), "test holder failed to take the flock" + + try: + start = time.monotonic() + entered = False + with caplog.at_level("ERROR", logger="cron.jobs"): + with _jobs_lock(): + entered = True + elapsed = time.monotonic() - start + + assert entered, "critical section must still run in degraded mode" + assert elapsed < 10, f"lock wait was not bounded (took {elapsed:.1f}s)" + assert any("Timed out" in r.message for r in caplog.records), ( + "degraded-mode fallback must be logged at ERROR" + ) + finally: + release.set() + holder.join(timeout=10) + + def test_uncontended_lock_is_fast_and_silent(self, caplog): + jobs_mod.ensure_dirs() + start = time.monotonic() + with caplog.at_level("ERROR", logger="cron.jobs"): + with _jobs_lock(): + pass + assert time.monotonic() - start < 5 + assert not [r for r in caplog.records if "Timed out" in r.message] + + def test_reentrant_nesting_still_works(self): + with _jobs_lock(): + with _jobs_lock(): # must not deadlock or re-flock + pass + + +class TestFutureDatedClaims: + def _make_job(self, **kw): + return create_job(name="claim job", schedule="0 7 * * *", prompt="x", **kw) + + def test_future_fire_claim_is_treated_as_stale(self): + """A fire_claim stamped in the future must not block claiming forever.""" + job = self._make_job() + jobs = load_jobs() + for j in jobs: + if j["id"] == job["id"]: + future = jobs_mod._hermes_now() + timedelta(hours=6) + j["fire_claim"] = {"at": future.isoformat(), "by": "other-host:1"} + save_jobs(jobs) + + assert claim_job_for_fire(job["id"]) is True, ( + "future-dated claim must be overwritable, not eternally fresh" + ) + + def test_fresh_past_fire_claim_still_blocks(self): + job = self._make_job() + assert claim_job_for_fire(job["id"]) is True + # Immediately re-claiming must be refused — claim is genuinely fresh. + assert claim_job_for_fire(job["id"]) is False + + def test_expired_fire_claim_is_reclaimable(self): + job = self._make_job() + jobs = load_jobs() + for j in jobs: + if j["id"] == job["id"]: + past = jobs_mod._hermes_now() - timedelta(hours=6) + j["fire_claim"] = {"at": past.isoformat(), "by": "other-host:1"} + save_jobs(jobs) + assert claim_job_for_fire(job["id"]) is True + + def test_future_run_claim_does_not_skip_oneshot_forever(self): + """A one-shot with a future-dated run_claim must still become due.""" + past_fire = (jobs_mod._hermes_now() - timedelta(seconds=30)).isoformat() + job = create_job(name="oneshot", schedule=past_fire, prompt="x") + jobs = load_jobs() + for j in jobs: + if j["id"] == job["id"]: + future = jobs_mod._hermes_now() + timedelta(hours=6) + j["run_claim"] = {"at": future.isoformat(), "by": "other-host:1"} + j["next_run_at"] = past_fire + save_jobs(jobs) + + due_ids = {j["id"] for j in get_due_jobs()} + assert job["id"] in due_ids, ( + "future-dated run_claim must be treated as stale, not fresh" + ) + + +class TestHonestRunSkipMessages: + def test_paused_job_not_reported_as_already_firing(self): + from tools.cronjob_tools import _execute_job_now + + job = create_job(name="paused job", schedule="0 7 * * *", prompt="x") + from cron.jobs import pause_job + + pause_job(job["id"]) + res = _execute_job_now(get_job(job["id"])) + assert res["claimed"] is False + assert "paused" in (res["error"] or "").lower() + assert "already being fired" not in (res["error"] or "").lower() + + def test_missing_job_not_reported_as_already_firing(self): + from tools.cronjob_tools import _execute_job_now + + res = _execute_job_now({"id": "does-not-exist-123"}) + assert res["claimed"] is False + assert "no longer exists" in (res["error"] or "").lower() diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index 02ac58f9c60..430cd1dda40 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -623,8 +623,18 @@ def _execute_job_now(job: Dict[str, Any]) -> Dict[str, Any]: # At-most-once claim: bail without running if a tick/other fire owns it. if not claim_job_for_fire(job_id): - return {"claimed": False, "success": False, - "error": "Job is already being fired by the scheduler; not run again."} + # claim_job_for_fire returns False for paused/disabled/missing + # jobs too — don't mislabel those as "already being fired" + # (#60703): that message sends the user chasing a phantom + # in-flight run when the job simply isn't runnable. + refreshed = get_job(job_id) + if refreshed is None: + reason = "Job no longer exists; nothing to run." + elif not refreshed.get("enabled", True) or refreshed.get("state") == "paused": + reason = "Job is paused/disabled; resume it before running." + else: + reason = "Job is already being fired by the scheduler; not run again." + return {"claimed": False, "success": False, "error": reason} # run_one_job records last_run_at/last_status via mark_job_run (which # also clears the fire claim) and returns True iff it processed the job. @@ -837,7 +847,7 @@ def cronjob( result["executed"] = exec_result.get("claimed", False) result["execution_success"] = exec_result.get("success", False) if not exec_result.get("claimed", False): - result["execution_skipped"] = ( + result["execution_skipped"] = exec_result.get("error") or ( "Already being fired by the scheduler; not run again." ) elif exec_result.get("error"):