From d592e0ca029273041501b447fdc8a0ae74e95ca3 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 6 Jun 2026 00:15:48 -0400 Subject: [PATCH] =?UTF-8?q?feat(patreon):=20within-pass=20transient=20retr?= =?UTF-8?q?y=20for=20media=20GETs=20=E2=80=94=20#8=20(was=20overstated=20a?= =?UTF-8?q?s=20done)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- backend/app/services/patreon_downloader.py | 73 ++++++++++++++++------ tests/test_patreon_downloader.py | 61 +++++++++++++++++- 2 files changed, 111 insertions(+), 23 deletions(-) diff --git a/backend/app/services/patreon_downloader.py b/backend/app/services/patreon_downloader.py index 9dda0f4..7125a7e 100644 --- a/backend/app/services/patreon_downloader.py +++ b/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 -# one throttled file from becoming a per-item error (plan #703). -_MAX_MEDIA_429_RETRIES = 2 +# Retry a media GET that hits a TRANSIENT failure within the same pass (plan +# #705 #8): a transport blip (connection reset / timeout / truncated stream), a +# 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 - falls through to raise_for_status → HTTPError, which download_post turns - into a resilient per-item error outcome. + Retried (backoff): transport blips (connection reset / timeout / + truncated stream — incl. mid-download), HTTP 429 (honoring Retry-After), + 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) - if resp.status_code == 429 and attempt < _MAX_MEDIA_429_RETRIES: + try: + 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)", - url, delay, attempt, _MAX_MEDIA_429_RETRIES, + "Patreon media transport error (%s) — backing off %.1fs " + "(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`. diff --git a/tests/test_patreon_downloader.py b/tests/test_patreon_downloader.py index ea89752..1d42285 100644 --- a/tests/test_patreon_downloader.py +++ b/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"