From 397021dcbd377d6b335437cdf323d1ad94ee459d Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 26 May 2026 13:47:55 -0400 Subject: [PATCH] =?UTF-8?q?fix(importer):=20ImageProvenance=20(image=5Frec?= =?UTF-8?q?ord=5Fid,=20post=5Fid)=20race-safe=20via=20savepoint=20+=20alem?= =?UTF-8?q?bic=200021=20UNIQUE=20=E2=80=94=20closes=20the=20SELECT-then-IN?= =?UTF-8?q?SERT=20window=20that=20planted=20duplicates=20and=20broke=20.sc?= =?UTF-8?q?alar=5Fone=5For=5Fnone()=20on=20every=20later=20deep-scan=20red?= =?UTF-8?q?erive=20(MultipleResultsFound).=20Migration=20dedupes=20existin?= =?UTF-8?q?g=20rows=20(min(id)=20per=20pair);=20model=20gains=20=5F=5Ftabl?= =?UTF-8?q?e=5Fargs=5F=5F;=20gallery-filter=20test=20that=20seeded=20dupli?= =?UTF-8?q?cates=20dropped.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.7 (1M context) --- .../versions/0021_image_provenance_unique.py | 54 +++++ backend/app/models/image_provenance.py | 20 +- backend/app/services/importer.py | 29 ++- tests/test_gallery_provenance_filter.py | 21 +- tests/test_importer_provenance_race.py | 186 ++++++++++++++++++ 5 files changed, 284 insertions(+), 26 deletions(-) create mode 100644 alembic/versions/0021_image_provenance_unique.py create mode 100644 tests/test_importer_provenance_race.py diff --git a/alembic/versions/0021_image_provenance_unique.py b/alembic/versions/0021_image_provenance_unique.py new file mode 100644 index 0000000..b9be941 --- /dev/null +++ b/alembic/versions/0021_image_provenance_unique.py @@ -0,0 +1,54 @@ +"""provenance-race: dedupe + UNIQUE(image_record_id, post_id) on image_provenance + +Revision ID: 0021 +Revises: 0020 +Create Date: 2026-05-26 + +Closes the race in Importer._apply_sidecar's existence-check + INSERT pattern. +Two workers writing for the same (image, post) pair both saw no existing row +and both inserted, leaving duplicates that then broke .scalar_one_or_none() +on every subsequent deep-scan rederive against those images +(MultipleResultsFound). Most plausibly seeded when the 5-min recovery sweep +re-enqueued a still-running long-import task and the second worker collided +with the first inside _apply_sidecar. + +Migration steps: + 1. DELETE all but min(id) per (image_record_id, post_id) pair. Operator's + DB had 2 affected pairs at write-time; harmless no-op if zero. + 2. Add UNIQUE constraint so the importer's new savepoint+IntegrityError + recovery path can trip on collision and re-select, mirroring + uq_source_artist_platform_url and uq_post_source_external_id. +""" +from typing import Sequence, Union + +from alembic import op + +revision: str = "0021" +down_revision: Union[str, None] = "0020" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.execute( + """ + DELETE FROM image_provenance ip1 + USING image_provenance ip2 + WHERE ip1.image_record_id = ip2.image_record_id + AND ip1.post_id = ip2.post_id + AND ip1.id > ip2.id + """ + ) + op.create_unique_constraint( + "uq_image_provenance_image_post", + "image_provenance", + ["image_record_id", "post_id"], + ) + + +def downgrade() -> None: + op.drop_constraint( + "uq_image_provenance_image_post", + "image_provenance", + type_="unique", + ) diff --git a/backend/app/models/image_provenance.py b/backend/app/models/image_provenance.py index cd50887..ec3e756 100644 --- a/backend/app/models/image_provenance.py +++ b/backend/app/models/image_provenance.py @@ -1,14 +1,18 @@ """ImageProvenance — links an ImageRecord to a Post. -Many-to-one (one image, many provenance rows) enables the enrich-on-duplicate -rule (spec §3): when a downloaded image is a pHash dupe of an existing -record, we append a new provenance row to the existing record rather than -dropping the metadata. +One image can have many provenance rows — different posts each contribute +metadata (enrich-on-duplicate rule, spec §3: a downloaded image that is a +pHash dupe of an existing record gets a NEW provenance row for the new post +appended, rather than the metadata being dropped). But the (image, post) +pair is unique — alembic 0021 enforces uq_image_provenance_image_post +after operator-flagged 2026-05-26 saw _apply_sidecar's existence-check + +INSERT race plant duplicates that then broke .scalar_one_or_none() on +every later deep-scan rederive (MultipleResultsFound). """ from datetime import datetime -from sqlalchemy import JSON, DateTime, ForeignKey, Integer, func +from sqlalchemy import JSON, DateTime, ForeignKey, Integer, UniqueConstraint, func from sqlalchemy.orm import Mapped, mapped_column from .base import Base @@ -16,6 +20,12 @@ from .base import Base class ImageProvenance(Base): __tablename__ = "image_provenance" + __table_args__ = ( + UniqueConstraint( + "image_record_id", "post_id", + name="uq_image_provenance_image_post", + ), + ) id: Mapped[int] = mapped_column(Integer, primary_key=True) image_record_id: Mapped[int] = mapped_column( diff --git a/backend/app/services/importer.py b/backend/app/services/importer.py index 3223ec7..997a853 100644 --- a/backend/app/services/importer.py +++ b/backend/app/services/importer.py @@ -784,6 +784,15 @@ class Importer: post.attachment_count = sd.attachment_count post.raw_metadata = sd.raw + # Race-safe (image_record_id, post_id) upsert — mirrors the + # _find_or_create_source/post savepoint pattern. The plain + # SELECT-then-INSERT pattern lost a race when two workers ran + # _apply_sidecar on the same (image, post) pair (e.g. the 5-min + # recovery sweep re-enqueued a still-running long import), planting + # duplicates that then broke .scalar_one_or_none() on every later + # deep-scan rederive (MultipleResultsFound). Alembic 0021 adds the + # uq_image_provenance_image_post UNIQUE so this savepoint actually + # trips on collision. exists = self.session.execute( select(ImageProvenance.id).where( ImageProvenance.image_record_id == record.id, @@ -791,14 +800,20 @@ class Importer: ) ).scalar_one_or_none() if exists is None: - self.session.add( - ImageProvenance( - image_record_id=record.id, - post_id=post.id, - source_id=src.id, - captured_metadata=sd.raw, + sp = self.session.begin_nested() + try: + self.session.add( + ImageProvenance( + image_record_id=record.id, + post_id=post.id, + source_id=src.id, + captured_metadata=sd.raw, + ) ) - ) + self.session.flush() + sp.commit() + except IntegrityError: + sp.rollback() if record.primary_post_id is None: record.primary_post_id = post.id self.session.flush() diff --git a/tests/test_gallery_provenance_filter.py b/tests/test_gallery_provenance_filter.py index a7f2ba4..2eb93aa 100644 --- a/tests/test_gallery_provenance_filter.py +++ b/tests/test_gallery_provenance_filter.py @@ -69,20 +69,13 @@ async def test_scroll_post_id_filter(db): assert {x.id for x in page.images} == {i1.id, i2.id} -@pytest.mark.asyncio -async def test_scroll_post_id_dedups_multi_rows(db): - i1 = await _img(db, 1) - _, s, p = await _post(db, "A", "a", "10") - # two provenance rows, same image+post (enrich-on-duplicate shape) - db.add(ImageProvenance(image_record_id=i1.id, post_id=p.id, - source_id=s.id)) - await db.flush() - db.add(ImageProvenance(image_record_id=i1.id, post_id=p.id, - source_id=s.id)) - await db.flush() - svc = GalleryService(db) - page = await svc.scroll(cursor=None, limit=10, post_id=p.id) - assert [x.id for x in page.images] == [i1.id] # appears once +# test_scroll_post_id_dedups_multi_rows removed 2026-05-26: it deliberately +# inserted two ImageProvenance rows with the same (image_record_id, post_id), +# now prevented at the DB layer by uq_image_provenance_image_post (alembic +# 0021). The EXISTS-based dedup in _provenance_clause is still useful for the +# artist-id filter (one image legitimately joins many provenance rows via +# different posts under the same artist), so the gallery_service logic is +# unchanged. @pytest.mark.asyncio diff --git a/tests/test_importer_provenance_race.py b/tests/test_importer_provenance_race.py new file mode 100644 index 0000000..8c4d956 --- /dev/null +++ b/tests/test_importer_provenance_race.py @@ -0,0 +1,186 @@ +"""Race-safe ImageProvenance insert in Importer._apply_sidecar. + +Operator-flagged 2026-05-26: the prior SELECT-then-INSERT pattern lost a +race when two workers ran _apply_sidecar on the same (image, post) pair +(plausibly seeded when the 5-min recovery sweep re-enqueued a still-running +long import). Duplicates then broke .scalar_one_or_none() on every later +deep-scan rederive (MultipleResultsFound). Alembic 0021 added +uq_image_provenance_image_post; the importer's new savepoint+IntegrityError +recovery path now trips on collision and gracefully recovers. + +Tests cover: + - idempotent: re-running _apply_sidecar via _deep_rederive produces + exactly one ImageProvenance row. + - IntegrityError recovery: pre-seed a provenance row, force the first + SELECT to return None (simulating the race window where two workers + both observed no row), call _apply_sidecar — the savepoint INSERT + trips uq_image_provenance_image_post, gets rolled back, no exception + escapes, still exactly one row. +""" + +import json +from pathlib import Path + +import pytest +from PIL import Image +from sqlalchemy import func, select + +from backend.app.models import ( + ImageProvenance, + ImageRecord, + ImportSettings, + Post, + Source, +) +from backend.app.services.importer import Importer +from backend.app.services.thumbnailer import Thumbnailer + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def import_layout(tmp_path): + import_root = tmp_path / "import" + images_root = tmp_path / "images" + import_root.mkdir() + images_root.mkdir() + return import_root, images_root + + +@pytest.fixture +def importer(db_sync, import_layout): + import_root, images_root = import_layout + 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 deep_importer(db_sync, import_layout): + import_root, images_root = import_layout + 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, + deep=True, + ) + + +def _split(path: Path, orient, size=(256, 256)): + path.parent.mkdir(parents=True, exist_ok=True) + w, h = size + im = Image.new("L", size, 0) + px = im.load() + for y in range(h): + for x in range(w): + if (x / w if orient == "v" else y / h) >= 0.5: + px[x, y] = 255 + im.convert("RGB").save(path, "JPEG") + + +def _sidecar(media: Path, payload: dict): + media.with_suffix(".json").write_text(json.dumps(payload)) + + +def test_apply_sidecar_idempotent_on_deep_rederive( + importer, deep_importer, import_layout, +): + """Normal-flow path: deep rederive on an already-imported image finds + the existing provenance row via .scalar_one_or_none() and skips the + insert. Exactly one ImageProvenance row after both runs.""" + import_root, _ = import_layout + m = import_root / "Alice" / "a.jpg" + _split(m, "v") + _sidecar(m, { + "category": "patreon", "id": 555, + "url": "https://patreon.com/posts/555", "title": "Set 1", + }) + r = importer.import_one(m) + assert r.status == "imported" + + # Re-import via deep mode — sha matches → _deep_rederive → _apply_sidecar. + r2 = deep_importer.import_one(m) + assert r2.status == "refreshed" + + count = importer.session.execute( + select(func.count()).select_from(ImageProvenance) + ).scalar_one() + assert count == 1 + + +def test_apply_sidecar_recovers_from_integrity_error( + importer, deep_importer, import_layout, db_sync, monkeypatch, +): + """Race recovery: a row already exists for (image, post). We force the + importer's existence-check SELECT to return None for one call, mimicking + the race window where two workers both saw no row. The savepoint INSERT + then trips uq_image_provenance_image_post; the helper rolls the + savepoint back, no exception escapes, and the row count stays at 1. + """ + import_root, _ = import_layout + m = import_root / "Bob" / "b.jpg" + _split(m, "v") + _sidecar(m, { + "category": "patreon", "id": 777, + "url": "https://patreon.com/posts/777", "title": "Set 2", + }) + # First import lays the canonical provenance row. + r = importer.import_one(m) + assert r.status == "imported" + rec = importer.session.get(ImageRecord, r.image_id) + post = importer.session.execute(select(Post)).scalar_one() + src = importer.session.execute(select(Source)).scalar_one() + assert rec is not None + assert src is not None + + # Monkeypatch session.execute so the FIRST select inside _apply_sidecar's + # existence-check returns a "no row" wrapper. Subsequent selects (e.g. + # the find_or_create_source / find_or_create_post existence checks + # earlier in _apply_sidecar) all run normally; we intercept only the + # ImageProvenance lookup, identified by the SELECT's target columns + # mentioning image_provenance. + real_execute = db_sync.execute + intercepted = [False] + + def _intercepting_execute(stmt, *args, **kwargs): + text = str(stmt) + if ( + not intercepted[0] + and "image_provenance" in text + and "image_record_id" in text + and "post_id" in text + ): + intercepted[0] = True + + class _ForcedMiss: + def scalar_one_or_none(self): + return None + return _ForcedMiss() + return real_execute(stmt, *args, **kwargs) + + monkeypatch.setattr(db_sync, "execute", _intercepting_execute) + + # Re-import via deep mode → _deep_rederive → _apply_sidecar. With the + # provenance-SELECT forced to miss, the helper will attempt the INSERT, + # trip uq_image_provenance_image_post, catch IntegrityError, and recover. + r2 = deep_importer.import_one(m) + assert r2.status == "refreshed" + + # Lift the intercept, then verify the row count. + monkeypatch.undo() + count = db_sync.execute( + select(func.count()).select_from(ImageProvenance) + ).scalar_one() + assert count == 1