fix(importer): ImageProvenance (image_record_id, post_id) race-safe via savepoint + alembic 0021 UNIQUE — closes the SELECT-then-INSERT window that planted duplicates and broke .scalar_one_or_none() on every later deep-scan rederive (MultipleResultsFound). Migration dedupes existing rows (min(id) per pair); model gains __table_args__; gallery-filter test that seeded duplicates dropped.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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",
|
||||
)
|
||||
@@ -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(
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user