NixReaperandClaude Sonnet 5
cd4dd98adb
CI / lint (push) Canceled after 0s
CI / typecheck (push) Canceled after 0s
CI / unit-tests (push) Canceled after 0s
CI / integration-tests (push) Canceled after 0s
CI / container-build (push) Canceled after 0s
CI / dependency-audit (push) Canceled after 0s
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]>
80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
"""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
|