fix(ext/kv): retry transaction on `SQLITE_BUSY` errors (#20189)

Properly handle the `SQLITE_BUSY` error code by retrying the
transaction.

Also wraps database initialization logic in a transaction to protect
against incomplete/concurrent initializations.

Fixes https://github.com/denoland/deno/issues/20116.

1 files changed, 50 insertions, 0 deletions

diff --git a/cli/tests/unit/kv_test.ts b/cli/tests/unit/kv_test.ts
index 3081917da..74a8ed6b3 100644
--- a/cli/tests/unit/kv_test.ts
+++ b/cli/tests/unit/kv_test.ts
@@ -1756,3 +1756,53 @@ dbTest("atomic operation is exposed", (db) => {
   const ao = db.atomic();
   assert(ao instanceof Deno.AtomicOperation);
 });
+
+Deno.test({
+  name: "racy open",
+  async fn() {
+    for (let i = 0; i < 100; i++) {
+      const filename = await Deno.makeTempFile({ prefix: "racy_open_db" });
+      try {
+        const [db1, db2, db3] = await Promise.all([
+          Deno.openKv(filename),
+          Deno.openKv(filename),
+          Deno.openKv(filename),
+        ]);
+        db1.close();
+        db2.close();
+        db3.close();
+      } finally {
+        await Deno.remove(filename);
+      }
+    }
+  },
+});
+
+Deno.test({
+  name: "racy write",
+  async fn() {
+    const filename = await Deno.makeTempFile({ prefix: "racy_write_db" });
+    const concurrency = 20;
+    const iterations = 5;
+    try {
+      const dbs = await Promise.all(
+        Array(concurrency).fill(0).map(() => Deno.openKv(filename)),
+      );
+      try {
+        for (let i = 0; i < iterations; i++) {
+          await Promise.all(
+            dbs.map((db) => db.atomic().sum(["counter"], 1n).commit()),
+          );
+        }
+        assertEquals(
+          ((await dbs[0].get(["counter"])).value as Deno.KvU64).value,
+          BigInt(concurrency * iterations),
+        );
+      } finally {
+        dbs.forEach((db) => db.close());
+      }
+    } finally {
+      await Deno.remove(filename);
+    }
+  },
+});