feat(journal): honor 429/Retry-After in outbox polling (§7.4)
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) <noreply@anthropic.com>
This commit is contained in:
parent
60afacc69e
commit
3c2bdfd2bd
4 changed files with 161 additions and 6 deletions
|
|
@ -3,10 +3,12 @@ import { eq, like } from "drizzle-orm";
|
||||||
import { randomUUID } from "node:crypto";
|
import { randomUUID } from "node:crypto";
|
||||||
import { getDb } from "./db.ts";
|
import { getDb } from "./db.ts";
|
||||||
import { users, follows, activities, remoteActors } from "@trails-cool/db/schema/journal";
|
import { users, follows, activities, remoteActors } from "@trails-cool/db/schema/journal";
|
||||||
|
import { FetchError } from "@fedify/fedify";
|
||||||
import {
|
import {
|
||||||
ingestRemoteActivities,
|
ingestRemoteActivities,
|
||||||
pollRemoteActor,
|
pollRemoteActor,
|
||||||
listActorsDuePolling,
|
listActorsDuePolling,
|
||||||
|
resetHostBackoff,
|
||||||
type ParsedRemoteActivity,
|
type ParsedRemoteActivity,
|
||||||
} from "./federation-ingest.server.ts";
|
} from "./federation-ingest.server.ts";
|
||||||
|
|
||||||
|
|
@ -56,6 +58,7 @@ describe.runIf(runIntegration)("outbox-poll ingestion (integration)", () => {
|
||||||
});
|
});
|
||||||
|
|
||||||
afterEach(async () => {
|
afterEach(async () => {
|
||||||
|
resetHostBackoff();
|
||||||
const db = getDb();
|
const db = getDb();
|
||||||
await db.delete(activities).where(like(activities.remoteOriginIri, `${ACTOR}%`));
|
await db.delete(activities).where(like(activities.remoteOriginIri, `${ACTOR}%`));
|
||||||
await db.delete(remoteActors).where(eq(remoteActors.actorIri, 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" });
|
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<unknown> {
|
||||||
|
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 () => {
|
it("listActorsDuePolling returns followed actors not polled within the hour", async () => {
|
||||||
await makeFollowerOf(ACTOR);
|
await makeFollowerOf(ACTOR);
|
||||||
expect(await listActorsDuePolling()).toContain(ACTOR);
|
expect(await listActorsDuePolling()).toContain(ACTOR);
|
||||||
|
|
|
||||||
|
|
@ -1,8 +1,15 @@
|
||||||
import { describe, it, expect, vi } from "vitest";
|
import { describe, it, expect, vi, afterEach } from "vitest";
|
||||||
|
|
||||||
vi.mock("./db.ts", () => ({ getDb: () => ({}) }));
|
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<string, unknown> = {}, noteOverrides: Record<string, unknown> = {}) {
|
function trailsCreate(overrides: Record<string, unknown> = {}, noteOverrides: Record<string, unknown> = {}) {
|
||||||
return {
|
return {
|
||||||
|
|
@ -78,3 +85,47 @@ describe("parseOutboxItem", () => {
|
||||||
expect(parsed?.name).not.toContain("<b>");
|
expect(parsed?.name).not.toContain("<b>");
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
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)",
|
||||||
|
});
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
|
||||||
|
|
@ -8,6 +8,7 @@
|
||||||
//
|
//
|
||||||
// Network steps are injectable for offline integration tests.
|
// 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 { and, eq, isNull, isNotNull, lt, or, sql } from "drizzle-orm";
|
||||||
import { randomUUID } from "node:crypto";
|
import { randomUUID } from "node:crypto";
|
||||||
import { activities, follows, remoteActors, users } from "@trails-cool/db/schema/journal";
|
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;
|
const HOST_PACE_MS = 5_000;
|
||||||
/** Stop early after this many consecutive already-seen items (7.2). */
|
/** Stop early after this many consecutive already-seen items (7.2). */
|
||||||
const CONFLICT_STREAK_LIMIT = 5;
|
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 {
|
export interface ParsedRemoteActivity {
|
||||||
originIri: string;
|
originIri: string;
|
||||||
|
|
@ -159,6 +168,59 @@ export interface PollDeps {
|
||||||
|
|
||||||
const lastFetchPerHost = new Map<string, number>();
|
const lastFetchPerHost = new Map<string, number>();
|
||||||
|
|
||||||
|
// 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<string, number>();
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 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 {
|
function defaultDeps(): PollDeps {
|
||||||
return {
|
return {
|
||||||
async fetchJson(url, signerUsername) {
|
async fetchJson(url, signerUsername) {
|
||||||
|
|
@ -186,6 +248,10 @@ export async function pollRemoteActor(
|
||||||
actorIri: string,
|
actorIri: string,
|
||||||
deps: PollDeps = defaultDeps(),
|
deps: PollDeps = defaultDeps(),
|
||||||
): Promise<{ inserted: number } | { skipped: string }> {
|
): 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();
|
const db = getDb();
|
||||||
|
|
||||||
// Signer: any local user with an accepted follow against this actor.
|
// Signer: any local user with an accepted follow against this actor.
|
||||||
|
|
@ -203,7 +269,6 @@ export async function pollRemoteActor(
|
||||||
.limit(1);
|
.limit(1);
|
||||||
if (!signer) return { skipped: "no accepted local follower" };
|
if (!signer) return { skipped: "no accepted local follower" };
|
||||||
|
|
||||||
const host = new URL(actorIri).host;
|
|
||||||
try {
|
try {
|
||||||
await deps.pace(host);
|
await deps.pace(host);
|
||||||
|
|
||||||
|
|
@ -251,7 +316,14 @@ export async function pollRemoteActor(
|
||||||
logger.info({ actorIri, fetched: items.length, inserted }, "federation: outbox poll");
|
logger.info({ actorIri, fetched: items.length, inserted }, "federation: outbox poll");
|
||||||
return { inserted };
|
return { inserted };
|
||||||
} catch (err) {
|
} 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");
|
logger.warn({ err, actorIri }, "federation: outbox poll failed; will retry on next sweep");
|
||||||
return { skipped: "fetch failed" };
|
return { skipped: "fetch failed" };
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -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.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.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)
|
- [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
|
- [x] 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.
|
> 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)
|
- [x] 7.5 First-poll trigger on `accepted_at` transition (already enqueued from 4.4)
|
||||||
|
|
||||||
## 8. Feed query update
|
## 8. Feed query update
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue