diff --git a/src/echo_certification_forge/subscriber.py b/src/echo_certification_forge/subscriber.py index 3c69fb4..3bba7a4 100644 --- a/src/echo_certification_forge/subscriber.py +++ b/src/echo_certification_forge/subscriber.py @@ -3955,6 +3955,61 @@ def bind_run( outcome="allowed", ) + def _terminalize_unstarted_run( + self, + connection: sqlite3.Connection, + *, + run_id: str, + organization_id: str, + idempotency_key: str, + metered_at: str, + now_text: str, + actor: str, + reason: str, + ) -> None: + """Finish a QUEUED run whose dispatch gave up before execution started (fleet #49604). + + Same outcome as lease-expiry exhaustion in _recover_expired_dispatches: the run becomes + INFRASTRUCTURE_FAILURE / INFRA_FAILED, its BOUND reservation is released and the metered + run is compensated. The compensation uses the same 'dispatch-crash' key, so a run is + never compensated twice whichever path reaches it first. + """ + self._transition_run_with_connection( + connection, + run_id, + organization_id, + "QUEUED", + "INFRASTRUCTURE_FAILURE", + actor, + reason, + "p7.subscriber", + now_text, + ) + connection.execute( + """ + UPDATE runs SET run_outcome = 'INFRA_FAILED', updated_at = ? + WHERE run_id = ? AND tenant_id = ? + """, + (now_text, run_id, organization_id), + ) + connection.execute( + """ + UPDATE subscriber_run_reservations + SET state = 'RELEASED', updated_at = ? + WHERE run_id = ? AND organization_id = ? AND state = 'BOUND' + """, + (now_text, run_id, organization_id), + ) + self._compensate_crash_meter( + connection, + organization_id, + idempotency_key, + run_id, + metered_at, + key_prefix="dispatch-crash", + reason="dispatch_attempts_exhausted_before_execution_start", + ) + def _recover_expired_dispatches( self, connection: sqlite3.Connection, now: datetime ) -> int: @@ -4094,7 +4149,61 @@ def _recover_expired_dispatches( "meter_compensated": next_state == "FAILED", }, ) - return terminalized + len(rows) + # Runs orphaned before fleet #49604: dispatch FAILED, run still QUEUED, reservation BOUND. + # Once terminalized they no longer match, so this pass is idempotent. + orphans = connection.execute( + """ + SELECT d.run_id, d.organization_id, d.idempotency_key, d.attempts, d.last_error, + COALESCE( + ( + SELECT u.occurred_at + FROM subscriber_usage_events u + WHERE u.organization_id = d.organization_id + AND u.meter = 'certification_runs' + AND u.idempotency_key = d.idempotency_key + AND u.quantity > 0 + LIMIT 1 + ), + d.created_at + ) AS metered_at + FROM subscriber_run_dispatches AS d + JOIN subscriber_run_reservations AS r + ON r.run_id = d.run_id + AND r.organization_id = d.organization_id + JOIN runs + ON runs.run_id = d.run_id + AND runs.tenant_id = d.organization_id + WHERE d.state = 'FAILED' + AND r.state = 'BOUND' + AND runs.state = 'QUEUED' + """ + ).fetchall() + for row in orphans: + self._terminalize_unstarted_run( + connection, + run_id=str(row["run_id"]), + organization_id=str(row["organization_id"]), + idempotency_key=str(row["idempotency_key"]), + metered_at=str(row["metered_at"]), + now_text=now_text, + actor="certforge.dispatch-recovery", + reason="dispatch_failed_run_orphaned", + ) + self._append_audit( + connection, + organization_id=str(row["organization_id"]), + actor_ref="certforge.dispatch-recovery", + action="certification.dispatch_orphan_reconciled", + resource_type="certification", + resource_id=str(row["run_id"]), + outcome="allowed", + details={ + "attempts": int(row["attempts"]), + "last_error": row["last_error"], + "meter_compensated": True, + }, + ) + return terminalized + len(rows) + len(orphans) def claim_dispatch( self, @@ -4291,6 +4400,21 @@ def finish_dispatch( state = "FAILED" elif error is not None and int(row["attempts"]) < self.policy.max_dispatch_attempts: state = "PENDING" + elif error is not None and run_state == "QUEUED" and row["reservation_state"] == "BOUND": + # Last attempt failed before execution started (for example + # subscriber_worker_claim_denied). Marking only the dispatch FAILED left the run + # QUEUED forever and uncompensated (fleet #49604). + self._terminalize_unstarted_run( + connection, + run_id=dispatch.run_id, + organization_id=dispatch.organization_id, + idempotency_key=str(row["idempotency_key"]), + metered_at=str(row["metered_at"]), + now_text=now, + actor="certforge.dispatcher", + reason="dispatch_attempts_exhausted", + ) + state = "FAILED" else: state = "FAILED" connection.execute( diff --git a/tests/test_dispatch_orphaned_runs.py b/tests/test_dispatch_orphaned_runs.py new file mode 100644 index 0000000..a80578e --- /dev/null +++ b/tests/test_dispatch_orphaned_runs.py @@ -0,0 +1,117 @@ +"""A dispatch that gives up must not leave its run QUEUED forever (Echo fleet #49604). + +On FORGE (2026-11-03) seven runs from 2026-10-09..10-30 were QUEUED while their dispatch rows were +FAILED after 3 attempts with `subscriber_worker_claim_denied`, their reservations still BOUND and +execution never started. finish_dispatch() marked only the dispatch FAILED, so the runs never +reached a terminal state, the customer was never compensated for the metered run, and every +"is the queue idle?" check (for example the swarm re-certification) stayed blocked. + +The lease-expiry path already does the right thing (INFRASTRUCTURE_FAILURE, run_outcome INFRA_FAILED, +reservation RELEASED, meter compensated). These tests pin the same outcome for finish_dispatch, and +for runs already orphaned by the old code. +""" +from __future__ import annotations + +import sqlite3 +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from echo_certification_forge.models import RunState # noqa: E402 +from echo_certification_forge.subscriber import Permission # noqa: E402 +from test_p7_subscriber_governance import ( # noqa: E402 + _owner, + _policy, + _provision, + _stack, + _submit_local_run, +) + + +def _run(tmp_path, manifest, name, *, max_attempts): + policy = _policy().model_copy(update={"max_dispatch_attempts": max_attempts}) + store, governance, client = _stack(tmp_path, manifest, policy=policy) + org = _provision(governance, name) + source = tmp_path / f"{name}-source" + source.mkdir() + (source / "hello.py").write_text("print('dispatch')\n", encoding="utf-8") + run_id, _ = _submit_local_run(client, manifest, org.organization_id, org.bootstrap_api_key, source, + key=f"{name}-0001") + return store, governance, org, run_id + + +def _usage(governance, org): + return governance.usage_summary(_owner(governance, org.organization_id, org.bootstrap_api_key, + Permission.USAGE_READ)) + + +def test_worker_error_on_the_last_attempt_terminalizes_and_compensates(tmp_path, manifest) -> None: + store, governance, org, run_id = _run(tmp_path, manifest, "exhausted-claim-denied", max_attempts=1) + dispatch = governance.claim_dispatch("dispatcher-a", run_id=run_id, organization_id=org.organization_id) + assert dispatch is not None + + assert governance.finish_dispatch(dispatch, error="subscriber_worker_claim_denied") == "FAILED" + + run = store.get_run(run_id, org.organization_id) + assert run["state"] == RunState.INFRASTRUCTURE_FAILURE.value + assert run["run_outcome"] == "INFRA_FAILED" + assert governance.dispatch_status(run_id, org.organization_id)["state"] == "FAILED" + usage = _usage(governance, org) + assert usage["meters"]["certification_runs"] == 0 + assert usage["active_run_reservations"] == 0 + + +def test_worker_error_with_attempts_left_still_retries(tmp_path, manifest) -> None: + store, governance, org, run_id = _run(tmp_path, manifest, "retry-left", max_attempts=3) + dispatch = governance.claim_dispatch("dispatcher-a", run_id=run_id, organization_id=org.organization_id) + + assert governance.finish_dispatch(dispatch, error="subscriber_worker_claim_denied") == "PENDING" + + assert store.get_run(run_id, org.organization_id)["state"] == RunState.QUEUED.value + assert _usage(governance, org)["meters"]["certification_runs"] == 1 + + +def _orphan_the_old_way(governance, org, run_id) -> None: + """Reproduce what the old finish_dispatch left behind: dispatch FAILED, run QUEUED, reservation BOUND.""" + dispatch = governance.claim_dispatch("dispatcher-a", run_id=run_id, organization_id=org.organization_id) + assert dispatch is not None + with sqlite3.connect(governance.db_path) as connection: + connection.execute( + """ + UPDATE subscriber_run_dispatches + SET state = 'FAILED', attempts = 3, claim_token = NULL, lease_owner = NULL, + lease_expires_at = NULL, last_error = 'subscriber_worker_claim_denied' + WHERE run_id = ? AND organization_id = ? + """, + (run_id, org.organization_id), + ) + connection.commit() + + +def test_runs_orphaned_by_the_old_code_are_reconciled_once(tmp_path, manifest) -> None: + store, governance, org, run_id = _run(tmp_path, manifest, "orphan-reconcile", max_attempts=3) + _orphan_the_old_way(governance, org, run_id) + assert store.get_run(run_id, org.organization_id)["state"] == RunState.QUEUED.value + assert _usage(governance, org)["meters"]["certification_runs"] == 1 + + assert governance.claim_dispatch("dispatcher-b", organization_id=org.organization_id) is None + + run = store.get_run(run_id, org.organization_id) + assert run["state"] == RunState.INFRASTRUCTURE_FAILURE.value + assert run["run_outcome"] == "INFRA_FAILED" + usage = _usage(governance, org) + assert usage["meters"]["certification_runs"] == 0 + assert usage["active_run_reservations"] == 0 + # Idempotent: a later pass neither re-transitions nor compensates twice. + assert governance.claim_dispatch("dispatcher-c", organization_id=org.organization_id) is None + assert _usage(governance, org)["meters"]["certification_runs"] == 0 + + +def test_a_failed_dispatch_whose_run_already_finished_is_left_alone(tmp_path, manifest) -> None: + store, governance, org, run_id = _run(tmp_path, manifest, "orphan-terminal", max_attempts=1) + dispatch = governance.claim_dispatch("dispatcher-a", run_id=run_id, organization_id=org.organization_id) + governance.finish_dispatch(dispatch, error="subscriber_worker_claim_denied") + before = _usage(governance, org)["meters"]["certification_runs"] + assert governance.claim_dispatch("dispatcher-b", organization_id=org.organization_id) is None + assert _usage(governance, org)["meters"]["certification_runs"] == before == 0