From 50febe97d94bdbc8473ba716cbe7b26b41823202 Mon Sep 17 00:00:00 2001 From: kshitij <82637225+kshitijk4poor@users.noreply.github.com> Date: Sat, 15 Aug 2026 01:48:11 +0530 Subject: [PATCH] =?UTF-8?q?fix:=20follow-up=20for=20salvaged=20PR=20#86129?= =?UTF-8?q?=20=E2=80=94=20config.yaml,=20eliminate=20redundant=20load=5Fjo?= =?UTF-8?q?bs,=20consolidate=20dispatch?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Read inflight_max_minutes from config.yaml first (cron.inflight_max_minutes), keep HERMES_CRON_INFLIGHT_MAX_MINUTES env var as internal escape hatch only. - Skip the redundant load_jobs() call in tick() when there are no in-flight claims or when due_jobs already covers the in-flight set (get_due_jobs calls load_jobs internally, so this avoids a second file read on every active tick). - Consolidate _job_interval_minutes by normalizing string schedule to dict first, eliminating ~8 lines of duplicated dispatch logic. --- cron/scheduler.py | 62 +++++++++++++++++++++++++++++------------------ 1 file changed, 38 insertions(+), 24 deletions(-) diff --git a/cron/scheduler.py b/cron/scheduler.py index 4291f53ad9..fc45dd013e 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -579,7 +579,21 @@ def _inflight_min_allowance_minutes() -> float: Effective allowance per job is ``max(2 * interval, this)``, so a slow-but-healthy long-interval job is never clipped by the sweep. + Reads ``cron.inflight_max_minutes`` from config.yaml; the + ``HERMES_CRON_INFLIGHT_MAX_MINUTES`` env var is kept as an internal + escape hatch only. """ + try: + _ucfg = load_config() or {} + _cfg_val = ( + _ucfg.get("cron", {}) if isinstance(_ucfg, dict) else {} + ).get("inflight_max_minutes") + if _cfg_val is not None: + val = float(_cfg_val) + if val > 0: + return val + except Exception: + pass raw = os.getenv("HERMES_CRON_INFLIGHT_MAX_MINUTES", "").strip() if raw: try: @@ -636,6 +650,10 @@ def _job_interval_minutes(job: dict) -> Optional[float]: """ try: schedule = job.get("schedule") + if isinstance(schedule, str) and schedule.strip(): + from cron.jobs import parse_schedule + + schedule = parse_schedule(schedule) or {} if isinstance(schedule, dict): kind = schedule.get("kind") if kind == "interval": @@ -643,18 +661,8 @@ def _job_interval_minutes(job: dict) -> Optional[float]: return float(minutes) if minutes else None if kind == "cron": return _cron_interval_minutes(str(schedule.get("expr") or "")) - return None - if isinstance(schedule, str) and schedule.strip(): - from cron.jobs import parse_schedule - - parsed = parse_schedule(schedule) or {} - if parsed.get("kind") == "interval": - minutes = parsed.get("minutes") - return float(minutes) if minutes else None - if parsed.get("kind") == "cron": - return _cron_interval_minutes(str(parsed.get("expr") or "")) except Exception: - return None + pass return None @@ -5478,20 +5486,26 @@ def tick( # Bound the in-flight set BEFORE the dedup guard is consulted, so a # leaked claim is force-released in-cycle rather than silently eating - # every subsequent fire until the gateway process restarts. Runs even - # when nothing is due — a wedged job's own fires are exactly what the - # leak suppresses (t_3778a491; recurring no_agent router/watchdog jobs, - # 2026-08-14 t_20e23f84). - try: - from cron.jobs import load_jobs as _load_all_jobs + # every subsequent fire until the gateway process restarts. Skips the + # extra load_jobs when there are no in-flight claims (the common idle + # tick) and reuses due_jobs when they already cover the in-flight set + # (get_due_jobs calls load_jobs internally, so this avoids a redundant + # second file read on every active tick). + if _running_job_ids: + _sweep_jobs = due_jobs + try: + _inflight_ids = set(_running_job_ids) + _due_ids = {j.get("id") for j in due_jobs if isinstance(j, dict)} + if not _inflight_ids <= _due_ids: + from cron.jobs import load_jobs as _load_all_jobs - _all_jobs = _load_all_jobs() - except Exception: - _all_jobs = due_jobs - try: - sweep_stale_inflight(_all_jobs) - except Exception as e: - logger.warning("Stale in-flight sweep failed: %s", e) + _sweep_jobs = _load_all_jobs() + except Exception: + pass + try: + sweep_stale_inflight(_sweep_jobs) + except Exception as e: + logger.warning("Stale in-flight sweep failed: %s", e) if not due_jobs: # Idle tick: skip config load + pool partitioning entirely