diff --git a/cron/scheduler.py b/cron/scheduler.py index c640b2505a..864e3f6f80 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -2730,15 +2730,30 @@ def run_one_job( logger.error("Job '%s': %s", job["id"], error) claim = job.get("fire_claim") owner = str(claim.get("by") or "") if isinstance(claim, dict) else "" + # The dispatch failure is a job failure like any other: it must open an + # incident and leave through the job's failure lane (#123401). Without + # this the outage is silent — no cron_incidents row, no ping — while + # executions.db keeps piling up failed rows. + delivery_error = None + delivery_outcome = "failed" + try: + delivery_error, delivery_outcome = _deliver_crash_failure( + job, error, adapters=adapters, loop=loop) + except Exception as notice_exc: + logger.error( + "Dispatch-failure notice failed for job %s: %s", job["id"], notice_exc) try: mark_job_run( job["id"], False, error, + delivery_error=delivery_error, **({"expected_fire_owner": owner} if owner else {}), ) finally: - finish_execution(execution_id, success=False, error=error) + finish_execution( + execution_id, success=False, error=error, + delivery_outcome=delivery_outcome) return True if extra_prompt is None: # Gateway-forwarded manual run stamps its prompt on the job via trigger_job; the fire that diff --git a/tests/cron/test_restart_safe_worker.py b/tests/cron/test_restart_safe_worker.py index f800a2b39b..b931cd5c22 100644 --- a/tests/cron/test_restart_safe_worker.py +++ b/tests/cron/test_restart_safe_worker.py @@ -663,6 +663,53 @@ def test_shared_run_path_hands_gateway_fire_to_external_worker(monkeypatch): run.assert_not_called() +def test_dispatch_failure_opens_incident_and_delivers_failure_notice( + execution_ledger, monkeypatch +): + """A failed external-worker handoff must surface like any other job failure: one + incident row plus one failure-lane notice, with repeats withheld by the alerted + cooldown (#123401) — not a silent outage while executions.db piles up failed rows.""" + import cron.incidents as incidents + import cron.scheduler as scheduler + + def _handoff_boom(_job): + raise RuntimeError("worker exited before ownership acknowledgement") + + monkeypatch.setattr(scheduler, "_launch_external_cron_worker", _handoff_boom) + monkeypatch.setattr(scheduler, "mark_job_run", lambda *_a, **_k: True) + delivered = [] + monkeypatch.setattr( + scheduler, "_deliver_result", + lambda job, content, **_kw: delivered.append(content) or None) + + record = execution_ledger.create_execution("job-dispatch", source="builtin") + job = {"id": "job-dispatch", "execution_id": record["id"], + "deliver": "telegram:123"} + assert scheduler.run_one_job(job, adapters=None) is True + + rows = incidents.list_incidents() + assert len(rows) == 1 + assert rows[0]["job_id"] == "job-dispatch" + assert rows[0]["state"] == "alerted" + assert "Restart-safe cron worker dispatch failed" in rows[0]["error"] + assert len(delivered) == 1 and delivered[0].strip() + finished = execution_ledger.get_execution(record["id"]) + assert finished["status"] == "failed" + assert "Restart-safe cron worker dispatch failed" in finished["error"] + assert finished["delivery_outcome"] == "delivered" + + # The same dispatch failure again: same signature -> incident dedup, and the + # alerted cooldown withholds the repeat ping. + repeat = execution_ledger.create_execution("job-dispatch", source="builtin") + job2 = {"id": "job-dispatch", "execution_id": repeat["id"], + "deliver": "telegram:123"} + assert scheduler.run_one_job(job2, adapters=None) is True + assert len(incidents.list_incidents()) == 1 + assert len(delivered) == 1 + assert execution_ledger.get_execution(repeat["id"])["delivery_outcome"] == \ + "suppressed_acked" + + def test_shutdown_does_not_interrupt_restart_safe_waiter(): import cron.scheduler as scheduler