fix: retry transient failures in the weekly online fixture refresh

.github/workflows/fetch-fixtures.yml

@@ -11,12 +11,45 @@ permissions:
 jobs:
   fetch:
     runs-on: ubuntu-latest
+    timeout-minutes: 30
+    permissions:
+      contents: read
+      issues: write
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5
         with:
           python-version: "3.12"
       - run: pip install -e ".[dev,ml]"
-      - run: python scripts/fetch_online_fixtures.py --refresh --update-manifest
+      - name: Refresh fixtures
+        # Downloads retry transient network errors. A couple of upstream hosts
+        # (UCI in particular) may still be unreachable in a given week; parse,
+        # empty, hash and permanent HTTP failures always fail the job.
+        run: python scripts/fetch_online_fixtures.py --refresh --update-manifest --max-failures 2
       - name: Verify tests (offline cache)
-        run: pytest tests/test_online_datasets.py -q --no-cov
+        if: always()
+        # Live-fetch tests are marked `online`; the refresh step above already
+        # exercised the network, so this step only checks the cached slices.
+        run: pytest tests/test_online_datasets.py -m "not online" -q --no-cov
+      - name: Open/refresh alert issue on failure
+        if: failure() && github.event_name == 'schedule'
+        uses: actions/github-script@v7
+        with:
+          script: |
+            const title = "weekly online fixture refresh failing";
+            const body = `The scheduled online fixture refresh failed.\n\n` +
+              `Run: ${context.serverUrl}/${context.repo.owner}/${context.repo.repo}` +
+              `/actions/runs/${context.runId}`;
+            const open = await github.rest.issues.listForRepo({
+              ...context.repo, state: "open", labels: "nightly-failure",
+            });
+            const existing = open.data.find(i => i.title === title);
+            if (existing) {
+              await github.rest.issues.createComment({
+                ...context.repo, issue_number: existing.number, body,
+              });
+            } else {
+              await github.rest.issues.create({
+                ...context.repo, title, body, labels: ["nightly-failure"],
+              });
+            }

scripts/fetch_online_fixtures.py

@@ -8,23 +8,25 @@
     python scripts/fetch_online_fixtures.py --only titanic
     python scripts/fetch_online_fixtures.py --update-manifest
     python scripts/fetch_online_fixtures.py --discover --update-manifest
+    python scripts/fetch_online_fixtures.py --refresh --max-failures 2
 """
 
 from __future__ import annotations
 
 import argparse
 import hashlib
 import json
+import os
 import sys
 import urllib.error
-import urllib.request
 from pathlib import Path
 
 import pandas as pd
 
 ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 from dataset_loader import load_dataframe, payload_bytes, registry_entry_to_manifest  # noqa: E402
+from fixture_download import TransientFetchError, download  # noqa: E402
 
 MANIFEST_PATH = ROOT / "tests" / "fixtures" / "online" / "manifest.json"
 REGISTRY_PATH = ROOT / "tests" / "fixtures" / "online" / "registry.json"
@@ -55,15 +57,19 @@ def _save_manifest(manifest: dict) -> None:
 
 
 def _download(url: str) -> bytes:
-    req = urllib.request.Request(url, headers={"User-Agent": "freshdata-fixture-fetch/1.0"})
-    with urllib.request.urlopen(req, timeout=120) as resp:  # noqa: S310
-        return resp.read()
+    return download(url)
 
 
 def _sha256(data: bytes) -> str:
     return hashlib.sha256(data).hexdigest()
 
 
+def _annotate(message: str) -> None:
+    """Surface a failure as a GitHub Actions warning annotation when in CI."""
+    if os.environ.get("GITHUB_ACTIONS") == "true":
+        print(f"::warning::{message}")
+
+
 def fetch_one(
     name: str,
     entry: dict,
@@ -139,6 +145,17 @@ def main(argv: list[str] | None = None) -> int:
         help="Fetch all registry entries (sync manifest from registry)",
     )
     parser.add_argument("--only", action="append", default=[], metavar="ID")
+    parser.add_argument(
+        "--max-failures",
+        type=int,
+        default=0,
+        metavar="N",
+        help=(
+            "Tolerate up to N datasets whose download still fails with transient "
+            "network errors after retries. Parse, empty, hash and permanent HTTP "
+            "failures always fail the run."
+        ),
+    )
     args = parser.parse_args(argv)
 
     if not REGISTRY_PATH.exists():
@@ -154,7 +171,8 @@ def main(argv: list[str] | None = None) -> int:
         return 1
 
     print(f"Fetching {len(names)} online fixture(s)...")
-    failures = 0
+    dataset_failures: list[str] = []
+    network_failures: list[str] = []
     for name in names:
         entry = dict(registry[name])
         if name in manifest and not args.discover:
@@ -168,20 +186,34 @@ def main(argv: list[str] | None = None) -> int:
                 update_manifest=args.update_manifest or args.discover,
             )
             if result is None:
-                failures += 1
+                dataset_failures.append(name)
                 manifest.pop(name, None)
+        except TransientFetchError as exc:
+            # Keep the existing manifest entry: a network blip says nothing
+            # about whether the dataset itself is still valid.
+            print(f"  {name}: FAILED (network) — {exc}", file=sys.stderr)
+            network_failures.append(name)
         except (urllib.error.URLError, ValueError, pd.errors.ParserError) as exc:
             print(f"  {name}: FAILED — {exc}", file=sys.stderr)
-            failures += 1
+            dataset_failures.append(name)
             if args.discover:
                 manifest.pop(name, None)
 
     if args.update_manifest or args.refresh or args.discover:
         _save_manifest(manifest)
         print(f"Updated {MANIFEST_PATH.relative_to(ROOT)} ({len(manifest)} entries)")
 
-    print(f"Done. {failures} failure(s).")
-    return 1 if failures else 0
+    for name in network_failures:
+        _annotate(f"online fixture {name!r}: download failed after retries (network)")
+    for name in dataset_failures:
+        _annotate(f"online fixture {name!r}: dataset failed to download or parse")
+    print(
+        f"Done. {len(dataset_failures)} dataset failure(s), "
+        f"{len(network_failures)} network failure(s) (tolerating {args.max_failures})."
+    )
+    if dataset_failures or len(network_failures) > args.max_failures:
+        return 1
+    return 0
 
 
 if __name__ == "__main__":

scripts/fixture_download.py

@@ -0,0 +1,58 @@
+"""Network download with retries for the online fixture tooling.
+
+Shared by ``scripts/fetch_online_fixtures.py`` and the live-fetch test helper so
+both survive transient upstream failures (connection resets, dropped responses,
+throttling, 5xx) the same way.
+"""
+
+from __future__ import annotations
+
+import http.client
+import random
+import time
+import urllib.error
+import urllib.request
+from typing import Callable
+
+USER_AGENT = "freshdata-fixture-fetch/1.0"
+
+
+class TransientFetchError(RuntimeError):
+    """A download kept failing with transient network errors after every retry."""
+
+
+def _is_transient(exc: BaseException) -> bool:
+    if isinstance(exc, urllib.error.HTTPError):
+        return exc.code == 429 or exc.code >= 500
+    # URLError (DNS, refused, reset), RemoteDisconnected / IncompleteRead
+    # (HTTPException) and socket timeouts are all worth another attempt.
+    return isinstance(exc, (OSError, http.client.HTTPException))
+
+
+def download(
+    url: str,
+    *,
+    attempts: int = 4,
+    timeout: float = 120.0,
+    base_delay: float = 2.0,
+    sleep: Callable[[float], None] = time.sleep,
+) -> bytes:
+    """Return the body of *url*, retrying transient failures with backoff.
+
+    Permanent HTTP errors (4xx other than 429) are raised immediately; after
+    *attempts* transient failures a :class:`TransientFetchError` is raised.
+    """
+    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
+    for attempt in range(1, attempts + 1):
+        try:
+            with urllib.request.urlopen(req, timeout=timeout) as resp:  # noqa: S310
+                return resp.read()
+        except (OSError, http.client.HTTPException) as exc:
+            if not _is_transient(exc):
+                raise
+            if attempt == attempts:
+                raise TransientFetchError(
+                    f"{url}: {type(exc).__name__}: {exc} (after {attempts} attempts)"
+                ) from exc
+            sleep(base_delay * 2 ** (attempt - 1) + random.uniform(0, 1))  # noqa: S311
+    raise AssertionError("unreachable: the loop always returns or raises")

tests/expectations.py

@@ -7,7 +7,6 @@
 import os
 import sys
 import time
-import urllib.request
 import warnings
 from pathlib import Path
 
@@ -27,6 +26,7 @@
 if str(_SCRIPTS) not in sys.path:
     sys.path.insert(0, str(_SCRIPTS))
 from dataset_loader import load_dataframe, payload_bytes  # noqa: E402
+from fixture_download import download  # noqa: E402
 
 ALL_FIXTURES = [
     "aqi_sample",
@@ -75,10 +75,7 @@ def _fetch_online_live(name: str) -> pd.DataFrame:
         raise KeyError(f"unknown online fixture: {name}")
     entry = manifest[name]
 
-    url = entry["url"]
-    req = urllib.request.Request(url, headers={"User-Agent": "freshdata-fixture-fetch/1.0"})
-    with urllib.request.urlopen(req, timeout=120) as resp:  # noqa: S310
-        raw = resp.read()
+    raw = download(entry["url"])
     expected = entry.get("sha256") or ""
     if expected:
         digest = hashlib.sha256(raw).hexdigest()

tests/test_fixture_download.py

@@ -0,0 +1,128 @@
+"""Retry and failure classification for the online fixture download tooling."""
+
+from __future__ import annotations
+
+import http.client
+import io
+import sys
+import urllib.error
+from pathlib import Path
+
+import pytest
+
+_SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
+if str(_SCRIPTS) not in sys.path:
+    sys.path.insert(0, str(_SCRIPTS))
+
+import fetch_online_fixtures as fof  # noqa: E402
+import fixture_download as net  # noqa: E402
+
+URL = "https://example.invalid/data.csv"
+BODY = b"a,b\n1,2\n"
+
+
+def _flaky_urlopen(monkeypatch, errors):
+    """Patch urlopen to raise *errors* in order, then return BODY."""
+    calls: list[str] = []
+
+    def urlopen(req, timeout):
+        calls.append(req.full_url)
+        if len(calls) <= len(errors):
+            raise errors[len(calls) - 1]
+        return io.BytesIO(BODY)
+
+    monkeypatch.setattr(net.urllib.request, "urlopen", urlopen)
+    return calls
+
+
+def test_download_retries_dropped_connections(monkeypatch):
+    calls = _flaky_urlopen(
+        monkeypatch,
+        [http.client.RemoteDisconnected("closed"), ConnectionResetError(104, "reset")],
+    )
+    sleeps: list[float] = []
+    assert net.download(URL, sleep=sleeps.append) == BODY
+    assert len(calls) == 3
+    assert len(sleeps) == 2
+    assert sleeps[1] > sleeps[0]  # exponential backoff
+
+
+def test_download_gives_up_after_all_attempts(monkeypatch):
+    calls = _flaky_urlopen(monkeypatch, [http.client.IncompleteRead(b"")] * 5)
+    with pytest.raises(net.TransientFetchError, match="after 3 attempts"):
+        net.download(URL, attempts=3, sleep=lambda _: None)
+    assert len(calls) == 3
+
+
+@pytest.mark.parametrize("code", [429, 503])
+def test_download_retries_throttling_and_server_errors(monkeypatch, code):
+    error = urllib.error.HTTPError(URL, code, "busy", hdrs=None, fp=None)
+    calls = _flaky_urlopen(monkeypatch, [error])
+    assert net.download(URL, sleep=lambda _: None) == BODY
+    assert len(calls) == 2
+
+
+def test_download_does_not_retry_permanent_http_errors(monkeypatch):
+    error = urllib.error.HTTPError(URL, 404, "Not Found", hdrs=None, fp=None)
+    calls = _flaky_urlopen(monkeypatch, [error])
+    with pytest.raises(urllib.error.HTTPError):
+        net.download(URL, sleep=lambda _: None)
+    assert len(calls) == 1
+
+
+def _run_main(monkeypatch, tmp_path, outcomes, *argv):
+    """Run main() over fake datasets whose fetch_one result/exception is *outcomes*."""
+    registry_path = tmp_path / "registry.json"
+    registry_path.write_text("{}")
+    monkeypatch.setattr(fof, "REGISTRY_PATH", registry_path)
+    monkeypatch.setattr(fof, "_load_registry", lambda: {name: {"url": URL} for name in outcomes})
+    monkeypatch.setattr(fof, "_load_manifest", dict)
+    monkeypatch.setattr(fof, "_save_manifest", lambda manifest: None)
+    attempted: list[str] = []
+
+    def fake_fetch_one(name, entry, manifest, *, refresh, update_manifest):
+        attempted.append(name)
+        outcome = outcomes[name]
+        if isinstance(outcome, BaseException):
+            raise outcome
+        return outcome
+
+    monkeypatch.setattr(fof, "fetch_one", fake_fetch_one)
+    return fof.main(list(argv)), attempted
+
+
+def test_main_keeps_going_after_a_network_failure(monkeypatch, tmp_path):
+    outcomes = {
+        "a_flaky": net.TransientFetchError("reset"),
+        "b_ok": tmp_path / "b_ok.csv",
+    }
+    code, attempted = _run_main(monkeypatch, tmp_path, outcomes)
+    assert attempted == ["a_flaky", "b_ok"]
+    assert code == 1  # default budget is zero
+
+
+def test_main_tolerates_network_failures_within_budget(monkeypatch, tmp_path):
+    outcomes = {
+        "a_flaky": net.TransientFetchError("reset"),
+        "b_flaky": net.TransientFetchError("reset"),
+        "c_ok": tmp_path / "c_ok.csv",
+    }
+    assert _run_main(monkeypatch, tmp_path, outcomes, "--max-failures", "2")[0] == 0
+    assert _run_main(monkeypatch, tmp_path, outcomes, "--max-failures", "1")[0] == 1
+
+
+@pytest.mark.parametrize(
+    "outcome",
+    [ValueError("unparseable"), urllib.error.URLError("404"), None],
+    ids=["parse-error", "permanent-http", "empty-after-parse"],
+)
+def test_main_never_tolerates_dataset_failures(monkeypatch, tmp_path, outcome):
+    outcomes = {"bad": outcome, "ok": tmp_path / "ok.csv"}
+    assert _run_main(monkeypatch, tmp_path, outcomes, "--max-failures", "5")[0] == 1
+
+
+def test_main_annotates_failures_in_github_actions(monkeypatch, tmp_path, capsys):
+    monkeypatch.setenv("GITHUB_ACTIONS", "true")
+    outcomes = {"flaky": net.TransientFetchError("reset")}
+    _run_main(monkeypatch, tmp_path, outcomes, "--max-failures", "1")
+    assert "::warning::online fixture 'flaky'" in capsys.readouterr().out