From 60f60bdc952e7059a2fd28a873b4620b79158060 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Thu, 5 Nov 2026 18:31:14 +0900 Subject: [PATCH 1/2] Await sibling deliveries after a fan-out failure The immediate fan-out paths in FederationImpl.sendActivity() and InboxContextImpl.forwardActivityInternal() used Promise.all() alone. When one delivery rejected, the method returned while the remaining deliveries were still in flight, so Cloudflare Workers could terminate the isolate and cancel them. Put the fan-out loop and Promise.all() in a try block and drain the started promises with Promise.allSettled() in finally. The original error still propagates after the siblings settle, and a synchronous failure while preparing a later recipient no longer abandons earlier deliveries. Attach a no-op catch to each started promise so an early rejection cannot surface as an unhandled rejection while a later Collection-Synchronization header is being built. Add regression tests for the sendActivity() and forwardActivity() immediate paths and for a fan-out setup failure. Document the fix in the 2.0.36 changelog. https://github.com/fedify-dev/fedify/issues/1380 Assisted-by: OpenCode:deepseek-flash Assisted-by: Codex:gpt-6-astra Assisted-by: Claude Code:claude-fable-5-1 Assisted-by: OpenCode:gemini-4-argon --- CHANGES.md | 9 + changes.d/fedify/await-sibling-deliveries.md | 12 + .../fedify/src/federation/middleware.test.ts | 270 ++++++++++++++++++ packages/fedify/src/federation/middleware.ts | 47 ++- 4 files changed, 326 insertions(+), 12 deletions(-) create mode 100644 changes.d/fedify/await-sibling-deliveries.md diff --git a/CHANGES.md b/CHANGES.md index 058fbedf1..30d4d6eae 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -10,6 +10,13 @@ To be released. ### @fedify/fedify + - Fixed `Context.sendActivity()` and `InboxContext.forwardActivity()` giving + up on the remaining deliveries when one recipient failed. The immediate + delivery paths now wait for every delivery they have already started before + propagating the error, so the remaining recipients are still reached on + runtimes such as Cloudflare Workers, which may terminate an isolate once + the error response is returned. This applies both to `immediate: true` and + to federations without an outbox queue. [[#1380], [#1388]] - Fixed Linked Data Signatures being emitted with a JSON-LD context that no longer resolves. The attached `signature` object now uses `https://w3id.org/security/v1` instead of @@ -46,7 +53,9 @@ To be released. [#664]: https://github.com/fedify-dev/fedify/issues/664 [#1339]: https://github.com/fedify-dev/fedify/issues/1339 [#1347]: https://github.com/fedify-dev/fedify/pull/1347 +[#1380]: https://github.com/fedify-dev/fedify/issues/1380 [#1381]: https://github.com/fedify-dev/fedify/issues/1381 +[#1388]: https://github.com/fedify-dev/fedify/pull/1388 [#1390]: https://github.com/fedify-dev/fedify/pull/1390 ### @fedify/vocab diff --git a/changes.d/fedify/await-sibling-deliveries.md b/changes.d/fedify/await-sibling-deliveries.md new file mode 100644 index 000000000..8766c3492 --- /dev/null +++ b/changes.d/fedify/await-sibling-deliveries.md @@ -0,0 +1,12 @@ +--- +links: + '#1380': https://github.com/fedify-dev/fedify/issues/1380 + '#1388': https://github.com/fedify-dev/fedify/pull/1388 +--- + - Fixed `Context.sendActivity()` and `InboxContext.forwardActivity()` giving + up on the remaining deliveries when one recipient failed. The immediate + delivery paths now wait for every delivery they have already started before + propagating the error, so the remaining recipients are still reached on + runtimes such as Cloudflare Workers, which may terminate an isolate once + the error response is returned. This applies both to `immediate: true` and + to federations without an outbox queue. [[#1380], [#1388]] diff --git a/packages/fedify/src/federation/middleware.test.ts b/packages/fedify/src/federation/middleware.test.ts index 7e4f06efd..aad33577e 100644 --- a/packages/fedify/src/federation/middleware.test.ts +++ b/packages/fedify/src/federation/middleware.test.ts @@ -65,6 +65,7 @@ import { import type { MessageQueue } from "./mq.ts"; import type { InboxMessage, Message, OutboxMessage } from "./queue.ts"; import { RouterError } from "./router.ts"; +import { SendActivityError } from "./send.ts"; test("createFederation()", async (t) => { const kv = new MemoryKvStore(); @@ -7275,6 +7276,275 @@ test("ContextImpl.sendActivity() honors the private-address policy", async (t) = } }); +interface DeferredSignal { + readonly promise: Promise; + readonly resolve: () => void; +} + +function deferredSignal(): DeferredSignal { + let resolve!: () => void; + const promise = new Promise((r) => resolve = r); + return { promise, resolve }; +} + +test("ContextImpl.sendActivity() waits for sibling deliveries on failure", async () => { + const failingInbox = "http://127.0.0.1:3001/inbox/fail"; + const slowInbox = "http://127.0.0.1:3002/inbox/slow"; + const failingEntered = deferredSignal(); + const slowEntered = deferredSignal(); + const slowCompleted = deferredSignal(); + let releaseSlow!: () => void; + const slowGate = new Promise((resolve) => releaseSlow = resolve); + let slowStarted = false; + let slowSettled = false; + + fetchMock.spyGlobal(); + fetchMock.post(failingInbox, () => { + failingEntered.resolve(); + return new Response(null, { status: 500 }); + }); + fetchMock.post(slowInbox, async () => { + slowStarted = true; + slowEntered.resolve(); + await slowGate; + slowSettled = true; + slowCompleted.resolve(); + return new Response(null, { status: 202 }); + }); + + let observed: Promise | undefined; + let released = false; + try { + const federation = createFederation({ + kv: new MemoryKvStore(), + allowPrivateAddress: true, + }); + const ctx = federation.createContext(new URL("https://example.com/")); + const activity = new Create({ + id: new URL("https://example.com/activity/fanout"), + actor: new URL("https://example.com/person"), + }); + let outerSettled = false; + let outerError: unknown; + const sending = ctx.sendActivity( + [{ privateKey: ed25519PrivateKey, keyId: ed25519Multikey.id! }], + [ + new Person({ + id: new URL("https://example.com/recipients/failing"), + inbox: new URL(failingInbox), + }), + new Person({ + id: new URL("https://example.com/recipients/slow"), + inbox: new URL(slowInbox), + }), + ], + activity, + { immediate: true }, + ); + observed = sending.then( + () => { + outerSettled = true; + }, + (error) => { + outerSettled = true; + outerError = error; + }, + ); + await Promise.all([failingEntered.promise, slowEntered.promise]); + await new Promise((resolve) => setTimeout(resolve, 0)); + assertFalse( + outerSettled, + "sendActivity() must wait for sibling deliveries before rejecting", + ); + releaseSlow(); + released = true; + await observed; + assert(slowSettled, "the delayed delivery should have settled"); + assertInstanceOf(outerError, SendActivityError); + } finally { + if (!released) releaseSlow(); + if (slowStarted) await slowCompleted.promise; + await observed; + fetchMock.hardReset(); + } +}); + +test("InboxContextImpl.forwardActivity() waits for sibling deliveries on failure", async () => { + const failingInbox = "http://127.0.0.1:3003/inbox/fail"; + const slowInbox = "http://127.0.0.1:3004/inbox/slow"; + const failingEntered = deferredSignal(); + const slowEntered = deferredSignal(); + const slowCompleted = deferredSignal(); + let releaseSlow!: () => void; + const slowGate = new Promise((resolve) => releaseSlow = resolve); + let slowStarted = false; + let slowSettled = false; + + fetchMock.spyGlobal(); + fetchMock.post(failingInbox, () => { + failingEntered.resolve(); + return new Response(null, { status: 500 }); + }); + fetchMock.post(slowInbox, async () => { + slowStarted = true; + slowEntered.resolve(); + await slowGate; + slowSettled = true; + slowCompleted.resolve(); + return new Response(null, { status: 202 }); + }); + + let observed: Promise | undefined; + let released = false; + try { + const federation = new FederationImpl({ + kv: new MemoryKvStore(), + allowPrivateAddress: true, + }); + const activity = await signObject( + new vocab.Create({ + id: new URL("https://example.com/activity/forward"), + actor: new URL("https://example.com/person2"), + }), + ed25519PrivateKey, + ed25519Multikey.id!, + { contextLoader: mockDocumentLoader, documentLoader: mockDocumentLoader }, + ); + const ctx = new InboxContextImpl( + null, + await activity.toJsonLd({ contextLoader: mockDocumentLoader }), + activity.id?.href, + getTypeId(activity).href, + { + data: undefined, + federation, + url: new URL("https://example.com/"), + documentLoader: mockDocumentLoader, + contextLoader: mockDocumentLoader, + }, + ); + let outerSettled = false; + let outerError: unknown; + const sending = ctx.forwardActivity( + [{ privateKey: ed25519PrivateKey, keyId: ed25519Multikey.id! }], + [ + { + id: new URL("https://example.com/recipients/failing"), + inboxId: new URL(failingInbox), + }, + { + id: new URL("https://example.com/recipients/slow"), + inboxId: new URL(slowInbox), + }, + ], + { immediate: true, skipIfUnsigned: true }, + ); + observed = sending.then( + () => { + outerSettled = true; + }, + (error) => { + outerSettled = true; + outerError = error; + }, + ); + await Promise.all([failingEntered.promise, slowEntered.promise]); + await new Promise((resolve) => setTimeout(resolve, 0)); + assertFalse( + outerSettled, + "forwardActivity() must wait for sibling deliveries before rejecting", + ); + releaseSlow(); + released = true; + await observed; + assert(slowSettled, "the delayed delivery should have settled"); + assertInstanceOf(outerError, SendActivityError); + } finally { + if (!released) releaseSlow(); + if (slowStarted) await slowCompleted.promise; + await observed; + fetchMock.hardReset(); + } +}); + +test("FederationImpl.sendActivity() drains deliveries when fan-out setup fails", async () => { + const slowInbox = "http://127.0.0.1:3005/inbox/slow"; + const slowEntered = deferredSignal(); + const slowCompleted = deferredSignal(); + let releaseSlow!: () => void; + const slowGate = new Promise((resolve) => releaseSlow = resolve); + let slowStarted = false; + let slowSettled = false; + + fetchMock.spyGlobal(); + fetchMock.post(slowInbox, async () => { + slowStarted = true; + slowEntered.resolve(); + await slowGate; + slowSettled = true; + slowCompleted.resolve(); + return new Response(null, { status: 202 }); + }); + + let observed: Promise | undefined; + let released = false; + try { + const federation = new FederationImpl({ + kv: new MemoryKvStore(), + allowPrivateAddress: true, + }); + const context = federation.createContext(new URL("https://example.com/")); + const activity = new Create({ + id: new URL("https://example.com/activity/setup-failure"), + actor: new URL("https://example.com/person"), + }); + let outerSettled = false; + let outerError: unknown; + const sending = federation.sendActivity( + [{ privateKey: ed25519PrivateKey, keyId: ed25519Multikey.id! }], + { + [slowInbox]: { + actorIds: ["https://example.com/recipients/slow"], + sharedInbox: false, + }, + // Synthetic invalid input: the fan-out fails while preparing the + // second recipient, after the first delivery has already started. + "not a url": { + actorIds: ["https://example.com/recipients/invalid"], + sharedInbox: false, + }, + }, + activity, + { context, immediate: true }, + ); + observed = sending.then( + () => { + outerSettled = true; + }, + (error) => { + outerSettled = true; + outerError = error; + }, + ); + await slowEntered.promise; + await new Promise((resolve) => setTimeout(resolve, 0)); + assertFalse( + outerSettled, + "sendActivity() must drain started deliveries when fan-out setup fails", + ); + releaseSlow(); + released = true; + await observed; + assert(slowSettled, "the started delivery should have settled"); + assertInstanceOf(outerError, TypeError); + } finally { + if (!released) releaseSlow(); + if (slowStarted) await slowCompleted.promise; + await observed; + fetchMock.hardReset(); + } +}); + test("Federation.fetch() bounds inbox bodies before dispatch", async () => { let dispatched = false; const json = JSON.stringify({ diff --git a/packages/fedify/src/federation/middleware.ts b/packages/fedify/src/federation/middleware.ts index af3b83b78..eff84d922 100644 --- a/packages/fedify/src/federation/middleware.ts +++ b/packages/fedify/src/federation/middleware.ts @@ -1473,9 +1473,9 @@ export class FederationImpl ); } const promises: Promise[] = []; - for (const inbox in inboxes) { - promises.push( - sendActivity({ + try { + for (const inbox in inboxes) { + const promise = sendActivity({ allowPrivateAddress: this.allowPrivateAddress, keys, activity: jsonLd, @@ -1496,10 +1496,22 @@ export class FederationImpl this.firstKnock, ), tracerProvider: this.tracerProvider, - }), - ); + }); + // Mark the delivery as handled so an early rejection does not + // surface as an unhandled rejection while a later recipient is + // still being prepared. Promise.all() and Promise.allSettled() + // still observe the original rejection. + promise.catch(() => {}); + promises.push(promise); + } + await Promise.all(promises); + } finally { + // Drain the deliveries that were already started before propagating + // the original error, so that a failing recipient cannot cancel its + // siblings when the runtime tears down the request context (e.g., + // Cloudflare Workers). + await Promise.allSettled(promises); } - await Promise.all(promises); return; } logger.debug( @@ -3432,9 +3444,9 @@ export class InboxContextImpl extends ContextImpl ); } const promises: Promise[] = []; - for (const inbox in inboxes) { - promises.push( - sendActivity({ + try { + for (const inbox in inboxes) { + const promise = sendActivity({ allowPrivateAddress: this.federation.allowPrivateAddress, keys, activity: this.activity, @@ -3448,10 +3460,21 @@ export class InboxContextImpl extends ContextImpl this.federation.kvPrefixes.httpMessageSignaturesSpec, this.federation.firstKnock, ), - }), - ); + }); + // Mark the delivery as handled so an early rejection does not + // surface as an unhandled rejection if the loop fails before + // Promise.all() attaches its handlers. + promise.catch(() => {}); + promises.push(promise); + } + await Promise.all(promises); + } finally { + // Drain the deliveries that were already started before propagating + // the original error, so that a failing recipient cannot cancel its + // siblings when the runtime tears down the request context (e.g., + // Cloudflare Workers). + await Promise.allSettled(promises); } - await Promise.all(promises); return; } logger.debug( From 5cdba4678b25686f795aab4ddb90f7223fe77a0f Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 6 Nov 2026 20:35:46 +0900 Subject: [PATCH 2/2] Test early rejection during header fan-out The no-op catch on each started delivery prevents an unhandled rejection while a later Collection-Synchronization header is still being built. Gate the second digest and make the first delivery fail before it, then assert that no unhandled rejection is reported. https://github.com/fedify-dev/fedify/pull/1388#discussion_r4491741455 Assisted-by: OpenCode:deepseek-flash Changelog: none --- .../fedify/src/federation/middleware.test.ts | 73 +++++++++++++++++++ 1 file changed, 73 insertions(+) diff --git a/packages/fedify/src/federation/middleware.test.ts b/packages/fedify/src/federation/middleware.test.ts index aad33577e..7861aee3f 100644 --- a/packages/fedify/src/federation/middleware.test.ts +++ b/packages/fedify/src/federation/middleware.test.ts @@ -7545,6 +7545,79 @@ test("FederationImpl.sendActivity() drains deliveries when fan-out setup fails", } }); +test("FederationImpl.sendActivity() handles rejections while building headers", async () => { + const federation = new FederationImpl({ + kv: new MemoryKvStore(), + }); + const context = federation.createContext(new URL("https://example.com/")); + const activity = await signObject( + new Create({ + id: new URL("https://example.com/activity/header-failure"), + actor: new URL("https://example.com/person"), + }), + ed25519PrivateKey, + ed25519Multikey.id!, + { contextLoader: mockDocumentLoader, documentLoader: mockDocumentLoader }, + ); + const subtle = crypto.subtle; + const originalDigest = subtle.digest; + let digestCalls = 0; + let releaseSecondDigest!: () => void; + const secondDigestGate = new Promise((resolve) => { + releaseSecondDigest = resolve; + }); + let secondDigestStarted!: () => void; + const secondDigestEntered = new Promise((resolve) => { + secondDigestStarted = resolve; + }); + subtle.digest = async (algorithm, data) => { + digestCalls++; + if (digestCalls === 2) { + secondDigestStarted(); + await secondDigestGate; + } + return await originalDigest.call(subtle, algorithm, data); + }; + const capture = captureUnhandledRejections(); + let released = false; + try { + const sending = federation.sendActivity( + [{ privateKey: ed25519PrivateKey, keyId: ed25519Multikey.id! }], + { + "ftp://example.com/fail": { + actorIds: ["https://example.com/recipients/failing"], + sharedInbox: false, + }, + "http://127.0.0.1:3006/inbox": { + actorIds: ["https://example.com/recipients/slow"], + sharedInbox: false, + }, + }, + activity, + { + context, + immediate: true, + collectionSync: "https://example.com/followers", + }, + ); + const observed = sending.catch((error) => error); + await secondDigestEntered; + await capture.settle(); + assertEquals( + capture.reasons, + [], + "an early rejection must not surface as an unhandled rejection", + ); + releaseSecondDigest(); + released = true; + await observed; + } finally { + if (!released) releaseSecondDigest(); + subtle.digest = originalDigest; + capture.stop(); + } +}); + test("Federation.fetch() bounds inbox bodies before dispatch", async () => { let dispatched = false; const json = JSON.stringify({