diff --git a/cron/executions.py b/cron/executions.py index b6afa0d379..32a5ce4728 100644 --- a/cron/executions.py +++ b/cron/executions.py @@ -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 diff --git a/cron/scheduler.py b/cron/scheduler.py index 99669e60ea..1e204111e5 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -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 diff --git a/tests/cron/test_due_scan_save_failure.py b/tests/cron/test_due_scan_save_failure.py index 9f2a87bff2..90193bfa80 100644 --- a/tests/cron/test_due_scan_save_failure.py +++ b/tests/cron/test_due_scan_save_failure.py @@ -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"