Add native retrial support to AmqpMessageQueue

- Add nativeRetrial option to AmqpMessageQueueOptions
- When enabled, messages are not acknowledged on processing failure
- Update documentation and examples in README
- Add comprehensive tests for retrial behavior
- Update dependencies to support new functionality

fedify-dev/fedify#250

Author: dahlia

README.md

@@ -30,6 +30,17 @@ const federation = createFederation({
 });
 ~~~~
 
+The `AmqpMessageQueue` constructor accepts options as the second
+parameter, which can be used to configure the message queue:
+
+~~~~ typescript
+new AmqpMessageQueue(await connect("amqp://localhost"), {
+  queue: "my_queue",
+})
+~~~~
+
+For more details, please refer to the docs of [`AmqpMessageQueueOptions`].
+
 [JSR]: https://jsr.io/@fedify/amqp
 [JSR badge]: https://jsr.io/badges/@fedify/amqp
 [npm]: https://www.npmjs.com/package/@fedify/amqp
@@ -41,6 +52,7 @@ const federation = createFederation({
 [`KvStore`]: https://jsr.io/@fedify/fedify/doc/federation/~/KvStore
 [`MessageQueue`]: https://jsr.io/@fedify/fedify/doc/federation/~/MessageQueue
 [`AmqpMessageQueue`]: https://jsr.io/@fedify/amqp/doc/mq/~/AmqpMessageQueue
+[`AmqpMessageQueueOptions`]: https://jsr.io/@fedify/amqp/doc/mq/~/AmqpMessageQueueOptions
 
 
 Installation
@@ -72,6 +84,9 @@ Changelog
 
 To be released.
 
+ -  Added `nativeRetrial` option to `AmqpMessageQueueOptions` to enable
+    native retrial of messages.
+
  -  The type of the `AmqpMessageQueue()` constructor's first parameter has been
     changed from `Connection` to `ChannelModel`.
 

deno.json

@@ -7,8 +7,8 @@
     "./mq": "./src/mq.ts"
   },
   "imports": {
-    "@alinea/suite": "jsr:@alinea/suite@^0.6.2",
-    "@fedify/fedify": "jsr:@fedify/fedify@^1.5.0",
+    "@fedify/fedify": "jsr:@fedify/fedify@^1.7.0-dev.887+013dabb5",
+    "@hongminhee/suite": "jsr:@hongminhee/suite@^0.6.3",
     "@std/assert": "jsr:@std/assert@^1.0.13",
     "@std/async": "jsr:@std/async@^1.0.13",
     "amqplib": "npm:amqplib@^0.10.8",

deno.lock

@@ -35,19 +35,19 @@
     },
     "./package.json": "./package.json"
   },
+  "peerDependencies": {
+    "@fedify/fedify": "^1.7.0-dev.887",
+    "amqplib": "^0.10.8"
+  },
   "devDependencies": {
-    "@alinea/suite": "^0.6.2",
+    "@hongminhee/suite": "^0.6.3",
     "@js-temporal/polyfill": "^0.5.1",
     "@std/assert": "jsr:^1.0.13",
     "@std/async": "jsr:^1.0.13",
     "@types/amqplib": "^0.10.7",
     "tsdown": "^0.12.7",
     "typescript": "^5.8.3"
   },
-  "peerDependencies": {
-    "@fedify/fedify": "^1.5.0",
-    "amqplib": "^0.10.8"
-  },
   "scripts": {
     "build": "tsdown",
     "prepack": "tsdown",

package.json

@@ -1,6 +1,6 @@
-import { suite } from "@alinea/suite";
+import { suite } from "@hongminhee/suite";
 import * as temporal from "@js-temporal/polyfill";
-import { assertEquals, assertGreater } from "@std/assert";
+import { assert, assertEquals, assertFalse, assertGreater } from "@std/assert";
 import { delay } from "@std/async/delay";
 // @deno-types="npm:@types/amqplib"
 import { type ChannelModel, connect } from "amqplib";
@@ -21,13 +21,16 @@ function getConnection(): Promise<ChannelModel> {
   return connect(url ?? "amqp://localhost");
 }
 
-test("AmqpMessageQueue", async () => {
+test("AmqpMessageQueue", {
+  sanitizeOps: false,
+  sanitizeExit: false,
+  sanitizeResources: false,
+}, async () => {
   const conn = await getConnection();
   const conn2 = await getConnection();
-  const queue = `fedify_queue_${Math.random().toString(36).slice(5)}`;
-  const delayedQueuePrefix = `fedify_delayed_${
-    Math.random().toString(36).slice(5)
-  }_`;
+  const randomSuffix = Math.random().toString(36).substring(2);
+  const queue = `fedify_queue_${randomSuffix}`;
+  const delayedQueuePrefix = `fedify_delayed_${randomSuffix}_`;
   const mq = new AmqpMessageQueue(conn, { queue, delayedQueuePrefix });
   const mq2 = new AmqpMessageQueue(conn2, { queue, delayedQueuePrefix });
 
@@ -99,6 +102,76 @@ test("AmqpMessageQueue", async () => {
   await conn2.close();
 });
 
+test(
+  "AmqpMessageQueue [nativeRetrial: false]",
+  { sanitizeOps: false, sanitizeExit: false, sanitizeResources: false },
+  async () => {
+    const conn = await getConnection();
+    const randomSuffix = Math.random().toString(36).substring(2);
+    const queue = `fedify_queue_${randomSuffix}`;
+    const delayedQueuePrefix = `fedify_delayed_${randomSuffix}_`;
+    const mq = new AmqpMessageQueue(conn, { queue, delayedQueuePrefix });
+    assertFalse(mq.nativeRetrial);
+
+    const controller = new AbortController();
+    let i = 0;
+    const listening = mq.listen((message: string) => {
+      if (message !== "Hello, world!") return;
+      if (i++ < 1) {
+        throw new Error("Test error to check native retrial");
+      }
+    }, { signal: controller.signal });
+
+    await mq.enqueue("Hello, world!");
+
+    await waitFor(() => i >= 1, 15_000);
+    assertEquals(i, 1);
+    await delay(5_000);
+
+    controller.abort();
+    await listening;
+    await conn.close();
+
+    assertEquals(i, 1);
+  },
+);
+
+test(
+  "AmqpMessageQueue [nativeRetrial: true]",
+  { sanitizeOps: false, sanitizeExit: false, sanitizeResources: false },
+  async () => {
+    const conn = await getConnection();
+    const randomSuffix = Math.random().toString(36).substring(2);
+    const queue = `fedify_queue_${randomSuffix}`;
+    const delayedQueuePrefix = `fedify_delayed_${randomSuffix}_`;
+    const mq = new AmqpMessageQueue(conn, {
+      queue,
+      delayedQueuePrefix,
+      nativeRetrial: true,
+    });
+    assert(mq.nativeRetrial);
+
+    const controller = new AbortController();
+    let i = 0;
+    const listening = mq.listen((message: string) => {
+      if (message !== "Hello, world!") return;
+      if (i++ < 1) {
+        throw new Error("Test error to check native retrial");
+      }
+    }, { signal: controller.signal });
+
+    await mq.enqueue("Hello, world!");
+
+    await waitFor(() => i > 1, 15_000);
+
+    controller.abort();
+    await listening;
+    await conn.close();
+
+    assertGreater(i, 1);
+  },
+);
+
 async function waitFor(
   predicate: () => boolean,
   timeoutMs: number,