Honor Retry-After on HTTP 429 (polite rest, not a failure)
Per Codex's spec — a publisher saying "slow down" shouldn't make a feed look broken, but repeated 429s stay visible via last_success_at / stale-source. * Schema: sources.retry_after_at (nullable) + migration. * feeds.parse_retry_after: delta-seconds OR HTTP-date → UTC stamp; ignores invalid/negative/past; caps at now + MAX_BACKOFF_MINUTES. * fetch_feed raises RateLimited (carrying the parsed time) on a 429. * poll_source: on 429 set retry_after_at + last_error, status='rate_limited', and do NOT increment consecutive_failures; on success clear retry_after_at; non-429 failures unchanged. * due_source_rows requires BOTH the streak backoff elapsed AND retry_after_at passed (i.e. the later of the two). * Admin: source_health returns retry_after_at; status reads "rate-limited · rests until …" rather than "failed/resting". Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -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 @@
|
||||
<td class="status">
|
||||
{#if st === 'retired'}<span class="paustxt">retired</span>
|
||||
{:else if st === 'paused'}<span class="paustxt">paused</span>
|
||||
{:else if rateLimited(s)}<span class="bad" title={s.last_error || ''}>rate-limited · rests until {fwhen(s.retry_after_at)}{#if s.review_flag} · review{/if}</span>
|
||||
{:else if s.failures > 0}<span class="bad" title={s.last_error || ''}>⚠ resting{#if s.review_flag} · review{/if}</span>
|
||||
{:else if s.review_flag}<span class="flagtxt">⚑ review</span>
|
||||
{:else}<span class="good">✓ ok</span>{/if}
|
||||
|
||||
@@ -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)")}
|
||||
|
||||
+70
-3
@@ -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
|
||||
|
||||
+1
-1
@@ -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,
|
||||
|
||||
@@ -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"<rss><channel></channel></rss>")
|
||||
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
|
||||
Reference in New Issue
Block a user