From 3c2bdfd2bdf9584527f0bfd310fc98af0975a0ad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ullrich=20Sch=C3=A4fer?= Date: Sun, 7 Jun 2026 12:35:13 +0200 Subject: [PATCH] =?UTF-8?q?feat(journal):=20honor=20429/Retry-After=20in?= =?UTF-8?q?=20outbox=20polling=20(=C2=A77.4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes the 7.4 remainder. Fedify's FetchError carries the failed Response, so a 429 from a remote's outbox now reads Retry-After (delta-seconds or HTTP-date per RFC 9110), arms a per-host backoff (15 min default, capped at the 1 h poll interval), and pollRemoteActor skips backing-off hosts before doing any DB or network work. State is in-process like the pacing map — a restart costs at most one extra request that re-arms the backoff. last_polled_at is deliberately not stamped on a rate-limited poll so the hourly sweep retries. Unit tests cover the header parser (incl. the V8 footgun where Date.parse reads bare negative integers as years), default/cap behavior, and window expiry; an integration test drives the full FetchError path and asserts the second poll never touches the network. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../lib/federation-ingest.integration.test.ts | 32 ++++++++ .../app/lib/federation-ingest.server.test.ts | 55 +++++++++++++- .../app/lib/federation-ingest.server.ts | 76 ++++++++++++++++++- openspec/changes/social-federation/tasks.md | 4 +- 4 files changed, 161 insertions(+), 6 deletions(-) diff --git a/apps/journal/app/lib/federation-ingest.integration.test.ts b/apps/journal/app/lib/federation-ingest.integration.test.ts index 1f746e1..d5fcd98 100644 --- a/apps/journal/app/lib/federation-ingest.integration.test.ts +++ b/apps/journal/app/lib/federation-ingest.integration.test.ts @@ -3,10 +3,12 @@ import { eq, like } from "drizzle-orm"; import { randomUUID } from "node:crypto"; import { getDb } from "./db.ts"; import { users, follows, activities, remoteActors } from "@trails-cool/db/schema/journal"; +import { FetchError } from "@fedify/fedify"; import { ingestRemoteActivities, pollRemoteActor, listActorsDuePolling, + resetHostBackoff, type ParsedRemoteActivity, } from "./federation-ingest.server.ts"; @@ -56,6 +58,7 @@ describe.runIf(runIntegration)("outbox-poll ingestion (integration)", () => { }); afterEach(async () => { + resetHostBackoff(); const db = getDb(); await db.delete(activities).where(like(activities.remoteOriginIri, `${ACTOR}%`)); await db.delete(remoteActors).where(eq(remoteActors.actorIri, ACTOR)); @@ -153,6 +156,35 @@ describe.runIf(runIntegration)("outbox-poll ingestion (integration)", () => { expect(result).toEqual({ skipped: "no accepted local follower" }); }); + it("a 429 with Retry-After backs off the host; the next poll skips without fetching (7.4)", async () => { + await makeFollowerOf(ACTOR); + let fetches = 0; + const deps = { + async pace() {}, + async fetchJson(url: string): Promise { + fetches++; + throw new FetchError( + new URL(url), + "rate limited", + new Response(null, { status: 429, headers: { "Retry-After": "120" } }), + ); + }, + }; + expect(await pollRemoteActor(ACTOR, deps)).toEqual({ skipped: "rate limited" }); + expect(fetches).toBe(1); + + // Host is now backing off — no network attempt at all. + expect(await pollRemoteActor(ACTOR, deps)).toEqual({ + skipped: "host rate-limited (backing off)", + }); + expect(fetches).toBe(1); + + // last_polled_at was NOT stamped, so the sweep will retry later. + const db = getDb(); + const [cached] = await db.select().from(remoteActors).where(eq(remoteActors.actorIri, ACTOR)); + expect(cached?.lastPolledAt ?? null).toBeNull(); + }); + it("listActorsDuePolling returns followed actors not polled within the hour", async () => { await makeFollowerOf(ACTOR); expect(await listActorsDuePolling()).toContain(ACTOR); diff --git a/apps/journal/app/lib/federation-ingest.server.test.ts b/apps/journal/app/lib/federation-ingest.server.test.ts index 8167e53..80f1fda 100644 --- a/apps/journal/app/lib/federation-ingest.server.test.ts +++ b/apps/journal/app/lib/federation-ingest.server.test.ts @@ -1,8 +1,15 @@ -import { describe, it, expect, vi } from "vitest"; +import { describe, it, expect, vi, afterEach } from "vitest"; vi.mock("./db.ts", () => ({ getDb: () => ({}) })); -const { parseOutboxItem } = await import("./federation-ingest.server.ts"); +const { + parseOutboxItem, + parseRetryAfter, + noteHostRateLimited, + isHostBackingOff, + resetHostBackoff, + pollRemoteActor, +} = await import("./federation-ingest.server.ts"); function trailsCreate(overrides: Record = {}, noteOverrides: Record = {}) { return { @@ -78,3 +85,47 @@ describe("parseOutboxItem", () => { expect(parsed?.name).not.toContain(""); }); }); + +describe("429 backoff (7.4)", () => { + afterEach(() => resetHostBackoff()); + + const NOW = Date.parse("2026-06-07T12:00:00Z"); + + it("parses Retry-After as delta-seconds and HTTP-date", () => { + expect(parseRetryAfter("120", NOW)).toBe(120_000); + expect(parseRetryAfter(" 5 ", NOW)).toBe(5_000); + expect(parseRetryAfter("Sun, 07 Jun 2026 12:01:00 GMT", NOW)).toBe(60_000); + // past HTTP-date clamps to zero + expect(parseRetryAfter("Sun, 07 Jun 2026 11:00:00 GMT", NOW)).toBe(0); + expect(parseRetryAfter(null, NOW)).toBeNull(); + expect(parseRetryAfter("soon", NOW)).toBeNull(); + expect(parseRetryAfter("-5", NOW)).toBeNull(); + }); + + it("falls back to the default and caps absurd Retry-After values", () => { + // default (no header): 15 min + expect(noteHostRateLimited("a.example", null, NOW)).toBe(15 * 60 * 1000); + // honored when reasonable + expect(noteHostRateLimited("b.example", "120", NOW)).toBe(120_000); + // capped at the poll interval (1 h) + expect(noteHostRateLimited("c.example", String(7 * 24 * 3600), NOW)).toBe(60 * 60 * 1000); + }); + + it("backs off the host until the window expires", () => { + noteHostRateLimited("d.example", "60", NOW); + expect(isHostBackingOff("d.example", NOW + 59_000)).toBe(true); + expect(isHostBackingOff("d.example", NOW + 61_000)).toBe(false); + // expiry clears the entry; a later check stays false + expect(isHostBackingOff("d.example", NOW)).toBe(false); + expect(isHostBackingOff("other.example", NOW)).toBe(false); + }); + + it("pollRemoteActor skips a backing-off host before touching the database", async () => { + // getDb is mocked to {} — any query would throw, so reaching the + // skip proves the backoff check runs first. + noteHostRateLimited("limited.example", "3600"); + await expect(pollRemoteActor("https://limited.example/users/alice")).resolves.toEqual({ + skipped: "host rate-limited (backing off)", + }); + }); +}); diff --git a/apps/journal/app/lib/federation-ingest.server.ts b/apps/journal/app/lib/federation-ingest.server.ts index ae7fa5d..767a95f 100644 --- a/apps/journal/app/lib/federation-ingest.server.ts +++ b/apps/journal/app/lib/federation-ingest.server.ts @@ -8,6 +8,7 @@ // // Network steps are injectable for offline integration tests. +import { FetchError } from "@fedify/fedify"; import { and, eq, isNull, isNotNull, lt, or, sql } from "drizzle-orm"; import { randomUUID } from "node:crypto"; import { activities, follows, remoteActors, users } from "@trails-cool/db/schema/journal"; @@ -25,6 +26,14 @@ const POLL_INTERVAL_MS = 60 * 60 * 1000; const HOST_PACE_MS = 5_000; /** Stop early after this many consecutive already-seen items (7.2). */ const CONFLICT_STREAK_LIMIT = 5; +/** Backoff after a 429 without a usable Retry-After (7.4). */ +const DEFAULT_BACKOFF_MS = 15 * 60 * 1000; +/** + * Cap an honored Retry-After at the poll interval — anything longer is + * equivalent to "skip until a later sweep", and an absurd header from + * a misbehaving remote shouldn't park a host for days. + */ +const MAX_BACKOFF_MS = POLL_INTERVAL_MS; export interface ParsedRemoteActivity { originIri: string; @@ -159,6 +168,59 @@ export interface PollDeps { const lastFetchPerHost = new Map(); +// 7.4: hosts that answered 429 are skipped until their backoff expires. +// In-process state, like the pacing map above — a restart forgets it, +// which is fine: the worst case is one extra request that gets another +// 429 and re-arms the backoff. +const hostBackoffUntil = new Map(); + +/** + * Parse a Retry-After header value (RFC 9110: delta-seconds or an + * HTTP-date) into a millisecond delay from `now`. Null if absent or + * unparseable. + */ +export function parseRetryAfter(value: string | null, now: number): number | null { + if (value == null) return null; + const trimmed = value.trim(); + // Bare integers are delta-seconds; negative ones are invalid (and + // must not fall through to Date.parse, which reads them as years). + if (/^-?\d+$/.test(trimmed)) { + return /^\d+$/.test(trimmed) ? Number(trimmed) * 1000 : null; + } + const date = Date.parse(trimmed); + if (!Number.isNaN(date)) return Math.max(0, date - now); + return null; +} + +/** + * Record a 429 from `host`: back off for the remote's Retry-After + * (capped) or a default. Returns the applied delay in ms. + */ +export function noteHostRateLimited( + host: string, + retryAfter: string | null, + now = Date.now(), +): number { + const delay = Math.min(parseRetryAfter(retryAfter, now) ?? DEFAULT_BACKOFF_MS, MAX_BACKOFF_MS); + hostBackoffUntil.set(host, now + delay); + return delay; +} + +/** Whether `host` is inside a 429 backoff window. */ +export function isHostBackingOff(host: string, now = Date.now()): boolean { + const until = hostBackoffUntil.get(host) ?? 0; + if (until <= now) { + hostBackoffUntil.delete(host); + return false; + } + return true; +} + +/** Test hook: forget all 429 backoff state. */ +export function resetHostBackoff(): void { + hostBackoffUntil.clear(); +} + function defaultDeps(): PollDeps { return { async fetchJson(url, signerUsername) { @@ -186,6 +248,10 @@ export async function pollRemoteActor( actorIri: string, deps: PollDeps = defaultDeps(), ): Promise<{ inserted: number } | { skipped: string }> { + // 7.4: respect an active 429 backoff before doing any work. + const host = new URL(actorIri).host; + if (isHostBackingOff(host)) return { skipped: "host rate-limited (backing off)" }; + const db = getDb(); // Signer: any local user with an accepted follow against this actor. @@ -203,7 +269,6 @@ export async function pollRemoteActor( .limit(1); if (!signer) return { skipped: "no accepted local follower" }; - const host = new URL(actorIri).host; try { await deps.pace(host); @@ -251,7 +316,14 @@ export async function pollRemoteActor( logger.info({ actorIri, fetched: items.length, inserted }, "federation: outbox poll"); return { inserted }; } catch (err) { - // 429 / network failures: log and let the hourly sweep retry. + // 429: honor Retry-After (or default) and skip the host until the + // backoff expires (7.4). + if (err instanceof FetchError && err.response?.status === 429) { + const delayMs = noteHostRateLimited(host, err.response.headers.get("retry-after")); + logger.warn({ actorIri, host, delayMs }, "federation: remote rate-limited poll; backing off host"); + return { skipped: "rate limited" }; + } + // Other network failures: log and let the hourly sweep retry. logger.warn({ err, actorIri }, "federation: outbox poll failed; will retry on next sweep"); return { skipped: "fetch failed" }; } diff --git a/openspec/changes/social-federation/tasks.md b/openspec/changes/social-federation/tasks.md index 0594e26..5ef6bab 100644 --- a/openspec/changes/social-federation/tasks.md +++ b/openspec/changes/social-federation/tasks.md @@ -60,8 +60,8 @@ - [x] 7.1 pg-boss recurring job `poll-remote-outboxes` (cron every 5 min): iterate distinct accepted remote actor IRIs in `follows` where `remote_actors.last_polled_at < now() - interval '1 hour'`; for each, signed GET on the outbox using one of the local accepted-followers' keys - [x] 7.2 Insert returned activities into `journal.activities` with `remote_origin_iri`, `remote_actor_iri`, `audience`. `ON CONFLICT (remote_origin_iri) DO NOTHING` for replay safety. Stop early on streak of conflicts - [x] 7.3 Refresh `remote_actors` cache row on each poll (display name, avatar, outbox URL, public key) -- [ ] 7.4 Per-remote-host rate limit (1 req / 5 sec); honor `Retry-After` / `429`; back off the host - > Partially done (2026-06-07): 1 req/5 s pacing implemented; any fetch failure (incl. 429) skips the host until the next 5-min sweep. Explicit Retry-After-duration backoff remains — needs header access through Fedify's document loader errors. +- [x] 7.4 Per-remote-host rate limit (1 req / 5 sec); honor `Retry-After` / `429`; back off the host + > Done (2026-06-07): 1 req/5 s pacing + explicit 429 backoff. Fedify's `FetchError` carries the failed `Response`, so `Retry-After` is read from `err.response.headers` (delta-seconds or HTTP-date, capped at the 1 h poll interval; 15 min default). Backing-off hosts are skipped before any DB/network work; in-process state (a restart costs at most one extra 429). - [x] 7.5 First-poll trigger on `accepted_at` transition (already enqueued from 4.4) ## 8. Feed query update