diff --git a/src/main/ai-vault-search/session-search-index-consumer.ts b/src/main/ai-vault-search/session-search-index-consumer.ts index bce9a3097b4..f2d10bb2f87 100644 --- a/src/main/ai-vault-search/session-search-index-consumer.ts +++ b/src/main/ai-vault-search/session-search-index-consumer.ts @@ -106,7 +106,7 @@ class SessionSearchReadConsumer implements TranscriptReadConsumer { this.store.writePublished(candidate) return } - this.store.markStale(candidate) + this.store.writeAbandoned(candidate) } } diff --git a/src/main/ai-vault-search/session-search-pending-deletes.ts b/src/main/ai-vault-search/session-search-pending-deletes.ts index 69de9f9f603..7de70a37940 100644 --- a/src/main/ai-vault-search/session-search-pending-deletes.ts +++ b/src/main/ai-vault-search/session-search-pending-deletes.ts @@ -28,8 +28,13 @@ export function discardSearchBatch( export function recoverSearchWrites(db: SyncDatabase): void { // A staging session or a batch row that outlived its writer is by definition unfinished: // publish clears both in the same transaction that makes the rows visible. + // The batch half skips a batch whose session is already retired: that tombstone + // covers the same rows, and without the guard every reopen adds one more pass + // over them. db.exec(`INSERT OR IGNORE INTO search_pending_deletes(path,session_row_id) SELECT char(0)||'session:'||id,id FROM sessions WHERE index_ready=0; INSERT OR IGNORE INTO search_pending_deletes(path,session_row_id,batch_id) - SELECT char(0)||'batch:'||id,session_row_id,id FROM search_write_batches`) + SELECT char(0)||'batch:'||b.id,b.session_row_id,b.id FROM search_write_batches b + WHERE NOT EXISTS (SELECT 1 FROM search_pending_deletes d + WHERE d.session_row_id=b.session_row_id AND d.batch_id IS NULL)`) } diff --git a/src/main/ai-vault-search/session-search-schema.test.ts b/src/main/ai-vault-search/session-search-schema.test.ts index cbcd24ef3ed..4702622ad8e 100644 --- a/src/main/ai-vault-search/session-search-schema.test.ts +++ b/src/main/ai-vault-search/session-search-schema.test.ts @@ -255,9 +255,29 @@ it('retires an unfinished batch as part of opening, not of using', async () => { const reopened = openSessionSearchDatabase(path) try { - expect(reopened.prepare('SELECT count(*) AS n FROM search_pending_deletes').get()).toEqual({ - n: 2 - }) + // One tombstone, not two: the session-keyed one already covers every row the + // batch holds, and a per-reopen duplicate would replay the same deletes. + expect(reopened.prepare('SELECT path FROM search_pending_deletes').all()).toEqual([ + { path: '\u0000session:1' } + ]) + } finally { + reopened.close() + } +}) + +it('retires a batch whose session survived it', async () => { + const path = await tempDatabasePath() + const crashed = openSessionSearchDatabase(path) + crashed.exec(`INSERT INTO sessions(id,index_ready,agent,session_id,file_path,title,resume_command) + VALUES (1,1,'claude','a','a','published',''); + INSERT INTO search_write_batches(id,session_row_id) VALUES (7,1)`) + crashed.close() + + const reopened = openSessionSearchDatabase(path) + try { + expect(reopened.prepare('SELECT path,batch_id FROM search_pending_deletes').all()).toEqual([ + { path: '\u0000batch:7', batch_id: 7 } + ]) } finally { reopened.close() } diff --git a/src/main/ai-vault-search/session-search-staged-write.test.ts b/src/main/ai-vault-search/session-search-staged-write.test.ts index a65d3409606..79334248cd9 100644 --- a/src/main/ai-vault-search/session-search-staged-write.test.ts +++ b/src/main/ai-vault-search/session-search-staged-write.test.ts @@ -176,6 +176,52 @@ it('retires a batch appended onto a live session when the writer dies', async () } }) +it('drains the rows an incomplete read staged, without waiting for another write', async () => { + replayTranscriptRead({ + messages: userMessages('incompleteread', 300), + outcome: { incomplete: true } + }) + // Two flushes reached disk before the read turned out not to be publishable. + expect(counts(index.db).rawMessages).toBe(2 * SEARCH_WRITE_ROWS_PER_STEP) + await store.settled() + + const after = counts(index.db) + expect(after.rawMessages).toBe(0) + expect(after.full).toBe(0) + expect(after.conversation).toBe(0) + expect(after.tombstones).toBe(0) + expect(after.batches).toBe(0) + // The file is still owed a whole re-read; only its staged rows are gone. + expect(store.pendingFileCount).toBe(1) + expect(errors).toEqual([]) +}) + +it('drains what a dead writer left, under one tombstone, when the store reopens', async () => { + const staged = store.beginWrite(syntheticCandidate(), 'replace', 0)! + for (const message of userMessages('crashedrows', 200)) { + staged.add(message) + } + // No discard and no publish: the process died mid-write. + store.close() + + const reopened = new SessionSearchStore(index.path, (error) => errors.push(error)) + try { + // One tombstone covers the staging session and its batch alike; a second + // would replay the same deletes, and would be written again on every reopen. + expect(counts(index.db).tombstones).toBe(1) + await reopened.settled() + const after = counts(index.db) + expect(after.rawMessages).toBe(0) + expect(after.full).toBe(0) + expect(after.tombstones).toBe(0) + expect(after.batches).toBe(0) + expect(errors).toEqual([]) + } finally { + reopened.close() + store = new SessionSearchStore(index.path, (error) => errors.push(error)) + } +}) + it('replaces the previous generation without ever showing both', async () => { replayTranscriptRead({ messages: userMessages('firstgeneration', 10) }) await store.settled() diff --git a/src/main/ai-vault-search/session-search-store.ts b/src/main/ai-vault-search/session-search-store.ts index b4f0d3db5e4..44ed7c46b71 100644 --- a/src/main/ai-vault-search/session-search-store.ts +++ b/src/main/ai-vault-search/session-search-store.ts @@ -55,6 +55,10 @@ export class SessionSearchStore { ) { this.db = openSessionSearchDatabase(path) this.writer = new SessionSearchIndexWriter(this.db, options.walBudgetBytes) + // Opening tombstones whatever a dead writer left staged. Nothing else will + // schedule that drain: a store that is only ever read from, or one whose + // next read declines, would carry those rows for the life of the index. + this.scheduleCleanup() } setAcceptingWrites(accept: boolean): void { @@ -117,6 +121,17 @@ export class SessionSearchStore { this.scheduleCleanup() } + /** + * A read that staged rows and then could not publish them. The tombstone its + * discard wrote needs the same drain a publish gets, or the staged rows sit in + * `messages` and both FTS tables until some unrelated write happens to + * schedule a pass — which for the last read before a shutdown is never. + */ + writeAbandoned(candidate: SessionFileCandidate): void { + this.markStale(candidate) + this.scheduleCleanup() + } + reportWriteFailure(error: unknown): void { this.writeFailures += 1 this.onError(error) @@ -263,8 +278,15 @@ export class SessionSearchStore { }) } - /** Tests only: the cleanup lane is fire-and-forget everywhere else. */ - settled(): Promise { - return this.cleanup ?? Promise.resolve() + /** + * Tests only: the cleanup lane is fire-and-forget everywhere else. Loops + * because a pass requested while one was running is scheduled from the + * finished pass's own continuation, so awaiting a single promise would return + * with work still queued. + */ + async settled(): Promise { + while (this.cleanup) { + await this.cleanup + } } }