NixReaperandClaude Sonnet 5
71388e254e
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
Found live: running bulk_import.py's fixed 50-worker pool against the real movie export, 49 of the first 50 concurrent workers crashed immediately with "duplicate key value violates unique constraint uq_providers_slug" -- all racing to create the one-and-only "tmdb" Provider row on first run. get_or_create_provider was a plain check-then-insert with no race recovery, unlike its sibling get_or_create_* helpers in reference_repository.py, which already handle exactly this class of race (arq's own concurrency and season- ingestion cascades already exercise it there). Extracted the SAVEPOINT-based create_or_recover helper out of reference_repository.py into app/repositories/concurrency.py so both modules share it, and applied it to get_or_create_provider and create_ingestion_job (the latter's idempotency-key race is less likely under bulk_import's own access pattern -- each of its workers processes a distinct external_id -- but a concurrent admin-triggered ingestion of the same id during the weeks this bulk import runs is a realistic way to hit it too). reference_repository.py re-exports the helper under its old private name so its 8 existing call sites and test_reference_repository.py need no changes. Co-Authored-By: Claude Sonnet 5 <[email protected]>
43 lines
1.8 KiB
Python
43 lines
1.8 KiB
Python
"""Shared helper for check-then-act get-or-create races (section 15/28).
|
|
|
|
Every get_or_create_* across the repository layer follows check-then-act
|
|
(look up by natural key, insert if missing) rather than a single atomic
|
|
UPSERT. That's fine as long as nothing ingests concurrently for
|
|
overlapping data — which isn't a safe assumption once more than one
|
|
ingestion job can run at the same time (arq's own concurrency, season
|
|
ingestion cascading via `include_seasons`, and this bulk-import script's
|
|
worker pool all create exactly this situation): two concurrent callers can
|
|
both see "missing" and both try to INSERT, and the loser's INSERT hits a
|
|
unique-constraint violation instead of finding the winner's row.
|
|
|
|
`create_or_recover` wraps the insert in a SAVEPOINT and, on that specific
|
|
failure, rolls back just the savepoint and re-queries for what the
|
|
concurrent caller committed — recovering instead of crashing. Originally
|
|
lived only in `reference_repository.py`; moved here once
|
|
`provider_repository.py:get_or_create_provider` needed the exact same
|
|
protection (found live: 50 concurrent bulk-import workers all racing to
|
|
create the one-and-only "tmdb" Provider row on first run crashed 49 of
|
|
them with `uq_providers_slug` violations)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Awaitable, Callable
|
|
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
|
|
async def create_or_recover[T](
|
|
session: AsyncSession,
|
|
create: Callable[[], Awaitable[T]],
|
|
recover: Callable[[], Awaitable[T | None]],
|
|
) -> T:
|
|
try:
|
|
async with session.begin_nested():
|
|
return await create()
|
|
except IntegrityError:
|
|
existing = await recover()
|
|
if existing is None:
|
|
raise
|
|
return existing
|