From bb10d30b2db41102695194e91493d67b14afd017 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 6 Nov 2026 18:35:51 -0500 Subject: [PATCH] fix: two copies of one picture imported at once are filed as duplicates, and an archive's row always ends The sha256 lookup before inserting an image row can lose a race to another worker importing a second copy of the same picture. The unique index then refused the insert. A media import retried behind the whole import queue, and the recovery sweep ran it a second time. An archive swallowed the error, failed on the dead transaction, and its row stayed 'processing' because the failure mark could not commit either. - Importer._store_new_record inserts under a savepoint; on a conflict the file is filed as a duplicate of the row that won (_file_as_duplicate, now shared by every duplicate branch). _import_media removes its stray library copy. - _import_archive commits its post before extracting; the member loop rolls back a failed member and lets SoftTimeLimitExceeded through. - _mark_failed rolls back before writing the failure. Co-Authored-By: Claude Opus 5.5 --- backend/app/services/importer.py | 85 +++++++++---- backend/app/tasks/import_file.py | 7 +- tests/test_importer_sha_race.py | 202 +++++++++++++++++++++++++++++++ 3 files changed, 271 insertions(+), 23 deletions(-) create mode 100644 tests/test_importer_sha_race.py diff --git a/backend/app/services/importer.py b/backend/app/services/importer.py index e5960ca..4114cca 100644 --- a/backend/app/services/importer.py +++ b/backend/app/services/importer.py @@ -16,6 +16,7 @@ from dataclasses import dataclass, field from enum import StrEnum from pathlib import Path +from celery.exceptions import SoftTimeLimitExceeded from PIL import Image from sqlalchemy import func, select, update from sqlalchemy.exc import IntegrityError @@ -307,6 +308,44 @@ class Importer: sp.rollback() return self.session.execute(stmt).scalar_one() + def _store_new_record(self, record: ImageRecord) -> ImageRecord | None: + """Insert `record`, or return the row that already holds its bytes. + + The sha256 lookup before an insert can lose a race: two copies of one + picture in different folders, imported side by side by two workers, + both find no row, and the unique index refuses the second insert. Only + that insert is undone (a savepoint), so the session stays usable and + the caller files its file as a duplicate of the winner. None means + `record` was stored.""" + sp = self.session.begin_nested() + try: + self.session.add(record) + self.session.flush() + sp.commit() + return None + except IntegrityError: + sp.rollback() + winner = self.session.execute( + select(ImageRecord).where(ImageRecord.sha256 == record.sha256) + ).scalar_one_or_none() + if winner is None: + raise + return winner + + def _file_as_duplicate( + self, existing: ImageRecord, path: Path, artist: Artist | None, + explicit_source: Source | None, + ) -> ImportResult: + """A byte-identical copy of `existing`: link the copy's post to the + image (enrich-on-duplicate), so a picture posted twice shows on both + posts, and report the skip.""" + self._apply_sidecar(existing, path, artist, explicit_source=explicit_source) + self.session.commit() + return ImportResult( + status="skipped", skip_reason=SkipReason.duplicate_hash, + image_id=existing.id, error="sha256 already present", + ) + def _find_or_create_source( self, *, artist_id: int, platform: str, url: str, ) -> Source: @@ -577,6 +616,9 @@ class Importer: artist_use = artist if artist is not None else self._resolve_artist(source) post = self._post_for_sidecar(source, artist_use, source_row=source_row) + # Committed before extraction, so a member that fails (and is rolled + # back on its own) can never take the archive's post with it. + self.session.commit() member_ids: list[int] = [] # Every member image touched (new + superseded + deduped), so the # from_attachment_id stamp below covers files that already existed in the @@ -703,7 +745,15 @@ class Importer: member_record_ids.add(res.image_id) else: counts["failed"] += 1 + except SoftTimeLimitExceeded: + # The task's time budget is spent: stop here so the task marks + # its row failed, rather than carrying on until the hard kill. + raise except Exception as exc: # noqa: BLE001 — defensive per level; keep going + # Each member commits on its own, so this undoes only the one in + # progress; without it the next query fails on the dead transaction + # and the archive is never stored. + self.session.rollback() log.warning( "archive extraction failed for %s (depth %d): %s", archive_path.name, depth, exc, @@ -884,18 +934,8 @@ class Importer: existing = self.session.execute(existing_stmt).scalar_one_or_none() if existing: if not self.deep: - # Enrich-on-duplicate (parity with attach_in_place): a re-scanned - # file that is a byte-dup of an existing image still links its - # post via _apply_sidecar, so a cross-posted image shows on every - # post. - self._apply_sidecar( - existing, attribution_path, path_artist, - explicit_source=explicit_source, - ) - self.session.commit() - return ImportResult( - status="skipped", skip_reason=SkipReason.duplicate_hash, - image_id=existing.id, error="sha256 already present", + return self._file_as_duplicate( + existing, attribution_path, path_artist, explicit_source, ) return self._deep_rederive(existing, source, attribution_path) @@ -981,8 +1021,14 @@ class Importer: origin="imported_filesystem", integrity_status="unknown", ) - self.session.add(record) - self.session.flush() + if (winner := self._store_new_record(record)) is not None: + # The copy just written is a stray unless it is the winner's own + # file (the same source imported twice lands on the same path). + if Path(winner.path) != dest: + dest.unlink(missing_ok=True) + return self._file_as_duplicate( + winner, attribution_path, path_artist, explicit_source, + ) self._phash_cache_append(phash, width, height, record.id) # Folder→artist (path_artist resolved up-front above); bind it to the @@ -1310,12 +1356,7 @@ class Importer: # only sets it when NULL. Without this the new post is left with no # image AND (if its other content also deduped) no attachment, i.e. a # bare shell — operator-flagged 2026-06-08 (1589 empty Anduo posts). - self._apply_sidecar(existing, path, artist, explicit_source=source) - self.session.commit() - return ImportResult( - status="skipped", skip_reason=SkipReason.duplicate_hash, - image_id=existing.id, error="sha256 already present", - ) + return self._file_as_duplicate(existing, path, artist, source) # phash dedup + supersede phash = None @@ -1395,8 +1436,8 @@ class Importer: ) if artist is not None: record.artist_id = artist.id - self.session.add(record) - self.session.flush() + if (winner := self._store_new_record(record)) is not None: + return self._file_as_duplicate(winner, path, artist, source) self._phash_cache_append(phash, width, height, record.id) # Sidecar provenance (best-effort). When `source` is passed, link diff --git a/backend/app/tasks/import_file.py b/backend/app/tasks/import_file.py index 4a46a86..7a4a878 100644 --- a/backend/app/tasks/import_file.py +++ b/backend/app/tasks/import_file.py @@ -46,8 +46,13 @@ def _mark_failed(session, task, error_msg: str) -> None: """Best-effort flip of a 'processing' row to 'failed' + batch counter increment. Wrapped in its own try because if the DB is what just broke, this commit will also fail — that's why the maintenance sweep - exists as a backstop.""" + exists as a backstop. + + The failure being recorded may have left the session's transaction dead + (an error mid-flush), so it is rolled back first; otherwise this commit + fails on it too and the row stays 'processing'.""" try: + session.rollback() task.status = "failed" task.error = error_msg task.finished_at = datetime.now(UTC) diff --git a/tests/test_importer_sha_race.py b/tests/test_importer_sha_race.py new file mode 100644 index 0000000..98f7354 --- /dev/null +++ b/tests/test_importer_sha_race.py @@ -0,0 +1,202 @@ +"""Two workers importing the same picture at once (services/importer.py). + +The importer looks a file's sha256 up before inserting its row. When two copies +of one picture are imported side by side, both lookups can find nothing; the +unique index on sha256 then refuses the second insert. The loser files its copy +as a duplicate of the winner, links its post, and leaves the session usable, +instead of failing the import with a dead transaction. + +The race is staged the way tests/test_importer_provenance_race.py stages its +own: the loser's sha256 lookup is forced to miss once, as it would if it ran +just before the winner's commit. The loser's near-duplicate cache is empty for +the same reason (it was loaded before the winner's row existed). +""" + +import io +import zipfile +from pathlib import Path + +import pytest +from PIL import Image +from sqlalchemy import func, select + +from backend.app.models import Artist, ImageRecord, ImportSettings, PostAttachment +from backend.app.services.importer import Importer, SkipReason +from backend.app.services.thumbnailer import Thumbnailer + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def layout(tmp_path): + import_root = tmp_path / "import" + images_root = tmp_path / "images" + import_root.mkdir() + images_root.mkdir() + return import_root, images_root + + +def _worker(db_sync, layout) -> Importer: + import_root, images_root = 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, + ) + + +def _jpeg(path: Path) -> bytes: + path.parent.mkdir(parents=True, exist_ok=True) + Image.new("RGB", (320, 240), (40, 90, 160)).save(path, "JPEG") + return path.read_bytes() + + +def _copy(src: Path, dst: Path) -> Path: + dst.parent.mkdir(parents=True, exist_ok=True) + dst.write_bytes(src.read_bytes()) + return dst + + +def _lose_the_race(importer: Importer, monkeypatch) -> None: + """Make `importer` miss its first sha256 lookup and see no near-duplicates.""" + real_execute = importer.session.execute + missed = [False] + + class _Miss: + def scalar_one_or_none(self): + return None + + def execute(stmt, *args, **kwargs): + sql = str(stmt) + if not missed[0] and "FROM image_record" in sql and "image_record.sha256 =" in sql: + missed[0] = True + return _Miss() + return real_execute(stmt, *args, **kwargs) + + monkeypatch.setattr(importer.session, "execute", execute) + monkeypatch.setattr(importer, "_phash_candidates_cache", lambda: []) + + +def _records(db_sync) -> int: + return db_sync.execute(select(func.count()).select_from(ImageRecord)).scalar_one() + + +def test_a_copy_imported_alongside_the_original_is_filed_as_its_duplicate( + db_sync, layout, monkeypatch, +): + import_root, images_root = layout + first = import_root / "Alice" / "set1" / "pic.jpg" + _jpeg(first) + second = _copy(first, import_root / "Alice" / "set2" / "pic.jpg") + + winner = _worker(db_sync, layout).import_one(first) + assert winner.status == "imported" + kept = Path(db_sync.get(ImageRecord, winner.image_id).path) + + loser = _worker(db_sync, layout) + _lose_the_race(loser, monkeypatch) + result = loser.import_one(second) + + assert result.status == "skipped" + assert result.skip_reason == SkipReason.duplicate_hash + assert result.image_id == winner.image_id + monkeypatch.undo() + assert _records(db_sync) == 1 + assert kept.is_file() + # The loser's copy in the library was a stray and is gone; only the + # winner's file remains. + library_files = [p for p in images_root.rglob("*.jpg") if p.is_file()] + assert library_files == [kept] + + +def test_the_same_file_imported_twice_at_once_keeps_its_library_copy( + db_sync, layout, monkeypatch, +): + """Both runs write the same library path, so the loser's copy IS the + winner's file and must not be deleted.""" + import_root, _ = layout + src = import_root / "Alice" / "pic.jpg" + _jpeg(src) + + winner = _worker(db_sync, layout).import_one(src) + kept = Path(db_sync.get(ImageRecord, winner.image_id).path) + + loser = _worker(db_sync, layout) + _lose_the_race(loser, monkeypatch) + result = loser.import_one(src) + + assert result.skip_reason == SkipReason.duplicate_hash + assert kept.is_file() + + +def test_a_copy_attached_in_place_alongside_the_original_is_its_duplicate( + db_sync, layout, monkeypatch, +): + _, images_root = layout + artist = Artist(name="Bea", slug="bea") + db_sync.add(artist) + db_sync.commit() + first = images_root / "bea" / "patreon" / "post1" / "pic.jpg" + _jpeg(first) + second = _copy(first, images_root / "bea" / "patreon" / "post2" / "pic.jpg") + + winner = _worker(db_sync, layout).attach_in_place(first, artist=artist) + assert winner.status == "imported" + + loser = _worker(db_sync, layout) + _lose_the_race(loser, monkeypatch) + result = loser.attach_in_place(second, artist=artist) + + assert result.status == "skipped" + assert result.skip_reason == SkipReason.duplicate_hash + assert result.image_id == winner.image_id + monkeypatch.undo() + assert _records(db_sync) == 1 + # Removing the loser's file is the caller's step (discard_after_attach). + assert first.is_file() and second.is_file() + + +def test_an_archive_whose_member_breaks_the_transaction_is_still_stored( + db_sync, layout, monkeypatch, +): + """A member whose import fails mid-flush leaves the transaction dead. The + member loop rolls it back, so the archive itself is still stored and the + import ends with a result instead of an exception.""" + import_root, _ = layout + seed = _worker(db_sync, layout).import_one(_jpeg_at(import_root / "Cy" / "seed.jpg")) + seed_path = db_sync.get(ImageRecord, seed.image_id).path + + arc = import_root / "Cy" / "set.zip" + with zipfile.ZipFile(arc, "w") as zf: + zf.writestr("a.jpg", _jpeg_bytes()) + + importer = _worker(db_sync, layout) + + def breaks_the_transaction(*args, **kwargs): + importer.session.add(ImageRecord( + path=seed_path, sha256="f" * 64, size_bytes=1, mime="image/jpeg", + origin="imported_filesystem", + )) + importer.session.flush() # refused: the path is taken + + monkeypatch.setattr(importer, "_import_media", breaks_the_transaction) + result = importer.import_one(arc) + + assert result.status == "attached" + stored = db_sync.execute( + select(func.count()).select_from(PostAttachment) + ).scalar_one() + assert stored == 1 + + +def _jpeg_at(path: Path) -> Path: + _jpeg(path) + return path + + +def _jpeg_bytes() -> bytes: + buf = io.BytesIO() + Image.new("RGB", (64, 48), (200, 30, 30)).save(buf, "JPEG") + return buf.getvalue()