144 lines
5.8 KiB
Python
144 lines
5.8 KiB
Python
"""Regression for #86721 — a one-shot `hermes cron run` invocation's
|
|
dispatched runner thread dies with the exiting process, leaving a stale
|
|
'claimed'/'running' row in cron/executions.db that blocks every subsequent
|
|
manual run of the same job. recover_interrupted_executions() already
|
|
existed and is correctly implemented, but was only ever called at the
|
|
long-lived scheduler ticker's own startup -- a one-shot CLI invocation has
|
|
no equivalent moment of its own, so the self-heal never ran for it.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
def test_stale_claim_from_a_dead_one_shot_process_blocks_new_execution_creation(
|
|
tmp_path,
|
|
):
|
|
"""Sanity/negative-control: reproduces the exact reported symptom in
|
|
isolation -- a genuinely dead-owner 'claimed' row for a job, with no
|
|
recovery step run, simply sits there. (create_execution() doesn't
|
|
itself gate on prior rows for the same job_id; the blocking happens
|
|
at a higher layer that checks for an existing claimed/running row --
|
|
this test establishes the stale row exists and stays 'claimed'
|
|
without recovery, matching the bug report's own SQL evidence.)"""
|
|
home = tmp_path / "home"
|
|
repo = Path(__file__).resolve().parents[2]
|
|
env = os.environ.copy()
|
|
env["HERMES_HOME"] = str(home)
|
|
env["PYTHONPATH"] = str(repo)
|
|
|
|
# Simulate the dispatched runner's owner process dying mid-flight,
|
|
# exactly as issue #86721 describes: a one-shot process creates the
|
|
# execution row (status='claimed') and exits before finishing it.
|
|
create = subprocess.run(
|
|
[
|
|
sys.executable, "-c",
|
|
"from cron.executions import create_execution; "
|
|
"r=create_execution('cron-run-job', source='direct'); "
|
|
"print(r['id'])",
|
|
],
|
|
cwd=repo, env=env, text=True, capture_output=True, check=True,
|
|
)
|
|
execution_id = create.stdout.strip()
|
|
|
|
# Without recovery, the row is exactly where the bug report found it.
|
|
check = subprocess.run(
|
|
[
|
|
sys.executable, "-c",
|
|
"import json; from cron.executions import list_executions; "
|
|
"print(json.dumps(list_executions(job_id='cron-run-job')))",
|
|
],
|
|
cwd=repo, env=env, text=True, capture_output=True, check=True,
|
|
)
|
|
records = json.loads(check.stdout.strip())
|
|
assert len(records) == 1
|
|
assert records[0]["id"] == execution_id
|
|
assert records[0]["status"] == "claimed" # stranded, matching the report
|
|
|
|
|
|
def test_recover_interrupted_executions_reaps_the_stale_claim_from_a_dead_process(
|
|
tmp_path,
|
|
):
|
|
"""The self-heal this issue's fix now triggers per one-shot
|
|
invocation: recover_interrupted_executions() correctly identifies and
|
|
clears a stale claim left by a process that has genuinely exited
|
|
(real dead PID, not a mock), unblocking the job for a new run."""
|
|
home = tmp_path / "home"
|
|
repo = Path(__file__).resolve().parents[2]
|
|
env = os.environ.copy()
|
|
env["HERMES_HOME"] = str(home)
|
|
env["PYTHONPATH"] = str(repo)
|
|
|
|
create = subprocess.run(
|
|
[
|
|
sys.executable, "-c",
|
|
"from cron.executions import create_execution; "
|
|
"r=create_execution('cron-run-job-2', source='direct'); "
|
|
"print(r['id'])",
|
|
],
|
|
cwd=repo, env=env, text=True, capture_output=True, check=True,
|
|
)
|
|
execution_id = create.stdout.strip()
|
|
|
|
# This is what #86721's fix now calls, from a FRESH process, mirroring
|
|
# exactly what a subsequent `hermes cron run` invocation would trigger
|
|
# before attempting its own claim.
|
|
recover = subprocess.run(
|
|
[
|
|
sys.executable, "-c",
|
|
"import json; "
|
|
"from cron.executions import recover_interrupted_executions, list_executions; "
|
|
"print(recover_interrupted_executions()); "
|
|
"print(json.dumps(list_executions(job_id='cron-run-job-2')))",
|
|
],
|
|
cwd=repo, env=env, text=True, capture_output=True, check=True,
|
|
)
|
|
lines = recover.stdout.strip().splitlines()
|
|
assert lines[0] == "1", "exactly one stale execution must be reaped"
|
|
records = json.loads(lines[1])
|
|
assert records[0]["id"] == execution_id
|
|
assert records[0]["status"] == "unknown", (
|
|
"the stale claim from the dead owner must be reclassified, "
|
|
"no longer blocking a fresh claim attempt for this job"
|
|
)
|
|
|
|
|
|
def test_try_dispatch_background_run_calls_recovery_before_claiming(monkeypatch):
|
|
"""Direct unit check on the fix's exact insertion point: the one-shot
|
|
dispatch path must call recover_interrupted_executions() before
|
|
proceeding to claim/create an execution for THIS run, so a stale row
|
|
left by a prior dead one-shot invocation is cleared first."""
|
|
import tools.cronjob_tools as cronjob_tools
|
|
|
|
calls = []
|
|
monkeypatch.setattr(
|
|
"cron.executions.recover_interrupted_executions",
|
|
lambda: calls.append("recovered") or 0,
|
|
)
|
|
# Force the function past its own early-return guards so execution
|
|
# reaches the point where recovery is called, without needing a real
|
|
# gateway/session runtime.
|
|
monkeypatch.setattr(
|
|
"gateway.session_context.async_delivery_supported", lambda: True
|
|
)
|
|
monkeypatch.setattr(
|
|
"tools.approval.get_current_session_key", lambda default="": ""
|
|
)
|
|
|
|
job = {"id": "unit-test-job", "name": "unit test job", "deliver": "local"}
|
|
# session_id is intentionally empty/None too, matching the direct-
|
|
# caller ("hermes cron run", tests) early return documented just
|
|
# below the recovery call -- this test only needs to confirm recovery
|
|
# ran before that point, not exercise the full dispatch.
|
|
cronjob_tools._try_dispatch_background_run(job, session_id=None)
|
|
|
|
assert calls == ["recovered"], (
|
|
"recover_interrupted_executions() must be called before any claim "
|
|
"attempt in the one-shot dispatch path"
|
|
)
|