Fix SQLite concurrency isolation

src/sqlite.test.ts

@@ -1,3 +1,6 @@
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
 import { describe, expect, it } from "vitest";
 import { defineDatabase, defineQuery, table, text } from "./index";
 import { jsonb } from "./postgres";
@@ -39,6 +42,66 @@ describe("SQLite dialect", () => {
     await adapter.close?.();
   });
 
+  it("should serialize concurrent sibling savepoints", async () => {
+    const adapter = await sqlite({ filename: ":memory:" }).open();
+    await adapter.execute({ text: "CREATE TABLE values_table (value text)", values: [] });
+
+    await expect(
+      adapter.transaction(async (transaction) => {
+        await Promise.all([
+          transaction.transaction(async (nested) => {
+            await nested.execute({ text: "INSERT INTO values_table VALUES ($1)", values: ["a"] });
+            await nested.transaction(async (recursive) => {
+              await recursive.execute({
+                text: "INSERT INTO values_table VALUES ($1)",
+                values: ["nested"],
+              });
+            });
+          }),
+          transaction.transaction(async (nested) => {
+            await nested.execute({ text: "INSERT INTO values_table VALUES ($1)", values: ["b"] });
+          }),
+        ]);
+      }),
+    ).resolves.toBeUndefined();
+
+    expect(
+      (
+        await adapter.execute<{ value: string }>({
+          text: "SELECT value FROM values_table ORDER BY value",
+          values: [],
+        })
+      ).rows,
+    ).toEqual([{ value: "a" }, { value: "b" }, { value: "nested" }]);
+    await adapter.close?.();
+  });
+
+  it("should serialize adapters that target the same SQLite file", async () => {
+    const directory = await mkdtemp(path.join(tmpdir(), "askr-orm-sqlite-"));
+    const filename = path.join(directory, "shared.sqlite");
+    const first = await sqlite({ filename }).open();
+    const second = await sqlite({ filename }).open();
+    try {
+      await first.execute({ text: "CREATE TABLE values_table (value text)", values: [] });
+      await expect(
+        Promise.all([
+          first.transaction(async (transaction) => {
+            await transaction.execute({
+              text: "INSERT INTO values_table VALUES ($1)",
+              values: ["a"],
+            });
+            await new Promise((resolve) => setTimeout(resolve, 20));
+          }),
+          second.execute({ text: "INSERT INTO values_table VALUES ($1)", values: ["b"] }),
+        ]),
+      ).resolves.toBeDefined();
+    } finally {
+      await first.close?.();
+      await second.close?.();
+      await rm(directory, { recursive: true, force: true });
+    }
+  });
+
   it("should register explicit keyed SQL and reject PostgreSQL-only columns before open", async () => {
     const users = table("users", { id: text().primaryKey(), name: text().notNull() });
     const byId = defineQuery<{ id: string }>("users.byId")`SELECT * FROM users WHERE id = ${"id"}`;

src/sqlite.ts

@@ -19,9 +19,12 @@ export interface SqliteOptions {
 
 interface QueueContext {
   readonly adapter: SqliteAdapter;
+  readonly nestedQueue: SqliteQueue;
 }
 const context = new AsyncLocalStorage<QueueContext>();
 
+const sharedQueues = new Map<string, { queue: SqliteQueue; users: number }>();
+
 function sqliteQuery(query: SqlQuery): SqlQuery {
   return rewritePlaceholders(query.text, query.values, { sqlite: true });
 }
@@ -60,6 +63,7 @@ class SqliteAdapter implements DatabaseAdapter {
     private readonly database: DatabaseSync,
     identity: string,
     queue = new SqliteQueue(),
+    private readonly releaseQueue = () => undefined,
   ) {
     this.identity = identity;
     this.queue = queue;
@@ -111,9 +115,14 @@ class SqliteAdapter implements DatabaseAdapter {
     callback: (adapter: DatabaseAdapter) => Promise<T>,
     _options?: TransactionOptions,
   ): Promise<T> {
-    if (context.getStore()?.adapter === this) return this.nested(callback);
+    const current = context.getStore();
+    if (current?.adapter === this) {
+      return current.nestedQueue.run(() =>
+        context.run({ adapter: this, nestedQueue: new SqliteQueue() }, () => this.nested(callback)),
+      );
+    }
     return this.queue.run(() =>
-      context.run({ adapter: this }, async () => {
+      context.run({ adapter: this, nestedQueue: new SqliteQueue() }, async () => {
         this.database.exec("BEGIN");
         try {
           const result = await callback(this);
@@ -129,7 +138,9 @@ class SqliteAdapter implements DatabaseAdapter {
 
   session<T>(callback: (adapter: DatabaseAdapter) => Promise<T>): Promise<T> {
     if (context.getStore()?.adapter === this) return callback(this);
-    return this.queue.run(() => context.run({ adapter: this }, () => callback(this)));
+    return this.queue.run(() =>
+      context.run({ adapter: this, nestedQueue: new SqliteQueue() }, () => callback(this)),
+    );
   }
 
   migrationLock<T>(callback: (adapter: DatabaseAdapter) => Promise<T>): Promise<T> {
@@ -154,7 +165,11 @@ class SqliteAdapter implements DatabaseAdapter {
     await this.queue.run(async () => {
       if (this.closed) return;
       this.closed = true;
-      this.database.close();
+      try {
+        this.database.close();
+      } finally {
+        this.releaseQueue();
+      }
     });
   }
 }
@@ -266,7 +281,19 @@ export function sqlite(options: SqliteOptions = {}): DatabaseDriver {
     targetIdentity: identity,
     shadowIdentity: ":memory:",
     async open() {
-      return new SqliteAdapter(new DatabaseSync(filename), identity);
+      const database = new DatabaseSync(filename);
+      if (identity === ":memory:") return new SqliteAdapter(database, identity);
+      const shared = sharedQueues.get(identity) ?? { queue: new SqliteQueue(), users: 0 };
+      shared.users += 1;
+      sharedQueues.set(identity, shared);
+      let released = false;
+      return new SqliteAdapter(database, identity, shared.queue, () => {
+        if (released) return;
+        released = true;
+        shared.users -= 1;
+        if (shared.users === 0 && sharedQueues.get(identity) === shared)
+          sharedQueues.delete(identity);
+      });
     },
     async shadow() {
       return tooling(":memory:");