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,