feat(ext/kv) add backoffSchedule to enqueue (#21474)

Also reduces the time to run `kv_queue_undelivered_test.ts` test from
100 seconds down to 3 seconds.

closes #21437

2 files changed, 37 insertions, 1 deletions

diff --git a/cli/tests/unit/kv_queue_undelivered_test.ts b/cli/tests/unit/kv_queue_undelivered_test.ts
index 2b1a2b0aa..27d9505b8 100644
--- a/cli/tests/unit/kv_queue_undelivered_test.ts
+++ b/cli/tests/unit/kv_queue_undelivered_test.ts
@@ -43,8 +43,9 @@ queueTest("queue with undelivered", async (db) => {
   try {
     await db.enqueue("test", {
       keysIfUndelivered: [["queue_failed", "a"], ["queue_failed", "b"]],
+      backoffSchedule: [10, 20],
     });
-    await sleep(100000);
+    await sleep(3000);
     const undelivered = await collect(db.list({ prefix: ["queue_failed"] }));
     assertEquals(undelivered.length, 2);
     assertEquals(undelivered[0].key, ["queue_failed", "a"]);
diff --git a/cli/tests/unit/kv_test.ts b/cli/tests/unit/kv_test.ts
index 68b3c4013..f46099ed1 100644
--- a/cli/tests/unit/kv_test.ts
+++ b/cli/tests/unit/kv_test.ts
@@ -1600,6 +1600,24 @@ queueTest("queue retries", async (db) => {
   assertEquals(4, count);
 });
 
+queueTest("queue retries with backoffSchedule", async (db) => {
+  let count = 0;
+  const listener = db.listenQueue((_msg) => {
+    count += 1;
+    throw new TypeError("dequeue error");
+  });
+  try {
+    await db.enqueue("test", { backoffSchedule: [1] });
+    await sleep(2000);
+  } finally {
+    db.close();
+    await listener;
+  }
+
+  // There should have been 1 attempt + 1 retry
+  assertEquals(2, count);
+});
+
 queueTest("multiple listenQueues", async (db) => {
   const numListens = 10;
   let count = 0;
@@ -1876,6 +1894,23 @@ Deno.test({
   },
 });
 
+dbTest("invalid backoffSchedule", async (db) => {
+  await assertRejects(
+    async () => {
+      await db.enqueue("foo", { backoffSchedule: [1, 1, 1, 1, 1, 1] });
+    },
+    TypeError,
+    "invalid backoffSchedule",
+  );
+  await assertRejects(
+    async () => {
+      await db.enqueue("foo", { backoffSchedule: [3600001] });
+    },
+    TypeError,
+    "invalid backoffSchedule",
+  );
+});
+
 dbTest("atomic operation is exposed", (db) => {
   assert(Deno.AtomicOperation);
   const ao = db.atomic();