Message byte estimation and batching on @fedify/cfworkers

CHANGES.md

@@ -8,6 +8,18 @@ Version 2.0.23
 
 To be released.
 
+### @fedify/cfworkers
+
+ -  Fixed `WorkersMessageQueue.enqueueMany()` failing when the given messages
+    exceeded Cloudflare Queues' batch limits of 100 messages or 256 KB per
+    batch, which could happen when delivering activities to a large audience.
+    The method now estimates the serialized size of each message and splits
+    the messages into multiple `sendBatch()` calls that stay within the
+    limits.  [[#958], [#960] by SJang1]
+
+[#958]: https://github.com/fedify-dev/fedify/issues/958
+[#960]: https://github.com/fedify-dev/fedify/pull/960
+
 
 Version 2.0.22
 --------------

packages/cfworkers/src/mod.ts

@@ -93,6 +93,13 @@ interface WrappedMessage {
   readonly __fedify_payload__: any;
 }
 
+/**
+ * enqueueMany Limit settings.
+ */
+const MAX_BATCH_MESSAGES = 100;
+const MAX_ESTIMATED_BATCH_BYTES = 240_000;
+const ESTIMATED_METADATA_BYTES_PER_MESSAGE = 128;
+
 /**
  * Result from {@link WorkersMessageQueue.processMessage}.
  * @since 2.0.0
@@ -339,16 +346,57 @@ export class WorkersMessageQueue implements MessageQueue {
     messages: readonly any[],
     options?: MessageQueueEnqueueOptions,
   ): Promise<void> {
-    const requests: MessageSendRequest[] = messages.map((msg) => ({
-      body: {
+    const utf8Encoder = new TextEncoder();
+
+    const estimateMessageBytes = (body: WrappedMessage): number => {
+      const serialized = JSON.stringify(body);
+
+      if (serialized === undefined) {
+        throw new TypeError("Queue message must be JSON-serializable.");
+      }
+
+      return utf8Encoder.encode(serialized).byteLength +
+        ESTIMATED_METADATA_BYTES_PER_MESSAGE;
+    };
+
+    const delaySeconds = options?.delay?.total("seconds") ?? 0;
+
+    let batch: MessageSendRequest[] = [];
+    let estimatedBatchBytes = 0;
+
+    const flush = async (): Promise<void> => {
+      if (batch.length === 0) return;
+
+      await this.#queue.sendBatch(batch, { delaySeconds });
+
+      batch = [];
+      estimatedBatchBytes = 0;
+    };
+
+    for (const message of messages) {
+      const body = {
         __fedify_ordering_key__: options?.orderingKey,
-        __fedify_payload__: msg,
-      } satisfies WrappedMessage,
-      contentType: "json",
-    }));
-    await this.#queue.sendBatch(requests, {
-      delaySeconds: options?.delay?.total("seconds") ?? 0,
-    });
+        __fedify_payload__: message,
+      } satisfies WrappedMessage;
+
+      const request = {
+        body,
+        contentType: "json",
+      } satisfies MessageSendRequest;
+
+      const messageBytes = estimateMessageBytes(body);
+      const exceedsBatchLimit = batch.length >= MAX_BATCH_MESSAGES ||
+        estimatedBatchBytes + messageBytes > MAX_ESTIMATED_BATCH_BYTES;
+
+      if (batch.length > 0 && exceedsBatchLimit) {
+        await flush();
+      }
+
+      batch.push(request);
+      estimatedBatchBytes += messageBytes;
+    }
+
+    await flush();
   }
 
   /**

packages/cfworkers/test/mq.test.ts

@@ -131,6 +131,104 @@ describe("WorkersMessageQueue", () => {
     sendBatchSpy.mockRestore();
   });
 
+  it("enqueueMany() splits batches at the 100-message limit", async () => {
+    const sendBatchSpy = vi
+      .spyOn(env.Q1, "sendBatch")
+      .mockImplementation(async () => {});
+
+    const queue = new WorkersMessageQueue(env.Q1);
+    const messages = Array.from({ length: 101 }, (_, id) => ({ id }));
+    await queue.enqueueMany(messages);
+
+    expect(sendBatchSpy).toHaveBeenCalledTimes(2);
+    expect(
+      sendBatchSpy.mock.calls.map(([requests]) =>
+        Array.from(requests).length
+      ),
+    ).toEqual([100, 1]);
+
+    sendBatchSpy.mockRestore();
+  });
+
+  it.each([
+    ["single-byte ASCII", "a", 90_000],
+    ["multi-byte Unicode", "🙂", 22_500],
+  ])(
+    "enqueueMany() splits %s payloads by serialized UTF-8 size",
+    async (_label, character, repetitions) => {
+      const sendBatchSpy = vi
+        .spyOn(env.Q1, "sendBatch")
+        .mockImplementation(async () => {});
+
+      const queue = new WorkersMessageQueue(env.Q1);
+      const messages = Array.from({ length: 3 }, (_, id) => ({
+        id,
+        content: character.repeat(repetitions),
+      }));
+      await queue.enqueueMany(messages);
+
+      expect(sendBatchSpy).toHaveBeenCalledTimes(2);
+      expect(
+        sendBatchSpy.mock.calls.map(([requests]) =>
+          Array.from(requests).length
+        ),
+      ).toEqual([2, 1]);
+
+      const queuedIds = sendBatchSpy.mock.calls.flatMap(([requests]) =>
+        Array.from(requests, (request) => {
+          const body = request.body as {
+            __fedify_payload__: { id: number };
+          };
+          return body.__fedify_payload__.id;
+        })
+      );
+      expect(queuedIds).toEqual([0, 1, 2]);
+
+      sendBatchSpy.mockRestore();
+    },
+  );
+
+  it("enqueueMany() preserves options across split batches", async () => {
+    const sendBatchSpy = vi
+      .spyOn(env.Q1, "sendBatch")
+      .mockImplementation(async () => {});
+
+    const queue = new WorkersMessageQueue(env.Q1);
+    const messages = Array.from({ length: 3 }, (_, id) => ({
+      id,
+      content: "a".repeat(90_000),
+    }));
+    await queue.enqueueMany(messages, {
+      orderingKey: "activity:123",
+      delay: mockDuration(5) as unknown as Temporal.Duration,
+    });
+
+    expect(sendBatchSpy).toHaveBeenCalledTimes(2);
+    for (const [requests, options] of sendBatchSpy.mock.calls) {
+      expect(options).toEqual({ delaySeconds: 5 });
+      for (const request of requests) {
+        expect(request.body).toMatchObject({
+          __fedify_ordering_key__: "activity:123",
+        });
+      }
+    }
+
+    sendBatchSpy.mockRestore();
+  });
+
+  it("enqueueMany() does not send an empty batch", async () => {
+    const sendBatchSpy = vi
+      .spyOn(env.Q1, "sendBatch")
+      .mockImplementation(async () => {});
+
+    const queue = new WorkersMessageQueue(env.Q1);
+    await queue.enqueueMany([]);
+
+    expect(sendBatchSpy).not.toHaveBeenCalled();
+
+    sendBatchSpy.mockRestore();
+  });
+
   it("listen() throws TypeError", () => {
     const queue = new WorkersMessageQueue(env.Q1);
     expect(() => queue.listen(() => {})).toThrow(TypeError);