From 96e984cded11d919eb2e58cb3242aaa9af10f2e3 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sun, 14 Jun 2026 13:44:07 -0400 Subject: [PATCH] feat(external): download worker for file-host links (Phase 4b) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tasks/external.py drives the external_link ledger: - fetch_external_link(link_id): atomic claim (pending/failed→downloading, so a duplicate enqueue no-ops), per-host Redis serialize lock (#720 pattern; requeue-with-countdown if busy), fetch via external_fetch into the artist library tree, then route each file through importer.attach_in_place via a synthesized sidecar so it links to the SAME post (archive→ImageRecords, else→PostAttachment; on-disk original removed for captured files, art stays); thumbnail+ML enqueue for new images; status downloaded | failed | dead with attempts/last_error/completed_at/duration. - sweep_external_links(): enqueue a bounded batch of actionable links. - recover_external_links() + prune_external_links(): recovery + retention (#89). - per-host enable read via getattr (forward-compatible; Settings UI adds the columns in 4d — defaults on, rule #26). Wiring: celery include + route (download lane) + beat (sweep 10m, recover + prune daily); download_service phase 3 enqueues a sweep after recording links. Integration tests: download+attach, failure, dead-letter, non-claimable, sweep. mega still needs the MEGAcmd binary in the runtime image (Phase 4c). Refs #830. Co-Authored-By: Claude Opus 4.8 (1M context) --- backend/app/celery_app.py | 19 ++ backend/app/services/download_service.py | 7 + backend/app/tasks/external.py | 292 +++++++++++++++++++++++ tests/test_external_worker.py | 137 +++++++++++ 4 files changed, 455 insertions(+) create mode 100644 backend/app/tasks/external.py create mode 100644 tests/test_external_worker.py diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 2d39f8f..0576d85 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -30,6 +30,7 @@ def make_celery() -> Celery: "backend.app.tasks.maintenance", "backend.app.tasks.ml", "backend.app.tasks.download", + "backend.app.tasks.external", "backend.app.tasks.backup", "backend.app.tasks.admin", "backend.app.tasks.library_audit", @@ -42,6 +43,9 @@ def make_celery() -> Celery: "backend.app.tasks.ml.*": {"queue": "ml"}, "backend.app.tasks.thumbnail.*": {"queue": "thumbnail"}, "backend.app.tasks.download.*": {"queue": "download"}, + # External file-host fetches are downloads — same lane (they can run + # long, but the download worker already tolerates long backfills). + "backend.app.tasks.external.*": {"queue": "download"}, "backend.app.tasks.scan.*": {"queue": "scan"}, # `maintenance` is the QUICK lane — recovery sweeps, vacuum, cleanup # (concurrency-1 on the scheduler). The long one-shots (DB backups, @@ -150,6 +154,21 @@ def make_celery() -> Celery: "task": "backend.app.tasks.thumbnail.backfill_thumbnails", "schedule": 86400.0, }, + # External file-host downloads (#830): a steady sweep catches links + # the post-download hook missed (worker down, etc.); recovery re-tries + # dead links daily; retention prunes long-dead rows. + "extdl-sweep": { + "task": "backend.app.tasks.external.sweep_external_links", + "schedule": 600.0, # every 10 min + }, + "extdl-recover-daily": { + "task": "backend.app.tasks.external.recover_external_links", + "schedule": 86400.0, + }, + "extdl-prune-daily": { + "task": "backend.app.tasks.external.prune_external_links", + "schedule": 86400.0, + }, }, timezone="UTC", ) diff --git a/backend/app/services/download_service.py b/backend/app/services/download_service.py index a913d83..479013f 100644 --- a/backend/app/services/download_service.py +++ b/backend/app/services/download_service.py @@ -396,6 +396,13 @@ class DownloadService: await loop.run_in_executor(None, _upsert) + # Kick the off-platform file-host downloader for any links this run + # recorded (mega/gdrive/…). Global + idempotent (only claims pending/ + # retryable rows); the beat sweep is the backstop. Lazy import dodges a + # task-module import cycle. + from ..tasks.external import sweep_external_links + sweep_external_links.delay() + ev = (await self.async_session.execute( select(DownloadEvent).where(DownloadEvent.id == event_id) )).scalar_one() diff --git a/backend/app/tasks/external.py b/backend/app/tasks/external.py new file mode 100644 index 0000000..af9ddd4 --- /dev/null +++ b/backend/app/tasks/external.py @@ -0,0 +1,292 @@ +"""Celery worker for off-platform file-host downloads (external_link ledger). + +Walks the external_link table and fetches each pending/retryable link via +external_fetch, then routes the downloaded file(s) through the existing importer +(attach_in_place) so an archive becomes ImageRecords and any other file a +PostAttachment — linked to the SAME post the link came from (via a synthesized +sidecar). Long-running-task hygiene (rule #89): per-fetch wall-clock timeout, +attempt tracking + dead-letter, a recovery sweep, and retention of dead rows. + +Concurrency: a per-host Redis lock serializes fetches to one-per-host (mega / +gdrive ban-avoidance), and an atomic status claim (pending/failed → +downloading) stops two workers grabbing the same link. + +Files land in the artist's library tree (so an attached art image stays in +place, mirroring the gallery-dl download path); a captured archive/attachment is +copied into its store and the on-disk original removed. +""" +from __future__ import annotations + +import json +import logging +import time +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import redis +from sqlalchemy import delete, select, update + +from ..celery_app import celery +from ..config import get_config +from ..models import Artist, ExternalLink, ImportSettings, Post, Source +from ..services.external_fetch import fetch_external +from ..services.importer import Importer +from ..services.thumbnailer import Thumbnailer +from ._sync_engine import sync_session_factory as _sync_session_factory + +log = logging.getLogger(__name__) + +IMAGES_ROOT = Path("/images") + +# After this many failed attempts a link is dead-lettered (skipped by routine +# sweeps; an operator recovery still re-attempts). Mirrors the ingester ledger. +DEAD_LETTER_THRESHOLD = 3 +# Per-fetch wall-clock budget — films/packs are large; the celery time_limit is +# the backstop above this. +_FETCH_TIMEOUT = 3000.0 +# Links enqueued per sweep — bounds the burst when a big backfill records many. +_SWEEP_BATCH = 50 +# Dead rows older than this are pruned (retention). +_RETENTION_DAYS = 30 +# Per-host serialize lock. +_LOCK_PREFIX = "fc:extdl_lock:" +_LOCK_TTL = 3600 +_SERIALIZE_COUNTDOWN = 120 +_MAX_SERIALIZE_WAITS = 30 + +_redis_client: redis.Redis | None = None + + +def _redis() -> redis.Redis: + global _redis_client + if _redis_client is None: + _redis_client = redis.from_url(get_config().celery_broker_url) + return _redis_client + + +def _host_enabled(settings: ImportSettings, host: str) -> bool: + """Per-host enable flag — defaults True (rule #26: works out of the box). + The Settings UI slice adds real columns; getattr keeps this forward- + compatible without a migration in this slice.""" + return bool(getattr(settings, f"extdl_{host}_enabled", True)) + + +def _write_link_sidecar(file: Path, post: Post, platform: str, artist: Artist) -> None: + """Sidecar next to a fetched file so the importer links it to `post` + (category+id resolve the source+post; matches gallery-dl's sidecar shape).""" + data = { + "category": platform, + "id": post.external_post_id, + "url": post.post_url, + "title": post.post_title, + "artist": artist.name, + } + Path(str(file) + ".json").write_text(json.dumps(data)) + + +@celery.task( + name="backend.app.tasks.external.fetch_external_link", + bind=True, + soft_time_limit=3300, + time_limit=3600, +) +def fetch_external_link(self, link_id: int, _serialize_waits: int = 0) -> dict: + """Fetch one external_link and route its file(s) through the importer.""" + SessionLocal = _sync_session_factory() + + # Atomic claim: only an actionable (pending/failed) row transitions to + # downloading, so a duplicate enqueue (sweep + post-download hook) no-ops. + with SessionLocal() as session: + claimed = session.execute( + update(ExternalLink) + .where( + ExternalLink.id == link_id, + ExternalLink.status.in_(("pending", "failed")), + ) + .values(status="downloading") + .returning(ExternalLink.id) + ).first() + session.commit() + if claimed is None: + return {"link_id": link_id, "skipped": "not claimable"} + + link = session.get(ExternalLink, link_id) + post = session.get(Post, link.post_id) if link.post_id else None + if post is None: + link.status = "dead" + link.last_error = "post missing" + link.completed_at = datetime.now(UTC) + session.commit() + return {"link_id": link_id, "error": "post missing"} + artist = session.get(Artist, post.artist_id) + source = session.get(Source, post.source_id) if post.source_id else None + platform = source.platform if source is not None else "patreon" + settings = ImportSettings.load_sync(session) + + if not _host_enabled(settings, link.host): + link.status = "skipped" + link.last_error = f"{link.host} disabled" + link.completed_at = datetime.now(UTC) + session.commit() + return {"link_id": link_id, "skipped": "host disabled"} + + host, url, attempts = link.host, link.url, link.attempts + + # Per-host serialize: one fetch per host at a time. If busy, requeue. + lock = None + try: + lock = _redis().lock(f"{_LOCK_PREFIX}{host}", timeout=_LOCK_TTL, blocking=False) + got = bool(lock.acquire(blocking=False)) + except redis.RedisError as exc: + log.warning("extdl lock unavailable for %s: %s — running uncapped", host, exc) + lock, got = None, True + if not got: + # Release the claim back to pending so a later run can pick it up, then + # requeue with a countdown (bounded, like download_source). + with SessionLocal() as session: + session.execute( + update(ExternalLink).where(ExternalLink.id == link_id) + .values(status="pending") + ) + session.commit() + if _serialize_waits < _MAX_SERIALIZE_WAITS: + fetch_external_link.apply_async( + (link_id,), {"_serialize_waits": _serialize_waits + 1}, + countdown=_SERIALIZE_COUNTDOWN, + ) + return {"link_id": link_id, "requeued": "host busy"} + + started = time.monotonic() + post_dir = IMAGES_ROOT / artist.slug / platform / "external" / str(post.external_post_id) + try: + result = fetch_external(host, url, post_dir, timeout=_FETCH_TIMEOUT) + with SessionLocal() as session: + link = session.get(ExternalLink, link_id) + if not result.ok: + _record_failure(link, attempts, result.error or "fetch failed") + session.commit() + return {"link_id": link_id, "error": result.error} + + # Re-load post/artist/source attached to THIS session before handing + # them to the importer (the claim block's instances are detached). + post = session.get(Post, link.post_id) + artist = session.get(Artist, post.artist_id) + source = session.get(Source, post.source_id) if post.source_id else None + image_ids = _route_files(session, result.files, post, platform, artist, source) + link.status = "downloaded" + link.last_error = None + link.completed_at = datetime.now(UTC) + link.duration_seconds = time.monotonic() - started + session.commit() + + # Thumbnails + ML for any newly-attached images (mirrors the download + # path). Lazy import to dodge a task-module import cycle. + if image_ids: + from .ml import tag_and_embed + from .thumbnail import generate_thumbnail + for img_id in image_ids: + generate_thumbnail.delay(img_id) + tag_and_embed.delay(img_id) + return {"link_id": link_id, "files": len(result.files), "images": len(image_ids)} + except Exception as exc: # never leave a link stuck in 'downloading' + log.exception("external fetch task failed for link %s", link_id) + with SessionLocal() as session: + link = session.get(ExternalLink, link_id) + if link is not None and link.status == "downloading": + _record_failure(link, attempts, f"{type(exc).__name__}: {exc}") + session.commit() + raise + finally: + if lock is not None: + try: + lock.release() + except Exception: # noqa: BLE001 — TTL may have already freed it + pass + + +def _record_failure(link: ExternalLink, attempts: int, error: str) -> None: + nxt = attempts + 1 + link.attempts = nxt + link.last_error = error[:1000] + link.status = "dead" if nxt >= DEAD_LETTER_THRESHOLD else "failed" + link.completed_at = datetime.now(UTC) + + +def _route_files(session, files, post, platform, artist, source) -> list[int]: + """Import each fetched file via the existing pipeline (in the artist tree, so + art stays in place); return new image ids for thumb/ML enqueue. A captured + archive/attachment is copied into its store, so its on-disk original is + removed (mirrors download_service); art images stay.""" + settings = ImportSettings.load_sync(session) + importer = Importer( + session=session, + images_root=IMAGES_ROOT, + import_root=IMAGES_ROOT, + thumbnailer=Thumbnailer(images_root=IMAGES_ROOT), + settings=settings, + ) + image_ids: list[int] = [] + for f in files: + _write_link_sidecar(f, post, platform, artist) + result = importer.attach_in_place(f, artist=artist, source=source) + session.commit() + if result.status in ("imported", "superseded"): + ids = list(getattr(result, "member_image_ids", []) or []) + if result.image_id is not None and result.image_id not in ids: + ids.append(result.image_id) + image_ids.extend(ids) + elif result.status in ("attached", "failed"): + # Copied into the attachment store (or failed) — drop the on-disk + # original so we don't keep two copies / an orphan. + f.unlink(missing_ok=True) + # The synthesized sidecar has done its job — don't litter the tree. + Path(str(f) + ".json").unlink(missing_ok=True) + return image_ids + + +@celery.task(name="backend.app.tasks.external.sweep_external_links") +def sweep_external_links() -> dict: + """Enqueue a bounded batch of actionable links (pending, or failed below the + dead-letter threshold). Driven by the post-download hook and a beat tick.""" + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + ids = session.execute( + select(ExternalLink.id).where( + ExternalLink.status.in_(("pending", "failed")), + ExternalLink.attempts < DEAD_LETTER_THRESHOLD, + ).order_by(ExternalLink.id.asc()).limit(_SWEEP_BATCH) + ).scalars().all() + for link_id in ids: + fetch_external_link.delay(link_id) + return {"enqueued": len(ids)} + + +@celery.task(name="backend.app.tasks.external.recover_external_links") +def recover_external_links() -> dict: + """Recovery sweep (rule #89): reset dead links back to retryable so a stuck + host outage or a since-fixed bug gets another pass, then enqueue. Bounded.""" + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + session.execute( + update(ExternalLink).where(ExternalLink.status == "dead") + .values(status="failed", attempts=0, last_error=None) + ) + session.commit() + return sweep_external_links() + + +@celery.task(name="backend.app.tasks.external.prune_external_links") +def prune_external_links() -> dict: + """Retention (rule #89): delete long-dead links so the ledger doesn't grow + unbounded. Downloaded links are kept (they're the record of what we have).""" + cutoff = datetime.now(UTC) - timedelta(days=_RETENTION_DAYS) + SessionLocal = _sync_session_factory() + with SessionLocal() as session: + res = session.execute( + delete(ExternalLink).where( + ExternalLink.status == "dead", + ExternalLink.created_at < cutoff, + ) + ) + session.commit() + return {"pruned": res.rowcount or 0} diff --git a/tests/test_external_worker.py b/tests/test_external_worker.py new file mode 100644 index 0000000..220154c --- /dev/null +++ b/tests/test_external_worker.py @@ -0,0 +1,137 @@ +"""Integration tests for the external-link download worker (tasks/external).""" + +import pytest +from sqlalchemy import func, select + +import backend.app.tasks.external as ext +from backend.app.models import Artist, ExternalLink, Post, PostAttachment, Source +from backend.app.services.external_fetch import FetchResult + +pytestmark = pytest.mark.integration + + +class _FakeLock: + def acquire(self, blocking=False): + return True + + def release(self): + pass + + +class _FakeRedis: + def lock(self, name, timeout=None, blocking=False): + return _FakeLock() + + +def _seed(db_sync, *, host="pixeldrain", status="pending"): + artist = Artist(name="Ext Artist", slug="ext-artist") + db_sync.add(artist) + db_sync.flush() + source = Source( + artist_id=artist.id, platform="patreon", + url="https://patreon.com/ext", enabled=True, config_overrides={}, + ) + db_sync.add(source) + db_sync.flush() + post = Post( + source_id=source.id, artist_id=artist.id, external_post_id="EXT1", + post_url="https://patreon.com/posts/EXT1", + ) + db_sync.add(post) + db_sync.flush() + link = ExternalLink( + post_id=post.id, artist_id=artist.id, host=host, + url=f"https://{host}.test/file", status=status, + ) + db_sync.add(link) + db_sync.commit() + return post, link + + +def test_fetch_external_link_downloads_and_attaches(db_sync, tmp_path, monkeypatch): + post, link = _seed(db_sync) + link_id = link.id + + monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path) + monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis()) + + def fake_fetch(host, url, dest_dir, *, timeout, should_stop=lambda: False): + dest_dir.mkdir(parents=True, exist_ok=True) + f = dest_dir / "film.bin" # non-art → PostAttachment (no thumb/ML enqueue) + f.write_bytes(b"a film pack") + return FetchResult(files=[f], bytes=f.stat().st_size) + + monkeypatch.setattr(ext, "fetch_external", fake_fetch) + + out = ext.fetch_external_link(link_id) + assert out.get("files") == 1 + + db_sync.expire_all() + refreshed = db_sync.get(ExternalLink, link_id) + assert refreshed.status == "downloaded" + assert refreshed.completed_at is not None + # The file was captured as a PostAttachment linked to the SAME post. + atts = db_sync.execute( + select(PostAttachment).where(PostAttachment.post_id == post.id) + ).scalars().all() + assert len(atts) == 1 + + +def test_fetch_external_link_records_failure(db_sync, tmp_path, monkeypatch): + _, link = _seed(db_sync) + link_id = link.id + monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path) + monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis()) + monkeypatch.setattr( + ext, "fetch_external", + lambda *a, **k: FetchResult(error="host 503"), + ) + + ext.fetch_external_link(link_id) + db_sync.expire_all() + refreshed = db_sync.get(ExternalLink, link_id) + assert refreshed.status == "failed" + assert refreshed.attempts == 1 + assert "503" in refreshed.last_error + + +def test_fetch_external_link_dead_letters_at_threshold(db_sync, tmp_path, monkeypatch): + _, link = _seed(db_sync) + link.attempts = ext.DEAD_LETTER_THRESHOLD - 1 + db_sync.commit() + link_id = link.id + monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path) + monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis()) + monkeypatch.setattr(ext, "fetch_external", lambda *a, **k: FetchResult(error="nope")) + + ext.fetch_external_link(link_id) + db_sync.expire_all() + assert db_sync.get(ExternalLink, link_id).status == "dead" + + +def test_fetch_external_link_skips_non_claimable(db_sync, tmp_path, monkeypatch): + _, link = _seed(db_sync, status="downloaded") + link_id = link.id + called = [] + monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis()) + monkeypatch.setattr( + ext, "fetch_external", + lambda *a, **k: called.append(1) or FetchResult(), + ) + out = ext.fetch_external_link(link_id) + assert out.get("skipped") == "not claimable" + assert called == [] # never fetched + + +def test_sweep_enqueues_pending_and_retryable(db_sync, monkeypatch): + _, l1 = _seed(db_sync, host="pixeldrain") + _, l2 = _seed(db_sync, host="mega", status="failed") + # A dead one must NOT be swept. + _, l3 = _seed(db_sync, host="dropbox", status="dead") + + enqueued = [] + monkeypatch.setattr(ext.fetch_external_link, "delay", lambda lid: enqueued.append(lid)) + + out = ext.sweep_external_links() + assert out["enqueued"] == 2 + assert set(enqueued) == {l1.id, l2.id}