From 4da8d1d774988a7d8cdd14356179d42e4685986e Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 26 May 2026 07:51:12 -0400 Subject: [PATCH] =?UTF-8?q?fix(importer):=20race-safe=20savepoint-based=20?= =?UTF-8?q?find-or-create=20for=20Source=20+=20Post=20(uq=5Fsource=5Fartis?= =?UTF-8?q?t=5Fplatform=5Furl=20UniqueViolation=20operator-flagged=202026-?= =?UTF-8?q?05-26)=20=E2=80=94=20Co-Authored-By:=20Claude=20Opus=204.7=20(1?= =?UTF-8?q?M=20context)=20?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/services/importer.py | 130 +++++++++++++++-------- tests/test_importer_upsert_helpers.py | 147 ++++++++++++++++++++++++++ 2 files changed, 234 insertions(+), 43 deletions(-) create mode 100644 tests/test_importer_upsert_helpers.py diff --git a/backend/app/services/importer.py b/backend/app/services/importer.py index fb9d0e3..3223ec7 100644 --- a/backend/app/services/importer.py +++ b/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: diff --git a/tests/test_importer_upsert_helpers.py b/tests/test_importer_upsert_helpers.py new file mode 100644 index 0000000..ca8a3bc --- /dev/null +++ b/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