fix(cron): settle the execution receipt when dispatch fails after creating it

_submit_with_guard persists a `claimed` receipt before the worker runs, and
run_one_job only owns the terminal writes once it starts. Two exits in
between left the row `claimed` forever (a permanently in-flight attempt
for a run that never happened): a fire claim raising anything other than
OSError in _process_due_job, and a failure in _submit_with_guard after
create_execution returned (note_cron_execution, context copy). Both now
close the receipt through a best-effort settle_unstarted_execution in
cron/executions.py (the ledger owner; scheduler.py is a capped facade); the
claim failure still re-raises so the real error surfaces.

Co-authored-by: PRELVAS94 <86016656+PRELVAS94@users.noreply.github.com>
This commit is contained in:
kshitijk4poor
2026-10-06 09:37:49 +05:30
committed by kshitij
parent f8639aed98
commit 5fff7380ab
3 changed files with 44 additions and 1 deletions

View File

@@ -7,6 +7,7 @@ to match the claim-time fingerprint is not proof of death. Terminal states are i
from __future__ import annotations
import logging
import math
import os
import sqlite3
@@ -23,6 +24,8 @@ from hermes_time import now as _hermes_now
from cron.constants import CLAIM_TTL_INACTIVITY_HEADROOM
from hermes_cli.observability.shared_metrics_gateway import record_cron_finish
logger = logging.getLogger(__name__)
# Optional test override. Production resolves the path at transaction time so dashboard operations
# that temporarily enter another profile cannot leak that profile's records into the import-time
# home.
@@ -319,6 +322,16 @@ _OWNER_WEDGED_REASON = (
)
def settle_unstarted_execution(execution_id: str, job_id: str, error: str) -> None:
"""Close the receipt of a run that never started: a ``claimed`` row never resolves. Best-effort
so a ledger write cannot mask the failure the caller logs or re-raises."""
try:
finish_execution(execution_id, success=False, error=error)
except (sqlite3.Error, OSError) as record_err:
logger.error("Job '%s': failed to close execution receipt %s (%s): %s",
job_id, execution_id, error, record_err)
def recover_interrupted_executions() -> int:
"""Mark abandoned attempts unknown without scheduling retries: rows whose owner is provably
dead, plus rows whose live owner holds a claim older than the derived stale bound (the

View File

@@ -519,7 +519,7 @@ from cron.execution_identity import enter_cron_execution, exit_cron_execution
from cron.executions import (
_TERMINAL_STATES, HANDOFF_ADOPTION_GRACE_SECONDS, create_execution, finish_execution,
get_execution, mark_execution_handoff_pending, mark_execution_running,
recover_interrupted_executions, terminalize_dead_owner)
recover_interrupted_executions, settle_unstarted_execution, terminalize_dead_owner)
# Response marker that suppresses delivery (output is still saved locally for audit).
SILENT_MARKER = "[SILENT]"
@@ -4269,6 +4269,10 @@ def _process_due_job(job: dict, adapters, loop, verbose: bool) -> bool:
finish_execution(
job["execution_id"], success=False, error=f"Cron store unwritable; not started: {exc}")
return False
except BaseException as exc: # settle first, then surface the real error
settle_unstarted_execution(
job["execution_id"], job["id"], f"Fire claim failed: {type(exc).__name__}: {exc}")
raise
if not claimed:
finish_execution(
job["execution_id"], success=False, error="Fire claim lost; execution was not started.")
@@ -4334,6 +4338,7 @@ def _submit_with_guard(job: dict, pool: concurrent.futures.ThreadPoolExecutor, p
# discard the LAUNCH home's key and leak every secondary profile's claim.
_claim_home = _get_hermes_home()
# Record the attempt before dispatch; recovery marks abandoned rows unknown (no retry).
execution = None
try:
execution = create_execution(
job_id, source="builtin", scheduled_instant=job.get("_scheduled_instant"))
@@ -4344,6 +4349,9 @@ def _submit_with_guard(job: dict, pool: concurrent.futures.ThreadPoolExecutor, p
# Release the claim so the next tick retries instead of wedging "already running".
release_running_job(job_id, home=_claim_home, owner=registration_owner)
_clear_run_claim_best_effort()
if execution is not None: # the receipt was persisted; a later step failed
settle_unstarted_execution(execution["id"], job_id, (
f"Dispatch preparation failed: {type(execution_err).__name__}: {execution_err}"))
logger.exception(
"Job '%s' not dispatched: execution creation failed: %s", job_label, execution_err)
return None

View File

@@ -126,3 +126,25 @@ def test_tick_on_unwritable_store_returns_cleanly_without_dispatch(cron_store, m
assert row["status"] == "failed" and "Cron store unwritable" in row["error"]
else:
assert row is None
@pytest.mark.parametrize("site", ["claim_job_for_fire", "note_cron_execution"])
def test_dispatch_failure_after_receipt_never_leaves_it_claimed(cron_store, monkeypatch, site):
"""A non-OSError fire-claim failure, or a failure after create_execution in _submit_with_guard,
must settle the receipt: a ``claimed`` row never resolves and reads forever as in flight."""
from cron import executions, scheduler
def boom(*_a, **_k):
raise RuntimeError("boom")
save_jobs([_due_job()])
monkeypatch.setattr(executions, "EXECUTIONS_FILE", cron_store / "cron" / "executions.db")
monkeypatch.setattr(scheduler, "run_one_job", lambda job, **k: True)
monkeypatch.setattr(scheduler, "_should_yield_tick_to_fresh_gateway", lambda: None)
monkeypatch.setattr(scheduler, site, boom)
try:
scheduler.tick(verbose=False, sync=True)
except RuntimeError:
pass # a claim failure may still surface; the receipt is what must be terminal
row = executions.latest_execution("due-job")
assert row is not None and row["status"] == "failed"