Merge pull request 'fix(external): recovery-sweep threshold + queue recording + split fetch timeouts (#883)' (#116) from dev into main

test_downloaded_archive_gets_provenance_and_tagging's fake_fetch still had the
old `*, timeout` signature; the task now calls fetch_external() without it, so
the stub raised TypeError in the integration lane (run 1191). Switch it to
**kwargs like the other two.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/app/celery_signals.py

@@ -69,7 +69,15 @@ def _queue_for(task) -> str:
         return "ml"
     if name.startswith("backend.app.tasks.thumbnail."):
         return "thumbnail"
-    if name.startswith("backend.app.tasks.download."):
+    if name.startswith((
+        "backend.app.tasks.download.",
+        # External file-host fetches share the download lane (celery_app
+        # routes external.* → download). Mirror it here or TaskRun.queue
+        # lies 'default' for them, so per-queue dashboard filters and the
+        # per-queue threshold override miss them — the same gap the
+        # 2026-06-02 audit fixed for backup/admin/library_audit.
+        "backend.app.tasks.external.",
+    )):
         return "download"
     if name.startswith("backend.app.tasks.scan."):
         return "scan"

backend/app/services/external_fetch.py

@@ -9,7 +9,8 @@ call `fetch_external()` directly.
 No single tool covers all five hosts, so a small registry maps host → fetch
 function behind one signature:
 
-    fetch_external(host, url, dest_dir, *, timeout, should_stop) -> FetchResult
+    fetch_external(host, url, dest_dir, *, read_timeout, total_timeout, should_stop)
+        -> FetchResult
 
 Backends:
   - dropbox    : force the direct-download variant (dl=1) + stream GET.
@@ -30,6 +31,7 @@ import logging
 import os
 import re
 import subprocess
+import time
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -40,7 +42,19 @@ import requests
 log = logging.getLogger(__name__)
 
 _CHUNK = 1 << 16
-_DEFAULT_TIMEOUT = 600.0
+# Two distinct limits, because conflating them (the old single 3000s value) meant
+# a stalled HTTP connection tied up a download-worker slot + the per-host lock for
+# ~50 min before failing (operator-flagged 2026-06-17):
+#   * READ timeout — max idle gap between bytes on an HTTP socket. A stalled host
+#     (socket open, nothing flowing) is the common failure mode; a short read
+#     timeout fails it fast. This is what requests' `timeout` actually enforces —
+#     per-read, never a total.
+#   * TOTAL budget — generous wall-clock cap for a file that IS actively
+#     transferring (big films/packs). Enforced as a deadline across chunks, since
+#     no HTTP client timeout bounds the total. Also the subprocess total for mega.
+_CONNECT_TIMEOUT = 30.0
+_READ_TIMEOUT = 60.0
+_TOTAL_TIMEOUT = 1800.0  # 30 min per fetch
 _USER_AGENT = (
     "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
     "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
@@ -122,9 +136,11 @@ def _filename_from(resp: requests.Response, url: str, fallback: str) -> str:
 
 
 def _stream_to_file(resp: requests.Response, dest: Path,
-                    should_stop: Callable[[], bool]) -> int:
+                    should_stop: Callable[[], bool],
+                    *, deadline: float | None = None) -> int:
     """Stream a response body to `dest` (atomic via .part). Returns byte count.
-    Honors should_stop between chunks (partial file removed)."""
+    Honors should_stop and the total-budget `deadline` (a time.monotonic() value)
+    between chunks; the partial file is removed on either abort."""
     part = dest.with_name(dest.name + ".part")
     total = 0
     try:
@@ -132,6 +148,8 @@ def _stream_to_file(resp: requests.Response, dest: Path,
             for chunk in resp.iter_content(chunk_size=_CHUNK):
                 if should_stop():
                     raise ExternalFetchError("stopped")
+                if deadline is not None and time.monotonic() > deadline:
+                    raise ExternalFetchError("exceeded total fetch budget")
                 if chunk:
                     fh.write(chunk)
                     total += len(chunk)
@@ -142,21 +160,29 @@ def _stream_to_file(resp: requests.Response, dest: Path,
     return total
 
 
-def _get_to_dir(url: str, dest_dir: Path, *, timeout: float,
+def _get_to_dir(url: str, dest_dir: Path, *, read_timeout: float,
-                should_stop: Callable[[], bool], fallback: str,
+                total_timeout: float, should_stop: Callable[[], bool],
-                headers: dict | None = None) -> FetchResult:
+                fallback: str, headers: dict | None = None) -> FetchResult:
-    resp = _http_get(url, timeout=timeout, headers=headers, stream=True)
+    # (connect, read): a short read timeout fails a stalled socket fast; the total
+    # budget is enforced separately as a deadline across chunks (requests has no
+    # total-download timeout).
+    resp = _http_get(
+        url, timeout=(_CONNECT_TIMEOUT, read_timeout), headers=headers, stream=True
+    )
     if resp.status_code != 200:
         return FetchResult(error=f"HTTP {resp.status_code} for {url}")
     name = _filename_from(resp, url, fallback)
     dest = dest_dir / name
-    written = _stream_to_file(resp, dest, should_stop)
+    written = _stream_to_file(
+        resp, dest, should_stop, deadline=time.monotonic() + total_timeout
+    )
     return FetchResult(files=[dest], bytes=written)
 
 
 # -- per-host fetchers -----------------------------------------------------
 
-def _fetch_dropbox(url: str, dest_dir: Path, *, timeout: float,
+def _fetch_dropbox(url: str, dest_dir: Path, *, read_timeout: float,
+                   total_timeout: float,
                    should_stop: Callable[[], bool]) -> FetchResult:
     # Force the direct-download variant: dl=1 (Dropbox serves an HTML preview
     # for dl=0). Rewrite/insert the param rather than string-replace so ?dl=0,
@@ -165,35 +191,44 @@ def _fetch_dropbox(url: str, dest_dir: Path, *, timeout: float,
     q = dict(parse_qsl(parts.query))
     q["dl"] = "1"
     direct = urlunsplit(parts._replace(query=urlencode(q)))
-    return _get_to_dir(direct, dest_dir, timeout=timeout,
+    return _get_to_dir(direct, dest_dir, read_timeout=read_timeout,
-                       should_stop=should_stop, fallback="dropbox-file")
+                       total_timeout=total_timeout, should_stop=should_stop,
+                       fallback="dropbox-file")
 
 
-def _fetch_pixeldrain(url: str, dest_dir: Path, *, timeout: float,
+def _fetch_pixeldrain(url: str, dest_dir: Path, *, read_timeout: float,
+                      total_timeout: float,
                       should_stop: Callable[[], bool]) -> FetchResult:
     # /u/{id} (and /l/{id}) → the API file endpoint.
     file_id = urlsplit(url).path.rstrip("/").split("/")[-1]
     if not file_id:
         return FetchResult(error=f"no pixeldrain id in {url}")
     api = f"https://pixeldrain.com/api/file/{file_id}"
-    return _get_to_dir(api, dest_dir, timeout=timeout,
+    return _get_to_dir(api, dest_dir, read_timeout=read_timeout,
-                       should_stop=should_stop, fallback=f"{file_id}.bin")
+                       total_timeout=total_timeout, should_stop=should_stop,
+                       fallback=f"{file_id}.bin")
 
 
-def _fetch_mediafire(url: str, dest_dir: Path, *, timeout: float,
+def _fetch_mediafire(url: str, dest_dir: Path, *, read_timeout: float,
+                     total_timeout: float,
                      should_stop: Callable[[], bool]) -> FetchResult:
-    page = _http_get(url, timeout=timeout, stream=False)
+    page = _http_get(url, timeout=(_CONNECT_TIMEOUT, read_timeout), stream=False)
     if page.status_code != 200:
         return FetchResult(error=f"HTTP {page.status_code} for mediafire page")
     m = _MEDIAFIRE_RE.search(page.text or "")
     if not m:
         return FetchResult(error="mediafire direct link not found on page")
-    return _get_to_dir(m.group(1), dest_dir, timeout=timeout,
+    return _get_to_dir(m.group(1), dest_dir, read_timeout=read_timeout,
-                       should_stop=should_stop, fallback="mediafire-file")
+                       total_timeout=total_timeout, should_stop=should_stop,
+                       fallback="mediafire-file")
 
 
-def _fetch_gdrive(url: str, dest_dir: Path, *, timeout: float,
+def _fetch_gdrive(url: str, dest_dir: Path, *, read_timeout: float,
+                  total_timeout: float,
                   should_stop: Callable[[], bool]) -> FetchResult:
+    # gdown manages its own HTTP session/timeouts; the task's celery hard limit is
+    # the outer backstop. read_timeout/total_timeout are accepted for a uniform
+    # registry signature but not separately enforceable here.
     out = _gdown_download(url, str(dest_dir))
     if not out:
         return FetchResult(error="gdown returned no file (quota / private?)")
@@ -203,10 +238,13 @@ def _fetch_gdrive(url: str, dest_dir: Path, *, timeout: float,
     return FetchResult(files=[p], bytes=p.stat().st_size)
 
 
-def _fetch_mega(url: str, dest_dir: Path, *, timeout: float,
+def _fetch_mega(url: str, dest_dir: Path, *, read_timeout: float,
+                total_timeout: float,
                 should_stop: Callable[[], bool]) -> FetchResult:
     before = set(dest_dir.iterdir()) if dest_dir.exists() else set()
-    _run_mega_get(url, str(dest_dir), timeout=timeout)
+    # megatools is a subprocess: its timeout IS a total wall-clock cap (the read
+    # timeout has no analogue here), so the total budget applies directly.
+    _run_mega_get(url, str(dest_dir), timeout=total_timeout)
     new = [p for p in dest_dir.iterdir() if p not in before and p.is_file()]
     if not new:
         return FetchResult(error="mega-get wrote no new file")
@@ -225,18 +263,24 @@ SUPPORTED_HOSTS = tuple(_REGISTRY)
 
 
 def fetch_external(host: str, url: str, dest_dir: Path, *,
-                   timeout: float = _DEFAULT_TIMEOUT,
+                   read_timeout: float = _READ_TIMEOUT,
+                   total_timeout: float = _TOTAL_TIMEOUT,
                    should_stop: Callable[[], bool] = lambda: False) -> FetchResult:
     """Fetch `url` (a `host` link) into `dest_dir`. Returns a FetchResult; never
-    raises — any backend error (transport, non-200, scrape miss, subprocess
+    raises — any backend error (transport, read/total timeout, non-200, scrape
-    failure, stop) is captured on `.error` so the worker can record it and move
+    miss, subprocess failure, stop) is captured on `.error` so the worker can
-    on."""
+    record it and move on.
+
+    `read_timeout` fails a stalled HTTP socket fast (idle gap between bytes);
+    `total_timeout` is the generous wall-clock cap for a large file that is
+    actively transferring (and the subprocess total for mega)."""
     fetcher = _REGISTRY.get(host)
     if fetcher is None:
         return FetchResult(error=f"unsupported host {host!r}")
     dest_dir.mkdir(parents=True, exist_ok=True)
     try:
-        return fetcher(url, dest_dir, timeout=timeout, should_stop=should_stop)
+        return fetcher(url, dest_dir, read_timeout=read_timeout,
+                       total_timeout=total_timeout, should_stop=should_stop)
     except requests.RequestException as exc:
         return FetchResult(error=f"transport error: {exc}")
     except subprocess.TimeoutExpired:

backend/app/tasks/external.py

@@ -41,16 +41,19 @@ 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
+# Per-fetch read + total budgets now live in external_fetch (a short read
-# the backstop above this.
+# timeout fails a stalled host fast; a generous total caps a big-but-flowing
-_FETCH_TIMEOUT = 3000.0
+# download). The celery soft/hard time_limit is the outer backstop above those.
 # 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.
+# Per-host serialize lock. TTL is a safety net for a worker that dies holding it
+# (normal completion/error releases it in `finally`); sized just past the fetch
+# total budget (30 min) so a dead worker can't wedge a host's links much longer
+# than one fetch would have taken.
 _LOCK_PREFIX = "fc:extdl_lock:"
-_LOCK_TTL = 3600
+_LOCK_TTL = 2400
 _SERIALIZE_COUNTDOWN = 120
 _MAX_SERIALIZE_WAITS = 30
 
@@ -177,7 +180,7 @@ def fetch_external_link(self, link_id: int, _serialize_waits: int = 0) -> dict:
         / str(post.external_post_id) / str(link_id)
     )
     try:
-        result = fetch_external(host, url, post_dir, timeout=_FETCH_TIMEOUT)
+        result = fetch_external(host, url, post_dir)
         with SessionLocal() as session:
             link = session.get(ExternalLink, link_id)
             if not result.ok:

backend/app/tasks/maintenance.py

@@ -147,6 +147,15 @@ TASK_STUCK_THRESHOLD_MINUTES: dict[str, int] = {
     "backend.app.tasks.backup.restore_images_task": 420,
     # Library audit scans the full library — 2h hard limit.
     "backend.app.tasks.library_audit.scan_library_for_rule": 130,
+    # External file-host fetches (mega/gdrive/film packs) can run to the task's
+    # 60-min hard limit (time_limit=3600) — the fetcher's own read/total budgets
+    # (external_fetch) cap a single fetch below that, but this stays the outer
+    # backstop. Without an override these healthy in-flight fetches were
+    # phantom-flagged 'RecoverySweep' before their own timeout/error could
+    # surface (operator-flagged 2026-06-17, target 414 swept at 6.6min). A
+    # task-name override beats the queue threshold whatever queue the row records
+    # (it recorded 'default' before the celery_signals fix → download). 65 = 60+5.
+    "backend.app.tasks.external.fetch_external_link": 65,
 }
 
 

tests/test_celery_routing.py

@@ -18,3 +18,21 @@ def test_long_one_shots_route_to_maintenance_long():
 def test_quick_maintenance_stays_on_maintenance():
     routes = celery.conf.task_routes
     assert routes["backend.app.tasks.maintenance.*"]["queue"] == "maintenance"
+
+
+def test_queue_for_mirrors_external_to_download():
+    """celery_signals._queue_for is a hand-maintained mirror of task_routes
+    that stamps TaskRun.queue. external.* routes to the download lane, so the
+    mirror must agree — else TaskRun.queue lies 'default' for external fetches
+    and per-queue dashboard filters / threshold overrides miss them
+    (operator-flagged 2026-06-17)."""
+    from backend.app.celery_signals import _queue_for
+
+    class _T:
+        name = "backend.app.tasks.external.fetch_external_link"
+
+    assert _queue_for(_T()) == "download"
+    assert (
+        celery.conf.task_routes["backend.app.tasks.external.*"]["queue"]
+        == "download"
+    )

tests/test_external_fetch.py

@@ -139,3 +139,43 @@ def test_should_stop_aborts_and_cleans_part(tmp_path, monkeypatch):
     )
     assert not res.ok
     assert list(tmp_path.glob("*.part")) == []
+
+
+def test_total_budget_exceeded_aborts_and_cleans_part(tmp_path, monkeypatch):
+    # A negative total_timeout puts the deadline in the past, so the first
+    # chunk's budget check trips — simulates a slow trickle blowing the budget.
+    monkeypatch.setattr(ef, "_http_get", lambda *a, **k: _resp(chunks=(b"a", b"b")))
+    res = fetch_external(
+        "pixeldrain", "https://pixeldrain.com/u/x", tmp_path, total_timeout=-1,
+    )
+    assert not res.ok
+    assert "budget" in res.error
+    assert list(tmp_path.glob("*.part")) == []
+
+
+def test_http_get_uses_connect_and_read_timeout_tuple(tmp_path, monkeypatch):
+    # A stalled socket must fail on the short read timeout, not the total — so
+    # the HTTP layer gets (connect, read), never the total budget.
+    seen = {}
+
+    def fake_get(url, *, timeout, headers=None, stream=True):
+        seen["timeout"] = timeout
+        return _resp()
+
+    monkeypatch.setattr(ef, "_http_get", fake_get)
+    fetch_external("pixeldrain", "https://pixeldrain.com/u/x", tmp_path, read_timeout=42)
+    assert seen["timeout"] == (ef._CONNECT_TIMEOUT, 42)
+
+
+def test_mega_uses_total_timeout(tmp_path, monkeypatch):
+    # megatools is a subprocess: its only timeout is a total wall-clock, so the
+    # total budget (not the read timeout) applies to it.
+    seen = {}
+
+    def fake_mega(url, out_dir, *, timeout):
+        seen["timeout"] = timeout
+        (Path(out_dir) / "film.mp4").write_bytes(b"vid")
+
+    monkeypatch.setattr(ef, "_run_mega_get", fake_mega)
+    fetch_external("mega", "https://mega.nz/file/x#k", tmp_path, total_timeout=123)
+    assert seen["timeout"] == 123

tests/test_external_worker.py

@@ -81,7 +81,7 @@ def test_fetch_external_link_downloads_and_attaches(db_sync, tmp_path, monkeypat
     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):
+    def fake_fetch(host, url, dest_dir, **kwargs):
         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")
@@ -115,7 +115,7 @@ def test_refetch_same_link_keeps_canonical_file(db_sync, tmp_path, monkeypatch):
     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):
+    def fake_fetch(host, url, dest_dir, **kwargs):
         dest_dir.mkdir(parents=True, exist_ok=True)
         f = dest_dir / "clip.jpg"  # art → ImageRecord (imported in place)
         f.write_bytes(_structured_jpeg_bytes())
@@ -222,7 +222,7 @@ def test_downloaded_archive_gets_provenance_and_tagging(db_sync, tmp_path, monke
     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):
+    def fake_fetch(host, url, dest_dir, **kwargs):
         dest_dir.mkdir(parents=True, exist_ok=True)
         zpath = dest_dir / "pack.zip"
         with zipfile.ZipFile(zpath, "w") as z:

tests/test_maintenance.py

@@ -453,6 +453,59 @@ def test_recover_stalled_task_runs_archive_task_uses_longer_threshold(db_sync):
     assert _status(archive_stale_id) == "error"
 
 
+def test_recover_stalled_task_runs_external_fetch_uses_longer_threshold(db_sync):
+    """fetch_external_link legitimately runs to its 60-min hard limit, but its
+    TaskRun records queue='default' (no queue override), so before the
+    task-name override (65 min) it fell to the 5-min default and healthy
+    in-flight fetches were phantom-flagged 'RecoverySweep' before their own
+    timeout/error could surface (operator-flagged 2026-06-17, target 414 swept
+    at 6.6min). A 10-min-old row must survive; a 70-min-old one is flagged."""
+    from sqlalchemy import select
+
+    from backend.app.models import TaskRun
+    from backend.app.tasks.maintenance import recover_stalled_task_runs
+
+    name = "backend.app.tasks.external.fetch_external_link"
+    now = datetime.now(UTC)
+    # 10-min-old: stale by the default 5-min rule but fresh by the 65-min
+    # task-name override. Must survive despite recording queue='default'.
+    fresh_id = _make_task_run(
+        db_sync, status="running", queue="default", task_name=name,
+        started_at=now - timedelta(minutes=10),
+    )
+    # 70-min-old: past even the 65-min override (a genuine hard kill). Flagged.
+    stale_id = _make_task_run(
+        db_sync, status="running", queue="default", task_name=name,
+        started_at=now - timedelta(minutes=70),
+    )
+    db_sync.commit()
+
+    recovered = recover_stalled_task_runs.apply().get()
+    assert recovered == 1
+
+    db_sync.expire_all()
+    def _status(_id):
+        return db_sync.execute(
+            select(TaskRun.status).where(TaskRun.id == _id)
+        ).scalar_one()
+    assert _status(fresh_id) == "running"
+    assert _status(stale_id) == "error"
+
+
+def test_external_fetch_stuck_threshold_exceeds_hard_time_limit():
+    """Invariant guard (maintenance.py:112): the fetch_external_link task-name
+    override MUST be ≥ its hard time_limit, else the sweep flags healthy long
+    fetches. Pins it so a future time_limit bump can't silently re-break it."""
+    from backend.app.tasks.external import fetch_external_link
+    from backend.app.tasks.maintenance import TASK_STUCK_THRESHOLD_MINUTES
+
+    hard_minutes = fetch_external_link.time_limit / 60
+    override = TASK_STUCK_THRESHOLD_MINUTES[
+        "backend.app.tasks.external.fetch_external_link"
+    ]
+    assert override >= hard_minutes
+
+
 def test_prune_task_runs_deletes_ok_older_than_24h(db_sync):
     from sqlalchemy import select