From eceebf9b91d37074afd423f2a8cd7abbc08639df Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 14:41:08 +0000 Subject: [PATCH] fix(cron): prevent silent ticker stall; heartbeat + supervisor + delivery timeout The cron ticker could silently stall (observed ~17h with no jobs firing while the gateway stayed up). Root cause: an unbounded await in the tick path (platform delivery send) could wedge the ticker with no timeout and no liveness signal, and an exception escaping the tick body would kill the loop with nothing to restart it. Fix: - run_cron_tick_once records a liveness heartbeat at the top of every tick before any job work, and isolates per-tick exceptions (logged, loop continues) - platform delivery send wrapped in asyncio.wait_for (_DELIVERY_TIMEOUT=60s) - _cron_ticker_supervisor watchdog restarts the ticker when the heartbeat goes stale (>5x tick interval) - tests cover exception isolation, heartbeat-before-tick, stale detection, and supervisor restart/no-restart behavior Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01PQKCc5mDedYAiCNyXnTezh --- cron/scheduler.py | 122 +++++++++++++++++++++++++++++++-- gateway/run.py | 97 ++++++++++++++++++++++---- tests/cron/test_scheduler.py | 128 +++++++++++++++++++++++++++++++++++ 3 files changed, 328 insertions(+), 19 deletions(-) diff --git a/cron/scheduler.py b/cron/scheduler.py index 6b511d38b77d..e5b3b7dcb954 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -17,6 +17,8 @@ import shutil import subprocess import sys +import threading +import time from contextlib import contextmanager # fcntl is Unix-only; on Windows use msvcrt for file locking @@ -44,6 +46,100 @@ logger = logging.getLogger(__name__) +# --------------------------------------------------------------------------- +# Ticker liveness heartbeat + self-healing support. +# +# The gateway drives cron from a single background thread (see +# gateway.run._start_cron_ticker). Historically that loop called tick() +# inline with no liveness signal and no watchdog: if a tick blocked forever +# (a hung delivery send, a stuck job, a pool that never drained) the thread +# wedged and cron silently stopped firing while the gateway itself stayed up. +# One incident ran ~17h with no jobs firing and no error surfaced. +# +# record_ticker_heartbeat() is called at the TOP of every loop iteration, +# BEFORE any job work, so the heartbeat measures the loop's own liveness — +# not whether individual jobs succeed. A supervisor (gateway side) watches +# the heartbeat age and restarts the ticker thread if it goes stale. +# --------------------------------------------------------------------------- +_ticker_heartbeat_lock = threading.Lock() +_ticker_heartbeat = {"monotonic": None, "wall": None, "count": 0} + + +def record_ticker_heartbeat() -> None: + """Record that the cron ticker loop is alive (called at the top of each tick). + + Uses a monotonic clock for age math (immune to wall-clock jumps) and also + stashes a human-readable wall timestamp for logs/diagnostics. Best-effort + file mirror lets operators inspect ticker liveness without attaching to the + process. + """ + with _ticker_heartbeat_lock: + _ticker_heartbeat["monotonic"] = time.monotonic() + _ticker_heartbeat["wall"] = _hermes_now().isoformat() + _ticker_heartbeat["count"] += 1 + count = _ticker_heartbeat["count"] + wall = _ticker_heartbeat["wall"] + + # Best-effort persisted mirror — never let heartbeat bookkeeping raise into + # the ticker loop. + try: + lock_dir, _ = _get_lock_paths() + lock_dir.mkdir(parents=True, exist_ok=True) + (lock_dir / ".ticker_heartbeat").write_text( + json.dumps({"wall": wall, "count": count}), encoding="utf-8" + ) + except Exception: + pass + + +def get_ticker_heartbeat_age() -> Optional[float]: + """Return seconds since the last recorded ticker heartbeat, or None if never set.""" + with _ticker_heartbeat_lock: + mono = _ticker_heartbeat["monotonic"] + if mono is None: + return None + return time.monotonic() - mono + + +def reset_ticker_heartbeat() -> None: + """Clear the heartbeat state (used by tests to start from a known state).""" + with _ticker_heartbeat_lock: + _ticker_heartbeat["monotonic"] = None + _ticker_heartbeat["wall"] = None + _ticker_heartbeat["count"] = 0 + + +def ticker_heartbeat_is_stale(interval: float, multiplier: float = 5.0) -> bool: + """Whether the ticker heartbeat is older than ``multiplier`` tick intervals. + + Returns False when no heartbeat has been recorded yet (the ticker may not + have started) so a supervisor doesn't thrash before the first tick. + """ + age = get_ticker_heartbeat_age() + if age is None: + return False + return age > max(interval, 0.0) * multiplier + + +def run_cron_tick_once(verbose: bool = False, adapters=None, loop=None) -> int: + """Run a single cron tick with heartbeat + exception isolation. + + This is the loop body the gateway ticker thread should call each interval. + The heartbeat is recorded FIRST, before any job work, so a hung or failing + job cannot make the loop look dead. Any exception escaping ``tick()`` is + logged and swallowed so a single bad tick can never kill the ticker thread. + """ + record_ticker_heartbeat() + try: + return tick(verbose=verbose, adapters=adapters, loop=loop) + except Exception: + # Escaped exceptions used to be logged at DEBUG and effectively lost. + # Log with traceback at ERROR so a recurring tick failure is visible, + # then continue — the next tick still fires. + logger.exception("Cron tick raised; loop continues to next interval") + return 0 + + class CronPromptInjectionBlocked(Exception): """Raised by _build_job_prompt when the fully-assembled prompt trips the injection scanner. Caught in run_job so the operator sees a clean @@ -511,6 +607,12 @@ def _resolve_delivery_target(job: dict) -> Optional[dict]: _VIDEO_EXTS = frozenset({'.mp4', '.mov', '.avi', '.mkv', '.webm', '.3gp'}) _IMAGE_EXTS = frozenset({'.jpg', '.jpeg', '.png', '.webp', '.gif'}) +# Upper bound (seconds) on the standalone delivery send. This send runs inline +# on the ticker's tick path, so an unbounded platform HTTP call could wedge the +# whole cron ticker — a delivery that never returns must fail as an error, not +# hang forever. +_DELIVERY_TIMEOUT = 60 + def _send_media_via_adapter( adapter, @@ -731,8 +833,20 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option ) if not delivered: - # Standalone path: run the async send in a fresh event loop (safe from any thread) - coro = _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files) + # Standalone path: run the async send in a fresh event loop (safe from any thread). + # Bound the send with a timeout — this runs inline on the ticker's tick + # path, so a platform whose HTTP send never returns would otherwise wedge + # the whole cron ticker (the 17h silent-stall failure mode). + def _bounded_send_coro(): + return asyncio.wait_for( + _send_to_platform( + platform, pconfig, chat_id, cleaned_delivery_content, + thread_id=thread_id, media_files=media_files, + ), + timeout=_DELIVERY_TIMEOUT, + ) + + coro = _bounded_send_coro() try: result = asyncio.run(coro) except RuntimeError: @@ -742,8 +856,8 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option # fresh thread that has no running loop. coro.close() with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: - future = pool.submit(asyncio.run, _send_to_platform(platform, pconfig, chat_id, cleaned_delivery_content, thread_id=thread_id, media_files=media_files)) - result = future.result(timeout=30) + future = pool.submit(lambda: asyncio.run(_bounded_send_coro())) + result = future.result(timeout=_DELIVERY_TIMEOUT + 5) except Exception as e: msg = f"delivery to {platform_name}:{chat_id} failed: {e}" logger.error("Job '%s': %s", job["id"], msg) diff --git a/gateway/run.py b/gateway/run.py index 67c50daddfe9..5b5f362363e9 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -17741,7 +17741,7 @@ def _start_cron_ticker(stop_event: threading.Event, adapters=None, loop=None, in image/audio/document cache + expired ``hermes debug share`` pastes once per hour. """ - from cron.scheduler import tick as cron_tick + from cron.scheduler import run_cron_tick_once from gateway.platforms.base import cleanup_image_cache, cleanup_document_cache from hermes_cli.debug import _sweep_expired_pastes @@ -17753,10 +17753,10 @@ def _start_cron_ticker(stop_event: threading.Event, adapters=None, loop=None, in logger.info("Cron ticker started (interval=%ds)", interval) tick_count = 0 while not stop_event.is_set(): - try: - cron_tick(verbose=False, adapters=adapters, loop=loop) - except Exception as e: - logger.debug("Cron tick error: %s", e) + # run_cron_tick_once records the liveness heartbeat at the top of the + # tick (before any job work) and swallows/logs any exception, so a + # single failing tick can never kill this loop or make it look dead. + run_cron_tick_once(verbose=False, adapters=adapters, loop=loop) tick_count += 1 @@ -17822,6 +17822,51 @@ def _start_cron_ticker(stop_event: threading.Event, adapters=None, loop=None, in logger.info("Cron ticker stopped") +def _cron_ticker_supervisor( + stop_event: threading.Event, + restart_ticker, + interval: int = 60, + stale_multiplier: float = 5.0, + check_interval: int = 60, +): + """Watchdog that restarts the cron ticker if its heartbeat goes stale. + + The ticker records a liveness heartbeat at the top of every tick + (cron.scheduler.record_ticker_heartbeat). If that heartbeat is older than + ``stale_multiplier`` tick intervals, the ticker thread has almost certainly + wedged (a hung delivery, a stuck job, a pool that never drained). We log a + warning and call ``restart_ticker`` to spin up a fresh ticker thread; the + file lock inside tick() prevents the old and new threads from overlapping. + + This closes the gap behind the 17h silent-stall incident: previously a hung + ticker was never detected and cron simply stopped firing while the gateway + stayed up. + """ + from cron.scheduler import get_ticker_heartbeat_age, ticker_heartbeat_is_stale + + logger.info( + "Cron ticker supervisor started (stale after %.0fs)", + interval * stale_multiplier, + ) + while not stop_event.is_set(): + stop_event.wait(timeout=check_interval) + if stop_event.is_set(): + break + try: + if ticker_heartbeat_is_stale(interval, stale_multiplier): + age = get_ticker_heartbeat_age() + logger.warning( + "Cron ticker heartbeat stale (%.0fs old, > %.0fs threshold) — " + "restarting ticker thread", + age if age is not None else -1.0, + interval * stale_multiplier, + ) + restart_ticker() + except Exception as e: + logger.debug("Cron ticker supervisor error: %s", e) + logger.info("Cron ticker supervisor stopped") + + async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool = False, verbosity: Optional[int] = 0) -> bool: """ Start the gateway and run until interrupted. @@ -18173,15 +18218,34 @@ def restart_signal_handler(): # Start background cron ticker so scheduled jobs fire automatically. # Pass the event loop so cron delivery can use live adapters (E2EE support). cron_stop = threading.Event() - cron_thread = threading.Thread( - target=_start_cron_ticker, - args=(cron_stop,), - kwargs={"adapters": runner.adapters, "loop": asyncio.get_running_loop()}, + _cron_loop = asyncio.get_running_loop() + # Mutable holder so the supervisor can swap in a fresh ticker thread if the + # current one wedges. daemon=True: a hung old thread is abandoned, not joined. + _cron_ticker_holder = {"thread": None} + + def _spawn_cron_ticker() -> None: + t = threading.Thread( + target=_start_cron_ticker, + args=(cron_stop,), + kwargs={"adapters": runner.adapters, "loop": _cron_loop}, + daemon=True, + name="cron-ticker", + ) + _cron_ticker_holder["thread"] = t + t.start() + + _spawn_cron_ticker() + + # Self-healing watchdog: restarts the ticker if its heartbeat goes stale + # (the 17h silent-stall failure mode). Runs in its own daemon thread. + cron_supervisor_thread = threading.Thread( + target=_cron_ticker_supervisor, + args=(cron_stop, _spawn_cron_ticker), daemon=True, - name="cron-ticker", + name="cron-ticker-supervisor", ) - cron_thread.start() - + cron_supervisor_thread.start() + # Wait for shutdown await runner.wait_for_shutdown() @@ -18189,10 +18253,13 @@ def restart_signal_handler(): if runner.exit_reason: logger.error("Gateway exiting with failure: %s", runner.exit_reason) return False - - # Stop cron ticker cleanly + + # Stop cron ticker + supervisor cleanly cron_stop.set() - cron_thread.join(timeout=5) + _current_ticker = _cron_ticker_holder.get("thread") + if _current_ticker is not None: + _current_ticker.join(timeout=5) + cron_supervisor_thread.join(timeout=5) # Close MCP server connections try: diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py index 62bc6b688a0e..59e65ac31481 100644 --- a/tests/cron/test_scheduler.py +++ b/tests/cron/test_scheduler.py @@ -2548,3 +2548,131 @@ def fake_run_coro(coro, _loop): # 2. Second file still got dispatched — one timeout doesn't abort the batch adapter.send_video.assert_called_once() assert adapter.send_video.call_args[1]["video_path"] == str(fast.resolve()) + + +class TestTickerHeartbeatAndSupervisor: + """Regression coverage for the 17h silent-stall incident. + + The cron ticker used to run tick() inline with no liveness signal and no + watchdog: a hung tick wedged the thread and cron silently stopped firing + while the gateway stayed up. These tests lock in the three guarantees of + the fix: per-tick exception isolation, a heartbeat that measures loop + liveness (not job success), and a supervisor that restarts a wedged ticker. + """ + + def setup_method(self): + from cron.scheduler import reset_ticker_heartbeat + reset_ticker_heartbeat() + + def teardown_method(self): + from cron.scheduler import reset_ticker_heartbeat + reset_ticker_heartbeat() + + def test_tick_exception_does_not_kill_loop(self, tmp_path): + """An exception escaping tick() is swallowed so the next tick still fires.""" + from cron import scheduler + + calls = {"n": 0} + + def boom(*args, **kwargs): + calls["n"] += 1 + raise RuntimeError("job execution exploded") + + with patch("cron.scheduler._hermes_home", tmp_path), \ + patch("cron.scheduler.tick", side_effect=boom): + # First tick raises internally but run_cron_tick_once must not. + assert scheduler.run_cron_tick_once(verbose=False) == 0 + # Loop would call it again next interval — it still runs. + assert scheduler.run_cron_tick_once(verbose=False) == 0 + + assert calls["n"] == 2 + + def test_heartbeat_recorded_before_tick_even_when_job_hangs(self, tmp_path): + """Heartbeat is written at the top of the tick, before job work, so a + hung/failing job still leaves a fresh heartbeat behind.""" + from cron import scheduler + + observed = {} + + def hang_then_fail(*args, **kwargs): + # Heartbeat must already be fresh by the time job work runs. + observed["age_during_tick"] = scheduler.get_ticker_heartbeat_age() + raise TimeoutError("job hung") + + assert scheduler.get_ticker_heartbeat_age() is None + with patch("cron.scheduler._hermes_home", tmp_path), \ + patch("cron.scheduler.tick", side_effect=hang_then_fail): + scheduler.run_cron_tick_once(verbose=False) + + # Heartbeat was recorded before the (failing) tick body ran. + assert observed["age_during_tick"] is not None + assert observed["age_during_tick"] < 5.0 + # And it remains fresh afterward. + assert scheduler.get_ticker_heartbeat_age() < 5.0 + + def test_heartbeat_stale_detection(self): + from cron import scheduler + + # No heartbeat yet -> never considered stale (ticker may not have started). + assert scheduler.ticker_heartbeat_is_stale(interval=60, multiplier=5) is False + + scheduler.record_ticker_heartbeat() + # Fresh heartbeat -> not stale. + assert scheduler.ticker_heartbeat_is_stale(interval=60, multiplier=5) is False + + # Simulate an aged heartbeat. + with patch("cron.scheduler.get_ticker_heartbeat_age", return_value=10_000.0): + assert scheduler.ticker_heartbeat_is_stale(interval=60, multiplier=5) is True + + def test_supervisor_restarts_stale_ticker(self): + """When the heartbeat is stale, the supervisor calls the restart callback.""" + import threading + from gateway.run import _cron_ticker_supervisor + + stop_event = threading.Event() + restarts = {"n": 0} + + def fake_restart(): + restarts["n"] += 1 + stop_event.set() # exit the supervisor loop after one restart + + with patch("cron.scheduler.ticker_heartbeat_is_stale", return_value=True): + _cron_ticker_supervisor( + stop_event, + fake_restart, + interval=60, + stale_multiplier=5.0, + check_interval=0, + ) + + assert restarts["n"] == 1 + + def test_supervisor_does_not_restart_when_healthy(self): + """A fresh heartbeat must not trigger a restart.""" + import threading + from gateway.run import _cron_ticker_supervisor + + stop_event = threading.Event() + restarts = {"n": 0} + + def fake_restart(): + restarts["n"] += 1 + + call_count = {"n": 0} + + def not_stale(*args, **kwargs): + call_count["n"] += 1 + if call_count["n"] >= 2: + stop_event.set() # end the loop after a couple of healthy checks + return False + + with patch("cron.scheduler.ticker_heartbeat_is_stale", side_effect=not_stale): + _cron_ticker_supervisor( + stop_event, + fake_restart, + interval=60, + stale_multiplier=5.0, + check_interval=0, + ) + + assert restarts["n"] == 0