diff --git a/app/services/ingestion/runner.py b/app/services/ingestion/runner.py index 6e76b8c..f018597 100644 --- a/app/services/ingestion/runner.py +++ b/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, diff --git a/docs/architecture.md b/docs/architecture.md index 67a9591..3cac28a 100644 --- a/docs/architecture.md +++ b/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 diff --git a/tests/integration/test_ingestion_retry.py b/tests/integration/test_ingestion_retry.py new file mode 100644 index 0000000..61bd12d --- /dev/null +++ b/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