From 061f0adce452e72e346bf5e3241d6ff510742508 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Thu, 5 Nov 2026 23:15:34 +0900 Subject: [PATCH] Start the fanout worker when only fanout is queued FederationImpl._startQueueInternal() began with a guard that only checked the inbox and outbox queues. The fanout queue was added in 1.5.0, but the guard was not updated, so a federation configured with only queue: { fanout: ... } returned before reaching the fanout branch. ContextImpl.sendActivityInternal() still enqueued a FanoutMessage, and because nothing called fanoutQueue.listen(), the messages stayed in the backend unprocessed. The failure looked intermittent because adding an inbox or outbox queue started the fanout worker as a side effect, and non-forced sends with fewer than five inboxes bypass the fanout queue entirely. Nothing was logged either, because the early return preceded the logger. Include fanoutQueue in the guard, and add regression tests for an explicit startQueue(..., { queue: "fanout" }), a no-selector startQueue(), and the implicit startup after a forced fanout enqueue. A manual-start test asserts that sending still enqueues without starting a worker. https://github.com/fedify-dev/fedify/issues/1389 Assisted-by: OpenCode:deepseek-flash Assisted-by: Codex:gpt-6-astra Assisted-by: Claude Code:claude-fable-5-1 Assisted-by: OpenCode:mimo-v2.6-flash-free --- CHANGES.md | 9 ++ changes.d/fedify/fanout-only-queue-worker.md | 12 ++ .../fedify/src/federation/middleware.test.ts | 126 ++++++++++++++++++ packages/fedify/src/federation/middleware.ts | 7 +- 4 files changed, 153 insertions(+), 1 deletion(-) create mode 100644 changes.d/fedify/fanout-only-queue-worker.md diff --git a/CHANGES.md b/CHANGES.md index 2d76a3c9a..f6b3dccee 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -30,10 +30,19 @@ To be released. `2024-02-31T10:00.5Z`, are now treated as malformed instead of being accepted with `closed` discarded. [[#1339], [#1347] by ChanHaeng Lee\] + - Fixed the fanout worker never starting when the `fanout` queue was the only + configured queue. `Federation.startQueue()` returned without calling + `fanoutQueue.listen()`, so fanout messages were enqueued but never + consumed; the implicit startup triggered by a fanout enqueue silently did + nothing. Both paths now start the fanout worker, and deployments that + configure only `queue: { fanout: ... }` no longer need to add an unused + `inbox` or `outbox` queue to make fanout delivery work. [[#1389], [#1392]] [#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 +[#1389]: https://github.com/fedify-dev/fedify/issues/1389 +[#1392]: https://github.com/fedify-dev/fedify/pull/1392 ### @fedify/vocab diff --git a/changes.d/fedify/fanout-only-queue-worker.md b/changes.d/fedify/fanout-only-queue-worker.md new file mode 100644 index 000000000..06d931742 --- /dev/null +++ b/changes.d/fedify/fanout-only-queue-worker.md @@ -0,0 +1,12 @@ +--- +links: + '#1389': https://github.com/fedify-dev/fedify/issues/1389 + '#1392': https://github.com/fedify-dev/fedify/pull/1392 +--- + - Fixed the fanout worker never starting when the `fanout` queue was the only + configured queue. `Federation.startQueue()` returned without calling + `fanoutQueue.listen()`, so fanout messages were enqueued but never + consumed; the implicit startup triggered by a fanout enqueue silently did + nothing. Both paths now start the fanout worker, and deployments that + configure only `queue: { fanout: ... }` no longer need to add an unused + `inbox` or `outbox` queue to make fanout delivery work. [[#1389], [#1392]] diff --git a/packages/fedify/src/federation/middleware.test.ts b/packages/fedify/src/federation/middleware.test.ts index c136f5339..7b668b436 100644 --- a/packages/fedify/src/federation/middleware.test.ts +++ b/packages/fedify/src/federation/middleware.test.ts @@ -200,6 +200,132 @@ test("createFederation()", async (t) => { }); }); +test("FederationImpl.startQueue()", async (t) => { + function createCountingQueue() { + const messages: Message[] = []; + let listenCount = 0; + const queue: MessageQueue & { + messages: Message[]; + readonly listenCount: number; + } = { + messages, + get listenCount() { + return listenCount; + }, + enqueue(message: Message) { + messages.push(message); + return Promise.resolve(); + }, + listen() { + listenCount++; + return Promise.resolve(); + }, + }; + return queue; + } + + await t.step( + "starts a fanout-only queue with an explicit selector", + async () => { + const queue = createCountingQueue(); + const federation = new FederationImpl({ + kv: new MemoryKvStore(), + queue: { fanout: queue }, + manuallyStartQueue: true, + }); + await federation.startQueue(undefined, { queue: "fanout" }); + assertEquals(queue.listenCount, 1); + await federation.startQueue(undefined, { queue: "fanout" }); + assertEquals(queue.listenCount, 1); + }, + ); + + await t.step( + "starts a fanout-only queue without a selector", + async () => { + const queue = createCountingQueue(); + const federation = new FederationImpl({ + kv: new MemoryKvStore(), + queue: { fanout: queue }, + manuallyStartQueue: true, + }); + await federation.startQueue(undefined); + assertEquals(queue.listenCount, 1); + }, + ); + + await t.step( + "starts a fanout-only queue after a fanout enqueue", + async () => { + const queue = createCountingQueue(); + const federation = new FederationImpl({ + kv: new MemoryKvStore(), + contextLoaderFactory: () => mockDocumentLoader, + queue: { fanout: queue }, + }); + const ctx = new ContextImpl({ + data: undefined, + federation, + url: new URL("https://example.com/"), + documentLoader: mockDocumentLoader, + contextLoader: mockDocumentLoader, + }); + const activity = new vocab.Create({ + id: new URL("https://example.com/activity/1"), + actor: new URL("https://example.com/person"), + }); + await ctx.sendActivity( + [{ privateKey: rsaPrivateKey2, keyId: rsaPublicKey2.id! }], + { + id: new URL("https://example.com/recipient"), + inboxId: new URL("https://example.com/inbox"), + }, + activity, + { fanout: "force" }, + ); + assertEquals(queue.listenCount, 1); + assertEquals(queue.messages.length, 1); + assertEquals(queue.messages[0].type, "fanout"); + }, + ); + + await t.step( + "does not start a fanout-only queue automatically when manually started", + async () => { + const queue = createCountingQueue(); + const federation = new FederationImpl({ + kv: new MemoryKvStore(), + contextLoaderFactory: () => mockDocumentLoader, + queue: { fanout: queue }, + manuallyStartQueue: true, + }); + const ctx = new ContextImpl({ + data: undefined, + federation, + url: new URL("https://example.com/"), + documentLoader: mockDocumentLoader, + contextLoader: mockDocumentLoader, + }); + const activity = new vocab.Create({ + id: new URL("https://example.com/activity/1"), + actor: new URL("https://example.com/person"), + }); + await ctx.sendActivity( + [{ privateKey: rsaPrivateKey2, keyId: rsaPublicKey2.id! }], + { + id: new URL("https://example.com/recipient"), + inboxId: new URL("https://example.com/inbox"), + }, + activity, + { fanout: "force" }, + ); + assertEquals(queue.listenCount, 0); + assertEquals(queue.messages.length, 1); + assertEquals(queue.messages[0].type, "fanout"); + }, + ); +}); + test({ name: "Federation.createContext()", permissions: { env: true, read: true }, diff --git a/packages/fedify/src/federation/middleware.ts b/packages/fedify/src/federation/middleware.ts index df299e728..b2d51dc27 100644 --- a/packages/fedify/src/federation/middleware.ts +++ b/packages/fedify/src/federation/middleware.ts @@ -509,7 +509,12 @@ export class FederationImpl signal?: AbortSignal, queue?: keyof FederationQueueOptions, ): Promise { - if (this.inboxQueue == null && this.outboxQueue == null) return; + if ( + this.inboxQueue == null && this.outboxQueue == null && + this.fanoutQueue == null + ) { + return; + } const logger = getLogger(["fedify", "federation", "queue"]); const promises: Promise[] = []; if (