From 815934ae52b95dfb6fc95fe8716bd8e0f79c94f1 Mon Sep 17 00:00:00 2001 From: webtecnica Date: Sun, 16 Aug 2026 12:48:26 -0300 Subject: [PATCH] fix(cron): scheduler self-heals after EMFILE instead of stalling silently (#87644) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tick() swallowed a real OSError at tick-lock acquisition as 'another instance holds the lock', so fd exhaustion (EMFILE/ENFILE) made the scheduler return 0 — recorded as a successful tick — while no job ever ran again. Heartbeat and success markers stayed fresh, masking the stall. - propagate lock-acquisition OSError to the ticker loop (records + backs off) - detect fd exhaustion, attempt gc.collect() + raise soft nofile limit - exponential backoff so an exhausted process stops hammering the store - preserve genuine lock contention (EWOULDBLOCK) silent-skip behavior - 11 regression tests --- cron/scheduler.py | 99 ++++++++- cron/scheduler_provider.py | 59 +++++- hermes_cli/cron.py | 12 +- tests/cron/test_cron_emfile_stall_87644.py | 221 +++++++++++++++++++++ 4 files changed, 383 insertions(+), 8 deletions(-) create mode 100644 tests/cron/test_cron_emfile_stall_87644.py diff --git a/cron/scheduler.py b/cron/scheduler.py index fe72208b22..61a04e22f3 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -13,6 +13,7 @@ import atexit import concurrent.futures import contextlib import contextvars +import errno import json import logging import os @@ -1334,6 +1335,68 @@ def _get_lock_paths() -> tuple[Path, Path]: return lock_dir, lock_dir / ".tick.lock" +# Errnos that mean "another ticker (or manual tick) holds the tick lock", +# as opposed to a real failure opening/locking the file. Everything else — +# most importantly EMFILE/ENFILE (fd exhaustion, #87644) and EACCES on +# open() — must be surfaced, never swallowed as lock contention. +def _is_lock_contention_errno(err: OSError) -> bool: + """Return True when *err* from the lock syscall means the lock is held. + + - POSIX: ``flock(LOCK_EX|LOCK_NB)`` reports EWOULDBLOCK/EAGAIN when + another process holds the lock (EACCES on some NFS implementations). + - Windows: ``msvcrt.locking(LK_NBLCK)`` reports EACCES/EDEADLK. + """ + if err.errno is None: + return False + if fcntl is not None: + return err.errno in (errno.EWOULDBLOCK, errno.EAGAIN, errno.EACCES) + if msvcrt is not None: + return err.errno in (errno.EACCES, errno.EDEADLK) + return False + + +def _is_fd_exhaustion(exc: BaseException) -> bool: + """Return True when *exc* indicates file-descriptor exhaustion. + + Recognizes EMFILE/ENFILE by errno, and the "Too many open files" wording + for wrapped exceptions (``load_jobs`` wraps the raw OSError in a + RuntimeError with that message, #87644). + """ + if isinstance(exc, OSError) and exc.errno in (errno.EMFILE, errno.ENFILE): + return True + text = str(exc).lower() + return "too many open files" in text or "emfile" in text + + +def _reclaim_fds_best_effort() -> None: + """Best-effort attempt to free leaked file descriptors. + + The cron FD-leak family (#60859, #79742, #80792) leaks descriptors from + abandoned workers/sessions. Two safe, idempotent levers: + + 1. ``gc.collect()`` — closes file-like objects held only in reference + cycles (the classic unclosed-file leak shape), which CPython would + otherwise never finalize. + 2. ``apply_nofile_soft_limit()`` — raise RLIMIT_NOFILE's soft limit + toward the configured target when the hard limit allows, giving the + process headroom to keep serving even before every leak is freed. + + Never raises: a reclamation attempt must not make the ticker worse. + """ + try: + import gc + + gc.collect() + except Exception: + pass + try: + from hermes_cli.resource_limits import apply_nofile_soft_limit + + apply_nofile_soft_limit(None) + except Exception: + pass + + def _resolve_origin(job: dict) -> Optional[dict]: """Extract origin info from a job, preserving any extra routing metadata. @@ -6471,7 +6534,12 @@ def tick( lock_dir, lock_file = _get_lock_paths() lock_dir.mkdir(parents=True, exist_ok=True) - # Cross-platform file locking: fcntl on Unix, msvcrt on Windows + # Cross-platform file locking: fcntl on Unix, msvcrt on Windows. + # Only genuine lock contention (another ticker holds the lock) skips the + # tick silently. A real OSError — most importantly EMFILE/ENFILE from fd + # exhaustion — must NOT be swallowed as "another instance holds the + # lock": that previously made the scheduler appear healthy (tick returned + # 0, heartbeat recorded success) while no job ever ran again (#87644). lock_fd = None try: lock_fd = open(lock_file, "w", encoding="utf-8") @@ -6479,11 +6547,32 @@ def tick( fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) elif msvcrt: msvcrt.locking(lock_fd.fileno(), msvcrt.LK_NBLCK, 1) - except (OSError, IOError): - logger.debug("Tick skipped — another instance holds the lock") + except OSError as exc: + if lock_fd is not None and _is_lock_contention_errno(exc): + logger.debug("Tick skipped — another instance holds the lock") + try: + lock_fd.close() + except OSError: + pass + return 0 + # Real failure: log loudly, attempt fd reclamation, and let the + # caller (ticker loop) see a FAILED tick so liveness degrades + # instead of reporting healthy-while-stalled. if lock_fd is not None: - lock_fd.close() - return 0 + try: + lock_fd.close() + except OSError: + pass + if _is_fd_exhaustion(exc): + logger.error( + "Cron tick could not acquire tick lock: %s — attempting fd " + "reclamation; scheduler will retry on the next tick", + exc, + ) + _reclaim_fds_best_effort() + else: + logger.error("Cron tick could not acquire tick lock: %s", exc) + raise try: # Global emergency stop (`hermes pause`): skip dispatch entirely while diff --git a/cron/scheduler_provider.py b/cron/scheduler_provider.py index e5923d5ed0..8b49ea894c 100644 --- a/cron/scheduler_provider.py +++ b/cron/scheduler_provider.py @@ -24,6 +24,13 @@ import threading from abc import ABC, abstractmethod from typing import Any +# Cap for the exponential tick backoff applied while consecutive ticks fail +# with fd exhaustion (EMFILE/ENFILE, #87644). Base is the tick interval +# (60s by default); each consecutive EMFILE failure doubles the wait, capped +# here so a still-alive-but-exhausted gateway never sleeps longer than this +# between recovery attempts. +_EMFILE_BACKOFF_MAX_SECONDS = 15 * 60 # 15 minutes + class CronScheduler(ABC): """Axis-B trigger provider. Decides WHEN a due cron job fires. @@ -348,6 +355,10 @@ class InProcessCronScheduler(CronScheduler): ): import logging from cron.scheduler import tick as cron_tick + from cron.scheduler import ( + _is_fd_exhaustion, + _reclaim_fds_best_effort, + ) from cron.jobs import ( clear_ticker_error, record_ticker_error, @@ -385,6 +396,12 @@ class InProcessCronScheduler(CronScheduler): # 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() + # Exponential backoff for consecutive tick failures — most importantly + # fd exhaustion (EMFILE/ENFILE, #87644). While FDs stay exhausted the + # ticker must NOT hammer the store every 60s; once they free (leak + # fixed, reclamation ran) the next tick succeeds and the backoff + # resets, so the scheduler self-heals without a gateway restart. + consecutive_failures = 0 while not stop_event.is_set(): ok = False try: @@ -415,13 +432,30 @@ class InProcessCronScheduler(CronScheduler): # uid went unnoticed for ~14h with the reason buried in the # gateway log (#68483). record_ticker_error(f"{type(e).__name__}: {e}") + if _is_fd_exhaustion(e): + # EMFILE: try to free descriptors (gc.collect + raise the + # soft nofile limit) so the NEXT tick can succeed; back off + # exponentially so the exhausted process stops hammering + # the store while it has no chance of making progress + # (#87644). + _reclaim_fds_best_effort() + consecutive_failures += 1 + else: + consecutive_failures = 0 # 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) if ok: clear_ticker_error() - stop_event.wait(interval) + consecutive_failures = 0 + wait_seconds = interval + if consecutive_failures > 0: + wait_seconds = min( + interval * (2 ** (consecutive_failures - 1)), + _EMFILE_BACKOFF_MAX_SECONDS, + ) + stop_event.wait(wait_seconds) def _start_multiplex( self, @@ -443,6 +477,10 @@ class InProcessCronScheduler(CronScheduler): """ import logging from cron.scheduler import tick as cron_tick + from cron.scheduler import ( + _is_fd_exhaustion, + _reclaim_fds_best_effort, + ) from cron.jobs import ( clear_ticker_error, record_ticker_error, @@ -475,8 +513,10 @@ class InProcessCronScheduler(CronScheduler): finally: reset_hermes_home_override(home_token) + consecutive_failures = 0 while not stop_event.is_set(): ok = False + _tick_error = None try: if can_dispatch is not None and not can_dispatch(): logger.debug("Cron dispatch paused while gateway drains existing work") @@ -499,6 +539,13 @@ class InProcessCronScheduler(CronScheduler): except BaseException as e: logger.error("Cron tick error: %s", e, exc_info=True) _tick_error = f"{type(e).__name__}: {e}" + if _is_fd_exhaustion(e): + # EMFILE: attempt fd reclamation so the next tick can + # succeed, and back off exponentially (#87644). + _reclaim_fds_best_effort() + consecutive_failures += 1 + else: + consecutive_failures = 0 else: _tick_error = None # Record per-profile heartbeat after each tick cycle. @@ -517,4 +564,12 @@ class InProcessCronScheduler(CronScheduler): record_ticker_error(_tick_error) finally: reset_hermes_home_override(home_token) - stop_event.wait(interval) + if ok: + consecutive_failures = 0 + wait_seconds = interval + if consecutive_failures > 0: + wait_seconds = min( + interval * (2 ** (consecutive_failures - 1)), + _EMFILE_BACKOFF_MAX_SECONDS, + ) + stop_event.wait(wait_seconds) diff --git a/hermes_cli/cron.py b/hermes_cli/cron.py index 9bbe8976ee..0557c8f096 100644 --- a/hermes_cli/cron.py +++ b/hermes_cli/cron.py @@ -301,7 +301,9 @@ def cron_status(): if last_error: # Show WHY ticks fail — e.g. a root-rewritten jobs.json # (PermissionError) that silently locked out the ticker's - # uid for ~14h in the field (#68483). + # uid for ~14h in the field (#68483), or fd exhaustion + # (EMFILE) that used to stall the scheduler invisibly + # (#87644). print(color(f" Last tick error: {last_error}", Colors.RED)) if "Permission denied" in last_error: print(color( @@ -311,6 +313,14 @@ def cron_status(): "gateway user, and prefer `docker exec -u :`.", Colors.YELLOW, )) + elif "Too many open files" in last_error or "EMFILE" in last_error: + print(color( + " Hint: the ticker hit file-descriptor exhaustion " + "(EMFILE). The scheduler now retries with backoff and " + "attempts fd reclamation, but if the leak persists, " + "restart the gateway to recover scheduling.", + Colors.YELLOW, + )) print(" Check the gateway log for 'Cron tick error'.") else: print(color("✓ Gateway is running — cron jobs will fire automatically", Colors.GREEN)) diff --git a/tests/cron/test_cron_emfile_stall_87644.py b/tests/cron/test_cron_emfile_stall_87644.py new file mode 100644 index 0000000000..2b895b386c --- /dev/null +++ b/tests/cron/test_cron_emfile_stall_87644.py @@ -0,0 +1,221 @@ +"""Regression tests for #87644 — cron scheduler silently stalls after EMFILE. + +Three fixes under test: + +1. ``tick()`` no longer swallows a real ``OSError`` at tick-lock acquisition as + "another instance holds the lock". An EMFILE/ENFILE (fd exhaustion) on + ``open(.tick.lock)`` previously made ``tick()`` return 0 — which the ticker + loop recorded as a *successful* tick and cleared the error — so the + scheduler looked perfectly healthy (heartbeat + success markers fresh) + while no job ever ran again. Now the failure propagates to the ticker + loop, which records it and backs off. + +2. The ticker loop detects fd exhaustion (EMFILE/ENFILE/"Too many open + files"), attempts best-effort fd reclamation (``gc.collect()`` + raising + the soft nofile limit), and applies exponential backoff so an exhausted + process stops hammering the store every 60s. Once FDs free up, the next + tick succeeds and the backoff resets — the scheduler self-heals without a + gateway restart. + +3. Genuine lock contention (EWOULDBLOCK/EAGAIN on flock) still skips the tick + silently — that behavior is preserved. +""" +import errno +import threading +import time +from unittest.mock import patch + +import pytest + +import cron.scheduler as scheduler_mod + +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 _wait_until(predicate, timeout=10.0, interval=0.005): + """Block until ``predicate()`` is truthy or ``timeout`` elapses.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + value = predicate() + if value: + return value + time.sleep(interval) + return predicate() + + +# ── Fix 1: tick() must not swallow EMFILE as "another instance holds the lock" ─ + + +class TestTickLockEmfileNotSwallowed: + def test_lock_open_emfile_raises_instead_of_silent_skip(self, monkeypatch): + """open(.tick.lock) raising EMFILE must propagate — not return 0 as if + another ticker held the lock (the pre-fix silent stall).""" + real_open = open + calls = [] + + def flaky_open(path, *args, **kwargs): + if str(path).endswith(".tick.lock"): + calls.append(("emfile", str(path))) + raise OSError(errno.EMFILE, "Too many open files") + return real_open(path, *args, **kwargs) + + monkeypatch.setattr("builtins.open", flaky_open) + with pytest.raises(OSError) as excinfo: + scheduler_mod.tick(verbose=False) + assert excinfo.value.errno == errno.EMFILE + assert calls, "the lock file open must have been attempted" + + def test_lock_flock_emfile_raises(self, monkeypatch): + """flock() raising EMFILE (fd exhaustion at lock syscall) must also + propagate — only EWOULDBLOCK/EAGAIN/EACCES mean contention.""" + with patch.object(fcntl, "flock", side_effect=OSError(errno.EMFILE, "Too many open files")): + with pytest.raises(OSError) as excinfo: + scheduler_mod.tick(verbose=False) + assert excinfo.value.errno == errno.EMFILE + + def test_lock_contention_ewouldblock_still_skips_silently(self, monkeypatch): + """Genuine contention (another ticker holds the lock) still returns 0 + silently — the cross-process mutual-exclusion contract is preserved.""" + with patch.object(fcntl, "flock", side_effect=OSError(errno.EWOULDBLOCK, "Resource temporarily unavailable")): + assert scheduler_mod.tick(verbose=False) == 0 + + def test_lock_contention_eagain_still_skips(self): + with patch.object(fcntl, "flock", side_effect=OSError(errno.EAGAIN, "Resource temporarily unavailable")): + assert scheduler_mod.tick(verbose=False) == 0 + + +# ── Fix 2: ticker loop survives EMFILE, reclaims fds, backs off, self-heals ── + + +class TestTickerEmfileBackoff: + def test_emfile_tick_records_error_and_keeps_looping(self, monkeypatch): + """An EMFILE tick must be recorded as a FAILED tick (not success) and + the loop must keep running — the pre-fix state where the scheduler + stalled silently while the heartbeat kept reporting healthy.""" + from cron.scheduler_provider import InProcessCronScheduler + + beats = [] + errors = [] + stop = threading.Event() + prov = InProcessCronScheduler() + + with patch( + "cron.scheduler.tick", + side_effect=OSError(errno.EMFILE, "Too many open files"), + ), patch( + "cron.jobs.record_ticker_heartbeat", + side_effect=lambda success=False: beats.append(success), + ), patch( + "cron.jobs.record_ticker_error", + side_effect=lambda msg: errors.append(msg), + ), patch("cron.jobs.clear_ticker_error") as clear: + t = threading.Thread(target=prov.start, args=(stop,), kwargs={"interval": 0}, daemon=True) + t.start() + assert _wait_until(lambda: len(beats) >= 3), "ticker did not keep beating" + stop.set() + t.join(timeout=5) + + assert not t.is_alive(), "ticker must survive EMFILE ticks" + assert errors, "ticker error must be persisted so `hermes cron status` can show it" + assert "Too many open files" in errors[0] + # Every post-failure beat must be success=False: liveness yes, success no. + assert beats[-1] is False + clear.assert_not_called() + + def test_emfile_triggers_fd_reclamation(self, monkeypatch): + """The ticker must attempt best-effort fd reclamation on EMFILE so the + next tick can succeed once descriptors free up.""" + from cron.scheduler_provider import InProcessCronScheduler + + reclaimed = [] + stop = threading.Event() + prov = InProcessCronScheduler() + + with patch( + "cron.scheduler.tick", + side_effect=OSError(errno.EMFILE, "Too many open files"), + ), patch( + "cron.scheduler._reclaim_fds_best_effort", + side_effect=lambda: reclaimed.append(1), + ), patch("cron.jobs.record_ticker_heartbeat"), patch("cron.jobs.record_ticker_error"): + t = threading.Thread(target=prov.start, args=(stop,), kwargs={"interval": 0}, daemon=True) + t.start() + assert _wait_until(lambda: len(reclaimed) >= 1), "fd reclamation never ran" + stop.set() + t.join(timeout=5) + + assert len(reclaimed) >= 1 + + def test_ticker_recovers_after_emfile_clears(self, monkeypatch): + """Once ticks stop raising EMFILE, the next tick succeeds, the error is + cleared and the success marker is bumped again — self-healing without a + gateway restart.""" + from cron.scheduler_provider import InProcessCronScheduler + + attempts = {"n": 0} + beats = [] + errors = [] + stop = threading.Event() + prov = InProcessCronScheduler() + + def flaky_tick(*args, **kwargs): + attempts["n"] += 1 + if attempts["n"] <= 2: + raise OSError(errno.EMFILE, "Too many open files") + return 0 + + with patch("cron.scheduler.tick", side_effect=flaky_tick), patch( + "cron.jobs.record_ticker_heartbeat", + side_effect=lambda success=False: beats.append(success), + ), patch( + "cron.jobs.record_ticker_error", + side_effect=lambda msg: errors.append(msg), + ), patch("cron.jobs.clear_ticker_error") as clear, patch( + "cron.scheduler._reclaim_fds_best_effort" + ): + t = threading.Thread(target=prov.start, args=(stop,), kwargs={"interval": 0}, daemon=True) + t.start() + # Wait until we've seen at least one success beat AFTER the failures. + assert _wait_until(lambda: len(beats) >= 4), "ticker never recovered" + stop.set() + t.join(timeout=5) + + assert not t.is_alive() + assert attempts["n"] >= 3 + assert errors, "failures must be recorded" + assert True in beats, "a successful tick must bump the success marker" + clear.assert_called(), "a successful tick must clear the recorded error" + + +# ── Helpers ──────────────────────────────────────────────────────────────────── + + +class TestEmfileHelpers: + def test_is_fd_exhaustion_errno(self): + assert scheduler_mod._is_fd_exhaustion(OSError(errno.EMFILE, "Too many open files")) + assert scheduler_mod._is_fd_exhaustion(OSError(errno.ENFILE, "file table overflow")) + assert not scheduler_mod._is_fd_exhaustion(OSError(errno.EACCES, "Permission denied")) + assert not scheduler_mod._is_fd_exhaustion(OSError(errno.EWOULDBLOCK, "Resource temporarily unavailable")) + + def test_is_fd_exhaustion_wrapped_message(self): + """load_jobs wraps the raw OSError in a RuntimeError whose message + carries the errno text — the ticker must classify that too.""" + assert scheduler_mod._is_fd_exhaustion( + RuntimeError("Failed to read cron database: [Errno 24] Too many open files") + ) + + def test_reclaim_fds_best_effort_never_raises(self): + # gc.collect + apply_nofile_soft_limit are both best-effort; the helper + # must tolerate missing/refusing platforms without raising. + assert scheduler_mod._reclaim_fds_best_effort() is None + + def test_lock_contention_errno_classification(self): + assert scheduler_mod._is_lock_contention_errno(OSError(errno.EWOULDBLOCK, "x")) + assert scheduler_mod._is_lock_contention_errno(OSError(errno.EAGAIN, "x")) + assert not scheduler_mod._is_lock_contention_errno(OSError(errno.EMFILE, "x")) + assert not scheduler_mod._is_lock_contention_errno(OSError(errno.ENFILE, "x"))