diff --git a/cron/executions.py b/cron/executions.py index f64945e560..dd3d7c8f64 100644 --- a/cron/executions.py +++ b/cron/executions.py @@ -386,6 +386,60 @@ def recover_interrupted_executions() -> int: return changed +def terminalize_dead_owner(execution_id: str, *, reason: str) -> bool: + """Record one attempt as ``unknown`` with a cause this process actually observed. + + ``recover_interrupted_executions`` sweeps every attempt whose owner is provably + dead, and knows nothing but that absence — so all it can write is + ``_OWNER_GONE_REASON``, which asserts a scheduler restart. A waiter that held the + worker's ``Popen`` knows more: the owner was that external worker, and it exited + with a known status. Without this, a manual run whose worker dies is filed as + "Scheduler restarted ..." (a restart that never happened) and, because the sweep + leaves the row terminal, the waiter reports success and never records the run — + the job's ``fire_claim`` then blocks the next manual fire for the whole lease + (#128509). + + The attempt stays ``unknown``, not ``failed``: whether side effects ran is still + unknown. Only the CAUSE becomes truthful. Returns False — leaving the caller to + fall back to the generic sweep — when the row is absent, already terminal, owned by + this process, inside the handoff adoption grace, or owned by a live process: a + worker that is still running must never be terminalized out from under itself. + """ + now = _hermes_now().isoformat() + with _transaction() as conn: + row = conn.execute( + """SELECT id, status, process_id, pid, process_started_at, + handoff_pending, handoff_started_at + FROM executions WHERE id=?""", + (execution_id,), + ).fetchone() + if row is None or row["status"] not in ("claimed", "running"): + return False + if row["process_id"] == _PROCESS_ID: + return False + if _owner_is_live(int(row["pid"]), row["process_started_at"]): + return False + handoff_started_at = row["handoff_started_at"] + if ( + row["handoff_pending"] + and handoff_started_at is not None + and time.time() - float(handoff_started_at) < HANDOFF_ADOPTION_GRACE_SECONDS + ): + return False + cur = conn.execute( + """UPDATE executions + SET status='unknown', finished_at=?, error=?, + handoff_pending=0, handoff_started_at=NULL + WHERE id=? AND status=? AND process_id=? AND pid=?""", + (now, reason, row["id"], row["status"], row["process_id"], row["pid"]), + ) + if cur.rowcount != 1: + return False + record = _fetch(conn, execution_id) + _emit_execution_state(record) + return True + + def list_executions( *, job_id: Optional[str] = None, limit: int = 50, before_claimed_at: Optional[str] = None, ) -> List[Dict[str, Any]]: diff --git a/cron/scheduler.py b/cron/scheduler.py index 6276f94772..443beec1b8 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -543,7 +543,7 @@ from cron.jobs import ( 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) + recover_interrupted_executions, terminalize_dead_owner) # Response marker that suppresses delivery (output is still saved locally for audit). SILENT_MARKER = "[SILENT]" @@ -3415,10 +3415,27 @@ def _run_one_job_body( reset_terminal_scope(_terminal_scope_token) +def _external_worker_exited_reason(returncode: Optional[int]) -> str: + """Cause for an attempt whose restart-safe worker this process watched exit. + + The generic dead-owner sweep can only say the owner vanished, so it names a + scheduler restart — but this waiter never lost its scheduler: it handed the + attempt to the external worker and then saw that worker exit with this status. + The run is still ``unknown`` rather than ``failed`` (whether side effects ran is + still unknown); only the cause becomes truthful (#128509). + """ + return ( + f"Restart-safe cron worker exited with status {returncode} after adopting this " + "execution, without writing a durable terminal state; whether side effects ran " + "is unknown." + ) + + def _wait_for_external_cron_worker_body( process: subprocess.Popen, *, execution_id: str, + stderr_path: Optional[Path] = None, ) -> bool: """Preserve ``run_one_job``'s synchronous contract after handoff. @@ -3455,12 +3472,27 @@ def _wait_for_external_cron_worker_body( # 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(): - return True + # + # Record the cause THIS waiter observed instead of letting the blanket + # sweep stamp "Scheduler restarted ...", and then RAISE: the sweep leaves + # the row terminal, so returning success here made run_one_job skip + # mark_job_run entirely and the job's fire claim outlived its lost + # execution, refusing the next manual fire with "Job is already being + # fired" (#128509). Raising routes the existing post-handoff bookkeeping + # instead — record the run, no dispatch incident, no crash notice. + reason = _external_worker_exited_reason(returncode) + if not terminalize_dead_owner(execution_id, reason=reason): + recover_interrupted_executions() + if _is_terminal(): + return True + stderr_tail = "" + if stderr_path is not None: + from cron.scheduler_diagnostics import external_worker_stderr_tail + + stderr_tail = external_worker_stderr_tail(stderr_path) raise RuntimeError( - "cron external worker exited before durable recovery could " - f"terminalize its execution state (exit {returncode})" + "cron external worker exited before durable recovery could terminalize " + f"its execution state (exit {returncode}): {reason}{stderr_tail}" ) @@ -3474,10 +3506,11 @@ def _wait_for_external_cron_worker( execution_id: str, job_id: Optional[str] = None, handoff_files: tuple[Path, ...] = (), + stderr_path: Optional[Path] = None, ) -> bool: try: return _wait_for_external_cron_worker_body( - process, execution_id=execution_id + process, execution_id=execution_id, stderr_path=stderr_path ) except Exception as wait_error: raise _ExternalWorkerPostHandoffError(str(wait_error)) from wait_error @@ -3657,6 +3690,7 @@ def _launch_external_cron_worker(job: dict) -> bool: execution_id=execution_id, job_id=job_id, handoff_files=(payload_path, stderr_path), + stderr_path=stderr_path, ) finally: ack_path.unlink(missing_ok=True) @@ -3674,6 +3708,7 @@ def _launch_external_cron_worker(job: dict) -> bool: execution_id=execution_id, job_id=job_id, handoff_files=(payload_path, stderr_path), + stderr_path=stderr_path, ) logger.info( "Cron job '%s' handed to restart-safe worker pid=%s execution=%s", @@ -3689,6 +3724,7 @@ def _launch_external_cron_worker(job: dict) -> bool: execution_id=execution_id, job_id=job_id, handoff_files=(payload_path, stderr_path), + stderr_path=stderr_path, ) returncode = process.poll() if returncode is not None: @@ -3728,6 +3764,7 @@ def _launch_external_cron_worker(job: dict) -> bool: execution_id=execution_id, job_id=job_id, handoff_files=(payload_path, ack_path, stderr_path), + stderr_path=stderr_path, ) diff --git a/tests/cron/test_restart_safe_worker.py b/tests/cron/test_restart_safe_worker.py index 982e9af9fe..461a69ad2b 100644 --- a/tests/cron/test_restart_safe_worker.py +++ b/tests/cron/test_restart_safe_worker.py @@ -1172,7 +1172,7 @@ def test_post_handoff_waiter_failure_records_bookkeeping_without_alert( import cron.incidents as incidents import cron.scheduler as scheduler - def _body_boom(_process, *, execution_id): + def _body_boom(_process, *, execution_id, **_kwargs): raise RuntimeError("cron external worker exited before durable recovery") monkeypatch.setattr(scheduler, "_wait_for_external_cron_worker_body", _body_boom) diff --git a/tests/cron/test_restart_safe_worker_death_evidence.py b/tests/cron/test_restart_safe_worker_death_evidence.py new file mode 100644 index 0000000000..116c9135d3 --- /dev/null +++ b/tests/cron/test_restart_safe_worker_death_evidence.py @@ -0,0 +1,157 @@ +"""Durable evidence when a restart-safe cron worker dies AFTER adopting its attempt. + +Regression for #128509. A ``cronjob(action='run')`` fired inside a gateway agent turn +returns ``executed=true / execution_mode=background``, then the attempt dies. The waiter +learns the adopted worker exited and calls ``recover_interrupted_executions()``, which only +knows "the owner is gone" and stamps the generic ``_OWNER_GONE_REASON`` -- "Scheduler +restarted after this execution's owner exited" -- a restart that never happened. The waiter +then returns success, so ``run_one_job`` short-circuits and never calls ``mark_job_run``: + +* the job's ``fire_claim`` is left behind, so the next manual fire is refused with + "Job is already being fired by the scheduler" until the 300s lease expires, and +* ``last_error`` stays empty, so the operator has no error and no cause anywhere. + +The worker is not in the calling turn's process group (``start_new_session=True`` plus a +``systemd-run --user --scope`` cgroup), so turn teardown never signalled it: the child +crashed on its own. What the gateway still owes the operator is a truthful cause and a +released claim. +""" + +from __future__ import annotations + +import json +import subprocess +import sys +import time +from pathlib import Path + +import pytest + +# A real worker process: it adopts the durable row (the ownership rewrite that makes the +# reported status line reachable), publishes the acknowledgement, then dies on its own. +# The sleep keeps the parent's ack poll (every 50ms) deterministically ahead of the exit. +_WORKER_THAT_DIES_AFTER_ADOPTION = """ +import json, os, sys, time +from pathlib import Path + +argv = sys.argv[1:] +payload_path = Path(argv[argv.index("--external-worker-file") + 1]) +ack_path = Path(argv[argv.index("--ack-file") + 1]) + +import cron.executions as executions + +job = json.loads(payload_path.read_text(encoding="utf-8-sig"))["job"] +assert executions.adopt_claimed_execution(job["execution_id"]) is not None +ack_path.write_text( + json.dumps({"pid": os.getpid(), "execution_id": job["execution_id"]}), + encoding="utf-8", +) +time.sleep(1.5) +os._exit(9) +""" + + +@pytest.fixture +def profile_store(tmp_path, monkeypatch): + """A real profile-scoped cron store, ledger and scheduler home, all on one temp path.""" + from cron.jobs import create_job, use_cron_store + + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + import cron.executions as executions + import cron.scheduler as scheduler + + monkeypatch.setattr(scheduler, "_hermes_home", home) + monkeypatch.setattr(executions, "EXECUTIONS_FILE", home / "cron" / "executions.db") + monkeypatch.setattr(scheduler, "load_config_readonly", lambda: {}) + return home + + +def test_worker_dying_after_adoption_reports_its_cause_and_releases_the_claim( + tmp_path, monkeypatch, profile_store +): + """The run is lost either way; the ledger must not blame a restart that never + happened, the job must be claimable again, and the operator must get a cause.""" + import cron.executions as executions + import cron.scheduler as scheduler + from cron.jobs import claim_job_for_fire, create_job, get_job, use_cron_store + from tools.process_registry import GatewayChildDispatch + + home = profile_store + with use_cron_store(home): + job = create_job(prompt="summarise the day", schedule="every 1h", name="daily brief") + claimed = claim_job_for_fire(job["id"], manual=True, return_job=True) + assert claimed, "the manual claim must succeed or this test proves nothing" + + worker = tmp_path / "worker_that_dies.py" + worker.write_text(_WORKER_THAT_DIES_AFTER_ADOPTION, encoding="utf-8") + + def dispatch(command, *, unit_suffix, require_restart_safe_scope=False): + # Keep the launcher's real payload/ack argv; swap only the child's body so the + # spawn, the acknowledgement poll and the wait all run for real. + flags = command[command.index("--external-worker-file"):] + return GatewayChildDispatch( + "degraded", [sys.executable, str(worker), *flags]) + + monkeypatch.setattr( + "tools.process_registry.restart_safe_gateway_child_argv", dispatch) + + assert scheduler.run_one_job(claimed, adapters=None) is True + execution_id = claimed["execution_id"] + + with use_cron_store(home): + record = get_job(job["id"]) + # The claim is what makes the next manual fire say "Job is already being fired"; + # only mark_job_run retires it, and the lost waiter used to never call it. + assert record["fire_claim"] is None, "the fire claim outlived its lost execution" + assert claim_job_for_fire(job["id"], manual=True) is True + + # A truthful cause: this waiter never lost its scheduler, it watched ITS worker + # exit. The generic dead-owner sweep can only assert a restart. + assert record["last_error"], "a lost run recorded no error at all" + assert "Scheduler restarted" not in record["last_error"] + assert "exit 9" in record["last_error"] + + # Still `unknown`, not a fabricated durable state: side effects really are unknown. + row = executions.get_execution(execution_id) + assert row["status"] == "unknown" + assert "Scheduler restarted" not in (row["error"] or "") + + +def test_terminalizer_refuses_an_attempt_whose_worker_is_still_alive( + profile_store, +): + """The targeted terminalizer must not steal a LIVE worker's attempt: refusing a live + owner is what keeps the fix from becoming a second way to lose side effects.""" + import cron.executions as executions + + script = ( + "import sys, time\n" + "from pathlib import Path\n" + "import cron.executions as executions\n" + f"executions.EXECUTIONS_FILE = Path({str(executions.EXECUTIONS_FILE)!r})\n" + "assert executions.adopt_claimed_execution(sys.argv[1]) is not None\n" + "sys.stdout.write('adopted\\n')\n" + "sys.stdout.flush()\n" + "time.sleep(60)\n" + ) + record = executions.create_execution("job-live", source="direct") + assert executions.mark_execution_handoff_pending(record["id"]) is not None + live_worker = subprocess.Popen( + [sys.executable, "-c", script, record["id"]], + stdout=subprocess.PIPE, text=True) + try: + deadline = time.monotonic() + 30.0 + while time.monotonic() < deadline: + if executions.get_execution(record["id"])["status"] == "running": + break + time.sleep(0.05) + assert executions.get_execution(record["id"])["status"] == "running" + + assert executions.terminalize_dead_owner( + record["id"], reason="exit 9") is False + assert executions.get_execution(record["id"])["status"] == "running" + finally: + live_worker.kill() + live_worker.wait(timeout=30)