From e0bb5d3d0c9b03cdad86610b9cf827975e1dacbe Mon Sep 17 00:00:00 2001 From: Jeff Repanich Date: Sun, 16 Aug 2026 09:22:04 -0400 Subject: [PATCH 1/2] fix: serialize sqlite concurrency --- src/sqlite.test.ts | 61 ++++++++++++++++++++++++++++++++++++++++++++++ src/sqlite.ts | 38 +++++++++++++++++++++++++---- 2 files changed, 94 insertions(+), 5 deletions(-) diff --git a/src/sqlite.test.ts b/src/sqlite.test.ts index 469d3fd..ac81b7e 100644 --- a/src/sqlite.test.ts +++ b/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,64 @@ 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"}`; diff --git a/src/sqlite.ts b/src/sqlite.ts index 63ffc41..c493f9d 100644 --- a/src/sqlite.ts +++ b/src/sqlite.ts @@ -19,9 +19,12 @@ export interface SqliteOptions { interface QueueContext { readonly adapter: SqliteAdapter; + readonly nestedQueue: SqliteQueue; } const context = new AsyncLocalStorage(); +const sharedQueues = new Map(); + 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,16 @@ class SqliteAdapter implements DatabaseAdapter { callback: (adapter: DatabaseAdapter) => Promise, _options?: TransactionOptions, ): Promise { - 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 +140,9 @@ class SqliteAdapter implements DatabaseAdapter { session(callback: (adapter: DatabaseAdapter) => Promise): Promise { 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(callback: (adapter: DatabaseAdapter) => Promise): Promise { @@ -154,7 +167,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 +283,18 @@ 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:"); From 7ee33e286658d84ff10af863247fe2fc8a89b3f8 Mon Sep 17 00:00:00 2001 From: Jeff Repanich Date: Sun, 16 Aug 2026 09:38:24 -0400 Subject: [PATCH 2/2] style: format SQLite concurrency changes --- src/sqlite.test.ts | 10 ++++++---- src/sqlite.ts | 7 +++---- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/src/sqlite.test.ts b/src/sqlite.test.ts index ac81b7e..ab22a19 100644 --- a/src/sqlite.test.ts +++ b/src/sqlite.test.ts @@ -66,10 +66,12 @@ describe("SQLite dialect", () => { ).resolves.toBeUndefined(); expect( - (await adapter.execute<{ value: string }>({ - text: "SELECT value FROM values_table ORDER BY value", - values: [], - })).rows, + ( + 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?.(); }); diff --git a/src/sqlite.ts b/src/sqlite.ts index c493f9d..f41e7cf 100644 --- a/src/sqlite.ts +++ b/src/sqlite.ts @@ -118,9 +118,7 @@ class SqliteAdapter implements DatabaseAdapter { const current = context.getStore(); if (current?.adapter === this) { return current.nestedQueue.run(() => - context.run({ adapter: this, nestedQueue: new SqliteQueue() }, () => - this.nested(callback), - ), + context.run({ adapter: this, nestedQueue: new SqliteQueue() }, () => this.nested(callback)), ); } return this.queue.run(() => @@ -293,7 +291,8 @@ export function sqlite(options: SqliteOptions = {}): DatabaseDriver { if (released) return; released = true; shared.users -= 1; - if (shared.users === 0 && sharedQueues.get(identity) === shared) sharedQueues.delete(identity); + if (shared.users === 0 && sharedQueues.get(identity) === shared) + sharedQueues.delete(identity); }); }, async shadow() {