diff --git a/docs/storage-architecture.md b/docs/storage-architecture.md index 3d5e9266..c76331b7 100644 --- a/docs/storage-architecture.md +++ b/docs/storage-architecture.md @@ -39,6 +39,49 @@ SQLite / workspace-portable files / application preferences | Application preferences | `electron-store` | user configuration directory | small values only; localStorage may be a disposable UI cache, never authority | | Runtime queues and previews | memory/sessionStorage | renderer lifetime | explicitly non-durable | +## SQLite boundaries + +Adnify owns exactly two SQLite domains. They intentionally do not share a file, +connection, schema, or retention policy: + +| Database | Location | Authority | Main access pattern | +| --- | --- | --- | --- | +| Workspace session database | user configuration `session-storage/.sqlite3` | authoritative | catalog ordering, one active thread's ordered messages, incremental tail replacement | +| Structural index cache | workspace cache `structural-index.sqlite` | disposable and rebuildable | generation-scoped bulk writes, composite-key pagination, per-file replacement | + +Both databases are accessed only by dedicated workers. Renderer and Electron's +main event loop never execute SQLite statements. The SQLite MCP preset is user +configuration for an external MCP server and is not an Adnify storage backend. + +The session schema normalizes threads, messages, branches, plans, large blobs, +and message-to-blob references. Blob liveness is determined by indexed foreign +key relationships; commits never scan or parse the complete message history. +The structural cache normalizes file metadata away from chunks and uses +`(generation, relative_path, id)` as its storage and pagination order. + +## I/O budget + +- No SQLite task walks a drive or searches outside its explicit database and + companion blob directory. +- WAL checkpoints are passive and triggered by write count or a 30-second quiet + period. Clean shutdown may truncate the WAL once. +- Incremental vacuum runs only when at least 1,024 pages are free and free pages + exceed 20% of the database. It reclaims at most 256 pages per maintenance pass. +- Recovery snapshots are copied only after committed changes and at most once per + 24 hours. They are never refreshed on every message. +- Full SQLite integrity checks run after an unclean shutdown, a schema migration, + or when the previous check is at least seven days old. Clean restarts do not + reread the complete session database. +- Structural index reads use 512-row keyset pages. There is no offset scan and no + periodic full-project polling associated with SQLite. +- The native workspace watcher updates an index only after that index has been + explicitly built or loaded. Editing a file cannot silently create a partial + index database. +- Dependency, build-output, and application-cache globs are passed into the + native watcher backend, so those trees do not generate events that are merely + discarded later. Full indexing stats files before reading and uses four + bounded consumers; files above the configured limit are never read in full. + JSONL is not an online session backend. The session worker may read the legacy format exactly once inside a transaction. A migration marker prevents old data from being imported again after the user clears the database. @@ -104,5 +147,5 @@ The reproducible write-amplification benchmark is: ```sh pnpm build -node scripts/benchmark-session-storage.cjs +node scripts/benchmark-sqlite-storage.cjs ``` diff --git a/scripts/benchmark-sqlite-storage.cjs b/scripts/benchmark-sqlite-storage.cjs new file mode 100644 index 00000000..0eb765f2 --- /dev/null +++ b/scripts/benchmark-sqlite-storage.cjs @@ -0,0 +1,184 @@ +const { Worker } = require('node:worker_threads') +const { mkdtemp, rm, stat } = require('node:fs/promises') +const { tmpdir } = require('node:os') +const path = require('node:path') +const { performance } = require('node:perf_hooks') +const { randomUUID } = require('node:crypto') + +function createClient(workerPath) { + const worker = new Worker(workerPath) + const pending = new Map() + worker.on('message', response => { + const request = pending.get(response.requestId) + if (!request) return + pending.delete(response.requestId) + if (response.ok) request.resolve(response.result) + else request.reject(new Error(response.error)) + }) + worker.on('error', error => { + for (const request of pending.values()) request.reject(error) + pending.clear() + }) + return { + request(operation) { + const requestId = randomUUID() + return new Promise((resolve, reject) => { + pending.set(requestId, { resolve, reject }) + worker.postMessage({ requestId, operation }) + }) + }, + terminate: () => worker.terminate(), + } +} + +async function fileBytes(filePath) { + return stat(filePath).then(value => value.size).catch(() => 0) +} + +async function benchmarkStructural(root) { + const workerPath = path.resolve('dist/main/structuralIndexStore.worker.js') + const databasePath = path.join(root, 'structural.sqlite') + const client = createClient(workerPath) + const generation = randomUUID() + const fileCount = 2_000 + const chunkCount = 10_000 + const chunks = Array.from({ length: chunkCount }, (_, index) => { + const fileIndex = Math.floor(index / 5) + const relativePath = `src/file-${fileIndex}.ts` + return { + id: `chunk-${index}`, + filePath: path.join(root, relativePath), + relativePath, + fileHash: `hash-${fileIndex}`, + content: `export function symbol${index}() { return ${index} }`, + startLine: index % 5, + endLine: index % 5, + type: 'function', + language: 'typescript', + symbols: [`symbol${index}`], + } + }) + + const writeStart = performance.now() + await client.request({ type: 'beginReplace', databasePath, generation }) + for (let offset = 0; offset < chunks.length; offset += 512) { + await client.request({ + type: 'appendReplace', databasePath, generation, + chunks: chunks.slice(offset, offset + 512), + }) + } + await client.request({ + type: 'commitReplace', databasePath, generation, + metadata: { totalFiles: fileCount, totalChunks: chunkCount, savedAt: Date.now() }, + }) + const writeMs = performance.now() - writeStart + + const readStart = performance.now() + let cursor + let loaded = 0 + let pages = 0 + do { + const result = await client.request({ type: 'loadPage', databasePath, cursor }) + loaded += result.chunks.length + pages += 1 + cursor = result.nextCursor || undefined + } while (cursor) + const readMs = performance.now() - readStart + await client.request({ type: 'close', databasePath }) + await client.terminate() + + return { + files: fileCount, + chunks: loaded, + pages, + writeMs: Math.round(writeMs), + readMs: Math.round(readMs), + databaseBytes: await fileBytes(databasePath), + } +} + +async function benchmarkSessions(root) { + const workerPath = path.resolve('dist/main/sessionStorage.worker.js') + const databasePath = path.join(root, 'sessions.sqlite3') + const client = createClient(workerPath) + const threadCount = 100 + const messagesPerThread = 100 + const payloadText = 'session-payload-'.repeat(32) + const threads = Array.from({ length: threadCount }, (_, threadIndex) => ({ + metadata: { + id: `thread-${threadIndex}`, + createdAt: threadIndex, + lastModified: threadIndex, + title: `Thread ${threadIndex}`, + messageCount: messagesPerThread, + data: {}, + }, + replaceFrom: 0, + messages: Array.from({ length: messagesPerThread }, (_, ordinal) => ({ + ordinal, + id: `message-${threadIndex}-${ordinal}`, + role: ordinal % 2 === 0 ? 'user' : 'assistant', + timestamp: ordinal, + payload: { id: `message-${threadIndex}-${ordinal}`, content: payloadText }, + })), + })) + + await client.request({ type: 'open', databasePath }) + const writeStart = performance.now() + await client.request({ + type: 'applyPatch', databasePath, + patch: { threads, deletedThreadIds: [], branchThreads: [] }, + }) + const writeMs = performance.now() - writeStart + + const catalogStart = performance.now() + await client.request({ type: 'loadCatalog', databasePath }) + const catalogMs = performance.now() - catalogStart + const threadStart = performance.now() + await client.request({ type: 'loadMessages', databasePath, threadId: 'thread-50' }) + const threadMs = performance.now() - threadStart + + const tailStart = performance.now() + await client.request({ + type: 'applyPatch', databasePath, + patch: { + deletedThreadIds: [], branchThreads: [], + threads: [{ + metadata: { ...threads[50].metadata, lastModified: Date.now() }, + replaceFrom: 99, + messages: [{ ...threads[50].messages[99], payload: { id: 'tail', content: 'updated' } }], + }], + }, + }) + const tailWriteMs = performance.now() - tailStart + const stats = await client.request({ type: 'getStats', databasePath }) + await client.request({ type: 'closeAll' }) + await client.terminate() + + return { + threads: threadCount, + messages: threadCount * messagesPerThread, + writeMs: Math.round(writeMs), + catalogMs: Math.round(catalogMs), + activeThreadMs: Math.round(threadMs), + tailWriteMs: Math.round(tailWriteMs), + databaseBytes: stats.stats.databaseBytes, + walBytesBeforeClose: stats.stats.walBytes, + } +} + +async function main() { + const root = await mkdtemp(path.join(tmpdir(), 'adnify-sqlite-benchmark-')) + try { + const structural = await benchmarkStructural(root) + const sessions = await benchmarkSessions(root) + process.stdout.write(`${JSON.stringify({ structural, sessions }, null, 2)}\n`) + } finally { + await rm(root, { recursive: true, force: true }) + } +} + +main().catch(error => { + process.stderr.write(`${error.stack || error}\n`) + process.exitCode = 1 +}) diff --git a/src/main/indexing/indexService.ts b/src/main/indexing/indexService.ts index 152432f0..a1a09387 100644 --- a/src/main/indexing/indexService.ts +++ b/src/main/indexing/indexService.ts @@ -11,8 +11,6 @@ import { randomUUID } from 'crypto' import type { BrowserWindow } from 'electron' import { Worker } from 'worker_threads' import { logger, normalizePath } from '@shared/utils' -import { TreeSitterChunker } from './treeSitterChunker' -import { ChunkerService } from './chunker' import { EmbeddingService } from './embedder' import { VectorStoreService } from './vectorStore' import { BM25Index, SymbolIndex, rerankCandidates } from './search' @@ -29,7 +27,19 @@ import { getUserConfigDir, getWorkspaceCacheDir } from '../services/configPath' interface WorkerStructuralResultMessage { type: 'structural_result'; requestId: number; chunks: CodeChunk[]; processed: number; total: number } interface WorkerResultMessage { type: 'result'; chunks: IndexedChunk[]; processed: number; total: number } interface WorkerUpdateResultMessage { type: 'update_result'; filePath: string; chunks: IndexedChunk[]; deleted: boolean } -interface WorkerBatchUpdateResultMessage { type: 'batch_update_result'; requestId: number; results: Array<{ filePath: string; chunks: IndexedChunk[]; deleted: boolean }> } +type WorkerBatchUpdateResultMessage = + | { + type: 'batch_update_result' + mode: 'structural' + requestId: number + results: Array<{ filePath: string; chunks: CodeChunk[]; deleted: boolean }> + } + | { + type: 'batch_update_result' + mode: 'semantic' + requestId: number + results: Array<{ filePath: string; chunks: IndexedChunk[]; deleted: boolean }> + } interface WorkerCompleteMessage { type: 'complete'; totalChunks: number } interface WorkerErrorMessage { type: 'error'; error: string; requestId?: number } type WorkerMessage = @@ -49,8 +59,6 @@ export class CodebaseIndexService { private mainWindow: BrowserWindow | null = null // 结构化索引组件 - private chunker: TreeSitterChunker - private fallbackChunker: ChunkerService private bm25Index: BM25Index private symbolIndex: SymbolIndex private summaryGenerator: ProjectSummaryGenerator @@ -95,8 +103,6 @@ export class CodebaseIndexService { this.status.mode = this.config.mode // 初始化结构化索引组件 - this.chunker = new TreeSitterChunker(this.config) - this.fallbackChunker = new ChunkerService(this.config) this.bm25Index = new BM25Index() this.symbolIndex = new SymbolIndex() this.summaryGenerator = new ProjectSummaryGenerator( @@ -130,8 +136,6 @@ export class CodebaseIndexService { if (this.initialized) return if (this.destroyed) throw new Error('Index service has been destroyed') - await this.chunker.init() - // 加载缓存的项目摘要 this.projectSummary = await this.summaryGenerator.loadCache() @@ -173,9 +177,8 @@ export class CodebaseIndexService { return } - this.bm25Index.build() this.status.totalChunks = metadata.totalChunks - this.status.indexedFiles = this.symbolIndex.fileCount + this.status.indexedFiles = this.bm25Index.fileCount this.status.totalFiles = metadata.totalFiles this.status.lastIndexedAt = metadata.savedAt logger.index.info(`[IndexService] Loaded structural index: ${this.bm25Index.size} chunks, ${this.symbolIndex.size} symbols`) @@ -430,74 +433,24 @@ export class CodebaseIndexService { /** 批量更新文件(用于文件监听) */ async updateFiles(filePaths: string[]): Promise { if (filePaths.length === 0 || this.destroyed) return - return this.enqueueMutation(() => this.performUpdateFiles(filePaths)) + // The shared native watcher exists independently of code search. Do not + // create a partial SQLite index merely because an ordinary file changed. + if (!this.initialized && !this.pendingFullIndex) return + if (!this.pendingFullIndex && this.status.lastIndexedAt === undefined) return + return this.enqueueMutation(async () => { + // A queued full build may have failed. Never turn its trailing watcher + // events into an incomplete index. + if (this.status.lastIndexedAt === undefined) return + await this.performUpdateFiles(filePaths) + }) } private async performUpdateFiles(filePaths: string[]): Promise { logger.index.info(`[IndexService] Updating ${filePaths.length} files...`) - // 结构化模式:增量更新 - if (this.config.mode === 'structural') { - let updated = 0 - const persistedFiles: Array<{ relativePath: string; chunks: CodeChunk[] }> = [] - for (const filePath of filePaths) { - try { - const ext = path.extname(filePath).toLowerCase() - if (!this.config.includedExts.includes(ext)) continue - const relativePath = path.relative(this.workspacePath, filePath) - - // 检查文件是否存在 - if (!fs.existsSync(filePath)) { - // 文件被删除,从索引中移除 - await this.deleteFileFromStructuralIndex(filePath) - persistedFiles.push({ relativePath, chunks: [] }) - updated++ - continue - } - - const content = await fs.promises.readFile(filePath, 'utf-8') - if (content.length > this.config.maxFileSize) { - await this.deleteFileFromStructuralIndex(filePath) - persistedFiles.push({ relativePath, chunks: [] }) - updated++ - continue - } - - const chunks = await this.chunkFile(filePath, content) - - // 先删除该文件的旧索引 - await this.deleteFileFromStructuralIndex(filePath) - this.addStructuralChunks(chunks) - persistedFiles.push({ relativePath, chunks }) - updated++ - } catch (e) { - logger.index.warn(`[IndexService] Failed to update ${filePath}: `, e) - } - } - - if (updated > 0) { - // 重建 BM25 索引(必须调用以更新 IDF) - this.bm25Index.build() - this.status.totalChunks = this.bm25Index.size - this.status.indexedFiles = this.symbolIndex.fileCount - await this.structuralStore.request({ - type: 'applyFiles', - databasePath: this.structuralIndexPath, - files: persistedFiles, - metadata: { - totalFiles: this.status.totalFiles, - totalChunks: this.status.totalChunks, - savedAt: Date.now(), - }, - }) - logger.index.info(`[IndexService] Updated ${updated} files in structural index`) - } - return - } - - // 语义模式:通过 worker 处理。Promise 只在向量存储提交完成后解析, + // 两种模式都通过 worker 处理。Promise 只在内存索引和持久化提交完成后解析, // 这样文件监听缓冲器才能提供真正的背压,而不是无限 postMessage。 - await this.initSemanticComponents() + if (this.config.mode === 'semantic') await this.initSemanticComponents() if (!this.worker) this.initWorker() if (!this.worker) throw new Error('Index worker is unavailable') @@ -528,7 +481,12 @@ export class CodebaseIndexService { /** 删除文件索引 */ async deleteFileIndex(filePath: string): Promise { if (this.destroyed) return - return this.enqueueMutation(() => this.performDeleteFileIndex(filePath)) + if (!this.initialized && !this.pendingFullIndex) return + if (!this.pendingFullIndex && this.status.lastIndexedAt === undefined) return + return this.enqueueMutation(async () => { + if (this.status.lastIndexedAt === undefined) return + await this.performDeleteFileIndex(filePath) + }) } private async performDeleteFileIndex(filePath: string): Promise { @@ -537,9 +495,8 @@ export class CodebaseIndexService { // 结构化模式:从索引中删除 if (this.config.mode === 'structural') { await this.deleteFileFromStructuralIndex(filePath) - this.bm25Index.build() this.status.totalChunks = this.bm25Index.size - this.status.indexedFiles = this.symbolIndex.fileCount + this.status.indexedFiles = this.bm25Index.fileCount await this.structuralStore.request({ type: 'applyFiles', databasePath: this.structuralIndexPath, @@ -614,7 +571,7 @@ export class CodebaseIndexService { await this.startWorkerIndex() if (this.destroyed) return - this.bm25Index.build() + this.status.indexedFiles = this.bm25Index.fileCount this.projectSummary = this.summaryGenerator.generate( this.structuralBuildFileSymbols, this.structuralBuildLanguages, @@ -747,11 +704,39 @@ export class CodebaseIndexService { break case 'batch_update_result': - for (const res of message.results) { - if (res.deleted) { - await this.vectorStore!.deleteFile(res.filePath) - } else if (res.chunks?.length > 0) { - await this.vectorStore!.upsertFile(res.filePath, res.chunks) + if (message.mode === 'structural') { + const persistedFiles: Array<{ relativePath: string; chunks: CodeChunk[] }> = [] + for (const result of message.results) { + const relativePath = path.relative(this.workspacePath, result.filePath) + await this.deleteFileFromStructuralIndex(result.filePath) + if (!result.deleted) this.addStructuralChunks(result.chunks) + persistedFiles.push({ + relativePath, + chunks: result.deleted ? [] : result.chunks, + }) + } + this.status.totalChunks = this.bm25Index.size + this.status.indexedFiles = this.bm25Index.fileCount + if (persistedFiles.length > 0) { + await this.structuralStore.request({ + type: 'applyFiles', + databasePath: this.structuralIndexPath, + files: persistedFiles, + metadata: { + totalFiles: this.status.totalFiles, + totalChunks: this.status.totalChunks, + savedAt: Date.now(), + }, + }) + } + logger.index.info(`[IndexService] Updated ${persistedFiles.length} files in structural index`) + } else { + for (const result of message.results) { + if (result.deleted) { + await this.vectorStore!.deleteFile(result.filePath) + } else if (result.chunks.length > 0) { + await this.vectorStore!.upsertFile(result.filePath, result.chunks) + } } } this.emitProgress() @@ -869,15 +854,6 @@ export class CodebaseIndexService { } // ==================== 工具方法 ==================== - - private async chunkFile(filePath: string, content: string): Promise { - let chunks = await this.chunker.chunkFile(filePath, content, this.workspacePath) - if (chunks.length === 0) { - chunks = this.fallbackChunker.chunkFile(filePath, content, this.workspacePath) - } - return chunks - } - private extractKeywords(query: string): string[] { return query.split(/[\s,.:;!?()[\]{}'"<>]+/).map(t => t.trim()).filter(t => t.length >= 2 && !/^\d+$/.test(t)) } diff --git a/src/main/indexing/indexer.worker.ts b/src/main/indexing/indexer.worker.ts index a6c3b2be..d9241151 100644 --- a/src/main/indexing/indexer.worker.ts +++ b/src/main/indexing/indexer.worker.ts @@ -27,7 +27,18 @@ type WorkerResponse = | { type: 'structural_result'; requestId: number; chunks: CodeChunk[]; processed: number; total: number } | { type: 'result'; chunks: IndexedChunk[]; processed: number; total: number } | { type: 'update_result'; filePath: string; chunks: IndexedChunk[]; deleted: boolean } - | { type: 'batch_update_result'; requestId: number; results: Array<{ filePath: string; chunks: IndexedChunk[]; deleted: boolean }> } + | { + type: 'batch_update_result' + mode: 'structural' + requestId: number + results: Array<{ filePath: string; chunks: CodeChunk[]; deleted: boolean }> + } + | { + type: 'batch_update_result' + mode: 'semantic' + requestId: number + results: Array<{ filePath: string; chunks: IndexedChunk[]; deleted: boolean }> + } | { type: 'complete'; totalChunks: number } | { type: 'error'; error: string; requestId?: number } @@ -131,7 +142,6 @@ async function handleIndex( const { regexChunker, tsChunker } = await getChunkers(config) const embedder = config.mode === 'semantic' ? new EmbeddingService(config.embedding) : null - const limit = pLimit(4) // 预热 Embedder (触发模型加载) 并通知 UI if (config.mode === 'semantic' && config.embedding.provider === 'transformers') { @@ -149,7 +159,7 @@ async function handleIndex( let skippedFiles = 0 let pendingChunks: IndexedChunk[] = [] let pendingStructuralChunks: CodeChunk[] = [] - const RESULT_BATCH_SIZE = 50 + const resultBatchSize = config.mode === 'structural' ? 512 : 50 const flushChunks = async (): Promise => { if (pendingStructuralChunks.length > 0) { @@ -162,8 +172,13 @@ async function handleIndex( } } - const tasks = files.map(filePath => limit(async () => { + const processFile = async (filePath: string): Promise => { try { + const stats = await fs.stat(filePath) + if (!stats.isFile() || stats.size > config.maxFileSize) { + processedFiles++ + return + } const content = await fs.readFile(filePath, 'utf-8') if (content.length > config.maxFileSize) { @@ -171,16 +186,17 @@ async function handleIndex( return } - const currentHash = crypto.createHash('sha256').update(content).digest('hex') - - // 使用对象属性访问(不是 Map.get) - if (existingHashes && existingHashes[filePath] === currentHash) { - skippedFiles++ - processedFiles++ - if (processedFiles % 10 === 0) { - postResponse({ type: 'progress', processed: processedFiles, total: totalFiles }) + if (existingHashes) { + const currentHash = crypto.createHash('sha256').update(content).digest('hex') + // 使用对象属性访问(不是 Map.get) + if (existingHashes[filePath] === currentHash) { + skippedFiles++ + processedFiles++ + if (processedFiles % 10 === 0) { + postResponse({ type: 'progress', processed: processedFiles, total: totalFiles }) + } + return } - return } const chunks = await chunkFile(tsChunker, regexChunker, filePath, content, workspacePath) @@ -203,7 +219,7 @@ async function handleIndex( processedFiles++ - if (pendingChunks.length >= RESULT_BATCH_SIZE || pendingStructuralChunks.length >= RESULT_BATCH_SIZE) { + if (pendingChunks.length >= resultBatchSize || pendingStructuralChunks.length >= resultBatchSize) { await flushChunks() } else if (processedFiles % 10 === 0) { postResponse({ type: 'progress', processed: processedFiles, total: totalFiles }) @@ -212,9 +228,16 @@ async function handleIndex( logger.index.error(`Error processing file ${filePath}:`, error) processedFiles++ } - })) + } - await Promise.all(tasks) + let nextFileIndex = 0 + const consumers = Array.from({ length: Math.min(4, files.length) }, async () => { + while (nextFileIndex < files.length) { + const filePath = files[nextFileIndex++] + await processFile(filePath) + } + }) + await Promise.all(consumers) await flushChunks() logger.index.info(`[Worker] Indexing complete. Total: ${totalFiles}, Skipped: ${skippedFiles}, Chunks: ${totalChunks}`) @@ -231,42 +254,60 @@ async function handleBatchUpdate( config: IndexConfig, ): Promise { const { regexChunker, tsChunker } = await getChunkers(config) - const embedder = new EmbeddingService(config.embedding) - const limit = pLimit(5) // 批量更新时降低并发 - - const results: Array<{ filePath: string; chunks: IndexedChunk[]; deleted: boolean }> = [] + const embedder = config.mode === 'semantic' ? new EmbeddingService(config.embedding) : null + const limit = pLimit(4) + const structuralResults: Array<{ filePath: string; chunks: CodeChunk[]; deleted: boolean }> = [] + const semanticResults: Array<{ filePath: string; chunks: IndexedChunk[]; deleted: boolean }> = [] const tasks = files.map(filePath => limit(async () => { try { - // 检查文件是否存在 + const ext = path.extname(filePath).toLowerCase() + if (!config.includedExts.includes(ext)) return + + let stats try { - await fs.access(filePath) + stats = await fs.stat(filePath) } catch { - results.push({ filePath, chunks: [], deleted: true }) + if (config.mode === 'structural') structuralResults.push({ filePath, chunks: [], deleted: true }) + else semanticResults.push({ filePath, chunks: [], deleted: true }) + return + } + + if (!stats.isFile() || stats.size > config.maxFileSize) { + if (config.mode === 'structural') structuralResults.push({ filePath, chunks: [], deleted: true }) + else semanticResults.push({ filePath, chunks: [], deleted: true }) return } const content = await fs.readFile(filePath, 'utf-8') if (content.length > config.maxFileSize) { + if (config.mode === 'structural') structuralResults.push({ filePath, chunks: [], deleted: true }) + else semanticResults.push({ filePath, chunks: [], deleted: true }) return } const chunks = await chunkFile(tsChunker, regexChunker, filePath, content, workspacePath) if (chunks.length === 0) { - results.push({ filePath, chunks: [], deleted: true }) + if (config.mode === 'structural') structuralResults.push({ filePath, chunks: [], deleted: true }) + else semanticResults.push({ filePath, chunks: [], deleted: true }) + return + } + + if (config.mode === 'structural') { + structuralResults.push({ filePath, chunks, deleted: false }) return } const texts = chunks.map(c => prepareTextForEmbedding(c)) - const vectors = await embedder.embedBatch(texts) + const vectors = await embedder!.embedBatch(texts) const indexedChunks: IndexedChunk[] = chunks .map((chunk, idx) => vectors[idx] ? { ...chunk, vector: vectors[idx] } : null) .filter((c): c is IndexedChunk => c !== null) - results.push({ filePath, chunks: indexedChunks, deleted: false }) + semanticResults.push({ filePath, chunks: indexedChunks, deleted: false }) } catch (error) { logger.index.error(`[Worker] Error updating file ${filePath}:`, error) // 出错的文件跳过,不影响其他文件 @@ -275,7 +316,11 @@ async function handleBatchUpdate( await Promise.all(tasks) - postResponse({ type: 'batch_update_result', requestId, results }) + if (config.mode === 'structural') { + postResponse({ type: 'batch_update_result', mode: 'structural', requestId, results: structuralResults }) + } else { + postResponse({ type: 'batch_update_result', mode: 'semantic', requestId, results: semanticResults }) + } } /** diff --git a/src/main/indexing/search/bm25.ts b/src/main/indexing/search/bm25.ts index 3330f4f8..7666c3fa 100644 --- a/src/main/indexing/search/bm25.ts +++ b/src/main/indexing/search/bm25.ts @@ -23,9 +23,11 @@ export interface BM25Document { } export class BM25Index { - private documents: BM25Document[] = [] - private avgDocLength = 0 - private idf: Map = new Map() + private documents = new Set() + private documentsByFile = new Map>() + private documentFrequency = new Map() + private totalDocumentLength = 0 + /** 添加文档 */ addDocument(doc: Omit): void { const terms = this.tokenize(doc.content) @@ -33,7 +35,15 @@ export class BM25Index { for (const term of terms) { termFreq.set(term, (termFreq.get(term) || 0) + 1) } - this.documents.push({ ...doc, termFreq, docLength: terms.length }) + const indexedDocument = { ...doc, termFreq, docLength: terms.length } + this.documents.add(indexedDocument) + const fileDocuments = this.documentsByFile.get(doc.relativePath) || new Set() + fileDocuments.add(indexedDocument) + this.documentsByFile.set(doc.relativePath, fileDocuments) + this.totalDocumentLength += indexedDocument.docLength + for (const term of termFreq.keys()) { + this.documentFrequency.set(term, (this.documentFrequency.get(term) || 0) + 1) + } } /** 批量添加文档 */ @@ -43,44 +53,17 @@ export class BM25Index { } } - /** 构建索引(计算 IDF) */ - build(): void { - // 先清空:增量更新会让部分词彻底离开语料库, - // 若保留旧条目,这些词的 IDF 会一直参与评分,且 Map 无界增长。 - this.idf.clear() - - if (this.documents.length === 0) { - this.avgDocLength = 0 - return - } - - // 平均文档长度 - const totalLength = this.documents.reduce((sum, doc) => sum + doc.docLength, 0) - this.avgDocLength = totalLength / this.documents.length - - // 计算 IDF - const docFreq = new Map() - for (const doc of this.documents) { - const seenTerms = new Set() - for (const term of doc.termFreq.keys()) { - if (!seenTerms.has(term)) { - docFreq.set(term, (docFreq.get(term) || 0) + 1) - seenTerms.add(term) - } - } - } - - const N = this.documents.length - for (const [term, df] of docFreq) { - this.idf.set(term, Math.log((N - df + 0.5) / (df + 0.5) + 1)) - } - } - /** 搜索 */ search(query: string, topK: number = 10): SearchResult[] { - if (this.documents.length === 0) return [] + if (this.documents.size === 0) return [] const queryTerms = this.tokenize(query) + const documentCount = this.documents.size + const averageDocumentLength = this.totalDocumentLength / documentCount + const queryIdf = new Map(queryTerms.map(term => { + const frequency = this.documentFrequency.get(term) || 0 + return [term, Math.log((documentCount - frequency + 0.5) / (frequency + 0.5) + 1)] + })) const scores: { doc: BM25Document; score: number }[] = [] for (const doc of this.documents) { @@ -90,9 +73,11 @@ export class BM25Index { const tf = doc.termFreq.get(term) || 0 if (tf === 0) continue - const idf = this.idf.get(term) || 0 + const idf = queryIdf.get(term) || 0 const numerator = tf * (BM25_K1 + 1) - const denominator = tf + BM25_K1 * (1 - BM25_B + BM25_B * (doc.docLength / this.avgDocLength)) + const denominator = tf + BM25_K1 * ( + 1 - BM25_B + BM25_B * (doc.docLength / averageDocumentLength) + ) score += idf * (numerator / denominator) } @@ -127,31 +112,46 @@ export class BM25Index { /** 清空 */ clear(): void { - this.documents = [] - this.idf.clear() - this.avgDocLength = 0 + this.documents.clear() + this.documentsByFile.clear() + this.documentFrequency.clear() + this.totalDocumentLength = 0 } /** * 删除文件的所有文档 * - * 返回是否真的删除了内容,供调用者判断能否跳过 build()。 - * 注意:文档数变化后 IDF 已失效,调用者需在批量删除结束后调用 build()。 + * 返回是否真的删除了内容。 */ deleteFile(relativePath: string): boolean { - const before = this.documents.length - this.documents = this.documents.filter(doc => doc.relativePath !== relativePath) - return this.documents.length !== before + const fileDocuments = this.documentsByFile.get(relativePath) + if (!fileDocuments) return false + + for (const doc of fileDocuments) { + this.documents.delete(doc) + this.totalDocumentLength -= doc.docLength + for (const term of doc.termFreq.keys()) { + const nextFrequency = (this.documentFrequency.get(term) || 0) - 1 + if (nextFrequency <= 0) this.documentFrequency.delete(term) + else this.documentFrequency.set(term, nextFrequency) + } + } + this.documentsByFile.delete(relativePath) + return true } /** 获取文档数量 */ get size(): number { - return this.documents.length + return this.documents.size + } + + get fileCount(): number { + return this.documentsByFile.size } /** Number of searchable terms, useful for status and invariant checks. */ get vocabularySize(): number { - return this.idf.size + return this.documentFrequency.size } /** 分词 */ diff --git a/src/main/indexing/structuralIndexStore.worker.ts b/src/main/indexing/structuralIndexStore.worker.ts index 75c7d828..668cf7e7 100644 --- a/src/main/indexing/structuralIndexStore.worker.ts +++ b/src/main/indexing/structuralIndexStore.worker.ts @@ -1,5 +1,5 @@ import { DatabaseSync } from 'node:sqlite' -import { mkdirSync } from 'fs' +import { mkdirSync, rmSync } from 'fs' import * as path from 'path' import { parentPort } from 'worker_threads' import type { CodeChunk } from './types' @@ -13,21 +13,51 @@ import type { } from './structuralIndexStore.types' const databases = new Map() -const LOAD_BATCH_SIZE = 50 +// Keep worker IPC amortized without creating multi-megabyte structured-clone +// payloads for the main process. The old size of 50 made a 1M-chunk cache +// require 20,000 request/response turns during startup. +const LOAD_BATCH_SIZE = 512 +const SCHEMA_VERSION = 1 -function openDatabase(databasePath: string): DatabaseSync { - const existing = databases.get(databasePath) - if (existing) return existing +interface StoredChunkRow { + id: string + relative_path: string + file_path: string + file_hash: string + content: string + start_line: number + end_line: number + type: CodeChunk['type'] + language: string + symbols_json: string +} - mkdirSync(path.dirname(databasePath), { recursive: true }) - const database = new DatabaseSync(databasePath) +function removeDatabaseFiles(databasePath: string): void { + for (const target of [databasePath, `${databasePath}-wal`, `${databasePath}-shm`]) { + rmSync(target, { force: true }) + } +} + +function hasTables(database: DatabaseSync): boolean { + return Boolean(database.prepare( + "SELECT 1 FROM sqlite_master WHERE type = 'table' LIMIT 1", + ).get()) +} + +function createSchema(database: DatabaseSync): void { database.exec(` + PRAGMA page_size = 8192; PRAGMA journal_mode = WAL; PRAGMA synchronous = NORMAL; + PRAGMA foreign_keys = ON; PRAGMA temp_store = MEMORY; PRAGMA busy_timeout = 5000; + PRAGMA cache_size = -32768; + PRAGMA mmap_size = 268435456; + PRAGMA wal_autocheckpoint = 4096; + PRAGMA journal_size_limit = 16777216; - CREATE TABLE IF NOT EXISTS index_state ( + CREATE TABLE index_state ( singleton INTEGER PRIMARY KEY CHECK (singleton = 1), active_generation TEXT NOT NULL, total_files INTEGER NOT NULL CHECK (total_files >= 0), @@ -35,21 +65,63 @@ function openDatabase(databasePath: string): DatabaseSync { saved_at INTEGER NOT NULL ) STRICT; - CREATE TABLE IF NOT EXISTS chunks ( + CREATE TABLE files ( generation TEXT NOT NULL, - id TEXT NOT NULL, relative_path TEXT NOT NULL, - payload_json TEXT NOT NULL, - PRIMARY KEY (generation, id) + file_path TEXT NOT NULL, + file_hash TEXT NOT NULL, + PRIMARY KEY (generation, relative_path) ) WITHOUT ROWID, STRICT; - CREATE INDEX IF NOT EXISTS chunks_generation_file - ON chunks(generation, relative_path); + CREATE TABLE chunks ( + generation TEXT NOT NULL, + relative_path TEXT NOT NULL, + id TEXT NOT NULL, + content TEXT NOT NULL, + start_line INTEGER NOT NULL CHECK (start_line >= 0), + end_line INTEGER NOT NULL CHECK (end_line >= start_line), + type TEXT NOT NULL, + language TEXT NOT NULL, + symbols_json TEXT NOT NULL, + PRIMARY KEY (generation, relative_path, id), + FOREIGN KEY (generation, relative_path) + REFERENCES files(generation, relative_path) ON DELETE CASCADE + ) WITHOUT ROWID, STRICT; + + PRAGMA user_version = ${SCHEMA_VERSION}; + `) +} + +function openDatabase(databasePath: string): DatabaseSync { + const existing = databases.get(databasePath) + if (existing) return existing + + mkdirSync(path.dirname(databasePath), { recursive: true }) + let database = new DatabaseSync(databasePath) + const version = Number( + (database.prepare('PRAGMA user_version').get() as { user_version: number }).user_version, + ) + if (version !== SCHEMA_VERSION && (version !== 0 || hasTables(database))) { + database.close() + removeDatabaseFiles(databasePath) + database = new DatabaseSync(databasePath) + } + if (!hasTables(database)) createSchema(database) + else database.exec(` + PRAGMA journal_mode = WAL; + PRAGMA synchronous = NORMAL; + PRAGMA foreign_keys = ON; + PRAGMA temp_store = MEMORY; + PRAGMA busy_timeout = 5000; + PRAGMA cache_size = -32768; + PRAGMA mmap_size = 268435456; + PRAGMA wal_autocheckpoint = 4096; + PRAGMA journal_size_limit = 16777216; `) // A terminated rebuild may leave an uncommitted generation behind. Only the // active generation is authoritative, so orphan rows are safe to remove. database.exec(` - DELETE FROM chunks + DELETE FROM files WHERE generation NOT IN (SELECT active_generation FROM index_state WHERE singleton = 1) `) databases.set(databasePath, database) @@ -67,13 +139,46 @@ function inTransaction(database: DatabaseSync, operation: () => void): void { } } -function insertChunks(database: DatabaseSync, generation: string, chunks: CodeChunk[]): void { - const insert = database.prepare(` - INSERT OR REPLACE INTO chunks(generation, id, relative_path, payload_json) +function createChunkStatements(database: DatabaseSync) { + const insertFile = database.prepare(` + INSERT INTO files(generation, relative_path, file_path, file_hash) VALUES (?, ?, ?, ?) + ON CONFLICT(generation, relative_path) DO UPDATE SET + file_path = excluded.file_path, + file_hash = excluded.file_hash `) + const insertChunk = database.prepare(` + INSERT INTO chunks( + generation, relative_path, id, content, start_line, end_line, + type, language, symbols_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + `) + return { insertFile, insertChunk } +} + +function insertChunks( + statements: ReturnType, + generation: string, + chunks: CodeChunk[], +): void { + const { insertFile, insertChunk } = statements + const insertedFiles = new Set() for (const chunk of chunks) { - insert.run(generation, chunk.id, chunk.relativePath, JSON.stringify(chunk)) + if (!insertedFiles.has(chunk.relativePath)) { + insertFile.run(generation, chunk.relativePath, chunk.filePath, chunk.fileHash) + insertedFiles.add(chunk.relativePath) + } + insertChunk.run( + generation, + chunk.relativePath, + chunk.id, + chunk.content, + chunk.startLine, + chunk.endLine, + chunk.type, + chunk.language, + JSON.stringify(chunk.symbols || []), + ) } } @@ -106,31 +211,45 @@ function loadPage( const rows = cursor ? database.prepare(` - SELECT id, relative_path, payload_json FROM chunks - WHERE generation = ? - AND (relative_path > ? OR (relative_path = ? AND id > ?)) - ORDER BY relative_path, id LIMIT ? + SELECT + c.id, c.relative_path, f.file_path, f.file_hash, c.content, + c.start_line, c.end_line, c.type, c.language, c.symbols_json + FROM chunks c + JOIN files f USING (generation, relative_path) + WHERE c.generation = ? + AND (c.relative_path, c.id) > (?, ?) + ORDER BY c.relative_path, c.id LIMIT ? `).all( metadata.generation, cursor.relativePath, - cursor.relativePath, cursor.id, LOAD_BATCH_SIZE, - ) as Array<{ id: string; relative_path: string; payload_json: string }> + ) as unknown as StoredChunkRow[] : database.prepare(` - SELECT id, relative_path, payload_json FROM chunks - WHERE generation = ? ORDER BY relative_path, id LIMIT ? - `).all(metadata.generation, LOAD_BATCH_SIZE) as Array<{ - id: string - relative_path: string - payload_json: string - }> + SELECT + c.id, c.relative_path, f.file_path, f.file_hash, c.content, + c.start_line, c.end_line, c.type, c.language, c.symbols_json + FROM chunks c + JOIN files f USING (generation, relative_path) + WHERE c.generation = ? ORDER BY c.relative_path, c.id LIMIT ? + `).all(metadata.generation, LOAD_BATCH_SIZE) as unknown as StoredChunkRow[] const { generation: _generation, ...result } = metadata const last = rows.at(-1) return { type: 'loadedPage', metadata: result, - chunks: rows.map(row => JSON.parse(row.payload_json) as CodeChunk), + chunks: rows.map(row => ({ + id: row.id, + filePath: row.file_path, + relativePath: row.relative_path, + fileHash: row.file_hash, + content: row.content, + startLine: row.start_line, + endLine: row.end_line, + type: row.type, + language: row.language, + symbols: JSON.parse(row.symbols_json) as string[], + })), nextCursor: rows.length === LOAD_BATCH_SIZE && last ? { relativePath: last.relative_path, id: last.id } : null, @@ -145,10 +264,12 @@ export function executeStructuralIndexStoreOperation( case 'loadPage': return loadPage(database, operation.cursor) case 'beginReplace': - database.prepare('DELETE FROM chunks WHERE generation = ?').run(operation.generation) + database.prepare('DELETE FROM files WHERE generation = ?').run(operation.generation) return { type: 'ok' } case 'appendReplace': - inTransaction(database, () => insertChunks(database, operation.generation, operation.chunks)) + inTransaction(database, () => { + insertChunks(createChunkStatements(database), operation.generation, operation.chunks) + }) return { type: 'ok' } case 'commitReplace': { const row = database.prepare( @@ -174,23 +295,24 @@ export function executeStructuralIndexStoreOperation( operation.metadata.totalChunks, operation.metadata.savedAt, ) - database.prepare('DELETE FROM chunks WHERE generation <> ?').run(operation.generation) + database.prepare('DELETE FROM files WHERE generation <> ?').run(operation.generation) }) return { type: 'ok' } } case 'abortReplace': - database.prepare('DELETE FROM chunks WHERE generation = ?').run(operation.generation) + database.prepare('DELETE FROM files WHERE generation = ?').run(operation.generation) return { type: 'ok' } case 'applyFiles': { const current = readMetadata(database) const generation = current?.generation || `incremental-${operation.metadata.savedAt}` inTransaction(database, () => { + const statements = createChunkStatements(database) const remove = database.prepare( - 'DELETE FROM chunks WHERE generation = ? AND relative_path = ?', + 'DELETE FROM files WHERE generation = ? AND relative_path = ?', ) for (const file of operation.files) { remove.run(generation, file.relativePath) - insertChunks(database, generation, file.chunks) + insertChunks(statements, generation, file.chunks) } const row = database.prepare( 'SELECT COUNT(*) AS count FROM chunks WHERE generation = ?', @@ -218,10 +340,11 @@ export function executeStructuralIndexStoreOperation( } case 'clear': inTransaction(database, () => { - database.exec('DELETE FROM chunks; DELETE FROM index_state;') + database.exec('DELETE FROM files; DELETE FROM index_state;') }) return { type: 'ok' } case 'close': + database.exec('PRAGMA optimize') database.exec('PRAGMA wal_checkpoint(TRUNCATE)') database.close() databases.delete(operation.databasePath) diff --git a/src/main/security/fileWatcher.ts b/src/main/security/fileWatcher.ts index 51e06fcb..78ab6d78 100644 --- a/src/main/security/fileWatcher.ts +++ b/src/main/security/fileWatcher.ts @@ -36,7 +36,10 @@ interface WatcherEntry { } const DEFAULT_CONFIG: FileWatcherConfig = { - ignored: [/node_modules/, /\.git/, /dist/, /build/, /\.adnify/, '**/*.tmp', '**/*.temp'], + // String globs are passed to the native backend, preventing dependency and + // build trees from generating events at all. Git stays post-filtered because + // repositories whose metadata lives inside the workspace still need state signals. + ignored: ['**/node_modules/**', /\.git/, '**/dist/**', '**/build/**', '**/.adnify/**', '**/*.tmp', '**/*.temp'], persistent: true, ignoreInitial: true, bufferTimeMs: 500, @@ -64,7 +67,7 @@ function createIgnoreMatcher(patterns: (string | RegExp)[]): (path: string) => b for (const regex of regexPatterns) { if (regex.test(filePath)) return true } - if (globMatcher && globMatcher(filePath)) return true + if (globMatcher && globMatcher(filePath.replace(/\\/g, '/'))) return true return false } } diff --git a/src/main/services/session/sessionStorage.worker.ts b/src/main/services/session/sessionStorage.worker.ts index ffb0efb5..ed060e7f 100644 --- a/src/main/services/session/sessionStorage.worker.ts +++ b/src/main/services/session/sessionStorage.worker.ts @@ -23,6 +23,7 @@ interface OpenDatabase { writesSinceCheckpoint: number dirtySinceBackup: boolean idleCheckpoint: ReturnType | null + protectedBackupBlobHashes: Set | null } interface BlobDescriptor { @@ -52,14 +53,16 @@ const DEFAULT_STATE: SessionStateRecord = { activeBranchId: {}, version: 0, } -const LATEST_SCHEMA_VERSION = 4 +const LATEST_SCHEMA_VERSION = 5 const BLOB_THRESHOLD_BYTES = 256 * 1024 const IDLE_CHECKPOINT_MS = 30_000 const CHECKPOINT_WRITE_THRESHOLD = 32 const BACKUP_MIN_INTERVAL_MS = 24 * 60 * 60 * 1_000 +const INTEGRITY_CHECK_INTERVAL_MS = 7 * 24 * 60 * 60 * 1_000 function configure(database: DatabaseSync): void { database.exec(` + PRAGMA page_size = 8192; PRAGMA journal_mode = WAL; PRAGMA synchronous = FULL; PRAGMA foreign_keys = ON; @@ -67,6 +70,9 @@ function configure(database: DatabaseSync): void { PRAGMA wal_autocheckpoint = 0; PRAGMA journal_size_limit = 16777216; PRAGMA temp_store = MEMORY; + PRAGMA cache_size = -32768; + PRAGMA mmap_size = 268435456; + PRAGMA trusted_schema = OFF; `) } @@ -233,6 +239,113 @@ function migrateV4(database: DatabaseSync): void { `) } +function migrateV5(database: DatabaseSync): void { + if (tableExists(database, 'message_blobs') && tableExists(database, 'branch_message_blobs')) { + database.exec(` + CREATE INDEX IF NOT EXISTS threads_last_modified_id + ON threads(last_modified DESC, id ASC); + DROP INDEX IF EXISTS plans_updated_at; + CREATE INDEX IF NOT EXISTS plans_updated_at_id + ON plans(updated_at DESC, id ASC); + DROP INDEX IF EXISTS messages_thread_id_id; + CREATE TABLE IF NOT EXISTS maintenance_state ( + singleton INTEGER PRIMARY KEY CHECK (singleton = 1), + last_quick_check_at INTEGER NOT NULL, + clean_shutdown INTEGER NOT NULL CHECK (clean_shutdown IN (0, 1)) + ) STRICT; + INSERT OR IGNORE INTO maintenance_state(singleton, last_quick_check_at, clean_shutdown) + VALUES (1, 0, 1); + `) + return + } + + database.exec(` + CREATE INDEX IF NOT EXISTS threads_last_modified_id + ON threads(last_modified DESC, id ASC); + + DROP INDEX IF EXISTS plans_updated_at; + DROP INDEX IF EXISTS messages_thread_id_id; + CREATE INDEX IF NOT EXISTS plans_updated_at_id + ON plans(updated_at DESC, id ASC); + + CREATE TABLE blobs_v5 ( + hash TEXT PRIMARY KEY, + encoding TEXT NOT NULL CHECK (encoding IN ('utf8', 'base64')), + byte_length INTEGER NOT NULL CHECK (byte_length >= 0), + created_at INTEGER NOT NULL + ) STRICT; + + INSERT INTO blobs_v5(hash, encoding, byte_length, created_at) + SELECT hash, encoding, byte_length, created_at FROM blobs; + + CREATE TABLE message_blobs ( + thread_id TEXT NOT NULL, + ordinal INTEGER NOT NULL, + hash TEXT NOT NULL REFERENCES blobs_v5(hash) ON DELETE RESTRICT, + PRIMARY KEY (thread_id, ordinal, hash), + FOREIGN KEY (thread_id, ordinal) + REFERENCES messages(thread_id, ordinal) ON DELETE CASCADE + ) WITHOUT ROWID, STRICT; + + CREATE INDEX message_blobs_hash ON message_blobs(hash); + + CREATE TABLE branch_message_blobs ( + thread_id TEXT NOT NULL, + branch_id TEXT NOT NULL, + ordinal INTEGER NOT NULL, + hash TEXT NOT NULL REFERENCES blobs_v5(hash) ON DELETE RESTRICT, + PRIMARY KEY (thread_id, branch_id, ordinal, hash), + FOREIGN KEY (thread_id, branch_id, ordinal) + REFERENCES branch_messages(thread_id, branch_id, ordinal) ON DELETE CASCADE + ) WITHOUT ROWID, STRICT; + + CREATE INDEX branch_message_blobs_hash ON branch_message_blobs(hash); + + CREATE TABLE IF NOT EXISTS maintenance_state ( + singleton INTEGER PRIMARY KEY CHECK (singleton = 1), + last_quick_check_at INTEGER NOT NULL, + clean_shutdown INTEGER NOT NULL CHECK (clean_shutdown IN (0, 1)) + ) STRICT; + + INSERT OR IGNORE INTO maintenance_state(singleton, last_quick_check_at, clean_shutdown) + VALUES (1, 0, 1); + `) + + const insertMessageBlob = database.prepare(` + INSERT OR IGNORE INTO message_blobs(thread_id, ordinal, hash) VALUES (?, ?, ?) + `) + const insertBranchMessageBlob = database.prepare(` + INSERT OR IGNORE INTO branch_message_blobs(thread_id, branch_id, ordinal, hash) + VALUES (?, ?, ?, ?) + `) + const messages = database.prepare( + 'SELECT thread_id, ordinal, payload_json FROM messages', + ).all() as Array<{ thread_id: string; ordinal: number; payload_json: string }> + for (const row of messages) { + for (const reference of blobReferences(row.payload_json)) { + insertMessageBlob.run(row.thread_id, row.ordinal, reference.hash) + } + } + const branchMessages = database.prepare( + 'SELECT thread_id, branch_id, ordinal, payload_json FROM branch_messages', + ).all() as Array<{ + thread_id: string + branch_id: string + ordinal: number + payload_json: string + }> + for (const row of branchMessages) { + for (const reference of blobReferences(row.payload_json)) { + insertBranchMessageBlob.run(row.thread_id, row.branch_id, row.ordinal, reference.hash) + } + } + + database.exec(` + DROP TABLE blobs; + ALTER TABLE blobs_v5 RENAME TO blobs; + `) +} + function migrateSchema(database: DatabaseSync): boolean { let version = Number((database.prepare('PRAGMA user_version').get() as { user_version: number }).user_version) if (version > LATEST_SCHEMA_VERSION) { @@ -240,7 +353,7 @@ function migrateSchema(database: DatabaseSync): boolean { } const startingVersion = version - const migrations = [migrateV1, migrateV2, migrateV3, migrateV4] + const migrations = [migrateV1, migrateV2, migrateV3, migrateV4, migrateV5] while (version < LATEST_SCHEMA_VERSION) { const nextVersion = version + 1 database.exec('BEGIN IMMEDIATE') @@ -264,11 +377,10 @@ function assertHealthy(database: DatabaseSync): void { } } -function openCheckedDatabase(databasePath: string): DatabaseSync { +function openConfiguredDatabase(databasePath: string): DatabaseSync { const database = new DatabaseSync(databasePath) try { configure(database) - assertHealthy(database) return database } catch (error) { try { database.close() } catch { /* preserve original failure */ } @@ -321,21 +433,43 @@ async function recoverFromBackup(databasePath: string, openError: unknown): Prom if (await fileExists(source)) await fs.rename(source, `${source}${suffix}`) } await fs.copyFile(backupPath, databasePath) - return openCheckedDatabase(databasePath) + return openConfiguredDatabase(databasePath) } async function openHealthyDatabase(databasePath: string): Promise<{ database: DatabaseSync; migrated: boolean }> { - let database: DatabaseSync + let database: DatabaseSync | undefined + let checkedAt = 0 try { - database = openCheckedDatabase(databasePath) + database = openConfiguredDatabase(databasePath) + const maintenance = tableExists(database, 'maintenance_state') + ? database.prepare(` + SELECT last_quick_check_at, clean_shutdown + FROM maintenance_state WHERE singleton = 1 + `).get() as { last_quick_check_at: number; clean_shutdown: number } | undefined + : undefined + const now = Date.now() + if (!maintenance || maintenance.clean_shutdown !== 1 || + now - maintenance.last_quick_check_at >= INTEGRITY_CHECK_INTERVAL_MS) { + assertHealthy(database) + checkedAt = now + } } catch (error) { + try { database?.close() } catch { /* recovery owns the next connection */ } database = await recoverFromBackup(databasePath, error) + checkedAt = Date.now() } + if (!database) throw new Error('Session database failed to open') try { if (!tableExists(database, 'threads')) database.exec('PRAGMA auto_vacuum = INCREMENTAL') const schemaMigrated = migrateSchema(database) const payloadsMigrated = await externalizeLegacyPayloads(database, databasePath) + database.prepare(` + UPDATE maintenance_state + SET last_quick_check_at = CASE WHEN ? > 0 THEN ? ELSE last_quick_check_at END, + clean_shutdown = 0 + WHERE singleton = 1 + `).run(checkedAt, checkedAt) return { database, migrated: schemaMigrated || payloadsMigrated } } catch (error) { database.close() @@ -356,6 +490,7 @@ async function getDatabase(databasePath: string): Promise { writesSinceCheckpoint: 0, dirtySinceBackup: initialized.migrated, idleCheckpoint: null, + protectedBackupBlobHashes: null, } databases.set(databasePath, opened) return opened @@ -370,7 +505,11 @@ function blobPath(databasePath: string, hash: string): string { return path.join(blobDirectory(databasePath), hash.slice(0, 2), hash) } -async function writeBlob(databasePath: string, bytes: Buffer): Promise { +async function writeBlob( + databasePath: string, + bytes: Buffer, + createdHashes: Set, +): Promise { const hash = createHash('sha256').update(bytes).digest('hex') const target = blobPath(databasePath, hash) if (await fileExists(target)) return hash @@ -387,6 +526,7 @@ async function writeBlob(databasePath: string, bytes: Buffer): Promise { try { await fs.rename(temporary, target) await syncDirectory(path.dirname(target)) + createdHashes.add(hash) } catch (error) { await fs.rm(temporary, { force: true }) if (!await fileExists(target)) throw error @@ -406,6 +546,7 @@ async function externalizeValue( value: unknown, databasePath: string, blobs: BlobDescriptor[], + createdHashes: Set, key?: string, parent?: Record, ): Promise { @@ -414,7 +555,7 @@ async function externalizeValue( const bytes = isBase64 ? Buffer.from(value, 'base64') : Buffer.from(value, 'utf8') if (bytes.byteLength < BLOB_THRESHOLD_BYTES) return value const descriptor: BlobDescriptor = { - hash: await writeBlob(databasePath, bytes), + hash: await writeBlob(databasePath, bytes, createdHashes), encoding: isBase64 ? 'base64' : 'utf8', byteLength: bytes.byteLength, } @@ -422,23 +563,27 @@ async function externalizeValue( return { __adnifyBlob: 1, hash: descriptor.hash, encoding: descriptor.encoding } satisfies BlobReference } if (Array.isArray(value)) { - return Promise.all(value.map(item => externalizeValue(item, databasePath, blobs))) + return Promise.all(value.map(item => externalizeValue(item, databasePath, blobs, createdHashes))) } if (!value || typeof value !== 'object') return value const record = value as Record const entries = await Promise.all(Object.entries(record).map(async ([childKey, child]) => [ childKey, - await externalizeValue(child, databasePath, blobs, childKey, record), + await externalizeValue(child, databasePath, blobs, createdHashes, childKey, record), ] as const)) return Object.fromEntries(entries) } -async function prepareMessage(databasePath: string, message: SessionMessageWrite): Promise { +async function prepareMessage( + databasePath: string, + message: SessionMessageWrite, + createdHashes: Set, +): Promise { // Preserve JSON.stringify semantics (including rejecting BigInt) before any transaction begins. const compatible = JSON.parse(JSON.stringify(message.payload)) as unknown const blobs: BlobDescriptor[] = [] - const externalized = await externalizeValue(compatible, databasePath, blobs) + const externalized = await externalizeValue(compatible, databasePath, blobs, createdHashes) return { ordinal: message.ordinal, id: message.id, @@ -461,20 +606,31 @@ async function externalizeLegacyPayloads(database: DatabaseSync, databasePath: s ).all() as Array<{ thread_id: string; branch_id: string; ordinal: number; payload_json: string }> const preparedMessages: Array = [] const preparedBranches: Array = [] + const createdHashes = new Set() - for (const row of messageRows) { - const blobs: BlobDescriptor[] = [] - const nextJson = JSON.stringify(await externalizeValue(JSON.parse(row.payload_json), databasePath, blobs)) - if (blobs.length > 0) preparedMessages.push({ ...row, nextJson, blobs }) - } - for (const row of branchRows) { - const blobs: BlobDescriptor[] = [] - const nextJson = JSON.stringify(await externalizeValue(JSON.parse(row.payload_json), databasePath, blobs)) - if (blobs.length > 0) preparedBranches.push({ ...row, nextJson, blobs }) + try { + for (const row of messageRows) { + const blobs: BlobDescriptor[] = [] + const nextJson = JSON.stringify(await externalizeValue( + JSON.parse(row.payload_json), databasePath, blobs, createdHashes, + )) + if (blobs.length > 0) preparedMessages.push({ ...row, nextJson, blobs }) + } + for (const row of branchRows) { + const blobs: BlobDescriptor[] = [] + const nextJson = JSON.stringify(await externalizeValue( + JSON.parse(row.payload_json), databasePath, blobs, createdHashes, + )) + if (blobs.length > 0) preparedBranches.push({ ...row, nextJson, blobs }) + } + } catch (error) { + await removeUnreferencedCreatedBlobs(database, databasePath, createdHashes) + throw error } database.exec('BEGIN IMMEDIATE') try { + const blobStatements = createBlobStatements(database) const updateMessage = database.prepare( 'UPDATE messages SET payload_json = ? WHERE thread_id = ? AND ordinal = ?', ) @@ -482,18 +638,24 @@ async function externalizeLegacyPayloads(database: DatabaseSync, databasePath: s 'UPDATE branch_messages SET payload_json = ? WHERE thread_id = ? AND branch_id = ? AND ordinal = ?', ) for (const row of preparedMessages) { - incrementBlobRefs(database, row.blobs) updateMessage.run(row.nextJson, row.thread_id, row.ordinal) + insertBlobReferences(blobStatements, { + type: 'message', threadId: row.thread_id, ordinal: row.ordinal, + }, row.blobs) } for (const row of preparedBranches) { - incrementBlobRefs(database, row.blobs) updateBranch.run(row.nextJson, row.thread_id, row.branch_id, row.ordinal) + insertBlobReferences(blobStatements, { + type: 'branchMessage', threadId: row.thread_id, + branchId: row.branch_id, ordinal: row.ordinal, + }, row.blobs) } database.prepare('INSERT INTO migration_log(name, completed_at) VALUES (?, ?)') .run(migrationName, Date.now()) database.exec('COMMIT') } catch (error) { database.exec('ROLLBACK') + await removeUnreferencedCreatedBlobs(database, databasePath, createdHashes) throw error } return true @@ -527,53 +689,126 @@ function blobReferences(payloadJson: string): BlobReference[] { return found } -function incrementBlobRefs(database: DatabaseSync, blobs: BlobDescriptor[]): void { - const statement = database.prepare(` - INSERT INTO blobs(hash, encoding, byte_length, ref_count, created_at) - VALUES (?, ?, ?, 1, ?) - ON CONFLICT(hash) DO UPDATE SET ref_count = ref_count + 1 - `) - for (const blob of blobs) statement.run(blob.hash, blob.encoding, blob.byteLength, Date.now()) -} - -function decrementRowsBlobRefs(database: DatabaseSync, rows: Array<{ payload_json: string }>): void { - const decrement = database.prepare(` - UPDATE blobs SET ref_count = ref_count - 1 WHERE hash = ? AND ref_count > 0 - `) - for (const row of rows) { - for (const reference of blobReferences(row.payload_json)) decrement.run(reference.hash) +function createBlobStatements(database: DatabaseSync) { + return { + insertBlob: database.prepare(` + INSERT OR IGNORE INTO blobs(hash, encoding, byte_length, created_at) + VALUES (?, ?, ?, ?) + `), + insertMessageBlob: database.prepare(` + INSERT OR IGNORE INTO message_blobs(thread_id, ordinal, hash) VALUES (?, ?, ?) + `), + insertBranchMessageBlob: database.prepare(` + INSERT OR IGNORE INTO branch_message_blobs(thread_id, branch_id, ordinal, hash) + VALUES (?, ?, ?, ?) + `), } } -async function collectGarbageBlobs(database: DatabaseSync, databasePath: string): Promise { +type BlobOwner = + | { type: 'message'; threadId: string; ordinal: number } + | { type: 'branchMessage'; threadId: string; branchId: string; ordinal: number } + +function insertBlobReferences( + statements: ReturnType, + owner: BlobOwner, + blobs: BlobDescriptor[], +): void { + const inserted = new Set() + for (const blob of blobs) { + if (inserted.has(blob.hash)) continue + inserted.add(blob.hash) + statements.insertBlob.run(blob.hash, blob.encoding, blob.byteLength, Date.now()) + if (owner.type === 'message') { + statements.insertMessageBlob.run(owner.threadId, owner.ordinal, blob.hash) + } else { + statements.insertBranchMessageBlob.run( + owner.threadId, owner.branchId, owner.ordinal, blob.hash, + ) + } + } +} + +async function removeUnreferencedCreatedBlobs( + database: DatabaseSync, + databasePath: string, + createdHashes: Set, +): Promise { + if (createdHashes.size === 0) return + const isTracked = database.prepare('SELECT 1 FROM blobs WHERE hash = ?') + await Promise.all([...createdHashes].map(async hash => { + if (isTracked.get(hash)) return + await fs.rm(blobPath(databasePath, hash), { force: true }).catch(() => undefined) + })) +} + +function readBackupBlobHashes(backup: DatabaseSync): Set { + const hashes = new Set() + if (tableExists(backup, 'message_blobs') && tableExists(backup, 'branch_message_blobs')) { + const rows = backup.prepare(` + SELECT hash FROM message_blobs + UNION + SELECT hash FROM branch_message_blobs + `).all() as Array<{ hash: string }> + for (const row of rows) hashes.add(row.hash) + return hashes + } + + // A pre-v5 recovery snapshot can survive the first application upgrade. + // Parse it once per worker lifetime, then the next snapshot rotation removes + // this compatibility path from the hot commit loop. + for (const table of ['messages', 'branch_messages'] as const) { + if (!tableExists(backup, table)) continue + const rows = backup.prepare(`SELECT payload_json FROM ${table}`).all() as Array<{ payload_json: string }> + for (const row of rows) { + for (const reference of blobReferences(row.payload_json)) hashes.add(reference.hash) + } + } + return hashes +} + +async function protectedBackupBlobHashes(opened: OpenDatabase): Promise | null> { + if (opened.protectedBackupBlobHashes) return opened.protectedBackupBlobHashes const protectedHashes = new Set() - for (const backupPath of [`${databasePath}.bak`, `${databasePath}.bak.previous`]) { + for (const backupPath of [`${opened.path}.bak`, `${opened.path}.bak.previous`]) { if (!await fileExists(backupPath)) continue let backup: DatabaseSync | undefined try { backup = new DatabaseSync(backupPath, { readOnly: true }) assertHealthy(backup) - for (const table of ['messages', 'branch_messages'] as const) { - if (!tableExists(backup, table)) continue - const rows = backup.prepare(`SELECT payload_json FROM ${table}`).all() as Array<{ payload_json: string }> - for (const row of rows) for (const reference of blobReferences(row.payload_json)) { - protectedHashes.add(reference.hash) - } - } + for (const hash of readBackupBlobHashes(backup)) protectedHashes.add(hash) } catch { // Fail closed: an unreadable snapshot must never cause companion data loss. - return + return null } finally { backup?.close() } } + opened.protectedBackupBlobHashes = protectedHashes + return protectedHashes +} - const rows = database.prepare('SELECT hash FROM blobs WHERE ref_count = 0').all() as Array<{ hash: string }> - const removeRow = database.prepare('DELETE FROM blobs WHERE hash = ? AND ref_count = 0') +async function collectGarbageBlobs(opened: OpenDatabase): Promise { + const rows = opened.database.prepare(` + SELECT b.hash + FROM blobs b + WHERE NOT EXISTS (SELECT 1 FROM message_blobs m WHERE m.hash = b.hash) + AND NOT EXISTS (SELECT 1 FROM branch_message_blobs bm WHERE bm.hash = b.hash) + `).all() as Array<{ hash: string }> + if (rows.length === 0) return + + const protectedHashes = await protectedBackupBlobHashes(opened) + if (!protectedHashes) return + const removeRow = opened.database.prepare(` + DELETE FROM blobs + WHERE hash = ? + AND NOT EXISTS (SELECT 1 FROM message_blobs m WHERE m.hash = blobs.hash) + AND NOT EXISTS (SELECT 1 FROM branch_message_blobs bm WHERE bm.hash = blobs.hash) + `) for (const row of rows) { if (protectedHashes.has(row.hash)) continue try { - await fs.rm(blobPath(databasePath, row.hash), { force: true }) + await fs.rm(blobPath(opened.path, row.hash), { force: true }) removeRow.run(row.hash) } catch { /* retry during the next maintenance cycle */ } } @@ -597,7 +832,7 @@ function readState(database: DatabaseSync): SessionStateRecord { function readThreads(database: DatabaseSync): SessionThreadMetadata[] { const rows = database.prepare(` SELECT id, created_at, last_modified, title, message_count, metadata_json - FROM threads ORDER BY last_modified DESC + FROM threads ORDER BY last_modified DESC, id ASC `).all() as Array<{ id: string created_at: number @@ -664,16 +899,21 @@ async function readBranchMessages( const branches = database.prepare( 'SELECT branch_id FROM branches WHERE thread_id = ? ORDER BY ordinal', ).all(threadId) as Array<{ branch_id: string }> - return Promise.all(branches.map(async branch => { - const rows = database.prepare(` - SELECT payload_json FROM branch_messages - WHERE thread_id = ? AND branch_id = ? ORDER BY ordinal - `).all(threadId, branch.branch_id) as Array<{ payload_json: string }> - return { - id: branch.branch_id, - messages: await Promise.all(rows.map(row => hydrateValue(JSON.parse(row.payload_json), databasePath))), - } - })) + const rows = database.prepare(` + SELECT branch_id, payload_json FROM branch_messages + WHERE thread_id = ? ORDER BY branch_id, ordinal + `).all(threadId) as Array<{ branch_id: string; payload_json: string }> + const payloadsByBranch = new Map() + for (const row of rows) { + const payloads = payloadsByBranch.get(row.branch_id) || [] + if (!payloadsByBranch.has(row.branch_id)) payloadsByBranch.set(row.branch_id, payloads) + payloads.push(row.payload_json) + } + return Promise.all(branches.map(async branch => ({ + id: branch.branch_id, + messages: await Promise.all((payloadsByBranch.get(branch.branch_id) || []) + .map(payload => hydrateValue(JSON.parse(payload), databasePath))), + }))) } function upsertThread(database: DatabaseSync, thread: SessionThreadMetadata): void { @@ -707,6 +947,7 @@ function insertMessages( throw new Error('branchId is required for branch messages') } const messageBranchId = branchId ?? '' + const blobStatements = createBlobStatements(database) const insert = table === 'messages' ? database.prepare(` INSERT INTO messages(thread_id, ordinal, message_id, role, timestamp, payload_json) @@ -718,51 +959,56 @@ function insertMessages( ) VALUES (?, ?, ?, ?, ?, ?, ?) `) for (const message of messages) { - incrementBlobRefs(database, message.blobs) if (table === 'messages') { insert.run(threadId, message.ordinal, message.id, message.role, message.timestamp, message.payloadJson) + insertBlobReferences(blobStatements, { + type: 'message', threadId, ordinal: message.ordinal, + }, message.blobs) } else { insert.run(threadId, messageBranchId, message.ordinal, message.id, message.role, message.timestamp, message.payloadJson) + insertBlobReferences(blobStatements, { + type: 'branchMessage', threadId, branchId: messageBranchId, ordinal: message.ordinal, + }, message.blobs) } } } -function deleteThreadBlobRefs(database: DatabaseSync, threadId: string): void { - const rows = [ - ...database.prepare('SELECT payload_json FROM messages WHERE thread_id = ?').all(threadId), - ...database.prepare('SELECT payload_json FROM branch_messages WHERE thread_id = ?').all(threadId), - ] as Array<{ payload_json: string }> - decrementRowsBlobRefs(database, rows) -} - -async function applyPatch(database: DatabaseSync, databasePath: string, patch: SessionPatch): Promise { - const preparedThreads = await Promise.all(patch.threads.map(async thread => ({ - ...thread, - messages: await Promise.all((thread.messages || []).map(message => prepareMessage(databasePath, message))), - }))) - const preparedBranchThreads = await Promise.all(patch.branchThreads.map(async entry => ({ - threadId: entry.threadId, - branches: await Promise.all(entry.branches.map(async branch => ({ - ...branch, - messages: await Promise.all(branch.messages.map(message => prepareMessage(databasePath, message))), - } satisfies PreparedBranch))), - }))) +async function applyPatch(opened: OpenDatabase, patch: SessionPatch): Promise { + const { database, path: databasePath } = opened + const createdHashes = new Set() + let preparedThreads: Array & { + messages: PreparedMessage[] + }> + let preparedBranchThreads: Array<{ threadId: string; branches: PreparedBranch[] }> + try { + preparedThreads = await Promise.all(patch.threads.map(async thread => ({ + ...thread, + messages: await Promise.all((thread.messages || []) + .map(message => prepareMessage(databasePath, message, createdHashes))), + }))) + preparedBranchThreads = await Promise.all(patch.branchThreads.map(async entry => ({ + threadId: entry.threadId, + branches: await Promise.all(entry.branches.map(async branch => ({ + ...branch, + messages: await Promise.all(branch.messages + .map(message => prepareMessage(databasePath, message, createdHashes))), + } satisfies PreparedBranch))), + }))) + } catch (error) { + await removeUnreferencedCreatedBlobs(database, databasePath, createdHashes) + throw error + } database.exec('BEGIN IMMEDIATE') try { const deleteThread = database.prepare('DELETE FROM threads WHERE id = ?') for (const threadId of patch.deletedThreadIds) { - deleteThreadBlobRefs(database, threadId) deleteThread.run(threadId) } for (const threadPatch of preparedThreads) { upsertThread(database, threadPatch.metadata) if (threadPatch.replaceFrom !== undefined) { - const oldRows = database.prepare(` - SELECT payload_json FROM messages WHERE thread_id = ? AND ordinal >= ? - `).all(threadPatch.metadata.id, threadPatch.replaceFrom) as Array<{ payload_json: string }> - decrementRowsBlobRefs(database, oldRows) database.prepare('DELETE FROM messages WHERE thread_id = ? AND ordinal >= ?') .run(threadPatch.metadata.id, threadPatch.replaceFrom) insertMessages(database, 'messages', threadPatch.metadata.id, threadPatch.messages) @@ -777,10 +1023,6 @@ async function applyPatch(database: DatabaseSync, databasePath: string, patch: S ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) `) for (const entry of preparedBranchThreads) { - const oldRows = database.prepare( - 'SELECT payload_json FROM branch_messages WHERE thread_id = ?', - ).all(entry.threadId) as Array<{ payload_json: string }> - decrementRowsBlobRefs(database, oldRows) deleteBranches.run(entry.threadId) for (const branch of entry.branches) { insertBranch.run( @@ -807,9 +1049,12 @@ async function applyPatch(database: DatabaseSync, databasePath: string, patch: S database.exec('COMMIT') } catch (error) { database.exec('ROLLBACK') + await removeUnreferencedCreatedBlobs(database, databasePath, createdHashes) throw error } - await collectGarbageBlobs(database, databasePath) + const mayHaveOrphanedBlobs = patch.deletedThreadIds.length > 0 || + patch.branchThreads.length > 0 || patch.threads.some(thread => thread.replaceFrom !== undefined) + if (mayHaveOrphanedBlobs) await collectGarbageBlobs(opened) } async function importLegacy(database: DatabaseSync, databasePath: string, sessionsDir: string): Promise { @@ -928,7 +1173,9 @@ async function importLegacy(database: DatabaseSync, databasePath: string, sessio }] }) - await applyPatch(database, databasePath, { state, threads, deletedThreadIds: [], branchThreads }) + const opened = databases.get(databasePath) + if (!opened) throw new Error('Session database is not registered') + await applyPatch(opened, { state, threads, deletedThreadIds: [], branchThreads }) markComplete() return true } @@ -1020,7 +1267,18 @@ function scheduleCheckpoint(opened: OpenDatabase): void { opened.idleCheckpoint = null try { checkpoint(opened, false) - opened.database.exec('PRAGMA incremental_vacuum(256); PRAGMA optimize;') + const pageCount = Number( + (opened.database.prepare('PRAGMA page_count').get() as { page_count: number }).page_count, + ) + const freePages = Number( + (opened.database.prepare('PRAGMA freelist_count').get() as { freelist_count: number }).freelist_count, + ) + // Reclaim only meaningful fragmentation. Running incremental_vacuum on + // every quiet period creates needless SSD writes for normal append-only use. + if (freePages >= 1024 && freePages / Math.max(1, pageCount) >= 0.2) { + opened.database.exec(`PRAGMA incremental_vacuum(${Math.min(256, freePages)})`) + } + opened.database.exec('PRAGMA optimize') } catch { /* retry on next idle/close */ } }, delay) } @@ -1028,6 +1286,11 @@ function scheduleCheckpoint(opened: OpenDatabase): void { async function closeAll(): Promise { for (const opened of databases.values()) { if (opened.idleCheckpoint) clearTimeout(opened.idleCheckpoint) + try { + opened.database.prepare( + 'UPDATE maintenance_state SET clean_shutdown = 1 WHERE singleton = 1', + ).run() + } catch { /* an older failed migration remains recoverable */ } try { checkpoint(opened, true) } catch { /* WAL remains recoverable */ } opened.database.close() try { @@ -1072,18 +1335,7 @@ async function clearDatabase(opened: OpenDatabase): Promise { await fs.rm(`${opened.path}.bak`, { force: true }) await fs.rm(`${opened.path}.bak.previous`, { force: true }) await fs.rm(blobDirectory(opened.path), { recursive: true, force: true }) -} - -async function directoryBytes(directory: string): Promise { - let total = 0 - let entries - try { entries = await fs.readdir(directory, { withFileTypes: true }) } catch { return 0 } - for (const entry of entries) { - const target = path.join(directory, entry.name) - if (entry.isDirectory()) total += await directoryBytes(target) - else if (entry.isFile()) total += await fs.stat(target).then(stat => stat.size).catch(() => 0) - } - return total + opened.protectedBackupBlobHashes = new Set() } async function readStats(database: DatabaseSync, databasePath: string): Promise { @@ -1096,11 +1348,11 @@ async function readStats(database: DatabaseSync, databasePath: string): Promise< return { databaseBytes: await fileBytes(databasePath), walBytes: await fileBytes(`${databasePath}-wal`), - blobBytes: await directoryBytes(blobDirectory(databasePath)), + blobBytes: scalar('SELECT COALESCE(SUM(byte_length), 0) AS value FROM blobs', 'value'), threadCount: scalar('SELECT COUNT(*) AS value FROM threads', 'value'), messageCount: scalar('SELECT COUNT(*) AS value FROM messages', 'value'), branchCount: scalar('SELECT COUNT(*) AS value FROM branches', 'value'), - blobCount: scalar('SELECT COUNT(*) AS value FROM blobs WHERE ref_count > 0', 'value'), + blobCount: scalar('SELECT COUNT(*) AS value FROM blobs', 'value'), planCount: scalar('SELECT COUNT(*) AS value FROM plans', 'value'), pageSize: scalar('PRAGMA page_size', 'page_size'), freePages: scalar('PRAGMA freelist_count', 'freelist_count'), @@ -1149,7 +1401,7 @@ export async function executeSessionStorageOperation( scheduleCheckpoint(opened) return { type: 'ok' } case 'applyPatch': - await applyPatch(opened.database, opened.path, operation.patch) + await applyPatch(opened, operation.patch) scheduleCheckpoint(opened) return { type: 'ok' } case 'clear': @@ -1162,7 +1414,9 @@ export async function executeSessionStorageOperation( } } -parentPort?.on('message', async (request: SessionWorkerRequest) => { +const operationQueues = new Map>() + +async function respond(request: SessionWorkerRequest): Promise { let response: SessionWorkerResponse try { response = { @@ -1178,4 +1432,20 @@ parentPort?.on('message', async (request: SessionWorkerRequest) => { } } parentPort?.postMessage(response) +} + +parentPort?.on('message', (request: SessionWorkerRequest) => { + if (request.operation.type === 'closeAll') { + const pending = [...operationQueues.values()] + void Promise.allSettled(pending).then(() => respond(request)) + return + } + + const databasePath = request.operation.databasePath + const previous = operationQueues.get(databasePath) || Promise.resolve() + const next = previous.catch(() => undefined).then(() => respond(request)) + operationQueues.set(databasePath, next) + void next.finally(() => { + if (operationQueues.get(databasePath) === next) operationQueues.delete(databasePath) + }) }) diff --git a/tests/main/indexing/bm25.test.ts b/tests/main/indexing/bm25.test.ts index 0501505d..5b09af6b 100644 --- a/tests/main/indexing/bm25.test.ts +++ b/tests/main/indexing/bm25.test.ts @@ -17,11 +17,10 @@ function doc(id: string, relativePath: string, content: string, symbols: string[ } describe('BM25Index', () => { - it('returns nothing before build and finds matches after', () => { + it('is searchable immediately after documents are added', () => { const idx = new BM25Index() idx.addDocument(doc('a', 'a.ts', 'authentication token refresh handler')) idx.addDocument(doc('b', 'b.ts', 'unrelated rendering pipeline code')) - idx.build() const results = idx.search('authentication token') expect(results.length).toBeGreaterThan(0) @@ -32,7 +31,6 @@ describe('BM25Index', () => { const idx = new BM25Index() idx.addDocument(doc('a', 'a.ts', 'cache invalidation cache eviction cache policy')) idx.addDocument(doc('b', 'b.ts', 'cache mentioned once here')) - idx.build() const results = idx.search('cache') expect(results[0].relativePath).toBe('a.ts') @@ -42,7 +40,6 @@ describe('BM25Index', () => { const idx = new BM25Index() idx.addDocument(doc('a', 'a.ts', 'shared body text alpha', ['parseWorkspace'])) idx.addDocument(doc('b', 'b.ts', 'shared body text alpha', [])) - idx.build() const results = idx.search('parseworkspace') expect(results[0].relativePath).toBe('a.ts') @@ -53,10 +50,8 @@ describe('BM25Index', () => { idx.addDocument(doc('a1', 'a.ts', 'authentication token part one')) idx.addDocument(doc('a2', 'a.ts', 'authentication token part two')) idx.addDocument(doc('b1', 'b.ts', 'unrelated rendering pipeline')) - idx.build() idx.deleteFile('a.ts') - idx.build() expect(idx.size).toBe(1) expect(idx.search('authentication token')).toHaveLength(0) @@ -66,11 +61,9 @@ describe('BM25Index', () => { const idx = new BM25Index() idx.addDocument(doc('a', 'a.ts', 'zzuniqueterm appears only here')) idx.addDocument(doc('b', 'b.ts', 'unrelated rendering pipeline')) - idx.build() const vocabularyBefore = idx.vocabularySize idx.deleteFile('a.ts') - idx.build() // 该词已随文件移除,不应残留 IDF 条目(否则评分失真且内存无界增长) expect(idx.vocabularySize).toBeLessThan(vocabularyBefore) @@ -82,14 +75,12 @@ describe('BM25Index', () => { for (let i = 0; i < 5; i++) { idx.addDocument(doc(`f${i}`, `f${i}.ts`, 'common filler text')) } - idx.build() const rareScoreBefore = idx.search('rare')[0].score // 让 "rare" 变成常见词,IDF 应当下降 for (let i = 0; i < 5; i++) { idx.addDocument(doc(`g${i}`, `g${i}.ts`, 'rare term everywhere now')) } - idx.build() const rareScoreAfter = idx.search('rare')[0].score expect(rareScoreAfter).toBeLessThan(rareScoreBefore) @@ -97,7 +88,6 @@ describe('BM25Index', () => { it('handles an empty index without throwing', () => { const idx = new BM25Index() - idx.build() expect(idx.search('anything')).toEqual([]) expect(idx.size).toBe(0) }) @@ -105,7 +95,6 @@ describe('BM25Index', () => { it('clears all state', () => { const idx = new BM25Index() idx.addDocument(doc('a', 'a.ts', 'authentication token refresh')) - idx.build() idx.clear() expect(idx.size).toBe(0) diff --git a/tests/main/indexing/indexServiceScheduling.test.ts b/tests/main/indexing/indexServiceScheduling.test.ts index eabe6aa9..8326eb4e 100644 --- a/tests/main/indexing/indexServiceScheduling.test.ts +++ b/tests/main/indexing/indexServiceScheduling.test.ts @@ -36,12 +36,6 @@ vi.mock('worker_threads', () => ({ }, })) -vi.mock('@main/indexing/treeSitterChunker', () => ({ - TreeSitterChunker: class { - init = vi.fn(async () => {}) - }, -})) - vi.mock('@main/indexing/vectorStore', () => ({ VectorStoreService: class {}, })) @@ -114,6 +108,7 @@ describe('CodebaseIndexService scheduling', () => { internals.performWorkspaceIndex = vi.fn(async () => { order.push('index:start') await fullIndex.promise + ;(service as unknown as { status: { lastIndexedAt?: number } }).status.lastIndexedAt = 1 order.push('index:end') }) internals.performUpdateFiles = vi.fn(async paths => { @@ -222,6 +217,57 @@ describe('CodebaseIndexService scheduling', () => { await expect(service.search('cachedNeedle')).resolves.toHaveLength(1) }) + it('processes structural watcher updates through the worker and persists them', async () => { + structuralStoreState.metadata = { totalFiles: 1, totalChunks: 0, savedAt: 123 } + await service.initialize() + const updating = service.updateFiles(['C:/workspace/src/updated.ts']) + await vi.waitFor(() => expect(workerState.instances).toHaveLength(1)) + const worker = workerState.instances[0] + expect(worker.postMessage).toHaveBeenCalledWith(expect.objectContaining({ + type: 'batch_update', + files: ['C:/workspace/src/updated.ts'], + config: expect.objectContaining({ mode: 'structural' }), + })) + + worker.emit('message', { + type: 'batch_update_result', + mode: 'structural', + requestId: 1, + results: [{ + filePath: 'C:/workspace/src/updated.ts', + deleted: false, + chunks: [{ + id: 'updated-chunk', + filePath: 'C:/workspace/src/updated.ts', + relativePath: 'src/updated.ts', + fileHash: 'hash', + content: 'function updatedNeedle() {}', + startLine: 1, + endLine: 1, + type: 'function', + language: 'typescript', + symbols: ['updatedNeedle'], + }], + }], + }) + + await updating + expect(service.searchSymbols('updatedNeedle')).toHaveLength(1) + await expect(service.search('updatedNeedle')).resolves.toHaveLength(1) + expect(structuralStoreState.operations.at(-1)).toMatchObject({ + type: 'applyFiles', + files: [{ relativePath: 'src\\updated.ts' }], + }) + }) + + it('does not create a partial index from watcher events before indexing is enabled', async () => { + await service.updateFiles(['C:/workspace/src/updated.ts']) + await service.deleteFileIndex('C:/workspace/src/deleted.ts') + + expect(workerState.instances).toHaveLength(0) + expect(structuralStoreState.operations).toHaveLength(0) + }) + it('finishes an active index with an error when the worker exits unexpectedly', async () => { const internals = service as unknown as { saveIndex(): Promise } internals.saveIndex = vi.fn(async () => {}) diff --git a/tests/main/indexing/structuralIndexStore.test.ts b/tests/main/indexing/structuralIndexStore.test.ts index 18f9670a..b50286e8 100644 --- a/tests/main/indexing/structuralIndexStore.test.ts +++ b/tests/main/indexing/structuralIndexStore.test.ts @@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest' import { mkdtemp, rm } from 'fs/promises' import { tmpdir } from 'os' import * as path from 'path' +import { DatabaseSync } from 'node:sqlite' import { executeStructuralIndexStoreOperation } from '@main/indexing/structuralIndexStore.worker' import type { CodeChunk } from '@main/indexing/types' @@ -57,7 +58,7 @@ describe('structural index SQLite store', () => { } it('streams a committed generation in bounded batches', () => { - const chunks = Array.from({ length: 125 }, (_, index) => chunk(`chunk-${index}`, `src/${index}.ts`)) + const chunks = Array.from({ length: 1025 }, (_, index) => chunk(`chunk-${index}`, `src/${index}.ts`)) executeStructuralIndexStoreOperation({ type: 'beginReplace', databasePath, generation: 'one' }) executeStructuralIndexStoreOperation({ type: 'appendReplace', databasePath, generation: 'one', chunks, @@ -66,13 +67,13 @@ describe('structural index SQLite store', () => { type: 'commitReplace', databasePath, generation: 'one', - metadata: { totalFiles: 125, totalChunks: 125, savedAt: 10 }, + metadata: { totalFiles: 1025, totalChunks: 1025, savedAt: 10 }, }) const loaded = load() - expect(loaded.chunks).toHaveLength(125) - expect(loaded.metadata).toEqual({ totalFiles: 125, totalChunks: 125, savedAt: 10 }) - expect(loaded.batchSizes).toEqual([50, 50, 25]) + expect(loaded.chunks).toHaveLength(1025) + expect(loaded.metadata).toEqual({ totalFiles: 1025, totalChunks: 1025, savedAt: 10 }) + expect(loaded.batchSizes).toEqual([512, 512, 1]) }) it('keeps the previous generation when a replacement is incomplete', () => { @@ -131,4 +132,48 @@ describe('structural index SQLite store', () => { expect(loaded.chunks.map(item => item.id)).toEqual(['a2']) expect(loaded.metadata).toEqual({ totalFiles: 2, totalChunks: 1, savedAt: 20 }) }) + + it('stores file metadata once and uses the tuned database layout', () => { + executeStructuralIndexStoreOperation({ type: 'beginReplace', databasePath, generation: 'layout' }) + executeStructuralIndexStoreOperation({ + type: 'appendReplace', + databasePath, + generation: 'layout', + chunks: [ + chunk('part-1', 'shared.ts'), + { ...chunk('part-2', 'shared.ts'), startLine: 2, endLine: 2 }, + ], + }) + executeStructuralIndexStoreOperation({ + type: 'commitReplace', + databasePath, + generation: 'layout', + metadata: { totalFiles: 1, totalChunks: 2, savedAt: 10 }, + }) + + const database = new DatabaseSync(databasePath, { readOnly: true }) + try { + const fileCount = database.prepare('SELECT COUNT(*) AS count FROM files').get() as { count: number } + const chunkCount = database.prepare('SELECT COUNT(*) AS count FROM chunks').get() as { count: number } + const columns = database.prepare("PRAGMA table_info('chunks')").all() as Array<{ name: string }> + const pageSize = database.prepare('PRAGMA page_size').get() as { page_size: number } + const journalMode = database.prepare('PRAGMA journal_mode').get() as { journal_mode: string } + const pagePlan = database.prepare(` + EXPLAIN QUERY PLAN + SELECT * FROM chunks + WHERE generation = ? AND (relative_path, id) > (?, ?) + ORDER BY relative_path, id LIMIT ? + `).all('generation-layout', '', '', 512) as Array<{ detail: string }> + + expect(Number(fileCount.count)).toBe(1) + expect(Number(chunkCount.count)).toBe(2) + expect(columns.map(column => column.name)).not.toContain('payload_json') + expect(Number(pageSize.page_size)).toBe(8192) + expect(journalMode.journal_mode).toBe('wal') + expect(pagePlan.map(row => row.detail).join('\n')) + .toContain('PRIMARY KEY (generation=? AND (relative_path,id)>(?,?))') + } finally { + database.close() + } + }) }) diff --git a/tests/services/sessionSqliteStore.test.ts b/tests/services/sessionSqliteStore.test.ts index 76988e68..bd25b713 100644 --- a/tests/services/sessionSqliteStore.test.ts +++ b/tests/services/sessionSqliteStore.test.ts @@ -1,7 +1,8 @@ import { afterEach, describe, expect, it } from 'vitest' -import { mkdtemp, mkdir, readdir, rm, writeFile } from 'fs/promises' +import { access, mkdtemp, mkdir, readdir, rm, writeFile } from 'fs/promises' import { tmpdir } from 'os' import { join } from 'path' +import { createHash } from 'crypto' import { DatabaseSync } from 'node:sqlite' import { executeSessionStorageOperation } from '@/main/services/session/sessionStorage.worker' import type { SessionPatch } from '@/shared/types/sessionPersistence' @@ -38,6 +39,34 @@ afterEach(async () => { }) describe('SQLite session store', () => { + it('records clean shutdowns and reuses a recent integrity check', async () => { + const { databasePath } = await temporaryDatabase() + await executeSessionStorageOperation({ type: 'open', databasePath }) + let database = new DatabaseSync(databasePath, { readOnly: true }) + const opened = database.prepare(` + SELECT last_quick_check_at, clean_shutdown FROM maintenance_state WHERE singleton = 1 + `).get() as { last_quick_check_at: number; clean_shutdown: number } + expect(opened.last_quick_check_at).toBeGreaterThan(0) + expect(opened.clean_shutdown).toBe(0) + database.close() + + await executeSessionStorageOperation({ type: 'closeAll' }) + database = new DatabaseSync(databasePath, { readOnly: true }) + expect((database.prepare(` + SELECT clean_shutdown FROM maintenance_state WHERE singleton = 1 + `).get() as { clean_shutdown: number }).clean_shutdown).toBe(1) + database.close() + + await executeSessionStorageOperation({ type: 'open', databasePath }) + database = new DatabaseSync(databasePath, { readOnly: true }) + const reopened = database.prepare(` + SELECT last_quick_check_at, clean_shutdown FROM maintenance_state WHERE singleton = 1 + `).get() as { last_quick_check_at: number; clean_shutdown: number } + expect(reopened.last_quick_check_at).toBe(opened.last_quick_check_at) + expect(reopened.clean_shutdown).toBe(0) + database.close() + }) + it('replaces only a changed message tail', async () => { const { databasePath } = await temporaryDatabase() await executeSessionStorageOperation({ type: 'open', databasePath }) @@ -88,6 +117,28 @@ describe('SQLite session store', () => { expect(result.type === 'messages' ? result.messages : []).toHaveLength(2) }) + it('removes only newly-created blob files when a database transaction rolls back', async () => { + const { databasePath } = await temporaryDatabase() + await executeSessionStorageOperation({ type: 'open', databasePath }) + const content = 'rollback-blob-'.repeat(24_000) + const broken = initialPatch() + broken.threads[0].metadata.messageCount = 2 + broken.threads[0].messages = [0, 1].map(index => ({ + ordinal: 0, + id: `duplicate-${index}`, + role: 'assistant', + timestamp: index, + payload: { id: `duplicate-${index}`, content }, + })) + + await expect(executeSessionStorageOperation({ + type: 'applyPatch', databasePath, patch: broken, + })).rejects.toThrow() + + const hash = createHash('sha256').update(Buffer.from(content, 'utf8')).digest('hex') + await expect(access(join(`${databasePath}.blobs`, hash.slice(0, 2), hash))).rejects.toThrow() + }) + it('stores branch messages independently and replaces a thread branch set atomically', async () => { const { databasePath } = await temporaryDatabase() await executeSessionStorageOperation({ type: 'open', databasePath }) @@ -145,6 +196,13 @@ describe('SQLite session store', () => { threadCount: 1, messageCount: 2, blobCount: 1, }) expect(stats.type === 'stats' ? stats.stats.blobBytes : 0).toBeGreaterThan(256 * 1024) + + const database = new DatabaseSync(databasePath, { readOnly: true }) + const blobColumns = database.prepare("PRAGMA table_info('blobs')").all() as Array<{ name: string }> + const references = database.prepare('SELECT COUNT(*) AS count FROM message_blobs').get() as { count: number } + expect(blobColumns.map(column => column.name)).not.toContain('ref_count') + expect(references.count).toBe(2) + database.close() }) it('keeps blobs referenced by recovery snapshots until those snapshots rotate', async () => { @@ -179,6 +237,52 @@ describe('SQLite session store', () => { ]) }) + it('migrates v4 blob counters to normalized message references', async () => { + const { databasePath } = await temporaryDatabase() + const content = 'v4-payload-'.repeat(30_000) + const patch = initialPatch() + patch.threads[0].metadata.messageCount = 1 + patch.threads[0].messages = [{ + ordinal: 0, id: 'large', role: 'assistant', timestamp: 1, + payload: { id: 'large', role: 'assistant', content }, + }] + await executeSessionStorageOperation({ type: 'open', databasePath }) + await executeSessionStorageOperation({ type: 'applyPatch', databasePath, patch }) + await executeSessionStorageOperation({ type: 'closeAll' }) + + const legacy = new DatabaseSync(databasePath) + legacy.exec(` + DROP TABLE branch_message_blobs; + DROP TABLE message_blobs; + DROP INDEX threads_last_modified_id; + DROP INDEX plans_updated_at_id; + CREATE INDEX plans_updated_at ON plans(updated_at DESC); + CREATE TABLE blobs_v4 ( + hash TEXT PRIMARY KEY, + encoding TEXT NOT NULL, + byte_length INTEGER NOT NULL, + ref_count INTEGER NOT NULL, + created_at INTEGER NOT NULL + ) STRICT; + INSERT INTO blobs_v4(hash, encoding, byte_length, ref_count, created_at) + SELECT hash, encoding, byte_length, 1, created_at FROM blobs; + DROP TABLE blobs; + ALTER TABLE blobs_v4 RENAME TO blobs; + PRAGMA user_version = 4; + `) + legacy.close() + + await executeSessionStorageOperation({ type: 'open', databasePath }) + const loaded = await executeSessionStorageOperation({ type: 'loadMessages', databasePath, threadId: 't1' }) + expect(loaded.type === 'messages' ? loaded.messages : []).toEqual([ + { id: 'large', role: 'assistant', content }, + ]) + const migrated = new DatabaseSync(databasePath, { readOnly: true }) + expect((migrated.prepare('SELECT COUNT(*) AS count FROM message_blobs').get() as { count: number }).count) + .toBe(1) + migrated.close() + }) + it('migrates legacy inline branches with an atomic user_version migration', async () => { const { databasePath } = await temporaryDatabase() await executeSessionStorageOperation({ type: 'open', databasePath }) @@ -201,10 +305,27 @@ describe('SQLite session store', () => { expect.objectContaining({ id: 'b1', name: 'Legacy', messageCount: 1 }), ]) const check = new DatabaseSync(databasePath, { readOnly: true }) - expect((check.prepare('PRAGMA user_version').get() as { user_version: number }).user_version).toBe(4) + expect((check.prepare('PRAGMA user_version').get() as { user_version: number }).user_version).toBe(5) check.close() }) + it('uses indexes for catalog and plan ordering without temporary sorting', async () => { + const { databasePath } = await temporaryDatabase() + await executeSessionStorageOperation({ type: 'open', databasePath }) + const database = new DatabaseSync(databasePath, { readOnly: true }) + const queryPlan = (sql: string): string => database.prepare(`EXPLAIN QUERY PLAN ${sql}`) + .all().map(row => String((row as { detail: string }).detail)).join('\n') + + expect(queryPlan('SELECT * FROM threads ORDER BY last_modified DESC, id ASC')) + .toContain('threads_last_modified_id') + expect(queryPlan('SELECT payload_json FROM plans ORDER BY updated_at DESC, id ASC')) + .not.toContain('USE TEMP B-TREE') + expect(database.prepare(` + SELECT 1 FROM sqlite_master WHERE type = 'index' AND name = 'messages_thread_id_id' + `).get()).toBeUndefined() + database.close() + }) + it('does not quarantine a healthy database when its schema is newer than the app', async () => { const { root, databasePath } = await temporaryDatabase() const database = new DatabaseSync(databasePath)