fix(audit-g5b): race-poisoning in three find-or-create sites

Audit 2026-06-02 flagged three SELECT-then-INSERT sites that lost the
race under concurrent writers / recovery-sweep replays, poisoning the
outer transaction with an unrecoverable IntegrityError and crashing
the calling task. Same shape as the banked 2026-05-26 ImageProvenance
incident (see reference_scalar_one_or_none_duplicates memory).

Mirrors the importer._get_or_create pattern in all three: savepoint
via begin_nested + IntegrityError rollback to that savepoint + re-
SELECT to grab the row the other worker committed first.

- importer._capture_attachment: PostAttachment.sha256 UNIQUE. attachments.store
  is sha-addressed so both workers race to write the same on-disk path
  (shutil.copy2 + rename is idempotent), so no extra cleanup needed.

- TagService.find_or_create: partial uniqueness index on
  (name, kind, COALESCE(fandom_id, -1)). The previous docstring
  claimed "INSERT ... ON CONFLICT" but the implementation was
  SELECT-then-INSERT with no recovery.

- ArtistService.find_or_create: already had IntegrityError handling
  but did session.rollback() (unwinds the WHOLE transaction); now
  uses begin_nested + sp.rollback() so the surrounding request's
  progress isn't lost.

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

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")
 

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,