From 8d75ade1d5e5dc0a95979629c482611f0fd299ab Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 2 Jun 2026 17:02:12 -0400 Subject: [PATCH] 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 | 25 +++++++++++++++---------- backend/app/services/importer.py | 22 ++++++++++++++++++---- backend/app/services/tag_service.py | 24 +++++++++++++++++------- 3 files changed, 50 insertions(+), 21 deletions(-) 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,