feat(patreon): within-pass transient retry for media GETs — #8 (was overstated as done)

Honest completion of roadmap #8. Previously a NON-429 media failure (a
connection reset, a timeout, a truncated stream, a 5xx) was an immediate
terminal "error" for the pass — only retried on the NEXT walk. Now
_fetch_to_file retries TRANSIENT failures in-place with backoff (transport
blips incl. mid-download, 429 honoring Retry-After, and 5xx; up to 3 tries),
while PERMANENT failures (404 gone / 403 forbidden) fail fast straight to the
error → dead-letter path — re-fetching them is pointless. This makes the
transient-vs-permanent split explicit instead of leaning on the next-tick
cycle. (#1's 429 backoff + #7's dead-letter covered most of #8's value; this
is the missing in-pass transient piece I'd loosely marked "folded".)

Tests: a connection blip / a 5xx is retried then succeeds; a 404 errors with
NO retry; an exhausted transient becomes a terminal error.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/app/services/patreon_downloader.py

@@ -43,16 +43,30 @@ from urllib.parse import urlsplit
 import requests
 
 from .file_validator import is_validatable, validate_file
-from .patreon_client import _load_session, _retry_after_seconds
+from .patreon_client import (
+    _BACKOFF_CAP_SECONDS,
+    _load_session,
+    _retry_after_seconds,
+)
 
 log = logging.getLogger(__name__)
 
 _TITLE_MAX = 40
 _TIMEOUT_SECONDS = 120.0
 _CHUNK = 1 << 16
-# A CDN media GET rarely 429s, but if it does, a couple of backoff retries keep
+# Retry a media GET that hits a TRANSIENT failure within the same pass (plan
-# one throttled file from becoming a per-item error (plan #703).
+# #705 #8): a transport blip (connection reset / timeout / truncated stream), a
-_MAX_MEDIA_429_RETRIES = 2
+# 429, or a 5xx. PERMANENT failures (404 gone, 403 forbidden) fail fast straight
+# to the error/dead-letter path — no point re-fetching them. Keeps a momentary
+# network hiccup from becoming a per-item error that waits for the next walk.
+_MAX_MEDIA_RETRIES = 3
+# requests transport errors worth retrying (vs. an HTTPError, which is a real
+# server response and is classified by status code).
+_TRANSIENT_TRANSPORT_EXC = (
+    requests.ConnectionError,
+    requests.Timeout,
+    requests.exceptions.ChunkedEncodingError,
+)
 
 # Referer/Origin yt-dlp must send for Mux-hosted Patreon video. Mux's JWT
 # playback policy checks Referer/Origin on every request, so yt-dlp must send
@@ -285,30 +299,49 @@ class PatreonDownloader:
         return dest
 
     def _fetch_to_file(self, url: str, dest: Path) -> None:
-        """Stream a non-video URL to `dest` via the (stubbable) session.
+        """Stream a non-video URL to `dest` via the (stubbable) session, retrying
+        TRANSIENT failures within the same pass (plan #705 #8).
 
-        Retries a transient 429 with backoff (plan #703); any other non-2xx
+        Retried (backoff): transport blips (connection reset / timeout /
-        falls through to raise_for_status → HTTPError, which download_post turns
+        truncated stream — incl. mid-download), HTTP 429 (honoring Retry-After),
-        into a resilient per-item error outcome.
+        and 5xx. Failed fast (no retry → HTTPError → per-item error → dead-letter
+        path): 4xx other than 429 (404 gone, 403 forbidden) — re-fetching a
+        permanent failure is pointless.
         """
         attempt = 0
         while True:
-            resp = self.session.get(url, stream=True, timeout=_TIMEOUT_SECONDS)
+            try:
-            if resp.status_code == 429 and attempt < _MAX_MEDIA_429_RETRIES:
+                resp = self.session.get(url, stream=True, timeout=_TIMEOUT_SECONDS)
+                if (resp.status_code == 429 or resp.status_code >= 500) \
+                        and attempt < _MAX_MEDIA_RETRIES:
+                    attempt += 1
+                    delay = _retry_after_seconds(resp, attempt)
+                    log.warning(
+                        "Patreon media transient HTTP %d (%s) — backing off "
+                        "%.1fs (retry %d/%d)",
+                        resp.status_code, url, delay, attempt, _MAX_MEDIA_RETRIES,
+                    )
+                    time.sleep(delay)
+                    continue
+                # 2xx → ok; 4xx-non-429 (or an exhausted 429/5xx) → HTTPError
+                # (permanent for this pass) → not caught below → per-item error.
+                resp.raise_for_status()
+                with open(dest, "wb") as fh:
+                    for chunk in resp.iter_content(chunk_size=_CHUNK):
+                        if chunk:
+                            fh.write(chunk)
+                return
+            except _TRANSIENT_TRANSPORT_EXC as exc:
+                if attempt >= _MAX_MEDIA_RETRIES:
+                    raise  # exhausted → terminal error outcome
                 attempt += 1
-                delay = _retry_after_seconds(resp, attempt)
+                delay = min(2.0 * (2 ** (attempt - 1)), _BACKOFF_CAP_SECONDS)
                 log.warning(
-                    "Patreon media 429 (%s) — backing off %.1fs (retry %d/%d)",
+                    "Patreon media transport error (%s) — backing off %.1fs "
-                    url, delay, attempt, _MAX_MEDIA_429_RETRIES,
+                    "(retry %d/%d): %s",
+                    url, delay, attempt, _MAX_MEDIA_RETRIES, exc,
                 )
                 time.sleep(delay)
-                continue
-            break
-        resp.raise_for_status()
-        with open(dest, "wb") as fh:
-            for chunk in resp.iter_content(chunk_size=_CHUNK):
-                if chunk:
-                    fh.write(chunk)
 
     def _run_ytdlp(self, url: str, dest: Path, headers: dict) -> Path | None:
         """Invoke yt-dlp to fetch a Mux/HLS stream to (around) `dest`.

tests/test_patreon_downloader.py

@@ -56,15 +56,19 @@ class _FakeSession:
 
 
 class _SeqSession:
-    """Returns the queued responses in order (for retry tests)."""
+    """Returns the queued items in order (for retry tests). An Exception item is
+    RAISED by get() instead of returned (simulates a transport error)."""
 
-    def __init__(self, responses: list[_FakeResponse]):
+    def __init__(self, responses):
         self._responses = list(responses)
         self.calls: list[str] = []
 
     def get(self, url, stream=False, timeout=None):
         self.calls.append(url)
-        return self._responses.pop(0)
+        item = self._responses.pop(0)
+        if isinstance(item, Exception):
+            raise item
+        return item
 
 
 def _post(post_id="1001", title="My Post Title", published="2026-05-01T12:00:00Z"):
@@ -314,3 +318,54 @@ def test_media_429_retried_then_succeeds(tmp_path, monkeypatch):
     assert outcomes[0].status == "downloaded"
     assert 4.0 in slept  # backed off before the retry
     assert len(session.calls) == 2
+
+
+def _dl(tmp_path, session, monkeypatch):
+    monkeypatch.setattr(pd_mod.time, "sleep", lambda s: None)
+    return PatreonDownloader(
+        images_root=tmp_path, cookies_path=None, validate=False, session=session,
+    )
+
+
+def test_transient_transport_error_retried_within_pass(tmp_path, monkeypatch):
+    """plan #705 #8: a connection blip on the GET is retried in-place, not a
+    terminal error that waits for the next walk."""
+    session = _SeqSession([
+        requests.ConnectionError("connection reset"),
+        _FakeResponse(_PNG_BYTES, status_code=200),
+    ])
+    dl = _dl(tmp_path, session, monkeypatch)
+    outcomes = dl.download_post(_post(), [_img("a.png")], "artist-x")
+    assert outcomes[0].status == "downloaded"
+    assert len(session.calls) == 2
+
+
+def test_5xx_retried_within_pass(tmp_path, monkeypatch):
+    session = _SeqSession([
+        _FakeResponse(b"", status_code=503),
+        _FakeResponse(_PNG_BYTES, status_code=200),
+    ])
+    dl = _dl(tmp_path, session, monkeypatch)
+    outcomes = dl.download_post(_post(), [_img("a.png")], "artist-x")
+    assert outcomes[0].status == "downloaded"
+    assert len(session.calls) == 2
+
+
+def test_404_fails_fast_no_retry(tmp_path, monkeypatch):
+    """A permanent 4xx (gone/forbidden) is NOT retried — it errors immediately
+    (and the ingester's dead-letter ledger takes over across walks)."""
+    session = _SeqSession([
+        _FakeResponse(b"", status_code=404),
+        _FakeResponse(_PNG_BYTES, status_code=200),  # would succeed if it retried
+    ])
+    dl = _dl(tmp_path, session, monkeypatch)
+    outcomes = dl.download_post(_post(), [_img("a.png")], "artist-x")
+    assert outcomes[0].status == "error"
+    assert len(session.calls) == 1  # no retry on a permanent failure
+
+
+def test_exhausted_transient_becomes_error(tmp_path, monkeypatch):
+    session = _SeqSession([requests.Timeout("t")] * 10)  # always times out
+    dl = _dl(tmp_path, session, monkeypatch)
+    outcomes = dl.download_post(_post(), [_img("a.png")], "artist-x")
+    assert outcomes[0].status == "error"