fix(db): keep the lock-free path for the transaction handle only

The global inTransaction flag let every execute/select bypass the FIFO
lock while a withTransaction callback ran, wherever it came from. The
callback now receives its own handle, which runs directly, refuses
close() and throws once the callback settled; any other call, getDb()
from a callback included, queues behind the transaction.

connectToProfile, initializeNewProfileDb and closeDb each run as one
exclusive section (close the old pool, prepare, load), and every open
or close bumps a connection generation: an operation queued for a
closed connection throws "connection closed by profile switch". An
operation waiting more than 15 s for the lock writes one Journal
warning naming the holder and since when.

findOrCreateSecurity now requires its executor: its getDb() fallback
would deadlock if called from a transaction callback.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
le king fu 2026-10-12 19:28:36 -04:00
parent 8ceb3de921
commit 08946e6d08
5 changed files with 532 additions and 112 deletions

View file

@ -2367,7 +2367,7 @@ describe("roundToCent", () => {
describe("findOrCreateSecurity", () => { describe("findOrCreateSecurity", () => {
it("rejects an empty/whitespace symbol", async () => { it("rejects an empty/whitespace symbol", async () => {
await expect( await expect(
findOrCreateSecurity({ symbol: " ", asset_type: "stock" }) findOrCreateSecurity({ symbol: " ", asset_type: "stock" }, mockDb)
).rejects.toMatchObject({ code: "security_symbol_required" }); ).rejects.toMatchObject({ code: "security_symbol_required" });
expect(mockExecute).not.toHaveBeenCalled(); expect(mockExecute).not.toHaveBeenCalled();
}); });
@ -2375,7 +2375,7 @@ describe("findOrCreateSecurity", () => {
it("rejects an invalid asset_type", async () => { it("rejects an invalid asset_type", async () => {
await expect( await expect(
// @ts-expect-error runtime guard // @ts-expect-error runtime guard
findOrCreateSecurity({ symbol: "AAPL", asset_type: "bond" }) findOrCreateSecurity({ symbol: "AAPL", asset_type: "bond" }, mockDb)
).rejects.toMatchObject({ code: "security_asset_type_invalid" }); ).rejects.toMatchObject({ code: "security_asset_type_invalid" });
expect(mockExecute).not.toHaveBeenCalled(); expect(mockExecute).not.toHaveBeenCalled();
}); });
@ -2393,10 +2393,10 @@ describe("findOrCreateSecurity", () => {
updated_at: "", updated_at: "",
}, },
]); ]);
const sec = await findOrCreateSecurity({ const sec = await findOrCreateSecurity(
symbol: " aapl ", { symbol: " aapl ", asset_type: "stock" },
asset_type: "stock", mockDb
}); );
expect(sec.id).toBe(3); expect(sec.id).toBe(3);
// The INSERT and the lookup both use the normalized symbol. // The INSERT and the lookup both use the normalized symbol.
expect(mockExecute.mock.calls[0][1]?.[0]).toBe("AAPL"); expect(mockExecute.mock.calls[0][1]?.[0]).toBe("AAPL");
@ -2410,26 +2410,29 @@ describe("findOrCreateSecurity", () => {
mockSelect.mockResolvedValueOnce([ mockSelect.mockResolvedValueOnce([
{ id: 9, symbol: "BTC", name: null, currency: "CAD", asset_type: "crypto", created_at: "", updated_at: "" }, { id: 9, symbol: "BTC", name: null, currency: "CAD", asset_type: "crypto", created_at: "", updated_at: "" },
]); ]);
const a = await findOrCreateSecurity({ symbol: "btc", asset_type: "crypto" }); const a = await findOrCreateSecurity({ symbol: "btc", asset_type: "crypto" }, mockDb);
// Second call, different casing — the normalized conflict target returns id 9. // Second call, different casing — the normalized conflict target returns id 9.
mockExecute.mockResolvedValueOnce({ rowsAffected: 1 }); mockExecute.mockResolvedValueOnce({ rowsAffected: 1 });
mockSelect.mockResolvedValueOnce([ mockSelect.mockResolvedValueOnce([
{ id: 9, symbol: "BTC", name: null, currency: "CAD", asset_type: "crypto", created_at: "", updated_at: "" }, { id: 9, symbol: "BTC", name: null, currency: "CAD", asset_type: "crypto", created_at: "", updated_at: "" },
]); ]);
const b = await findOrCreateSecurity({ symbol: " BtC ", asset_type: "crypto" }); const b = await findOrCreateSecurity({ symbol: " BtC ", asset_type: "crypto" }, mockDb);
expect(a.id).toBe(b.id); expect(a.id).toBe(b.id);
expect(b.id).toBe(9); expect(b.id).toBe(9);
}); });
it("threads an executor when given (in-txn create)", async () => { it("runs on the executor it is given and never reaches for getDb() (#408)", async () => {
vi.mocked(getDb).mockClear();
const execSelect = vi.fn().mockResolvedValueOnce([ const execSelect = vi.fn().mockResolvedValueOnce([
{ id: 1, symbol: "VEQT.TO", name: null, currency: "CAD", asset_type: "stock", created_at: "", updated_at: "" }, { id: 1, symbol: "VEQT.TO", name: null, currency: "CAD", asset_type: "stock", created_at: "", updated_at: "" },
]); ]);
const execExecute = vi.fn().mockResolvedValueOnce({ rowsAffected: 1 }); const execExecute = vi.fn().mockResolvedValueOnce({ rowsAffected: 1 });
const exec = { select: execSelect, execute: execExecute }; const exec = { select: execSelect, execute: execExecute };
await findOrCreateSecurity({ symbol: "veqt.to", asset_type: "stock" }, exec); await findOrCreateSecurity({ symbol: "veqt.to", asset_type: "stock" }, exec);
// The provided executor was used, NOT the top-level db handle. // The provided executor was used, NOT the top-level db handle: a getDb()
// call from a transaction callback would deadlock behind the DB lock.
expect(execExecute).toHaveBeenCalledTimes(1); expect(execExecute).toHaveBeenCalledTimes(1);
expect(getDb).not.toHaveBeenCalled();
expect(mockExecute).not.toHaveBeenCalled(); expect(mockExecute).not.toHaveBeenCalled();
expect(mockSelect).not.toHaveBeenCalled(); expect(mockSelect).not.toHaveBeenCalled();
}); });

View file

@ -977,16 +977,18 @@ const ASSET_TYPES: readonly BalanceAssetType[] = ["stock", "crypto"];
* UPSERT a security by NORMALIZED symbol, returning its row. Idempotent: a * UPSERT a security by NORMALIZED symbol, returning its row. Idempotent: a
* second call with the same symbol (any casing) returns the existing row, * second call with the same symbol (any casing) returns the existing row,
* updating `name`/`asset_type`/`currency` only when the caller passes new * updating `name`/`asset_type`/`currency` only when the caller passes new
* values. Callable inside an open transaction by threading `exec` (so a * values. `exec` is mandatory: inside a transaction, the handle the
* brand-new symbol can create its security then its holding atomically inside * `withTransaction` callback received (so a brand-new symbol can create its
* `saveSnapshotAtomic`); omit `exec` to use the top-level db handle. * security then its holding atomically inside `saveSnapshotAtomic`); outside,
* `await getDb()`. No `getDb()` fallback here: called from a transaction
* callback, it would deadlock behind the DB lock (#408).
* *
* @throws `security_symbol_required` on empty symbol, * @throws `security_symbol_required` on empty symbol,
* `security_asset_type_invalid` on a non-enum asset_type. * `security_asset_type_invalid` on a non-enum asset_type.
*/ */
export async function findOrCreateSecurity( export async function findOrCreateSecurity(
input: FindOrCreateSecurityInput, input: FindOrCreateSecurityInput,
exec?: SqlExecutor exec: SqlExecutor
): Promise<BalanceSecurity> { ): Promise<BalanceSecurity> {
const symbol = normalizeSecuritySymbol(input.symbol ?? ""); const symbol = normalizeSecuritySymbol(input.symbol ?? "");
if (symbol.length === 0) { if (symbol.length === 0) {
@ -1003,7 +1005,7 @@ export async function findOrCreateSecurity(
} }
const currency = input.currency ?? BALANCE_CURRENCY_CAD; const currency = input.currency ?? BALANCE_CURRENCY_CAD;
const name = input.name ? input.name.trim() || null : null; const name = input.name ? input.name.trim() || null : null;
const db: SqlExecutor = exec ?? (await getDb()); const db = exec;
// ON CONFLICT(symbol) DO UPDATE keeps the row's metadata fresh (e.g. a name // ON CONFLICT(symbol) DO UPDATE keeps the row's metadata fresh (e.g. a name
// arriving on a later save) while preserving its id. COALESCE keeps an // arriving on a later save) while preserving its id. COALESCE keeps an
// existing name when the new payload omits one. The column is UNIQUE COLLATE // existing name when the new payload omits one. The column is UNIQUE COLLATE

View file

@ -1,4 +1,4 @@
import { describe, it, expect, vi, beforeEach } from "vitest"; import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
// Regression coverage for the "database is locked" fix (#228 follow-up). // Regression coverage for the "database is locked" fix (#228 follow-up).
// //
@ -30,22 +30,28 @@ vi.mock("@tauri-apps/plugin-sql", () => ({
vi.mock("@tauri-apps/api/core", () => ({ vi.mock("@tauri-apps/api/core", () => ({
invoke: vi.fn(async (command: string) => (command === "prepare_profile_db" ? h.prepare() : false)), invoke: vi.fn(async (command: string) => (command === "prepare_profile_db" ? h.prepare() : false)),
})); }));
vi.mock("./logService", () => ({ logInfo: vi.fn(), logError: vi.fn() })); vi.mock("./logService", () => ({ logInfo: vi.fn(), logWarn: vi.fn(), logError: vi.fn() }));
import Database from "@tauri-apps/plugin-sql"; import Database from "@tauri-apps/plugin-sql";
import { invoke } from "@tauri-apps/api/core"; import { invoke } from "@tauri-apps/api/core";
import { logError, logInfo } from "./logService"; import { logError, logInfo, logWarn } from "./logService";
import { import {
BLOCKED_ALERT_MS,
closeDb,
connectToProfile, connectToProfile,
DbConnectionClosedError,
describePreparation, describePreparation,
getDb, getDb,
initializeNewProfileDb,
TransactionHandleCloseError,
TransactionHandleRevokedError,
withTransaction, withTransaction,
type ProfilePreparation, type ProfilePreparation,
} from "./db"; } from "./db";
// Each op records start, yields a real macrotask (a window for interleaving if // Each op records start, yields a real macrotask (a window for interleaving if
// the lock were broken), then records end. The lock must keep them ordered. // the lock were broken), then records end. The lock must keep them ordered.
function makeFakeDb(): FakeDb { function makeFakeDb(name = "db"): FakeDb {
return { return {
execute: vi.fn(async (q: string) => { execute: vi.fn(async (q: string) => {
events.push(`start:${q}`); events.push(`start:${q}`);
@ -59,15 +65,19 @@ function makeFakeDb(): FakeDb {
events.push(`end:${q}`); events.push(`end:${q}`);
return []; return [];
}), }),
close: vi.fn(async () => {}), close: vi.fn(async () => {
events.push(`close:${name}`);
}),
}; };
} }
beforeEach(async () => { beforeEach(async () => {
events.length = 0;
h.fakeDb = makeFakeDb(); h.fakeDb = makeFakeDb();
h.prepare = async () => null; h.prepare = async () => null;
await connectToProfile("test.db"); await connectToProfile("test.db");
// Cleared after connecting: the connection opens by closing the previous
// test's fake, whose close is logged too.
events.length = 0;
}); });
describe("db serialization (database-locked fix)", () => { describe("db serialization (database-locked fix)", () => {
@ -103,17 +113,93 @@ describe("db serialization (database-locked fix)", () => {
]); ]);
}); });
it("lets a nested getDb() call run directly inside a transaction — no deadlock", async () => { it("makes a getDb() call from inside a transaction wait behind it — strict lock (#408)", async () => {
// A service called inside withTransaction might reach for getDb() (the // The lock is not reentrant: only the callback's handle runs directly. A
// serialized proxy) instead of the threaded handle. The reentrancy guard // nested service reaching for getDb() queues behind the transaction (this
// must run it directly rather than queue it behind the held lock (deadlock). // test releases the callback by hand; a callback awaiting it deadlocks,
const result = await withTransaction(async () => { // which the blocking alert reports).
const inner = await getDb(); let release: () => void = () => {};
await inner.execute("NESTED"); const gate = new Promise<void>((resolve) => {
return "ok"; release = resolve;
}); });
expect(result).toBe("ok"); let nested: Promise<unknown> = Promise.resolve();
expect(events).toEqual(["start:NESTED", "end:NESTED"]); const txn = withTransaction(async (tx) => {
await tx.execute("BEGIN");
nested = (await getDb()).execute("NESTED");
await new Promise((r) => setTimeout(r, 10));
await tx.execute("COMMIT");
await gate;
});
await vi.waitFor(() => expect(events).toContain("end:COMMIT"));
expect(events).not.toContain("start:NESTED");
release();
await txn;
await nested;
expect(events).toEqual(["start:BEGIN", "end:BEGIN", "start:COMMIT", "end:COMMIT", "start:NESTED", "end:NESTED"]);
});
it("runs a read launched from elsewhere during the callback after the COMMIT (#408)", async () => {
// The pre-#408 global flag let ANY query bypass the lock while a callback
// ran: a read started after BEGIN ran in parallel with the transaction.
const db = await getDb();
let outside: Promise<unknown> = Promise.resolve();
const txn = withTransaction(async (tx) => {
await tx.execute("BEGIN");
outside = db.select("READ");
await tx.execute("WRITE");
await tx.execute("COMMIT");
});
await txn;
await outside;
expect(events).toEqual([
"start:BEGIN",
"end:BEGIN",
"start:WRITE",
"end:WRITE",
"start:COMMIT",
"end:COMMIT",
"start:READ",
"end:READ",
]);
});
it("lets the callback's own handle run its statements without waiting", async () => {
const rows = await withTransaction(async (tx) => {
await tx.execute("BEGIN");
const result = await tx.select<unknown[]>("SELECT");
await tx.execute("COMMIT");
return result;
});
expect(rows).toEqual([]);
expect(events).toEqual(["start:BEGIN", "end:BEGIN", "start:SELECT", "end:SELECT", "start:COMMIT", "end:COMMIT"]);
});
it("revokes the transaction handle once the callback settled", async () => {
let kept: Awaited<ReturnType<typeof getDb>> | null = null;
await withTransaction(async (tx) => {
kept = tx;
});
await expect(kept!.execute("LATE")).rejects.toBeInstanceOf(TransactionHandleRevokedError);
await expect(kept!.select("LATE")).rejects.toBeInstanceOf(TransactionHandleRevokedError);
expect(events).not.toContain("start:LATE");
});
it("revokes the handle of a failed transaction too", async () => {
let kept: Awaited<ReturnType<typeof getDb>> | null = null;
await expect(
withTransaction(async (tx) => {
kept = tx;
throw new Error("boom");
})
).rejects.toThrow("boom");
await expect(kept!.execute("LATE")).rejects.toBeInstanceOf(TransactionHandleRevokedError);
});
it("refuses close() on the transaction handle", async () => {
await withTransaction(async (tx) => {
await expect(tx.close()).rejects.toBeInstanceOf(TransactionHandleCloseError);
});
expect(h.fakeDb.close).not.toHaveBeenCalled();
}); });
it("releases the lock after a failed transaction so later ops still run", async () => { it("releases the lock after a failed transaction so later ops still run", async () => {
@ -139,6 +225,184 @@ describe("db serialization (database-locked fix)", () => {
}); });
}); });
// ---------------------------------------------------------------------------
// Profile switch through the lock, connection generation, blocking alert (#408)
// ---------------------------------------------------------------------------
describe("profile switch goes through the lock (#408)", () => {
afterEach(() => {
vi.mocked(Database.load).mockImplementation(async () => h.fakeDb as never);
});
it("closes the connection only after a running transaction's COMMIT", async () => {
const old = h.fakeDb;
let switching: Promise<void> = Promise.resolve();
const txn = withTransaction(async (tx) => {
await tx.execute("BEGIN");
h.fakeDb = makeFakeDb("next");
switching = connectToProfile("next.db");
await new Promise((r) => setTimeout(r, 10));
await tx.execute("COMMIT");
});
await txn;
await switching;
expect(events).toEqual(["start:BEGIN", "end:BEGIN", "start:COMMIT", "end:COMMIT", "close:db"]);
expect(old.close).toHaveBeenCalledTimes(1);
});
it("two concurrent switches leave one pool open, on the last file asked for", async () => {
const pools = new Map<string, FakeDb>();
vi.mocked(Database.load).mockImplementation(async (url: string) => {
const fake = makeFakeDb(url);
pools.set(url, fake);
return fake as never;
});
await Promise.all([connectToProfile("a.db"), connectToProfile("b.db")]);
expect([...pools.keys()]).toEqual(["sqlite:a.db", "sqlite:b.db"]);
expect(pools.get("sqlite:a.db")!.close).toHaveBeenCalledTimes(1);
expect(pools.get("sqlite:b.db")!.close).not.toHaveBeenCalled();
await (await getDb()).execute("Q");
expect(pools.get("sqlite:b.db")!.execute).toHaveBeenCalledWith("Q", undefined);
});
it("invokes prepare_profile_db only once the old connection's close() resolved", async () => {
const old = h.fakeDb;
vi.mocked(old.close).mockImplementation(async () => {
events.push("close-start");
await new Promise((r) => setTimeout(r, 10));
events.push("close-end");
});
h.prepare = async () => {
events.push("prepare");
return null;
};
h.fakeDb = makeFakeDb("next");
await connectToProfile("next.db");
expect(events).toEqual(["close-start", "close-end", "prepare"]);
});
it("fails a transaction queued behind a switch instead of running it on the new database", async () => {
const old = h.fakeDb;
const next = makeFakeDb("next");
h.fakeDb = next;
const switching = connectToProfile("next.db");
const callback = vi.fn(async () => "ran");
const txn = withTransaction(callback);
await switching;
await expect(txn).rejects.toBeInstanceOf(DbConnectionClosedError);
await expect(txn).rejects.toThrow("connection closed by profile switch");
expect(callback).not.toHaveBeenCalled();
expect(next.execute).not.toHaveBeenCalled();
expect(old.execute).not.toHaveBeenCalled();
});
it("fails a query queued on the old handle behind a switch", async () => {
const before = await getDb();
const next = makeFakeDb("next");
h.fakeDb = next;
const switching = connectToProfile("next.db");
const late = before.select("OLD");
await switching;
await expect(late).rejects.toBeInstanceOf(DbConnectionClosedError);
expect(next.select).not.toHaveBeenCalled();
});
it("closeDb waits for the queue, then the old handle refuses to run", async () => {
const db = await getDb();
const first = db.execute("FIRST");
const closing = closeDb();
await first;
await closing;
expect(events).toEqual(["start:FIRST", "end:FIRST", "close:db"]);
await expect(db.execute("AFTER")).rejects.toBeInstanceOf(DbConnectionClosedError);
});
it("initializeNewProfileDb runs its SQL inside the section, before a queued query", async () => {
const next = makeFakeDb("new");
h.fakeDb = next;
const init = initializeNewProfileDb("new.db", ["CREATE A", "CREATE B"]);
const after = (async () => {
await init;
return (await getDb()).select("READ");
})();
await after;
expect(events).toEqual([
"close:db",
"start:CREATE A",
"end:CREATE A",
"start:CREATE B",
"end:CREATE B",
"start:READ",
"end:READ",
]);
});
});
describe("blocking alert (#408)", () => {
beforeEach(() => {
vi.mocked(logWarn).mockClear();
vi.useFakeTimers();
});
afterEach(() => {
vi.useRealTimers();
});
it("warns once, naming the transaction that holds the lock, when getDb() is called from a callback", async () => {
let release: () => void = () => {};
const gate = new Promise<void>((resolve) => {
release = resolve;
});
let nested: Promise<unknown> = Promise.resolve();
const startedAt = new Date(Date.now()).toISOString();
const txn = withTransaction(async () => {
nested = (await getDb()).execute("NESTED");
await gate;
});
await vi.advanceTimersByTimeAsync(BLOCKED_ALERT_MS - 1);
expect(logWarn).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1);
expect(logWarn).toHaveBeenCalledTimes(1);
const message = String(vi.mocked(logWarn).mock.calls[0][0]);
expect(message).toContain("a query has waited 15 s");
expect(message).toContain(`held by a transaction since ${startedAt}`);
// Once per operation: still one warning much later.
await vi.advanceTimersByTimeAsync(BLOCKED_ALERT_MS * 3);
expect(logWarn).toHaveBeenCalledTimes(1);
release();
await vi.advanceTimersByTimeAsync(10);
await txn;
await nested;
});
it("says nothing for an operation that got the lock in time", async () => {
const db = await getDb();
const run = db.execute("QUICK");
await vi.advanceTimersByTimeAsync(10);
await run;
await vi.advanceTimersByTimeAsync(BLOCKED_ALERT_MS * 2);
expect(logWarn).not.toHaveBeenCalled();
});
it("names a profile opening that holds the lock (long catch-up)", async () => {
let finish: () => void = () => {};
h.prepare = () =>
new Promise((resolve) => {
finish = () => resolve(null);
});
h.fakeDb = makeFakeDb("next");
const switching = connectToProfile("next.db");
// Settled outcome caught at once: the rejection lands while timers advance.
const waiting = withTransaction(async () => "late").catch((e: unknown) => e);
await vi.advanceTimersByTimeAsync(BLOCKED_ALERT_MS);
expect(logWarn).toHaveBeenCalledTimes(1);
expect(String(vi.mocked(logWarn).mock.calls[0][0])).toContain("a transaction has waited 15 s, the lock is held by a profile opening");
finish();
await vi.advanceTimersByTimeAsync(10);
await switching;
expect(await waiting).toBeInstanceOf(DbConnectionClosedError);
});
});
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// Profile preparation (#390) // Profile preparation (#390)
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------

View file

@ -1,12 +1,12 @@
import Database from "@tauri-apps/plugin-sql"; import Database from "@tauri-apps/plugin-sql";
import { invoke } from "@tauri-apps/api/core"; import { invoke } from "@tauri-apps/api/core";
import { logError, logInfo } from "./logService"; import { logError, logInfo, logWarn } from "./logService";
let rawInstance: Database | null = null; let rawInstance: Database | null = null;
let serializedInstance: Database | null = null; let serializedInstance: Database | null = null;
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
// DB access serialization (#228 follow-up — fix "database is locked"). // DB access serialization (#228 follow-up — fix "database is locked"; #408).
// //
// tauri-plugin-sql exposes NO transaction primitive to the JS side and loads // tauri-plugin-sql exposes NO transaction primitive to the JS side and loads
// SQLite through a default multi-connection sqlx pool (`Pool::connect` => // SQLite through a default multi-connection sqlx pool (`Pool::connect` =>
@ -23,20 +23,101 @@ let serializedInstance: Database | null = null;
// statements all run on the same connection and can never strand. Serializing // statements all run on the same connection and can never strand. Serializing
// all DB access is a non-issue for a local single-user app where queries are // all DB access is a non-issue for a local single-user app where queries are
// tiny and infrequent — and it makes the "atomic" saves genuinely atomic. // tiny and infrequent — and it makes the "atomic" saves genuinely atomic.
//
// Rules (#408):
// - The lock is strict and NOT reentrant. Inside a `withTransaction` callback,
// use ONLY the handle the callback receives: it runs directly on the
// connection the transaction holds. Calling `getDb()` from a callback (or
// from anything it calls) queues behind the transaction that waits for it:
// a deadlock, reported in the Journal after BLOCKED_ALERT_MS (see below).
// - The transaction handle is revoked when the callback settles: a call on it
// afterwards (an orphan promise) throws `TransactionHandleRevokedError`
// instead of running outside the lock. `close()` on it always throws.
// - Opening, closing and switching a profile each run as ONE exclusive section
// (`connectToProfile`, `initializeNewProfileDb`, `closeDb`), so they wait for
// the operations already queued and for a running transaction, and nothing
// runs between closing the old pool and opening the new one.
// - Every open or close bumps a connection generation. A handle checks its
// generation when its operation leaves the queue: an operation queued for a
// connection that was closed meanwhile throws `DbConnectionClosedError`
// rather than running on the next profile's database or on a closed pool.
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
let lockTail: Promise<unknown> = Promise.resolve(); /** An operation queued for a connection closed before it could run. */
export class DbConnectionClosedError extends Error {
constructor() {
super("connection closed by profile switch");
this.name = "DbConnectionClosedError";
}
}
// True only while a `withTransaction` callback is executing. The lock is held /** A transaction handle used after its `withTransaction` callback settled. */
// for that whole callback, so anything it triggers (the threaded handle, or an export class TransactionHandleRevokedError extends Error {
// accidental `getDb()` call from a nested service) must run DIRECTLY rather constructor() {
// than re-entering the non-reentrant lock — otherwise it would deadlock behind super("transaction handle used after its withTransaction callback ended");
// the transaction that is waiting for it. this.name = "TransactionHandleRevokedError";
let inTransaction = false; }
}
/** `close()` called on a transaction handle: only the profile functions close. */
export class TransactionHandleCloseError extends Error {
constructor() {
super("close() is not allowed on a transaction handle; use closeDb()");
this.name = "TransactionHandleCloseError";
}
}
/** What holds the lock, named in the blocking alert. */
type LockKind =
| "query"
| "transaction"
| "profile opening"
| "profile initialization"
| "connection close";
/**
* An operation still waiting for the lock after this delay writes ONE warning
* to the Journal, naming what holds the lock and since when. The typical cause
* is a `getDb()` call from inside a `withTransaction` callback (deadlock). A
* profile catch-up (`prepare_profile_db`) that takes longer also triggers it,
* naming a "profile opening": expected, nothing can run during a switch.
* `logWarn` only writes to memory and sessionStorage, never to the DB, so it
* is safe to call while the lock is held.
*/
export const BLOCKED_ALERT_MS = 15_000;
let lockTail: Promise<unknown> = Promise.resolve();
let lockHolder: { kind: LockKind; since: number } | null = null;
let generation = 0;
/** Run `fn` after every previously-queued DB operation has settled (FIFO). */ /** Run `fn` after every previously-queued DB operation has settled (FIFO). */
function runExclusive<T>(fn: () => Promise<T>): Promise<T> { function runExclusive<T>(kind: LockKind, fn: () => Promise<T>): Promise<T> {
const result = lockTail.then(fn, fn); const queuedAt = Date.now();
let started = false;
const alert = setTimeout(() => {
if (started) return;
const holder = lockHolder;
const waited = Math.round((Date.now() - queuedAt) / 1000);
logWarn(
holder
? `DB lock: a ${kind} has waited ${waited} s, the lock is held by a ${holder.kind} since ${new Date(holder.since).toISOString()}` +
(holder.kind === "transaction"
? " (a getDb() call inside a withTransaction callback deadlocks: use the callback's handle)"
: "")
: `DB lock: a ${kind} has waited ${waited} s behind the queue`
);
}, BLOCKED_ALERT_MS);
const run = async (): Promise<T> => {
started = true;
clearTimeout(alert);
lockHolder = { kind, since: Date.now() };
try {
return await fn();
} finally {
lockHolder = null;
}
};
const result = lockTail.then(run, run);
// Keep the chain alive regardless of fn's outcome; only the tail swallows. // Keep the chain alive regardless of fn's outcome; only the tail swallows.
lockTail = result.then( lockTail = result.then(
() => undefined, () => undefined,
@ -45,26 +126,38 @@ function runExclusive<T>(fn: () => Promise<T>): Promise<T> {
return result; return result;
} }
/** Throw when the connection a handle was made for is gone. */
function assertGeneration(handleGeneration: number): void {
if (handleGeneration !== generation) {
throw new DbConnectionClosedError();
}
}
/** /**
* Wrap the raw Database so `execute`/`select` are serialized through the lock. * Wrap the raw Database so `execute`/`select` are serialized through the lock
* While a transaction holds the lock (`inTransaction`), calls run directly so a * and checked against the connection generation once they leave the queue.
* nested call never deadlocks. The proxy keeps the `Database` type, so call * The proxy keeps the `Database` type, so call sites (including
* sites (including `db.select<T>(...)`) are unchanged. * `db.select<T>(...)`) are unchanged.
*/ */
function serialize(db: Database): Database { function serialize(db: Database, handleGeneration: number): Database {
return new Proxy(db, { return new Proxy(db, {
get(target, prop, receiver) { get(target, prop, receiver) {
if (prop === "execute") { if (prop === "execute") {
return (query: string, bindValues?: unknown[]) => return (query: string, bindValues?: unknown[]) =>
inTransaction runExclusive("query", async () => {
? target.execute(query, bindValues) assertGeneration(handleGeneration);
: runExclusive(() => target.execute(query, bindValues)); return target.execute(query, bindValues);
});
} }
if (prop === "select") { if (prop === "select") {
return (query: string, bindValues?: unknown[]) => return (query: string, bindValues?: unknown[]) =>
inTransaction runExclusive("query", async () => {
? target.select(query, bindValues) assertGeneration(handleGeneration);
: runExclusive(() => target.select(query, bindValues)); return target.select(query, bindValues);
});
}
if (prop === "close") {
return () => closeDb();
} }
const value = Reflect.get(target, prop, receiver); const value = Reflect.get(target, prop, receiver);
return typeof value === "function" ? value.bind(target) : value; return typeof value === "function" ? value.bind(target) : value;
@ -72,6 +165,51 @@ function serialize(db: Database): Database {
}); });
} }
/**
* The handle a `withTransaction` callback receives: runs directly on the
* connection (the transaction already holds the lock), refuses `close()`, and
* throws once `revoke()` ran at the end of the callback.
*/
function transactionHandle(
db: Database,
handleGeneration: number
): { handle: Database; revoke: () => void } {
let active = true;
const check = () => {
if (!active) throw new TransactionHandleRevokedError();
assertGeneration(handleGeneration);
};
const handle = new Proxy(db, {
get(target, prop, receiver) {
if (prop === "execute") {
return async (query: string, bindValues?: unknown[]) => {
check();
return target.execute(query, bindValues);
};
}
if (prop === "select") {
return async (query: string, bindValues?: unknown[]) => {
check();
return target.select(query, bindValues);
};
}
if (prop === "close") {
return async () => {
throw new TransactionHandleCloseError();
};
}
const value = Reflect.get(target, prop, receiver);
return typeof value === "function" ? value.bind(target) : value;
},
});
return {
handle,
revoke: () => {
active = false;
},
};
}
export async function getDb(): Promise<Database> { export async function getDb(): Promise<Database> {
if (!serializedInstance) { if (!serializedInstance) {
throw new Error("No database connection. Call connectToProfile() first."); throw new Error("No database connection. Call connectToProfile() first.");
@ -82,28 +220,51 @@ export async function getDb(): Promise<Database> {
/** /**
* Run `fn` as one serialized transaction. The global DB lock is held for the * Run `fn` as one serialized transaction. The global DB lock is held for the
* ENTIRE callback so no other db.execute/select can interleave and strand the * ENTIRE callback so no other db.execute/select can interleave and strand the
* transaction on a different pooled connection. `fn` receives the db handle and * transaction on a different pooled connection. `fn` receives the transaction
* is responsible for issuing BEGIN/COMMIT (and ROLLBACK on error), exactly as * handle and is responsible for issuing BEGIN/COMMIT (and ROLLBACK on error)
* the inline pattern did before — `withTransaction` only guarantees exclusive, * on it — `withTransaction` only guarantees exclusive, non-interleaved access
* non-interleaved access for the whole BEGIN..COMMIT. * for the whole BEGIN..COMMIT. Never call `getDb()` from `fn` (deadlock).
*/ */
export async function withTransaction<T>( export async function withTransaction<T>(
fn: (db: Database) => Promise<T> fn: (db: Database) => Promise<T>
): Promise<T> { ): Promise<T> {
if (!serializedInstance) { if (!rawInstance) {
throw new Error("No database connection. Call connectToProfile() first."); throw new Error("No database connection. Call connectToProfile() first.");
} }
const db = serializedInstance; const db = rawInstance;
return runExclusive(async () => { const handleGeneration = generation;
inTransaction = true; return runExclusive("transaction", async () => {
assertGeneration(handleGeneration);
const { handle, revoke } = transactionHandle(db, handleGeneration);
try { try {
return await fn(db); return await fn(handle);
} finally { } finally {
inTransaction = false; revoke();
} }
}); });
} }
// Internal helpers for the profile sections below. They NEVER take the lock:
// only call them from inside a `runExclusive` section.
async function closeUnlocked(): Promise<void> {
if (rawInstance) {
const closing = rawInstance;
rawInstance = null;
serializedInstance = null;
generation += 1;
await closing.close();
}
}
async function loadUnlocked(dbFilename: string): Promise<Database> {
const db = await Database.load(`sqlite:${dbFilename}`);
generation += 1;
rawInstance = db;
serializedInstance = serialize(db, generation);
return db;
}
/** What `prepare_profile_db` did to one profile database (#390). */ /** What `prepare_profile_db` did to one profile database (#390). */
export interface ProfilePreparation { export interface ProfilePreparation {
catch_up: { catch_up: {
@ -159,12 +320,10 @@ export function describePreparation(
return { info, errors }; return { info, errors };
} }
export async function connectToProfile(dbFilename: string): Promise<void> { export function connectToProfile(dbFilename: string): Promise<void> {
if (rawInstance) { return runExclusive("profile opening", async () => {
await rawInstance.close(); // The old pool is fully closed before the preparation touches the file.
rawInstance = null; await closeUnlocked();
serializedInstance = null;
}
// Repair migration checksums before loading (fixes "migration was modified" error) // Repair migration checksums before loading (fixes "migration was modified" error)
try { try {
const repaired = await invoke<boolean>("repair_migrations", { dbFilename }); const repaired = await invoke<boolean>("repair_migrations", { dbFilename });
@ -187,27 +346,20 @@ export async function connectToProfile(dbFilename: string): Promise<void> {
`Profile ${dbFilename}: database preparation failed: ${e instanceof Error ? e.message : String(e)}` `Profile ${dbFilename}: database preparation failed: ${e instanceof Error ? e.message : String(e)}`
); );
} }
rawInstance = await Database.load(`sqlite:${dbFilename}`); await loadUnlocked(dbFilename);
serializedInstance = serialize(rawInstance); });
} }
export async function initializeNewProfileDb(dbFilename: string, sqlStatements: string[]): Promise<void> { export function initializeNewProfileDb(dbFilename: string, sqlStatements: string[]): Promise<void> {
if (rawInstance) { return runExclusive("profile initialization", async () => {
await rawInstance.close(); await closeUnlocked();
rawInstance = null; const db = await loadUnlocked(dbFilename);
serializedInstance = null;
}
rawInstance = await Database.load(`sqlite:${dbFilename}`);
serializedInstance = serialize(rawInstance);
for (const sql of sqlStatements) { for (const sql of sqlStatements) {
await rawInstance.execute(sql); await db.execute(sql);
} }
});
} }
export async function closeDb(): Promise<void> { export function closeDb(): Promise<void> {
if (rawInstance) { return runExclusive("connection close", () => closeUnlocked());
await rawInstance.close();
rawInstance = null;
serializedInstance = null;
}
} }

View file

@ -126,9 +126,8 @@ export async function getOnboardingSnapshot(): Promise<OnboardingSnapshot> {
* it. * it.
* *
* The flag is read first, outside any transaction: once D9 ran, every later * The flag is read first, outside any transaction: once D9 ran, every later
* opening costs that one queued read and opens no transaction (while a * opening costs that one queued read and opens no transaction. Without the
* `withTransaction` callback runs, `db.ts` lets every other query bypass its * flag, D9 runs as one transaction under the DB lock, flag
* lock). Without the flag, D9 runs as one transaction under the DB lock, flag
* re-read inside it, so two concurrent first loads cannot apply it twice * re-read inside it, so two concurrent first loads cannot apply it twice
* (CWE-362): the hidden preference is written first, then the flag, then * (CWE-362): the hidden preference is written first, then the flag, then
* COMMIT; any failure rolls both back and leaves the flag unset, so the next * COMMIT; any failure rolls both back and leaves the flag unset, so the next