fix(scan): recovery sweep for stranded download events

The scan tick (scan.py:_tick_due_sources_async) inserts
DownloadEvent(status='pending') and fires download_source.delay(). If the
task dies before finalizing the event — worker OOM/SIGKILL, lost task, or
a gallery-dl that didn't unwind on the 1200s hard time_limit — the event
stays in-flight forever. Every later tick then skips the source via the
in-flight guard (scan.py:168), so Source.last_checked_at is never written
and the operator sees "last check never" in the Subscriptions health
column, permanently.

cleanup_old_download_events only prunes terminal events (by design); no
existing sweep covered the pending/running case. Operator confirmed
2026-05-29 with a diagnostic query: all 43 "never checked" sources were
stranded behind stale in-flight events (eligible_stuck_inflight = 43,
every other bucket zero).

New recover_stalled_download_events task (Beat every 5 min):
- Flips DownloadEvent rows pending/running > 30 min (10 min past the
  download_source 1200s hard kill, so legitimately-running tasks are
  never touched) to status='error' with a sentinel message.
- Bumps each affected Source's consecutive_failures ONCE per source —
  backoff is 2^N on that counter so per-event bumps would needlessly
  inflate the next interval — sets last_error, stamps last_checked_at.

UPDATE...RETURNING source_id avoids a SELECT-then-UPDATE-WHERE-IN that
would hit the psycopg 65535-param ceiling on a large strand pile.

Net: the 43 currently-stranded sources unstick on the first sweep after
deploy, their health dots flip amber instead of unchecked, and the next
scan tick re-queues them.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/app/celery_app.py

@@ -85,6 +85,10 @@ def make_celery() -> Celery:
                 "task": "backend.app.tasks.maintenance.cleanup_old_download_events",
                 "schedule": 86400.0,  # daily
             },
+            "recover-stalled-download-events": {
+                "task": "backend.app.tasks.maintenance.recover_stalled_download_events",
+                "schedule": 300.0,  # every 5 min, matches recover-interrupted-tasks
+            },
             "recover-stalled-task-runs": {
                 "task": "backend.app.tasks.maintenance.recover_stalled_task_runs",
                 "schedule": 300.0,  # every 5 min, matches recover-interrupted-tasks

backend/app/tasks/maintenance.py

@@ -10,7 +10,14 @@ from PIL import Image
 from sqlalchemy import and_, delete, or_, select, update
 
 from ..celery_app import celery
-from ..models import DownloadEvent, ImageRecord, ImportSettings, ImportTask, TaskRun
+from ..models import (
+    DownloadEvent,
+    ImageRecord,
+    ImportSettings,
+    ImportTask,
+    Source,
+    TaskRun,
+)
 from ..utils.phash import compute_phash
 from ._sync_engine import sync_session_factory as _sync_session_factory
 
@@ -34,6 +41,13 @@ ARCHIVE_STUCK_THRESHOLD_MINUTES = 40
 # flip to terminal 'failed' and never enter this loop.
 MAX_RECOVERY_ATTEMPTS = 3
 ORPHAN_PENDING_THRESHOLD_MINUTES = 30
+
+# DownloadEvent (pending|running) recovery threshold. download_source has
+# time_limit=1200s (20 min); 30 min is 10 min past that, so a legitimately-
+# running task is never killed by the sweep. Operator-confirmed 2026-05-29
+# after 43 sources stranded at "last check never" by the in-flight guard.
+DOWNLOAD_STALL_THRESHOLD_MINUTES = 30
+
 OLD_TASK_DAYS = 7
 PHASH_PAGE = 500
 VERIFY_PAGE = 200
@@ -448,6 +462,65 @@ def verify_integrity() -> int:
     return total
 
 
+@celery.task(name="backend.app.tasks.maintenance.recover_stalled_download_events")
+def recover_stalled_download_events() -> int:
+    """Recover DownloadEvent rows stuck pending/running past the worker hard kill.
+
+    The scan tick (scheduler_service.select_due_sources →
+    tasks.scan._tick_due_sources_async) inserts DownloadEvent(status='pending')
+    and fires download_source.delay(). If that task dies before finalizing the
+    event — worker OOM/SIGKILL, lost task, or a gallery-dl that didn't unwind
+    on the 1200s hard time_limit — the event stays in-flight forever. The next
+    tick then skips that source because of the in-flight guard (scan.py:168)
+    and Source.last_checked_at never updates; the operator sees "last check
+    never" in the Subscriptions health column, permanently.
+
+    This sweep flips matching events to 'error', stamps each affected Source's
+    last_checked_at + last_error and bumps consecutive_failures (once per
+    source, not per event — backoff is exponential on that count so an N-event
+    bump would inflate the next interval by 2^N for no reason). The source
+    becomes re-queueable on the next tick and the health dot goes amber.
+
+    Operator-confirmed 2026-05-29 (43-row strand pile in production).
+    """
+    SessionLocal = _sync_session_factory()
+    now = datetime.now(UTC)
+    cutoff = now - timedelta(minutes=DOWNLOAD_STALL_THRESHOLD_MINUTES)
+    msg = "stranded by recovery sweep (no terminal status after time_limit)"
+    with SessionLocal() as session:
+        # UPDATE...RETURNING the source_ids in one round trip — keeps us off
+        # the psycopg 65535-param ceiling that SELECT-then-UPDATE-WHERE-IN
+        # would hit on a large strand pile.
+        result = session.execute(
+            update(DownloadEvent)
+            .where(DownloadEvent.status.in_(["pending", "running"]))
+            .where(DownloadEvent.started_at < cutoff)
+            .values(status="error", finished_at=now, error=msg)
+            .returning(DownloadEvent.source_id)
+        )
+        returned = result.all()
+        if not returned:
+            session.commit()
+            return 0
+        events_recovered = len(returned)
+        source_ids = list({row.source_id for row in returned})
+        session.execute(
+            update(Source)
+            .where(Source.id.in_(source_ids))
+            .values(
+                consecutive_failures=Source.consecutive_failures + 1,
+                last_error=msg,
+                last_checked_at=now,
+            )
+        )
+        session.commit()
+        log.info(
+            "recover_stalled_download_events: recovered %d events across %d sources",
+            events_recovered, len(source_ids),
+        )
+        return events_recovered
+
+
 @celery.task(name="backend.app.tasks.maintenance.cleanup_old_download_events")
 def cleanup_old_download_events() -> int:
     """FC-3d: delete terminal DownloadEvent rows older than the configured

tests/test_maintenance.py

@@ -495,3 +495,146 @@ def test_prune_task_runs_never_deletes_running(db_sync):
         select(TaskRun.id).where(TaskRun.id == ancient_id)
     ).scalar_one_or_none()
     assert surviving == ancient_id
+
+
+# ---- recover_stalled_download_events ----------------------------------
+
+
+def _make_source(session, *, slug: str) -> int:
+    """Create an Artist + Source pair for the download-recovery tests."""
+    from backend.app.models import Artist, Source
+
+    artist = Artist(name=f"Artist {slug}", slug=slug)
+    session.add(artist)
+    session.flush()
+    source = Source(
+        artist_id=artist.id, platform="patreon",
+        url=f"https://example.com/{slug}",
+    )
+    session.add(source)
+    session.flush()
+    return source.id
+
+
+def test_recover_stalled_download_skips_fresh(db_sync):
+    """A pending event whose started_at is under the 30-min threshold is
+    left alone — the worker may still legitimately be processing it."""
+    from sqlalchemy import select
+
+    from backend.app.models import DownloadEvent, Source
+    from backend.app.tasks.maintenance import recover_stalled_download_events
+
+    sid = _make_source(db_sync, slug="fresh")
+    now = datetime.now(UTC)
+    db_sync.add(DownloadEvent(
+        source_id=sid, status="pending", started_at=now - timedelta(seconds=30),
+    ))
+    db_sync.commit()
+
+    recovered = recover_stalled_download_events.apply().get()
+
+    assert recovered == 0
+    db_sync.expire_all()
+    status = db_sync.execute(
+        select(DownloadEvent.status).where(DownloadEvent.source_id == sid)
+    ).scalar_one()
+    assert status == "pending"
+    failures = db_sync.execute(
+        select(Source.consecutive_failures).where(Source.id == sid)
+    ).scalar_one()
+    assert failures == 0
+
+
+def test_recover_stalled_download_flips_stale_pending(db_sync):
+    """A 2-hour-old pending event flips to error AND the source is bumped
+    (consecutive_failures, last_error, last_checked_at) so the next scan
+    tick can re-queue it (the in-flight guard no longer blocks)."""
+    from sqlalchemy import select
+
+    from backend.app.models import DownloadEvent, Source
+    from backend.app.tasks.maintenance import recover_stalled_download_events
+
+    sid = _make_source(db_sync, slug="stale-p")
+    now = datetime.now(UTC)
+    db_sync.add(DownloadEvent(
+        source_id=sid, status="pending", started_at=now - timedelta(hours=2),
+    ))
+    db_sync.commit()
+
+    recovered = recover_stalled_download_events.apply().get()
+
+    assert recovered == 1
+    db_sync.expire_all()
+    ev_row = db_sync.execute(
+        select(
+            DownloadEvent.status, DownloadEvent.finished_at, DownloadEvent.error,
+        ).where(DownloadEvent.source_id == sid)
+    ).one()
+    assert ev_row.status == "error"
+    assert ev_row.finished_at is not None
+    assert "stranded" in ev_row.error
+    src_row = db_sync.execute(
+        select(
+            Source.consecutive_failures, Source.last_error, Source.last_checked_at,
+        ).where(Source.id == sid)
+    ).one()
+    assert src_row.consecutive_failures == 1
+    assert "stranded" in src_row.last_error
+    assert src_row.last_checked_at is not None
+
+
+def test_recover_stalled_download_flips_stale_running(db_sync):
+    """'running' is the other in-flight state — recovery covers it equally."""
+    from sqlalchemy import select
+
+    from backend.app.models import DownloadEvent
+    from backend.app.tasks.maintenance import recover_stalled_download_events
+
+    sid = _make_source(db_sync, slug="stale-r")
+    now = datetime.now(UTC)
+    db_sync.add(DownloadEvent(
+        source_id=sid, status="running", started_at=now - timedelta(hours=2),
+    ))
+    db_sync.commit()
+
+    recovered = recover_stalled_download_events.apply().get()
+
+    assert recovered == 1
+    db_sync.expire_all()
+    status = db_sync.execute(
+        select(DownloadEvent.status).where(DownloadEvent.source_id == sid)
+    ).scalar_one()
+    assert status == "error"
+
+
+def test_recover_stalled_download_dedupes_per_source(db_sync):
+    """Two stale events on one source bump consecutive_failures ONCE.
+    Backoff is exponential on that counter (2^failures), so per-event bumps
+    would inflate the next check interval by 2^N for no real reason."""
+    from sqlalchemy import select
+
+    from backend.app.models import DownloadEvent, Source
+    from backend.app.tasks.maintenance import recover_stalled_download_events
+
+    sid = _make_source(db_sync, slug="dedupe")
+    now = datetime.now(UTC)
+    db_sync.add_all([
+        DownloadEvent(
+            source_id=sid, status="pending",
+            started_at=now - timedelta(hours=2),
+        ),
+        DownloadEvent(
+            source_id=sid, status="running",
+            started_at=now - timedelta(hours=3),
+        ),
+    ])
+    db_sync.commit()
+
+    recovered = recover_stalled_download_events.apply().get()
+
+    assert recovered == 2
+    db_sync.expire_all()
+    failures = db_sync.execute(
+        select(Source.consecutive_failures).where(Source.id == sid)
+    ).scalar_one()
+    assert failures == 1