diff --git a/CHANGES.md b/CHANGES.md index 30d4d6eae..2c6319d47 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -49,6 +49,13 @@ 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 @@ -56,7 +63,9 @@ To be released. [#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 +[#1389]: https://github.com/fedify-dev/fedify/issues/1389 [#1390]: https://github.com/fedify-dev/fedify/pull/1390 +[#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 7861aee3f..d47521ae2 100644 --- a/packages/fedify/src/federation/middleware.test.ts +++ b/packages/fedify/src/federation/middleware.test.ts @@ -202,6 +202,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 eff84d922..f78da81ba 100644 --- a/packages/fedify/src/federation/middleware.ts +++ b/packages/fedify/src/federation/middleware.ts @@ -534,7 +534,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 (