Start the fanout worker when only the fanout queue is configured

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
 

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]]

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<void>({
+        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<void>({
+        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<void>({
+        kv: new MemoryKvStore(),
+        contextLoaderFactory: () => mockDocumentLoader,
+        queue: { fanout: queue },
+      });
+      const ctx = new ContextImpl<void>({
+        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<void>({
+        kv: new MemoryKvStore(),
+        contextLoaderFactory: () => mockDocumentLoader,
+        queue: { fanout: queue },
+        manuallyStartQueue: true,
+      });
+      const ctx = new ContextImpl<void>({
+        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 },

packages/fedify/src/federation/middleware.ts

@@ -509,7 +509,12 @@ export class FederationImpl<TContextData>
     signal?: AbortSignal,
     queue?: keyof FederationQueueOptions,
   ): Promise<void> {
-    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<void>[] = [];
     if (