diff --git a/cron/incidents.py b/cron/incidents.py index 22a107da5ce54..c1e174b5505d6 100644 --- a/cron/incidents.py +++ b/cron/incidents.py @@ -124,8 +124,14 @@ def _redact_error(error: str) -> str: def _error_signature(job_id: str, error: str) -> str: - """Dedup key: stable for same job + same normalized error prefix.""" - normalized = _normalize_error(error)[:_MAX_SIGNATURE_ERROR_CHARS] + """Dedup key: stable for same job + same normalized error prefix. + + A recovered worker's exit code is excluded: a kill that flaps between signals + (137 vs 143) must refresh ONE incident for the recovery reason rather than mint a + fresh row per code and bypass the repeat-alert cooldown (#120328).""" + normalized = re.sub( + r"\s*\(cron worker exited with code \d+\)$", "", _normalize_error(error) + )[:_MAX_SIGNATURE_ERROR_CHARS] return hashlib.sha256(job_id.encode() + normalized.encode()).hexdigest()[:12] diff --git a/cron/scheduler.py b/cron/scheduler.py index c70b868a534af..b14e2a4869cf5 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -2723,7 +2723,7 @@ def run_one_job( external_owner = os.environ.get("_HERMES_CRON_EXTERNAL_WORKER") == execution_id if not external_owner: try: - if _launch_external_cron_worker(job): + if _launch_external_cron_worker(job, adapters=adapters, loop=loop): return True except Exception as handoff_error: error = f"Restart-safe cron worker dispatch failed: {handoff_error}" @@ -3358,10 +3358,99 @@ def _teardown_deferred() -> None: reset_terminal_scope(_terminal_scope_token) +def _record_killed_worker_run( + record: dict, + returncode: Optional[int], + *, + job_id: Optional[str] = None, + fire_owner: Optional[str] = None, + adapters=None, + loop=None, +) -> None: + """Gateway-side job bookkeeping for an adopted worker that died without a terminal row. + + The execution was recovered to ``unknown`` (#120328) — by the waiter's own recovery + or by a competing reaper (tick dead-owner reap / manual pre-dispatch). Without this, + the job record keeps the previous run's values, no incident opens, nothing is + delivered and nothing is logged — every job-level surface says the scheduled run + never happened. A worker that terminalized itself (``completed``/``failed``) owns + its own bookkeeping and is left alone. + + Best-effort and fenced, so it never breaks the waiter's contract and never touches a + replacement claim: when the fire owner seen at launch no longer holds the claim (or the + claim is already consumed), this is a no-op. Idempotent for the same reason — the first + ``mark_job_run`` consumes the claim, so a repeat finds nothing to fence. + + ``returncode`` is None when the worker has not exited yet (a reaper released the + claim while the process was still alive or in final teardown); the recovery reason + on the record already says why the run was abandoned. + """ + if not isinstance(record, dict) or record.get("status") != "unknown": + return + resolved_job_id = job_id or record.get("job_id") + base_error = str( + record.get("error") + or "Cron worker exited before a durable terminal state; " + "whether side effects ran is unknown." + ) + error = base_error if returncode is None else ( + f"{base_error} (cron worker exited with code {returncode})" + ) + try: + if not resolved_job_id: + logger.error( + "Cron external worker %s died without terminalizing: %s", + record.get("id"), error) + return + logger.error("Job '%s': %s", resolved_job_id, error) + from cron.jobs import get_job + job = get_job(str(resolved_job_id)) + if job is None: + return + claim = job.get("fire_claim") + live_owner = str(claim.get("by") or "") if isinstance(claim, dict) else "" + if fire_owner is not None: + if live_owner != fire_owner: + return # superseded or already recorded; never touch a replacement run + if not fire_owner: + return # launch carried no claim and none exists now; nothing to attribute + fence_owner: Optional[str] = fire_owner + else: + if not live_owner: + return + fence_owner = live_owner + incident_acked, failure_incident_id = _upsert_incident_for_failure(job, error) + delivery_outcome = "suppressed_acked" if incident_acked else "suppressed" + delivery_error: Optional[str] = None + if not incident_acked: + try: + delivery_error, delivery_outcome = _deliver_crash_failure( + job, error, adapters=adapters, loop=loop) + except Exception as delivery_exc: + delivery_error, delivery_outcome = str(delivery_exc), "failed" + mark_kwargs: dict = ( + {"delivery_error": delivery_error} if delivery_error else {} + ) + marked = mark_job_run( + str(resolved_job_id), False, error, + expected_fire_owner=fence_owner, **mark_kwargs, + ) + if marked and delivery_outcome in ("delivered", "not_configured"): + _mark_incident_alerted(failure_incident_id) + except Exception: + logger.error( + "Failed recording killed-worker run for job %s", + resolved_job_id, exc_info=True) + + def _wait_for_external_cron_worker_body( process: subprocess.Popen, *, execution_id: str, + job_id: Optional[str] = None, + fire_owner: Optional[str] = None, + adapters=None, + loop=None, ) -> bool: """Preserve ``run_one_job``'s synchronous contract after handoff. @@ -3371,9 +3460,15 @@ def _wait_for_external_cron_worker_body( A gateway replacement may kill this waiter; it does not kill the scoped worker or change its ledger ownership. """ - def _is_terminal() -> bool: + def _terminal_record() -> Optional[dict]: + """The terminal execution row, or None. ``unknown`` counts as terminal: a + competing reaper (tick dead-owner reap / manual pre-dispatch) can move a dead + worker's row there before this waiter's own recovery runs (#120328), so every + exit that can see one must still record the killed run.""" current = get_execution(execution_id) - return bool(current and current.get("status") in _TERMINAL_STATES) + if current and current.get("status") in _TERMINAL_STATES: + return current + return None # The worker commits its terminal row before its process exits, so exit is # the correct wakeup. Each ledger read opens a connection and re-runs @@ -3383,7 +3478,15 @@ def _is_terminal() -> bool: try: returncode = process.wait(timeout=1.0) except subprocess.TimeoutExpired: - if _is_terminal(): + terminal = _terminal_record() + if terminal is not None: + # A competing reaper can have already recovered the row to ``unknown`` + # (#120328): record the killed run from this exit too. No exit code + # exists yet — the process is still alive or in its final teardown. + _record_killed_worker_run( + terminal, None, job_id=job_id, + fire_owner=fire_owner, adapters=adapters, loop=loop, + ) from cron.scheduler_detached_worker import reap_terminal_worker_in_background reap_terminal_worker_in_background(process) @@ -3392,14 +3495,29 @@ def _is_terminal() -> bool: # The worker can commit its terminal row and exit between the first # read and wait(). Re-read the exact attempt before declaring that # it died without terminalizing. - if _is_terminal(): + terminal = _terminal_record() + if terminal is not None: + # Same race on this exit: the row may already be ``unknown`` because a + # competing reaper won the recovery first (#120328). + _record_killed_worker_run( + terminal, returncode, job_id=job_id, + fire_owner=fire_owner, adapters=adapters, loop=loop, + ) return True # If the adopted worker died without terminalizing, its owner is # now provably gone. Recover to ``unknown`` rather than routing the # exception through the pre-handoff dispatch-failure path, which # would falsely assert that no side effect could have happened. recover_interrupted_executions() - if _is_terminal(): + try: + recovered = get_execution(execution_id) + except Exception: + recovered = None + if bool(recovered and recovered.get("status") in _TERMINAL_STATES): + _record_killed_worker_run( + recovered, returncode, job_id=job_id, + fire_owner=fire_owner, adapters=adapters, loop=loop, + ) return True raise RuntimeError( "cron external worker exited before durable recovery could " @@ -3412,11 +3530,15 @@ def _wait_for_external_cron_worker( *, execution_id: str, job_id: Optional[str] = None, + fire_owner: Optional[str] = None, + adapters=None, + loop=None, handoff_files: tuple[Path, ...] = (), ) -> bool: try: return _wait_for_external_cron_worker_body( - process, execution_id=execution_id + process, execution_id=execution_id, job_id=job_id, + fire_owner=fire_owner, adapters=adapters, loop=loop, ) finally: if job_id is not None: @@ -3431,7 +3553,7 @@ def _wait_for_external_cron_worker( pass -def _launch_external_cron_worker(job: dict) -> bool: +def _launch_external_cron_worker(job: dict, *, adapters=None, loop=None) -> bool: """Launch *job* outside the managed gateway process when required. Returns ``False`` outside a managed systemd gateway (in-process path). In @@ -3442,6 +3564,11 @@ def _launch_external_cron_worker(job: dict) -> bool: """ execution_id = str(job["execution_id"]) job_id = str(job["id"]) + _launch_claim = job.get("fire_claim") + fire_owner = ( + str(_launch_claim.get("by") or "") + if isinstance(_launch_claim, dict) else "" + ) handoff_dir = _get_hermes_home() / "cron" / "external-workers" payload_path = handoff_dir / f"{execution_id}.json" ack_path = handoff_dir / f"{execution_id}.ready" @@ -3582,6 +3709,9 @@ def _launch_external_cron_worker(job: dict) -> bool: process, execution_id=execution_id, job_id=job_id, + fire_owner=fire_owner, + adapters=adapters, + loop=loop, handoff_files=(payload_path, stderr_path), ) finally: @@ -3599,6 +3729,9 @@ def _launch_external_cron_worker(job: dict) -> bool: process, execution_id=execution_id, job_id=job_id, + fire_owner=fire_owner, + adapters=adapters, + loop=loop, handoff_files=(payload_path, stderr_path), ) logger.info( @@ -3614,6 +3747,9 @@ def _launch_external_cron_worker(job: dict) -> bool: process, execution_id=execution_id, job_id=job_id, + fire_owner=fire_owner, + adapters=adapters, + loop=loop, handoff_files=(payload_path, stderr_path), ) returncode = process.poll() @@ -3653,6 +3789,9 @@ def _launch_external_cron_worker(job: dict) -> bool: process, execution_id=execution_id, job_id=job_id, + fire_owner=fire_owner, + adapters=adapters, + loop=loop, handoff_files=(payload_path, ack_path, stderr_path), ) diff --git a/tests/cron/test_cron_killed_worker_job_update.py b/tests/cron/test_cron_killed_worker_job_update.py new file mode 100644 index 0000000000000..444059a9098b7 --- /dev/null +++ b/tests/cron/test_cron_killed_worker_job_update.py @@ -0,0 +1,192 @@ +"""#120328: a cron worker killed after adoption must not leave the job untouched. + +A restart-safe worker that dies after adopting its execution (e.g. systemd +OOM-kills its scope) is recovered to ``unknown`` by the gateway waiter — but +nothing updates the job: ``last_run_at``/``last_status``/``last_error`` keep +the previous run's values, no ``cron_incidents`` row opens, nothing is +delivered, and the gateway logs no error. +""" +from __future__ import annotations + +import logging +from unittest.mock import Mock + + +def test_pre_recovered_row_still_recorded_on_worker_exit( + tmp_path, monkeypatch, caplog +): + """#120328 follow-up: a competing reaper (tick dead-owner reap / manual pre-dispatch) + can move the dead worker's row to ``unknown`` before this waiter's post-wait read. + The exit then sees a terminal row and must still record the killed run — otherwise + the job keeps saying the run never happened.""" + import cron.scheduler as scheduler + import cron.executions as executions + import cron.incidents as incidents + from cron.jobs import ( + claim_job_for_fire, create_job, get_job, use_cron_store, + ) + + monkeypatch.setattr( + executions, "EXECUTIONS_FILE", tmp_path / "executions.db") + with use_cron_store(tmp_path): + job = create_job(prompt="x", schedule="every 5m", name="killed-race") + jid = job["id"] + assert claim_job_for_fire(jid) is True + pre_recovered = { + "id": "exec-1", "job_id": jid, "status": "unknown", + "error": executions._OWNER_GONE_REASON, + } + monkeypatch.setattr( + scheduler, "get_execution", lambda _eid: dict(pre_recovered)) + recover = Mock(return_value=0) # recovery already ran elsewhere + monkeypatch.setattr( + scheduler, "recover_interrupted_executions", recover) + delivered = Mock(return_value=None) + monkeypatch.setattr(scheduler, "_deliver_result", delivered) + + process = Mock() + process.wait.return_value = 143 + + with caplog.at_level(logging.ERROR): + assert scheduler._wait_for_external_cron_worker( + process, execution_id="exec-1", job_id=jid) is True + + after = get_job(jid) + assert after["last_run_at"] is not None + assert after["last_status"] == "error" + assert "143" in (after["last_error"] or "") + rows = [r for r in incidents.list_incidents() + if r["job_id"] == jid] + assert rows, "reaper-pre-recovered killed run must still open an incident" + assert delivered.called, "reaper-pre-recovered killed run must deliver" + + +def test_pre_recovered_row_still_recorded_on_wait_timeout( + tmp_path, monkeypatch, caplog +): + """The worker-liveness exit (wait timeout) can also see a row a competing reaper + already recovered to ``unknown`` (dead-owner / wedged release): the bookkeeping + must fire there too, with no exit code available yet.""" + import subprocess + import cron.scheduler as scheduler + import cron.executions as executions + import cron.incidents as incidents + from cron.jobs import ( + claim_job_for_fire, create_job, get_job, use_cron_store, + ) + + monkeypatch.setattr( + executions, "EXECUTIONS_FILE", tmp_path / "executions.db") + with use_cron_store(tmp_path): + job = create_job(prompt="x", schedule="every 5m", name="killed-wedged") + jid = job["id"] + assert claim_job_for_fire(jid) is True + pre_recovered = { + "id": "exec-1", "job_id": jid, "status": "unknown", + "error": executions._OWNER_WEDGED_REASON, + } + monkeypatch.setattr( + scheduler, "get_execution", lambda _eid: dict(pre_recovered)) + recover = Mock(return_value=0) + monkeypatch.setattr( + scheduler, "recover_interrupted_executions", recover) + delivered = Mock(return_value=None) + monkeypatch.setattr(scheduler, "_deliver_result", delivered) + + process = Mock() + process.wait.side_effect = [ + subprocess.TimeoutExpired(cmd="cron-worker", timeout=1.0), 137, + ] + + with caplog.at_level(logging.ERROR): + assert scheduler._wait_for_external_cron_worker( + process, execution_id="exec-1", job_id=jid) is True + + after = get_job(jid) + assert after["last_run_at"] is not None + assert after["last_status"] == "error" + rows = [r for r in incidents.list_incidents() + if r["job_id"] == jid] + assert rows, "wedged pre-recovered killed run must still open an incident" + assert delivered.called, "wedged pre-recovered killed run must deliver" + + +def test_flapping_kill_codes_share_one_incident(tmp_path, monkeypatch): + """A kill that flaps between signals (137 vs 143) must refresh ONE incident for the + recovery reason — not mint a fresh row per code and reset the repeat-alert cooldown.""" + import cron.scheduler as scheduler + import cron.executions as executions + import cron.incidents as incidents + from cron.jobs import create_job, use_cron_store + + monkeypatch.setattr( + executions, "EXECUTIONS_FILE", tmp_path / "executions.db") + with use_cron_store(tmp_path): + job = create_job(prompt="x", schedule="every 5m", name="flap") + jid = job["id"] + for base in (executions._OWNER_GONE_REASON, + executions._OWNER_WEDGED_REASON): + scheduler._upsert_incident_for_failure( + job, f"{base} (cron worker exited with code 137)") + scheduler._upsert_incident_for_failure( + job, f"{base} (cron worker exited with code 143)") + rows = [r for r in incidents.list_incidents() + if r["job_id"] == jid] + assert len(rows) == 2, ( + "one incident per recovery reason across kill codes, got " + f"{len(rows)}: {[r['error'] for r in rows]}") + + +def test_killed_worker_after_adopt_updates_job_incident_and_log( + tmp_path, monkeypatch, caplog +): + import cron.scheduler as scheduler + import cron.executions as executions + import cron.incidents as incidents + from cron.jobs import ( + claim_job_for_fire, create_job, get_job, use_cron_store, + ) + + monkeypatch.setattr( + executions, "EXECUTIONS_FILE", tmp_path / "executions.db") + with use_cron_store(tmp_path): + job = create_job(prompt="x", schedule="every 5m", name="killed") + jid = job["id"] + assert claim_job_for_fire(jid) is True + assert get_job(jid)["fire_claim"]["by"] + + recovery_error = executions._OWNER_GONE_REASON + states = iter([ + {"id": "exec-1", "job_id": jid, "status": "running"}, + {"id": "exec-1", "job_id": jid, "status": "unknown", + "error": recovery_error}, + {"id": "exec-1", "job_id": jid, "status": "unknown", + "error": recovery_error}, + ]) + monkeypatch.setattr( + scheduler, "get_execution", lambda _eid: next(states)) + recover = Mock(return_value=1) + monkeypatch.setattr( + scheduler, "recover_interrupted_executions", recover) + delivered = Mock(return_value=None) + monkeypatch.setattr(scheduler, "_deliver_result", delivered) + + process = Mock() + process.wait.return_value = 137 + process.poll.return_value = 137 + + with caplog.at_level(logging.ERROR): + assert scheduler._wait_for_external_cron_worker( + process, execution_id="exec-1", job_id=jid) is True + + after = get_job(jid) + assert after["last_run_at"] is not None + assert after["last_status"] == "error" + assert "137" in (after["last_error"] or "") + rows = [r for r in incidents.list_incidents() + if r["job_id"] == jid] + assert rows, "killed run must open a cron_incidents row" + assert delivered.called, "killed run must attempt failure delivery" + assert any(jid in r.getMessage() or "exec-1" in r.getMessage() + for r in caplog.records), \ + "killed run must log a gateway error" diff --git a/tests/cron/test_restart_safe_worker.py b/tests/cron/test_restart_safe_worker.py index aa3d0b2505c23..61172b48b165e 100644 --- a/tests/cron/test_restart_safe_worker.py +++ b/tests/cron/test_restart_safe_worker.py @@ -657,9 +657,10 @@ def test_shared_run_path_hands_gateway_fire_to_external_worker(monkeypatch): monkeypatch.setattr(scheduler, "run_job", run) job = {"id": "job-1", "execution_id": "exec-1"} - assert scheduler.run_one_job(job, adapters={"discord": object()}) is True + adapters = {"discord": object()} + assert scheduler.run_one_job(job, adapters=adapters) is True - launch.assert_called_once_with(job) + launch.assert_called_once_with(job, adapters=adapters, loop=None) run.assert_not_called() @@ -748,7 +749,7 @@ def test_gateway_tool_run_without_adapter_objects_hands_off(monkeypatch): created.assert_called_once_with("tool-job", source="direct", scheduled_instant=None) assert job["execution_id"] == "exec-tool" - launch.assert_called_once_with(job) + launch.assert_called_once_with(job, adapters=None, loop=None) run.assert_not_called()