fix(importer): race-safe savepoint-based find-or-create for Source + Post (uq_source_artist_platform_url UniqueViolation operator-flagged 2026-05-26) — Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/app/services/importer.py

@@ -18,6 +18,7 @@ from pathlib import Path
 
 from PIL import Image
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
 from ..models import (
@@ -202,6 +203,80 @@ class Importer:
                 (phash, width or 0, height or 0, image_id)
             )
 
+    def _find_or_create_source(
+        self, *, artist_id: int, platform: str, url: str,
+    ) -> Source:
+        """Race-safe find-or-create on `source` keyed by
+        (artist_id, platform, url) — the same key as the
+        `uq_source_artist_platform_url` constraint.
+
+        Two concurrent workers processing different files in the same
+        post can both find no existing Source row then both INSERT,
+        which trips the unique constraint and poisons the session with
+        `psycopg.errors.UniqueViolation`. Operator-flagged 2026-05-26.
+
+        Pattern: select; if absent, open a savepoint and INSERT.
+        On IntegrityError, roll the savepoint back (NOT the outer
+        transaction, which would lose the surrounding scan's progress)
+        and re-select — the concurrent op just created the row we
+        wanted, so the second select will find it.
+        """
+        existing = 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
+        sp = self.session.begin_nested()
+        try:
+            row = Source(artist_id=artist_id, platform=platform, url=url)
+            self.session.add(row)
+            self.session.flush()
+            sp.commit()
+            return row
+        except IntegrityError:
+            sp.rollback()
+            return self.session.execute(
+                select(Source).where(
+                    Source.artist_id == artist_id,
+                    Source.platform == platform,
+                    Source.url == url,
+                )
+            ).scalar_one()
+
+    def _find_or_create_post(
+        self, *, source_id: int, external_post_id: str,
+    ) -> Post:
+        """Race-safe find-or-create on `post` keyed by
+        (source_id, external_post_id). Mirrors `_find_or_create_source`
+        — same savepoint + IntegrityError-recovery pattern."""
+        existing = self.session.execute(
+            select(Post).where(
+                Post.source_id == source_id,
+                Post.external_post_id == external_post_id,
+            )
+        ).scalar_one_or_none()
+        if existing is not None:
+            return existing
+        sp = self.session.begin_nested()
+        try:
+            row = Post(source_id=source_id, external_post_id=external_post_id)
+            self.session.add(row)
+            self.session.flush()
+            sp.commit()
+            return row
+        except IntegrityError:
+            sp.rollback()
+            return self.session.execute(
+                select(Post).where(
+                    Post.source_id == source_id,
+                    Post.external_post_id == external_post_id,
+                )
+            ).scalar_one()
+
     def import_one(self, source: Path) -> ImportResult:
         """Dispatch by kind. Media → normal pipeline. Archive → extract
         media members (one Post via the archive-adjacent sidecar) and
@@ -241,29 +316,13 @@ class Importer:
         sd = parse_sidecar(data)
         platform = sd.platform or "unknown"
         url = sd.post_url or f"sidecar:{platform}"
-        src = self.session.execute(
-            select(Source).where(
-                Source.artist_id == artist.id,
-                Source.platform == platform,
-                Source.url == url,
-            )
-        ).scalar_one_or_none()
-        if src is None:
-            src = Source(artist_id=artist.id, platform=platform, url=url)
-            self.session.add(src)
-            self.session.flush()
+        src = self._find_or_create_source(
+            artist_id=artist.id, platform=platform, url=url,
+        )
         epid = sd.external_post_id or sc.stem
-        post = self.session.execute(
-            select(Post).where(
-                Post.source_id == src.id,
-                Post.external_post_id == epid,
-            )
-        ).scalar_one_or_none()
-        if post is None:
-            post = Post(source_id=src.id, external_post_id=epid)
-            self.session.add(post)
-            self.session.flush()
-        return post
+        return self._find_or_create_post(
+            source_id=src.id, external_post_id=epid,
+        )
 
     def _capture_attachment(
         self, source: Path, *, post: Post | None = None,
@@ -705,29 +764,14 @@ class Importer:
         else:
             platform = sd.platform or "unknown"
             url = sd.post_url or f"sidecar:{platform}"
-            src = self.session.execute(
-                select(Source).where(
-                    Source.artist_id == artist.id,
-                    Source.platform == platform,
-                    Source.url == url,
-                )
-            ).scalar_one_or_none()
-            if src is None:
-                src = Source(artist_id=artist.id, platform=platform, url=url)
-                self.session.add(src)
-                self.session.flush()
+            src = self._find_or_create_source(
+                artist_id=artist.id, platform=platform, url=url,
+            )
 
         epid = sd.external_post_id or sc.stem
-        post = self.session.execute(
-            select(Post).where(
-                Post.source_id == src.id,
-                Post.external_post_id == epid,
-            )
-        ).scalar_one_or_none()
-        if post is None:
-            post = Post(source_id=src.id, external_post_id=epid)
-            self.session.add(post)
-            self.session.flush()
+        post = self._find_or_create_post(
+            source_id=src.id, external_post_id=epid,
+        )
         if sd.post_url is not None:
             post.post_url = sd.post_url
         if sd.post_title is not None:

tests/test_importer_upsert_helpers.py

@@ -0,0 +1,147 @@
+"""Tests for Importer._find_or_create_source / _find_or_create_post —
+the race-safe savepoint-based helpers that replaced the previous
+check-then-insert pattern.
+
+Operator-flagged 2026-05-26: concurrent workers processing different
+files in the same post both found no existing Source row, then both
+INSERTed, tripping uq_source_artist_platform_url and poisoning the
+session with `psycopg.errors.UniqueViolation`. The new helpers wrap
+the INSERT in a savepoint and recover from IntegrityError by
+re-selecting the row that the concurrent op committed.
+
+Tests cover:
+  - idempotent return: same (artist_id, platform, url) → same Source row
+  - idempotent return for Post: same (source_id, external_post_id) → same Post
+  - IntegrityError recovery: monkeypatched flush raises once, helper
+    finds the row a concurrent op committed
+"""
+
+from pathlib import Path
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+
+from backend.app.models import Artist, ImportSettings, Post, Source
+from backend.app.services.importer import Importer
+from backend.app.services.thumbnailer import Thumbnailer
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+def importer(db_sync, tmp_path):
+    import_root = tmp_path / "import"
+    images_root = tmp_path / "images"
+    import_root.mkdir()
+    images_root.mkdir()
+    settings = db_sync.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    ).scalar_one()
+    return Importer(
+        session=db_sync,
+        images_root=images_root,
+        import_root=import_root,
+        thumbnailer=Thumbnailer(images_root=images_root),
+        settings=settings,
+    )
+
+
+@pytest.fixture
+def artist_row(db_sync):
+    a = Artist(name="TestArtist", slug="testartist")
+    db_sync.add(a)
+    db_sync.flush()
+    return a
+
+
+def test_find_or_create_source_creates_then_returns_existing(importer, artist_row):
+    s1 = importer._find_or_create_source(
+        artist_id=artist_row.id, platform="patreon",
+        url="https://www.patreon.com/posts/test-1",
+    )
+    s2 = importer._find_or_create_source(
+        artist_id=artist_row.id, platform="patreon",
+        url="https://www.patreon.com/posts/test-1",
+    )
+    assert s1.id == s2.id
+
+
+def test_find_or_create_source_distinct_urls_yield_distinct_rows(
+    importer, artist_row,
+):
+    a = importer._find_or_create_source(
+        artist_id=artist_row.id, platform="patreon",
+        url="https://www.patreon.com/posts/a",
+    )
+    b = importer._find_or_create_source(
+        artist_id=artist_row.id, platform="patreon",
+        url="https://www.patreon.com/posts/b",
+    )
+    assert a.id != b.id
+
+
+def test_find_or_create_post_idempotent(importer, artist_row, db_sync):
+    src = importer._find_or_create_source(
+        artist_id=artist_row.id, platform="patreon",
+        url="https://www.patreon.com/posts/post-test",
+    )
+    p1 = importer._find_or_create_post(
+        source_id=src.id, external_post_id="ext-001",
+    )
+    p2 = importer._find_or_create_post(
+        source_id=src.id, external_post_id="ext-001",
+    )
+    assert p1.id == p2.id
+
+
+def test_find_or_create_source_recovers_from_integrity_error(
+    importer, artist_row, db_sync, monkeypatch,
+):
+    """Simulate the race: another worker has already inserted a Source row
+    matching our (artist_id, platform, url) just before our flush would
+    have. Our flush raises IntegrityError; the helper rolls back the
+    savepoint and re-selects, returning the row the concurrent op created.
+    """
+    canonical_url = "https://www.patreon.com/posts/race-141226276"
+    pre_existing = Source(
+        artist_id=artist_row.id, platform="patreon", url=canonical_url,
+    )
+    db_sync.add(pre_existing)
+    db_sync.flush()
+
+    # Force a fresh select within the helper to MISS the existing row by
+    # detaching it from the identity map; SQLAlchemy's first-level cache
+    # would otherwise return the pre_existing row immediately.
+    # Easier: monkeypatch the FIRST select inside the helper to return
+    # None on first call, real result on subsequent. We do that by
+    # patching session.execute with a single-shot wrapper.
+    real_execute = db_sync.execute
+    skip_count = [0]
+
+    def execute_with_first_select_miss(stmt, *args, **kwargs):
+        # Strip-down heuristic: the first SELECT issued by the helper is
+        # the existence check. Force it to return a "no row" result.
+        result = real_execute(stmt, *args, **kwargs)
+        if skip_count[0] == 0:
+            skip_count[0] += 1
+            # Wrap result so .scalar_one_or_none() returns None for this
+            # one call, then unwrap on subsequent uses.
+            class _ForcedMiss:
+                def scalar_one_or_none(self):
+                    return None
+
+                def scalar_one(self):
+                    return result.scalar_one()
+
+                def __getattr__(self, name):
+                    return getattr(result, name)
+            return _ForcedMiss()
+        return result
+
+    monkeypatch.setattr(db_sync, "execute", execute_with_first_select_miss)
+
+    recovered = importer._find_or_create_source(
+        artist_id=artist_row.id, platform="patreon", url=canonical_url,
+    )
+    assert recovered.id == pre_existing.id