Make failed ingestion jobs actually retry, not just say "retrying"

arq only auto-retries a job on its own Retry exception (or
CancelledError) -- never on an arbitrary exception a task raises.
_run_ingestion caught failures, recorded them via mark_job_failed
(which sets ingestion_jobs.status to "retrying" when attempts hasn't
hit max_attempts), and then just re-raised the original exception --
which arq treats as a terminal failure. The DB said "retrying"
forever; nothing ever put the job back on the queue.

Confirmed live: cascading Better Call Saul's 6 seasons hit the
get_or_create race fixed in the previous commit on 4 of them, and the
worker's own shutdown summary showed "4 failed, 0 retries" -- every
one of those was permanently lost despite max_attempts=5 suggesting
otherwise.

_run_ingestion now raises arq.Retry (capped exponential backoff) when
mark_job_failed left the job in RETRYING state, so arq actually
requeues it; only once attempts are exhausted does the original
exception propagate as a genuine terminal failure.

Co-Authored-By: Claude Sonnet 5 <[email protected]>

app/services/ingestion/runner.py

@@ -10,10 +10,12 @@ from __future__ import annotations
 import uuid
 from collections.abc import Awaitable, Callable
 
+from arq import Retry
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.database import session_scope
 from app.core.logging import get_logger
+from app.models.enums import IngestionJobStatus
 from app.models.provider import IngestionJob
 from app.providers.base import MetadataProvider
 from app.providers.tmdb import TMDbProvider
@@ -94,11 +96,24 @@ async def _run_ingestion(
         async with session_scope() as session:
             job = await _require_job(session, job_id)
             await mark_job_failed(session, job, str(exc))
+            attempts = job.attempts
+            still_retryable = job.status == IngestionJobStatus.RETRYING
+        if still_retryable:
+            # arq only auto-retries on its own Retry (or CancelledError),
+            # never on an arbitrary exception — mark_job_failed setting the
+            # DB status to RETRYING was, on its own, purely cosmetic; this
+            # is what actually gets the job back on the queue instead of
+            # leaving it permanently "retrying" without ever being retried.
+            raise Retry(defer=_retry_backoff_seconds(attempts)) from exc
         raise
     finally:
         await provider.aclose()
 
 
+def _retry_backoff_seconds(attempt: int) -> float:
+    return min(60.0, 2.0**attempt)
+
+
 async def run_movie_ingestion(job_id: uuid.UUID, external_id: str) -> uuid.UUID:
     return await _run_ingestion(
         job_id,

docs/architecture.md

@@ -76,6 +76,20 @@ lifecycle in `ingestion_jobs` (idempotent via a unique `idempotency_key` —
 re-triggering ingestion for the same provider + external ID reuses the same
 job record rather than double-processing).
 
+Failures actually get retried, not just labeled retryable: arq only
+auto-retries a job on its own `arq.Retry` exception (or `CancelledError`) —
+never on an arbitrary exception a task raises — so `_run_ingestion` catches
+the failure, records it, and if `ingestion_jobs.attempts` hasn't hit
+`max_attempts` yet, raises `Retry(defer=...)` (capped exponential backoff)
+so arq actually puts the job back on the queue instead of the DB saying
+"retrying" while nothing retries it. Series ingestion cascading into
+concurrent season jobs (`include_seasons`) is exactly what surfaces
+transient failures worth retrying this way: two seasons sharing a guest
+star can race to create the same `Person` row (see
+`app/repositories/reference_repository.py:_create_or_recover`, which
+recovers from that specific race directly; the arq retry here is the
+backstop for whatever it doesn't).
+
 ## Search (section 16)
 
 `app/services/search/base.py:SearchProvider` is the abstraction; API

tests/integration/test_ingestion_retry.py

@@ -0,0 +1,79 @@
+"""Tests that a failed ingestion job actually gets resubmitted to arq.
+
+arq only auto-retries a job on its own `Retry` exception (or
+CancelledError) — never on an arbitrary exception a task raises. Before
+this, app/services/ingestion/runner.py:_run_ingestion just recorded
+IngestionJob.status = "retrying" in the database and re-raised the
+original exception, which arq treated as a terminal failure: the DB said
+"retrying" forever, but nothing ever put the job back on the queue.
+"""
+
+from __future__ import annotations
+
+import uuid
+
+import pytest
+from arq import Retry
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.database import session_scope
+from app.repositories.provider_repository import (
+    create_ingestion_job,
+    get_ingestion_job,
+    get_or_create_provider,
+)
+from app.services.ingestion.runner import _run_ingestion
+
+
+class _BoomError(Exception):
+    pass
+
+
+async def _make_job(*, attempts_already_used: int = 0) -> uuid.UUID:
+    async with session_scope() as session:
+        provider = await get_or_create_provider(session, "tmdb", "TMDB")
+        job = await create_ingestion_job(
+            session,
+            provider_id=provider.id,
+            job_type="movie_sync",
+            idempotency_key=f"test:{uuid.uuid4()}",
+            entity_type="movie",
+            external_id="603",
+        )
+        job.attempts = attempts_already_used
+        await session.flush()
+        return job.id
+
+
+@pytest.mark.asyncio
+async def test_run_ingestion_raises_arq_retry_while_attempts_remain(db_session: AsyncSession):
+    job_id = await _make_job(attempts_already_used=0)
+
+    async def failing_work(session, provider):
+        raise _BoomError("simulated failure")
+
+    with pytest.raises(Retry):
+        await _run_ingestion(job_id, "test_ingestion", failing_work)
+
+    async with session_scope() as session:
+        job = await get_ingestion_job(session, job_id)
+        assert job.status.value == "retrying"
+        assert job.attempts == 1
+
+
+@pytest.mark.asyncio
+async def test_run_ingestion_raises_original_error_once_attempts_exhausted(
+    db_session: AsyncSession,
+):
+    job_id = await _make_job(attempts_already_used=4)  # default max_attempts is 5
+
+    async def failing_work(session, provider):
+        raise _BoomError("simulated failure")
+
+    with pytest.raises(_BoomError):
+        await _run_ingestion(job_id, "test_ingestion", failing_work)
+
+    async with session_scope() as session:
+        job = await get_ingestion_job(session, job_id)
+        assert job.status.value == "failed"
+        assert job.attempts == 5