feat(feed): Retry logic + Stale check

AGENTS.md

@@ -104,6 +104,7 @@ changing shared scanners, ledger logic, or schemas — not only this tree.
 | **traust-engine** | sibling repository `traust-engine` |
 | **traust-ledger** | sibling repository `traust-ledger` |
 | **traust-contracts** | sibling repository `traust-contracts` |
+| **traust-core** | sibling repository `traust-core` |
 
 **Pins live only in `pyproject.toml`** — do not duplicate shas/version tags in docs.
 `[tool.uv.sources]` pins by commit `rev`, not `tag`: a tag requires a release

config/feeds.yaml

@@ -45,6 +45,19 @@
 # check-licensing walks the same intake checklist when a source is added.
 version: 1
 
+# Download-retry policy for cached-feed refreshes (traust.cli.fetch_feeds).
+# Defaults for every cached source; a source may override any key under its
+# own `retry:` block. Keys are traust_core.v1.providers.http.RetryPolicy fields;
+# 429 honours Retry-After (capped by max_wait_seconds); 5xx and timeouts use
+# capped exponential backoff (backoff_base_seconds * 2**attempt, + up to
+# jitter_seconds). A non-429 4xx is not retried.
+retry:
+  max_attempts: 4
+  timeout_seconds: 120
+  backoff_base_seconds: 1
+  max_wait_seconds: 60
+  jitter_seconds: 0.5
+
 sources:
 
   # ---------------------------------------------------------------- cached

docs/external-dependencies.md

@@ -34,6 +34,7 @@ packages, and `harnessing/`.
 | `traust-engine` | processing core — validators, scanners, reporting, impact analysis (`traust_engine.*`) | Same license as this harness | sibling repository, installed from its git tag via `[tool.uv.sources]` | Not a third-party licensing surface |
 | `traust-contracts` | contract surface — bundled JSON schemas, shared enums, identity vectors | Same license as this harness | sibling repository, installed from its git tag via `[tool.uv.sources]` | Not a third-party licensing surface |
 | `traust-ledger` | disposition-ledger kernel — finding identity, Merkle integrity, signing | Same license as this harness | sibling repository, installed from its git tag via `[tool.uv.sources]` | Not a third-party licensing surface |
+| `traust-core` | platform API — providers (`HttpFetcher`, feed `validate_payload`), errors (`traust_core.v1.*`) | Same license as this harness | sibling repository, installed from its git commit via `[tool.uv.sources]` | Not a third-party licensing surface |
 | `jsonschema` | all validators | MIT | [PyPI](https://pypi.org/pypi/jsonschema/json) | Permissive |
 | `referencing` | schema registry for the validators | MIT | [PyPI](https://pypi.org/pypi/referencing/json) | Permissive |
 | `pyyaml` | config, scope and targets parsing | MIT | [PyPI](https://pypi.org/pypi/PyYAML/json) | Permissive |

pyproject.toml

@@ -28,6 +28,7 @@ dependencies = [
     # itself a candidate for removal, which retires most of the callers.
     "traust-ledger",
     "traust-contracts",
+    "traust-core",
     "jsonschema",
     "referencing",
     "pyyaml",
@@ -53,10 +54,18 @@ traust = "traust.cli.__main__:main"
 [dependency-groups]
 dev = ["ruff>=0.6", "pytest-xdist>=3", "traust-ledger[db]"]
 
+[tool.uv]
+# traust-core (development) is built on the one-owner contracts (72e641c); engine
+# and ledger development still pin 84bd8e4. Unify on 72e641c until they bump.
+override-dependencies = [
+    "traust-contracts @ git+https://github.com/traust-security/traust-contracts.git@72e641c992cc08ef5381e436d48ce457887ede32",
+]
+
 [tool.uv.sources]
-traust-contracts = { git = "https://github.com/traust-security/traust-contracts.git", rev = "84bd8e4f6cda7431fb4a1321b1d10715f2220860" }
+traust-contracts = { git = "https://github.com/traust-security/traust-contracts.git", rev = "72e641c992cc08ef5381e436d48ce457887ede32" }
 traust-engine = { git = "https://github.com/traust-security/traust-engine.git", rev = "88c8065f4509cc3d3b228cce3413584425b44413" }
 traust-ledger = { git = "https://github.com/traust-security/traust-ledger.git", rev = "c3e6d2ebb1c3105e3ddf9c262198a9514d94291e" }
+traust-core = { git = "https://github.com/traust-security/traust-core.git", rev = "fc6ae2433cc81444be62345ed863e957fc6638c8" }
 
 # Multi-skill CLIs live under src/traust/; harnessing/<skill>/ for single-skill
 # scripts. Install for deps + importable path helpers.

src/traust/cli/fetch_feeds.py

@@ -75,14 +75,15 @@
 import json
 import re
 import sys
-import urllib.error
-import urllib.request
 from pathlib import Path
 
+from traust_core.v1.errors import ServiceError
+from traust_core.v1.providers.feeds import validate_payload
+from traust_core.v1.providers.http import HttpFetcher, RetryPolicy
 from traust_engine import locations
 
 from traust.context import add_config_home_arg, load_engine
-from traust.registry.feeds_config import cached_sources, feeds_cache_dir
+from traust.registry.feeds_config import cached_sources, feeds_cache_dir, retry_defaults
 
 USER_AGENT = "traust fetch_feeds.py"
 
@@ -92,6 +93,16 @@
 # window is ~12 pages, so 60 is far above any real fetch.
 MAX_FEED_PAGES = 60
 
+
+def _retry_policy(name: str) -> RetryPolicy:
+    """Global retry defaults from feeds.yaml, overridden by the feed's own `retry:`."""
+    return RetryPolicy(**{**retry_defaults(), **(_feeds().get(name, {}).get("retry") or {})})
+
+
+def _fetcher(name: str) -> HttpFetcher:
+    return HttpFetcher(USER_AGENT, _retry_policy(name))
+
+
 # The vulnerability-data sources come from config/feeds.yaml — one
 # registry shared with fetch_advisory.py (live tier) and check_drift.py
 # (freshness + liveness rows), so adding a source is a config change and
@@ -199,6 +210,16 @@ def _write_meta(cache: Path, meta: dict):
     (cache / "feeds-meta.json").write_text(json.dumps(meta, indent=2) + "\n", encoding="utf-8")
 
 
+def _record_attempt(cache: Path, meta: dict, name: str, *, error: str) -> None:
+    """Persist a failed refresh so consumers can see a feed went stale and why."""
+    entry = dict(meta.get(name) or {})
+    entry["last_attempt_at"] = _iso(_now())
+    entry["last_error"] = error
+    entry["stale"] = True
+    meta[name] = entry
+    _write_meta(cache, meta)
+
+
 def _age_hours(meta_entry: dict) -> float | None:
     try:
         t = datetime.datetime.strptime(meta_entry["retrieved_at"], "%Y-%m-%dT%H:%M:%SZ").replace(
@@ -249,7 +270,9 @@ def _feed_extra(name: str, raw: bytes) -> dict:
     return {}
 
 
-def _fetch_paginated(spec: dict, since: str | None) -> tuple[list, str | None]:
+def _fetch_paginated(
+    spec: dict, since: str | None, fetcher: HttpFetcher
+) -> tuple[list, str | None]:
     """Page a date-windowed JSON list endpoint. Returns (records, watermark).
 
     Pages *within* the window rather than assuming a window fits one page:
@@ -265,9 +288,7 @@ def _fetch_paginated(spec: dict, since: str | None) -> tuple[list, str | None]:
         q = f"?{pag['page_param']}={page}&per_page={per_page}"
         if since and inc.get("since_param"):
             q += f"&{inc['since_param']}={since}"
-        req = urllib.request.Request(spec["url"] + q, headers={"User-Agent": USER_AGENT})
-        with urllib.request.urlopen(req, timeout=120) as r:
-            batch = json.load(r)
+        batch = json.loads(fetcher.get(spec["url"] + q))
         if not isinstance(batch, list):
             raise ValueError(f"expected a JSON list, got {type(batch).__name__}")
         for rec in batch:
@@ -314,8 +335,9 @@ def fetch(name: str, cache: Path, max_age_hours: float, offline: bool = False) -
         inc = spec.get("incremental") or {}
         since = entry.get("watermark") or inc.get("backfill_from")
         try:
-            fresh, watermark = _fetch_paginated(spec, since)
-        except (urllib.error.URLError, OSError, TimeoutError, ValueError) as e:
+            fresh, watermark = _fetch_paginated(spec, since, _fetcher(name))
+        except (ServiceError, ValueError) as e:
+            _record_attempt(cache, meta, name, error=str(e))
             if path.is_file():
                 return True, (
                     f"STALE ({'unknown age' if age is None else f'{age:.0f}h old'})"
@@ -343,6 +365,9 @@ def fetch(name: str, cache: Path, max_age_hours: float, offline: bool = False) -
             "retrieved_at": _iso(_now()),
             "sha256": hashlib.sha256(raw).hexdigest(),
             "size": len(raw),
+            "last_attempt_at": _iso(_now()),
+            "last_error": None,
+            "stale": False,
             "watermark": watermark,
             "count": len(records),
             "added_last_run": added,
@@ -351,13 +376,12 @@ def fetch(name: str, cache: Path, max_age_hours: float, offline: bool = False) -
         _write_meta(cache, meta)
         return True, (f"refreshed (+{added} new, {len(records)} total, watermark {watermark})")
 
-    req = urllib.request.Request(
-        spec.get("url") or spec["index_url"], headers={"User-Agent": USER_AGENT}
-    )
+    url = spec.get("url") or spec["index_url"]
     try:
-        with urllib.request.urlopen(req, timeout=120) as r:
-            raw = r.read()
-    except (urllib.error.URLError, OSError, TimeoutError) as e:
+        raw = _fetcher(name).get(url)
+        validate_payload(raw, spec.get("file") or spec.get("index_file") or "")
+    except (ServiceError, ValueError) as e:
+        _record_attempt(cache, meta, name, error=str(e))
         if path.is_file():
             return True, (
                 f"STALE ({'unknown age' if age is None else f'{age:.0f}h old'})"
@@ -367,10 +391,13 @@ def fetch(name: str, cache: Path, max_age_hours: float, offline: bool = False) -
 
     path.write_bytes(raw)
     meta[name] = {
-        "url": spec.get("url") or spec["index_url"],
+        "url": url,
         "retrieved_at": _iso(_now()),
         "sha256": hashlib.sha256(raw).hexdigest(),
         "size": len(raw),
+        "last_attempt_at": _iso(_now()),
+        "last_error": None,
+        "stale": False,
         **_feed_extra(name, raw),
     }
     _write_meta(cache, meta)
@@ -403,6 +430,8 @@ def feed_status(cache: Path, max_age_hours: float | None = None) -> dict:
             "age_hours": None if age is None else round(age, 1),
             "max_age_hours": threshold,
             "internal": bool(spec.get("internal")),
+            "last_error": entry.get("last_error"),
+            "last_attempt_at": entry.get("last_attempt_at"),
             "stale": (not present or age is None or age > threshold),
         }
     return out
@@ -502,9 +531,7 @@ def load_vex(cache: Path, cve: str, offline: bool = False) -> dict:
     if offline:
         raise FileNotFoundError(f"{cve} not in the VEX cache and --offline")
     url = spec["doc_url_template"].format(year=year, ident_lower=ident)
-    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
-    with urllib.request.urlopen(req, timeout=60) as r:
-        raw = r.read()
+    raw = _fetcher("vex").get(url)
     doc_path.parent.mkdir(parents=True, exist_ok=True)
     doc_path.write_bytes(raw)
     return json.loads(raw.decode("utf-8"))

src/traust/registry/feeds_config.py

@@ -130,6 +130,19 @@ def load(path: Path | None = None) -> dict[str, dict]:
     return sources
 
 
+def retry_defaults(path: Path | None = None) -> dict:
+    """Global download-retry policy from feeds.yaml ``retry:`` (empty when unset).
+
+    A feed's own ``retry:`` block overrides these; fetch_feeds merges the two.
+    Returning {} rather than raising when absent keeps core's RetryPolicy
+    defaults in force for a registry that predates this block.
+    """
+    retry = _load_raw(path).get("retry") or {}
+    if not isinstance(retry, dict):
+        raise FeedsConfigError("feeds.yaml `retry:` must be a mapping")
+    return retry
+
+
 def cached_sources(path: Path | None = None) -> dict[str, dict]:
     return {k: v for k, v in load(path).items() if v["tier"] == "cached"}
 

tests/test_fetch_feeds.py

@@ -7,12 +7,25 @@
 
 import pytest
 from traust_contracts import Locations
+from traust_core.v1.errors import ServiceError
+from traust_core.v1.providers.http import HttpFetcher
 
 from traust.paths import skill_dir
 
 _ROOT = Path(__file__).resolve().parents[1]
 
 
+def _fake_get(monkeypatch, outcome):
+    """Stub core's HttpFetcher.get: return ``outcome`` bytes or raise it."""
+
+    def _get(self, url):
+        if isinstance(outcome, Exception):
+            raise outcome
+        return outcome
+
+    monkeypatch.setattr(HttpFetcher, "get", _get)
+
+
 def _load(name):
     import importlib
 
@@ -145,24 +158,14 @@ def test_offline_without_copy_fails(tmp_path):
 
 def test_fresh_copy_skips_download(tmp_path, monkeypatch):
     cache = _mk_cache(tmp_path, retrieved_at=ff._iso(ff._now()))
-    monkeypatch.setattr(
-        ff.urllib.request,
-        "urlopen",
-        lambda *a, **k: (_ for _ in ()).throw(AssertionError("network touched")),
-    )
+    _fake_get(monkeypatch, AssertionError("network touched"))
     usable, status = ff.fetch("epss", cache, 24)
     assert usable and status.startswith("fresh")
 
 
 def test_failed_refresh_falls_back(tmp_path, monkeypatch):
     cache = _mk_cache(tmp_path, retrieved_at="2026-01-01T00:00:00Z")
-    import urllib.error
-
-    monkeypatch.setattr(
-        ff.urllib.request,
-        "urlopen",
-        lambda *a, **k: (_ for _ in ()).throw(urllib.error.URLError("no vpn")),
-    )
+    _fake_get(monkeypatch, ServiceError("no vpn"))
     usable, status = ff.fetch("epss", cache, 24)
     assert usable and "STALE" in status and "refresh failed" in status
 
@@ -378,27 +381,12 @@ def _fake_pages(monkeypatch, pages):
     """Serve canned JSON pages, recording the query string of each call."""
     calls = []
 
-    class _Resp:
-        def __init__(self, body):
-            self._b = body
-
-        def read(self):
-            return self._b
-
-        def __enter__(self):
-            return self
-
-        def __exit__(self, *a):
-            return False
-
-    def _open(req, timeout=None):
-        url = req.full_url if hasattr(req, "full_url") else str(req)
+    def _get(self, url):
         calls.append(url)
         page = int(url.split("page=")[1].split("&")[0])
-        body = json.dumps(pages[page - 1] if page <= len(pages) else [])
-        return _Resp(body.encode())
+        return json.dumps(pages[page - 1] if page <= len(pages) else []).encode()
 
-    monkeypatch.setattr(ff.urllib.request, "urlopen", _open)
+    monkeypatch.setattr(HttpFetcher, "get", _get)
     return calls
 
 
@@ -503,3 +491,41 @@ def test_rh_cve_is_watched_by_drift():
     from traust.cli.check_drift import watched_feeds
 
     assert ("rh-cve", False) in watched_feeds()
+
+
+# ---------------- retry / stale (XWING-2223) ----------------
+# Retry/backoff behaviour is tested in traust-core (providers/test_http.py);
+# these cover what fetch_feeds does with the outcome.
+
+
+def test_shipped_retry_config_builds_core_policy():
+    from traust_core.v1.providers.http import RetryPolicy
+
+    from traust.registry.feeds_config import retry_defaults
+
+    shipped = retry_defaults(_ROOT / "config" / "feeds.yaml")
+    assert (
+        RetryPolicy(**shipped).max_attempts == shipped["max_attempts"]
+    )  # unknown key -> TypeError
+
+
+def test_stale_fallback_records_last_error(tmp_path, monkeypatch):
+    cache = _mk_cache(tmp_path, retrieved_at="2026-01-01T00:00:00Z")
+    _fake_get(monkeypatch, ServiceError("http://feed.test: gave up after 4 attempts (no vpn)"))
+    usable, status = ff.fetch("epss", cache, 24)
+    assert usable and "STALE" in status
+    meta = json.loads((cache / "feeds-meta.json").read_text())["epss"]
+    assert meta["stale"] is True
+    assert "no vpn" in meta["last_error"]
+    assert meta["last_attempt_at"]
+
+
+def test_corrupt_download_keeps_old_copy(tmp_path, monkeypatch):
+    cache = _mk_cache(tmp_path, retrieved_at="2026-01-01T00:00:00Z")
+    good = (cache / "epss_scores-current.csv.gz").read_bytes()
+    _fake_get(monkeypatch, b"not a gzip payload")
+    usable, status = ff.fetch("epss", cache, 24)
+    assert usable and "STALE" in status  # corrupt refresh falls back, never unusable here
+    assert (cache / "epss_scores-current.csv.gz").read_bytes() == good  # cache not overwritten
+    meta = json.loads((cache / "feeds-meta.json").read_text())["epss"]
+    assert meta["stale"] is True and "corrupt" in meta["last_error"]

tests/test_ledger_db_snapshot.py

@@ -22,7 +22,11 @@
 
 pytest.importorskip("sqlalchemy")
 from traust_contracts.v1.models.layer import LayerActor
-from traust_contracts.v1.storage import Store
+
+try:
+    from traust_contracts.v1.storage import Store
+except ImportError:  # retired by the one-owner contracts (72e641c); rebuilt upstream
+    pytest.skip("traust-contracts Store retired; awaiting its rebuild", allow_module_level=True)
 
 HARNESS = Path(__file__).resolve().parents[1]
 FIXTURE_CONFIG = Path(__file__).parent / "fixtures" / "config"