diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 42f72ec..eba1a64 100644 --- a/backend/app/celery_app.py +++ b/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 diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 8bd321b..1c4cf22 100644 --- a/backend/app/tasks/maintenance.py +++ b/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 diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py index 0f5164e..58ab626 100644 --- a/tests/test_maintenance.py +++ b/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