diff --git a/cron/scheduler.py b/cron/scheduler.py index b7edf52882..ed3a1cf86d 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -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 diff --git a/tests/cron/test_inflight_stale_guard.py b/tests/cron/test_inflight_stale_guard.py index 7787da685d..08e4f2e1c1 100644 --- a/tests/cron/test_inflight_stale_guard.py +++ b/tests/cron/test_inflight_stale_guard.py @@ -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