From 0dd4d1d64232f3268a45127be1e69efbc27b10a9 Mon Sep 17 00:00:00 2001 From: mesalogo Date: Sun, 4 Oct 2026 03:07:16 +0800 Subject: [PATCH] perf(db): relax frequent local writes, checkpoint WAL off Main, index startup recovery (schema 58) --- build/run-startup-perf.cjs | 23 +- build/run-write-perf.cjs | 235 ++++++++++++++++++ ...performance-experience-improvement-plan.md | 33 ++- package.json | 1 + .../activity-history-repository.test.ts | 3 +- src/main/assistant/assistant-database.test.ts | 83 ++++++- src/main/assistant/assistant-database.ts | 163 +++++++++++- src/main/index.ts | 1 + src/main/readonly-query-reader.test.ts | 35 +++ src/main/readonly-query-worker.ts | 26 +- 10 files changed, 586 insertions(+), 17 deletions(-) create mode 100644 build/run-write-perf.cjs diff --git a/build/run-startup-perf.cjs b/build/run-startup-perf.cjs index ca5239d5..96835940 100644 --- a/build/run-startup-perf.cjs +++ b/build/run-startup-perf.cjs @@ -21,6 +21,7 @@ // extra CPU-profiled launch per variant, excluded from the table), // GB_PERF_OUTPUT (report directory), GB_PERF_KEEP_PROFILE=1, // GB_PERF_STARTUP_APP_ROOT (checkout/worktree to build and launch; default .). +// Variant "real" (opt-in) launches on a fresh copy of GB_PERF_STARTUP_REAL_DATA. const { spawn, spawnSync, execFileSync } = require('node:child_process') const { cpSync, existsSync, mkdirSync, mkdtempSync, readFileSync, readdirSync, rmSync, statSync, writeFileSync } = require('node:fs') const { createServer } = require('node:http') @@ -262,6 +263,25 @@ async function main() { // launch starts from an identical, never-used-before directory. for (const variant of variants) { if (variant === 'first-run') continue + if (variant === 'real') { + // A copy of a real data directory (GB_PERF_STARTUP_REAL_DATA, for + // example the portable build's data folder). Only the stores startup + // reads are copied; the source is never opened by the app. The driver + // blocks every non-loopback request, so configured remote models are + // not called. + const source = process.env.GB_PERF_STARTUP_REAL_DATA + if (!source || !existsSync(join(source, 'assistant.sqlite'))) throw new Error('variant "real" needs GB_PERF_STARTUP_REAL_DATA with assistant.sqlite') + const directory = join(runDirectory, 'template-real') + const profile = join(directory, 'profile') + mkdirSync(profile, { recursive: true }) + for (const entry of readdirSync(source)) { + const from = join(source, entry) + const keep = /\.(sqlite|sqlite-wal|sqlite-shm|json)$/i.test(entry) || entry === 'notes' || entry === 'Local Storage' + if (keep && entry !== 'DevToolsActivePort') cpSync(from, join(profile, entry), { recursive: true }) + } + templates[variant] = directory + continue + } const seed = variant === 'empty' ? 0 : variant === seededName ? seedCount : null if (seed === null) throw new Error(`Unknown variant ${variant}`) const directory = join(runDirectory, `template-${variant}`) @@ -284,7 +304,8 @@ async function main() { return directory } - const expectedFor = variant => (variant === seededName ? seedCount : 0) + // The real profile's sidebar is ready once its first rows show. + const expectedFor = variant => (variant === seededName ? seedCount : variant === 'real' ? 10 : 0) const results = Object.fromEntries(variants.map(v => [v, []])) // A discarded warm-up launch per invocation so the first measured launch // does not also pay for loading Electron binaries from a cold disk cache. diff --git a/build/run-write-perf.cjs b/build/run-write-perf.cjs new file mode 100644 index 00000000..fa373dc4 --- /dev/null +++ b/build/run-write-perf.cjs @@ -0,0 +1,235 @@ +// Main-process database WRITE benchmark on a real data directory (PERF-15). +// Copies assistant.sqlite (+ -wal/-shm) from GB_WPERF_DATA (for example the +// portable build's data folder) into a temporary directory - the source is +// never opened - then measures, with the production AssistantDatabase, the +// writes Main performs synchronously while a run streams: +// task-events:subagent replay of the largest real subagent progress stream +// (restored events, re-compacted on write, as live) +// task-events:tool replay of real tool events +// task-events:text 250 ms text batches (persisted buffer) +// conversations:save the renderer's 500 ms local save (header + the +// streaming assistant message) for a real conversation +// activity:update one incremental activity change +// For each: per-call time and the longest Main event-loop block. +// Event replays are paced at GB_WPERF_RATE events/s (default 270, the busiest +// second in real data; 0 = back to back). +// Env: GB_WPERF_DATA (required), GB_WPERF_EVENTS (default 2000), +// GB_WPERF_OUTPUT (JSON), GB_WPERF_KEEP=1 (keep the copy). +const { buildSync } = require('esbuild') +const { mkdtempSync, rmSync, writeFileSync, mkdirSync, copyFileSync, cpSync, existsSync } = require('node:fs') +const { tmpdir } = require('node:os') +const { join, resolve } = require('node:path') +const { randomUUID } = require('node:crypto') +const { performance } = require('node:perf_hooks') + +const root = resolve(__dirname, '..') + +if (process.env.GB_WPERF_NODE !== '1') { + let electron + let version + try { electron = require('electron'); version = require('electron/package.json').version } catch { electron = undefined } + if (typeof electron === 'string' && process.versions.electron !== version) { + const { status } = require('node:child_process').spawnSync(electron, [__filename], { + stdio: 'inherit', env: { ...process.env, ELECTRON_RUN_AS_NODE: '1' } + }) + process.exit(status ?? 1) + } +} + +const source = process.env.GB_WPERF_DATA +if (!source || !existsSync(join(source, 'assistant.sqlite'))) { + console.error('[wperf] set GB_WPERF_DATA to a GoodBuddy data directory containing assistant.sqlite') + process.exit(2) +} +const eventLimit = Number(process.env.GB_WPERF_EVENTS || 2000) + +const directory = mkdtempSync(join(tmpdir(), 'goodbuddy-write-perf-')) +const bundle = join(directory, 'assistant-database.cjs') +const progressBundle = join(directory, 'subagent-progress.cjs') +for (const [entry, outfile] of [ + ['src/main/assistant/assistant-database.ts', bundle], + ['src/main/assistant/subagent-progress-storage.ts', progressBundle], + ['src/main/readonly-query-worker.ts', join(directory, 'readonly-query-worker.cjs')] +]) { + buildSync({ + entryPoints: [join(root, entry)], bundle: true, platform: 'node', format: 'cjs', + target: 'node22', outfile, logLevel: 'error', external: ['electron'] + }) +} +const { AssistantDatabase } = require(bundle) +const { restoreSubagentPayload } = require(progressBundle) + +const round = value => Math.round(value * 10) / 10 +const percentile = (values, p) => { + const sorted = [...values].sort((a, b) => a - b) + return sorted[Math.min(sorted.length - 1, Math.max(0, Math.ceil((p / 100) * sorted.length) - 1))] +} + +/** Per-call times plus the longest gap between event-loop turns. */ +async function measure(count, body, { yieldEvery = 1, rate = 0 } = {}) { + let probing = true + let last = performance.now() + let maxGap = 0 + const tick = () => { + const now = performance.now() + maxGap = Math.max(maxGap, now - last) + last = now + if (probing) setImmediate(tick) + } + setImmediate(tick) + const samples = [] + const slow = [] + const started = performance.now() + for (let index = 0; index < count; index += 1) { + const at = performance.now() + body(index) + const elapsed = performance.now() - at + samples.push(elapsed) + if (elapsed > 50) slow.push([index, round(elapsed)]) + if (rate > 0) { + // Real streams arrive at a rate; wait until this call's scheduled time. + const due = started + ((index + 1) * 1000) / rate + const wait = due - performance.now() + await new Promise(resolve => (wait > 1 ? setTimeout(resolve, wait) : setImmediate(resolve))) + } else if ((index + 1) % yieldEvery === 0) await new Promise(resolve => setImmediate(resolve)) + } + const total = performance.now() - started + await new Promise(resolve => setTimeout(resolve, 20)) + probing = false + return { + calls: count, totalMs: round(total), callP50: round(percentile(samples, 50)), + callP95: round(percentile(samples, 95)), callP99: round(percentile(samples, 99)), + callMax: round(Math.max(...samples)), loopMax: round(maxGap), slowCalls: slow.slice(0, 10) + } +} + +async function main() { + console.log(`[wperf] runtime: node ${process.version}${process.versions.electron ? ` (electron ${process.versions.electron})` : ''}`) + const copyStarted = performance.now() + for (const name of ['assistant.sqlite', 'assistant.sqlite-wal', 'assistant.sqlite-shm']) { + if (existsSync(join(source, name))) copyFileSync(join(source, name), join(directory, name)) + } + // Initialization checks note files against the database. + if (existsSync(join(source, 'notes'))) cpSync(join(source, 'notes'), join(directory, 'notes'), { recursive: true }) + console.log(`[wperf] copied data in ${round((performance.now() - copyStarted) / 1000)} s`) + const workspace = join(directory, 'workspace') + mkdirSync(workspace) + const database = new AssistantDatabase(join(directory, 'assistant.sqlite')) + const openStarted = performance.now() + database.initialize(workspace) + const raw = database.database + const results = { openMs: round(performance.now() - openStarted) } + // Experiment switch: override the durability level after initialization. + if (process.env.GB_WPERF_SYNCHRONOUS) raw.exec(`PRAGMA synchronous = ${process.env.GB_WPERF_SYNCHRONOUS}`) + results.synchronous = raw.prepare('PRAGMA synchronous').get().synchronous + if (process.env.GB_WPERF_AUTOCHECKPOINT) raw.exec(`PRAGMA wal_autocheckpoint = ${process.env.GB_WPERF_AUTOCHECKPOINT}`) + // As in production: checkpoints run on a worker thread (GB_WPERF_CHECKPOINT_WORKER=0 disables). + if (process.env.GB_WPERF_CHECKPOINT_WORKER !== '0' && typeof database.enableWalCheckpointWorker === 'function') { + database.enableWalCheckpointWorker(join(directory, 'readonly-query-worker.cjs')) + await new Promise(resolve => setTimeout(resolve, 500)) + } + results.walAutocheckpoint = raw.prepare('PRAGMA wal_autocheckpoint').get().wal_autocheckpoint + const count = (sql, ...args) => Number(raw.prepare(sql).get(...args).n) + results.taskEvents = count('SELECT COUNT(*) AS n FROM task_events') + results.messages = count('SELECT COUNT(*) AS n FROM messages') + results.activityRecords = count('SELECT COUNT(*) AS n FROM activity_history_records') + + // The real task with the most subagent events, restored in order. + const sourceTask = raw.prepare( + `SELECT task_id FROM task_events WHERE kind = 'subagent' GROUP BY task_id ORDER BY COUNT(*) DESC LIMIT 1` + ).get()?.task_id + const blocks = new Map() + const subagentEvents = sourceTask ? raw.prepare( + `SELECT payload_json FROM task_events WHERE task_id = ? AND kind = 'subagent' ORDER BY id LIMIT ?` + ).all(sourceTask, eventLimit).map(row => restoreSubagentPayload(JSON.parse(row.payload_json), blocks)) : [] + const toolEvents = raw.prepare( + `SELECT payload_json FROM task_events WHERE kind = 'tool' ORDER BY id DESC LIMIT ?` + ).all(eventLimit).map(row => JSON.parse(row.payload_json)) + + const conversation = raw.prepare( + `SELECT conversation_id AS id FROM messages GROUP BY conversation_id ORDER BY COUNT(*) DESC LIMIT 1` + ).get()?.id + const taskId = randomUUID() + database.createTask({ id: taskId, title: 'write perf', instructions: 'replay', workMode: 'ask', conversationId: conversation }) + + const replay = (events, kind) => index => { + const event = { ...events[index % events.length], requestId: taskId } + database.appendTaskEvent(taskId, kind, event) + } + if (process.env.GB_WPERF_INSPECT) { + const at = Number(process.env.GB_WPERF_INSPECT) + const { compactSubagentPayload } = require(progressBundle) + const event = subagentEvents[at] + const t0 = performance.now() + const json = JSON.stringify(event) + const t1 = performance.now() + const restoredBlocks = new Map() + for (let i = 0; i < at; i += 1) restoreSubagentPayload(JSON.parse(JSON.stringify(compactSubagentPayload(subagentEvents[i], restoredBlocks))), restoredBlocks) + const t2 = performance.now() + const compact = JSON.stringify(compactSubagentPayload(event, restoredBlocks)) + const t3 = performance.now() + console.log(`[wperf] event ${at}: full ${json.length} B (stringify ${round(t1 - t0)} ms), compact ${compact.length} B (compact ${round(t3 - t2)} ms)`) + console.log(`[wperf] event ${at} progress blocks: ${(event.progress ?? []).length}, updates: ${(event.progressUpdates ?? []).length}`) + } + // Paced at the busiest second seen in real data (269 subagent events/s); 0 = back to back. + const rate = Number(process.env.GB_WPERF_RATE ?? 270) + if (subagentEvents.length) { + results.subagentEvents = await measure(subagentEvents.length, replay(subagentEvents, 'subagent'), { rate }) + } + if (toolEvents.length) results.toolEvents = await measure(toolEvents.length, replay(toolEvents, 'tool'), { rate }) + const delta = 'streamed text '.repeat(40) + results.textBatches = await measure(400, index => database.appendTaskEvent(taskId, 'text', + { type: 'text', requestId: taskId, delta: `${delta}${index}` })) + + // Renderer local save: the conversation header and the streaming assistant message growing. + if (conversation) { + const snapshot = database.getConversation(conversation) + const header = { + id: snapshot.id, projectId: snapshot.projectId, title: snapshot.title, + runtimeSelection: snapshot.runtimeSelection, updatedAt: Date.now() + } + const messageId = randomUUID() + let content = '' + results.conversationSave = await measure(200, index => { + content += 'Lorem ipsum dolor sit amet, streaming reply chunk. '.repeat(8) + database.saveLocalConversations([{ + header: { ...header, updatedAt: Date.now() + index }, + messages: [{ id: messageId, role: 'assistant', state: 'streaming', content, createdAt: Date.now() }] + }]) + }) + results.conversationGet = await measure(20, () => database.getConversation(conversation)) + } + + const sample = database.getActivityHistoryPage({ limit: 50 }).records + if (sample.length) { + results.activityUpdate = await measure(200, index => database.updateActivityHistory({ + changes: [{ type: 'upsert', position: 'in-place', record: { ...sample[index % sample.length], detail: `perf ${index}` } }] + })) + } + results.walBytes = existsSync(join(directory, 'assistant.sqlite-wal')) + ? require('node:fs').statSync(join(directory, 'assistant.sqlite-wal')).size : 0 + database.close() + + console.log(`[wperf] ${results.taskEvents} task events, ${results.messages} messages, ${results.activityRecords} activity records; open ${results.openMs} ms`) + const rows = [ + ['task event: subagent (real stream)', results.subagentEvents], + ['task event: tool (real)', results.toolEvents], + ['task event: text batch', results.textBatches], + ['conversation save (500 ms tick)', results.conversationSave], + ['conversation get (largest)', results.conversationGet], + ['activity update, 1 record', results.activityUpdate] + ].filter(([, value]) => value) + console.table(rows.map(([name, r]) => ({ + path: name, calls: r.calls, 'p50 ms': r.callP50, 'p95 ms': r.callP95, 'p99 ms': r.callP99, + 'max ms': r.callMax, 'loop max ms': r.loopMax, 'calls/s': Math.round(r.calls / (r.totalMs / 1000)) + }))) + for (const [name, r] of rows) if (r.slowCalls.length) console.log(`[wperf] slow calls (index, ms) ${name}: ${JSON.stringify(r.slowCalls)}`) + console.log(`[wperf] synchronous=${results.synchronous} wal_autocheckpoint=${results.walAutocheckpoint} wal=${results.walBytes} B`) + const output = process.env.GB_WPERF_OUTPUT + if (output) writeFileSync(output, JSON.stringify({ kind: 'goodbuddy-main-write-benchmark', node: process.version, results }, null, 2)) +} + +main().catch(error => { process.exitCode = 1; console.error(error) }).finally(() => { + if (process.env.GB_WPERF_KEEP === '1') console.log(`[wperf] kept ${directory}`) + else rmSync(directory, { recursive: true, force: true, maxRetries: 5, retryDelay: 200 }) +}) diff --git a/docs/roadmap/product-performance-experience-improvement-plan.md b/docs/roadmap/product-performance-experience-improvement-plan.md index 400fdc58..bfb72813 100644 --- a/docs/roadmap/product-performance-experience-improvement-plan.md +++ b/docs/roadmap/product-performance-experience-improvement-plan.md @@ -951,7 +951,36 @@ Renderer 主线程降速 4 倍（Main 不降速，每次 reload 后重新应用 测试：随机动作序列（含翻页、截断、刷盘和注入的保存失败），在真实数据库上与改动前的整份列表逻辑逐步对比，最终数据库与整份 `replace` 的结果一致（含重复 ID 和移到最前）；启动后界面不超过一页；保存载荷只取决于变化的记录（含未加载的）；随机数据集（含重复 ID）下，分页加汇总渲染出的计数、分组、标题、状态和清空数量与整份列表渲染一致；重启后遗留的进行中记录被标为中断，说明文本相同；清空后数据库、界面和汇总都为空；SQL 变更语义与参考实现一致。`perf:app` 60 项阈值全部通过。 -未做：汇总按已加载的会话查询，加载 5,000 个以上会话时只取前 5,000 个；分组里“会话活动数”统计的仍是已加载的记录，和以前统计已显示的记录一致。 +后续（2026-10-04）：汇总覆盖全部已加载会话，超过 5,000 个时分多次请求再合并，不再截断（单元测试：5,003 个会话分 5,000 + 3 两次请求，全部有标题和状态）。分组里“会话活动数”统计的仍是已加载的记录，和以前统计已显示的记录一致。 + +### 7.2.2 Main 写库与启动：真实数据测量（2026-10-04，PERF-15） + +用便携版 `dist/GoodBuddy-windows-x64/data` 的副本测量（源目录从不被打开）：`assistant.sqlite` 2.4 GB，175.5 万条任务事件（其中 148 万条子任务进度、1.1 GB），4,812 条消息（metadata 约 400 MB，单条最大 4 MB），2.4 万条活动记录。新增 `npm run perf:write`（`build/run-write-perf.cjs`，按真实数据中最忙一秒的速率 270 条/秒回放真实事件），`perf:startup` 新增 `real` 变体（`GB_PERF_STARTUP_REAL_DATA`，每次启动用新副本，外网请求全部拦截）。 + +**测到的问题：** +1. 每次写库都按 `synchronous = FULL` 同步落盘，单次约 2.4 ms，与写入内容无关：子任务进度、工具、文本批、会话保存、活动记录都一样。 +2. SQLite 自动 checkpoint 发生在越过 1,000 页的那次提交里：大库上每隔几百次写入就有一次卡住 Main 100–350 ms（3 轮里 18 次超过 50 ms）。 +3. 启动时两条恢复语句、一条图片操作恢复语句各扫描全部消息的 metadata：在 Main 上同步约 2.3 秒，启动到输入框可用 3.7 秒。 + +**改动：** +- **频繁的本地写入改用 `synchronous = NORMAL`**：本地运行事件（生命周期事件 `status`/`done`/`error` 除外）、渲染进程 500 ms 一次的会话保存、活动记录增量。WAL 模式下仍是原子提交，应用崩溃不丢；断电最多丢最近一段这类写入，任何 FULL 提交及其之前的数据不受影响。远程事件（Agent ACK 依赖其持久性）、通道投递、任务状态等其余写入仍为 FULL。 +- **WAL checkpoint 移到 worker 线程**：复用只读查询 worker 的入口，每 250 ms 做一次 PASSIVE checkpoint；运行期间 Main 的自动 checkpoint 阈值提到 4,000 页作为兜底（持续写入时被动 checkpoint 追不上 WAL 重置点），worker 退出即恢复默认 1,000。 +- **启动恢复按索引读候选消息**：第 58 版 schema 新增部分索引 `messages_recovery_idx`（流式中，或 metadata 含未结束的工具、子任务、图片操作状态）；没有用户队列项时跳过按 queueItemId 的全表删除。metadata 只由 `JSON.stringify` 写入，状态总以紧凑形式出现，精确判断仍在候选行上用原逻辑执行。一次性迁移在升级 worker 中执行，真实数据约 2 秒。 + +**结果（真实数据）：** + +| 指标 | 之前 | 之后 | +| --- | --- | --- | +| 单次写库 p50（子任务 / 工具 / 文本 / 会话保存 / 活动记录） | 2.2–2.6 ms | 0.1–0.3 ms | +| 写入期间 Main 最长阻塞（270 条/秒，3 轮） | 8–346 ms，超过 50 ms 共 37 次 | 2–34 ms，超过 50 ms 0 次 | +| 启动：Main 上的初始化（`assistant-db-init`） | 2,564 ms | 232 ms | +| 启动：到输入框可用（中位数，4 次） | 3,702 ms | 1,359 ms | +| 启动：侧栏出现会话 | 4,081 ms | 1,749 ms | +| 启动期间 Main 最长阻塞 | 2.5–2.8 s | 236–257 ms | + +部分索引的代价：metadata 更新时如果新旧值都命中候选条件，要多维护一次索引（4 MB metadata 的更新 36 → 59 ms，233 KB 的 0.7 → 1.6 ms）；不命中时没有额外开销。启动时剩下的约 0.6 s 在本地工具环境探测（已在后台初始化，不阻塞 Main）。 + +未做：读取最大的会话（407 条消息、42 MB metadata）在 worker 上约 170 ms，不阻塞 Main，但切换到这个会话时要等这么久；需要按需加载 metadata，单独立项。 ### 7.3 体验倒退复查（2026-10-03） @@ -984,7 +1013,7 @@ Renderer 主线程降速 4 倍（Main 不降速，每次 reload 后重新应用 ### PERF-15 存储接口异步化 -- **优先级 / 状态：** P1 / 待开始 +- **优先级 / 状态：** P1 / 进行中（热点读走 worker；高频写降为 NORMAL、checkpoint 移出 Main、启动恢复按索引，见 7.2.2；接口异步化未开始） - **范围：** 在原进程内把 `AssistantDatabase`、`KnowledgeDatabase` 等对外接口改为异步， 调用方全部 `await`；会话列表只返回摘要，搜索下推到 SQL，活动记录改为增量追加。 远程 Runtime 实时事件改用已有的 `appendRemoteTaskEventsBatch` 批量写入，不再逐条同步事务。 diff --git a/package.json b/package.json index 904a0650..2c5f22a1 100644 --- a/package.json +++ b/package.json @@ -31,6 +31,7 @@ "perf:knowledge": "node build/run-knowledge-perf.cjs", "perf:main": "node build/run-main-perf.cjs", "perf:startup": "node build/run-startup-perf.cjs", + "perf:write": "node build/run-write-perf.cjs", "eval:retrieval": "vitest run --config tests/support/knowledge-retrieval-evaluation.ts tests/knowledge-retrieval-metrics.test.ts tests/knowledge-retrieval-evaluation.test.ts", "build": "npm run typecheck && npm run build:bundle", "build:bundle": "electron-vite build && node build/check-control-plane-package-installer.cjs", diff --git a/src/main/assistant/activity-history-repository.test.ts b/src/main/assistant/activity-history-repository.test.ts index 713f5071..a5f109d3 100644 --- a/src/main/assistant/activity-history-repository.test.ts +++ b/src/main/assistant/activity-history-repository.test.ts @@ -19,7 +19,8 @@ const directories: string[] = [] const opened: AssistantDatabase[] = [] afterEach(() => { for (const database of opened.splice(0)) database.close() - for (const directory of directories.splice(0)) rmSync(directory, { recursive: true, force: true }) + // A readonly worker ends asynchronously after close and may still hold the file briefly. + for (const directory of directories.splice(0)) rmSync(directory, { recursive: true, force: true, maxRetries: 10, retryDelay: 100 }) }) function open(): { database: AssistantDatabase; raw: DatabaseSync; path: string; directory: string } { diff --git a/src/main/assistant/assistant-database.test.ts b/src/main/assistant/assistant-database.test.ts index 84bdab63..290ab6be 100644 --- a/src/main/assistant/assistant-database.test.ts +++ b/src/main/assistant/assistant-database.test.ts @@ -14,7 +14,8 @@ import { } from '../../shared/assistant-contracts' import { AssistantDatabase, - ASSISTANT_DATABASE_SCHEMA_VERSION + ASSISTANT_DATABASE_SCHEMA_VERSION, + MESSAGE_RECOVERY_CANDIDATE_SQL } from './assistant-database' import { agentRuntimeSelectionKey } from '../../shared/runtime-selection-contracts' import type { ImageOperation } from '../../shared/image-generation-contracts' @@ -530,6 +531,86 @@ function claimManualScheduleQueueItem( } describe('AssistantDatabase', () => { + it('writes frequent local saves with synchronous NORMAL and keeps FULL for everything else', async () => { + const directory = await mkdtemp(join(tmpdir(), 'goodbuddy-relaxed-')) + temporaryDirectories.push(directory) + const path = join(directory, 'assistant.sqlite') + const database = new AssistantDatabase(path) + database.initialize(process.cwd()) + try { + const raw = (database as unknown as { database: DatabaseSync }).database + const levels: number[] = [] + const exec = raw.exec.bind(raw) + raw.exec = (sql: string) => { + exec(sql) + if (/^\s*COMMIT/i.test(sql)) levels.push((raw.prepare('PRAGMA synchronous').get() as { synchronous: number }).synchronous) + } + const level = (): number => (raw.prepare('PRAGMA synchronous').get() as { synchronous: number }).synchronous + expect(level()).toBe(2) + const task = database.createTask({ id: randomUUID(), title: 't', instructions: 'i', workMode: 'ask' }) + database.appendTaskEvent(task.id, 'text', { type: 'text', requestId: task.id, delta: 'a' }) + expect(level()).toBe(2) + const id = randomUUID() + database.saveLocalConversations([{ header: { id, title: 'Relaxed', updatedAt: 1 }, + messages: [{ id: randomUUID(), role: 'user', state: 'complete', content: 'hi', createdAt: 1 }] }]) + expect(levels.at(-1)).toBe(1) + expect(level()).toBe(2) + // A failed relaxed write still restores FULL. + expect(() => database.saveLocalConversations([{ header: { id, title: 'Relaxed', updatedAt: 2 }, + messages: [{ id: randomUUID(), role: 'user', state: 'complete', content: 'x', createdAt: 2 }, { id: 'dup', role: 'user', state: 'complete', content: 'y', createdAt: 2 }, { id: 'dup', role: 'assistant', state: 'complete', content: 'z', createdAt: 2 }] }])) + .toThrow() + expect(level()).toBe(2) + // Lifecycle events and task status changes commit with FULL. + levels.length = 0 + database.updateTaskStatus(task.id, 'completed') + expect(level()).toBe(2) + expect(levels.every((value) => value === 2)).toBe(true) + expect(database.getConversation(id).messages.map((message) => message.content)).toEqual(['hi']) + } finally { database.close() } + }) + + it('startup recovery reads only candidate messages and ends the same ones as a full scan', async () => { + const directory = await mkdtemp(join(tmpdir(), 'goodbuddy-recovery-index-')) + temporaryDirectories.push(directory) + const path = join(directory, 'assistant.sqlite') + const database = new AssistantDatabase(path) + database.initialize(process.cwd()) + const id = randomUUID() + const message = (content: string, extra: Record = {}) => ({ + id: randomUUID(), role: 'assistant' as const, state: 'complete' as const, content, createdAt: 1, ...extra }) + const messages = [ + message('streaming', { state: 'streaming' as const }), + message('running tool', { tools: [{ name: 'a', state: 'running', summary: 's' }] }), + message('pending block', { blocks: [{ type: 'tool', tool: { name: 'b', state: 'pending', summary: 's' } }] }), + message('queued subagent', { subagents: [{ childTaskId: randomUUID(), expertId: randomUUID(), expertName: 'e', routingMode: 'native', state: 'queued' }] }), + // Mentions "running" only inside text: a candidate, but nothing to end. + message('"state":"running" in text'), + message('done', { tools: [{ name: 'c', state: 'completed', summary: 's' }] }) + ] + database.saveLocalConversations([{ header: { id, title: 'Recovery', updatedAt: 1 }, messages }]) + const raw = (database as unknown as { database: DatabaseSync }).database + const plan = (raw.prepare(`EXPLAIN QUERY PLAN SELECT id FROM messages WHERE ${MESSAGE_RECOVERY_CANDIDATE_SQL}`) + .all() as Array<{ detail: string }>).map((row) => row.detail).join(' ') + expect(plan).toContain('messages_recovery_idx') + const imagePlan = (raw.prepare(`EXPLAIN QUERY PLAN SELECT id FROM messages + WHERE ${MESSAGE_RECOVERY_CANDIDATE_SQL} AND json_type(metadata_json, '$.imageOperations') = 'array'`) + .all() as Array<{ detail: string }>).map((row) => row.detail).join(' ') + expect(imagePlan).toContain('messages_recovery_idx') + database.close() + + const recovered = new AssistantDatabase(path) + recovered.initialize(process.cwd()) + try { + const after = recovered.getConversation(id).messages + expect(after.map((item) => item.state)).toEqual(['error', 'complete', 'complete', 'complete', 'complete', 'complete']) + expect(after[1]?.tools?.[0]?.state).toBe('interrupted') + expect((after[2]?.blocks?.[0] as { tool: { state: string } }).tool.state).toBe('interrupted') + expect(after[3]?.subagents?.[0]?.state).toBe('cancelled') + expect(after[4]?.content).toBe('"state":"running" in text') + expect(after[5]?.tools?.[0]?.state).toBe('completed') + } finally { recovered.close() } + }) + it('round trips long checklist metadata and explicit clears across reopen', async () => { const directory = await mkdtemp(join(tmpdir(), 'goodbuddy-checklist-')) temporaryDirectories.push(directory) diff --git a/src/main/assistant/assistant-database.ts b/src/main/assistant/assistant-database.ts index 4486d76c..06d08db9 100644 --- a/src/main/assistant/assistant-database.ts +++ b/src/main/assistant/assistant-database.ts @@ -4,6 +4,7 @@ import { magicNoteCanvasAnalysisText } from '../../shared/magic-note-canvas-text import { knowledgeReferenceKey } from '../../shared/knowledge-reference' import { appendConversationQuestionBlock } from '../../shared/conversation-question-blocks' import { DatabaseSync } from 'node:sqlite' +import { Worker } from 'node:worker_threads' import { ExecutionStatsReader } from './execution-stats-reader' import { ReadonlyQueryReader, readWithFallback } from '../readonly-query-reader' import { ActivityHistoryRepository, migrateActivityHistoryOrder } from './activity-history-repository' @@ -138,7 +139,23 @@ import { SUBAGENT_PROGRESS_STORAGE_SCHEMA_VERSION } from './subagent-progress-storage' -export const ASSISTANT_DATABASE_SCHEMA_VERSION = 57 +export const ASSISTANT_DATABASE_SCHEMA_VERSION = 58 +/** + * Messages startup recovery may have to end (schema 58): still streaming, or + * metadata naming an unfinished tool/subagent (pending, running, queued) or + * image operation (running, saving, cancelling). Metadata is always minified + * JSON.stringify output, so every such state appears literally; the exact + * checks run on these rows only, through the partial index + * messages_recovery_idx. + */ +export const MESSAGE_RECOVERY_CANDIDATE_SQL = `(state = 'streaming' + OR instr(metadata_json, '"state":"running"') > 0 + OR instr(metadata_json, '"state":"pending"') > 0 + OR instr(metadata_json, '"state":"queued"') > 0 + OR instr(metadata_json, '"state":"saving"') > 0 + OR instr(metadata_json, '"state":"cancelling"') > 0)` +/** Main's automatic checkpoint threshold while the checkpoint worker runs (16 MB of 4 KB pages). */ +const WAL_SAFETY_CHECKPOINT_PAGES = 4_000 export type RemoteTaskEventInput = { taskId: string @@ -1973,6 +1990,7 @@ export class AssistantDatabase { private activityHistoryRepository?: { database: DatabaseSync; repository: ActivityHistoryRepository } private executionStatsReader?: ExecutionStatsReader private readonlyReader?: ReadonlyQueryReader + private checkpointWorker?: Worker private readonlyWorkerPath?: string private foldedSearchConnection?: DatabaseSync private readonly noteStorage: MagicNoteStorage @@ -2271,12 +2289,15 @@ export class AssistantDatabase { )` ) .run() - database.exec(`DELETE FROM messages WHERE json_extract(metadata_json, '$.queueItemId') IN ( - SELECT id FROM conversation_queue_items WHERE source = 'user' - ) AND EXISTS ( - SELECT 1 FROM json_each(messages.metadata_json, '$.attachments') attachment - WHERE json_extract(attachment.value, '$.resourceId') IS NOT NULL - )`) + // This scans every message; with no user queue item it cannot match anything. + if (database.prepare("SELECT 1 FROM conversation_queue_items WHERE source = 'user' LIMIT 1").get()) { + database.exec(`DELETE FROM messages WHERE json_extract(metadata_json, '$.queueItemId') IN ( + SELECT id FROM conversation_queue_items WHERE source = 'user' + ) AND EXISTS ( + SELECT 1 FROM json_each(messages.metadata_json, '$.attachments') attachment + WHERE json_extract(attachment.value, '$.resourceId') IS NOT NULL + )`) + } database.exec(` INSERT OR IGNORE INTO conversation_queue_items (id, conversation_id, source, label, payload_json, @@ -2318,9 +2339,13 @@ export class AssistantDatabase { const recoverableMessages = database .prepare( + // The partial index messages_recovery_idx holds exactly the rows + // matching the first term (a superset of those updated below), so + // startup no longer reads every message's metadata. `SELECT id, state, metadata_json FROM messages - WHERE NOT EXISTS ( + WHERE ${MESSAGE_RECOVERY_CANDIDATE_SQL} + AND NOT EXISTS ( SELECT 1 FROM tasks WHERE tasks.id = messages.request_id AND tasks.remote_recoverable = 1 @@ -2402,6 +2427,7 @@ export class AssistantDatabase { } close(): void { + this.stopWalCheckpointWorker() this.executionStatsReader?.close() this.executionStatsReader = undefined this.readonlyReader?.close() @@ -3337,6 +3363,11 @@ export class AssistantDatabase { } saveLocalConversations(batch: LocalConversationSaveBatch): void { + // The renderer saves every 500 ms and retries until acknowledged. + this.relaxedWrite(() => this.saveLocalConversationsNow(batch)) + } + + private saveLocalConversationsNow(batch: LocalConversationSaveBatch): void { const database = this.requireDatabase() const findConversation = database.prepare( 'SELECT channel, context_state_json FROM conversations WHERE id = ?' @@ -5013,6 +5044,69 @@ export class AssistantDatabase { this.database = new DatabaseSync(this.databasePath, { readOnly: true, timeout: 5_000 }) } + /** + * Opens a connection used only to checkpoint the WAL from a worker thread + * (PERF-15). It never writes rows; see enableWalCheckpointWorker. + */ + openCheckpointer(): void { + if (this.database) throw new Error('Database already open') + this.database = new DatabaseSync(this.databasePath, { timeout: 5_000 }) + this.database.exec('PRAGMA busy_timeout = 5000; PRAGMA wal_autocheckpoint = 0') + } + + /** One PASSIVE checkpoint: copies committed WAL frames without blocking writers. */ + checkpointWal(): { busy: number; log: number; checkpointed: number } { + if (!this.database) throw new Error('Database is not open') + return this.database.prepare('PRAGMA wal_checkpoint(PASSIVE)').get() as + { busy: number; log: number; checkpointed: number } + } + + /** + * Moves WAL checkpoints off Main (PERF-15). SQLite's automatic checkpoint + * runs inside the commit that crosses 1,000 WAL pages, so on a large + * database one ordinary write now and then blocked Main for 100-250 ms + * while pages were copied and synced. A worker thread checkpoints on its + * own connection instead. While it runs, Main's automatic checkpoint stays + * as a safety net at a higher threshold: under sustained writes a passive + * checkpoint never catches up exactly, so the WAL could not reset; Main's + * checkpoint then finds almost every frame already copied and only + * finishes the rest. The default threshold comes back if the worker stops. + */ + enableWalCheckpointWorker(workerPath: string, intervalMs = 250): void { + const database = this.database + if (!database || this.checkpointWorker || this.databasePath === ':memory:') return + let worker: Worker + try { + worker = new Worker(workerPath, { + workerData: { kind: 'checkpoint', databasePath: this.databasePath, intervalMs } + }) + } catch { + return + } + this.checkpointWorker = worker + worker.unref() + const restore = (): void => { + if (this.checkpointWorker !== worker) return + this.checkpointWorker = undefined + if (this.database === database) database.exec('PRAGMA wal_autocheckpoint = 1000') + } + worker.on('message', (message: { ready?: boolean }) => { + if (message.ready && this.checkpointWorker === worker && this.database === database) { + database.exec(`PRAGMA wal_autocheckpoint = ${WAL_SAFETY_CHECKPOINT_PAGES}`) + } + }) + worker.once('error', restore) + worker.once('exit', restore) + } + + private stopWalCheckpointWorker(): void { + const worker = this.checkpointWorker + if (!worker) return + this.checkpointWorker = undefined + this.database?.exec('PRAGMA wal_autocheckpoint = 1000') + void worker.terminate() + } + /** * Runs multi-statement reads against one WAL snapshot. Used by the readonly * worker, where Main may commit between statements; on Main itself the @@ -5312,7 +5406,30 @@ export class AssistantDatabase { /** Applies incremental changes in one transaction (PERF-15). */ updateActivityHistory(input: unknown): ActivityHistoryUpdateResult { - return this.activityHistory().update(input) + return this.relaxedWrite(() => this.activityHistory().update(input)) + } + + /** + * Runs a high-frequency local write with `synchronous = NORMAL` instead of + * FULL (PERF-15). In WAL mode NORMAL skips the fsync per commit; the commit + * is still atomic and survives an application crash, and the next FULL + * commit or checkpoint makes it durable. A power loss can only drop a + * suffix of such local writes, never a FULL commit or anything before it, + * because the WAL is synced as a whole. Writes another party may rely on + * (remote acknowledgements, channel delivery) keep using FULL. + */ + private relaxedWrite(body: () => T): T { + const database = this.requireDatabase() + // Inside a caller's transaction the caller's commit decides durability. + if (database.isTransaction) return body() + const previous = (database.prepare('PRAGMA synchronous').get() as { synchronous: number }).synchronous + if (previous <= 1) return body() + database.exec('PRAGMA synchronous = NORMAL') + try { + return body() + } finally { + database.exec(`PRAGMA synchronous = ${previous}`) + } } /** Deletes every activity record in one transaction. */ @@ -6184,6 +6301,20 @@ export class AssistantDatabase { taskId: string, kind: string, payload: unknown + ): void { + // Local run events stream many times per second. Lifecycle events stay + // FULL; their commit also makes every earlier relaxed event durable. + if (kind === 'status' || kind === 'done' || kind === 'error') { + this.appendTaskEventNow(taskId, kind, payload) + } else { + this.relaxedWrite(() => this.appendTaskEventNow(taskId, kind, payload)) + } + } + + private appendTaskEventNow( + taskId: string, + kind: string, + payload: unknown ): void { const database = this.requireDatabase() const payloadJson = this.subagentProgress!.serialize(taskId, kind, payload) @@ -7109,7 +7240,9 @@ export class AssistantDatabase { const database = this.requireDatabase() database.exec('BEGIN IMMEDIATE') try { - const rows = database.prepare("SELECT id, metadata_json FROM messages WHERE json_type(metadata_json, '$.imageOperations') = 'array'") + // Candidates come from messages_recovery_idx instead of a scan of every message. + const rows = database.prepare(`SELECT id, metadata_json FROM messages + WHERE ${MESSAGE_RECOVERY_CANDIDATE_SQL} AND json_type(metadata_json, '$.imageOperations') = 'array'`) .all() as { id: string; metadata_json: string }[] for (const row of rows) { const metadata = JSON.parse(row.metadata_json) as MessageMetadata @@ -12257,6 +12390,16 @@ export class AssistantDatabase { database.exec('PRAGMA user_version = 57; COMMIT;') } catch (error) { database.exec('ROLLBACK'); throw error } } + if (version.user_version < 58) { + // Startup recovery reads candidate messages by index instead of scanning all (PERF-15). + database.exec(` + BEGIN IMMEDIATE; + CREATE INDEX IF NOT EXISTS messages_recovery_idx ON messages(id) + WHERE ${MESSAGE_RECOVERY_CANDIDATE_SQL}; + PRAGMA user_version = 58; + COMMIT; + `) + } } supervisionExperiences(): SupervisionExperienceStore { diff --git a/src/main/index.ts b/src/main/index.ts index 52695698..ccfa6b5c 100644 --- a/src/main/index.ts +++ b/src/main/index.ts @@ -1275,6 +1275,7 @@ if (hasSingleInstanceLock) { startupMark('main:assistant-db-init:start') startupAssistantDatabase.initialize(defaultWorkspace) startupAssistantDatabase.enableReadonlyWorker(readonlyQueryWorkerPath) + startupAssistantDatabase.enableWalCheckpointWorker(readonlyQueryWorkerPath) imageGenerationService!.initialize() startupAssistantDatabase.ensureChannelProjects( defaultWorkspace, diff --git a/src/main/readonly-query-reader.test.ts b/src/main/readonly-query-reader.test.ts index d0719b05..5651e6aa 100644 --- a/src/main/readonly-query-reader.test.ts +++ b/src/main/readonly-query-reader.test.ts @@ -1,5 +1,6 @@ // @vitest-environment node import { createHash, randomUUID } from 'node:crypto' +import { statSync } from 'node:fs' import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join, resolve } from 'node:path' @@ -292,3 +293,37 @@ describe('assistant readonly worker', () => { } finally { database.close() } }, 60_000) }) + +describe('WAL checkpoint worker', () => { + it('checkpoints off Main, keeps a raised safety threshold and restores the default when stopped', async () => { + const path = join(directory, `checkpoint-${randomUUID()}.sqlite`) + const database = new AssistantDatabase(path) + database.initialize(directory) + const raw = (database as unknown as { database: import('node:sqlite').DatabaseSync }).database + const autocheckpoint = (): number => + (raw.prepare('PRAGMA wal_autocheckpoint').get() as { wal_autocheckpoint: number }).wal_autocheckpoint + try { + expect(autocheckpoint()).toBe(1000) + database.enableWalCheckpointWorker(workerPath, 20) + await expect.poll(autocheckpoint, { timeout: 10_000 }).toBe(4000) + const task = database.createTask({ id: randomUUID(), title: 't', instructions: 'i', workMode: 'ask' }) + const sizeBefore = statSync(path).size + for (let index = 0; index < 300; index += 1) { + database.appendTaskEvent(task.id, 'text', { type: 'text', requestId: task.id, delta: 'x'.repeat(4_000) }) + } + // ~300 WAL pages stay below Main's 4,000-page threshold, so only the + // worker's checkpoint can copy them into the database file. + await expect.poll(() => statSync(path).size, { timeout: 10_000 }).toBeGreaterThan(sizeBefore + 1_000_000) + expect((raw.prepare("SELECT COUNT(*) AS n FROM task_events WHERE task_id = ? AND kind = 'text'").get(task.id) as { n: number }).n).toBe(300) + } finally { + database.close() + } + // A reopened database starts with SQLite's default threshold again. + const reopened = new AssistantDatabase(path) + reopened.initialize(directory) + try { + const again = (reopened as unknown as { database: import('node:sqlite').DatabaseSync }).database + expect((again.prepare('PRAGMA wal_autocheckpoint').get() as { wal_autocheckpoint: number }).wal_autocheckpoint).toBe(1000) + } finally { reopened.close() } + }, 60_000) +}) diff --git a/src/main/readonly-query-worker.ts b/src/main/readonly-query-worker.ts index 0197135d..e53be30f 100644 --- a/src/main/readonly-query-worker.ts +++ b/src/main/readonly-query-worker.ts @@ -8,13 +8,35 @@ import type { ReadonlyQueryKind, ReadonlyQueryRequest } from './readonly-query-r // It reuses the production classes on a read-only connection, so results are // produced by exactly the same code as the synchronous path. -const { kind, databasePath } = workerData as { kind: ReadonlyQueryKind; databasePath: string } +const { kind, databasePath, intervalMs } = workerData as { + kind: ReadonlyQueryKind | 'checkpoint' + databasePath: string + intervalMs?: number +} type Handler = (args: unknown[], signal: AbortSignal) => unknown let handlers: Record let close: () => void -if (kind === 'knowledge') { +if (kind === 'checkpoint') { + // WAL checkpointer for AssistantDatabase.enableWalCheckpointWorker (PERF-15): + // PASSIVE checkpoints never wait for, or block, Main's readers and writers. + const database = new AssistantDatabase(databasePath) + database.openCheckpointer() + const timer = setInterval(() => { + try { + database.checkpointWal() + } catch { + // SQLITE_BUSY and similar: try again on the next tick. + } + }, intervalMs ?? 1_000) + close = () => { + clearInterval(timer) + database.close() + } + handlers = {} + parentPort!.postMessage({ ready: true }) +} else if (kind === 'knowledge') { const database = new KnowledgeDatabase(databasePath) database.openReadOnly() close = () => database.close()