diff --git a/src/sqlite.test.ts b/src/sqlite.test.ts index 469d3fd..ab22a19 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,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"}`; diff --git a/src/sqlite.ts b/src/sqlite.ts index 63ffc41..f41e7cf 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,14 @@ 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 +138,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 +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:");