diff --git a/backend/app/services/artist_service.py b/backend/app/services/artist_service.py index e72e3c0..5be2dcb 100644 --- a/backend/app/services/artist_service.py +++ b/backend/app/services/artist_service.py @@ -10,7 +10,6 @@ from dataclasses import dataclass from sqlalchemy import and_, case, func, or_, select from sqlalchemy.dialects.postgresql import insert as pg_insert -from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from ..models import ( @@ -24,6 +23,7 @@ from ..models import ( ) from ..models.tag import image_tag from ..utils.slug import slugify +from .db_helpers import get_or_create from .gallery_service import decode_cursor, encode_cursor, thumbnail_url @@ -250,12 +250,10 @@ class ArtistService: ) async def find_or_create(self, name: str) -> tuple[Artist, bool]: - """Return (artist, created). Slug-keyed; idempotent under races. - - Audit 2026-06-02: switched from session.rollback() to a - begin_nested savepoint + IntegrityError recovery so a lost - race doesn't unwind the calling request's surrounding work. - Mirrors importer._get_or_create. + """Return (artist, created). Slug-keyed; idempotent under races via the + shared race-safe db_helpers.get_or_create (savepoint + IntegrityError + recovery). A new artist also seeds an ArtistVisit so the directory's + `+N new` badge starts at 0. """ cleaned = (name or "").strip() if not cleaned: @@ -263,12 +261,8 @@ class ArtistService: slug = slugify(cleaned) select_existing = select(Artist).where(Artist.slug == slug) - existing = (await self.session.execute(select_existing)).scalar_one_or_none() - if existing is not None: - return existing, False - sp = await self.session.begin_nested() - try: + async def _create() -> Artist: artist = Artist(name=cleaned, slug=slug) self.session.add(artist) await self.session.flush() @@ -279,13 +273,14 @@ class ArtistService: # count every image imported in the same session. self.session.add(ArtistVisit(artist_id=artist.id)) await self.session.flush() - await sp.commit() - except IntegrityError: - await sp.rollback() - existing = (await self.session.execute(select_existing)).scalar_one() - return existing, False - await self.session.commit() - return artist, True + return artist + + artist, created = await get_or_create( + self.session, select_existing, _create + ) + if created: + await self.session.commit() + return artist, created async def autocomplete(self, prefix: str, limit: int = 20) -> list[Artist]: cleaned = (prefix or "").strip() diff --git a/backend/app/services/db_helpers.py b/backend/app/services/db_helpers.py new file mode 100644 index 0000000..2b934c4 --- /dev/null +++ b/backend/app/services/db_helpers.py @@ -0,0 +1,55 @@ +"""Shared DB-access helpers for the async services. + +`get_or_create` centralizes the race-safe find-or-create dance — SELECT, then on +a miss a savepoint INSERT that recovers (NOT a full rollback) when a concurrent +worker inserted the same row first. It was hand-rolled identically in +ArtistService, TagService and ExtensionService; divergent copies are exactly how +the duplicate-row / race bugs in [[reference_scalar_one_or_none_duplicates]] crept +in, so it lives in one place now (DRY pattern sweep 2026-06-09). + +Note: this is the ASYNC sibling of `Importer._get_or_create` (sync, used by the +filesystem-import path). The two can't share an implementation across the +sync/async boundary; the importer one stays as the lone sync consumer. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from typing import TypeVar + +from sqlalchemy import Select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +T = TypeVar("T") + + +async def get_or_create( + session: AsyncSession, + select_stmt: Select, + factory: Callable[[], Awaitable[T]], +) -> tuple[T, bool]: + """Race-safe find-or-create. Returns ``(row, created)``. + + Run ``select_stmt`` (scalar_one_or_none); if a row exists, return it with + ``created=False``. Otherwise open a SAVEPOINT and ``await factory()`` — which + must add its row(s), flush, and return the primary row. On ``IntegrityError`` + (a concurrent worker inserted the same row first) roll back the SAVEPOINT — + NOT the outer transaction, which would lose the caller's surrounding work — + and re-run ``select_stmt`` (scalar_one) to return the row the other worker + created. The caller owns the outer commit. + + A UNIQUE/partial-unique constraint matching ``select_stmt``'s predicate is + required for the recovery to trip; without it a duplicate slips through. + """ + existing = (await session.execute(select_stmt)).scalar_one_or_none() + if existing is not None: + return existing, False + sp = await session.begin_nested() + try: + row = await factory() + await sp.commit() + return row, True + except IntegrityError: + await sp.rollback() + return (await session.execute(select_stmt)).scalar_one(), False diff --git a/backend/app/services/extension_service.py b/backend/app/services/extension_service.py index 25d02a8..a5a932f 100644 --- a/backend/app/services/extension_service.py +++ b/backend/app/services/extension_service.py @@ -11,11 +11,11 @@ from __future__ import annotations import re from sqlalchemy import select -from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from ..models import Artist, Source from ..utils.slug import slugify +from .db_helpers import get_or_create from .source_service import BACKFILL_MAX_CHUNKS @@ -169,24 +169,16 @@ class ExtensionService: 500 against uq_artist_slug. """ slug = slugify(raw_name) - existing = (await self.session.execute( - select(Artist).where(Artist.slug == slug) - )).scalar_one_or_none() - if existing is not None: - return existing, False - sp = await self.session.begin_nested() - try: + + async def _create() -> Artist: artist = Artist(name=raw_name, slug=slug, is_subscription=True) self.session.add(artist) await self.session.flush() - await sp.commit() - return artist, True - except IntegrityError: - await sp.rollback() - recovered = (await self.session.execute( - select(Artist).where(Artist.slug == slug) - )).scalar_one() - return recovered, False + return artist + + return await get_or_create( + self.session, select(Artist).where(Artist.slug == slug), _create + ) async def _find_or_create_source( self, *, artist_id: int, platform: str, url: str, @@ -194,17 +186,13 @@ class ExtensionService: """Race-safe — same pattern as _find_or_create_artist above. The uq_source_artist_platform_url constraint catches the duplicate insert; we roll the savepoint back and re-select.""" - existing = (await self.session.execute( - select(Source).where( - Source.artist_id == artist_id, - Source.platform == platform, - Source.url == url, - ) - )).scalar_one_or_none() - if existing is not None: - return existing, False - sp = await self.session.begin_nested() - try: + select_existing = select(Source).where( + Source.artist_id == artist_id, + Source.platform == platform, + Source.url == url, + ) + + async def _create() -> Source: # New subscription sources arm run-until-done backfill (plan #693) # so the first ticks walk the full history (otherwise gallery-dl's # exit:20 short-circuits before the archive is built). Mirrors @@ -219,16 +207,11 @@ class ExtensionService: ) self.session.add(src) await self.session.flush() - await sp.commit() - except IntegrityError: - await sp.rollback() - recovered = (await self.session.execute( - select(Source).where( - Source.artist_id == artist_id, - Source.platform == platform, - Source.url == url, - ) - )).scalar_one() - return recovered, False - await self.session.commit() - return src, True + return src + + src, created = await get_or_create( + self.session, select_existing, _create + ) + if created: + await self.session.commit() + return src, created diff --git a/backend/app/services/tag_service.py b/backend/app/services/tag_service.py index 335afc3..91dd7cb 100644 --- a/backend/app/services/tag_service.py +++ b/backend/app/services/tag_service.py @@ -7,12 +7,12 @@ from dataclasses import dataclass from sqlalchemy import and_, case, exists, func, select, text, update from sqlalchemy.dialects.postgresql import insert as pg_insert -from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from ..models import Tag, TagKind, image_tag from ..models.tag_allowlist import TagAllowlist from ..models.tag_reference_embedding import TagReferenceEmbedding +from .db_helpers import get_or_create log = logging.getLogger(__name__) @@ -130,20 +130,14 @@ class TagService: .order_by(Tag.id) .limit(1) ) - existing = (await self.session.execute(stmt)).scalar_one_or_none() - if existing: - return existing - - sp = await self.session.begin_nested() - try: + async def _create() -> Tag: new_tag = Tag(name=name, kind=kind, fandom_id=fandom_id) self.session.add(new_tag) await self.session.flush() - await sp.commit() return new_tag - except IntegrityError: - await sp.rollback() - return (await self.session.execute(stmt)).scalar_one() + + tag, _created = await get_or_create(self.session, stmt, _create) + return tag async def autocomplete( self, diff --git a/tests/test_db_helpers.py b/tests/test_db_helpers.py new file mode 100644 index 0000000..e0073b0 --- /dev/null +++ b/tests/test_db_helpers.py @@ -0,0 +1,34 @@ +"""Race-safe get_or_create helper (shared by the async services).""" + +import pytest +from sqlalchemy import select + +from backend.app.models import Artist +from backend.app.services.db_helpers import get_or_create + +pytestmark = pytest.mark.integration + + +@pytest.mark.asyncio +async def test_get_or_create_creates_then_returns_existing(db): + stmt = select(Artist).where(Artist.slug == "goc-test") + + factory_calls = {"n": 0} + + async def _make(): + factory_calls["n"] += 1 + a = Artist(name="GOC Test", slug="goc-test") + db.add(a) + await db.flush() + return a + + row1, created1 = await get_or_create(db, stmt, _make) + assert created1 is True + assert row1.slug == "goc-test" + assert factory_calls["n"] == 1 + + # Second call finds the existing row and does NOT invoke the factory. + row2, created2 = await get_or_create(db, stmt, _make) + assert created2 is False + assert row2.id == row1.id + assert factory_calls["n"] == 1 # factory not called when the row exists