fix(coverage-evidence): retry trusted-uv download on transient network errors

CHANGELOG.md

@@ -13,6 +13,7 @@ Semantic Versioning where the repository publishes a release.
 
 ### Fixed
 
+- Retried the trusted-uv archive download a bounded number of times on transient network failures instead of failing the whole `coverage-evidence` job on a single `HTTPError`/`URLError` from the shared `releases.astral.sh` origin, which every pull request's review across the organization fetches; every attempt still runs the unchanged redirect-rejection, host/port pin, size bound, and checksum/member verification, so the fix adds resilience without weakening the trust boundary.
 - Bounded the Strix quality self-test's deterministic timeout fixtures to 3-second process and 5-second fake-sleep budgets so exact-head policy evidence completes inside the existing job limit without changing production Strix scanner timeouts, providers, credentials, or review semantics.
 - Allowed commas and ASCII parentheses in the bounded Strix changed-file path policy so legal tracked Packrat fixtures can receive exact-head security analysis, while rejecting raw `..` components before normalization and keeping controls, backslashes, whitespace ambiguity, and shell punctuation fail-closed.
 - Bound each review-agent invocation key to the wrapper's complete canonical payload, including the base branch and requesting actor; altered fields with a valid-format key now fail before durable-leader election or forwarding, and wrapper write permission is job-scoped.

scripts/ci/materialize_base_python_requirements.py

@@ -19,6 +19,7 @@
 import sys
 import tarfile
 import tempfile
+import time
 import urllib.parse
 import urllib.request
 from typing import Any
@@ -50,6 +51,8 @@
 TRUSTED_UV_DOWNLOAD_MAX_BYTES = 64 * 1024 * 1024
 TRUSTED_UV_BINARY_MAX_BYTES = 64 * 1024 * 1024
 TRUSTED_UV_VERSION_TIMEOUT_SECONDS = 10
+TRUSTED_UV_DOWNLOAD_ATTEMPTS = 3
+TRUSTED_UV_DOWNLOAD_RETRY_DELAY_SECONDS = 2.0
 
 
 class _RejectTrustedUvRedirects(urllib.request.HTTPRedirectHandler):
@@ -165,53 +168,79 @@ def _git(repo_root: pathlib.Path, *args: str) -> bytes:
     return completed.stdout
 
 
-def _download_trusted_uv_archive() -> bytes:
-    """Download the fixed uv release archive through one HTTPS trust boundary."""
-    _install_trusted_uv_url_opener()
-    try:
-        # Keep the audited URL literal at the network sink so static analysis can
-        # prove that neither user data nor repository content selects a scheme,
-        # host, path, query, fragment, method, or request header.
-        with urllib.request.urlopen(  # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected  # nosec B310
-            "https://releases.astral.sh/github/uv/releases/download/0.12.1/"
-            "uv-x86_64-unknown-linux-gnu.tar.gz",
-            timeout=TRUSTED_UV_DOWNLOAD_TIMEOUT_SECONDS,
-        ) as response:
-            final_url = urllib.parse.urlparse(response.geturl())
-            try:
-                final_port = final_url.port
-            except ValueError as exc:
-                raise RuntimeError(
-                    "trusted uv archive redirected outside the fixed "
-                    "releases.astral.sh HTTPS origin"
-                ) from exc
-            if (
-                (final_url.scheme, final_url.hostname)
-                != ("https", "releases.astral.sh")
-                or final_port not in (None, 443)
-            ):
-                raise RuntimeError(
-                    "trusted uv archive redirected outside the fixed "
-                    "releases.astral.sh HTTPS origin"
-                )
-            payload = bytearray()
-            while len(payload) <= TRUSTED_UV_DOWNLOAD_MAX_BYTES:
-                chunk = response.read(
-                    TRUSTED_UV_DOWNLOAD_MAX_BYTES + 1 - len(payload)
-                )
-                if not chunk:
-                    break
-                payload.extend(chunk)
-    except OSError as exc:
-        raise RuntimeError(
-            f"trusted uv archive download failed: {type(exc).__name__}"
-        ) from exc
+def _fetch_trusted_uv_archive_once() -> bytes:
+    """Perform one download attempt through the fixed HTTPS trust boundary.
+
+    Raises ``OSError`` (the network-level failure) on a transient fetch
+    problem so the caller can decide whether to retry, and ``RuntimeError``
+    for every trust-boundary violation (unsafe redirect, oversized payload),
+    which must fail closed on the first occurrence and is never retried.
+    """
+    # Keep the audited URL literal at the network sink so static analysis can
+    # prove that neither user data nor repository content selects a scheme,
+    # host, path, query, fragment, method, or request header.
+    with urllib.request.urlopen(  # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected  # nosec B310
+        "https://releases.astral.sh/github/uv/releases/download/0.12.1/"
+        "uv-x86_64-unknown-linux-gnu.tar.gz",
+        timeout=TRUSTED_UV_DOWNLOAD_TIMEOUT_SECONDS,
+    ) as response:
+        final_url = urllib.parse.urlparse(response.geturl())
+        try:
+            final_port = final_url.port
+        except ValueError as exc:
+            raise RuntimeError(
+                "trusted uv archive redirected outside the fixed "
+                "releases.astral.sh HTTPS origin"
+            ) from exc
+        if (
+            (final_url.scheme, final_url.hostname) != ("https", "releases.astral.sh")
+            or final_port not in (None, 443)
+        ):
+            raise RuntimeError(
+                "trusted uv archive redirected outside the fixed "
+                "releases.astral.sh HTTPS origin"
+            )
+        payload = bytearray()
+        while len(payload) <= TRUSTED_UV_DOWNLOAD_MAX_BYTES:
+            chunk = response.read(TRUSTED_UV_DOWNLOAD_MAX_BYTES + 1 - len(payload))
+            if not chunk:
+                break
+            payload.extend(chunk)
 
     if len(payload) > TRUSTED_UV_DOWNLOAD_MAX_BYTES:
         raise RuntimeError("trusted uv archive exceeded the bounded download size")
     return bytes(payload)
 
 
+def _download_trusted_uv_archive() -> bytes:
+    """Download the fixed uv release archive through one HTTPS trust boundary.
+
+    The single shared ``releases.astral.sh`` origin is fetched by every
+    coverage-evidence run across the organization, so it occasionally answers
+    a transient network or HTTP error under concurrent load even though the
+    file itself is healthy (observed directly: repeated
+    ``trusted uv archive download failed: HTTPError`` failures on
+    unmodified, otherwise-passing pull requests). A bounded number of
+    attempts absorbs that transient condition; every attempt still runs the
+    full trust boundary unchanged (redirect rejection, host/port pin, size
+    bound, and the checksum/member verification performed by the caller), so
+    retrying strictly adds resilience and never weakens a check.
+    """
+    _install_trusted_uv_url_opener()
+    last_error: OSError | None = None
+    for attempt in range(TRUSTED_UV_DOWNLOAD_ATTEMPTS):
+        if attempt:
+            time.sleep(TRUSTED_UV_DOWNLOAD_RETRY_DELAY_SECONDS * attempt)
+        try:
+            return _fetch_trusted_uv_archive_once()
+        except OSError as exc:
+            last_error = exc
+            continue
+    raise RuntimeError(
+        f"trusted uv archive download failed: {type(last_error).__name__}"
+    ) from last_error
+
+
 def _verified_uv_binary(archive_payload: bytes) -> bytes:
     """Return the bounded uv executable after archive and member verification."""
     digest = hashlib.sha256(archive_payload).hexdigest()

tests/test_materialize_base_python_requirements.py

@@ -514,31 +514,87 @@ def test_download_trusted_uv_archive_accepts_fixed_https_origin(
 def test_download_trusted_uv_archive_rejects_unsafe_redirect(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """A redirect away from the fixed HTTPS release host fails closed."""
+    """A redirect away from the fixed HTTPS release host fails closed, unretried."""
     response = FakeHttpResponse("https://example.invalid/uv.tar.gz")
-    monkeypatch.setattr(materializer.urllib.request, "urlopen", lambda *_a, **_k: response)
+    calls = 0
+
+    def _urlopen(*_a: object, **_k: object) -> FakeHttpResponse:
+        nonlocal calls
+        calls += 1
+        return response
+
+    monkeypatch.setattr(materializer.urllib.request, "urlopen", _urlopen)
 
     with pytest.raises(RuntimeError, match="redirected outside"):
         materializer._download_trusted_uv_archive()
+    assert calls == 1
 
 
-def test_download_trusted_uv_archive_rejects_network_and_size_failures(
+def test_download_trusted_uv_archive_rejects_oversized_archive_unretried(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """Network errors and oversized archives cannot enter the trusted tool path."""
-    monkeypatch.setattr(
-        materializer.urllib.request,
-        "urlopen",
-        lambda *_a, **_k: (_ for _ in ()).throw(OSError("offline")),
-    )
-    with pytest.raises(RuntimeError, match="download failed"):
-        materializer._download_trusted_uv_archive()
-
+    """An oversized archive fails closed on the first attempt without retrying."""
     response = FakeHttpResponse(materializer.TRUSTED_UV_ARCHIVE_URL, b"12345")
-    monkeypatch.setattr(materializer.urllib.request, "urlopen", lambda *_a, **_k: response)
+    calls = 0
+
+    def _urlopen(*_a: object, **_k: object) -> FakeHttpResponse:
+        nonlocal calls
+        calls += 1
+        return response
+
+    monkeypatch.setattr(materializer.urllib.request, "urlopen", _urlopen)
     monkeypatch.setattr(materializer, "TRUSTED_UV_DOWNLOAD_MAX_BYTES", 4)
     with pytest.raises(RuntimeError, match="bounded download size"):
         materializer._download_trusted_uv_archive()
+    assert calls == 1
+
+
+def test_download_trusted_uv_archive_exhausts_retries_and_reports_download_failed(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A persistent transient network error retries a bounded number of times."""
+    calls = 0
+
+    def _urlopen(*_a: object, **_k: object) -> None:
+        nonlocal calls
+        calls += 1
+        raise OSError("offline")
+
+    monkeypatch.setattr(materializer.urllib.request, "urlopen", _urlopen)
+    monkeypatch.setattr(materializer.time, "sleep", lambda _seconds: None)
+    with pytest.raises(RuntimeError, match="download failed"):
+        materializer._download_trusted_uv_archive()
+    assert calls == materializer.TRUSTED_UV_DOWNLOAD_ATTEMPTS
+
+
+def test_download_trusted_uv_archive_retries_transient_failure_then_succeeds(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A transient failure on the first attempt is absorbed by a later retry.
+
+    The shared ``releases.astral.sh`` origin serves every coverage-evidence
+    run org-wide, so a lone transient error (observed directly in production
+    as ``trusted uv archive download failed: HTTPError``) must not fail the
+    whole job when a subsequent attempt would have succeeded.
+    """
+    payload = b"archive"
+    response = FakeHttpResponse(materializer.TRUSTED_UV_ARCHIVE_URL, payload)
+    calls = 0
+
+    def _urlopen(*_a: object, **_k: object) -> FakeHttpResponse:
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            raise OSError("transient")
+        return response
+
+    monkeypatch.setattr(materializer.urllib.request, "urlopen", _urlopen)
+    sleeps: list[float] = []
+    monkeypatch.setattr(materializer.time, "sleep", sleeps.append)
+
+    assert materializer._download_trusted_uv_archive() == payload
+    assert calls == 2
+    assert sleeps == [materializer.TRUSTED_UV_DOWNLOAD_RETRY_DELAY_SECONDS]
 
 
 def test_verified_uv_binary_accepts_exact_archive(

tests/test_trusted_uv_download_contract.py

@@ -24,11 +24,16 @@ def _module_tree() -> ast.Module:
 
 
 def _download_function() -> ast.FunctionDef:
-    """Return the trusted-uv downloader function from the parsed module."""
+    """Return the trusted-uv single-attempt network-sink function from the module.
+
+    ``_download_trusted_uv_archive`` is a bounded-retry wrapper around this
+    function; the literal-URL network sink itself lives in
+    ``_fetch_trusted_uv_archive_once`` so it runs unchanged on every attempt.
+    """
     for node in _module_tree().body:
-        if isinstance(node, ast.FunctionDef) and node.name == "_download_trusted_uv_archive":
+        if isinstance(node, ast.FunctionDef) and node.name == "_fetch_trusted_uv_archive_once":
             return node
-    raise AssertionError("trusted uv downloader function is missing")
+    raise AssertionError("trusted uv single-attempt downloader function is missing")
 
 
 def _assigned_literal(name: str) -> object: