fix(ext/kv): throw error if already closed (#21459)

If KV is closed and tries to listen queue should throw an error

closes #20991

cli/tests/integration/js_unit_tests.rs

@@ -48,6 +48,7 @@ util::unit_test_factory!(
     jupyter_test,
     kv_test,
     kv_queue_test_no_db_close,
+    kv_queue_test,
     kv_queue_undelivered_test,
     link_test,
     make_temp_test,

cli/tests/unit/kv_queue_test.ts

@@ -0,0 +1,13 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+import { assertEquals, assertFalse } from "./test_util.ts";
+
+Deno.test({}, async function queueTestDbClose() {
+  const db: Deno.Kv = await Deno.openKv(":memory:");
+  db.close();
+  try {
+    await db.listenQueue(() => {});
+    assertFalse(false);
+  } catch (e) {
+    assertEquals(e.message, "already closed");
+  }
+});

ext/kv/01_db.ts

@@ -64,6 +64,7 @@ const kvSymbol = Symbol("KvRid");
 
 class Kv {
   #rid: number;
+  #isClosed: boolean;
 
   constructor(rid: number = undefined, symbol: symbol = undefined) {
     if (kvSymbol !== symbol) {
@@ -72,6 +73,7 @@ class Kv {
       );
     }
     this.#rid = rid;
+    this.#isClosed = false;
   }
 
   atomic() {
@@ -251,6 +253,9 @@ class Kv {
   async listenQueue(
     handler: (message: unknown) => Promise<void> | void,
   ): Promise<void> {
+    if (this.#isClosed) {
+      throw new Error("already closed");
+    }
     const finishMessageOps = new Map<number, Promise<void>>();
     while (true) {
       // Wait for the next message.
@@ -366,6 +371,7 @@ class Kv {
 
   close() {
     core.close(this.#rid);
+    this.#isClosed = true;
   }
 
   [SymbolDispose]() {