fix(external): split fetch timeout into read (60s) + total (30m) budgets (#883)

The single _FETCH_TIMEOUT=3000s meant different things per host: a TOTAL
wall-clock for mega (subprocess), but only a per-read socket timeout for HTTP
hosts (requests' timeout is the idle gap between bytes, never a total). So a
stalled HTTP connection tied up a download-worker slot AND the per-host
serialize lock for ~50 min before failing (operator-flagged 2026-06-17).

Split into two limits in external_fetch:
- read timeout (_READ_TIMEOUT=60s, with _CONNECT_TIMEOUT=30s) → requests gets
  (connect, read); a stalled socket now fails in ~60s.
- total budget (_TOTAL_TIMEOUT=30min) → enforced as a wall-clock deadline
  across chunks in _stream_to_file (HTTP has no total-download timeout), and
  passed as the subprocess total for mega.
fetch_external() signature: timeout= → read_timeout=/total_timeout=. gdrive
(gdown) self-manages; the celery hard limit is the outer backstop.

Also lowered the per-host lock TTL 3600→2400 so a worker that dies holding it
can't wedge a host's links much past one fetch's budget.

Each external link is already one Celery task (sweep enqueues one
fetch_external_link.delay per link), so these budgets are per-link.

Tests: total-budget-exceeded cleans the .part; HTTP gets (connect, read);
mega gets the total. Worker fakes updated to **kwargs.

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

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,
-                should_stop: Callable[[], bool], fallback: str,
-                headers: dict | None = None) -> FetchResult:
-    resp = _http_get(url, timeout=timeout, headers=headers, stream=True)
+def _get_to_dir(url: str, dest_dir: Path, *, read_timeout: float,
+                total_timeout: float, should_stop: Callable[[], bool],
+                fallback: str, headers: dict | None = None) -> FetchResult:
+    # (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,
-                       should_stop=should_stop, fallback="dropbox-file")
+    return _get_to_dir(direct, dest_dir, read_timeout=read_timeout,
+                       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,
-                       should_stop=should_stop, fallback=f"{file_id}.bin")
+    return _get_to_dir(api, dest_dir, read_timeout=read_timeout,
+                       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,
-                       should_stop=should_stop, fallback="mediafire-file")
+    return _get_to_dir(m.group(1), dest_dir, read_timeout=read_timeout,
+                       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
-    failure, stop) is captured on `.error` so the worker can record it and move
-    on."""
+    raises — any backend error (transport, read/total timeout, non-200, scrape
+    miss, subprocess failure, stop) is captured on `.error` so the worker can
+    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
-# the backstop above this.
-_FETCH_TIMEOUT = 3000.0
+# Per-fetch read + total budgets now live in external_fetch (a short read
+# timeout fails a stalled host fast; a generous total caps a big-but-flowing
+# 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,13 +147,14 @@ 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) legitimately run to
-    # the task's 60-min hard limit (time_limit=3600; per-fetch _FETCH_TIMEOUT
-    # is 50min). Its TaskRun records queue='default' (no queue override), so
-    # without this it fell to the 5-min default and healthy in-flight fetches
-    # were phantom-flagged 'RecoverySweep' before their own timeout/error could
+    # 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 is robust whatever queue the row records. 65 = 60 + 5.
+    # 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_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())