diff --git a/backend/app/services/external_fetch.py b/backend/app/services/external_fetch.py index 265092b..4707858 100644 --- a/backend/app/services/external_fetch.py +++ b/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: diff --git a/backend/app/tasks/external.py b/backend/app/tasks/external.py index efbb028..5f4c2f4 100644 --- a/backend/app/tasks/external.py +++ b/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: diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 988c9ca..b24fe3a 100644 --- a/backend/app/tasks/maintenance.py +++ b/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, } diff --git a/tests/test_external_fetch.py b/tests/test_external_fetch.py index e7c1993..c456f75 100644 --- a/tests/test_external_fetch.py +++ b/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 diff --git a/tests/test_external_worker.py b/tests/test_external_worker.py index 0db0029..d54ba12 100644 --- a/tests/test_external_worker.py +++ b/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())