From 281cc00db669e2403d07e15ac20e93a549362f91 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 19:10:25 -0700 Subject: [PATCH] refactor(cron): scheduler_provider module logger, adapter-selection helper, drop dead provider_supports_fire_cancel --- cron/scheduler_provider.py | 149 ++++++++--------------------- tests/plugins/test_chronos_cron.py | 7 +- 2 files changed, 40 insertions(+), 116 deletions(-) diff --git a/cron/scheduler_provider.py b/cron/scheduler_provider.py index 6b1add10cf..3b006d427f 100644 --- a/cron/scheduler_provider.py +++ b/cron/scheduler_provider.py @@ -7,13 +7,17 @@ from __future__ import annotations import contextlib import inspect +import logging import threading from abc import ABC, abstractmethod from pathlib import Path from typing import Any +logger = logging.getLogger(__name__) + # Cap for exponential tick backoff during fd exhaustion (interval doubled per failure). -_EMFILE_BACKOFF_MAX_SECONDS = 15 * 60 # 15 minutes +_EMFILE_BACKOFF_MAX_SECONDS = 15 * 60 +DEFAULT_MISFIRE_GRACE_MINUTES = 10 def _backoff_wait_seconds(interval: float, consecutive_failures: int) -> float: @@ -203,38 +207,13 @@ def provider_supports_split_fire(provider: Any) -> bool: return fire_due_impl is None or fire_due_impl is CronScheduler.fire_due -def provider_supports_fire_cancel(provider: Any) -> bool: - """Return whether ``fire_claimed`` accepts a ``cancel_event`` kwarg.""" - try: - parameters = inspect.signature(provider.fire_claimed).parameters.values() - except (TypeError, ValueError): - return False - return any( - parameter.kind is inspect.Parameter.VAR_KEYWORD - or ( - parameter.name == "cancel_event" - and parameter.kind - in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY) - ) - for parameter in parameters - ) - - -DEFAULT_MISFIRE_GRACE_MINUTES = 10 - - def _misfire_grace_minutes() -> float: """``cron.misfire_grace_minutes`` from config; non-positive disables the catch-up sweep.""" try: from hermes_cli.config import cfg_get, load_config return float( - cfg_get( - load_config(), - "cron", - "misfire_grace_minutes", - default=DEFAULT_MISFIRE_GRACE_MINUTES, - ) + cfg_get(load_config(), "cron", "misfire_grace_minutes", default=DEFAULT_MISFIRE_GRACE_MINUTES) ) except Exception: return float(DEFAULT_MISFIRE_GRACE_MINUTES) @@ -253,12 +232,8 @@ def fire_overdue_jobs( concurrent late external retry is de-duplicated by the store CAS; waits out ``cron.misfire_grace_minutes`` so the external retry gets first right. Returns jobs dispatched. """ - import logging - import threading from datetime import datetime - logger = logging.getLogger("cron.scheduler_provider") - if isinstance(provider, InProcessCronScheduler): return 0 @@ -266,7 +241,7 @@ def fire_overdue_jobs( if grace_minutes <= 0: return 0 - from cron.jobs import _ensure_aware, _hermes_now, is_job_runnable, load_jobs + from cron.jobs import ONESHOT_GRACE_SECONDS, _ensure_aware, _hermes_now, is_job_runnable, load_jobs if now is None: now = _hermes_now() @@ -288,21 +263,18 @@ def fire_overdue_jobs( job_id = str(job.get("id") or "") # One-shots past ONESHOT_GRACE_SECONDS "will never fire"; don't resurrect them hours late. schedule = job.get("schedule") or {} - if str(schedule.get("kind") or "") == "once": - from cron.jobs import ONESHOT_GRACE_SECONDS - - if overdue_seconds > ONESHOT_GRACE_SECONDS: - logger.warning( - "Misfire catch-up: one-shot job %s (%s) was due %s " - "(%.0f min overdue) — outside the %ss one-shot grace " - "window, not firing.", - job_id, - job.get("name") or "unnamed", - next_run_at, - overdue_seconds / 60, - ONESHOT_GRACE_SECONDS, - ) - continue + if str(schedule.get("kind") or "") == "once" and overdue_seconds > ONESHOT_GRACE_SECONDS: + logger.warning( + "Misfire catch-up: one-shot job %s (%s) was due %s " + "(%.0f min overdue) — outside the %ss one-shot grace " + "window, not firing.", + job_id, + job.get("name") or "unnamed", + next_run_at, + overdue_seconds / 60, + ONESHOT_GRACE_SECONDS, + ) + continue logger.warning( "Misfire catch-up: job %s (%s) was due %s (%.0f min overdue) and " "no external fire arrived — firing locally.", @@ -335,10 +307,6 @@ def fire_overdue_jobs( def resolve_cron_scheduler() -> "CronScheduler": """Resolve ``cron.provider``; missing/failing/unavailable providers fall back to the built-in with a warning — cron must never be left without a trigger.""" - import logging - - logger = logging.getLogger("cron.scheduler_provider") - name = "" try: from hermes_cli.config import cfg_get, load_config @@ -372,10 +340,7 @@ def scheduler_for_profile_mode( stores: fail closed to the built-in multiplex ticker until the API carries profile identity.""" if not multiplex_profiles or isinstance(provider, InProcessCronScheduler): return provider - - import logging - - logging.getLogger("cron.scheduler_provider").warning( + logger.warning( "cron.provider '%s' does not support multiplex_profiles; using built-in ticker", provider.name, ) @@ -403,16 +368,10 @@ class InProcessCronScheduler(CronScheduler): default_profile=None, profile_gate=None, ): - import logging from cron.scheduler import CronTickYielded from cron.scheduler import tick as cron_tick - from cron.jobs import ( - clear_ticker_error, - record_ticker_error, - record_ticker_heartbeat, - ) + from cron.jobs import clear_ticker_error, record_ticker_error, record_ticker_heartbeat - logger = logging.getLogger("cron.scheduler_provider") logger.info("In-process cron scheduler started (interval=%ds)", interval) # Multiplex: tick EACH profile's store every cycle, heartbeats/recovery scoped per profile. @@ -432,10 +391,7 @@ class InProcessCronScheduler(CronScheduler): recovered = self.recover_interrupted() if recovered: - logger.warning( - "Marked %d interrupted cron execution(s) unknown after restart", - recovered, - ) + logger.warning("Marked %d interrupted cron execution(s) unknown after restart", recovered) # Heartbeat before the first sleep so `hermes cron status` sees a live ticker immediately. record_ticker_heartbeat() # EMFILE backoff: don't hammer the store while fds are exhausted; a clean tick resets it. @@ -446,13 +402,7 @@ class InProcessCronScheduler(CronScheduler): if can_dispatch is not None and not can_dispatch(): logger.debug("Cron dispatch paused while gateway drains existing work") else: - cron_tick( - verbose=False, - adapters=adapters, - loop=loop, - sync=False, - can_dispatch=can_dispatch, - ) + cron_tick(verbose=False, adapters=adapters, loop=loop, sync=False, can_dispatch=can_dispatch) ok = True except BaseException as e: # BaseException, not Exception: a SystemExit must not silently kill the ticker; @@ -488,7 +438,6 @@ class InProcessCronScheduler(CronScheduler): """Tick every profile's store, each scoped via ``_profile_cron_scope``. ``profile_gate(name, home)``, when given, is consulted every cycle; a rejected profile is neither ticked nor heartbeated.""" - import logging from cron.scheduler import tick as cron_tick from cron.scheduler import ( CronTickYielded, @@ -496,19 +445,25 @@ class InProcessCronScheduler(CronScheduler): _is_fd_exhaustion, _primary_profile_routes_for_current_home, ) - from cron.jobs import ( - clear_ticker_error, - record_ticker_error, - record_ticker_heartbeat, - ) + from cron.jobs import clear_ticker_error, record_ticker_error, record_ticker_heartbeat - logger = logging.getLogger("cron.scheduler_provider") logger.info( "Multiplex cron scheduler started for %d profile(s): %s", len(profile_homes), [p[0] if isinstance(p, tuple) else p for p in profile_homes], ) + def tick_adapters_for(profile_name): + # Deliver via the profile's OWN adapters; NEVER fall back to the default profile's (wrong + # bot). A credentialless satellite may ride the PRIMARY adapter only for targets an exact + # enabled route maps here; else fail closed (delivery skipped this tick). + if profile_name is None or profile_name == default_profile: + return adapters + tick_adapters = (profile_adapters or {}).get(profile_name) or {} + if not tick_adapters and adapters: + return SharedRouteAdapters(adapters, _primary_profile_routes_for_current_home()) + return tick_adapters + # Recovery + heartbeat per profile; one broken store must not abort startup for the others. for entry in _existing_profile_homes(profile_homes): _, home = _profile_entry(entry) @@ -517,18 +472,11 @@ class InProcessCronScheduler(CronScheduler): recovered = self.recover_interrupted() if recovered: logger.warning( - "Marked %d interrupted cron execution(s) for profile at %s", - recovered, - home, + "Marked %d interrupted cron execution(s) for profile at %s", recovered, home ) record_ticker_heartbeat() except BaseException as e: - logger.error( - "Cron startup recovery error for profile at %s: %s", - home, - e, - exc_info=True, - ) + logger.error("Cron startup recovery error for profile at %s: %s", home, e, exc_info=True) consecutive_failures = 0 while not stop_event.is_set(): @@ -547,23 +495,9 @@ class InProcessCronScheduler(CronScheduler): for _pname, home in cycle_homes: try: with _profile_cron_scope(home): - # Deliver via the profile's OWN adapters; NEVER fall back to the - # default profile's (wrong bot) — just skip delivery this tick. - if _pname is None or _pname == default_profile: - _tick_adapters = adapters - else: - _tick_adapters = (profile_adapters or {}).get(_pname) or {} - if not _tick_adapters and adapters: - # Credentialless satellite may ride the PRIMARY adapter only - # for targets an exact enabled route maps here; else fail - # closed. - _tick_adapters = SharedRouteAdapters( - adapters, - _primary_profile_routes_for_current_home(), - ) cron_tick( verbose=False, - adapters=_tick_adapters, + adapters=tick_adapters_for(_pname), loop=loop, sync=False, can_dispatch=can_dispatch, @@ -574,12 +508,7 @@ class InProcessCronScheduler(CronScheduler): _profile_errors[str(home)] = f"{type(e).__name__}: {e}" except BaseException as e: # THIS profile only; BaseException as in the single-profile loop. - logger.error( - "Cron tick error for profile at %s: %s", - home, - e, - exc_info=True, - ) + logger.error("Cron tick error for profile at %s: %s", home, e, exc_info=True) _profile_errors[str(home)] = f"{type(e).__name__}: {e}" if _cycle_exc is None or _is_fd_exhaustion(e): _cycle_exc = e diff --git a/tests/plugins/test_chronos_cron.py b/tests/plugins/test_chronos_cron.py index 75971ea2d3..a98ad24628 100644 --- a/tests/plugins/test_chronos_cron.py +++ b/tests/plugins/test_chronos_cron.py @@ -204,16 +204,11 @@ def test_chronos_is_split_fire_capable(chronos): fire webhook uses durable claim admission (not the legacy fire_due path). Chronos deliberately has NO fire_due override — its re-arm logic lives in fire_claimed, which the split path invokes.""" - from cron.scheduler_provider import ( - provider_supports_fire_cancel, - provider_supports_force_fire, - provider_supports_split_fire, - ) + from cron.scheduler_provider import provider_supports_force_fire, provider_supports_split_fire prov, _fake = chronos assert provider_supports_split_fire(prov) is True assert provider_supports_force_fire(prov) is True - assert provider_supports_fire_cancel(prov) is True def test_fire_claimed_no_rearm_when_run_failed(chronos, monkeypatch):