From abc3c47a3d6d67147e8ed6981fec113d8cb1a390 Mon Sep 17 00:00:00 2001 From: Thales <> Date: Fri, 17 Jul 2026 01:31:01 +0100 Subject: [PATCH] fix(download): retry the metadata probe; set socket timeouts everywhere The pre-download metadata probe (duration check) ran outside the retry loop: a transient network blip on that single request failed the whole job immediately, even though the actual download had a 3-attempt backoff. The probe and the download now share one retry policy (_with_retries), with the same retriable/non-retriable classification, cancel translation, and user-visible "retrying" stage message. Every YoutubeDL instance (probe, audio download, video track) now sets an explicit 30 s socket_timeout so a stalled TCP connection can never hang a job indefinitely. Closes #279 --- app/pipeline/download.py | 87 ++++++++++++++++++++++++------------ tests/test_download_retry.py | 86 +++++++++++++++++++++++++++++++++++ 2 files changed, 145 insertions(+), 28 deletions(-) create mode 100644 tests/test_download_retry.py diff --git a/app/pipeline/download.py b/app/pipeline/download.py index 6562b289..bc958a4e 100644 --- a/app/pipeline/download.py +++ b/app/pipeline/download.py @@ -50,6 +50,42 @@ def _is_retriable(exc: Exception) -> bool: return any(s in msg for s in _RETRIABLE) +# yt-dlp's default socket timeout is 20 s but only applies where it plumbs the +# option through; set it explicitly on every YoutubeDL we build so a stalled +# TCP connection can never hang a job indefinitely (#279). +_SOCKET_TIMEOUT_SEC = 30 + + +def _with_retries(job: Job, fn, *, what: str): + """Run `fn` with the shared transient-network retry policy (#279). + + Retries _MAX_RETRIES times with backoff on retriable errors; re-raises + immediately on non-retriable ones. A cancel arriving mid-attempt is + surfaced as JobCancelled. Shared by the metadata probe and the download + itself so both survive the same network blips.""" + for attempt in range(_MAX_RETRIES + 1): + try: + return fn() + except Exception as exc: + if job.cancel_requested: + raise JobCancelled() from exc + if attempt < _MAX_RETRIES and _is_retriable(exc): + wait = _RETRY_BACKOFF[attempt] + logger.warning( + "[%s] %s attempt %d/%d failed (%s), retrying in %ds", + job.id, + what, + attempt + 1, + _MAX_RETRIES, + exc, + wait, + ) + _set(job, stage=f"Network error — retrying ({attempt + 1}/{_MAX_RETRIES})...") + time.sleep(wait) + else: + raise + + _VIDEO_ID_RE = re.compile(r"^[A-Za-z0-9_-]{11}$") _YOUTUBE_HOSTS = frozenset( ( @@ -192,6 +228,7 @@ def vhook(d: dict) -> None: "noplaylist": True, "allowed_extractors": _ALLOWED_EXTRACTORS, "progress_hooks": [vhook], + "socket_timeout": _SOCKET_TIMEOUT_SEC, } # Point yt-dlp at the bundled ffmpeg in case a DASH stream needs remuxing; # in portable builds ffmpeg is not on PATH. @@ -224,11 +261,21 @@ def download(job: Job, url: str, job_dir: Path) -> Path: _set(job, status="downloading", progress=0.0, stage="Processing...") # Fetch metadata first (no download) so we can reject videos that are - # too long before wasting bandwidth and disk. - with YoutubeDL( - {"quiet": True, "noplaylist": True, "allowed_extractors": _ALLOWED_EXTRACTORS} - ) as ydl: - meta = ydl.extract_info(url, download=False) or {} + # too long before wasting bandwidth and disk. Runs under the same retry + # policy as the download itself -- a transient blip on this first request + # used to fail the whole job immediately (#279). + def _probe() -> dict: + with YoutubeDL( + { + "quiet": True, + "noplaylist": True, + "allowed_extractors": _ALLOWED_EXTRACTORS, + "socket_timeout": _SOCKET_TIMEOUT_SEC, + } + ) as ydl: + return ydl.extract_info(url, download=False) or {} + + meta = _with_retries(job, _probe, what="metadata probe") duration = meta.get("duration") or 0 max_duration = get_max_duration_sec() if duration > max_duration: @@ -263,30 +310,14 @@ def hook(d: dict) -> None: "noplaylist": True, "allowed_extractors": _ALLOWED_EXTRACTORS, "progress_hooks": [hook], + "socket_timeout": _SOCKET_TIMEOUT_SEC, } - info: dict = {} - for attempt in range(_MAX_RETRIES + 1): - try: - with YoutubeDL(ydl_opts) as ydl: - info = ydl.extract_info(url, download=True) or {} - break - except Exception as exc: - if job.cancel_requested: - raise JobCancelled() from exc - if attempt < _MAX_RETRIES and _is_retriable(exc): - wait = _RETRY_BACKOFF[attempt] - logger.warning( - "[%s] download attempt %d/%d failed (%s), retrying in %ds", - job.id, - attempt + 1, - _MAX_RETRIES, - exc, - wait, - ) - _set(job, stage=f"Network error — retrying ({attempt + 1}/{_MAX_RETRIES})...") - time.sleep(wait) - else: - raise + + def _fetch() -> dict: + with YoutubeDL(ydl_opts) as ydl: + return ydl.extract_info(url, download=True) or {} + + info: dict = _with_retries(job, _fetch, what="download") _set( job, diff --git a/tests/test_download_retry.py b/tests/test_download_retry.py new file mode 100644 index 00000000..ca2010b7 --- /dev/null +++ b/tests/test_download_retry.py @@ -0,0 +1,86 @@ +"""Tests for the shared download retry policy (#279).""" + +from __future__ import annotations + +import pytest + +from app.core.models import Job, JobCancelled +from app.pipeline import download as dl_mod + + +@pytest.fixture(autouse=True) +def _no_sleep(monkeypatch): + """Backoff sleeps are pointless in tests.""" + monkeypatch.setattr(dl_mod.time, "sleep", lambda _s: None) + + +@pytest.fixture() +def job(): + return Job(id="abcdefabc279") + + +def test_retries_transient_error_then_succeeds(job): + calls: list[int] = [] + + def flaky(): + calls.append(1) + if len(calls) < 3: + raise OSError("Connection reset by peer") + return {"title": "ok"} + + result = dl_mod._with_retries(job, flaky, what="metadata probe") + + assert result == {"title": "ok"} + assert len(calls) == 3 + # The retry stage message reached the job (user-visible feedback). + assert "retrying" in job.stage_message + + +def test_non_retriable_raises_immediately(job): + calls: list[int] = [] + + def private_video(): + calls.append(1) + raise RuntimeError("ERROR: Private video. Sign in if you have access") + + with pytest.raises(RuntimeError, match="Private video"): + dl_mod._with_retries(job, private_video, what="metadata probe") + + assert len(calls) == 1 # never retried + + +def test_exhausted_retries_reraise_last_error(job): + calls: list[int] = [] + + def always_down(): + calls.append(1) + raise OSError("Read timed out") + + with pytest.raises(OSError, match="timed out"): + dl_mod._with_retries(job, always_down, what="download") + + assert len(calls) == dl_mod._MAX_RETRIES + 1 + + +def test_cancel_mid_attempt_becomes_jobcancelled(job): + def fails(): + job.cancel_requested = True # POST /cancel raced the attempt + raise OSError("connection reset") + + with pytest.raises(JobCancelled): + dl_mod._with_retries(job, fails, what="download") + + +def test_unrecognized_error_is_not_retried(job): + """Errors matching neither list are treated as permanent -- retrying an + unknown failure mode would just triple the wait for the same outcome.""" + calls: list[int] = [] + + def weird(): + calls.append(1) + raise ValueError("some novel explosion") + + with pytest.raises(ValueError): + dl_mod._with_retries(job, weird, what="download") + + assert len(calls) == 1