feat(poller): honor Retry-After on HTTP 429 with bounded retries

Pyker · Pyker · commit 3dbfa282ef7f · 2026-05-13T16:20:34.000+01:00
Replace the single-shot 4xx raise in fetch_page with up to 3 attempts.
Honor server Retry-After (RFC 9110 §10.2.3, both delta-seconds and all
three HTTP-date forms, never capped) and fall back to exponential
backoff capped at 60s when the header is missing or unparseable. Final
give-up still raises NEMPException so existing error handling keeps
working. Per-attempt cool-downs log at INFO with attempt counter;
exhaustion logs at WARNING.

Fixes recurring HTTP 429 polling failures for the seven
raw.githubusercontent.com hosted mods.
diff --git a/mod_polling/poller.py b/mod_polling/poller.py
@@ -4,9 +4,11 @@
 import logging
 import re
 import urllib.parse
+from datetime import UTC, datetime
 
 import aiohttp
 import yaml
+from aiohttp.helpers import parse_http_date
 from jinja2 import Environment, FileSystemLoader, select_autoescape
 
 logger = logging.getLogger("mod_polling.poller")
@@ -21,6 +23,29 @@ def __str__(self):
         return repr(self.args[0])
 
 
+MAX_FETCH_ATTEMPTS = 3
+RETRY_AFTER_FALLBACK_CAP = 60.0
+
+
+def _retry_after_seconds(value: str | None) -> float | None:
+    """Parse a Retry-After header value (RFC 9110 §10.2.3) into a non-negative
+    seconds delta, or ``None`` if missing/unparseable.
+
+    Accepts delta-seconds or HTTP-date in any of the three forms allowed by
+    §5.6.7 (IMF-fixdate, RFC 850, or asctime).
+    """
+    if not value:
+        return None
+    try:
+        return max(0.0, float(value))
+    except ValueError:
+        pass
+    dt = parse_http_date(value)
+    if dt is None:
+        return None
+    return max(0.0, (dt - datetime.now(UTC)).total_seconds())
+
+
 class ModPoller:
     def __init__(self):
         self.nem_versions = []
@@ -48,19 +73,44 @@ async def fetch_page(self, url, timeout=10, decode_json=False, ratelimit=True, *
         lock = self._host_locks[host] if ratelimit else contextlib.nullcontext()
 
         async with lock:
-            async with self.session.get(url, timeout=aiohttp.ClientTimeout(total=timeout), **kwargs) as response:
-                if 400 <= response.status < 500:
-                    raise NEMPException(f"HTTP {response.status} for {url}")
-                response.raise_for_status()
+            for attempt in range(MAX_FETCH_ATTEMPTS):
+                delay = None
+                async with self.session.get(url, timeout=aiohttp.ClientTimeout(total=timeout), **kwargs) as response:
+                    if response.status == 429:
+                        if attempt == MAX_FETCH_ATTEMPTS - 1:
+                            logger.warning(
+                                "Got HTTP 429 for %s and exhausted %d attempts; giving up",
+                                url,
+                                MAX_FETCH_ATTEMPTS,
+                            )
+                            raise NEMPException(f"HTTP 429 for {url}")
+                        delay = _retry_after_seconds(response.headers.get("Retry-After"))
+                        if delay is None:
+                            delay = min(self._host_delay * (2**attempt), RETRY_AFTER_FALLBACK_CAP)
+                        logger.info(
+                            "Got HTTP 429 for %s; sleeping %.1fs before retry (attempt %d/%d)",
+                            url,
+                            delay,
+                            attempt + 1,
+                            MAX_FETCH_ATTEMPTS,
+                        )
+                    elif 400 <= response.status < 500:
+                        raise NEMPException(f"HTTP {response.status} for {url}")
+                    else:
+                        response.raise_for_status()
 
-                if decode_json:
-                    result = await response.json(content_type=None)
-                else:
-                    result = await response.text()
+                        if decode_json:
+                            result = await response.json(content_type=None)
+                        else:
+                            result = await response.text()
 
-            if ratelimit:
-                await asyncio.sleep(self._host_delay)
-            return result
+                        if ratelimit:
+                            await asyncio.sleep(self._host_delay)
+                        return result
+
+                # Outside response context (socket returned to pool); only
+                # reached via the 429 branch above, which always sets `delay`.
+                await asyncio.sleep(delay)
 
     async def fetch_json(self, *args, **kwargs):
         return await self.fetch_page(*args, decode_json=True, **kwargs)
diff --git a/tests/test_poller_fetch.py b/tests/test_poller_fetch.py
@@ -1,3 +1,7 @@
+from datetime import UTC, datetime, timedelta
+from email.utils import format_datetime
+from unittest.mock import AsyncMock, patch
+
 import aiohttp
 import pytest
 from aioresponses import aioresponses
@@ -45,6 +49,143 @@ async def test_timeout_propagation(self, poller_with_session):
                 await poller_with_session.fetch_page("https://example.com/slow", timeout=1)
 
 
+class TestFetchPageRetryAfter:
+    async def test_429_with_integer_retry_after_waits_and_retries(self, poller_with_session):
+        url = "https://example.com/throttled"
+        with aioresponses() as mocked:
+            mocked.get(url, status=429, headers={"Retry-After": "5"})
+            mocked.get(url, body="recovered")
+            with patch("mod_polling.poller.asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
+                result = await poller_with_session.fetch_page(url)
+            assert result == "recovered"
+            mock_sleep.assert_any_await(5.0)
+
+    async def test_429_with_http_date_retry_after_waits_and_retries(self, poller_with_session):
+        url = "https://example.com/throttled-date"
+        future = datetime.now(UTC) + timedelta(seconds=30)
+        http_date = format_datetime(future, usegmt=True)
+        with aioresponses() as mocked:
+            mocked.get(url, status=429, headers={"Retry-After": http_date})
+            mocked.get(url, body="recovered")
+            with patch("mod_polling.poller.asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
+                result = await poller_with_session.fetch_page(url)
+            assert result == "recovered"
+            slept = [call.args[0] for call in mock_sleep.await_args_list]
+            assert any(29 <= v <= 31 for v in slept), f"Expected ~30s sleep, got {slept}"
+
+    async def test_429_with_asctime_obs_date_retry_after(self, poller_with_session):
+        """RFC 9110 §5.6.7 obs-date allows asctime form, which has no timezone."""
+        url = "https://example.com/throttled-asctime"
+        future = datetime.now(UTC) + timedelta(seconds=20)
+        asctime_str = future.strftime("%a %b %d %H:%M:%S %Y")
+        with aioresponses() as mocked:
+            mocked.get(url, status=429, headers={"Retry-After": asctime_str})
+            mocked.get(url, body="recovered")
+            with patch("mod_polling.poller.asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
+                result = await poller_with_session.fetch_page(url)
+            assert result == "recovered"
+            slept = [call.args[0] for call in mock_sleep.await_args_list]
+            assert any(18 <= v <= 22 for v in slept), f"Expected ~20s sleep, got {slept}"
+
+    async def test_429_without_retry_after_uses_exponential_backoff_fallback(self, poller_with_session):
+        poller_with_session._host_delay = 0.5
+        url = "https://example.com/throttled-no-header"
+        with aioresponses() as mocked:
+            mocked.get(url, status=429)
+            mocked.get(url, body="recovered")
+            with patch("mod_polling.poller.asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
+                result = await poller_with_session.fetch_page(url)
+            assert result == "recovered"
+            # Backoff fires BEFORE the retry, so it's the first sleep call.
+            # attempt 0 fallback: min(0.5 * 2^0, 60) = 0.5
+            slept = [c.args[0] for c in mock_sleep.await_args_list]
+            assert slept[0] == 0.5, f"Expected backoff 0.5 as first sleep, got: {slept}"
+
+    async def test_429_with_garbled_retry_after_uses_exponential_backoff_fallback(self, poller_with_session):
+        poller_with_session._host_delay = 0.5
+        url = "https://example.com/throttled-garbled"
+        with aioresponses() as mocked:
+            mocked.get(url, status=429, headers={"Retry-After": "soon-ish"})
+            mocked.get(url, body="recovered")
+            with patch("mod_polling.poller.asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
+                result = await poller_with_session.fetch_page(url)
+            assert result == "recovered"
+            slept = [c.args[0] for c in mock_sleep.await_args_list]
+            assert slept[0] == 0.5, f"Expected backoff 0.5 as first sleep, got: {slept}"
+
+    async def test_429_fallback_caps_at_60s_with_large_host_delay(self, poller_with_session):
+        poller_with_session._host_delay = 1000
+        url = "https://example.com/throttled-big-delay"
+        with aioresponses() as mocked:
+            mocked.get(url, status=429)
+            mocked.get(url, body="recovered")
+            with patch("mod_polling.poller.asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
+                result = await poller_with_session.fetch_page(url)
+            assert result == "recovered"
+            # min(1000 * 2^0, 60) = 60 — the cap kicks in
+            mock_sleep.assert_any_await(60.0)
+
+    async def test_429_then_429_then_success(self, poller_with_session):
+        url = "https://example.com/twice-throttled"
+        with aioresponses() as mocked:
+            mocked.get(url, status=429, headers={"Retry-After": "5"})
+            mocked.get(url, status=429, headers={"Retry-After": "3"})
+            mocked.get(url, body="finally")
+            with patch("mod_polling.poller.asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
+                result = await poller_with_session.fetch_page(url)
+            assert result == "finally"
+            mock_sleep.assert_any_await(5.0)
+            mock_sleep.assert_any_await(3.0)
+
+    async def test_429_three_times_raises_after_two_retries(self, poller_with_session):
+        url = "https://example.com/persistently-throttled"
+        with aioresponses() as mocked:
+            mocked.get(url, status=429, headers={"Retry-After": "1"})
+            mocked.get(url, status=429, headers={"Retry-After": "1"})
+            mocked.get(url, status=429, headers={"Retry-After": "1"})
+            with (
+                patch("mod_polling.poller.asyncio.sleep", new_callable=AsyncMock),
+                pytest.raises(NEMPException, match="HTTP 429"),
+            ):
+                await poller_with_session.fetch_page(url)
+
+    async def test_429_exhausted_logs_warning(self, poller_with_session, caplog):
+        url = "https://example.com/exhausted"
+        with aioresponses() as mocked:
+            mocked.get(url, status=429, headers={"Retry-After": "1"})
+            mocked.get(url, status=429, headers={"Retry-After": "1"})
+            mocked.get(url, status=429, headers={"Retry-After": "1"})
+            with (
+                patch("mod_polling.poller.asyncio.sleep", new_callable=AsyncMock),
+                caplog.at_level("WARNING", logger="mod_polling.poller"),
+                pytest.raises(NEMPException),
+            ):
+                await poller_with_session.fetch_page(url)
+        matching = [
+            r
+            for r in caplog.records
+            if r.levelname == "WARNING" and url in r.getMessage() and "exhausted" in r.getMessage().lower()
+        ]
+        messages = [r.getMessage() for r in caplog.records]
+        assert matching, f"Expected WARNING log mentioning exhausted attempts, got: {messages}"
+
+    async def test_429_logs_cooldown_at_info(self, poller_with_session, caplog):
+        url = "https://example.com/throttled-logged"
+        with aioresponses() as mocked:
+            mocked.get(url, status=429, headers={"Retry-After": "7"})
+            mocked.get(url, body="recovered")
+            with (
+                patch("mod_polling.poller.asyncio.sleep", new_callable=AsyncMock),
+                caplog.at_level("INFO", logger="mod_polling.poller"),
+            ):
+                await poller_with_session.fetch_page(url)
+        matching = [
+            r for r in caplog.records if r.levelname == "INFO" and url in r.getMessage() and "7" in r.getMessage()
+        ]
+        messages = [r.getMessage() for r in caplog.records]
+        assert matching, f"Expected INFO log mentioning URL and duration, got: {messages}"
+
+
 class TestFetchJson:
     async def test_delegates_to_fetch_page(self, poller_with_session):
         with aioresponses() as mocked:
