ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

fix(cron): reconcile stale in-flight claim against executions ledger (t_8b5480b3)

Browse Source 
The age-only stale-claim sweep (t_3778a491, already on main) force-releases
an in-memory _running_job_ids claim only once it is older than
max(2*interval, 30m). A leaked claim that is YOUNG (inside its allowance)
while the durable executions ledger already proves the last run ended stays
wedged: the job is returned as due every tick, _submit_with_guard short-
circuits on 'already running', and next_run_at keeps fast-forwarding with no
execution — the exact 2026-08-14 recurring-router incident (t_20e23f84),
which survived a gateway restart because the in-memory age bound alone could
not see a run the ledger had already finished.

sweep_stale_inflight now reconciles each in-flight claim against the durable
executions ledger (cron/executions.db): if the job's MOST RECENT execution
row is terminal (completed/failed/unknown), the run provably ended, so the
claim is stale by construction regardless of its in-memory age and is force-
released. This is a persisted-state recovery path: the ledger is written by
the worker that ran the job and read by ANY ticker process (including one
that started AFTER the leak), so a leaked claim is recoverable without
force-run/resume and without depending on which process holds it in memory.
A ledger-terminal release is authoritative — it does not write a synthetic
mark_job_run failure (the ledger already records the outcome).

Added TestLedgerTerminalReconciliation (4 tests): young+terminal -> released
(RED on main, GREEN here), no-ledger-row -> not released, running-row -> not
released, old+terminal -> released once without synthetic failure.
This commit is contained in:
devops
2026-08-15 21:42:29 +01:00
committed by kshitij
parent 0a8e703701
commit 9b9cfbc1fd
2 changed files with 159 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cron/scheduler.py
+55 -6
View File
@@ -855,6 +855,23 @@ def sweep_stale_inflight(due_jobs: Optional[list] = None) -> list:
now = time.time()
stale: list = []
# Latest durable execution per in-flight job id, loaded in one indexed
# query. Used for the persisted-state reconciliation below (t_8b5480b3):
# an in-memory claim whose job's MOST RECENT execution row is terminal
# (completed/failed/unknown) cannot represent a live run — the durable
# ledger proves that run already ended — so the claim is stale by
# construction, regardless of its in-memory age. This closes the
# restart-survival gap that the age-only sweep left open: the ledger is
# written by the worker that ran the job and read by ANY ticker process
# (including one that started AFTER the leak), so a leaked claim is
# recoverable without force-run/resume and without depending on which
# process happens to still hold it in memory.
try:
from cron.executions import latest_executions as _latest_execs
_latest = _latest_execs(list(_running_job_ids))
except Exception:
_latest = {}
# Precompute job intervals OUTSIDE _running_lock so croniter evaluation
# does not block try_register/release_running_job for other jobs.
_intervals = {jid: _job_interval_minutes(j) for jid, j in by_id.items()}
@@ -872,8 +889,6 @@ def sweep_stale_inflight(due_jobs: Optional[list] = None) -> list:
allowance = floor_seconds
if interval_minutes:
allowance = max(allowance, 2.0 * interval_minutes * 60.0)
if age < allowance:
continue
fut = _running_futures.get(job_id)
if fut is _FUTURE_PENDING:
# The claim is past its allowance and the owning future still
@@ -883,13 +898,34 @@ def sweep_stale_inflight(due_jobs: Optional[list] = None) -> list:
pass
elif fut is not None and not fut.done():
continue # genuinely still executing
# Persisted-state reconciliation: if the durable executions ledger
# shows this job's last run reached a terminal state, the claim is
# provably stale even if it is still inside its in-memory age
# allowance (or was adopted fresh this tick). Release it now so
# the job re-dispatches on the next tick without force-run/resume
# (t_8b5480b3 — the 2026-08-14 recurring-router wedge that
# survived a gateway restart because the in-memory age bound alone
# could not see a run the ledger had already finished).
if fut is None or fut is _FUTURE_PENDING or fut.done():
latest = _latest.get(job_id)
if latest is not None and latest.get("status") in (
"completed", "failed", "unknown",
):
_running_job_ids.discard(job_id)
_running_since.pop(job_id, None)
_running_futures.pop(job_id, None)
_forced_release_count += 1
stale.append((job_id, age, allowance, fut, "ledger-terminal"))
continue
if age < allowance:
continue
_running_job_ids.discard(job_id)
_running_since.pop(job_id, None)
_running_futures.pop(job_id, None)
_forced_release_count += 1
stale.append((job_id, age, allowance, fut))
stale.append((job_id, age, allowance, fut, "age"))
for job_id, age, allowance, fut in stale:
for job_id, age, allowance, fut, _reason in stale:
job = by_id.get(job_id) or {}
name = job.get("name") or job_id
if fut is _FUTURE_PENDING:
@@ -899,9 +935,10 @@ def sweep_stale_inflight(due_jobs: Optional[list] = None) -> list:
else:
future_state = "finished"
logger.warning(
"cron.inflight.forced_release event=forced_release job='%s' id=%s "
"age=%.0fs allowance=%.0fs future=%s — stale in-flight claim "
"cron.inflight.forced_release event=forced_release reason=%s job='%s' "
"id=%s age=%.0fs allowance=%.0fs future=%s — stale in-flight claim "
"released; the job was skipping every fire with 'already running'",
_reason,
name,
job_id,
age,
@@ -909,6 +946,18 @@ def sweep_stale_inflight(due_jobs: Optional[list] = None) -> list:
future_state,
)
_record_forced_release(job_id, name, age, allowance)
# A ledger-terminal release is authoritative: the durable executions
# ledger ALREADY records how the last run ended (completed/failed/
# unknown), so we must NOT call mark_job_run here — doing so would
# clobber an honest completed/ok status with a synthetic failure, or
# double-write an already-recorded failure. We only release the claim
# so the job re-dispatches on its next due tick; the ledger is the
# record of record for the outcome. The age-based release below keeps
# the original wedge-surfacing mark_job_run behaviour (an age-release
# may have no ledger row at all, so surfacing last_error is the only
# way the wedge becomes visible).
if _reason == "ledger-terminal":
continue
# Finite-repeat guard: a forced release is NOT a real run, so it must
# not consume a finite one-shot's repeat budget or let mark_job_run
# auto-delete the row (completed >= times). The claim is released and
tests/cron/test_inflight_stale_guard.py
+104
View File
@@ -348,3 +348,107 @@ class TestWedgedJobRefiresWithoutRestart:
assert n == 1, "wedged job must fire again without a gateway restart"
assert job["id"] not in sched.get_running_job_ids()
assert sched.get_inflight_guard_stats()["forced_releases"] == 1
class TestLedgerTerminalReconciliation:
"""Persisted-state recovery path (t_8b5480b3).
The age-only sweep released claims older than ``max(2 * interval, floor)``,
but a leaked claim could be YOUNG (inside its allowance) while the durable
executions ledger already proved the last run ended — e.g. the 2026-08-14
recurring-router wedge, which survived a gateway restart because the age
bound alone could not see a run the ledger had already finished. This
class tests the ledger reconciliation: an in-memory claim whose job's
MOST RECENT execution row is terminal (completed/failed/unknown) is stale
by construction and is force-released regardless of in-memory age, so the
recurring job re-dispatches on the next tick without force-run/resume.
RED first: on the age-only sweep (main before this change) a young leaked
claim with a terminal ledger row is NOT released — it stays wedged. With
the ledger reconciliation it IS released (and, because a terminal ledger
row is authoritative, WITHOUT a synthetic mark_job_run failure).
"""
def _inject_young_claim(self, job_id: str) -> None:
"""Claim is YOUNG (inside the 30m floor) so only the ledger-terminal
path, never the age path, can release it."""
sched._running_job_ids.add(job_id)
if hasattr(sched, "_running_since"):
sched._running_since[job_id] = time.time() - 60 # 1 minute old
def test_young_claim_with_terminal_ledger_row_is_released(self, tmp_path):
"""RED/GREEN: a terminal ledger row proves the run ended, so a young
leaked claim must be force-released even though its in-memory age is
inside the allowance (the age-only sweep alone would leave it)."""
job = _job(job_id="ledger-terminal", minutes=10)
job_id = job["id"]
self._inject_young_claim(job_id)
with patch.object(sched, "_get_hermes_home", return_value=tmp_path), \
patch("cron.executions.latest_executions", return_value={
job_id: {"status": "failed", "id": "exec-x"},
}), \
patch.object(sched, "mark_job_run") as mark:
released = sched.sweep_stale_inflight([job])
# RED on the age-only sweep: the young claim is NOT released. GREEN
# with the ledger reconciliation: it IS released.
assert job_id in released, "terminal ledger row must force-release the claim"
assert job_id not in sched.get_running_job_ids()
# Ledger-terminal is authoritative: no synthetic failure written.
mark.assert_not_called()
def test_young_claim_without_ledger_row_is_not_released(self, tmp_path):
"""A young claim whose job has NO execution row at all (the claim was
taken but create_execution never ran) is left to the age bound — the
ledger reconciliation must not release claims it cannot prove ended."""
job = _job(job_id="no-ledger-row", minutes=10)
job_id = job["id"]
self._inject_young_claim(job_id)
with patch.object(sched, "_get_hermes_home", return_value=tmp_path), \
patch("cron.executions.latest_executions", return_value={}), \
patch.object(sched, "mark_job_run"):
released = sched.sweep_stale_inflight([job])
assert job_id not in released
assert job_id in sched.get_running_job_ids()
def test_young_claim_with_running_ledger_row_is_not_released(self, tmp_path):
"""A young claim whose job is genuinely still running per the ledger
('claimed'/'running' row) is never released — reconciliation must not
double-dispatch a healthy long-running job."""
job = _job(job_id="still-running", minutes=10)
job_id = job["id"]
self._inject_young_claim(job_id)
with patch.object(sched, "_get_hermes_home", return_value=tmp_path), \
patch("cron.executions.latest_executions", return_value={
job_id: {"status": "running", "id": "exec-y"},
}), \
patch.object(sched, "mark_job_run"):
released = sched.sweep_stale_inflight([job])
assert job_id not in released
assert job_id in sched.get_running_job_ids()
def test_old_claim_with_terminal_ledger_row_still_released_once(self, tmp_path):
"""An old claim with a terminal ledger row is released by the ledger
path (one release) — it must not double-release or double-count."""
job = _job(job_id="old-terminal", minutes=10)
job_id = job["id"]
sched._running_job_ids.add(job_id)
if hasattr(sched, "_running_since"):
sched._running_since[job_id] = time.time() - 6 * 60 * 60 # 6h old
with patch.object(sched, "_get_hermes_home", return_value=tmp_path), \
patch("cron.executions.latest_executions", return_value={
job_id: {"status": "completed", "id": "exec-z"},
}), \
patch.object(sched, "mark_job_run") as mark:
released = sched.sweep_stale_inflight([job])
assert job_id in released
assert job_id not in sched.get_running_job_ids()
assert sched.get_inflight_guard_stats()["forced_releases"] == 1
mark.assert_not_called() # authoritative ledger row, no synthetic failure