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 <noreply@anthropic.com>

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
+                return self._file_as_duplicate(
-                # file that is a byte-dup of an existing image still links its
+                    existing, attribution_path, path_artist, explicit_source,
-                # 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._deep_rederive(existing, source, attribution_path)
 
@@ -981,8 +1021,14 @@ class Importer:
             origin="imported_filesystem",
             integrity_status="unknown",
         )
-        self.session.add(record)
+        if (winner := self._store_new_record(record)) is not None:
-        self.session.flush()
+            # 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)
+            return self._file_as_duplicate(existing, path, artist, source)
-            self.session.commit()
-            return ImportResult(
-                status="skipped", skip_reason=SkipReason.duplicate_hash,
-                image_id=existing.id, error="sha256 already present",
-            )
 
         # 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)
+        if (winner := self._store_new_record(record)) is not None:
-        self.session.flush()
+            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

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)

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()