From 6295351af2e2d1d1c6877e50411327b03f02202e Mon Sep 17 00:00:00 2001 From: Kevin Costner <120246174+kevincostner17@users.noreply.github.com> Date: Mon, 14 Sep 2026 23:09:06 +0530 Subject: [PATCH] fix: retry transient failures in the weekly online fixture refresh The weekly "Refresh online fixtures" workflow failed three weeks running. _download had no retries, and http.client.RemoteDisconnected is not a URLError, so a single dropped connection from archive.ics.uci.edu escaped the per-dataset handler and aborted the whole run. - scripts/fixture_download.py: shared download() with exponential backoff for OSError/HTTPException, 429 and 5xx; permanent 4xx raise immediately. - fetch_online_fixtures.py: continue per dataset, separate network failures from dataset (parse/empty/hash/permanent HTTP) failures, add --max-failures for the network budget, emit GitHub warning annotations. - tests/expectations.py: the live-fetch helper reuses download(). - fetch-fixtures.yml: --max-failures 2, always run the offline verify step with -m "not online", open a nightly-failure issue on scheduled failures. --- .github/workflows/fetch-fixtures.yml | 37 +++++++- scripts/fetch_online_fixtures.py | 50 +++++++++-- scripts/fixture_download.py | 58 ++++++++++++ tests/expectations.py | 7 +- tests/test_fixture_download.py | 128 +++++++++++++++++++++++++++ 5 files changed, 264 insertions(+), 16 deletions(-) create mode 100644 scripts/fixture_download.py create mode 100644 tests/test_fixture_download.py diff --git a/.github/workflows/fetch-fixtures.yml b/.github/workflows/fetch-fixtures.yml index 56e5d415..71e04320 100644 --- a/.github/workflows/fetch-fixtures.yml +++ b/.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"], + }); + } diff --git a/scripts/fetch_online_fixtures.py b/scripts/fetch_online_fixtures.py index bc8aca1e..790d61e2 100644 --- a/scripts/fetch_online_fixtures.py +++ b/scripts/fetch_online_fixtures.py @@ -8,6 +8,7 @@ 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 @@ -15,9 +16,9 @@ import argparse import hashlib import json +import os import sys import urllib.error -import urllib.request from pathlib import Path import pandas as pd @@ -25,6 +26,7 @@ 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,11 +186,16 @@ 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) @@ -180,8 +203,17 @@ def main(argv: list[str] | None = None) -> int: _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__": diff --git a/scripts/fixture_download.py b/scripts/fixture_download.py new file mode 100644 index 00000000..1410e6c5 --- /dev/null +++ b/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") diff --git a/tests/expectations.py b/tests/expectations.py index 3a74d329..2cdd2464 100644 --- a/tests/expectations.py +++ b/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() diff --git a/tests/test_fixture_download.py b/tests/test_fixture_download.py new file mode 100644 index 00000000..965a3c74 --- /dev/null +++ b/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