tests/test_retry.py
Rama main · réplica de lectura del repositorio original.
"""Opt-in per-job retry (`max_retries`).
The broker still never re-runs a failed job on its own: the default is 0 and
every test here that asserts a retry also asserts that the same report without
the opt-in stays failed. What is retried is exactly one thing -- the workload
exited non-zero with no termination reason. A timeout, a preempt, a cancel and a
worker-side fault each carry a reason or a different final state, and re-running
those would either repeat a verdict the broker already reached or hide a broken
host behind a second attempt.
"""
from __future__ import annotations
import json
from datetime import UTC, datetime, timedelta
import pytest
HB = {
"free_vram_gb": 0,
"unregistered_vram_gb": 0,
"free_ram_gb": 8,
"idle_cpus": 4,
}
def submit(client, **extra):
r = client.post(
"/submit", json={"cmd": ["false"], "cwd": "/tmp", "project": "project-a", **extra}
)
assert r.status_code == 200, r.text
return r.json()["id"]
def claim(client, host="w1"):
client.post(
"/heartbeat",
json={
"host": host,
**HB,
"arch": "x86_64",
"os": "linux",
"gpu": False,
"tags": [],
"host_aliases": [],
},
)
body = client.post("/next-job", json={"host": host, **HB}).json()
if body is not None:
client.post(f"/jobs/{body['id']}/started", headers={"X-Jobd-Worker": host})
return body
def complete(client, job_id, host="w1", **payload):
r = client.post(f"/jobs/{job_id}/complete", json=payload, headers={"X-Jobd-Worker": host})
assert r.status_code == 200, r.text
return r.json()
def events(logs_dir, kind):
path = logs_dir / "events.jsonl"
rows = [json.loads(x) for x in path.read_text().splitlines()] if path.exists() else []
return [r for r in rows if r.get("event") == kind]
def test_failed_exit_is_requeued_until_retries_run_out(client_logs):
client, logs = client_logs
jid = submit(client, max_retries=2)
seen = []
for expected_attempt in (0, 1, 2):
job = claim(client)
assert job is not None and job["id"] == jid
assert job["attempt"] == expected_attempt
seen.append(complete(client, jid, exit_code=3))
assert [(j["state"], j["attempt"], j["exit_code"]) for j in seen] == [
("queued", 1, None),
("queued", 2, None),
("failed", 2, 3),
]
# A requeued attempt is a fresh claim: no worker, no start stamp, no finish.
assert (seen[0]["worker"], seen[0]["started_at"], seen[0]["finished_at"]) == (None, None, None)
assert claim(client) is None
retries = events(logs, "job_retry_scheduled")
assert [(e["job_id"], e["payload"]) for e in retries] == [
(jid, {"exit_code": 3, "attempt": n, "max_retries": 2, "retry_delay_s": 0, "worker": "w1"})
for n in (1, 2)
]
# One terminal event, for the attempt that actually ended the job.
assert [e["payload"]["final_state"] for e in events(logs, "job_completed")] == ["failed"]
def test_without_the_opt_in_a_failed_job_stays_failed(client_logs):
client, logs = client_logs
jid = submit(client)
claim(client)
job = complete(client, jid, exit_code=3)
assert (job["state"], job["attempt"], job["max_retries"]) == ("failed", 0, 0)
assert events(logs, "job_retry_scheduled") == []
@pytest.mark.parametrize(
"payload, final",
[
({"exit_code": 0}, "completed"),
(
{
"exit_code": 124,
"final_state": "failed",
"termination_reason": "wall_clock_exceeded",
},
"failed",
),
(
{"exit_code": 1, "final_state": "failed", "termination_reason": "launcher_missing"},
"failed",
),
({"exit_code": -15, "final_state": "cancelled"}, "cancelled"),
({"exit_code": -15, "final_state": "preempted"}, "preempted"),
],
)
def test_only_a_plain_nonzero_exit_is_retried(client, payload, final):
jid = submit(client, max_retries=5)
claim(client)
job = complete(client, jid, **payload)
assert (job["state"], job["attempt"]) == (final, 0)
# Control: same submission, plain failure -> the opt-in does fire.
jid2 = submit(client, max_retries=5)
assert claim(client)["id"] == jid2
assert complete(client, jid2, exit_code=1)["state"] == "queued"
def test_retry_delay_holds_the_job_back(client, app):
from jobd.db import Job
jid = submit(client, max_retries=1, retry_delay_s=600)
claim(client)
before = datetime.now(UTC).replace(tzinfo=None)
job = complete(client, jid, exit_code=1)
assert job["state"] == "queued"
not_before = datetime.fromisoformat(job["not_before"]).replace(tzinfo=None)
assert timedelta(seconds=595) < not_before - before < timedelta(seconds=605)
assert claim(client) is None # still inside the delay
# Once the delay has passed the same job is offered again.
with app.state.SessionLocal() as s:
row = s.get(Job, jid)
row.not_before = datetime.now(UTC) - timedelta(seconds=1)
s.commit()
again = claim(client)
# The hold is spent once the job is claimed; a finished job must not go on
# advertising a "next attempt" (seen on a real broker run, not in a unit test).
assert (again["id"], again["not_before"]) == (jid, None)
def test_cancel_pending_at_failure_wins_over_the_retry(client_logs):
client, logs = client_logs
jid = submit(client, max_retries=3)
claim(client)
assert client.post(f"/jobs/{jid}/cancel").status_code == 200 # running -> signal=cancel
job = complete(client, jid, exit_code=1) # worker reports a plain failure first
# The user was told "cancelling": whatever the terminal label, it must not run again.
assert (job["state"], job["attempt"]) == ("failed", 0)
assert events(logs, "job_retry_scheduled") == []
assert claim(client) is None
def test_children_wait_for_the_retry_instead_of_being_cancelled(client):
parent = submit(client, max_retries=1)
child = client.post(
"/submit",
json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a", "depends_on": [parent]},
).json()["id"]
claim(client)
complete(client, parent, exit_code=1)
assert client.get(f"/jobs/{child}").json()["state"] == "queued"
assert claim(client)["id"] == parent # the child is still blocked on it
complete(client, parent, exit_code=1) # retries exhausted -> terminal -> cascade
assert client.get(f"/jobs/{parent}").json()["state"] == "failed"
assert client.get(f"/jobs/{child}").json()["state"] == "cancelled"
@pytest.mark.parametrize("bad", [{"max_retries": -1}, {"max_retries": 21}, {"retry_delay_s": -1}])
def test_out_of_range_retry_options_are_rejected(client, bad):
r = client.post("/submit", json={"cmd": ["true"], "cwd": "/tmp", "project": "project-a", **bad})
assert r.status_code == 422
def test_a_job_row_from_before_the_upgrade_is_never_retried(tmp_path):
"""The live database is upgraded in place by ALTER TABLE. SQLite back-fills an
added column with its DEFAULT, and a NULL that slipped through must still read
as "no retries" rather than crash the /complete handler or the matcher."""
from sqlalchemy import create_engine, inspect, text
from sqlalchemy.orm import sessionmaker
from jobd.broker.jobinfo import _to_info
from jobd.db import Job, init_db, migrate
url = f"sqlite:///{tmp_path / 'old.db'}"
engine = create_engine(url)
init_db(engine)
with sessionmaker(engine)() as s:
s.add(
Job(
project="project-a",
priority=50,
state="running",
cmd_json='["false"]',
cwd="/tmp",
submitted_at=datetime(2026, 1, 1),
)
)
s.commit()
with engine.begin() as conn: # back to the pre-upgrade table shape
for col in ("max_retries", "retry_delay_s", "attempt", "not_before"):
conn.execute(text(f"ALTER TABLE jobs DROP COLUMN {col}"))
assert "attempt" not in {c["name"] for c in inspect(engine).get_columns("jobs")}
migrate(engine)
with engine.begin() as conn: # the worst case: the default did not take
conn.execute(
text("UPDATE jobs SET max_retries = NULL, attempt = NULL, retry_delay_s = NULL")
)
with sessionmaker(engine)() as s:
info = _to_info(s.get(Job, 1))
assert (info.attempt, info.max_retries, info.retry_delay_s, info.not_before) == (0, 0, 0, None)