Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
128 changes: 95 additions & 33 deletions cron/scheduler_provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -664,6 +664,11 @@ def _start_multiplex(
agent execution to that profile's home — mirroring how
``_profile_runtime_scope`` scopes the multiplexed inbound path and
``web_server.py`` scopes per-profile cron API calls.

Failures are isolated per profile, at startup and in steady state:
neither a recovery error nor a tick that raises for one profile skips
the profiles after it in ``profile_homes``, prevents the tick loop from
starting, or marks the other profiles as failing in their heartbeats.
"""
import logging
from cron.scheduler import tick as cron_tick
Expand All @@ -682,7 +687,18 @@ def _start_multiplex(
[p[0] if isinstance(p, tuple) else p for p in profile_homes],
)

# Recovery + initial heartbeat for every profile.
# Recovery + initial heartbeat for every profile, isolated per profile
# for the same reason the tick loop below is — and with a wider blast
# radius if it is not: this pass runs BEFORE the loop is entered.
# recover_interrupted() opens a transaction against that profile's own
# ledger (cron/executions.py::recover_interrupted_executions), and
# neither it nor _transaction() has a catch-all, so an unreadable or
# locked store raises straight out of start(). The gateway runs
# start() as a bare thread target (gateway/run.py::start_gateway) with
# nothing above it but threading's excepthook, so that kills the ticker
# thread outright: the loop below never runs a single cycle and EVERY
# profile stops firing — including the ones whose recovery succeeded —
# until the gateway is restarted.
# A profile may have been deleted since this snapshot was taken;
# never recreate a deleted home's cron workspace via the heartbeat
# below (#47368).
Expand All @@ -699,41 +715,87 @@ def _start_multiplex(
home,
)
record_ticker_heartbeat()
except BaseException as e:
# BaseException for the same reason as the tick loop (#32612).
# The profile stays in profile_homes: a ledger that cannot be
# recovered may still tick, and if it cannot, the per-profile
# handler below records that on its own terms.
logger.error(
"Cron startup recovery failed (profile at %s); continuing "
"with the remaining profiles: %s",
home, e, exc_info=True,
)
try:
with use_cron_store(home):
record_ticker_error(f"{type(e).__name__}: {e}")
except BaseException:
# Best-effort: the store is exactly what may be
# unreachable, and letting the report of a failure raise
# would reintroduce the failure it is reporting.
logger.debug(
"Could not record startup recovery error for profile at %s",
home, exc_info=True,
)
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")
else:
for entry in _existing_profile_homes(profile_homes):
home = entry[1] if isinstance(entry, tuple) else entry
home_token = set_hermes_home_override(str(home))
try:
with use_cron_store(home):
cron_tick(
verbose=False,
adapters=adapters,
loop=loop,
sync=False,
can_dispatch=can_dispatch,
)
finally:
reset_hermes_home_override(home_token)
ok = True
except BaseException as e:
logger.error("Cron tick error: %s", e, exc_info=True)
_tick_error = f"{type(e).__name__}: {e}"
# EMFILE: reclaim fds + exponential backoff (#87644).
consecutive_failures = _note_tick_failure(e, consecutive_failures)
# One outcome per profile for this cycle: (entry, ok, error|None).
# Keeping the result per profile is what stops a failure in one
# store from being attributed to — or blocking — the others.
# Deleted homes are filtered out via _existing_profile_homes so a
# removed profile's cron workspace is never recreated (#47368).
outcomes = []
if can_dispatch is not None and not can_dispatch():
logger.debug("Cron dispatch paused while gateway drains existing work")
# Nothing was attempted this cycle; treat it as clean, matching
# the single-profile path above.
outcomes = [(entry, True, None) for entry in _existing_profile_homes(profile_homes)]
else:
_tick_error = None
# Record per-profile heartbeat after each tick cycle.
for entry in _existing_profile_homes(profile_homes):
for entry in _existing_profile_homes(profile_homes):
home = entry[1] if isinstance(entry, tuple) else entry
home_token = set_hermes_home_override(str(home))
try:
with use_cron_store(home):
cron_tick(
verbose=False,
adapters=adapters,
loop=loop,
sync=False,
can_dispatch=can_dispatch,
)
outcomes.append((entry, True, None))
except BaseException as e:
# Isolate per profile. tick() has no catch-all of its
# own — its body is try/finally — so anything raised
# inside it lands here; without this the remaining
# profiles in profile_homes are skipped for the cycle,
# and a persistent per-profile failure (e.g. a
# root-rewritten jobs.json) starves them indefinitely.
# BaseException for the same reason the single-profile
# path catches it (#32612): a SystemExit out of a
# provider SDK must not kill the ticker thread.
logger.error(
"Cron tick error (profile at %s): %s",
home, e, exc_info=True,
)
outcomes.append((entry, False, f"{type(e).__name__}: {e}"))
# EMFILE: reclaim fds + exponential backoff (#87644).
# Per-profile integration: fd exhaustion is
# process-wide, but attribution stays per profile —
# the failing profile carries the error record, and
# the counter still feeds the shared backoff wait.
consecutive_failures = _note_tick_failure(
e, consecutive_failures
)
finally:
reset_hermes_home_override(home_token)

# Record each profile's OWN result, so `hermes cron status` cannot
# report a healthy profile as failing because a different profile
# raised.
for entry, ok, tick_error in outcomes:
home = entry[1] if isinstance(entry, tuple) else entry
home_token = set_hermes_home_override(str(home))
try:
Expand All @@ -744,10 +806,10 @@ def _start_multiplex(
# (#68483).
if ok:
clear_ticker_error()
elif _tick_error:
record_ticker_error(_tick_error)
elif tick_error:
record_ticker_error(tick_error)
finally:
reset_hermes_home_override(home_token)
if ok:
if all(ok for _, ok, _ in outcomes):
consecutive_failures = 0
stop_event.wait(_backoff_wait_seconds(interval, consecutive_failures))
184 changes: 184 additions & 0 deletions tests/cron/test_scheduler_provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -691,3 +691,187 @@ def test_existing_profile_homes_filters_deleted(tmp_path):

as_paths = _existing_profile_homes([live, deleted])
assert [p for p in as_paths] == [live]


def test_multiplex_tick_error_does_not_stop_other_profiles(tmp_path, monkeypatch):
"""A tick that raises for one profile must not skip the profiles after it.

The per-profile loop used to sit inside a single try/except for the whole
cycle, so the first store that raised aborted the rest of the cycle — and a
persistent failure (e.g. a root-rewritten jobs.json the ticker's uid can no
longer lock) starved every profile ordered after it indefinitely.

Deleted homes are excluded up front by _existing_profile_homes (#47368), so
this test only exercises live profiles.
"""
from cron.scheduler_provider import InProcessCronScheduler
from cron import jobs as cron_jobs

profile_homes = []
for name in ("alpha", "beta", "gamma"):
d = tmp_path / name
(d / "cron").mkdir(parents=True)
profile_homes.append((name, d))

ticked: list[str] = []

def _tick(*args, **kwargs):
who = cron_jobs._current_cron_store().cron_dir.parent.name
if who == "alpha":
raise RuntimeError("boom in alpha")
ticked.append(who)
return 0

stop = threading.Event()
prov = InProcessCronScheduler()

with patch("cron.scheduler.tick", side_effect=_tick), \
patch("cron.jobs.record_ticker_heartbeat", lambda *a, **kw: None), \
patch("cron.jobs.record_ticker_error", lambda *a, **kw: None), \
patch("cron.jobs.clear_ticker_error", lambda *a, **kw: None):
t = threading.Thread(
target=prov.start,
args=(stop,),
kwargs={"interval": 0, "profile_homes": profile_homes},
daemon=True,
)
t.start()
deadline = time.monotonic() + 10
while not {"beta", "gamma"}.issubset(set(ticked)) and time.monotonic() < deadline:
time.sleep(0.005)
stop.set()
t.join(timeout=5)

assert not t.is_alive()
assert {"beta", "gamma"}.issubset(set(ticked)), (
"profiles ordered after the failing one were never ticked; "
f"ticked={sorted(set(ticked))}"
)


def test_multiplex_heartbeat_success_is_per_profile(tmp_path, monkeypatch):
"""Each profile's heartbeat carries that profile's own tick result.

A single cycle-wide success flag marked every profile as failing the moment
any one of them raised, so `hermes cron status` reported profiles as broken
that had ticked cleanly.
"""
from cron.scheduler_provider import InProcessCronScheduler
from cron import jobs as cron_jobs

profile_homes = []
for name in ("alpha", "beta"):
d = tmp_path / name
(d / "cron").mkdir(parents=True)
profile_homes.append((name, d))

beats: list[tuple[str, bool]] = []

def _tick(*args, **kwargs):
if cron_jobs._current_cron_store().cron_dir.parent.name == "alpha":
raise RuntimeError("boom in alpha")
return 0

def _beat(*args, **kwargs):
# The startup liveness beat is called with no arguments; only the
# per-cycle beats carry a result, and those are the ones under test.
if "success" not in kwargs:
return
who = cron_jobs._current_cron_store().cron_dir.parent.name
beats.append((who, kwargs["success"]))

stop = threading.Event()
prov = InProcessCronScheduler()

with patch("cron.scheduler.tick", side_effect=_tick), \
patch("cron.jobs.record_ticker_heartbeat", _beat), \
patch("cron.jobs.record_ticker_error", lambda *a, **kw: None), \
patch("cron.jobs.clear_ticker_error", lambda *a, **kw: None):
t = threading.Thread(
target=prov.start,
args=(stop,),
kwargs={"interval": 0, "profile_homes": profile_homes},
daemon=True,
)
t.start()
deadline = time.monotonic() + 10
while not {("alpha", False), ("beta", True)}.issubset(set(beats)) \
and time.monotonic() < deadline:
time.sleep(0.005)
stop.set()
t.join(timeout=5)

assert not t.is_alive()
assert ("beta", True) in beats, f"clean profile not recorded as healthy: {beats}"
assert ("alpha", False) in beats, f"raising profile not recorded as failing: {beats}"
assert ("beta", False) not in beats, (
f"clean profile marked failing because another profile raised: {beats}"
)


def test_multiplex_startup_recovery_error_does_not_stop_other_profiles(tmp_path):
"""A recovery failure at startup must not stop any profile from ticking.

The recovery pass runs BEFORE the tick loop is entered, so an unhandled
raise there does not merely skip the profiles after it — it propagates out
of ``start()``, which the gateway runs as a bare thread target
(``gateway/run.py::start_gateway``). The ticker thread dies before the loop
runs a single cycle, so every profile stops firing, including the ones
whose own recovery succeeded.

``recover_interrupted_executions()`` opens a SQLite transaction against the
profile's own ledger and has no catch-all, so an unreadable or locked store
is enough to trigger this.
"""
import sqlite3

from cron.scheduler_provider import InProcessCronScheduler
from cron import jobs as cron_jobs

profile_homes = []
for name in ("alpha", "beta", "gamma"):
d = tmp_path / name
(d / "cron").mkdir(parents=True)
profile_homes.append((name, d))

def _current_profile():
return cron_jobs._current_cron_store().cron_dir.parent.name

def _recover(self):
if _current_profile() == "alpha":
raise sqlite3.OperationalError("unable to open database file")
return 0

ticked: list[str] = []

def _tick(*args, **kwargs):
ticked.append(_current_profile())
return 0

stop = threading.Event()
prov = InProcessCronScheduler()

with patch.object(InProcessCronScheduler, "recover_interrupted", _recover), \
patch("cron.scheduler.tick", side_effect=_tick), \
patch("cron.jobs.record_ticker_heartbeat", lambda *a, **kw: None), \
patch("cron.jobs.record_ticker_error", lambda *a, **kw: None), \
patch("cron.jobs.clear_ticker_error", lambda *a, **kw: None):
t = threading.Thread(
target=prov.start,
args=(stop,),
kwargs={"interval": 0, "profile_homes": profile_homes},
daemon=True,
)
t.start()
_wait_until(lambda: {"alpha", "beta", "gamma"}.issubset(set(ticked)))
stop.set()
t.join(timeout=5)

assert not t.is_alive()
# gamma is ordered after the failing profile; alpha is the failing profile
# itself, which stays in the rotation because a ledger that cannot be
# recovered may still hold tickable jobs.
assert {"alpha", "beta", "gamma"}.issubset(set(ticked)), (
"a startup recovery failure stopped the tick loop; "
f"ticked={sorted(set(ticked))}"
)
Loading