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

@@ -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

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()

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"):