fix(cron): scheduler self-heals after EMFILE instead of stalling silently (#87644)

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
This commit is contained in:
webtecnica
2026-08-16 12:48:26 -03:00
committed by kshitij
parent d153bfe6cb
commit 815934ae52
4 changed files with 383 additions and 8 deletions
+94 -5
View File
@@ -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
+57 -2
View File
@@ -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)
+11 -1
View File
@@ -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 <uid>:<gid>`.",
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))
+221
View File
@@ -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"))