Jonathan
d7ea29a97a
An unhandled exception in run_pipeline (e.g. a MusicBrainz TLS drop surfacing as SSL: UNEXPECTED_EOF, after the MB retries are exhausted) left the job stranded in its mid-pipeline 'matching' state: the except handler logged "pipeline failed" and rolled back, but never reset the job. Since claim_next only picks 'requested', the job was orphaned — stuck showing "Searching sources…" on the press until the next worker restart's reclaim_stuck_jobs happened to sweep it up. Add requeue_or_fail: on an unhandled failure, requeue to 'requested' for a bounded retry (transient network blips clear on retry), then fall to 'needs_attention' at the attempt cap (5) so a persistently-broken job stays visible instead of looping. Mirrors reclaim_stuck_jobs (requeue) and pipeline._fail (needs_attention), and resets Request.status to match. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
70 lines
2.4 KiB
Python
70 lines
2.4 KiB
Python
from lyra_worker.claim import claim_next, requeue_or_fail
|
|
from tests.conftest import insert_request
|
|
|
|
|
|
def _set_attempts(conn, job_id, attempts, state="matching", stage="match"):
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
'UPDATE "Job" SET attempts = %s, state = %s, "currentStage" = %s, "claimedAt" = now() '
|
|
'WHERE id = %s',
|
|
(attempts, state, stage, job_id),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def _job(conn, job_id):
|
|
with conn.cursor() as cur:
|
|
cur.execute('SELECT state, "currentStage", "claimedAt", error FROM "Job" WHERE id = %s', (job_id,))
|
|
return cur.fetchone()
|
|
|
|
|
|
def _request_status(conn, job_id):
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
'SELECT status FROM "Request" WHERE id = (SELECT "requestId" FROM "Job" WHERE id = %s)',
|
|
(job_id,),
|
|
)
|
|
return cur.fetchone()[0]
|
|
|
|
|
|
def test_below_cap_requeues_to_requested(conn):
|
|
# A transient failure (e.g. MB SSL EOF) mid-'matching' must not strand the job — it returns
|
|
# to 'requested' so claim_next picks it up again instead of orphaning it.
|
|
jid = insert_request(conn)
|
|
_set_attempts(conn, jid, 2)
|
|
requeue_or_fail(conn, jid, "SSL: UNEXPECTED_EOF_WHILE_READING")
|
|
st, stage, claimed_at, err = _job(conn, jid)
|
|
assert st == "requested"
|
|
assert stage == "intake"
|
|
assert claimed_at is None
|
|
assert err is None
|
|
assert _request_status(conn, jid) == "pending"
|
|
assert claim_next(conn) == jid # immediately re-claimable
|
|
|
|
|
|
def test_at_cap_marks_needs_attention_with_error(conn):
|
|
jid = insert_request(conn)
|
|
_set_attempts(conn, jid, 5) # hit the cap
|
|
requeue_or_fail(conn, jid, "SSL: UNEXPECTED_EOF_WHILE_READING")
|
|
st, _stage, claimed_at, err = _job(conn, jid)
|
|
assert st == "needs_attention"
|
|
assert err == "SSL: UNEXPECTED_EOF_WHILE_READING"
|
|
assert claimed_at is None
|
|
assert _request_status(conn, jid) == "needs_attention"
|
|
assert claim_next(conn) is None # not re-claimed
|
|
|
|
|
|
def test_requeues_until_cap_then_fails(conn):
|
|
jid = insert_request(conn)
|
|
for attempts in (1, 2, 3, 4):
|
|
_set_attempts(conn, jid, attempts)
|
|
requeue_or_fail(conn, jid, "boom")
|
|
assert _job(conn, jid)[0] == "requested"
|
|
_set_attempts(conn, jid, 5)
|
|
requeue_or_fail(conn, jid, "boom")
|
|
assert _job(conn, jid)[0] == "needs_attention"
|
|
|
|
|
|
def test_missing_job_is_noop(conn):
|
|
requeue_or_fail(conn, "does-not-exist", "boom") # must not raise
|