diff --git a/frontend/src/routes/admin/+page.svelte b/frontend/src/routes/admin/+page.svelte index cd6bae2..4f4d53f 100644 --- a/frontend/src/routes/admin/+page.svelte +++ b/frontend/src/routes/admin/+page.svelte @@ -64,6 +64,12 @@ let sources = $derived(stats?.sources ?? []); // status falls back to the legacy active flag during the migration window. const sstatus = (s) => s.status || (s.active ? 'active' : 'paused'); + // A 429 rest window that hasn't elapsed — polite wait, not breakage. + function rateLimited(s) { + if (!s.retry_after_at) return false; + const d = new Date(s.retry_after_at.replace(' ', 'T') + 'Z'); + return !isNaN(d) && d > new Date(); + } let healthy = $derived(sources.filter((s) => sstatus(s) === 'active' && !s.failures && !s.review_flag).length); let resting = $derived(sources.filter((s) => sstatus(s) === 'active' && s.failures > 0).length); let flagged = $derived(sources.filter((s) => s.review_flag).length); @@ -460,6 +466,7 @@ {#if st === 'retired'}retired {:else if st === 'paused'}paused + {:else if rateLimited(s)}rate-limited · rests until {fwhen(s.retry_after_at)}{#if s.review_flag} · review{/if} {:else if s.failures > 0}⚠ resting{#if s.review_flag} · review{/if} {:else if s.review_flag}⚑ review {:else}✓ ok{/if} diff --git a/goodnews/db.py b/goodnews/db.py index 23afd72..da35a0a 100644 --- a/goodnews/db.py +++ b/goodnews/db.py @@ -25,6 +25,7 @@ CREATE TABLE IF NOT EXISTS sources ( last_error_at TEXT, last_error TEXT, consecutive_failures INTEGER NOT NULL DEFAULT 0, + retry_after_at TEXT, review_flag INTEGER NOT NULL DEFAULT 0, review_reason TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, @@ -326,6 +327,8 @@ def _migrate(conn: sqlite3.Connection) -> None: conn.execute("UPDATE sources SET status = CASE WHEN active = 1 THEN 'active' ELSE 'paused' END") if "content_visible" not in source_cols: conn.execute("ALTER TABLE sources ADD COLUMN content_visible INTEGER NOT NULL DEFAULT 1") + if "retry_after_at" not in source_cols: + conn.execute("ALTER TABLE sources ADD COLUMN retry_after_at TEXT") # feedback.read_at (admin inbox read/unread) added later. fb_cols = {row["name"] for row in conn.execute("PRAGMA table_info(feedback)")} diff --git a/goodnews/feeds.py b/goodnews/feeds.py index 6f05ff1..1e6113f 100644 --- a/goodnews/feeds.py +++ b/goodnews/feeds.py @@ -8,7 +8,7 @@ import urllib.request import xml.etree.ElementTree as ET from collections import Counter from dataclasses import dataclass -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from urllib.parse import urljoin, urlsplit from .enrich import MAX_REDIRECTS, _NoRedirect, _host_is_public @@ -45,6 +45,47 @@ def poll_all_sources(conn: sqlite3.Connection, limit: int | None = None) -> dict MAX_BACKOFF_MINUTES = 1440 +class RateLimited(RuntimeError): + """A feed returned HTTP 429. Carries the parsed retry-after timestamp (a UTC + 'YYYY-MM-DD HH:MM:SS' string, or None) so the caller can rest politely rather + than treating it as operational breakage.""" + + def __init__(self, message: str, retry_after_at: str | None = None) -> None: + super().__init__(message) + self.retry_after_at = retry_after_at + + +def parse_retry_after(value: str | None, now: datetime | None = None) -> str | None: + """Parse a Retry-After header → UTC 'YYYY-MM-DD HH:MM:SS', or None. + + Accepts delta-seconds or an HTTP-date; ignores invalid/negative/past values; + caps the result at now + MAX_BACKOFF_MINUTES so a feed can't park itself. + """ + if not value: + return None + now = now or datetime.now(UTC) + value = value.strip() + target = None + if value.isdigit(): + target = now + timedelta(seconds=int(value)) + else: + try: + dt = email.utils.parsedate_to_datetime(value) + except (TypeError, ValueError, IndexError): + dt = None + if dt is not None: + target = dt if dt.tzinfo else dt.replace(tzinfo=UTC) + if target is None: + return None + target = target.astimezone(UTC) + if target <= now: + return None # negative / past → ignore + cap = now + timedelta(minutes=MAX_BACKOFF_MINUTES) + if target > cap: + target = cap + return target.strftime("%Y-%m-%d %H:%M:%S") + + def poll_due_sources(conn: sqlite3.Connection, limit: int | None = None) -> dict: """Poll only active sources whose last *attempt* (success OR failure) is older than their effective interval, or that have never been polled. @@ -68,6 +109,7 @@ def due_source_rows(conn: sqlite3.Connection) -> list[sqlite3.Row]: SELECT s.* FROM sources s WHERE s.active = 1 + AND (s.retry_after_at IS NULL OR s.retry_after_at <= datetime('now')) AND ( NOT EXISTS ( SELECT 1 FROM ingest_runs r @@ -133,17 +175,39 @@ def poll_source(conn: sqlite3.Connection, source: sqlite3.Row) -> dict: """, (seen, inserted, duplicate, run_id), ) - # A clean poll resets the failure streak and records the success time. + # A clean poll resets the failure streak, clears any rate-limit rest, and + # records the success time. conn.execute( """ UPDATE sources - SET last_success_at = CURRENT_TIMESTAMP, consecutive_failures = 0 + SET last_success_at = CURRENT_TIMESTAMP, consecutive_failures = 0, retry_after_at = NULL WHERE id = ? """, (source["id"],), ) conn.commit() return {"status": "ok", "seen": seen, "inserted": inserted, "duplicate": duplicate} + except RateLimited as exc: + # The publisher asked us to wait — not operational breakage. Record a + # polite rest window WITHOUT inflating the failure streak. + conn.execute( + """ + UPDATE ingest_runs + SET finished_at = CURRENT_TIMESTAMP, status = 'rate_limited', + items_seen = ?, items_inserted = ?, items_duplicate = ?, error = ? + WHERE id = ? + """, + (seen, inserted, duplicate, str(exc), run_id), + ) + conn.execute( + "UPDATE sources SET last_error_at = CURRENT_TIMESTAMP, last_error = ?, retry_after_at = ? WHERE id = ?", + (str(exc), exc.retry_after_at, source["id"]), + ) + conn.commit() + return { + "status": "rate_limited", "seen": seen, "inserted": inserted, + "duplicate": duplicate, "retry_after_at": exc.retry_after_at, + } except Exception as exc: conn.execute( """ @@ -286,6 +350,9 @@ def fetch_feed(url: str, timeout: int = 20) -> bytes: with urllib.request.urlopen(request, timeout=timeout) as response: return response.read() except urllib.error.HTTPError as exc: + if exc.code == 429: + ra = parse_retry_after(exc.headers.get("Retry-After")) if exc.headers else None + raise RateLimited(f"HTTP 429 rate limited fetching {url}", retry_after_at=ra) from exc raise RuntimeError(f"HTTP {exc.code} fetching {url}") from exc except urllib.error.URLError as exc: raise RuntimeError(f"failed fetching {url}: {exc.reason}") from exc diff --git a/goodnews/queries.py b/goodnews/queries.py index 7a51ab1..fb76234 100644 --- a/goodnews/queries.py +++ b/goodnews/queries.py @@ -295,7 +295,7 @@ def source_health(conn: sqlite3.Connection) -> list[dict]: """ SELECT s.id, s.name, s.default_category AS category, s.active, - s.status, s.content_visible, + s.status, s.content_visible, s.retry_after_at, s.consecutive_failures AS failures, s.review_flag, s.review_reason, s.poll_interval_minutes AS interval_minutes, s.last_success_at, s.last_error_at, substr(s.last_error, 1, 160) AS last_error, diff --git a/tests/test_retry_after.py b/tests/test_retry_after.py new file mode 100644 index 0000000..957fdd6 --- /dev/null +++ b/tests/test_retry_after.py @@ -0,0 +1,65 @@ +from datetime import UTC, datetime, timedelta + +from goodnews import feeds +from goodnews.db import connect, init_db + + +NOW = datetime(2026, 6, 9, 12, 0, 0, tzinfo=UTC) + + +def test_parse_delta_seconds(): + assert feeds.parse_retry_after("120", now=NOW) == "2026-06-09 12:02:00" + + +def test_parse_http_date(): + assert feeds.parse_retry_after("Tue, 09 Jun 2026 12:30:00 GMT", now=NOW) == "2026-06-09 12:30:00" + + +def test_parse_ignores_invalid_and_past(): + assert feeds.parse_retry_after("", now=NOW) is None + assert feeds.parse_retry_after("soon", now=NOW) is None + assert feeds.parse_retry_after("-30", now=NOW) is None # negative + assert feeds.parse_retry_after("Tue, 09 Jun 2020 00:00:00 GMT", now=NOW) is None # past + + +def test_parse_caps_at_max_backoff(): + capped = feeds.parse_retry_after(str(60 * 60 * 24 * 30), now=NOW) # 30 days + assert capped == (NOW + timedelta(minutes=feeds.MAX_BACKOFF_MINUTES)).strftime("%Y-%m-%d %H:%M:%S") + + +def _src(c): + c.execute("INSERT INTO sources (id,name,feed_url,active) VALUES (1,'S','http://s/f',1)") + c.commit() + return c.execute("SELECT * FROM sources WHERE id=1").fetchone() + + +def test_429_sets_retry_after_without_streak(monkeypatch): + c = connect(":memory:"); init_db(c); src = _src(c) + def boom(url, timeout=20): + raise feeds.RateLimited("HTTP 429", retry_after_at="2030-01-01 00:00:00") + monkeypatch.setattr(feeds, "fetch_feed", boom) + res = feeds.poll_source(c, src) + assert res["status"] == "rate_limited" + row = c.execute("SELECT consecutive_failures, retry_after_at FROM sources WHERE id=1").fetchone() + assert row["consecutive_failures"] == 0 # NOT inflated + assert row["retry_after_at"] == "2030-01-01 00:00:00" + # and it's not due while resting + assert [s["id"] for s in feeds.due_source_rows(c)] == [] + + +def test_success_clears_retry_after(monkeypatch): + c = connect(":memory:"); init_db(c); src = _src(c) + c.execute("UPDATE sources SET retry_after_at='2030-01-01 00:00:00', consecutive_failures=2 WHERE id=1") + c.commit() + monkeypatch.setattr(feeds, "fetch_feed", lambda url, timeout=20: b"") + feeds.poll_source(c, src) + row = c.execute("SELECT consecutive_failures, retry_after_at FROM sources WHERE id=1").fetchone() + assert row["retry_after_at"] is None and row["consecutive_failures"] == 0 + + +def test_non_429_failure_still_increments_streak(monkeypatch): + c = connect(":memory:"); init_db(c); src = _src(c) + monkeypatch.setattr(feeds, "fetch_feed", lambda url, timeout=20: (_ for _ in ()).throw(RuntimeError("HTTP 500"))) + res = feeds.poll_source(c, src) + assert res["status"] == "failed" + assert c.execute("SELECT consecutive_failures FROM sources WHERE id=1").fetchone()[0] == 1