From 772fbeec381c448ecca489ceddeee9b0ec6c5924 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Thu, 5 Nov 2026 18:41:48 +0900 Subject: [PATCH] Report automatic queue startup failures Automatic queue worker startup is best-effort: inbox handling and Context.sendActivity() call _startQueueInternal() without awaiting it. WorkersMessageQueue.listen() throws synchronously because Cloudflare Queues cannot be polled, so the async call produced an unhandled rejection. The synchronous throw also aborted the loop that collects listener promises, leaving listeners started before the throw unobserved and later queues never started. Normalize a synchronous throw from listen() into a rejected promise so every started listener reaches Promise.all, and route the three implicit call sites through a helper that logs the failure once instead of leaving the rejection unhandled. Explicit Federation.startQueue() keeps rejecting so deliberate worker startup failures stay visible. The failed worker is not retried, which the changelog and the manuallyStartQueue reference now state. Cloudflare Workers cannot start workers in the request process at all, so the example and the deployment and message queue documentation now set manuallyStartQueue to true. Fixes https://github.com/fedify-dev/fedify/issues/1381 Assisted-by: OpenCode:deepseek-flash Assisted-by: Codex:gpt-6-astra Assisted-by: Claude Code:claude-fable-5-1 --- CHANGES.md | 14 + changes.d/fedify/queue-startup-failures.md | 17 + docs/manual/deploy.md | 16 +- docs/manual/federation.md | 15 +- docs/manual/mq.md | 14 + examples/cloudflare-workers/src/index.ts | 7 + packages/cfworkers/README.md | 43 +- packages/fedify/src/federation/federation.ts | 6 + .../fedify/src/federation/middleware.test.ts | 414 ++++++++++++++++++ packages/fedify/src/federation/middleware.ts | 72 ++- 10 files changed, 577 insertions(+), 41 deletions(-) create mode 100644 changes.d/fedify/queue-startup-failures.md diff --git a/CHANGES.md b/CHANGES.md index 2d76a3c9a..058fbedf1 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -22,6 +22,18 @@ To be released. fanout before the upgrade also keep their original proof-covered JSON when replayed. Incoming signatures that still use `identity/v1` continue to verify. [[#664]] + - Fixed automatic queue worker startup failures becoming unhandled promise + rejections. When `manuallyStartQueue` was left `false` and a `MessageQueue` + implementation could not start listening, such as `WorkersMessageQueue` on + Cloudflare Workers whose `listen()` throws a `TypeError`, the rejection from + the implicit startup in inbox handling and in `Context.sendActivity()` was + discarded. It is now logged as a warning once per `Federation` instance, + and a synchronous throw from one queue no longer prevents the remaining + queues from starting. The failed worker is not retried, so restart the + process after fixing the queue. `Federation.startQueue()` still rejects so + callers that start workers deliberately see the error, and Cloudflare + Workers deployments should set `manuallyStartQueue: true` to avoid the + warning entirely. [[#1381], [#1390]] - Fixed inbox handling that treated valid date-times without seconds, such as `2024-01-01T10:00Z` or `2024-01-01T10:00.5Z`, as permanently malformed. Activity Streams 2.0 allows seconds to be omitted, so activities with such @@ -34,6 +46,8 @@ 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 +[#1381]: https://github.com/fedify-dev/fedify/issues/1381 +[#1390]: https://github.com/fedify-dev/fedify/pull/1390 ### @fedify/vocab diff --git a/changes.d/fedify/queue-startup-failures.md b/changes.d/fedify/queue-startup-failures.md new file mode 100644 index 000000000..6bee5b62e --- /dev/null +++ b/changes.d/fedify/queue-startup-failures.md @@ -0,0 +1,17 @@ +--- +links: + '#1381': https://github.com/fedify-dev/fedify/issues/1381 + '#1390': https://github.com/fedify-dev/fedify/pull/1390 +--- + - Fixed automatic queue worker startup failures becoming unhandled promise + rejections. When `manuallyStartQueue` was left `false` and a `MessageQueue` + implementation could not start listening, such as `WorkersMessageQueue` on + Cloudflare Workers whose `listen()` throws a `TypeError`, the rejection from + the implicit startup in inbox handling and in `Context.sendActivity()` was + discarded. It is now logged as a warning once per `Federation` instance, + and a synchronous throw from one queue no longer prevents the remaining + queues from starting. The failed worker is not retried, so restart the + process after fixing the queue. `Federation.startQueue()` still rejects so + callers that start workers deliberately see the error, and Cloudflare + Workers deployments should set `manuallyStartQueue: true` to avoid the + warning entirely. [[#1381], [#1390]] diff --git a/docs/manual/deploy.md b/docs/manual/deploy.md index b4f43d03d..557def34d 100644 --- a/docs/manual/deploy.md +++ b/docs/manual/deploy.md @@ -359,6 +359,9 @@ export default { const federation = await builder.build({ kv: new WorkersKvStore(env.KV_NAMESPACE), queue: new WorkersMessageQueue(env.QUEUE), + // Cloudflare Queues cannot be polled, so the queue() handler below + // consumes the messages instead of Federation.startQueue(): + manuallyStartQueue: true, // Other options... }); @@ -371,10 +374,16 @@ export default { Cloudflare Queues don't provide polling-based APIs, so the `WorkersMessageQueue` cannot implement a traditional `~MessageQueue.listen()` method. Instead, you -must manually connect queue handlers: +must manually connect queue handlers. + +Because `WorkersMessageQueue.listen()` throws a `TypeError`, set +`~FederationOptions.manuallyStartQueue` to `true` as well. Otherwise Fedify +tries to start the queue workers automatically the first time an activity is +received or sent, and logs a warning through [LogTape](./log.md) that the +workers could not be started. The `queue()` handler below is what actually +processes the messages, so the option only prevents that unsuccessful attempt: ~~~~ typescript twoslash -// @noErrors: 2345 import { createFederationBuilder, type Message } from "@fedify/fedify"; import { Person } from "@fedify/vocab"; import { WorkersKvStore, WorkersMessageQueue } from "@fedify/cfworkers"; @@ -394,13 +403,14 @@ export default { const federation = await builder.build({ kv: new WorkersKvStore(env.KV_NAMESPACE), queue: new WorkersMessageQueue(env.QUEUE), + manuallyStartQueue: true, }); for (const message of batch.messages) { try { await federation.processQueuedTask( - message.body as unknown as Message, env, + message.body as unknown as Message, ); message.ack(); } catch (error) { diff --git a/docs/manual/federation.md b/docs/manual/federation.md index 489629e70..ebce7af99 100644 --- a/docs/manual/federation.md +++ b/docs/manual/federation.md @@ -194,13 +194,26 @@ Further details are explained in the [*Message queue* section](./mq.md). Whether to start the task queue manually or automatically. If `true`, the task queue will not start automatically and you need to -manually start it by calling the `Federation.startQueue()` method. +manually start it by calling the `Federation.startQueue()` method. On runtimes +that cannot start workers in the request process, such as Cloudflare Workers, +`Federation.processQueuedTask()` consumes the messages instead, so you still set +this option to `true`. If `false`, the task queue will start automatically as soon as the first task is enqueued. By default, the queue starts automatically. +> [!NOTE] +> Automatic startup is best-effort. If a queue cannot start listening +> (for example, when `MessageQueue.listen()` throws), Fedify logs a warning +> through [LogTape](./log.md) once per `Federation` instance instead of +> producing an unhandled promise rejection. The failed worker is not retried by +> later requests, and the application keeps running without it, so restart the +> process or recreate the federation after fixing the queue. Explicitly calling +> `Federation.startQueue()` reports the error to the caller instead of logging +> it. + > [!TIP] > This option is useful when you want to separately deploy the web server > and the task queue worker. In this case, you can start the task queue diff --git a/docs/manual/mq.md b/docs/manual/mq.md index 56de3b5b0..985dd763c 100644 --- a/docs/manual/mq.md +++ b/docs/manual/mq.md @@ -477,6 +477,9 @@ export default { kv: undefined as unknown as KvStore, // ---cut-end--- queue: new WorkersMessageQueue(env.QUEUE_BINDING), + // Cloudflare Queues cannot be polled, so the queue() handler below + // consumes the messages instead of Federation.startQueue(): + manuallyStartQueue: true, }); // Omit the rest of the code for brevity }, @@ -491,6 +494,10 @@ export default { kv: undefined as unknown as KvStore, // ---cut-end--- queue: new WorkersMessageQueue(env.QUEUE_BINDING), + // Inbox listeners run inside processQueuedTask() and may call + // ctx.sendActivity(), which would otherwise start the queue workers + // automatically: + manuallyStartQueue: true, }); for (const msg of batch.messages) { await federation.processQueuedTask( @@ -521,6 +528,13 @@ export default { > method, you need to call `Federation.processQueuedTask()` method to manually > process the messages. The `queue()` method is the only way to consume > messages from the queue in Cloudflare Workers. +> +> Set `~FederationOptions.manuallyStartQueue` to `true` for the same reason. +> Otherwise Fedify tries to start the queue workers automatically the first time +> an activity is received or sent, and since `listen()` throws, it logs a +> warning through [LogTape](./log.md) that the workers could not be started. +> The workers are not started on this runtime either way; the option prevents +> that unsuccessful attempt and the warning. [Cloudflare Workers]: https://workers.cloudflare.com/ [Cloudflare Queues]: https://developers.cloudflare.com/queues/ diff --git a/examples/cloudflare-workers/src/index.ts b/examples/cloudflare-workers/src/index.ts index 1660a77c7..4ac8ab802 100644 --- a/examples/cloudflare-workers/src/index.ts +++ b/examples/cloudflare-workers/src/index.ts @@ -149,6 +149,9 @@ export default { const federation: Federation = await builder.build({ kv: new WorkersKvStore(env.FEDIFY_KV), queue: new WorkersMessageQueue(env.FEDIFY_QUEUE), + // Cloudflare Queues cannot be polled, so the queue() handler below + // consumes the messages instead of Federation.startQueue(): + manuallyStartQueue: true, }); const url = new URL(req.url); if (url.pathname === "/") return await listFollowers(req, env); @@ -163,6 +166,10 @@ export default { const federation: Federation = await builder.build({ kv: new WorkersKvStore(env.FEDIFY_KV), queue: new WorkersMessageQueue(env.FEDIFY_QUEUE), + // Inbox listeners run inside processQueuedTask() and may call + // ctx.sendActivity(), which would otherwise start the queue workers + // automatically: + manuallyStartQueue: true, }); for (const message of batch.messages) { await federation.processQueuedTask( diff --git a/packages/cfworkers/README.md b/packages/cfworkers/README.md index a9743337d..db9d184f5 100644 --- a/packages/cfworkers/README.md +++ b/packages/cfworkers/README.md @@ -16,35 +16,44 @@ implementations for [Cloudflare Workers]: - [`WorkersMessageQueue`] ~~~~ typescript -import type { Federation } from "@fedify/fedify"; +import { createFederation, type Message } from "@fedify/fedify"; import { WorkersKvStore, WorkersMessageQueue } from "@fedify/cfworkers"; +type Env = { + KV_BINDING: KVNamespace; + QUEUE_BINDING: Queue; +}; + export default { - async fetch(request, env, ctx) { - const federation = createFederation({ + async fetch(request: Request, env: Env): Promise { + const federation = createFederation({ kv: new WorkersKvStore(env.KV_BINDING), queue: new WorkersMessageQueue(env.QUEUE_BINDING), + // Cloudflare Queues cannot be polled, so the queue() handler below + // consumes the messages instead of Federation.startQueue(): + manuallyStartQueue: true, // ... other options }); - - return federation.handle(request, { contextData: env }); + + return await federation.fetch(request, { contextData: env }); }, - - async queue(batch, env, ctx) { - const federation = createFederation({ + + async queue(batch: MessageBatch, env: Env): Promise { + const federation = createFederation({ kv: new WorkersKvStore(env.KV_BINDING), queue: new WorkersMessageQueue(env.QUEUE_BINDING), + // Inbox listeners run inside processQueuedTask() and may call + // ctx.sendActivity(), which would otherwise start the queue workers + // automatically: + manuallyStartQueue: true, // ... other options }); - + for (const message of batch.messages) { - await federation.processQueuedTask(message.body); + await federation.processQueuedTask(env, message.body as Message); } - } -} satisfies ExportedHandler<{ - KV_BINDING: KVNamespace; - QUEUE_BINDING: Queue; -}>; + }, +} satisfies ExportedHandler; ~~~~ [JSR badge]: https://jsr.io/badges/@fedify/cfworkers @@ -99,7 +108,9 @@ management. > API when new messages are available in the queue. Inside the `queue()` > method, you need to call `Federation.processQueuedTask()` method to manually > process the messages. The `queue()` method is the only way to consume -> messages from the queue in Cloudflare Workers. +> messages from the queue in Cloudflare Workers. Set `manuallyStartQueue` to +> `true` as well; otherwise Fedify tries to start the workers automatically, +> and since `listen()` throws, it logs a warning that they could not be started. [Cloudflare Queues]: https://developers.cloudflare.com/queues/ diff --git a/packages/fedify/src/federation/federation.ts b/packages/fedify/src/federation/federation.ts index 7a517b52b..7b7f523c5 100644 --- a/packages/fedify/src/federation/federation.ts +++ b/packages/fedify/src/federation/federation.ts @@ -814,6 +814,12 @@ export interface FederationOptions { * * By default, the queue starts automatically. * + * Automatic startup is best-effort: if a queue cannot start listening, + * a warning is logged once per federation instance and the failed worker is + * not retried, while an explicit {@link Federation.startQueue} call rejects. + * Runtimes that consume the queue elsewhere, such as Cloudflare Workers with + * {@link Federation.processQueuedTask}, should set this to `true`. + * * @since 0.12.0 */ manuallyStartQueue?: boolean; diff --git a/packages/fedify/src/federation/middleware.test.ts b/packages/fedify/src/federation/middleware.test.ts index c136f5339..7e4f06efd 100644 --- a/packages/fedify/src/federation/middleware.test.ts +++ b/packages/fedify/src/federation/middleware.test.ts @@ -7,6 +7,7 @@ import type { DocumentLoader } from "@fedify/vocab-runtime"; import type { TracerProvider } from "@opentelemetry/api"; import * as vocab from "@fedify/vocab"; import { Create, getTypeId, lookupObject, Offer, Person } from "@fedify/vocab"; +import { configure, type LogRecord, reset } from "@logtape/logtape"; import { assert, assertEquals, @@ -2275,6 +2276,419 @@ test("FederationImpl.sendActivity()", async (t) => { fetchMock.hardReset(); }); +const QUEUE_STARTUP_WARNING = + "Failed to start queue workers automatically: {error}"; + +interface UnhandledRejectionCapture { + reasons: unknown[]; + settle(): Promise; + stop(): void; +} + +/** + * Captures unhandled promise rejections for the duration of a test step. + * + * Deno dispatches `unhandledrejection` on `globalThis`, while Node.js and Bun + * dispatch `unhandledRejection` on `process`. Bun's test runner currently + * fires neither, so there the native runner's own unhandled-rejection failure + * remains the signal; the assertion is harmless otherwise. + */ +function captureUnhandledRejections(): UnhandledRejectionCapture { + const reasons: unknown[] = []; + const isDeno = "Deno" in globalThis; + interface RejectionEvent { + reason: unknown; + preventDefault(): void; + } + const onRejectionEvent = (event: RejectionEvent): void => { + // Keep Deno from treating a deliberate pre-fix failure as a fatal error + // that cancels the remaining tests in this file. + event.preventDefault(); + reasons.push(event.reason); + }; + const onProcessRejection = (reason: unknown): void => { + reasons.push(reason); + }; + const globalTarget = globalThis as unknown as { + addEventListener?( + type: string, + listener: (event: RejectionEvent) => void, + ): void; + removeEventListener?( + type: string, + listener: (event: RejectionEvent) => void, + ): void; + }; + const proc = (globalThis as unknown as { + process?: { + on?(type: string, listener: (reason: unknown) => void): void; + off?(type: string, listener: (reason: unknown) => void): void; + }; + }).process; + if (isDeno) { + globalTarget.addEventListener?.("unhandledrejection", onRejectionEvent); + } else { + proc?.on?.("unhandledRejection", onProcessRejection); + } + async function settle(): Promise { + // Deno dispatches `unhandledrejection` only after two timer turns. + await new Promise((resolve) => setTimeout(resolve, 0)); + await new Promise((resolve) => setTimeout(resolve, 0)); + } + return { + reasons, + settle, + stop() { + if (isDeno) { + globalTarget.removeEventListener?.( + "unhandledrejection", + onRejectionEvent, + ); + } else { + proc?.off?.("unhandledRejection", onProcessRejection); + } + }, + }; +} + +async function configureQueueStartupWarningCapture( + records: LogRecord[], +): Promise { + await configure({ + sinks: { + buffer(record: LogRecord): void { + records.push(record); + }, + }, + loggers: [ + { category: "fedify", sinks: ["buffer"], lowestLevel: "warning" }, + { category: ["logtape", "meta"], sinks: [] }, + ], + reset: true, + }); +} + +function queueStartupWarnings( + records: LogRecord[], + error: unknown, +): LogRecord[] { + return records.filter((record) => + record.level === "warning" && + record.category.length === 3 && + record.category[0] === "fedify" && + record.category[1] === "federation" && + record.category[2] === "queue" && + record.rawMessage === QUEUE_STARTUP_WARNING && + record.properties.error === error + ); +} + +test("Federation.fetch() reports automatic queue startup failures", async (t) => { + const listenError = new TypeError("simulated listen() failure"); + const queue: MessageQueue = { + enqueue() { + return Promise.resolve(); + }, + listen() { + throw listenError; + }, + }; + const capture = captureUnhandledRejections(); + const records: LogRecord[] = []; + await configureQueueStartupWarningCapture(records); + try { + const federation = createFederation({ + kv: new MemoryKvStore(), + queue, + skipSignatureVerification: true, + documentLoaderFactory: () => mockDocumentLoader, + contextLoaderFactory: () => mockDocumentLoader, + }); + federation.setActorDispatcher( + "/actors/{identifier}", + () => new Person({ id: new URL("https://example.com/actor") }), + ); + federation.setInboxListeners("/actors/{identifier}/inbox", "/inbox") + .on(Create, () => {}); + const post = (id: string) => + federation.fetch( + new Request("https://example.com/inbox", { + method: "POST", + headers: { "Content-Type": "application/activity+json" }, + body: JSON.stringify({ + "@context": "https://www.w3.org/ns/activitystreams", + id, + type: "Create", + actor: "https://example.com/actor", + }), + }), + { contextData: undefined }, + ); + + await t.step("synchronous failure", async () => { + const response = await post("https://example.com/activities/queued-1"); + assertEquals(response.status, 202); + await capture.settle(); + assertEquals(queueStartupWarnings(records, listenError).length, 1); + assertEquals(capture.reasons, []); + + const second = await post("https://example.com/activities/queued-2"); + assertEquals(second.status, 202); + await capture.settle(); + assertEquals( + queueStartupWarnings(records, listenError).length, + 1, + "the warning must be logged only once per federation instance", + ); + assertEquals(capture.reasons, []); + }); + + await t.step("asynchronous failure", async () => { + const asyncError = new TypeError("simulated async listen() failure"); + let rejectListen: ((error: unknown) => void) | undefined; + const asyncQueue: MessageQueue = { + enqueue() { + return Promise.resolve(); + }, + listen() { + return new Promise((_resolve, reject) => { + rejectListen = reject; + }); + }, + }; + const asyncFederation = createFederation({ + kv: new MemoryKvStore(), + queue: asyncQueue, + skipSignatureVerification: true, + documentLoaderFactory: () => mockDocumentLoader, + contextLoaderFactory: () => mockDocumentLoader, + }); + asyncFederation.setActorDispatcher( + "/actors/{identifier}", + () => new Person({ id: new URL("https://example.com/actor") }), + ); + asyncFederation.setInboxListeners( + "/actors/{identifier}/inbox", + "/inbox", + ).on(Create, () => {}); + const response = await asyncFederation.fetch( + new Request("https://example.com/inbox", { + method: "POST", + headers: { "Content-Type": "application/activity+json" }, + body: JSON.stringify({ + "@context": "https://www.w3.org/ns/activitystreams", + id: "https://example.com/activities/queued-async", + type: "Create", + actor: "https://example.com/actor", + }), + }), + { contextData: undefined }, + ); + assertEquals(response.status, 202); + await capture.settle(); + assertEquals(queueStartupWarnings(records, asyncError).length, 0); + assert(rejectListen != null); + rejectListen(asyncError); + await capture.settle(); + assertEquals( + queueStartupWarnings(records, asyncError).length, + 1, + "a listener that rejects after startup must still be reported", + ); + assertEquals(capture.reasons, []); + }); + } finally { + await capture.settle(); + capture.stop(); + await reset(); + } +}); + +test("Federation.startQueue() rejects when a queue cannot listen", async () => { + const listenError = new TypeError("simulated listen() failure"); + const federation = createFederation({ + kv: new MemoryKvStore(), + queue: { + enqueue() { + return Promise.resolve(); + }, + listen() { + throw listenError; + }, + }, + manuallyStartQueue: true, + }); + const error = await assertRejects(() => federation.startQueue(undefined)); + assertStrictEquals(error, listenError); +}); + +test("Federation.startQueue() keeps other queue workers running", async () => { + const outboxError = new TypeError("simulated outbox listen() failure"); + const lateInboxError = new TypeError("simulated late inbox failure"); + let rejectInbox: ((error: unknown) => void) | undefined; + let fanoutStarted = false; + const inbox: MessageQueue = { + enqueue() { + return Promise.resolve(); + }, + listen() { + return new Promise((_resolve, reject) => { + rejectInbox = reject; + }); + }, + }; + const outbox: MessageQueue = { + enqueue() { + return Promise.resolve(); + }, + listen() { + throw outboxError; + }, + }; + const fanout: MessageQueue = { + enqueue() { + return Promise.resolve(); + }, + listen() { + fanoutStarted = true; + return new Promise(() => {}); + }, + }; + const capture = captureUnhandledRejections(); + try { + const federation = createFederation({ + kv: new MemoryKvStore(), + queue: { inbox, outbox, fanout }, + manuallyStartQueue: true, + }); + const error = await assertRejects(() => federation.startQueue(undefined)); + assertStrictEquals(error, outboxError); + assert( + fanoutStarted, + "a synchronous throw from one queue must not prevent later queues from starting", + ); + assert(rejectInbox != null); + rejectInbox(lateInboxError); + await capture.settle(); + assertEquals( + capture.reasons, + [], + "an earlier listener rejection must remain handled after startup fails", + ); + } finally { + await capture.settle(); + capture.stop(); + } +}); + +test("ContextImpl.sendActivity() reports automatic queue startup failures", async (t) => { + const sender = [{ + privateKey: ed25519PrivateKey, + keyId: ed25519Multikey.id!, + }]; + const recipient = { + id: new URL("https://example.com/recipient"), + inboxId: new URL("https://example.com/inbox"), + }; + const createActivity = () => + new Create({ + id: new URL("https://example.com/activities/queue-startup-send"), + actor: new URL("https://example.com/actor"), + }); + + await t.step("outbox", async () => { + const listenError = new TypeError("simulated outbox listen() failure"); + const messages: Message[] = []; + const queue: MessageQueue = { + enqueue(message) { + messages.push(message as Message); + return Promise.resolve(); + }, + listen() { + throw listenError; + }, + }; + const capture = captureUnhandledRejections(); + const records: LogRecord[] = []; + await configureQueueStartupWarningCapture(records); + try { + const federation = new FederationImpl({ + kv: new MemoryKvStore(), + queue: { outbox: queue }, + documentLoaderFactory: () => mockDocumentLoader, + contextLoaderFactory: () => mockDocumentLoader, + }); + const ctx = new ContextImpl({ + data: undefined, + federation, + url: new URL("https://example.com/"), + documentLoader: mockDocumentLoader, + contextLoader: mockDocumentLoader, + }); + await ctx.sendActivity(sender, recipient, createActivity()); + assertEquals(messages.map((message) => message.type), ["outbox"]); + await capture.settle(); + assertEquals(queueStartupWarnings(records, listenError).length, 1); + assertEquals(capture.reasons, []); + } finally { + await capture.settle(); + capture.stop(); + await reset(); + } + }); + + await t.step("fanout", async () => { + const listenError = new TypeError("simulated fanout listen() failure"); + const messages: Message[] = []; + const outbox: MessageQueue = { + enqueue() { + return Promise.resolve(); + }, + listen() { + return Promise.resolve(); + }, + }; + const fanout: MessageQueue = { + enqueue(message) { + messages.push(message as Message); + return Promise.resolve(); + }, + listen() { + throw listenError; + }, + }; + const capture = captureUnhandledRejections(); + const records: LogRecord[] = []; + await configureQueueStartupWarningCapture(records); + try { + const federation = new FederationImpl({ + kv: new MemoryKvStore(), + queue: { outbox, fanout }, + documentLoaderFactory: () => mockDocumentLoader, + contextLoaderFactory: () => mockDocumentLoader, + }); + const ctx = new ContextImpl({ + data: undefined, + federation, + url: new URL("https://example.com/"), + documentLoader: mockDocumentLoader, + contextLoader: mockDocumentLoader, + }); + await ctx.sendActivity(sender, recipient, createActivity(), { + fanout: "force", + }); + assertEquals(messages.map((message) => message.type), ["fanout"]); + await capture.settle(); + assertEquals(queueStartupWarnings(records, listenError).length, 1); + assertEquals(capture.reasons, []); + } finally { + await capture.settle(); + capture.stop(); + await reset(); + } + }); +}); + test("FederationImpl.processQueuedTask()", async (t) => { await t.step("with MessageQueue having nativeRetrial", async () => { const kv = new MemoryKvStore(); diff --git a/packages/fedify/src/federation/middleware.ts b/packages/fedify/src/federation/middleware.ts index df299e728..af3b83b78 100644 --- a/packages/fedify/src/federation/middleware.ts +++ b/packages/fedify/src/federation/middleware.ts @@ -504,6 +504,31 @@ export class FederationImpl return this.tracerProvider.getTracer(metadata.name, metadata.version); } + /** + * Starts listening on a queue, converting a synchronous throw into + * a rejected promise. + * + * `MessageQueue.listen()` is documented to never reject, but some + * implementations (e.g., `WorkersMessageQueue`) throw synchronously. Without + * this normalization, one such throw aborts {@link _startQueueInternal} while + * it is still collecting listener promises, so listeners started earlier are + * never awaited and later queues are never started. + */ + #listenQueue( + queue: MessageQueue, + ctxData: TContextData, + signal: AbortSignal | undefined, + ): Promise { + try { + return queue.listen( + (msg) => this.processQueuedTask(ctxData, msg), + { signal }, + ); + } catch (error) { + return Promise.reject(error); + } + } + async _startQueueInternal( ctxData: TContextData, signal?: AbortSignal, @@ -518,12 +543,7 @@ export class FederationImpl ) { logger.debug("Starting an inbox task worker."); this.inboxQueueStarted = true; - promises.push( - this.inboxQueue.listen( - (msg) => this.processQueuedTask(ctxData, msg), - { signal }, - ), - ); + promises.push(this.#listenQueue(this.inboxQueue, ctxData, signal)); } if ( this.outboxQueue != null && @@ -533,12 +553,7 @@ export class FederationImpl ) { logger.debug("Starting an outbox task worker."); this.outboxQueueStarted = true; - promises.push( - this.outboxQueue.listen( - (msg) => this.processQueuedTask(ctxData, msg), - { signal }, - ), - ); + promises.push(this.#listenQueue(this.outboxQueue, ctxData, signal)); } if ( this.fanoutQueue != null && @@ -549,16 +564,29 @@ export class FederationImpl ) { logger.debug("Starting a fanout task worker."); this.fanoutQueueStarted = true; - promises.push( - this.fanoutQueue.listen( - (msg) => this.processQueuedTask(ctxData, msg), - { signal }, - ), - ); + promises.push(this.#listenQueue(this.fanoutQueue, ctxData, signal)); } await Promise.all(promises); } + /** + * Starts the queue workers automatically, reporting a failure instead of + * leaving the returned promise unobserved. + * + * Automatic startup is best-effort: call sites cannot await it, so a failing + * listener (e.g., `WorkersMessageQueue`, whose `listen()` throws) must not + * become an unhandled rejection. {@link startQueue} keeps rejecting for + * callers that start workers deliberately. + */ + _startQueueAutomatically(ctxData: TContextData): void { + this._startQueueInternal(ctxData).catch((error) => { + getLogger(["fedify", "federation", "queue"]).warn( + "Failed to start queue workers automatically: {error}", + { error }, + ); + }); + } + processQueuedTask( contextData: TContextData, message: Message, @@ -1483,7 +1511,7 @@ export class FederationImpl const privateKeyJwk = await exportJwk(privateKey); keyJwkPairs.push({ keyId: keyId.href, privateKey: privateKeyJwk }); } - if (!this.manuallyStartQueue) this._startQueueInternal(ctx.data); + if (!this.manuallyStartQueue) this._startQueueAutomatically(ctx.data); const carrier: Record = {}; propagation.inject(context.active(), carrier); const messages: { message: OutboxMessage; orderingKey?: string }[] = []; @@ -1781,7 +1809,9 @@ export class FederationImpl }); } } - if (!this.manuallyStartQueue) this._startQueueInternal(contextData); + if (!this.manuallyStartQueue) { + this._startQueueAutomatically(contextData); + } const inboxContextFactory = context.toInboxContext.bind(context) as & typeof context.toInboxContext & { @@ -2764,7 +2794,7 @@ export class ContextImpl implements Context { traceContext: carrier, }; if (!this.federation.manuallyStartQueue) { - this.federation._startQueueInternal(this.data); + this.federation._startQueueAutomatically(this.data); } await this.federation.fanoutQueue.enqueue( message,