fix: retry transient network failures

scripts/upstream.py

@@ -8,9 +8,11 @@ structure.
 from __future__ import annotations
 
 import hashlib
+import http.client
 import json
 import os
 import tempfile
+import time
 import urllib.error
 import urllib.parse
 import urllib.request
@@ -20,6 +22,8 @@ from pathlib import Path
 USER_AGENT = "retrobios-profile-sync/1.0"
 ABSENT = "\0absent\0"
 GITHUB_COMPARE_CAP = 300
+RETRIES = 3
+RETRY_BACKOFF = (0.5, 2.0)
 GITHUB_HOSTS = frozenset(
     {"github.com", "api.github.com", "raw.githubusercontent.com"}
 )
@@ -49,6 +53,9 @@ _HOSTS: dict[str, tuple[str, str, str]] = {
 }
 
 
+_sleep = time.sleep
+
+
 class UpstreamError(Exception):
     """Any failure while talking to a forge."""
 
@@ -156,32 +163,42 @@ def _http_failure(url: str, exc: urllib.error.HTTPError) -> UpstreamError:
     return UpstreamError(f"{url}: HTTP {exc.code}")
 
 
+def _fetch(url: str, accept_json: bool = False) -> bytes | None:
+    """Body of a GET, or None on 404.
+
+    A pass over every profile issues thousands of requests, so a dropped
+    connection or a transient 5xx is a certainty rather than an accident.
+    Those are retried; a definitive answer from the forge is not.
+    """
+    failure: UpstreamError | None = None
+    for attempt in range(RETRIES):
+        req = urllib.request.Request(url, headers=_headers(url, accept_json))
+        try:
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                return resp.read()
+        except urllib.error.HTTPError as exc:
+            if exc.code == 404:
+                return None
+            failure = _http_failure(url, exc)
+            if isinstance(failure, RateLimitError) or exc.code < 500:
+                raise failure from exc
+        except (urllib.error.URLError, http.client.HTTPException, OSError) as exc:
+            failure = UpstreamError(f"{url}: {exc}")
+        if attempt + 1 < RETRIES:
+            _sleep(RETRY_BACKOFF[attempt])
+    raise failure
+
+
 def _http_text(url: str) -> str | None:
-    """Body of a GET, or None on 404. Replaced in tests."""
-    req = urllib.request.Request(url, headers=_headers(url))
-    try:
-        with urllib.request.urlopen(req, timeout=30) as resp:
-            return resp.read().decode("utf-8", errors="replace")
-    except urllib.error.HTTPError as exc:
-        if exc.code == 404:
-            return None
-        raise _http_failure(url, exc) from exc
-    except urllib.error.URLError as exc:
-        raise UpstreamError(f"{url}: {exc.reason}") from exc
+    """Text body of a GET, or None on 404. Replaced in tests."""
+    body = _fetch(url)
+    return None if body is None else body.decode("utf-8", errors="replace")
 
 
 def _http_json(url: str) -> object | None:
     """Parsed JSON body of a GET, or None on 404. Replaced in tests."""
-    req = urllib.request.Request(url, headers=_headers(url, accept_json=True))
-    try:
-        with urllib.request.urlopen(req, timeout=30) as resp:
-            return json.loads(resp.read().decode())
-    except urllib.error.HTTPError as exc:
-        if exc.code == 404:
-            return None
-        raise _http_failure(url, exc) from exc
-    except urllib.error.URLError as exc:
-        raise UpstreamError(f"{url}: {exc.reason}") from exc
+    body = _fetch(url, accept_json=True)
+    return None if body is None else json.loads(body.decode())
 
 
 def cache_path(cache_dir: str, repo: Repo, sha: str, path: str) -> Path:

tests/test_upstream.py

@@ -2,11 +2,14 @@
 
 from __future__ import annotations
 
+import contextlib
+import http.client
 import os
 import sys
 import tempfile
 import unittest
 import urllib.error
+import urllib.request
 from pathlib import Path
 
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "scripts"))
@@ -20,6 +23,18 @@ def _no_network(url: str):
     raise AssertionError(f"test reached the network: {url}")
 
 
+class _Body:
+    def __init__(self, payload: bytes):
+        self._payload = payload
+
+    def read(self) -> bytes:
+        return self._payload
+
+
+def _http_error(code: int, headers=None) -> urllib.error.HTTPError:
+    return urllib.error.HTTPError("https://host/x", code, "msg", headers, None)
+
+
 class TestParseRepo(unittest.TestCase):
     def test_github(self):
         repo = parse_repo("https://github.com/libretro/beetle-psx-libretro")
@@ -141,40 +156,94 @@ class TestTokenScope(unittest.TestCase):
 class TestHttpFailure(unittest.TestCase):
     """Only an actual quota signal may abort a whole run."""
 
-    @staticmethod
-    def _error(code, headers=None):
-        return urllib.error.HTTPError(
-            "https://host/x", code, "msg", headers, None
-        )
-
     def test_429_is_a_rate_limit(self):
         self.assertIsInstance(
-            upstream._http_failure("u", self._error(429)), upstream.RateLimitError
+            upstream._http_failure("u", _http_error(429)), upstream.RateLimitError
         )
 
     def test_403_with_exhausted_quota_is_a_rate_limit(self):
-        exc = self._error(403, {"X-RateLimit-Remaining": "0"})
+        exc = _http_error(403, {"X-RateLimit-Remaining": "0"})
         self.assertIsInstance(
             upstream._http_failure("u", exc), upstream.RateLimitError
         )
 
     def test_403_with_quota_left_is_not(self):
-        exc = self._error(403, {"X-RateLimit-Remaining": "4970"})
+        exc = _http_error(403, {"X-RateLimit-Remaining": "4970"})
         failure = upstream._http_failure("u", exc)
         self.assertIsInstance(failure, upstream.UpstreamError)
         self.assertNotIsInstance(failure, upstream.RateLimitError)
 
     def test_bare_403_from_a_forge_is_not_a_rate_limit(self):
-        failure = upstream._http_failure("u", self._error(403))
+        failure = upstream._http_failure("u", _http_error(403))
         self.assertIsInstance(failure, upstream.UpstreamError)
         self.assertNotIsInstance(failure, upstream.RateLimitError)
 
     def test_525_is_a_plain_upstream_error(self):
-        failure = upstream._http_failure("u", self._error(525))
+        failure = upstream._http_failure("u", _http_error(525))
         self.assertIsInstance(failure, upstream.UpstreamError)
         self.assertNotIsInstance(failure, upstream.RateLimitError)
 
 
+class TestFetchRetry(unittest.TestCase):
+    """Transient network failures are retried, definitive answers are not."""
+
+    def setUp(self):
+        self._orig = (urllib.request.urlopen, upstream._sleep)
+        self.slept: list[float] = []
+        upstream._sleep = self.slept.append
+
+    def tearDown(self):
+        urllib.request.urlopen, upstream._sleep = self._orig
+
+    def _serve(self, outcomes):
+        self.calls = 0
+
+        def opener(req, timeout=None):
+            outcome = outcomes[min(self.calls, len(outcomes) - 1)]
+            self.calls += 1
+            if isinstance(outcome, Exception):
+                raise outcome
+            return contextlib.nullcontext(_Body(outcome))
+
+        urllib.request.urlopen = opener
+
+    def test_dropped_connection_is_retried_then_succeeds(self):
+        dropped = http.client.RemoteDisconnected("closed")
+        self._serve([dropped, b"payload"])
+        self.assertEqual(upstream._fetch("https://host/x"), b"payload")
+        self.assertEqual(self.calls, 2)
+        self.assertEqual(len(self.slept), 1)
+
+    def test_retries_are_bounded(self):
+        self._serve([http.client.RemoteDisconnected("closed")])
+        with self.assertRaises(upstream.UpstreamError):
+            upstream._fetch("https://host/x")
+        self.assertEqual(self.calls, upstream.RETRIES)
+
+    def test_server_error_is_retried(self):
+        self._serve([_http_error(503)])
+        with self.assertRaises(upstream.UpstreamError):
+            upstream._fetch("https://host/x")
+        self.assertEqual(self.calls, upstream.RETRIES)
+
+    def test_forbidden_is_not_retried(self):
+        self._serve([_http_error(403)])
+        with self.assertRaises(upstream.UpstreamError):
+            upstream._fetch("https://host/x")
+        self.assertEqual(self.calls, 1)
+
+    def test_missing_file_is_not_retried(self):
+        self._serve([_http_error(404)])
+        self.assertIsNone(upstream._fetch("https://host/x"))
+        self.assertEqual(self.calls, 1)
+
+    def test_rate_limit_is_not_retried(self):
+        self._serve([_http_error(429)])
+        with self.assertRaises(upstream.RateLimitError):
+            upstream._fetch("https://host/x")
+        self.assertEqual(self.calls, 1)
+
+
 class TestCache(unittest.TestCase):
     def setUp(self):
         self.tmp = tempfile.TemporaryDirectory()