feat(import-resilience L1): poison-pill circuit breaker — cap stuck-task re-queues

Layer 1 of the import-task resilience work (operator-requested
2026-05-28). The recover_interrupted_tasks sweep re-queues rows stuck
in 'processing' — correct for a worker crash, but without a cap a row
that RELIABLY hard-crashes the worker (OOM/segfault/SIGKILL on a
corrupt or oversized input) loops forever: re-queue → crash → re-queue,
burning a worker slot every 5 min. A caught exception flips to terminal
'failed' and never enters this loop; only process-killing inputs do.

- alembic 0026: import_task.recovery_count (int, default 0) +
  import_task.refetched (bool, default false — backs Layer 2).
- recover_interrupted_tasks now runs a poison-pill UPDATE FIRST: stuck
  rows whose recovery_count has already reached MAX_RECOVERY_ATTEMPTS-1
  are marked 'failed' with a diagnostic ("crashed or stalled the worker
  N times … likely a corrupt or oversized input … inspect/replace the
  file, then retry via /api/import/retry-failed") instead of re-queued.
  The re-queue pass then handles the remaining stuck rows and bumps
  recovery_count. Shared stuck_predicate (and_/or_) keeps the
  media-5min / archive-40min split.
- MAX_RECOVERY_ATTEMPTS=3 (two recoveries then give up).

The failed poison pill surfaces in the existing import-failures view
with its file path, directly answering "help me identify them."

Test test_recover_interrupted_poison_pill_caps_at_max pins both
branches: a row at the cap is failed (not re-enqueued, diagnostic
present), a row one short is re-queued + incremented.

alembic/versions/0026_import_task_recovery_count_refetched.py

@@ -0,0 +1,53 @@
+"""import_task.recovery_count + refetched — poison-pill circuit breaker
+
+Revision ID: 0026
+Revises: 0025
+Create Date: 2026-05-28
+
+Backs the import-task resilience work (operator-flagged 2026-05-28):
+
+- recovery_count: how many times recover_interrupted_tasks has
+  re-queued this row from a stuck 'processing' state. A row that
+  hard-crashes the worker (OOM / segfault on a corrupt or oversized
+  input) leaves no terminal flip, so the sweep re-queues it — and
+  without a cap it would loop forever, re-crashing the worker each
+  time. After MAX_RECOVERY_ATTEMPTS the sweep marks it 'failed' with a
+  diagnostic instead.
+
+- refetched: whether a one-shot re-download has already been attempted
+  for this task's file. Bounds the Layer-2 re-fetch remediation to a
+  single attempt so source-side corruption doesn't loop.
+
+Both default to 0 / false; additive, no backfill needed.
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0026"
+down_revision: Union[str, None] = "0025"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "import_task",
+        sa.Column(
+            "recovery_count", sa.Integer(), nullable=False,
+            server_default="0",
+        ),
+    )
+    op.add_column(
+        "import_task",
+        sa.Column(
+            "refetched", sa.Boolean(), nullable=False,
+            server_default=sa.false(),
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("import_task", "refetched")
+    op.drop_column("import_task", "recovery_count")

backend/app/models/import_task.py

@@ -8,7 +8,16 @@ been processing longer than the stuck-task threshold.
 
 from datetime import datetime
 
-from sqlalchemy import BigInteger, DateTime, ForeignKey, Integer, String, Text, func
+from sqlalchemy import (
+    BigInteger,
+    Boolean,
+    DateTime,
+    ForeignKey,
+    Integer,
+    String,
+    Text,
+    func,
+)
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from .base import Base
@@ -26,6 +35,13 @@ class ImportTask(Base):
     task_type: Mapped[str] = mapped_column(String(16), nullable=False)  # media|archive
     status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending", index=True)
 
+    # Poison-pill circuit breaker (alembic 0026). recovery_count tracks
+    # how many times the stuck-task sweep has re-queued this row; after
+    # the cap it's failed with a diagnostic instead of looping. refetched
+    # bounds the one-shot re-download remediation to a single attempt.
+    recovery_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    refetched: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+
     result_image_id: Mapped[int | None] = mapped_column(
         ForeignKey("image_record.id", ondelete="SET NULL"), nullable=True
     )

backend/app/tasks/maintenance.py

@@ -6,7 +6,7 @@ from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 from PIL import Image
-from sqlalchemy import delete, select, update
+from sqlalchemy import and_, delete, or_, select, update
 
 from ..celery_app import celery
 from ..models import DownloadEvent, ImageRecord, ImportSettings, ImportTask, TaskRun
@@ -23,6 +23,15 @@ STUCK_THRESHOLD_MINUTES = 5
 # 40 min = 5-min buffer past the archive task's hard kill.
 # Operator-flagged 2026-05-28 (target 1645019, a big archive).
 ARCHIVE_STUCK_THRESHOLD_MINUTES = 40
+
+# Poison-pill cap. After being recovered (re-queued from a stuck
+# 'processing' state) MAX_RECOVERY_ATTEMPTS-1 times, the next sweep
+# marks the row 'failed' instead of looping. 3 = two recoveries then
+# give up. A row reaches this only if it leaves NO terminal flip each
+# run — i.e. it hard-crashes the worker (OOM/segfault/SIGKILL), the
+# signature of a corrupt or oversized input. Caught exceptions already
+# flip to terminal 'failed' and never enter this loop.
+MAX_RECOVERY_ATTEMPTS = 3
 ORPHAN_PENDING_THRESHOLD_MINUTES = 30
 OLD_TASK_DAYS = 7
 PHASH_PAGE = 500
@@ -98,18 +107,54 @@ def recover_interrupted_tasks() -> int:
         # pipeline inline and can legitimately take up to its 35-min hard
         # limit, so it gets ARCHIVE_STUCK_THRESHOLD_MINUTES (40) to avoid
         # re-queueing a still-running archive.
+        stuck_predicate = and_(
+            ImportTask.status == "processing",
+            or_(
+                and_(ImportTask.task_type != "archive",
+                     ImportTask.started_at < media_cutoff),
+                and_(ImportTask.task_type == "archive",
+                     ImportTask.started_at < archive_cutoff),
+            ),
+        )
+
+        # POISON-PILL CIRCUIT BREAKER (Layer 1, 2026-05-28). A row that
+        # leaves no terminal flip (hard worker crash: OOM/segfault/SIGKILL
+        # on a corrupt or oversized input) gets re-queued by this sweep —
+        # and would loop forever, re-crashing the worker each pass,
+        # without a cap. Once a row has already been recovered
+        # MAX_RECOVERY_ATTEMPTS-1 times, stop re-queueing it and mark it
+        # 'failed' with a diagnostic so the operator can find + replace
+        # the offending file. This UPDATE runs FIRST so the rows it
+        # claims drop out of 'processing' before the re-queue pass.
+        poison_result = session.execute(
+            update(ImportTask)
+            .where(stuck_predicate)
+            .where(ImportTask.recovery_count >= MAX_RECOVERY_ATTEMPTS - 1)
+            .values(
+                status="failed",
+                finished_at=now,
+                error=(
+                    f"crashed or stalled the worker {MAX_RECOVERY_ATTEMPTS} "
+                    f"times without completing — likely a corrupt or "
+                    f"oversized input. Not re-queued. Inspect/replace the "
+                    f"file, then retry via /api/import/retry-failed."
+                ),
+            )
+            .returning(ImportTask.id)
+        )
+        poison_ids = [r[0] for r in poison_result.all()]
+
+        # Re-queue the remaining stuck rows (under the cap) and bump
+        # their recovery_count. RETURNING (id, task_type) so the requeue
+        # routes media vs archive correctly.
         stuck_result = session.execute(
             update(ImportTask)
-            .where(ImportTask.status == "processing")
-            .where(
-                ((ImportTask.task_type != "archive")
-                 & (ImportTask.started_at < media_cutoff))
-                | ((ImportTask.task_type == "archive")
-                   & (ImportTask.started_at < archive_cutoff))
-            )
+            .where(stuck_predicate)
+            .where(ImportTask.recovery_count < MAX_RECOVERY_ATTEMPTS - 1)
             .values(
                 status="queued",
                 started_at=None,
+                recovery_count=ImportTask.recovery_count + 1,
                 error="recovered from stuck state",
             )
             .returning(ImportTask.id, ImportTask.task_type)
@@ -138,7 +183,7 @@ def recover_interrupted_tasks() -> int:
             for tid, task_type in stuck:
                 enqueue_import(tid, task_type)
 
-        return len(stuck) + orphan_count
+        return len(stuck) + len(poison_ids) + orphan_count
 
 
 @celery.task(name="backend.app.tasks.maintenance.cleanup_old_tasks")

tests/test_maintenance.py

@@ -164,6 +164,52 @@ def test_recover_interrupted_handles_both_stuck_and_orphans(db_sync, monkeypatch
     assert dispatched == [stuck.id]  # stuck rows re-enqueue; orphans don't
 
 
+def test_recover_interrupted_poison_pill_caps_at_max(db_sync, monkeypatch):
+    """A stuck row that's already been recovered MAX_RECOVERY_ATTEMPTS-1
+    times is marked 'failed' (with a diagnostic) instead of re-queued —
+    the circuit breaker against an input that hard-crashes the worker
+    every run. Operator-flagged 2026-05-28."""
+    from backend.app.tasks import import_file
+    from backend.app.tasks.maintenance import (
+        MAX_RECOVERY_ATTEMPTS,
+        recover_interrupted_tasks,
+    )
+    dispatched: list[int] = []
+    monkeypatch.setattr(
+        import_file.import_media_file, "delay", dispatched.append
+    )
+
+    batch_id = _make_batch(db_sync)
+    now = datetime.now(UTC)
+
+    # At the cap already (recovered MAX-1 times) → fail, don't re-queue.
+    poison = ImportTask(
+        batch_id=batch_id, source_path="/import/poison.jpg", task_type="media",
+        status="processing", started_at=now - timedelta(hours=2),
+        recovery_count=MAX_RECOVERY_ATTEMPTS - 1,
+    )
+    # One recovery short of the cap → re-queue + increment.
+    recoverable = ImportTask(
+        batch_id=batch_id, source_path="/import/ok.jpg", task_type="media",
+        status="processing", started_at=now - timedelta(hours=2),
+        recovery_count=MAX_RECOVERY_ATTEMPTS - 2,
+    )
+    db_sync.add_all([poison, recoverable])
+    db_sync.commit()
+
+    touched = recover_interrupted_tasks.apply().get()
+    assert touched == 2  # one failed + one re-queued
+
+    db_sync.refresh(poison)
+    db_sync.refresh(recoverable)
+    assert poison.status == "failed"
+    assert "corrupt or" in (poison.error or "")
+    assert recoverable.status == "queued"
+    assert recoverable.recovery_count == MAX_RECOVERY_ATTEMPTS - 1
+    # Only the recoverable row re-enqueues; the poison pill does not.
+    assert dispatched == [recoverable.id]
+
+
 def test_cleanup_old_deletes_finished_old(db_sync):
     batch_id = _make_batch(db_sync)
     now = datetime.now(UTC)