diff --git a/worker/lyra_worker/claim.py b/worker/lyra_worker/claim.py index 788d128..135ae43 100644 --- a/worker/lyra_worker/claim.py +++ b/worker/lyra_worker/claim.py @@ -24,6 +24,43 @@ def reclaim_stuck_jobs(conn: psycopg.Connection) -> int: return n +_MAX_ATTEMPTS = 5 + + +def requeue_or_fail( + conn: psycopg.Connection, job_id: str, error: str, max_attempts: int = _MAX_ATTEMPTS +) -> None: + """Recover a job whose pipeline raised an *unhandled* exception (e.g. a dropped MusicBrainz + TLS connection surfacing as an SSL EOF). Without this the job is left in its mid-pipeline + 'matching' state; since claim_next only picks 'requested' it is orphaned until the next + restart's reclaim_stuck_jobs — showing "Searching sources…" on the press indefinitely. + + Such errors are usually transient, so requeue to 'requested' until the attempt cap, then fall + to 'needs_attention' so a persistently-failing job stays visible instead of looping forever. + `attempts` is bumped by claim_next on every claim, so it already counts tries.""" + with conn.cursor() as cur: + cur.execute('SELECT attempts, "requestId" FROM "Job" WHERE id = %s', (job_id,)) + row = cur.fetchone() + if row is None: + return + attempts, request_id = row + if attempts >= max_attempts: + cur.execute( + 'UPDATE "Job" SET state = \'needs_attention\', error = %s, "claimedAt" = NULL, ' + '"updatedAt" = now() WHERE id = %s', + (error, job_id), + ) + cur.execute('UPDATE "Request" SET status = \'needs_attention\' WHERE id = %s', (request_id,)) + else: + cur.execute( + 'UPDATE "Job" SET state = \'requested\', "currentStage" = \'intake\', error = NULL, ' + '"claimedAt" = NULL, "downloadProgress" = 0, "updatedAt" = now() WHERE id = %s', + (job_id,), + ) + cur.execute('UPDATE "Request" SET status = \'pending\' WHERE id = %s', (request_id,)) + conn.commit() + + def claim_next(conn: psycopg.Connection) -> str | None: """Atomically claim the oldest queued job. Returns its id or None.""" with conn.cursor() as cur: diff --git a/worker/lyra_worker/main.py b/worker/lyra_worker/main.py index 30b7e4b..ad7ca44 100644 --- a/worker/lyra_worker/main.py +++ b/worker/lyra_worker/main.py @@ -6,7 +6,7 @@ import uuid import psycopg -from lyra_worker.claim import claim_next, reclaim_stuck_jobs +from lyra_worker.claim import claim_next, reclaim_stuck_jobs, requeue_or_fail from lyra_worker.config import get_config from lyra_worker.crypto import secret_key_problem from lyra_worker.db import wait_for_db @@ -323,6 +323,9 @@ def run_forever() -> None: except Exception as e: # one bad job must not take down the worker loop print(f"worker: pipeline failed for job {job_id}: {e}", flush=True) conn.rollback() + # Don't orphan the job in its mid-pipeline 'matching' state (claim_next only + # picks 'requested'): requeue for a bounded retry, then needs_attention. + requeue_or_fail(conn, job_id, str(e)) finish_job(conn, job_id, mcfg) print(f"worker: finished job {job_id}", flush=True) next_job_allowed = time.monotonic() + job_delay_seconds(config) diff --git a/worker/tests/test_requeue.py b/worker/tests/test_requeue.py new file mode 100644 index 0000000..8383ca1 --- /dev/null +++ b/worker/tests/test_requeue.py @@ -0,0 +1,69 @@ +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