fix(dataflows): honour Reddit Retry-After: 0 and jitter 429 backoff

- a valid Retry-After: 0 means retry at once but was treated as absent
  (`or 5.0`) and waited 5s; honour it exactly now
- jitter our own headerless fallback and the inter-subreddit pacing so several
  analyses sharing an IP don't retry in lockstep and re-collide on the limit;
  keep the single-retry ceiling (more retries can't fix an exhausted IP budget) #1193
This commit is contained in:
Yijia-Xiao
2026-09-01 05:02:36 +00:00
parent 70b58c21dc
commit 2322dd9baa
2 changed files with 44 additions and 5 deletions

View File

@@ -147,6 +147,26 @@ class TestRss429Backoff:
reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0) reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0)
slept.assert_called_once_with(12.0) slept.assert_called_once_with(12.0)
def test_retry_after_zero_is_honoured_not_treated_as_absent(self):
# A valid "Retry-After: 0" means retry at once; it must not fall through
# to the fallback wait (the earlier `or 5.0` bug turned 0 into 5s).
err = HTTPError("url", 429, "Too Many Requests", {"Retry-After": "0"}, None)
with patch.object(reddit, "urlopen", side_effect=[err, _atom_resp()]), \
patch.object(reddit.time, "sleep") as slept:
reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0)
slept.assert_called_once_with(0.0)
def test_headerless_429_fallback_is_jittered(self):
# No Retry-After -> our own ~5s fallback, jittered so concurrent runs
# don't retry in lockstep (kept within a tight band).
err = HTTPError("url", 429, "Too Many Requests", {}, None)
with patch.object(reddit, "urlopen", side_effect=[err, _atom_resp()]), \
patch.object(reddit.time, "sleep") as slept:
reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0)
slept.assert_called_once()
(wait,), _ = slept.call_args
assert 4.0 <= wait <= 6.0 # 5s +/-20% jitter
@pytest.mark.unit @pytest.mark.unit
class TestChunkedTransferErrorsHandled: class TestChunkedTransferErrorsHandled:

View File

@@ -21,6 +21,7 @@ import html
import http.client import http.client
import json import json
import logging import logging
import random
import re import re
import time import time
import xml.etree.ElementTree as ET import xml.etree.ElementTree as ET
@@ -101,11 +102,26 @@ def _strip_html(content: str) -> str:
return " ".join(html.unescape(text).split()) return " ".join(html.unescape(text).split())
# Headerless-429 backoff when Reddit gives no Retry-After. Jittered so several
# analyses sharing an IP don't retry in lockstep and re-collide on the limit.
_RETRY_FALLBACK_SECONDS = 5.0
def _jitter(seconds: float, frac: float = 0.2) -> float:
"""Return ``seconds`` with +/-``frac`` random jitter, to desynchronize
concurrent runs pacing against the same per-IP limit."""
return seconds * (1.0 + random.uniform(-frac, frac))
def _retry_after_seconds(exc: HTTPError) -> float | None: def _retry_after_seconds(exc: HTTPError) -> float | None:
"""Seconds to wait from a 429's ``Retry-After`` header, capped at 30s.""" """Seconds to wait from a 429's ``Retry-After`` header, capped at 30s.
Returns ``None`` only when the header is absent or unparseable; a valid
``Retry-After: 0`` returns ``0.0`` (retry at once), not ``None``.
"""
try: try:
val = exc.headers.get("Retry-After") if getattr(exc, "headers", None) else None val = exc.headers.get("Retry-After") if getattr(exc, "headers", None) else None
return min(float(val), 30.0) if val else None return min(float(val), 30.0) if val is not None else None
except (ValueError, TypeError, AttributeError): except (ValueError, TypeError, AttributeError):
return None return None
@@ -131,7 +147,10 @@ def _fetch_subreddit_rss(
root = ET.fromstring(resp.read()) root = ET.fromstring(resp.read())
except HTTPError as exc: except HTTPError as exc:
if exc.code == 429 and _retry: if exc.code == 429 and _retry:
wait = _retry_after_seconds(exc) or 5.0 # Honour a server-supplied Retry-After exactly (including 0); jitter
# only our own fallback so concurrent runs don't retry in lockstep.
retry_after = _retry_after_seconds(exc)
wait = retry_after if retry_after is not None else _jitter(_RETRY_FALLBACK_SECONDS)
logger.warning( logger.warning(
"Reddit RSS 429 for r/%s · %s — backing off %.1fs then retrying once", "Reddit RSS 429 for r/%s · %s — backing off %.1fs then retrying once",
sub, ticker, wait, sub, ticker, wait,
@@ -234,8 +253,8 @@ def fetch_reddit_posts(
blocks = [] blocks = []
total_posts = 0 total_posts = 0
for i, sub in enumerate(subreddits): for i, sub in enumerate(subreddits):
if i > 0: if i > 0 and inter_request_delay:
time.sleep(inter_request_delay) time.sleep(_jitter(inter_request_delay))
posts = _within_window(_fetch_subreddit(ticker, sub, limit_per_sub, timeout), posts = _within_window(_fetch_subreddit(ticker, sub, limit_per_sub, timeout),
start_date, end_date) start_date, end_date)
total_posts += len(posts) total_posts += len(posts)