diff --git a/CHANGELOG.md b/CHANGELOG.md index 382db1efd..b039b53f8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,16 @@ ## [Unreleased] +### Fixed + +- `cleanupOrphanedVectors` now runs its orphan count and both DELETEs in a + single immediate transaction. An interruption between the two DELETEs + (crash, `SQLITE_BUSY`) could desync `vectors_vec` from `content_vectors`, + leaving stale metadata rows that make a later reactivation of the same + content hash look already-embedded — so `qmd embed` skips it and the + document becomes silently unsearchable by vector, with no orphan left to + clean up. + ## [2.6.3] - 2026-06-24 ### Added diff --git a/src/db.ts b/src/db.ts index 65b210495..c1242b1a5 100644 --- a/src/db.ts +++ b/src/db.ts @@ -129,7 +129,9 @@ export interface Database { exec(sql: string): void; prepare(sql: string): Statement; loadExtension(path: string): void; - transaction unknown>(fn: T): T; + // Both drivers return the wrapped function with variant methods attached + // (better-sqlite3 and bun:sqlite each expose .immediate/.deferred/.exclusive). + transaction unknown>(fn: T): T & { immediate: T }; close(): void; } diff --git a/src/store.ts b/src/store.ts index e4e3a622b..5a759046f 100644 --- a/src/store.ts +++ b/src/store.ts @@ -2431,36 +2431,55 @@ export function cleanupOrphanedVectors(db: Database): number { } return withLazyContentVectorMigration(db, () => { - // Count orphaned vectors first - const countResult = db.prepare(` - SELECT COUNT(*) as c FROM content_vectors cv - WHERE NOT EXISTS ( - SELECT 1 FROM documents d WHERE d.hash = cv.hash AND d.active = 1 - ) - `).get() as { c: number }; - - if (countResult.c === 0) { - return 0; - } - - // Delete from vectors_vec first - db.exec(` - DELETE FROM vectors_vec WHERE hash_seq IN ( - SELECT cv.hash || '_' || cv.seq FROM content_vectors cv + // Count and both DELETEs share one transaction. An interruption between the + // two DELETEs (crash, SQLITE_BUSY) desyncs the tables: vectors_vec loses + // the rows while content_vectors still records the chunks as embedded. + // These rows are orphaned (no active document), so live vector search — + // which post-filters on documents.active = 1 — is unaffected right away. + // The failure is latent: if that content hash is later reactivated (qmd is + // content-addressable, so the same content returning revives the hash), the + // stale content_vectors rows make getHashesNeedingEmbedding treat it as + // already embedded, so qmd embed skips it and the document is silently + // unsearchable by vector with no orphan left to clean up. Keeping the count + // inside the same transaction also makes the returned number match the rows + // the DELETEs actually remove if another connection mutates documents + // concurrently. Run it BEGIN IMMEDIATE: the count reads before the DELETEs + // write, and upgrading a deferred read snapshot under a concurrent WAL + // writer fails with SQLITE_BUSY_SNAPSHOT instead of honoring the busy + // timeout. Nested callers still get a savepoint. + const cleanup = db.transaction(() => { + const countResult = db.prepare(` + SELECT COUNT(*) as c FROM content_vectors cv WHERE NOT EXISTS ( SELECT 1 FROM documents d WHERE d.hash = cv.hash AND d.active = 1 ) - ) - `); + `).get() as { c: number }; - // Delete from content_vectors - db.exec(` - DELETE FROM content_vectors WHERE hash NOT IN ( - SELECT hash FROM documents WHERE active = 1 - ) - `); + if (countResult.c === 0) { + return 0; + } + + // Delete from vectors_vec first + db.exec(` + DELETE FROM vectors_vec WHERE hash_seq IN ( + SELECT cv.hash || '_' || cv.seq FROM content_vectors cv + WHERE NOT EXISTS ( + SELECT 1 FROM documents d WHERE d.hash = cv.hash AND d.active = 1 + ) + ) + `); + + // Delete from content_vectors + db.exec(` + DELETE FROM content_vectors WHERE hash NOT IN ( + SELECT hash FROM documents WHERE active = 1 + ) + `); + + return countResult.c; + }); - return countResult.c; + return cleanup.immediate(); }); } diff --git a/test/store.test.ts b/test/store.test.ts index 601e15e83..7de0e33ec 100644 --- a/test/store.test.ts +++ b/test/store.test.ts @@ -52,6 +52,7 @@ import { STRONG_SIGNAL_MIN_GAP, insertContent, insertDocument, + cleanupOrphanedVectors, generateEmbeddings, getHybridRrfWeights, _resetProductionModeForTesting, @@ -2730,6 +2731,138 @@ describe("Vector Table", () => { }); }); +describe("cleanupOrphanedVectors atomicity", () => { + // Seeds one active document (1 chunk) and one inactive document (2 chunks), + // so cleanup should remove exactly the 2 orphaned chunks from both tables. + async function seedOrphanFixture(store: Store): Promise { + const collectionName = await createTestCollection(); + const now = new Date().toISOString(); + + store.ensureVecTable(3); + await insertTestDocument(store.db, collectionName, { name: "kept-doc", hash: "keephash" }); + await insertTestDocument(store.db, collectionName, { name: "orphaned-doc", hash: "orphanhash", active: 0 }); + store.insertEmbedding("keephash", 0, 0, new Float32Array([1, 2, 3]), "test-model", now, 1); + store.insertEmbedding("orphanhash", 0, 0, new Float32Array([4, 5, 6]), "test-model", now, 2); + store.insertEmbedding("orphanhash", 1, 10, new Float32Array([7, 8, 9]), "test-model", now, 2); + } + + function vecCounts(db: Database): { vec: number; meta: number } { + const vec = (db.prepare(`SELECT COUNT(*) AS c FROM vectors_vec`).get() as { c: number }).c; + const meta = (db.prepare(`SELECT COUNT(*) AS c FROM content_vectors`).get() as { c: number }).c; + return { vec, meta }; + } + + // Fault injection: same connection, but the content_vectors DELETE throws — + // after the vectors_vec DELETE already executed inside the transaction. + function makeFailingDb(db: Database): Database { + return { + prepare: (sql: string) => db.prepare(sql), + transaction: (fn: () => unknown) => db.transaction(fn), + exec: (sql: string) => { + if (sql.includes("DELETE FROM content_vectors")) { + throw new Error("injected failure between deletes"); + } + return db.exec(sql); + }, + } as unknown as Database; + } + + test("removes orphaned chunks from both tables and returns the count", async () => { + const store = await createTestStore(); + try { + await seedOrphanFixture(store); + expect(vecCounts(store.db)).toEqual({ vec: 3, meta: 3 }); + + expect(cleanupOrphanedVectors(store.db)).toBe(2); + + expect(vecCounts(store.db)).toEqual({ vec: 1, meta: 1 }); + const survivor = store.db.prepare(`SELECT hash FROM content_vectors`).get() as { hash: string }; + expect(survivor.hash).toBe("keephash"); + const survivorVec = store.db.prepare(`SELECT hash_seq FROM vectors_vec`).get() as { hash_seq: string }; + expect(survivorVec.hash_seq).toBe("keephash_0"); + } finally { + await cleanupTestDb(store); + } + }); + + test("rolls back the vectors_vec DELETE when the content_vectors DELETE fails", async () => { + const store = await createTestStore(); + try { + await seedOrphanFixture(store); + const db = store.db; + + // Without the transaction wrap this used to leave vectors_vec already + // purged while content_vectors still claimed the chunks were embedded + // (silent desync). + expect(() => cleanupOrphanedVectors(makeFailingDb(db))).toThrow("injected failure between deletes"); + + // Both tables must be untouched — the vectors_vec DELETE was rolled back. + expect(vecCounts(db)).toEqual({ vec: 3, meta: 3 }); + + // The connection is left in a clean state: a plain retry succeeds. + expect(cleanupOrphanedVectors(db)).toBe(2); + expect(vecCounts(db)).toEqual({ vec: 1, meta: 1 }); + } finally { + await cleanupTestDb(store); + } + }); + + test("participates in an outer transaction via savepoint and rolls back with it", async () => { + const store = await createTestStore(); + try { + await seedOrphanFixture(store); + const db = store.db; + + const outer = db.transaction(() => { + const removed = cleanupOrphanedVectors(db); + if (removed !== 2) { + throw new Error(`expected 2 removed inside outer transaction, got ${removed}`); + } + throw new Error("outer rollback"); + }); + + expect(() => outer()).toThrow("outer rollback"); + + // The outer rollback must also restore the cleanup's deletions. + expect(vecCounts(db)).toEqual({ vec: 3, meta: 3 }); + } finally { + await cleanupTestDb(store); + } + }); + + test("runs as an inner savepoint: a caught cleanup failure rolls back alone, the outer transaction commits", async () => { + const store = await createTestStore(); + try { + await seedOrphanFixture(store); + const db = store.db; + + const outer = db.transaction(() => { + try { + cleanupOrphanedVectors(makeFailingDb(db)); + throw new Error("expected the injected failure to propagate"); + } catch (error) { + if (!(error instanceof Error) || error.message !== "injected failure between deletes") { + throw error; + } + } + // If the cleanup ran inline instead of inside its own savepoint, the + // vectors_vec DELETE would survive the caught failure and commit with + // the outer transaction below. + db.prepare(`INSERT INTO content (hash, doc, created_at) VALUES (?, ?, ?)`) + .run("outer-survivor", "outer doc", new Date().toISOString()); + }); + outer(); + + // Cleanup rolled back alone; the unrelated outer write committed. + expect(vecCounts(db)).toEqual({ vec: 3, meta: 3 }); + const kept = db.prepare(`SELECT COUNT(*) AS c FROM content WHERE hash = 'outer-survivor'`).get() as { c: number }; + expect(kept.c).toBe(1); + } finally { + await cleanupTestDb(store); + } + }); +}); + // ============================================================================= // Integration Tests // =============================================================================