Cron ticker silent failure detection and resilience

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #50016 fix(cron): keep ticker alive on BaseException + heartbeat-aware status

  merged to main

  files: cron/jobs.py, cron/scheduler.py, cron/scheduler_provider.py, hermes_cli/cron.py, scripts/release.py, tests/cron/test_scheduler_provider.py

  Show the diff

  • cron/jobs.py

    @@ -12,6 +12,7 @@
     import shutil
     import tempfile
     import threading
    +import time
     import os
     import re
     import uuid
    @@ -51,6 +52,20 @@
     HERMES_DIR = get_hermes_home().resolve()
     CRON_DIR = HERMES_DIR / "cron"
     JOBS_FILE = CRON_DIR / "jobs.json"
    +# Heartbeat file the in-process ticker touches on every loop iteration. The
    +# gateway process and the (separate) ``hermes cron status`` process share it
    +# so status can tell whether the ticker THREAD is alive, not just whether the
    +# gateway PROCESS exists — a ticker that dies silently inside a live gateway
    +# would otherwise report healthy (#32612, #32895).
    +TICKER_HEARTBEAT_FILE = CRON_DIR / "ticker_heartbeat"
    +# Last tick that completed WITHOUT raising. Distinguishing this from the plain
    +# heartbeat lets status detect a ticker that is alive but failing every tick.
    +TICKER_SUCCESS_FILE = CRON_DIR / "ticker_last_success"
    +# Default ticker loop interval (seconds). The single source of truth shared by
    +# the in-process ticker (cron/scheduler_provider.py) and the staleness
    +# threshold in `hermes cron status` (hermes_cli/cron.py), so the two never
    +# drift apart.
    +TICKER_INTERVAL_SECONDS = 60
     
     # In-process lock protecting load_jobs→modify→save_jobs cycles.
     # Required when tick() runs jobs in parallel threads — without this,
    @@ -499,6 +514,78 @@ def compute_next_run(schedule: Dict[str, Any], last_run_at: Optional[str] = None
         return None
     
     
    +# =============================================================================
    +# Ticker heartbeat (liveness signal for `hermes cron status`)
    +# =============================================================================
    +
    +def _atomic_write_epoch(path: Path) -> None:
    +    """Atomically write the current epoch time to ``path``.
    +
    +    Uses the same tmpfile + ``atomic_replace`` pattern as ``save_jobs`` so a
    +    concurrent reader in another process (``hermes cron status``) never sees a
    +    torn/truncated file. Best-effort: failures are swallowed by callers.
    +    """
    +    ensure_dirs()
    +    fd, tmp_path = tempfile.mkstemp(dir=str(CRON_DIR), suffix=".tmp", prefix=".hb_")
    +    try:
    +        with os.fdopen(fd, "w", encoding="utf-8") as f:
    +            f.write(str(time.time()))
    +            f.flush()
    +            os.fsync(f.fileno())
    +        atomic_replace(tmp_path, path)
    +    except BaseException:
    +        try:
    +            os.unlink(tmp_path)
    +        except OSError:
    +            pass
    +        raise
    +
    +
    +def record_ticker_heartbeat(success: bool = False) -> None:
    +    """Record a ticker liveness signal, and optionally a successful-tick signal.
    +
    +    The ticker calls this once per loop iteration. ``success=True`` additionally
    +    bumps the *last successful tick* marker. We track two distinct signals so
    +    `hermes cron status` can tell a thread that is merely *alive and looping*
    +    (heartbeat fresh, success stale) from one that is actually *firing jobs*
    +    (both fresh) — a ticker stuck failing every tick would otherwise keep the
    +    plain heartbeat fresh and falsely report healthy (#32612, #32895).
    +
    +    Best-effort: a write failure must never disrupt the tick loop.
    +    """
    +    try:
    +        _atomic_write_epoch(TICKER_HEARTBEAT_FILE)
    +    except Exception:
    +        pass
    +    if success:
    +        try:
    +            _atomic_write_epoch(TICKER_SUCCESS_FILE)
    +        except Exception:
    +            pass
    +
    +
    +def _epoch_file_age(path: Path) -> Optional[float]:
    +    try:
    +        raw = path.read_text(encoding="utf-8").strip()
    +        return max(0.0, time.time() - float(raw))
    +    except Exception:
    +        return None
    +
    +
    +def get_ticker_heartbeat_age() -> Optional[float]:
    +    """Seconds since the ticker loop last iterated, or None if unknown.
    +
    +    None = heartbeat file missing/unreadable (older build, never ran, or a
    +    torn read). Callers treat None as "cannot determine", not "dead".
    +    """
    +    return _epoch_file_age(TICKER_HEARTBEAT_FILE)
    +
    +
    +def get_ticker_success_age() -> Optional[float]:
    +    """Seconds since the ticker last completed a tick WITHOUT raising, or None."""
    +    return _epoch_file_age(TICKER_SUCCESS_FILE)
    +
    +
     # =============================================================================
     # Job CRUD Operations
     # =============================================================================

  • cron/scheduler.py

    @@ -2332,6 +2332,12 @@ def _sweep_mcp_orphans() -> None:
     
                 def _on_done(_f: concurrent.futures.Future) -> None:
                     _remaining[0] -= 1
    +                try:
    +                    _exc = _f.exception()
    +                    if _exc is not None:
    +                        logger.error("Cron job future failed in async mode: %s", _exc, exc_info=(type(_exc), _exc, _exc.__traceback__))
    +                except Exception:
    +                    pass
                     if _remaining[0] <= 0:
                         _sweep_mcp_orphans()
     

  • cron/scheduler_provider.py

    @@ -166,12 +166,29 @@ def name(self) -> str:
         def start(self, stop_event, *, adapters=None, loop=None, interval=60):
             import logging
             from cron.scheduler import tick as cron_tick
    +        from cron.jobs import record_ticker_heartbeat
     
             logger = logging.getLogger("cron.scheduler_provider")
             logger.info("In-process cron scheduler started (interval=%ds)", interval)
    +        # Heartbeat once before the first sleep so `hermes cron status` sees a
    +        # live ticker immediately after startup, not only after the first tick.
    +        record_ticker_heartbeat()
             while not stop_event.is_set():
    +            ok = False
                 try:
                     cron_tick(verbose=False, adapters=adapters, loop=loop, sync=False)
    -            except Exception as e:
    -                logger.debug("Cron tick error: %s", e)
    +                ok = True
    +            except BaseException as e:
    +                # Catch BaseException (not just Exception) so a SystemExit from
    +                # a misbehaving provider SDK / agent retry path does not kill
    +                # the ticker thread silently (#32612). KeyboardInterrupt is
    +                # intentionally caught here too — gateway shutdown is driven by
    +                # stop_event (set by the main thread's signal handler), not by
    +                # an exception in this daemon thread, so swallowing it and
    +                # re-checking stop_event keeps shutdown clean.
    +                logger.error("Cron tick error: %s", e, exc_info=True)
    +            # Record liveness every iteration; bump the success marker only on a
    +            # clean tick, so status can tell "alive but failing every tick" from
    +            # "actually firing jobs" (#32612, #32895).
    +            record_ticker_heartbeat(success=ok)
                 stop_event.wait(interval)

  • hermes_cli/cron.py

    @@ -160,8 +160,48 @@ def cron_status():
     
         pids = find_gateway_pids()
         if pids:
    -        print(color("✓ Gateway is running — cron jobs will fire automatically", Colors.GREEN))
    -        print(f"  PID: {', '.join(map(str, pids))}")
    +        # The gateway PROCESS is alive — but the cron ticker THREAD inside it
    +        # can die silently, or stay alive while every tick fails. Check both
    +        # the liveness heartbeat and the last-successful-tick marker so we
    +        # don't report "will fire" when the ticker is dead or failing
    +        # (#32612, #32895).
    +        from cron.jobs import (
    +            get_ticker_heartbeat_age,
    +            get_ticker_success_age,
    +            TICKER_INTERVAL_SECONDS,
    +        )
    +
    +        # Allow ~3 missed ticker iterations (+ a little slack) before declaring
    +        # trouble. Derived from the shared interval constant so this threshold
    +        # tracks the ticker cadence instead of assuming a hardcoded 60s.
    +        STALE_AFTER = TICKER_INTERVAL_SECONDS * 3 + 20  # = 200s at the 60s default
    +        hb_age = get_ticker_heartbeat_age()
    +        ok_age = get_ticker_success_age()
    +
    +        if hb_age is not None and hb_age > STALE_AFTER:
    +            # No heartbeat at all → the ticker thread is gone.
    +            print(color(
    +                "⚠ Gateway is running but the cron ticker looks STALLED — "
    +                f"no heartbeat for {int(hb_age)}s (expected every ~60s).",
    +                Colors.YELLOW,
    +            ))
    +            print(f"  PID: {', '.join(map(str, pids))}")
    +            print("  Cron jobs may NOT be firing. Restart: hermes gateway restart")
    +        elif hb_age is not None and ok_age is not None and ok_age > STALE_AFTER:
    +            # Loop is alive (fresh heartbeat) but no tick has SUCCEEDED in a
    +            # long time → ticks are failing every iteration.
    +            print(color(
    +                "⚠ Gateway and cron ticker are running, but no tick has "
    +                f"succeeded in {int(ok_age)}s — ticks may be failing.",
    +                Colors.YELLOW,
    +            ))
    +            print(f"  PID: {', '.join(map(str, pids))}")
    +            print("  Check the gateway log for 'Cron tick error'.")
    +        else:
    +            print(color("✓ Gateway is running — cron jobs will fire automatically", Colors.GREEN))
    +            print(f"  PID: {', '.join(map(str, pids))}")
    +            if hb_age is not None:
    +                print(f"  Ticker heartbeat: {int(hb_age)}s ago")
         else:
             print(color("✗ Gateway is not running — cron jobs will NOT fire", Colors.RED))
             print()

  • scripts/release.py

    @@ -47,6 +47,7 @@
     AUTHOR_MAP = {
         "[redacted:email]": "szzhoujiarui-sketch",  # cron model.default salvage co-author (#45550)
         "[redacted:email]": "rayjun",  # cron model.default salvage co-author (#43952)
    +    "[redacted:email]": "sweetcornna",  # cron ticker-liveness salvage co-author (#33849)
         "[redacted:email]": "izumi0uu",  # PR #49544 salvage (native rich reply echo; #49534)
         "[redacted:email]": "w31rdm4ch1nZ",
         "[redacted:email]": "x7peeps",

  • tests/cron/test_scheduler_provider.py

    @@ -332,3 +332,192 @@ def test_fire_due_missing_job_does_not_run(monkeypatch):
     
         assert InProcessCronScheduler().fire_due("gone") is False
         assert ran == []
    +
    +
    +# ── F2a: ticker liveness — survival, heartbeat, honest status (#32612, #32895) ──
    +
    +
    +def test_ticker_survives_baseexception_from_tick():
    +    """A BaseException (e.g. SystemExit from a provider SDK) raised by tick()
    +    must NOT kill the ticker loop — it logs and keeps looping (#32612)."""
    +    from cron.scheduler_provider import InProcessCronScheduler
    +
    +    calls = []
    +
    +    def _boom(*a, **k):
    +        calls.append(1)
    +        if len(calls) == 1:
    +            raise SystemExit("provider SDK called sys.exit")
    +        return 0
    +
    +    stop = threading.Event()
    +    prov = InProcessCronScheduler()
    +    with patch("cron.scheduler.tick", side_effect=_boom), \
    +         patch("cron.jobs.record_ticker_heartbeat"):
    +        t = threading.Thread(target=prov.start, args=(stop,), kwargs={"interval": 0}, daemon=True)
    +        t.start()
    +        time.sleep(0.2)
    +        stop.set()
    +        t.join(timeout=5)
    +
    +    assert not t.is_alive(), "ticker thread died on BaseException instead of surviving"
    +    assert len(calls) >= 2, "ticker did not keep ticking after the BaseException"
    +
    +
    +def test_ticker_records_heartbeat_each_iteration():
    +    """The loop records a liveness heartbeat on start and after each tick,
    +    bumping the success marker only on a clean tick."""
    +    from cron.scheduler_provider import InProcessCronScheduler
    +
    +    beats = []  # (success,) per call
    +    stop = threading.Event()
    +    prov = InProcessCronScheduler()
    +    with patch("cron.scheduler.tick", side_effect=lambda *a, **k: 0), \
    +         patch("cron.jobs.record_ticker_heartbeat",
    +               side_effect=lambda success=False: beats.append(success)):
    +        t = threading.Thread(target=prov.start, args=(stop,), kwargs={"interval": 0}, daemon=True)
    +        t.start()
    +        time.sleep(0.2)
    +        stop.set()
    +        t.join(timeout=5)
    +
    +    # one pre-loop liveness beat (success=False) + post-tick beats with success=True
    +    assert len(beats) >= 2, "ticker did not record heartbeats"
    +    assert beats[0] is False, "pre-loop beat should be liveness-only"
    +    assert any(b is True for b in beats[1:]), "successful tick did not bump success marker"
    +
    +
    +def test_failing_tick_records_liveness_but_not_success():
    +    """A tick that raises bumps the liveness heartbeat but NOT the success
    +    marker — so status can distinguish 'alive but failing' from 'firing'."""
    +    from cron.scheduler_provider import InProcessCronScheduler
    +
    +    beats = []
    +    stop = threading.Event()
    +    prov = InProcessCronScheduler()
    +    with patch("cron.scheduler.tick", side_effect=RuntimeError("every tick fails")), \
    +         patch("cron.jobs.record_ticker_heartbeat",
    +               side_effect=lambda success=False: beats.append(success)):
    +        t = threading.Thread(target=prov.start, args=(stop,), kwargs={"interval": 0}, daemon=True)
    +        t.start()
    +        time.sleep(0.2)
    +        stop.set()
    +        t.join(timeout=5)
    +
    +    # every post-tick beat must be success=False (ticks always failed)
    +    assert len(beats) >= 2
    +    assert all(b is False for b in beats), "a failing tick wrongly bumped the success marker"
    +
    +
    +def test_heartbeat_roundtrip_and_age(tmp_path, monkeypatch):
    +    """record_ticker_heartbeat writes fresh timestamps atomically; the age
    +    getters read them back as small positive ages."""
    +    import cron.jobs as jobs
    +
    +    cron_dir = tmp_path / "cron"
    +    monkeypatch.setattr(jobs, "CRON_DIR", cron_dir)
    +    monkeypatch.setattr(jobs, "OUTPUT_DIR", cron_dir / "output")
    +    monkeypatch.setattr(jobs, "TICKER_HEARTBEAT_FILE", cron_dir / "ticker_heartbeat")
    +    monkeypatch.setattr(jobs, "TICKER_SUCCESS_FILE", cron_dir / "ticker_last_success")
    +
    +    # No files yet -> unknown (None), NOT "dead"
    +    assert jobs.get_ticker_heartbeat_age() is None
    +    assert jobs.get_ticker_success_age() is None
    +
    +    # liveness-only: heartbeat set, success still unknown
    +    jobs.record_ticker_heartbeat(success=False)
    +    hb = jobs.get_ticker_heartbeat_age()
    +    assert hb is not None and 0.0 <= hb < 5.0
    +    assert jobs.get_ticker_success_age() is None
    +
    +    # success: both set
    +    jobs.record_ticker_heartbeat(success=True)
    +    ok = jobs.get_ticker_success_age()
    +    assert ok is not None and 0.0 <= ok < 5.0
    +
    +
    +def test_heartbeat_age_detects_staleness(tmp_path, monkeypatch):
    +    """A heartbeat written far in the past reads back as a large age."""
    +    import cron.jobs as jobs
    +
    +    cron_dir = tmp_path / "cron"
    +    cron_dir.mkdir(parents=True)
    +    hb = cron_dir / "ticker_heartbeat"
    +    monkeypatch.setattr(jobs, "CRON_DIR", cron_dir)
    +    monkeypatch.setattr(jobs, "TICKER_HEARTBEAT_FILE", hb)
    +
    +    import time as _t
    +    hb.write_text(str(_t.time() - 10_000), encoding="utf-8")
    +    age = jobs.get_ticker_heartbeat_age()
    +    assert age is not None and age > 9_000
    +
    +
    +def test_heartbeat_write_failure_is_silent(tmp_path, monkeypatch):
    +    """A real atomic-write failure must be swallowed AND leave no temp file.
    +
    +    Point CRON_DIR at a path that cannot be created (its parent is a regular
    +    file), so ensure_dirs()/mkstemp inside _atomic_write_epoch genuinely fail.
    +    record_ticker_heartbeat must not raise, and no stray .hb_*.tmp may leak.
    +    """
    +    import cron.jobs as jobs
    +
    +    blocker = tmp_path / "not_a_dir"
    +    blocker.write_text("i am a file, not a directory")
    +    bad_cron_dir = blocker / "cron"  # parent is a file -> mkdir/mkstemp fail
    +    monkeypatch.setattr(jobs, "CRON_DIR", bad_cron_dir)
    +    monkeypatch.setattr(jobs, "OUTPUT_DIR", bad_cron_dir / "output")
    +    monkeypatch.setattr(jobs, "TICKER_HEARTBEAT_FILE", bad_cron_dir / "ticker_heartbeat")
    +    monkeypatch.setattr(jobs, "TICKER_SUCCESS_FILE", bad_cron_dir / "ticker_last_success")
    +
    +    jobs.record_ticker_heartbeat(success=True)  # must not raise
    +
    +    # The write never succeeded, so no heartbeat is recorded...
    +    assert jobs.get_ticker_heartbeat_age() is None
    +    # ...and no stray temp file leaked anywhere under tmp_path.
    +    assert not list(tmp_path.rglob(".hb_*.tmp")), "atomic write leaked a temp file on failure"
    +
    +
    +def test_cron_status_reports_alive_but_failing(tmp_path, monkeypatch, capsys):
    +    """cron_status warns when the ticker is alive (fresh heartbeat) but no tick
    +    has succeeded recently (#32612: alive-but-failing must not look healthy)."""
    +    import cron.jobs as jobs
    +    from hermes_cli import cron as cron_cli
    +
    +    monkeypatch.setattr("hermes_cli.gateway.find_gateway_pids", lambda: [4321])
    +    monkeypatch.setattr(jobs, "get_ticker_heartbeat_age", lambda: 5.0)      # fresh
    +    monkeypatch.setattr(jobs, "get_ticker_success_age", lambda: 9_999.0)    # stale
    +    monkeypatch.setattr("cron.jobs.list_jobs", lambda **k: [])
    +
    +    cron_cli.cron_status()
    +    out = capsys.readouterr().out
    +    assert "no tick has succeeded" in out
    +    assert "will fire automatically" not in out
    +
    +
    +def test_cron_status_healthy_when_both_fresh(tmp_path, monkeypatch, capsys):
    +    import cron.jobs as jobs
    +    from hermes_cli import cron as cron_cli
    +
    +    monkeypatch.setattr("hermes_cli.gateway.find_gateway_pids", lambda: [4321])
    +    monkeypatch.setattr(jobs, "get_ticker_heartbeat_age", lambda: 5.0)
    +    monkeypatch.setattr(jobs, "get_ticker_success_age", lambda: 5.0)
    +    monkeypatch.setattr("cron.jobs.list_jobs", lambda **k: [])
    +
    +    cron_cli.cron_status()
    +    out = capsys.readouterr().out
    +    assert "will fire automatically" in out
    +
    +
    +def test_cron_status_reports_stalled_when_no_heartbeat(tmp_path, monkeypatch, capsys):
    +    import cron.jobs as jobs
    +    from hermes_cli import cron as cron_cli
    +
    +    monkeypatch.setattr("hermes_cli.gateway.find_gateway_pids", lambda: [4321])
    +    monkeypatch.setattr(jobs, "get_ticker_heartbeat_age", lambda: 9_999.0)  # dead
    +    monkeypatch.setattr(jobs, "get_ticker_success_age", lambda: 9_999.0)
    +    monkeypatch.setattr("cron.jobs.list_jobs", lambda **k: [])
    +
    +    cron_cli.cron_status()
    +    out = capsys.readouterr().out
    +    assert "STALLED" in out
    +    assert "will fire automatically" not in out

• #60855 fix(cron): bound the jobs lock and claim freshness so the ticker can't stall silently (#60703)

  merged to main

  files: cron/jobs.py, tests/cron/test_ticker_stall_60703.py, tools/cronjob_tools.py

  Show the diff

  • cron/jobs.py

    @@ -84,6 +84,15 @@
     # 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"):