diff --git a/CHANGELOG.md b/CHANGELOG.md index b367f4080..d72247ac0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,27 @@ - The embed session `maxDuration` is now env-configurable via `QMD_EMBED_MAX_DURATION_MS` (default: 30 min). This prevents large-corpus embeddings from being aborted by the hardcoded 30-minute ceiling (#673). +- `qmd query`, `qmd update`, and other commands no longer fail with + `SQLiteError: database is locked` when multiple processes run against the + same index in parallel (e.g. an `update` racing a long `embed`, the + first-open schema migration racing a routine command, or an agent fanning + out searches). `openDatabase` now sets `PRAGMA busy_timeout = 120000` on + every connection, so a writer that loses the race queues at batch + boundaries instead of throwing immediately. WAL handles read/write + concurrency but does not serialise concurrent writers, and `bun:sqlite` + and `better-sqlite3` both default the timeout to 0, so the loser + previously failed on the first DDL statement in `initializeDatabase`. + Override the default with `QMD_SQLITE_BUSY_TIMEOUT` (milliseconds; `0` + restores fail-fast). Two more crashes on the same concurrent-open + path are fixed: `trigger documents_ai already exists` (the FTS sync + triggers were dropped and recreated as separate statements on every + open, so two processes interleaved between the `DROP` and the + `CREATE`; `busy_timeout` serialises individual statements but not the + pair) and `database is locked` while migrating a cold database to WAL + (the `journal_mode` switch needs a brief exclusive lock and does not + invoke the busy handler). FTS trigger setup is now gated behind + `PRAGMA user_version` inside one `IMMEDIATE` transaction, and the WAL + migration retries within the busy-timeout budget. ## [2.5.3] - 2026-05-28 diff --git a/src/db.ts b/src/db.ts index b23a65ba2..a290a9fe8 100644 --- a/src/db.ts +++ b/src/db.ts @@ -61,11 +61,65 @@ if (isBun) { _sqliteVecLoad = (db: LoadableSqliteDatabase) => sqliteVec.load(db as Parameters[0]); } +function isBusyError(err: unknown): boolean { + if (typeof err !== "object" || err === null) return false; + const code = (err as { code?: unknown }).code; + if (code === "SQLITE_BUSY" || code === "SQLITE_BUSY_SNAPSHOT") return true; + const message = (err as { message?: unknown }).message; + return typeof message === "string" && /database is locked|database is busy|SQLITE_BUSY/i.test(message); +} + +function sleepSync(ms: number): void { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, ms); +} + +/** + * Switch a connection to WAL, retrying on `SQLITE_BUSY` within the busy-timeout + * budget. Unlike ordinary writes, migrating the journal needs a brief exclusive + * lock and does NOT invoke the busy handler, so concurrent first-time opens of a + * cold database throw "database is locked" even with `busy_timeout` set. Once the + * database is already WAL the pragma is a cheap no-op that does not contend. + */ +function enableWal(db: Database, budgetMs: number): void { + const deadline = Date.now() + Math.max(budgetMs, 0); + for (let attempt = 0; ; attempt++) { + try { + db.exec("PRAGMA journal_mode = WAL"); + return; + } catch (err) { + if (!isBusyError(err) || Date.now() >= deadline) throw err; + sleepSync(Math.min(5 + attempt, 25)); + } + } +} + /** * Open a SQLite database. Works with both bun:sqlite and better-sqlite3. + * + * `bun:sqlite` and `better-sqlite3` both default `busy_timeout` to 0, so + * concurrent writers throw `SQLITE_BUSY` instead of waiting. WAL improves + * read-while-write concurrency but does not serialise writers. Setting the + * timeout at connection open makes parallel processes (e.g. an `update` or + * `query` racing a long `embed`, or a first-open schema migration racing any + * routine command) queue at batch boundaries instead of failing on contact. + * + * WAL is enabled here too (with a bounded retry) so connection-level pragmas + * live in one place and the cold-database journal migration survives concurrent + * opens. + * + * Default 120_000 ms outlasts the worst-case batch commit on a multi-GB + * index. Override with `QMD_SQLITE_BUSY_TIMEOUT` (value in milliseconds; `0` + * restores the upstream fail-fast behaviour). See + * https://bun.sh/docs/api/sqlite#busy-timeout. */ export function openDatabase(path: string): Database { - return new _Database(path) as Database; + const db = new _Database(path) as Database; + const raw = process.env.QMD_SQLITE_BUSY_TIMEOUT; + const parsed = raw !== undefined && raw !== "" ? Number(raw) : Number.NaN; + const busyTimeoutMs = Number.isFinite(parsed) && parsed >= 0 ? Math.floor(parsed) : 120_000; + db.exec(`PRAGMA busy_timeout = ${busyTimeoutMs}`); + enableWal(db, busyTimeoutMs); + return db; } /** diff --git a/src/store.ts b/src/store.ts index 99e36b861..0a0e9e062 100644 --- a/src/store.ts +++ b/src/store.ts @@ -755,6 +755,10 @@ const CJK_CHAR_PATTERN = /[\p{Script=Han}\p{Script=Hiragana}\p{Script=Katakana}\ const CJK_RUN_PATTERN = /[\p{Script=Han}\p{Script=Hiragana}\p{Script=Katakana}\p{Script=Hangul}]+/gu; const FTS_CJK_NORMALIZED_VERSION = "1"; +// Bump when any FTS sync trigger body in applyFtsSyncTriggers changes, so the +// new definition is reapplied to existing databases on next open. +const STORE_SCHEMA_VERSION = 1; + /** * FTS5's unicode61 tokenizer does not segment CJK text into searchable words. * Normalize CJK runs by spacing every character so exact CJK queries can be @@ -818,6 +822,76 @@ function rebuildFTSForCjkNormalization(db: Database): void { `).run(FTS_CJK_NORMALIZED_VERSION); } +function getUserVersion(db: Database): number { + const row = db.prepare(`PRAGMA user_version`).get() as Record | undefined; + const value = row ? Object.values(row)[0] : 0; + return typeof value === "number" ? value : Number(value) || 0; +} + +// FTS sync triggers keep documents_fts current for callers that write directly +// to documents (production indexing rebuilds FTS in TypeScript to normalize CJK +// first). The bodies use DROP+CREATE rather than CREATE IF NOT EXISTS so a +// changed body propagates to existing databases. DROP and CREATE are separate +// autocommit statements, so concurrent opens of one database interleave across +// connections (A drops, B drops, A creates, B creates -> "trigger already +// exists"); busy_timeout serializes individual statements but not the pair. +// Gate the work behind PRAGMA user_version and apply it inside one IMMEDIATE +// transaction: the DROP+CREATE pair is atomic across connections, and a +// double-checked read skips it once any process has stamped the version. +function applyFtsSyncTriggers(db: Database): void { + if (getUserVersion(db) >= STORE_SCHEMA_VERSION) return; + db.exec(`BEGIN IMMEDIATE`); + try { + if (getUserVersion(db) < STORE_SCHEMA_VERSION) { + db.exec(`DROP TRIGGER IF EXISTS documents_ai`); + db.exec(` + CREATE TRIGGER documents_ai AFTER INSERT ON documents + WHEN new.active = 1 + BEGIN + INSERT INTO documents_fts(rowid, filepath, title, body) + SELECT + new.id, + new.collection || '/' || new.path, + new.title, + (SELECT doc FROM content WHERE hash = new.hash) + WHERE new.active = 1; + END + `); + + db.exec(`DROP TRIGGER IF EXISTS documents_ad`); + db.exec(` + CREATE TRIGGER documents_ad AFTER DELETE ON documents BEGIN + DELETE FROM documents_fts WHERE rowid = old.id; + END + `); + + db.exec(`DROP TRIGGER IF EXISTS documents_au`); + db.exec(` + CREATE TRIGGER documents_au AFTER UPDATE ON documents + BEGIN + -- Delete from FTS if no longer active + DELETE FROM documents_fts WHERE rowid = old.id AND new.active = 0; + + -- Update FTS if still/newly active + INSERT OR REPLACE INTO documents_fts(rowid, filepath, title, body) + SELECT + new.id, + new.collection || '/' || new.path, + new.title, + (SELECT doc FROM content WHERE hash = new.hash) + WHERE new.active = 1; + END + `); + + db.exec(`PRAGMA user_version = ${STORE_SCHEMA_VERSION}`); + } + db.exec(`COMMIT`); + } catch (err) { + db.exec(`ROLLBACK`); + throw err; + } +} + function initializeDatabase(db: Database): void { try { loadSqliteVec(db); @@ -830,7 +904,6 @@ function initializeDatabase(db: Database): void { _sqliteVecUnavailableReason = getErrorMessage(err); console.warn(_sqliteVecUnavailableReason); } - db.exec("PRAGMA journal_mode = WAL"); db.exec("PRAGMA foreign_keys = ON"); // Drop legacy tables that are now managed in YAML @@ -920,48 +993,7 @@ function initializeDatabase(db: Database): void { ) `); - // Triggers keep FTS in sync for callers that write directly to documents. - // Production indexing paths rebuild entries in TypeScript so CJK text can be - // normalized before it reaches the unicode61 tokenizer. - db.exec(`DROP TRIGGER IF EXISTS documents_ai`); - db.exec(` - CREATE TRIGGER documents_ai AFTER INSERT ON documents - WHEN new.active = 1 - BEGIN - INSERT INTO documents_fts(rowid, filepath, title, body) - SELECT - new.id, - new.collection || '/' || new.path, - new.title, - (SELECT doc FROM content WHERE hash = new.hash) - WHERE new.active = 1; - END - `); - - db.exec(`DROP TRIGGER IF EXISTS documents_ad`); - db.exec(` - CREATE TRIGGER documents_ad AFTER DELETE ON documents BEGIN - DELETE FROM documents_fts WHERE rowid = old.id; - END - `); - - db.exec(`DROP TRIGGER IF EXISTS documents_au`); - db.exec(` - CREATE TRIGGER documents_au AFTER UPDATE ON documents - BEGIN - -- Delete from FTS if no longer active - DELETE FROM documents_fts WHERE rowid = old.id AND new.active = 0; - - -- Update FTS if still/newly active - INSERT OR REPLACE INTO documents_fts(rowid, filepath, title, body) - SELECT - new.id, - new.collection || '/' || new.path, - new.title, - (SELECT doc FROM content WHERE hash = new.hash) - WHERE new.active = 1; - END - `); + applyFtsSyncTriggers(db); rebuildFTSForCjkNormalization(db); } diff --git a/test/_helpers/store-init-worker.ts b/test/_helpers/store-init-worker.ts new file mode 100644 index 000000000..c08ab063b --- /dev/null +++ b/test/_helpers/store-init-worker.ts @@ -0,0 +1,30 @@ +/** + * store-init-worker - one subprocess that opens a store and exits. + * + * Spawned N-up by store-concurrency.test.ts to reproduce the cross-connection + * schema-init race. Each worker busy-waits until a shared wall-clock start time + * so every process hits initializeDatabase at once, then opens the store. Exits + * 0 on success, 1 if createStore throws (e.g. "trigger already exists"). + */ +import { createStore } from "../../src/store.ts"; + +const dbPath = process.argv[2]; +const startAtMs = Number(process.argv[3] ?? "0"); + +if (!dbPath) { + console.error("usage: store-init-worker [startAtMs]"); + process.exit(2); +} + +while (Date.now() < startAtMs) { + // Align all workers to the same start so their DROP/CREATE windows overlap. +} + +try { + const store = createStore(dbPath); + store.close(); + process.exit(0); +} catch (err) { + console.error(err instanceof Error ? err.message : String(err)); + process.exit(1); +} diff --git a/test/db.test.ts b/test/db.test.ts new file mode 100644 index 000000000..f2b3abca8 --- /dev/null +++ b/test/db.test.ts @@ -0,0 +1,128 @@ +/** + * db.test.ts - openDatabase configuration + */ + +import { describe, test, expect, afterEach } from "vitest"; +import { openDatabase } from "../src/db.js"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +const DEFAULT_BUSY_TIMEOUT_MS = 120_000; + +function readBusyTimeout(db: ReturnType): number { + const row = db.prepare("PRAGMA busy_timeout").get() as Record; + const value = Object.values(row)[0]; + return typeof value === "number" ? value : Number(value); +} + +describe("openDatabase", () => { + const originalEnv = process.env.QMD_SQLITE_BUSY_TIMEOUT; + afterEach(() => { + if (originalEnv === undefined) delete process.env.QMD_SQLITE_BUSY_TIMEOUT; + else process.env.QMD_SQLITE_BUSY_TIMEOUT = originalEnv; + }); + + test("sets the default busy_timeout so concurrent writers wait for the lock", () => { + delete process.env.QMD_SQLITE_BUSY_TIMEOUT; + const db = openDatabase(":memory:"); + try { + expect(readBusyTimeout(db)).toBe(DEFAULT_BUSY_TIMEOUT_MS); + } finally { + db.close(); + } + }); + + test("applies the busy_timeout to each independently opened connection", async () => { + delete process.env.QMD_SQLITE_BUSY_TIMEOUT; + const dir = await mkdtemp(join(tmpdir(), "qmd-busy-")); + const dbPath = join(dir, "shared.sqlite"); + try { + const a = openDatabase(dbPath); + const b = openDatabase(dbPath); + try { + expect(readBusyTimeout(a)).toBe(DEFAULT_BUSY_TIMEOUT_MS); + expect(readBusyTimeout(b)).toBe(DEFAULT_BUSY_TIMEOUT_MS); + } finally { + a.close(); + b.close(); + } + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); + + test("QMD_SQLITE_BUSY_TIMEOUT overrides the default", () => { + process.env.QMD_SQLITE_BUSY_TIMEOUT = "750"; + const db = openDatabase(":memory:"); + try { + expect(readBusyTimeout(db)).toBe(750); + } finally { + db.close(); + } + }); + + test("QMD_SQLITE_BUSY_TIMEOUT=0 restores fail-fast", () => { + process.env.QMD_SQLITE_BUSY_TIMEOUT = "0"; + const db = openDatabase(":memory:"); + try { + expect(readBusyTimeout(db)).toBe(0); + } finally { + db.close(); + } + }); + + test("ignores unparseable QMD_SQLITE_BUSY_TIMEOUT and falls back to the default", () => { + process.env.QMD_SQLITE_BUSY_TIMEOUT = "not-a-number"; + const db = openDatabase(":memory:"); + try { + expect(readBusyTimeout(db)).toBe(DEFAULT_BUSY_TIMEOUT_MS); + } finally { + db.close(); + } + }); + + test("SQLite honors the configured busy_timeout when another connection holds the write lock", async () => { + const dir = await mkdtemp(join(tmpdir(), "qmd-busy-")); + const dbPath = join(dir, "contention.sqlite"); + try { + const setup = openDatabase(dbPath); + setup.exec("PRAGMA journal_mode = WAL"); + setup.exec("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)"); + setup.close(); + + const holder = openDatabase(dbPath); + const waiter = openDatabase(dbPath); + try { + // The synchronous SQLite API blocks the thread while it waits for the + // lock, so the test can't release the holder mid-wait. Shorten the + // waiter's timeout so the test finishes quickly; openDatabase already + // proved (above) that the default is the full 120_000ms. + waiter.exec("PRAGMA busy_timeout = 250"); + + holder.exec("BEGIN IMMEDIATE"); + holder.prepare("INSERT INTO t (v) VALUES ('holder')").run(); + + const start = Date.now(); + let threw: unknown = null; + try { + waiter.exec("BEGIN IMMEDIATE"); + } catch (err) { + threw = err; + } + const elapsed = Date.now() - start; + + expect(threw).toBeTruthy(); + expect(elapsed).toBeGreaterThanOrEqual(200); + expect(elapsed).toBeLessThan(2000); + + holder.exec("ROLLBACK"); + } finally { + holder.close(); + waiter.close(); + } + } finally { + await rm(dir, { recursive: true, force: true }); + } + }); +}); diff --git a/test/store-concurrency.test.ts b/test/store-concurrency.test.ts new file mode 100644 index 000000000..3ee2bf8e2 --- /dev/null +++ b/test/store-concurrency.test.ts @@ -0,0 +1,106 @@ +/** + * store-concurrency.test.ts - concurrent schema-init safety + * + * Reproduces the cross-connection race where two processes opening the same + * database interleave the FTS trigger DROP+CREATE and collide with "trigger + * already exists". JavaScript is single-threaded, so the race only appears + * across OS processes — each case spawns N workers that open the store at once. + * + * Fails against the pre-fix DROP+CREATE-on-every-open code; passes once the + * trigger rebuild is gated by PRAGMA user_version inside an IMMEDIATE + * transaction. + */ +import { describe, test, expect } from "vitest"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, dirname } from "node:path"; +import { fileURLToPath } from "node:url"; +import { spawn } from "node:child_process"; +import { openDatabase } from "../src/db.ts"; + +const thisDir = dirname(fileURLToPath(import.meta.url)); +const projectRoot = join(thisDir, ".."); +const workerScript = join(thisDir, "_helpers", "store-init-worker.ts"); +const tsxCli = join(projectRoot, "node_modules", "tsx", "dist", "cli.mjs"); +const isBunRuntime = typeof (globalThis as { Bun?: unknown }).Bun !== "undefined"; + +const WORKERS = 12; + +type WorkerResult = { code: number | null; stderr: string }; + +function runWorker(dbPath: string, startAtMs: number): Promise { + const args = isBunRuntime + ? [workerScript, dbPath, String(startAtMs)] + : [tsxCli, workerScript, dbPath, String(startAtMs)]; + return new Promise((resolve) => { + const proc = spawn(process.execPath, args, { stdio: ["ignore", "ignore", "pipe"] }); + let stderr = ""; + proc.stderr.on("data", (d: Buffer) => { stderr += d.toString(); }); + proc.on("close", (code) => resolve({ code, stderr })); + }); +} + +async function openConcurrently(dbPath: string, n: number): Promise { + const startAtMs = Date.now() + 1000; + return Promise.all(Array.from({ length: n }, () => runWorker(dbPath, startAtMs))); +} + +function expectAllSucceeded(results: WorkerResult[]): void { + const failed = results.filter(r => r.code !== 0); + // On failure the joined worker stderr is surfaced by the assertion below. + expect(failed.map(r => r.stderr.trim()).join("\n---\n")).toBe(""); + expect(failed).toHaveLength(0); +} + +function expectSchemaIntact(dbPath: string): void { + const db = openDatabase(dbPath); + try { + const triggers = db + .prepare(`SELECT name FROM sqlite_master WHERE type = 'trigger'`) + .all() as { name: string }[]; + expect(new Set(triggers.map(t => t.name))).toEqual( + new Set(["documents_ai", "documents_ad", "documents_au"]) + ); + + const fts = db + .prepare(`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'documents_fts'`) + .get(); + expect(fts).toBeTruthy(); + + const versionRow = db.prepare(`PRAGMA user_version`).get() as Record; + expect(Object.values(versionRow)[0]).toBeGreaterThanOrEqual(1); + } finally { + db.close(); + } +} + +describe("concurrent store initialization", () => { + test("cold database: N processes initialize without colliding on triggers", async () => { + const dir = await mkdtemp(join(tmpdir(), "qmd-store-concurrency-")); + const dbPath = join(dir, "index.sqlite"); + try { + const results = await openConcurrently(dbPath, WORKERS); + expectAllSucceeded(results); + expectSchemaIntact(dbPath); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }, 60_000); + + test("existing database: N processes reopen without rebuilding triggers", async () => { + const dir = await mkdtemp(join(tmpdir(), "qmd-store-concurrency-")); + const dbPath = join(dir, "index.sqlite"); + try { + // Stamp the schema once, single-process, so every concurrent reopen takes + // the version-gated fast path. + const [seed] = await openConcurrently(dbPath, 1); + expect(seed.code).toBe(0); + + const results = await openConcurrently(dbPath, WORKERS); + expectAllSucceeded(results); + expectSchemaIntact(dbPath); + } finally { + await rm(dir, { recursive: true, force: true }); + } + }, 60_000); +});