fix(worker): requeue jobs that fail on an unhandled pipeline exception

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>

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:

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)

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