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