fix(download): retry the metadata probe; set socket timeouts everywhere

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,

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