diff --git a/backend/app/services/importer.py b/backend/app/services/importer.py index 0e366d9..b722f67 100644 --- a/backend/app/services/importer.py +++ b/backend/app/services/importer.py @@ -204,6 +204,32 @@ class Importer: (phash, width or 0, height or 0, image_id) ) + def _get_or_create(self, stmt, factory): + """Race-safe find-or-create. Run `stmt` (scalar_one_or_none); if a + row exists, return it. Otherwise open a savepoint and INSERT + ``factory()``; on IntegrityError (a concurrent worker inserted the + same row first) roll the savepoint back — NOT the outer transaction, + which would lose the surrounding scan's progress — and re-run `stmt` + (scalar_one) to return the row the other worker created. + + Centralizes the pattern shared by _find_or_create_source, + _source_for_sidecar, and _find_or_create_post. The plain + SELECT-then-INSERT version lost races under the 5-min recovery sweep + (operator-flagged 2026-05-26).""" + existing = self.session.execute(stmt).scalar_one_or_none() + if existing is not None: + return existing + sp = self.session.begin_nested() + try: + row = factory() + self.session.add(row) + self.session.flush() + sp.commit() + return row + except IntegrityError: + sp.rollback() + return self.session.execute(stmt).scalar_one() + def _find_or_create_source( self, *, artist_id: int, platform: str, url: str, ) -> Source: @@ -222,31 +248,15 @@ class Importer: 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() + stmt = select(Source).where( + Source.artist_id == artist_id, + Source.platform == platform, + Source.url == url, + ) + return self._get_or_create( + stmt, + lambda: Source(artist_id=artist_id, platform=platform, url=url), + ) def _source_for_sidecar( self, *, artist_id: int, platform: str, artist_slug: str, @@ -268,7 +278,7 @@ class Importer: ONE synthetic anchor with url='sidecar::' and enabled=False (so the subscription checker doesn't poll it). """ - existing = self.session.execute( + stmt = ( select(Source) .where( Source.artist_id == artist_id, @@ -276,33 +286,16 @@ class Importer: ) .order_by(Source.id.asc()) .limit(1) - ).scalar_one_or_none() - if existing is not None: - return existing - synthetic_url = f"sidecar:{platform}:{artist_slug}" - sp = self.session.begin_nested() - try: - row = Source( + ) + return self._get_or_create( + stmt, + lambda: Source( artist_id=artist_id, platform=platform, - url=synthetic_url, + url=f"sidecar:{platform}:{artist_slug}", enabled=False, - ) - 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, - ) - .order_by(Source.id.asc()) - .limit(1) - ).scalar_one() + ), + ) def _find_or_create_post( self, *, source_id: int, external_post_id: str, @@ -310,29 +303,14 @@ class Importer: """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() + stmt = select(Post).where( + Post.source_id == source_id, + Post.external_post_id == external_post_id, + ) + return self._get_or_create( + stmt, + lambda: Post(source_id=source_id, external_post_id=external_post_id), + ) def import_one(self, source: Path) -> ImportResult: """Dispatch by kind. Media → normal pipeline. Archive → extract