cron: record killed post-adopt worker on the job (incident, delivery, log)

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]
 
 

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),
     )