From e37ef52a97c6512cbf88c0bb1392487a89c09e29 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 27 Oct 2026 23:20:42 +0000 Subject: [PATCH 1/3] feat(activitypub): double-knock outbound delivery signatures Sign each outbound inbox POST with RFC 9421 first and, when the peer answers 400/401/403, re-sign the same body with draft-cavage and send once more. Hosts that only accept draft-cavage are remembered in the actor DO's SQLite (`sig_pref`) for 7 days so later deliveries skip the first knock. Outbound GETs stay draft-cavage. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_015LRGjmHE99WESN5HBERxSj --- .changeset/activitypub-double-knock.md | 5 + packages/activitypub/README.md | 18 ++-- packages/activitypub/src/delivery.test.ts | 93 +++++++++++++++- packages/activitypub/src/delivery.ts | 102 ++++++++++++++---- packages/activitypub/src/index.test.ts | 86 ++++++++++++++- packages/activitypub/src/index.ts | 1 + packages/activitypub/src/object.ts | 48 +++++++++ packages/activitypub/src/signature.test.ts | 32 ++++++ packages/activitypub/src/signature.ts | 49 ++++++++- ...central-do-activitypub.integration.test.ts | 4 +- .../phase5-activitypub.integration.test.ts | 4 +- spec/packages/activitypub.md | 9 ++ 12 files changed, 413 insertions(+), 38 deletions(-) create mode 100644 .changeset/activitypub-double-knock.md diff --git a/.changeset/activitypub-double-knock.md b/.changeset/activitypub-double-knock.md new file mode 100644 index 00000000..89df43f8 --- /dev/null +++ b/.changeset/activitypub-double-knock.md @@ -0,0 +1,5 @@ +--- +"@dwk/activitypub": minor +--- + +Sign outbound deliveries with RFC 9421 HTTP Message Signatures first, falling back to draft-cavage when a peer rejects them ("double-knock"). Peers that only accept draft-cavage are remembered per actor for 7 days. Adds `signRequestRfc9421`; `DeliveryResult` now reports the accepted `profile` on success. diff --git a/packages/activitypub/README.md b/packages/activitypub/README.md index 528e7e6f..a3c4b031 100644 --- a/packages/activitypub/README.md +++ b/packages/activitypub/README.md @@ -96,17 +96,21 @@ follower lists, dedup, and the delivery queue all live in the DO's SQLite. ## HTTP signatures -Inbound `POST /inbox` deliveries are authenticated, and outbound deliveries are -signed, with the de-facto fediverse `draft-cavage-http-signatures` profile -(RSA-SHA256 over a covered header set, body integrity via `Digest`). The +Inbound `POST /inbox` deliveries are verified against either the de-facto +fediverse `draft-cavage-http-signatures` profile (RSA-SHA256 over a covered +header set, body integrity via `Digest`) or +[RFC 9421](https://www.rfc-editor.org/rfc/rfc9421) HTTP Message Signatures +(via [`@dwk/http-signatures`](../http-signatures/README.md)). The implementation is RSA-only with an explicit algorithm allow-list (no `none`, no symmetric algorithms), mirroring the [`@dwk/dpop`](../dpop/README.md) hardening posture. -Signing/verification sit behind the `verifyInboxSignature` config seam, so the -forthcoming cross-standard `@dwk/http-signatures` package (RFC 9421 + -draft-cavage; [#59](https://github.com/davidwkeith/workers/issues/59)) can be -swapped in unchanged once it lands. +Outbound deliveries **double-knock**: each is signed with RFC 9421 first, and +if the peer answers `400`/`401`/`403` the same body is re-signed with +draft-cavage and sent once more. A host that only accepts draft-cavage is +remembered in the actor's Durable Object for 7 days, so later deliveries skip +the first knock (the same strategy [Fedify](https://fedify.dev/manual/send#double-knocking-http-signatures) +uses). Outbound `GET`s (authorized-fetch key resolution) stay draft-cavage. Every outbound request (key resolution, delivery, discovery) sends a descriptive `User-Agent` (`userAgent` in `ActivityPubConfig`, defaulting to diff --git a/packages/activitypub/src/delivery.test.ts b/packages/activitypub/src/delivery.test.ts index 982d9bae..280d582a 100644 --- a/packages/activitypub/src/delivery.test.ts +++ b/packages/activitypub/src/delivery.test.ts @@ -99,13 +99,20 @@ describe("deliverActivity", () => { TEST_UA, ); - expect(result).toEqual({ ok: true, status: 202 }); + expect(result).toEqual({ + ok: true, + status: 202, + profile: "rfc9421", + }); expect(fetchImpl).toHaveBeenCalledOnce(); const init = (fetchImpl as unknown as ReturnType).mock .calls[0]![1] as RequestInit; const headers = init.headers as Record; - expect(headers.Signature).toContain('keyId="'); - expect(headers.Digest).toMatch(/^SHA-256=/); + expect(headers["Signature-Input"]).toContain( + 'keyid="https://example.com/users/bob#main-key"', + ); + expect(headers.Signature).toMatch(/^sig1=:/); + expect(headers["Content-Digest"]).toMatch(/^sha-256=:/); expect(headers.Date).toBeTruthy(); }); @@ -131,7 +138,7 @@ describe("deliverActivity", () => { .calls[0]![1] as RequestInit; const headers = init.headers as Record; expect(headers["User-Agent"]).toBe(TEST_UA); - expect(headers.Signature).toContain('keyId="'); + expect(headers["Signature-Input"]).toContain('keyid="'); }); it("releases its timeout after a successful delivery", async () => { @@ -161,8 +168,86 @@ describe("deliverActivity", () => { } }); + it("double-knocks with draft-cavage when RFC 9421 is rejected", async () => { + const pem = await privateKeyPem(); + const fetchImpl = vi.fn(async (_url: string, init?: RequestInit) => { + const headers = init?.headers as Record; + return new Response(null, { + status: headers["Signature-Input"] ? 401 : 202, + }); + }) as unknown as typeof fetch; + + const result = await deliverActivity( + "https://remote.example/inbox", + JSON.stringify({ type: "Accept" }), + { keyId: "https://example.com/users/bob#main-key", privateKeyPem: pem }, + fetchImpl, + TEST_UA, + ); + + expect(result).toEqual({ + ok: true, + status: 202, + profile: "cavage", + }); + const calls = (fetchImpl as unknown as ReturnType).mock + .calls as [string, RequestInit][]; + expect(calls).toHaveLength(2); + const second = calls[1]![1].headers as Record; + expect(second["Signature-Input"]).toBeUndefined(); + expect(second.Signature).toContain( + 'keyId="https://example.com/users/bob#main-key"', + ); + expect(second.Digest).toMatch(/^SHA-256=/); + }); + + it("skips the RFC 9421 knock for a known draft-cavage peer", async () => { + const pem = await privateKeyPem(); + const fetchImpl = vi.fn( + async () => new Response(null, { status: 202 }), + ) as unknown as typeof fetch; + + const result = await deliverActivity( + "https://remote.example/inbox", + "{}", + { keyId: "k", privateKeyPem: pem }, + fetchImpl, + TEST_UA, + undefined, + undefined, + "cavage", + ); + + expect(result).toEqual({ ok: true, status: 202, profile: "cavage" }); + expect(fetchImpl).toHaveBeenCalledOnce(); + const init = (fetchImpl as unknown as ReturnType).mock + .calls[0]![1] as RequestInit; + const headers = init.headers as Record; + expect(headers["Signature-Input"]).toBeUndefined(); + expect(headers.Signature).toContain('keyId="k"'); + }); + + it.each([404, 410, 422, 500])( + "does not knock again on a non-signature %i", + async (status) => { + const pem = await privateKeyPem(); + const fetchImpl = vi.fn( + async () => new Response(null, { status }), + ) as unknown as typeof fetch; + await deliverActivity( + "https://remote.example/inbox", + "{}", + { keyId: "k", privateKeyPem: pem }, + fetchImpl, + TEST_UA, + ); + expect(fetchImpl).toHaveBeenCalledOnce(); + }, + ); + it("classifies a 4xx as a permanent (non-retryable) failure", async () => { const pem = await privateKeyPem(); + // Both knocks rejected: the draft-cavage answer is the one reported. const fetchImpl = (async () => new Response("bad", { status: 400 })) as unknown as typeof fetch; const result = await deliverActivity( diff --git a/packages/activitypub/src/delivery.ts b/packages/activitypub/src/delivery.ts index 74408c48..e70f324f 100644 --- a/packages/activitypub/src/delivery.ts +++ b/packages/activitypub/src/delivery.ts @@ -8,13 +8,34 @@ * malicious actor cannot point a delivery at the Worker's own network. (DNS * rebinding is out of scope: the Workers runtime does not expose name * resolution to user code; see `@dwk/webmention`'s safe-fetch for the same - * limitation.) The forthcoming `@dwk/http-signatures` package owns the signing - * primitive; this module wires it to the network. + * limitation.) `signature.ts` owns the signing primitives; this module wires + * them to the network. + * + * Signing "double-knocks": a peer with no known preference is sent an RFC 9421 + * signature first, and if it rejects that with a signature-shaped status + * (`400`/`401`/`403`) the same body is re-signed with draft-cavage and sent + * once more. The result reports which profile the peer accepted so the caller + * can remember draft-cavage peers and skip the first knock next time. */ import { assertPublicUrl, safeFetch, SsrfError } from "@dwk/safe-fetch"; -import { signRequest, type SignerKey } from "./signature.js"; +import type { SignatureProfile } from "@dwk/http-signatures"; + +import { + signRequest, + signRequestRfc9421, + type SignedRequest, + type SignerKey, +} from "./signature.js"; + +/** + * Statuses that may mean "I could not verify that signature", and so warrant a + * second knock with draft-cavage. Mastodon answers `401`; others use `400` or + * `403`. A `404`/`410`/`422`/`5xx` is not about the signature, so it never + * triggers the fallback. + */ +const KNOCK_AGAIN_STATUSES: ReadonlySet = new Set([400, 401, 403]); /** Machine-readable cause of a blocked delivery target. */ export type BlockedReason = @@ -62,7 +83,12 @@ function reasonFor(error: SsrfError): BlockedReason { /** The outcome of a single delivery attempt. */ export type DeliveryResult = - | { readonly ok: true; readonly status: number } + | { + readonly ok: true; + readonly status: number; + /** The signature profile the peer accepted. */ + readonly profile: SignatureProfile; + } | { readonly ok: false; readonly status: number; @@ -73,6 +99,10 @@ export type DeliveryResult = * Sign and `POST` one activity to a remote inbox. A `2xx` is success; `4xx` * (except `408`/`429`) is a permanent failure (drop) since the peer rejected the * request; `5xx`, `408`, `429`, and network errors are retryable. + * + * `preference` is the profile the peer is known to want: `"cavage"` skips the + * RFC 9421 knock; anything else (including unknown) double-knocks — RFC 9421 + * first, then draft-cavage if the first answer is in {@link KNOCK_AGAIN_STATUSES}. */ export async function deliverActivity( inboxUrl: string, @@ -82,14 +112,56 @@ export async function deliverActivity( userAgent: string, now: () => number = () => Date.now(), timeoutMs = 10_000, + preference?: SignatureProfile, ): Promise { // Throws DeliveryBlockedError for an unsafe target — the caller drops the row. assertPublicHttpsTarget(inboxUrl); const body = new TextEncoder().encode(activityJson); - const signed = await signRequest(inboxUrl, body, signer, { now }); + const post = (signed: SignedRequest): Promise => + postSigned(fetchImpl, inboxUrl, signed, userAgent, timeoutMs); - let response: Response; + let profile: SignatureProfile = + preference === "cavage" ? "cavage" : "rfc9421"; + let response = + profile === "cavage" + ? await post(await signRequest(inboxUrl, body, signer, { now })) + : await post(await signRequestRfc9421(inboxUrl, body, signer, { now })); + if ( + profile === "rfc9421" && + response !== null && + KNOCK_AGAIN_STATUSES.has(response.status) + ) { + // The first knock's body is never read; release it before knocking again. + await response.body?.cancel(); + profile = "cavage"; + response = await post(await signRequest(inboxUrl, body, signer, { now })); + } + // Network error or timeout: worth retrying. + if (response === null) return { ok: false, status: 0, retryable: true }; + + if (response.status >= 200 && response.status < 300) { + return { ok: true, status: response.status, profile }; + } + const retryable = + response.status >= 500 || + response.status === 408 || + response.status === 429; + return { ok: false, status: response.status, retryable }; +} + +/** + * `POST` one signed delivery. Returns the response, or `null` on a network + * error or timeout; throws {@link DeliveryBlockedError} if a redirect hop lands + * on a blocked target. + */ +async function postSigned( + fetchImpl: typeof fetch, + inboxUrl: string, + signed: SignedRequest, + userAgent: string, + timeoutMs: number, +): Promise { try { // Route the POST itself through the SSRF-safe wrapper too, not just the // pre-flight check on `inboxUrl`: a hostile inbox can 3xx-redirect a @@ -102,30 +174,20 @@ export async function deliverActivity( { method: "POST", // `User-Agent` is layered on top of (not part of) the signed header - // set `signRequest` returns — it isn't covered by the signature, so - // adding it here doesn't touch signing at all. + // set — it isn't covered by the signature, so adding it here doesn't + // touch signing at all. headers: { ...signed.headers, "User-Agent": userAgent }, body: signed.body as BufferSource, }, { allowedSchemes: ["https:"], timeoutMs }, ); - response = result.response; + return result.response; } catch (error) { if (error instanceof SsrfError) { // A redirect hop landed on a blocked target — this inbox can never be // delivered to safely, same as failing the pre-flight check. throw new DeliveryBlockedError(reasonFor(error)); } - // Network error or timeout: worth retrying. - return { ok: false, status: 0, retryable: true }; + return null; } - - if (response.status >= 200 && response.status < 300) { - return { ok: true, status: response.status }; - } - const retryable = - response.status >= 500 || - response.status === 408 || - response.status === 429; - return { ok: false, status: response.status, retryable }; } diff --git a/packages/activitypub/src/index.test.ts b/packages/activitypub/src/index.test.ts index 891016ac..9a1d5501 100644 --- a/packages/activitypub/src/index.test.ts +++ b/packages/activitypub/src/index.test.ts @@ -1263,8 +1263,9 @@ describe("Accept-on-Follow delivery", () => { expect(delivery).toBeDefined(); expect(delivery?.init?.method).toBe("POST"); const headers = delivery?.init?.headers as Record; - expect(headers.Signature).toContain(`keyId="${iris.keyId}"`); - expect(headers.Digest).toMatch(/^SHA-256=/); + // First knock is RFC 9421 (see delivery.ts's double-knock). + expect(headers["Signature-Input"]).toContain(`keyid="${iris.keyId}"`); + expect(headers["Content-Digest"]).toMatch(/^sha-256=:/); const raw = delivery?.init?.body as ArrayBufferView; const body = JSON.parse(new TextDecoder().decode(raw)); expect(body.type).toBe("Accept"); @@ -1330,6 +1331,87 @@ describe("Accept-on-Follow delivery", () => { } }); }); + + it("remembers a draft-cavage peer after a double-knock and skips RFC 9421 next time", async () => { + const username = `bob-${crypto.randomUUID().slice(0, 8)}`; + const iris = deriveIris(BASE, username); + const stub = testEnv.ACTOR.get(testEnv.ACTOR.idFromName(iris.id)); + + await runInDurableObject(stub, async (instance, state) => { + const original = globalThis.fetch; + const inboxCalls: Record[] = []; + globalThis.fetch = (async (url: string | URL, init?: RequestInit) => { + const href = typeof url === "string" ? url : url.toString(); + if (href === REMOTE) { + return new Response( + JSON.stringify({ + id: REMOTE, + inbox: `${REMOTE}/inbox`, + publicKey: { publicKeyPem, owner: REMOTE }, + }), + { headers: { "content-type": "application/activity+json" } }, + ); + } + const headers = init?.headers as Record; + inboxCalls.push(headers); + // A draft-cavage-only peer: rejects RFC 9421 the way Mastodon does. + return new Response(null, { + status: headers["Signature-Input"] ? 401 : 202, + }); + }) as unknown as typeof fetch; + + const deliver = () => + instance.fetch( + new Request(`${iris.id}/__deliver`, { + headers: { [INTERNAL_HEADERS.config]: forwardedHeader(username) }, + }), + ); + + try { + await instance.fetch( + new Request(iris.inbox, { + method: "POST", + headers: { + "content-type": "application/activity+json", + [INTERNAL_HEADERS.config]: forwardedHeader(username), + }, + body: JSON.stringify({ + id: "https://remote.example/activities/77", + type: "Follow", + actor: REMOTE, + object: iris.id, + }), + }), + ); + await deliver(); + + expect(inboxCalls.map((h) => Boolean(h["Signature-Input"]))).toEqual([ + true, + false, + ]); + const pref = state.storage.sql + .exec<{ host: string; profile: string }>( + "SELECT host, profile FROM sig_pref", + ) + .toArray(); + expect(pref).toEqual([{ host: "remote.example", profile: "cavage" }]); + + inboxCalls.length = 0; + state.storage.sql.exec( + "INSERT INTO delivery (inbox, json, attempts, next_at) VALUES (?, ?, 0, 0)", + `${REMOTE}/inbox`, + JSON.stringify({ type: "Update" }), + ); + await deliver(); + + expect(inboxCalls).toHaveLength(1); + expect(inboxCalls[0]!["Signature-Input"]).toBeUndefined(); + expect(inboxCalls[0]!.Signature).toContain(`keyId="${iris.keyId}"`); + } finally { + globalThis.fetch = original; + } + }); + }); }); // --------------------------------------------------------------------------- diff --git a/packages/activitypub/src/index.ts b/packages/activitypub/src/index.ts index 06b8646c..d91c3902 100644 --- a/packages/activitypub/src/index.ts +++ b/packages/activitypub/src/index.ts @@ -114,6 +114,7 @@ export { export { signRequest, + signRequestRfc9421, verifyInboxSignature, digestHeader, importPublicKey, diff --git a/packages/activitypub/src/object.ts b/packages/activitypub/src/object.ts index 1b490df7..6faf8d33 100644 --- a/packages/activitypub/src/object.ts +++ b/packages/activitypub/src/object.ts @@ -26,6 +26,7 @@ import { type ActorIris, type JsonValue, } from "./as2.js"; +import type { SignatureProfile } from "@dwk/http-signatures"; import { hostFromUrl } from "@dwk/log"; import { readBodyCapped, safeFetch } from "@dwk/safe-fetch"; @@ -59,6 +60,12 @@ import type { ActivityPubEnv } from "./config.js"; /** How long a seen activity `id` is remembered for dedup (7 days). */ const SEEN_TTL_MS = 7 * 24 * 60 * 60 * 1000; +/** + * How long a peer's draft-cavage signature preference (learned when it rejected + * an RFC 9421 delivery) is remembered before double-knocking it again (7 days), + * so a peer that later adds RFC 9421 support is eventually rediscovered. + */ +const SIG_PREF_TTL_MS = 7 * 24 * 60 * 60 * 1000; /** Max delivery rows processed per alarm wake. */ const DELIVERY_BATCH = 20; /** Timeout (ms) bounding any single outbound fetch (actor lookup / delivery). */ @@ -331,6 +338,14 @@ export class ActivityPubObject extends DurableObject { event TEXT NOT NULL, fields TEXT NOT NULL, n INTEGER NOT NULL, PRIMARY KEY (event, fields))`, ); + // Delivery hosts that rejected an RFC 9421 signature but accepted + // draft-cavage (see `deliverActivity`'s double-knock), so later deliveries + // skip the doomed first knock. Rows older than SIG_PREF_TTL_MS are ignored + // and pruned on the next delivery pass. + this.#sql.exec( + `CREATE TABLE IF NOT EXISTS sig_pref ( + host TEXT PRIMARY KEY, profile TEXT NOT NULL, at INTEGER NOT NULL)`, + ); // Additive-column migrations for objects created before fediverse interop // phases 1–2 (#274/#275); fresh objects already get these from the CREATE // TABLEs above. Nullable by design: `object_type`/`audience` classify @@ -3241,6 +3256,22 @@ export class ActivityPubObject extends DurableObject { return drained; } + /** + * The signature profile `host` is known to want for deliveries, or + * `undefined` to double-knock (RFC 9421 first). Only draft-cavage peers are + * recorded; an expired row reads as unknown. + */ + #signaturePreference(host: string): SignatureProfile | undefined { + const row = this.#sql + .exec<{ profile: string }>( + `SELECT profile FROM sig_pref WHERE host = ? AND at > ?`, + host, + Date.now() - SIG_PREF_TTL_MS, + ) + .toArray()[0]; + return row?.profile === "cavage" ? "cavage" : undefined; + } + /** * Process every due delivery row once: sign and `POST` it, deleting on * success or permanent failure and rescheduling with exponential backoff on a @@ -3264,6 +3295,12 @@ export class ActivityPubObject extends DurableObject { DELIVERY_BATCH, ) .toArray(); + if (due.length > 0) { + this.#sql.exec( + `DELETE FROM sig_pref WHERE at <= ?`, + now - SIG_PREF_TTL_MS, + ); + } for (const row of due) { if (!signer) { @@ -3272,6 +3309,7 @@ export class ActivityPubObject extends DurableObject { continue; } const targetHost = hostFromUrl(row.inbox) ?? "unknown"; + const preference = this.#signaturePreference(targetHost); try { const result = await deliverActivity( row.inbox, @@ -3280,9 +3318,19 @@ export class ActivityPubObject extends DurableObject { fetch, userAgent, () => Date.now(), + OUTBOUND_TIMEOUT_MS, + preference, ); if (result.ok) { this.#sql.exec(`DELETE FROM delivery WHERE seq = ?`, row.seq); + if (result.profile === "cavage" && preference !== "cavage") { + this.#sql.exec( + `INSERT INTO sig_pref (host, profile, at) VALUES (?, 'cavage', ?) + ON CONFLICT (host) DO UPDATE SET profile = excluded.profile, at = excluded.at`, + targetHost, + Date.now(), + ); + } this.#logDelivery(ActivityPubLogEvent.DeliverySucceeded, { targetHost, status: result.status, diff --git a/packages/activitypub/src/signature.test.ts b/packages/activitypub/src/signature.test.ts index 7ee16bb7..2c47f0a5 100644 --- a/packages/activitypub/src/signature.test.ts +++ b/packages/activitypub/src/signature.test.ts @@ -8,6 +8,7 @@ import { parseSignatureHeader, signGetRequest, signRequest, + signRequestRfc9421, verifyInboxSignature, type InboxRequest, type KeyResolver, @@ -319,6 +320,37 @@ describe("verifyInboxSignature (RFC 9421)", () => { }); }); +describe("signRequestRfc9421", () => { + const now = Date.UTC(2026, 5, 4, 12, 0, 0); + const url = "https://example.com/users/bob/inbox"; + const body = JSON.stringify({ type: "Accept", id: "https://remote/2" }); + + it("produces a delivery our own RFC 9421 inbox verifier accepts", async () => { + const bytes = new TextEncoder().encode(body); + const signed = await signRequestRfc9421( + url, + bytes, + { keyId: KEY_ID, privateKeyPem }, + { now: () => now }, + ); + expect(signed.headers["Signature-Input"]).toMatch( + /^sig1=\("@method" "@target-uri" "content-digest" "content-type"\);created=\d+;keyid="[^"]+";alg="rsa-v1_5-sha256"$/, + ); + const result = await verifyInboxSignature( + { + method: "POST", + path: "/users/bob/inbox", + url, + headers: new Headers(signed.headers), + body: bytes, + }, + resolver(publicKeyPem), + { now: () => now }, + ); + expect(result).toEqual({ ok: true, keyId: KEY_ID, actor: ACTOR }); + }); +}); + describe("digestHeader", () => { it("matches a known SHA-256 of the empty body", async () => { expect(await digestHeader(new Uint8Array(0))).toBe( diff --git a/packages/activitypub/src/signature.ts b/packages/activitypub/src/signature.ts index 792ee420..e19789c5 100644 --- a/packages/activitypub/src/signature.ts +++ b/packages/activitypub/src/signature.ts @@ -12,15 +12,19 @@ * cross-standard `@dwk/http-signatures` package (issue #59) rather than * reimplemented here — see {@link verifyInboxSignature}. * - * Outbound signing stays on the draft-cavage profile: every fediverse peer - * this package interoperates with today accepts it, so there is no correctness - * reason to change it, and it keeps `signRequest` self-contained. + * Outbound `POST` deliveries can be signed with either profile — + * {@link signRequest} (draft-cavage) or {@link signRequestRfc9421} — and + * `delivery.ts` "double-knocks": RFC 9421 first, falling back to draft-cavage + * when a peer rejects it. Outbound `GET`s ({@link signGetRequest}) stay on + * draft-cavage, which every authorized-fetch peer accepts. * * Inputs are plain data — method, URL, headers, body bytes, and a resolved key — * so the crypto unit-tests without any ActivityPub assumptions. */ import { + createContentDigest, + signMessage, verifyMessage, type HttpMessage, type KeyResolver as HttpSigKeyResolver, @@ -501,6 +505,45 @@ export async function signRequest( }; } +/** + * Sign an outbound `POST` (a delivery to a remote inbox) with the RFC 9421 + * profile via `@dwk/http-signatures`. Covers `@method`, `@target-uri`, + * `content-digest`, and `content-type` with a `created` parameter — a superset + * of what {@link verifyInboxSignature} requires of an RFC 9421 peer — and + * returns `Date`, `Content-Digest`, `Content-Type`, `Signature-Input`, and + * `Signature` to send. `Date` is not covered (`created` is the timestamp + * anchor) but is sent for peers that log or sanity-check it. + */ +export async function signRequestRfc9421( + url: string, + body: Uint8Array, + signer: SignerKey, + options: { readonly now?: () => number; readonly contentType?: string } = {}, +): Promise { + const now = options.now ?? (() => Date.now()); + const contentType = options.contentType ?? "application/activity+json"; + const nowMs = now(); + const contentDigest = await createContentDigest(body); + const headers: Record = { + Date: new Date(nowMs).toUTCString(), + "Content-Digest": contentDigest, + "Content-Type": contentType, + }; + const key = await importPrivateKey(signer.privateKeyPem); + const signature = await signMessage( + { method: "POST", url, headers }, + { + profile: "rfc9421", + key, + keyId: signer.keyId, + alg: "rsa-v1_5-sha256", + components: ["@method", "@target-uri", "content-digest", "content-type"], + created: Math.floor(nowMs / 1000), + }, + ); + return { headers: { ...headers, ...signature }, body }; +} + /** * Sign an outbound `GET` (fetching a remote actor, key, or object) with the * draft-cavage profile, covering `(request-target) host date`. Servers running diff --git a/packages/server/src/central-do-activitypub.integration.test.ts b/packages/server/src/central-do-activitypub.integration.test.ts index 5db4c9e9..1ea3305d 100644 --- a/packages/server/src/central-do-activitypub.integration.test.ts +++ b/packages/server/src/central-do-activitypub.integration.test.ts @@ -240,7 +240,9 @@ describe("central mode — activitypub's actor Durable Object across two replica const delivery = seen.find((s) => s.url === `${REMOTE}/inbox`); expect(delivery?.init?.method).toBe("POST"); const headers = delivery?.init?.headers as Record; - expect(headers.Signature).toContain(`keyId="${actorUrl}#main-key"`); + expect(headers["Signature-Input"]).toContain( + `keyid="${actorUrl}#main-key"`, + ); const raw = delivery?.init?.body as ArrayBufferView; const body = JSON.parse(new TextDecoder().decode(raw)) as { type: string; diff --git a/packages/server/src/phase5-activitypub.integration.test.ts b/packages/server/src/phase5-activitypub.integration.test.ts index 7718e690..496e0f90 100644 --- a/packages/server/src/phase5-activitypub.integration.test.ts +++ b/packages/server/src/phase5-activitypub.integration.test.ts @@ -219,7 +219,9 @@ describe("Phase 5 — activitypub on the emulated Durable Object", () => { const delivery = seen.find((s) => s.url === `${REMOTE}/inbox`); expect(delivery?.init?.method).toBe("POST"); const headers = delivery?.init?.headers as Record; - expect(headers.Signature).toContain(`keyId="${actor.actorUrl}#main-key"`); + expect(headers["Signature-Input"]).toContain( + `keyid="${actor.actorUrl}#main-key"`, + ); const raw = delivery?.init?.body as ArrayBufferView; const body = JSON.parse(new TextDecoder().decode(raw)) as { type: string; diff --git a/spec/packages/activitypub.md b/spec/packages/activitypub.md index c53b5e92..685df125 100644 --- a/spec/packages/activitypub.md +++ b/spec/packages/activitypub.md @@ -108,6 +108,15 @@ large media; bodies **MUST** stream and **MUST NOT** be buffered in the DO. outbound fetch (key resolution, delivery, discovery) carries a configurable `userAgent` (default: a string identifying this package), since some peers' anti-scraper challenge layers treat an unidentified client as a bot. +- Outbound delivery signatures **double-knock**: sign with RFC 9421 + (`@method @target-uri content-digest content-type`, `created`, + `rsa-v1_5-sha256`) first; if the peer answers `400`/`401`/`403`, re-sign the + same body with draft-cavage (`(request-target) host date digest + content-type`) and send once more. No other status triggers a second knock. + A host that rejected RFC 9421 but accepted draft-cavage is recorded in the + actor DO's SQLite (`sig_pref`) for 7 days; deliveries to it skip the RFC 9421 + knock until the row expires, so a peer that later adds RFC 9421 support is + rediscovered. Hosts are not recorded for any other outcome. - Outbound `GET`s (key resolution, inbox resolution, actor-profile refresh, relayed-object verification, the `activitypub_resolve` actor fetch) are HTTP-signed as the actor (draft-cavage, `(request-target) host date`) From d4cc978401ad93c34db2e9dc2450464be8238aee Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 27 Oct 2026 23:24:50 +0000 Subject: [PATCH 2/3] refactor(activitypub): address double-knock review feedback Move deliverActivity's optional trailing arguments into a DeliveryOptions object, document the accepted 403 double-traffic trade-off, regroup the type import, and test the sig_pref TTL expiry and prune path. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_015LRGjmHE99WESN5HBERxSj --- .changeset/activitypub-double-knock.md | 2 +- packages/activitypub/src/delivery.test.ts | 7 ++--- packages/activitypub/src/delivery.ts | 32 ++++++++++++++++++----- packages/activitypub/src/index.test.ts | 28 ++++++++++++++++++++ packages/activitypub/src/index.ts | 1 + packages/activitypub/src/object.ts | 6 ++--- 6 files changed, 60 insertions(+), 16 deletions(-) diff --git a/.changeset/activitypub-double-knock.md b/.changeset/activitypub-double-knock.md index 89df43f8..67b1a444 100644 --- a/.changeset/activitypub-double-knock.md +++ b/.changeset/activitypub-double-knock.md @@ -2,4 +2,4 @@ "@dwk/activitypub": minor --- -Sign outbound deliveries with RFC 9421 HTTP Message Signatures first, falling back to draft-cavage when a peer rejects them ("double-knock"). Peers that only accept draft-cavage are remembered per actor for 7 days. Adds `signRequestRfc9421`; `DeliveryResult` now reports the accepted `profile` on success. +Sign outbound deliveries with RFC 9421 HTTP Message Signatures first, falling back to draft-cavage when a peer rejects them ("double-knock"). Peers that only accept draft-cavage are remembered per actor for 7 days. Adds `signRequestRfc9421`; `deliverActivity`'s trailing `now`/`timeoutMs` arguments move into a `DeliveryOptions` object (with the new `preference`), and `DeliveryResult` now reports the accepted `profile` on success. diff --git a/packages/activitypub/src/delivery.test.ts b/packages/activitypub/src/delivery.test.ts index 280d582a..b29e9be3 100644 --- a/packages/activitypub/src/delivery.test.ts +++ b/packages/activitypub/src/delivery.test.ts @@ -157,8 +157,7 @@ describe("deliverActivity", () => { { keyId: "https://example.com/users/bob#main-key", privateKeyPem: pem }, fetchImpl, TEST_UA, - undefined, - 10, + { timeoutMs: 10 }, ); await vi.advanceTimersByTimeAsync(10); @@ -213,9 +212,7 @@ describe("deliverActivity", () => { { keyId: "k", privateKeyPem: pem }, fetchImpl, TEST_UA, - undefined, - undefined, - "cavage", + { preference: "cavage" }, ); expect(result).toEqual({ ok: true, status: 202, profile: "cavage" }); diff --git a/packages/activitypub/src/delivery.ts b/packages/activitypub/src/delivery.ts index e70f324f..252c78a7 100644 --- a/packages/activitypub/src/delivery.ts +++ b/packages/activitypub/src/delivery.ts @@ -34,6 +34,12 @@ import { * second knock with draft-cavage. Mastodon answers `401`; others use `400` or * `403`. A `404`/`410`/`422`/`5xx` is not about the signature, so it never * triggers the fallback. + * + * `403` is also what a peer that blocks our domain or actor answers, so each + * delivery to such a peer is sent twice (both knocks fail, and the row is + * dropped as permanent). That doubled traffic is accepted — the same trade-off + * Fedify makes — rather than dropping `403`, which would leave peers that use + * it for signature failures unreachable. */ const KNOCK_AGAIN_STATUSES: ReadonlySet = new Set([400, 401, 403]); @@ -95,14 +101,27 @@ export type DeliveryResult = readonly retryable: boolean; }; +/** Optional knobs for {@link deliverActivity}. */ +export interface DeliveryOptions { + /** Clock, epoch milliseconds (deterministic tests). Defaults to `Date.now`. */ + readonly now?: () => number; + /** Timeout for each `POST`, in milliseconds. Defaults to 10 s. */ + readonly timeoutMs?: number; + /** + * The signature profile the peer is known to want: `"cavage"` skips the + * RFC 9421 knock; anything else (including unknown) double-knocks — RFC 9421 + * first, then draft-cavage if the first answer is in + * {@link KNOCK_AGAIN_STATUSES}. + */ + readonly preference?: SignatureProfile; +} + /** * Sign and `POST` one activity to a remote inbox. A `2xx` is success; `4xx` * (except `408`/`429`) is a permanent failure (drop) since the peer rejected the * request; `5xx`, `408`, `429`, and network errors are retryable. * - * `preference` is the profile the peer is known to want: `"cavage"` skips the - * RFC 9421 knock; anything else (including unknown) double-knocks — RFC 9421 - * first, then draft-cavage if the first answer is in {@link KNOCK_AGAIN_STATUSES}. + * {@link DeliveryOptions.preference} selects the signing strategy — see there. */ export async function deliverActivity( inboxUrl: string, @@ -110,10 +129,11 @@ export async function deliverActivity( signer: SignerKey, fetchImpl: typeof fetch, userAgent: string, - now: () => number = () => Date.now(), - timeoutMs = 10_000, - preference?: SignatureProfile, + options: DeliveryOptions = {}, ): Promise { + const now = options.now ?? (() => Date.now()); + const timeoutMs = options.timeoutMs ?? 10_000; + const { preference } = options; // Throws DeliveryBlockedError for an unsafe target — the caller drops the row. assertPublicHttpsTarget(inboxUrl); diff --git a/packages/activitypub/src/index.test.ts b/packages/activitypub/src/index.test.ts index 9a1d5501..c9294160 100644 --- a/packages/activitypub/src/index.test.ts +++ b/packages/activitypub/src/index.test.ts @@ -1407,6 +1407,34 @@ describe("Accept-on-Follow delivery", () => { expect(inboxCalls).toHaveLength(1); expect(inboxCalls[0]!["Signature-Input"]).toBeUndefined(); expect(inboxCalls[0]!.Signature).toContain(`keyId="${iris.keyId}"`); + + // Once the row is older than the 7-day TTL it is ignored (the peer is + // double-knocked again) and pruned on that delivery pass. + inboxCalls.length = 0; + state.storage.sql.exec( + "UPDATE sig_pref SET at = ?", + Date.now() - 7 * 24 * 60 * 60 * 1000 - 1, + ); + state.storage.sql.exec( + "INSERT INTO delivery (inbox, json, attempts, next_at) VALUES (?, ?, 0, 0)", + `${REMOTE}/inbox`, + JSON.stringify({ type: "Update" }), + ); + globalThis.fetch = (async (_url: string | URL, init?: RequestInit) => { + const headers = init?.headers as Record; + inboxCalls.push(headers); + // The peer now accepts RFC 9421, so the first knock lands. + return new Response(null, { status: 202 }); + }) as unknown as typeof fetch; + await deliver(); + + expect(inboxCalls).toHaveLength(1); + expect(inboxCalls[0]!["Signature-Input"]).toContain( + `keyid="${iris.keyId}"`, + ); + expect( + state.storage.sql.exec("SELECT host FROM sig_pref").toArray(), + ).toEqual([]); } finally { globalThis.fetch = original; } diff --git a/packages/activitypub/src/index.ts b/packages/activitypub/src/index.ts index d91c3902..3ad6872f 100644 --- a/packages/activitypub/src/index.ts +++ b/packages/activitypub/src/index.ts @@ -131,6 +131,7 @@ export { deliverActivity, assertPublicHttpsTarget, DeliveryBlockedError, + type DeliveryOptions, type DeliveryResult, } from "./delivery.js"; diff --git a/packages/activitypub/src/object.ts b/packages/activitypub/src/object.ts index 6faf8d33..033578ce 100644 --- a/packages/activitypub/src/object.ts +++ b/packages/activitypub/src/object.ts @@ -26,9 +26,9 @@ import { type ActorIris, type JsonValue, } from "./as2.js"; -import type { SignatureProfile } from "@dwk/http-signatures"; import { hostFromUrl } from "@dwk/log"; import { readBodyCapped, safeFetch } from "@dwk/safe-fetch"; +import type { SignatureProfile } from "@dwk/http-signatures"; import { ApOutcome, @@ -3317,9 +3317,7 @@ export class ActivityPubObject extends DurableObject { signer, fetch, userAgent, - () => Date.now(), - OUTBOUND_TIMEOUT_MS, - preference, + { timeoutMs: OUTBOUND_TIMEOUT_MS, preference }, ); if (result.ok) { this.#sql.exec(`DELETE FROM delivery WHERE seq = ?`, row.seq); From 2ec181b0c425d22549261c0abebd0a3376e417ef Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 28 Oct 2026 00:14:52 +0000 Subject: [PATCH 3/3] feat(activitypub): per-host delivery circuit breaker Open a host's delivery circuit after 5 consecutive retryable failures and defer its rows (no retry attempt spent) for a cooldown that starts at 10 minutes and doubles per failed half-open probe, capped at 24 h. A success, a non-retryable answer, or a signed inbound activity from the host closes it; after 7 days of failure, rows reaching an open circuit are dropped. State lives in the actor DO's SQLite. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_015LRGjmHE99WESN5HBERxSj --- .../activitypub-delivery-circuit-breaker.md | 5 + packages/activitypub/README.md | 12 ++ packages/activitypub/src/index.test.ts | 174 ++++++++++++++++++ packages/activitypub/src/log.ts | 13 ++ packages/activitypub/src/object.ts | 136 +++++++++++++- spec/packages/activitypub.md | 14 ++ 6 files changed, 353 insertions(+), 1 deletion(-) create mode 100644 .changeset/activitypub-delivery-circuit-breaker.md diff --git a/.changeset/activitypub-delivery-circuit-breaker.md b/.changeset/activitypub-delivery-circuit-breaker.md new file mode 100644 index 00000000..d3ee505e --- /dev/null +++ b/.changeset/activitypub-delivery-circuit-breaker.md @@ -0,0 +1,5 @@ +--- +"@dwk/activitypub": minor +--- + +Add a per-host delivery circuit breaker. After 5 consecutive retryable failures to a host, its deliveries are deferred, without spending retry attempts, for a cooldown that starts at 10 minutes and doubles on each failed probe up to 24 hours. A success or a signed inbound activity from the host closes the circuit, and a host failing for 7 days has its deliveries dropped while the circuit is open. New log/metric events: `activitypub.delivery.deferred`, `activitypub.delivery.circuit_opened`, `activitypub.delivery.circuit_closed`. diff --git a/packages/activitypub/README.md b/packages/activitypub/README.md index a3c4b031..6dd558aa 100644 --- a/packages/activitypub/README.md +++ b/packages/activitypub/README.md @@ -94,6 +94,18 @@ No KV is used for any authoritative state, per [`spec/non-functional-requirements.md`](../../spec/non-functional-requirements.md): follower lists, dedup, and the delivery queue all live in the DO's SQLite. +## Unreachable peers + +Each actor's Durable Object keeps a per-host delivery circuit breaker. After 5 +consecutive retryable failures to a host, its queued deliveries are deferred +(not attempted, and not charged a retry) for a cooldown that starts at 10 +minutes and doubles on each failed probe, up to 24 hours. A successful +delivery, or any signed activity received from that host, closes the circuit. +A host that has been failing for 7 days has its deliveries dropped while the +circuit is open, the same window Mastodon uses before it marks a domain +unavailable. This keeps one dead peer from spending every alarm wake on +10-second timeouts. + ## HTTP signatures Inbound `POST /inbox` deliveries are verified against either the de-facto diff --git a/packages/activitypub/src/index.test.ts b/packages/activitypub/src/index.test.ts index c9294160..471f7609 100644 --- a/packages/activitypub/src/index.test.ts +++ b/packages/activitypub/src/index.test.ts @@ -1743,3 +1743,177 @@ describe("Event RSVP (Join/Leave)", () => { }); }); }); + +// --------------------------------------------------------------------------- +// Delivery circuit breaker: a host that keeps failing stops being attempted, +// is probed again after a cooldown, and closes on success or inbound traffic. +// --------------------------------------------------------------------------- + +describe("delivery circuit breaker", () => { + const DEAD_INBOX = "https://dead.example/inbox"; + + /** A fresh actor DO with `rows` deliveries queued to {@link DEAD_INBOX}. */ + function withQueuedDeliveries( + rows: number, + body: (ctx: { + instance: { fetch(request: Request): Promise }; + sql: SqlStorage; + calls: string[]; + respond: (status: number) => void; + deliver: () => Promise; + username: string; + iris: ReturnType; + }) => Promise, + ): Promise { + const username = `bob-${crypto.randomUUID().slice(0, 8)}`; + const iris = deriveIris(BASE, username); + const stub = testEnv.ACTOR.get(testEnv.ACTOR.idFromName(iris.id)); + return runInDurableObject(stub, async (instance, state) => { + const original = globalThis.fetch; + const calls: string[] = []; + let status = 503; + globalThis.fetch = (async (url: string | URL) => { + calls.push(typeof url === "string" ? url : url.toString()); + return new Response(null, { status }); + }) as unknown as typeof fetch; + const deliver = () => + instance.fetch( + new Request(`${iris.id}/__deliver`, { + headers: { [INTERNAL_HEADERS.config]: forwardedHeader(username) }, + }), + ); + try { + // A first request lets the DO pick up its forwarded config + schema. + await deliver(); + for (let i = 0; i < rows; i++) { + state.storage.sql.exec( + "INSERT INTO delivery (inbox, json, attempts, next_at) VALUES (?, ?, 0, 0)", + DEAD_INBOX, + JSON.stringify({ type: "Note", n: i }), + ); + } + await body({ + instance, + sql: state.storage.sql, + calls, + respond: (s) => { + status = s; + }, + deliver, + username, + iris, + }); + } finally { + globalThis.fetch = original; + } + }); + } + + const circuit = (sql: SqlStorage) => + sql + .exec<{ + failures: number; + open_until: number; + trips: number; + }>("SELECT failures, open_until, trips FROM peer_circuit") + .toArray(); + + it("opens after 5 consecutive failures and defers the rest without spending attempts", async () => { + await withQueuedDeliveries(7, async ({ sql, calls, deliver }) => { + const before = Date.now(); + await deliver(); + + expect(calls).toHaveLength(5); + const [row] = circuit(sql); + expect(row?.failures).toBe(5); + expect(row?.trips).toBe(1); + expect(row!.open_until).toBeGreaterThanOrEqual(before + 10 * 60_000); + + const deferred = sql + .exec<{ attempts: number; next_at: number }>( + "SELECT attempts, next_at FROM delivery WHERE attempts = 0", + ) + .toArray(); + expect(deferred).toHaveLength(2); + for (const d of deferred) expect(d.next_at).toBe(row!.open_until); + }); + }); + + it("re-opens with a doubled cooldown when the half-open probe fails", async () => { + await withQueuedDeliveries(7, async ({ sql, calls, deliver }) => { + await deliver(); + // Cooldown elapses; every row is due again. + sql.exec("UPDATE peer_circuit SET open_until = 0"); + sql.exec("UPDATE delivery SET next_at = 0"); + calls.length = 0; + const before = Date.now(); + await deliver(); + + expect(calls).toHaveLength(1); + const [row] = circuit(sql); + expect(row?.trips).toBe(2); + expect(row!.open_until).toBeGreaterThanOrEqual(before + 20 * 60_000); + }); + }); + + it("closes on a successful probe and delivers the backlog", async () => { + await withQueuedDeliveries(7, async ({ sql, calls, deliver, respond }) => { + await deliver(); + sql.exec("UPDATE peer_circuit SET open_until = 0"); + sql.exec("UPDATE delivery SET next_at = 0"); + respond(202); + calls.length = 0; + await deliver(); + + expect(calls).toHaveLength(7); + expect(circuit(sql)).toEqual([]); + expect(sql.exec("SELECT seq FROM delivery").toArray()).toEqual([]); + }); + }); + + it("drops rows to a host failing for over 7 days while its circuit is open", async () => { + await withQueuedDeliveries(7, async ({ sql, calls, deliver }) => { + await deliver(); + sql.exec( + "UPDATE peer_circuit SET first_failed_at = ?", + Date.now() - 8 * 24 * 60 * 60 * 1000, + ); + sql.exec("UPDATE delivery SET next_at = 0"); + calls.length = 0; + await deliver(); + + expect(calls).toEqual([]); + expect(sql.exec("SELECT seq FROM delivery").toArray()).toEqual([]); + }); + }); + + it("closes when the host sends us a signed activity", async () => { + await withQueuedDeliveries( + 5, + async ({ instance, sql, deliver, username, iris }) => { + await deliver(); + expect(circuit(sql)).toHaveLength(1); + + await instance.fetch( + new Request(iris.inbox, { + method: "POST", + headers: { + "content-type": "application/activity+json", + [INTERNAL_HEADERS.config]: forwardedHeader(username), + [INTERNAL_HEADERS.signedActor]: + "https://dead.example/users/carol", + }, + body: JSON.stringify({ + id: "https://dead.example/activities/1", + type: "Like", + actor: "https://dead.example/users/carol", + object: `${iris.id}/notes/1`, + }), + }), + ); + + expect(circuit(sql)).toEqual([]); + }, + ); + }); +}); diff --git a/packages/activitypub/src/log.ts b/packages/activitypub/src/log.ts index adfdb17a..be14dd06 100644 --- a/packages/activitypub/src/log.ts +++ b/packages/activitypub/src/log.ts @@ -45,6 +45,19 @@ export const ActivityPubLogEvent = { DeliveryFailed: "activitypub.delivery.failed", /** A delivery target was refused on SSRF grounds. Field: `reason`. */ DeliveryBlocked: "activitypub.delivery.blocked", + /** + * A delivery was not attempted because its host's circuit is open (the host + * keeps failing); the row was pushed back to when the circuit half-opens, + * without spending a retry attempt. Field: `targetHost`. + */ + DeliveryDeferred: "activitypub.delivery.deferred", + /** + * A host's delivery circuit opened after consecutive retryable failures. + * Fields: `targetHost`, `failures`, `cooldownMs`. + */ + CircuitOpened: "activitypub.delivery.circuit_opened", + /** A host whose circuit had opened is reachable again. Field: `targetHost`. */ + CircuitClosed: "activitypub.delivery.circuit_closed", /** * The default key resolver could not resolve an inbound signer's key — * surfaces *why*, since `SignatureRejected`'s `key_unresolved` reason alone diff --git a/packages/activitypub/src/object.ts b/packages/activitypub/src/object.ts index 033578ce..2bd17d59 100644 --- a/packages/activitypub/src/object.ts +++ b/packages/activitypub/src/object.ts @@ -66,6 +66,24 @@ const SEEN_TTL_MS = 7 * 24 * 60 * 60 * 1000; * so a peer that later adds RFC 9421 support is eventually rediscovered. */ const SIG_PREF_TTL_MS = 7 * 24 * 60 * 60 * 1000; +/** + * Delivery circuit breaker. After this many consecutive retryable failures + * (network error, `5xx`, `408`, `429`) to one host — counted across all its + * delivery rows — the host's circuit opens and its rows are deferred, not + * attempted, so a dead peer stops tying up the alarm with timeouts. + */ +const CIRCUIT_FAILURE_THRESHOLD = 5; +/** First open-circuit cooldown; doubles on each re-trip, capped below. */ +const CIRCUIT_BASE_COOLDOWN_MS = 10 * 60_000; +/** Longest open-circuit cooldown — a dead host is still probed daily. */ +const CIRCUIT_MAX_COOLDOWN_MS = 24 * 60 * 60 * 1000; +/** + * A host failing continuously for this long is treated as gone (Mastodon's + * "unavailable domain" window): rows that reach it while its circuit is open + * are dropped instead of deferred. One probe per cooldown still goes out, and + * a success — or any signed activity from the host — closes the circuit. + */ +const CIRCUIT_DEAD_AFTER_MS = 7 * 24 * 60 * 60 * 1000; /** Max delivery rows processed per alarm wake. */ const DELIVERY_BATCH = 20; /** Timeout (ms) bounding any single outbound fetch (actor lookup / delivery). */ @@ -346,6 +364,16 @@ export class ActivityPubObject extends DurableObject { `CREATE TABLE IF NOT EXISTS sig_pref ( host TEXT PRIMARY KEY, profile TEXT NOT NULL, at INTEGER NOT NULL)`, ); + // Per-host delivery circuit breaker (see CIRCUIT_* above). A row exists + // only while the host is failing; any success deletes it. `trips` counts + // how many times the circuit has opened during this failure streak, so + // each re-open doubles the cooldown. + this.#sql.exec( + `CREATE TABLE IF NOT EXISTS peer_circuit ( + host TEXT PRIMARY KEY, failures INTEGER NOT NULL, + first_failed_at INTEGER NOT NULL, open_until INTEGER NOT NULL DEFAULT 0, + trips INTEGER NOT NULL DEFAULT 0)`, + ); // Additive-column migrations for objects created before fediverse interop // phases 1–2 (#274/#275); fresh objects already get these from the CREATE // TABLEs above. Nullable by design: `object_type`/`audience` classify @@ -680,6 +708,11 @@ export class ActivityPubObject extends DurableObject { if (author && this.#isBlocked(author)) { return text(403, "Actor is blocked"); } + // A host that just sent us a validly-signed activity is up: close its + // delivery circuit so queued deliveries to it resume at once rather than + // waiting out the cooldown (Mastodon does the same for unavailable domains). + const signerHost = signer ? hostFromUrl(signer) : undefined; + if (signerHost) this.#closeCircuit(signerHost); const id = activity.id; if (typeof id === "string" && id.length > 0) { @@ -3126,7 +3159,9 @@ export class ActivityPubObject extends DurableObject { fields: Record, ): void { const level: "info" | "warn" | "error" = - event === ActivityPubLogEvent.DeliverySucceeded + event === ActivityPubLogEvent.DeliverySucceeded || + event === ActivityPubLogEvent.DeliveryDeferred || + event === ActivityPubLogEvent.CircuitClosed ? "info" : fields.dropped === true ? "error" @@ -3256,6 +3291,77 @@ export class ActivityPubObject extends DurableObject { return drained; } + /** + * `host`'s open delivery circuit, or `null` when deliveries to it should be + * attempted (no failure streak, below the threshold, or the cooldown has + * elapsed — the half-open probe). `dead` marks a host failing for longer + * than {@link CIRCUIT_DEAD_AFTER_MS}. + */ + #openCircuit(host: string): { until: number; dead: boolean } | null { + const now = Date.now(); + const row = this.#sql + .exec<{ open_until: number; first_failed_at: number }>( + `SELECT open_until, first_failed_at FROM peer_circuit + WHERE host = ? AND open_until > ?`, + host, + now, + ) + .toArray()[0]; + if (!row) return null; + return { + until: row.open_until, + dead: now - row.first_failed_at >= CIRCUIT_DEAD_AFTER_MS, + }; + } + + /** + * Count one retryable delivery failure to `host`, opening its circuit once + * the streak reaches {@link CIRCUIT_FAILURE_THRESHOLD}. A failed half-open + * probe re-opens it at once with double the previous cooldown. + */ + #recordCircuitFailure(host: string): void { + const now = Date.now(); + const row = this.#sql + .exec<{ failures: number; trips: number }>( + `INSERT INTO peer_circuit (host, failures, first_failed_at) VALUES (?, 1, ?) + ON CONFLICT (host) DO UPDATE SET failures = failures + 1 + RETURNING failures, trips`, + host, + now, + ) + .toArray()[0]!; + if (row.failures < CIRCUIT_FAILURE_THRESHOLD) return; + const cooldownMs = Math.min( + CIRCUIT_BASE_COOLDOWN_MS * 2 ** row.trips, + CIRCUIT_MAX_COOLDOWN_MS, + ); + this.#sql.exec( + `UPDATE peer_circuit SET open_until = ?, trips = trips + 1 WHERE host = ?`, + now + cooldownMs, + host, + ); + this.#logDelivery(ActivityPubLogEvent.CircuitOpened, { + targetHost: host, + failures: row.failures, + cooldownMs, + }); + } + + /** Clear `host`'s failure streak; logs when a circuit that had opened closes. */ + #closeCircuit(host: string): void { + const closed = this.#sql + .exec<{ trips: number }>( + `DELETE FROM peer_circuit WHERE host = ? RETURNING trips`, + host, + ) + .toArray()[0]; + if (closed && closed.trips > 0) { + this.#logDelivery(ActivityPubLogEvent.CircuitClosed, { + targetHost: host, + }); + } + } + /** * The signature profile `host` is known to want for deliveries, or * `undefined` to double-knock (RFC 9421 first). Only draft-cavage peers are @@ -3309,6 +3415,30 @@ export class ActivityPubObject extends DurableObject { continue; } const targetHost = hostFromUrl(row.inbox) ?? "unknown"; + const circuit = this.#openCircuit(targetHost); + if (circuit) { + if (circuit.dead) { + this.#sql.exec(`DELETE FROM delivery WHERE seq = ?`, row.seq); + this.#logDelivery(ActivityPubLogEvent.DeliveryFailed, { + targetHost, + status: 0, + attempts: row.attempts, + dropped: true, + reason: "circuit_open", + }); + } else { + // Not attempted, so no retry attempt is spent. + this.#sql.exec( + `UPDATE delivery SET next_at = ? WHERE seq = ?`, + circuit.until, + row.seq, + ); + this.#logDelivery(ActivityPubLogEvent.DeliveryDeferred, { + targetHost, + }); + } + continue; + } const preference = this.#signaturePreference(targetHost); try { const result = await deliverActivity( @@ -3319,6 +3449,10 @@ export class ActivityPubObject extends DurableObject { userAgent, { timeoutMs: OUTBOUND_TIMEOUT_MS, preference }, ); + // Any answer but a retryable one proves the host is up; a retryable + // one (network error, 5xx, 408, 429) counts toward its circuit. + if (result.ok || !result.retryable) this.#closeCircuit(targetHost); + else this.#recordCircuitFailure(targetHost); if (result.ok) { this.#sql.exec(`DELETE FROM delivery WHERE seq = ?`, row.seq); if (result.profile === "cavage" && preference !== "cavage") { diff --git a/spec/packages/activitypub.md b/spec/packages/activitypub.md index 685df125..c3b6b111 100644 --- a/spec/packages/activitypub.md +++ b/spec/packages/activitypub.md @@ -117,6 +117,20 @@ large media; bodies **MUST** stream and **MUST NOT** be buffered in the DO. actor DO's SQLite (`sig_pref`) for 7 days; deliveries to it skip the RFC 9421 knock until the row expires, so a peer that later adds RFC 9421 support is rediscovered. Hosts are not recorded for any other outcome. +- Outbound delivery has a per-host **circuit breaker** (actor DO SQLite, + `peer_circuit`). Five consecutive retryable failures to one host (network + error, `5xx`, `408`, `429`), counted across all of its delivery rows, open + the circuit for 10 minutes. While open, that host's due rows are deferred to + the end of the cooldown without spending a retry attempt. When the cooldown + elapses, the next row is a half-open probe: a failure re-opens the circuit + with double the previous cooldown (capped at 24 h), and a success closes it. + Any non-retryable answer (`2xx`, or a `4xx` other than `408`/`429`) also + closes it, as does a validly-signed inbound activity from the host. Once a + host has been failing for 7 days, rows that reach it while its circuit is + open are dropped instead of deferred (one probe per cooldown still goes + out). Events: `activitypub.delivery.deferred`, + `activitypub.delivery.circuit_opened`, `activitypub.delivery.circuit_closed`, + and `activitypub.delivery.failed` with `reason: "circuit_open"` for a drop. - Outbound `GET`s (key resolution, inbox resolution, actor-profile refresh, relayed-object verification, the `activitypub_resolve` actor fetch) are HTTP-signed as the actor (draft-cavage, `(request-target) host date`)