diff --git a/backend/app/services/artist_service.py b/backend/app/services/artist_service.py index 0947411..1203f51 100644 --- a/backend/app/services/artist_service.py +++ b/backend/app/services/artist_service.py @@ -208,27 +208,32 @@ class ArtistService: ) async def find_or_create(self, name: str) -> tuple[Artist, bool]: - """Return (artist, created). Slug-keyed; idempotent under races.""" + """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. + """ cleaned = (name or "").strip() if not cleaned: raise ValueError("artist name must not be empty") slug = slugify(cleaned) - existing = (await self.session.execute( - select(Artist).where(Artist.slug == slug) - )).scalar_one_or_none() + 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 - artist = Artist(name=cleaned, slug=slug) - self.session.add(artist) + sp = await self.session.begin_nested() try: + artist = Artist(name=cleaned, slug=slug) + self.session.add(artist) await self.session.flush() + await sp.commit() except IntegrityError: - await self.session.rollback() - existing = (await self.session.execute( - select(Artist).where(Artist.slug == slug) - )).scalar_one() + await sp.rollback() + existing = (await self.session.execute(select_existing)).scalar_one() return existing, False await self.session.commit() return artist, True diff --git a/backend/app/services/importer.py b/backend/app/services/importer.py index 13614c3..72713c5 100644 --- a/backend/app/services/importer.py +++ b/backend/app/services/importer.py @@ -374,10 +374,19 @@ class Importer: artist = self._resolve_artist(source) post = self._post_for_sidecar(source, artist) sha = _sha256_of(source) - existing = self.session.execute( - select(PostAttachment).where(PostAttachment.sha256 == sha) - ).scalar_one_or_none() - if existing is None: + select_existing = select(PostAttachment).where(PostAttachment.sha256 == sha) + existing = self.session.execute(select_existing).scalar_one_or_none() + if existing is not None: + self.session.commit() + return ImportResult(status="attached") + # Savepoint + IntegrityError recovery — PostAttachment.sha256 is + # UNIQUE, so two workers can both pass the SELECT and only the + # second INSERT fails. Without savepoint, the outer transaction + # poisons and the calling task crashes. attachments.store is + # sha-addressed so both workers race to write the same target + # path; shutil.copy2 + rename is idempotent. Audit 2026-06-02. + sp = self.session.begin_nested() + try: stored = self.attachments.store(source, sha) self.session.add(PostAttachment( post_id=post.id if post else None, @@ -390,6 +399,11 @@ class Importer: size_bytes=source.stat().st_size, )) self.session.flush() + sp.commit() + except IntegrityError: + sp.rollback() + # Lost the race — the other worker's row is canonical. + self.session.execute(select_existing).scalar_one() self.session.commit() return ImportResult(status="attached") diff --git a/backend/app/services/tag_service.py b/backend/app/services/tag_service.py index da9e499..efb0add 100644 --- a/backend/app/services/tag_service.py +++ b/backend/app/services/tag_service.py @@ -5,6 +5,7 @@ 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 @@ -86,9 +87,12 @@ class TagService: f"fandom_id {fandom_id} does not reference a fandom tag" ) - # Upsert via INSERT ... ON CONFLICT DO NOTHING. We can't use the - # uniqueness index name directly (it's a partial coalesce-based - # expression), so we re-select after insert. + # Audit 2026-06-02: race-safe upsert via savepoint + + # IntegrityError recovery. The partial uniqueness index on + # (name, kind, COALESCE(fandom_id, -1)) catches concurrent + # inserts; without the savepoint the outer transaction would + # poison and the calling request crashes. Mirrors + # importer._get_or_create. stmt = ( select(Tag) .where(Tag.name == name) @@ -101,10 +105,16 @@ class TagService: if existing: return existing - new_tag = Tag(name=name, kind=kind, fandom_id=fandom_id) - self.session.add(new_tag) - await self.session.flush() - return new_tag + sp = await self.session.begin_nested() + try: + 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() async def autocomplete( self,