fix(ext/kv): send queue wake messages accross different kv instances (#20465)

fixes #20454

Current KV queues implementation assumes that `enqueue` and
`listenQueue` are called on the same instance of `Deno.Kv`. It's
possible that the same Deno process opens multiple KV instances pointing
to the same fs path, and in that case `listenQueue` should still get
notified of messages enqueued through a different KV instance.

1 files changed, 34 insertions, 0 deletions

diff --git a/cli/tests/unit/kv_test.ts b/cli/tests/unit/kv_test.ts
index 25abf4cde..7cb8ebccf 100644
--- a/cli/tests/unit/kv_test.ts
+++ b/cli/tests/unit/kv_test.ts
@@ -1820,6 +1820,40 @@ Deno.test({
 });
 
 Deno.test({
+  name: "different kv instances for enqueue and queueListen",
+  async fn() {
+    const filename = await Deno.makeTempFile({ prefix: "queue_db" });
+    try {
+      const db0 = await Deno.openKv(filename);
+      const db1 = await Deno.openKv(filename);
+      const promise = deferred();
+      let dequeuedMessage: unknown = null;
+      const listener = db0.listenQueue((msg) => {
+        dequeuedMessage = msg;
+        promise.resolve();
+      });
+      try {
+        const res = await db1.enqueue("test");
+        assert(res.ok);
+        assertNotEquals(res.versionstamp, null);
+        await promise;
+        assertEquals(dequeuedMessage, "test");
+      } finally {
+        db0.close();
+        await listener;
+        db1.close();
+      }
+    } finally {
+      try {
+        await Deno.remove(filename);
+      } catch {
+        // pass
+      }
+    }
+  },
+});
+
+Deno.test({
   name: "queue graceful close",
   async fn() {
     const db: Deno.Kv = await Deno.openKv(":memory:");