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..7861aee3f 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,348 @@ 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("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({ 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(