perf: run knowledge, story graph and conversation search reads in a read-only worker

This commit is contained in:
mesalogo
2026-10-03 02:53:51 +08:00
parent 53b9ecbcae
commit f571301764
13 changed files with 799 additions and 65 deletions
+67 -15
View File
@@ -8,9 +8,24 @@ const { createHash } = require('node:crypto')
const { mkdtempSync, rmSync, writeFileSync } = require('node:fs')
const { tmpdir } = require('node:os')
const { join, resolve } = require('node:path')
const { performance } = require('node:perf_hooks')
const { monitorEventLoopDelay, performance } = require('node:perf_hooks')
const root = resolve(__dirname, '..')
// Measure on the project's Electron (the production runtime). Older runtimes
// (Node 24.18 / Electron 43) stall worker threads for hundreds of ms on many
// small ArrayBuffer allocations (SQLite BLOB rows); GB_KPERF_NODE=1 opts out.
if (process.env.GB_KPERF_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 sizes = (process.env.GB_KPERF_SIZES || '5000,20000,50000').split(',').map(Number)
const dimensions = Number(process.env.GB_KPERF_DIMENSIONS || 384)
const chunksPerDocument = 100
@@ -18,15 +33,23 @@ const queries = Number(process.env.GB_KPERF_QUERIES || 15)
const directory = mkdtempSync(join(tmpdir(), 'goodbuddy-knowledge-perf-'))
const bundle = join(directory, 'knowledge-database.cjs')
buildSync({
entryPoints: [join(root, 'src/main/knowledge/knowledge-database.ts')],
bundle: true,
platform: 'node',
format: 'cjs',
target: 'node22',
outfile: bundle,
logLevel: 'error'
})
const workerBundle = join(directory, 'readonly-query-worker.cjs')
for (const [entry, outfile] of [
['src/main/knowledge/knowledge-database.ts', bundle],
['src/main/readonly-query-worker.ts', workerBundle]
]) {
buildSync({
entryPoints: [join(root, entry)],
bundle: true,
platform: 'node',
format: 'cjs',
target: 'node22',
outfile,
logLevel: 'error',
// The worker reuses AssistantDatabase, which imports Electron-only modules lazily.
external: ['electron']
})
}
const { KnowledgeDatabase } = require(bundle)
// Deterministic pseudo-random generator so runs are comparable.
@@ -53,8 +76,30 @@ function time(body) {
}
const round = value => Math.round(value * 10) / 10
// Main event-loop delay while `queries` searches run back to back. A sync
// search blocks the loop for its whole duration; a worker search should not.
async function loopDelay(body) {
const histogram = monitorEventLoopDelay({ resolution: 5 })
await body() // warm worker / caches outside the measurement
// Let the warm-up's blocked interval drain before sampling.
await new Promise(resolve => setTimeout(resolve, 50))
histogram.enable()
const started = performance.now()
for (let index = 0; index < queries; index += 1) {
await body()
// Yield like a real IPC handler, so the histogram samples between queries.
await new Promise(resolve => setImmediate(resolve))
}
const wallMs = performance.now() - started
await new Promise(resolve => setTimeout(resolve, 20))
histogram.disable()
const ms = value => round(value / 1e6)
return { p50: ms(histogram.percentile(50)), p99: ms(histogram.percentile(99)), max: ms(histogram.max), perQueryMs: round(wallMs / queries) }
}
const results = []
try {
async function main() {
console.log(`[kperf] runtime: node ${process.version}${process.versions.electron ? ` (electron ${process.versions.electron})` : ''}`)
for (const size of sizes) {
const path = join(directory, `knowledge-${size}.sqlite`)
const database = new KnowledgeDatabase(path)
@@ -82,15 +127,22 @@ try {
seedSeconds: round(seedMs / 1000),
ftsMs: time(() => reopened.search({ knowledgeBaseId: library.id, query: 'lighthouse budget', limit: 40 })),
vectorMs: time(() => reopened.vectorSearch({ knowledgeBaseId: library.id, provider: 'perf', model: 'perf-model', vector: query, limit: 40 })),
hybridMs: time(() => reopened.hybridSearchWithDiagnostics({ knowledgeBaseId: library.id, query: 'lighthouse budget', limit: 40, provider: 'perf', model: 'perf-model', vector: query, graphEnabled: false, candidateMultiplier: 1 }))
hybridMs: time(() => reopened.hybridSearchWithDiagnostics(hybridOptions(library.id, query)))
}
// Main event-loop delay: synchronous search on Main vs the readonly worker.
entry.mainLoopDelaySyncMs = await loopDelay(async () => reopened.hybridSearchWithDiagnostics(hybridOptions(library.id, query)))
reopened.enableReadonlyWorker(workerBundle)
entry.mainLoopDelayWorkerMs = await loopDelay(() => reopened.hybridSearchWithDiagnosticsAsync(hybridOptions(library.id, query)))
reopened.close()
results.push(entry)
console.log(`[kperf] ${size} chunks: fts p95 ${entry.ftsMs.p95} ms, vector p95 ${entry.vectorMs.p95} ms, hybrid p95 ${entry.hybridMs.p95} ms (seed ${entry.seedSeconds} s)`)
console.log(`[kperf] ${size} chunks: Main loop delay max sync ${entry.mainLoopDelaySyncMs.max} ms vs worker ${entry.mainLoopDelayWorkerMs.max} ms (p99 ${entry.mainLoopDelaySyncMs.p99} / ${entry.mainLoopDelayWorkerMs.p99} ms; per query ${entry.mainLoopDelaySyncMs.perQueryMs} / ${entry.mainLoopDelayWorkerMs.perQueryMs} ms)`)
}
const output = process.env.GB_KPERF_OUTPUT
if (output) writeFileSync(output, JSON.stringify({ kind: 'goodbuddy-knowledge-retrieval-benchmark', node: process.version, results }, null, 2))
console.table(results.map(r => ({ chunks: r.chunks, 'fts p50/p95': `${r.ftsMs.p50} / ${r.ftsMs.p95}`, 'vector p50/p95': `${r.vectorMs.p50} / ${r.vectorMs.p95}`, 'hybrid p50/p95': `${r.hybridMs.p50} / ${r.hybridMs.p95}` })))
} finally {
rmSync(directory, { recursive: true, force: true, maxRetries: 5, retryDelay: 200 })
console.table(results.map(r => ({ chunks: r.chunks, 'fts p50/p95': `${r.ftsMs.p50} / ${r.ftsMs.p95}`, 'vector p50/p95': `${r.vectorMs.p50} / ${r.vectorMs.p95}`, 'hybrid p50/p95': `${r.hybridMs.p50} / ${r.hybridMs.p95}`, 'loop max sync/worker': `${r.mainLoopDelaySyncMs.max} / ${r.mainLoopDelayWorkerMs.max}` })))
}
const hybridOptions = (knowledgeBaseId, vector) => ({ knowledgeBaseId, query: 'lighthouse budget', limit: 40, provider: 'perf', model: 'perf-model', vector, graphEnabled: false, candidateMultiplier: 1 })
main().catch(error => { process.exitCode = 1; console.error(error) }).finally(() => {
rmSync(directory, { recursive: true, force: true, maxRetries: 5, retryDelay: 200 })
})
+1
View File
@@ -133,6 +133,7 @@ export default defineConfig({
index: resolve('src/main/index.ts'),
'assistant-storage-worker': resolve('src/main/assistant-storage-worker.ts'),
'execution-stats-worker': resolve('src/main/execution-stats-worker.ts'),
'readonly-query-worker': resolve('src/main/readonly-query-worker.ts'),
'wechat-sidecar': resolve(
'src/main/channels/wechat-sidecar.ts'
),
+2 -2
View File
@@ -10,7 +10,7 @@ export type StoryGraphRemoteBinding = {
}
export type StoryGraphService = {
available(binding: StoryGraphBinding): Promise<boolean>
read(name: StoryGraphToolName, input: unknown, projectId?: string, signal?: AbortSignal): Record<string, unknown>
read(name: StoryGraphToolName, input: unknown, projectId?: string, signal?: AbortSignal): Record<string, unknown> | Promise<Record<string, unknown>>
}
import { knowledgeReferenceKey, toKnowledgeReference } from '../../shared/knowledge-reference'
import {
@@ -1611,7 +1611,7 @@ export class KnowledgeMcpGateway {
effectiveSignal.throwIfAborted()
if (!await this.isStoryGraphAvailable(token)) throw new Error('story_graph_unavailable: Supervisor or runtime capability is disabled')
effectiveSignal.throwIfAborted()
const result = this.storyGraphService!.read(name, input, capability.storyGraph!.projectId, effectiveSignal)
const result = await this.storyGraphService!.read(name, input, capability.storyGraph!.projectId, effectiveSignal)
effectiveSignal.throwIfAborted()
if (!await this.isStoryGraphAvailable(token)) throw new Error('story_graph_unavailable: Supervisor or runtime capability is disabled')
effectiveSignal.throwIfAborted()
+112 -41
View File
@@ -5,6 +5,7 @@ import { knowledgeReferenceKey } from '../../shared/knowledge-reference'
import { appendConversationQuestionBlock } from '../../shared/conversation-question-blocks'
import { DatabaseSync } from 'node:sqlite'
import { ExecutionStatsReader } from './execution-stats-reader'
import { ReadonlyQueryReader, readWithFallback } from '../readonly-query-reader'
import { statSync } from 'node:fs'
import { MagicNoteStorage } from '../magic-notes/magic-note-storage'
import type { AssistantStorageProgress } from '../../shared/assistant-storage-contracts'
@@ -1964,6 +1965,9 @@ export class AssistantDatabase {
private database?: DatabaseSync
private activityHistoryCache?: { records: Map<string, string>; dataVersion: number }
private executionStatsReader?: ExecutionStatsReader
private readonlyReader?: ReadonlyQueryReader
private readonlyWorkerPath?: string
private foldedSearchConnection?: DatabaseSync
private readonly noteStorage: MagicNoteStorage
private readonly dirtyMagicNotes = new Set<string>()
private readonly pendingMagicNoteCleanup = new Set<string | undefined>()
@@ -2393,6 +2397,9 @@ export class AssistantDatabase {
close(): void {
this.executionStatsReader?.close()
this.executionStatsReader = undefined
this.readonlyReader?.close()
this.readonlyReader = undefined
this.foldedSearchConnection = undefined
this.executionStatsCache.clear()
this.activityHistoryCache = undefined
this.database?.close()
@@ -3021,61 +3028,99 @@ export class AssistantDatabase {
private readConversationList(detailIds?: ReadonlySet<string>): import('../../shared/assistant-contracts').ConversationListSnapshot[] {
const database = this.requireDatabase()
const conversations = this.listConversationRows()
const messageStatement = database.prepare(
`SELECT id, conversation_id, role, content, state, metadata_json,
created_at
FROM (
SELECT id, conversation_id, role, content, state, metadata_json,
created_at, sequence
FROM messages
WHERE conversation_id = ?
ORDER BY sequence DESC
)
ORDER BY sequence ASC`
)
const summaryStatement = database.prepare(
`SELECT count(*) AS count,
max(state = 'streaming') AS streaming,
max(created_at) AS latestMessageAt,
(SELECT role FROM messages WHERE conversation_id = ? ORDER BY sequence LIMIT 1) AS firstRole
FROM messages WHERE conversation_id = ?`
)
type MessageSummaryRow = {
conversation_id: string; count: number; streaming: number | null
firstRole: ConversationMessage['role'] | null; latestMessageAt: string | null
}
const activeIds = new Set(
(database.prepare(`SELECT conversation_id FROM (${activeVisibleTaskSelect})`).all() as
{ conversation_id: string }[]).map(row => row.conversation_id)
)
// Batched reads (PERF-15): one aggregate for every summary and one ordered
// scan for every detailed conversation, instead of 1-2 statements each.
const summaries = new Map<string, MessageSummaryRow>()
if (detailIds) {
for (const row of database.prepare(
`SELECT m.conversation_id AS conversation_id, count(*) AS count,
max(m.state = 'streaming') AS streaming,
max(m.created_at) AS latestMessageAt,
(SELECT f.role FROM messages f WHERE f.conversation_id = m.conversation_id
ORDER BY f.sequence LIMIT 1) AS firstRole
FROM messages m JOIN conversations c ON c.id = m.conversation_id
WHERE c.status = 'active'
GROUP BY m.conversation_id`
).all() as MessageSummaryRow[]) summaries.set(row.conversation_id, row)
}
const summaryOf = (id: string): MessageSummaryRow => summaries.get(id) ??
{ conversation_id: id, count: 0, streaming: null, firstRole: null, latestMessageAt: null }
const detailed = conversations.filter(conversation =>
!detailIds || detailIds.has(conversation.id) || activeIds.has(conversation.id) ||
Boolean(summaryOf(conversation.id).streaming))
const messagesByConversation = new Map<string, MessageRow[]>()
if (detailed.length > 0) {
for (const row of database.prepare(
`SELECT id, conversation_id, role, content, state, metadata_json, created_at
FROM messages
WHERE conversation_id IN (SELECT value FROM json_each(?))
ORDER BY conversation_id, sequence ASC`
).all(JSON.stringify(detailed.map(conversation => conversation.id))) as MessageRow[]) {
const list = messagesByConversation.get(row.conversation_id)
if (list) list.push(row)
else messagesByConversation.set(row.conversation_id, [row])
}
}
const detailedIds = new Set(detailed.map(conversation => conversation.id))
return conversations.map((conversation) => {
if (detailIds && !detailIds.has(conversation.id) && !activeIds.has(conversation.id)) {
const summary = summaryStatement.get(conversation.id, conversation.id) as {
count: number; streaming: number | null; firstRole: ConversationMessage['role'] | null; latestMessageAt: string | null
}
if (!summary.streaming) {
return {
...toConversationSnapshot(conversation, []),
messageSummary: {
count: summary.count,
firstRole: summary.firstRole ?? undefined,
latestMessageAt: summary.latestMessageAt === null ? undefined : Date.parse(summary.latestMessageAt)
}
if (!detailedIds.has(conversation.id)) {
const summary = summaryOf(conversation.id)
return {
...toConversationSnapshot(conversation, []),
messageSummary: {
count: summary.count,
firstRole: summary.firstRole ?? undefined,
latestMessageAt: summary.latestMessageAt === null ? undefined : Date.parse(summary.latestMessageAt)
}
}
}
return toConversationSnapshot(conversation, messageStatement.all(conversation.id) as MessageRow[])
return toConversationSnapshot(conversation, messagesByConversation.get(conversation.id) ?? [])
})
}
/** Registers the Unicode case-folding matcher used by conversation search. */
private foldedSearchDatabase(): DatabaseSync {
const database = this.requireDatabase()
if (this.foldedSearchConnection !== database) {
// Same folding as the renderer's search (String#toLocaleLowerCase), which
// SQLite's ASCII-only lower()/LIKE cannot reproduce for non-ASCII text.
database.function('goodbuddy_folded_includes', { deterministic: true }, (text, needle) =>
typeof text === 'string' && typeof needle === 'string' &&
text.toLocaleLowerCase().includes(needle) ? 1 : 0)
this.foldedSearchConnection = database
}
return database
}
searchConversations(query: string): string[] {
const normalized = query.trim().toLocaleLowerCase()
if (!normalized) return []
const database = this.requireDatabase()
// Read only searchable text, never tool/subagent/image metadata. Use the
// same Unicode case folding as the renderer's existing search.
const messages = database.prepare('SELECT content FROM messages WHERE conversation_id = ?')
return this.listConversationRows().filter(conversation =>
conversation.title.toLocaleLowerCase().includes(normalized) ||
(messages.all(conversation.id) as { content: string }[]).some(message =>
message.content.toLocaleLowerCase().includes(normalized))
).map(conversation => conversation.id)
// Read only searchable text, never tool/subagent/image metadata. One query
// with the same Unicode case folding as the renderer's existing search.
return (this.foldedSearchDatabase().prepare(
`SELECT c.id FROM conversations c
WHERE c.status = 'active' AND (
goodbuddy_folded_includes(c.title, ?1)
OR EXISTS (SELECT 1 FROM messages m WHERE m.conversation_id = c.id
AND goodbuddy_folded_includes(m.content, ?1)))
ORDER BY c.pinned DESC, c.updated_at DESC`
).all(normalized) as { id: string }[]).map(row => row.id)
}
/** Conversation search on the readonly worker; falls back to the synchronous query. */
searchConversationsAsync(query: string, signal?: AbortSignal): Promise<string[]> {
this.requireDatabase()
if (!query.trim()) return Promise.resolve([])
return readWithFallback(this.readonlyQueryReader(), 'searchConversations', [query], signal,
() => this.searchConversations(query))
}
getConversation(conversationId: string): ConversationSnapshot {
@@ -4924,6 +4969,25 @@ export class AssistantDatabase {
this.database = new DatabaseSync(this.databasePath, { readOnly: true, timeout: 5_000 })
}
/**
* Routes heavy read-only queries (story graph, conversation search) through a
* read-only worker thread. In-memory databases keep the synchronous path.
*/
enableReadonlyWorker(workerPath: string): void {
this.readonlyWorkerPath = workerPath
}
private readonlyQueryReader(): ReadonlyQueryReader | undefined {
if (!this.readonlyWorkerPath || this.databasePath === ':memory:') return undefined
this.readonlyReader ??= new ReadonlyQueryReader('assistant', this.databasePath, this.readonlyWorkerPath)
return this.readonlyReader
}
/** Test hook for the worker crash and cancellation paths. */
get readonlyWorkerForTest(): ReadonlyQueryReader | undefined {
return this.readonlyQueryReader()
}
private readExecutionStatsSnapshot(
scope: ExecutionStatsInput,
activeRequestIds: ReadonlySet<string>
@@ -9236,6 +9300,13 @@ export class AssistantDatabase {
return readStoryGraph(this.requireDatabase(), name, input, projectId, signal)
}
/** Story graph read on the readonly worker; falls back to the synchronous read. */
readStoryGraphAsync(name: import('../../shared/story-graph-tools').StoryGraphToolName, input: unknown, projectId?: string, signal?: AbortSignal): Promise<Record<string, unknown>> {
this.requireDatabase()
return readWithFallback(this.readonlyQueryReader(), 'readStoryGraph', [name, input, projectId], signal,
() => this.readStoryGraph(name, input, projectId, signal))
}
getSupervisionResult(id: string): { id: string; summary: string } | undefined {
return this.requireDatabase().prepare('SELECT id, summary FROM supervision_results WHERE id = ?').get(id) as { id: string; summary: string } | undefined
}
@@ -138,3 +138,25 @@ it('pages Unicode snapshots, reads explicit current versions, and reports deleti
f.sql.prepare('UPDATE supervision_sources SET locator_json = ? WHERE id = ?').run(JSON.stringify({ detail: 'x'.repeat(100_001) }), source.id!)
expect(() => f.read('story_graph_read_source', { source_reference_id: source.id })).toThrow('response_too_large')
})
it('prefilters story lines by scope in SQL with the exact normalized match, and caps the scanned objects', async () => {
const f = await fixture()
f.save('Decision A', '2026-09-01T00:00:00Z')
const insert = f.sql.prepare('INSERT INTO story_lines (id, scope_json, title, created_at, updated_at) VALUES (?, ?, ?, ?, ?)')
const now = '2026-09-01T00:00:00.000Z'
// Unsorted / duplicate ids still normalize to the requested scope; supersets and other kinds do not.
insert.run('dup', JSON.stringify({ kind: 'projects', projectIds: [f.project.id, f.project.id] }), 'Decision dup', now, now)
insert.run('superset', JSON.stringify({ kind: 'projects', projectIds: [f.other.id, f.project.id] }), 'Decision superset', now, now)
insert.run('global', JSON.stringify({ kind: 'global' }), 'Decision global', now, now)
const lines = (scope?: unknown) => ((f.read('story_graph_search', { query: 'Decision', page_size: 50, object_types: ['story_line'], ...(scope ? { scope } : {}) }) as Page)
.items.map(item => item.object_ref.id)).sort()
const all = f.sql.prepare('SELECT id FROM story_lines').all().map(row => String(row.id))
expect(lines()).toEqual(expect.arrayContaining(['dup']))
expect(lines()).not.toContain('superset')
expect(lines()).not.toContain('global')
expect(lines({ kind: 'global' })).toEqual(['global'])
expect(lines({ kind: 'projects', projectIds: [f.project.id, f.other.id] })).toEqual(['superset'])
expect(all.length).toBeGreaterThanOrEqual(4)
expect(() => readStoryGraph(f.sql, 'story_graph_search', { query: 'Decision' }, f.project.id, undefined, 2)).toThrow('scan_limit_exceeded')
expect(readStoryGraph(f.sql, 'story_graph_search', { query: 'Decision' }, f.project.id)).toMatchObject({ page: { complete: true } })
})
+15 -2
View File
@@ -21,6 +21,8 @@ type Fact = {
source_reference_ids: string[]
data: Row
}
/** Upper bound on projected objects per read (PERF-15). Normal scopes are far below it. */
export const STORY_GRAPH_MAX_FACTS = 50_000
const digest = (value: unknown): string => createHash('sha256').update(JSON.stringify(value)).digest('hex')
const cursorSchema = z.object({ key: z.string(), revision: z.string(), index: z.number().int().nonnegative(), offset: z.number().int().nonnegative() }).strict()
const normalizeScope = (scope: HeartbeatScope): HeartbeatScope => scope.kind === 'global' ? scope : { kind: 'projects', projectIds: [...new Set(scope.projectIds)].sort() }
@@ -48,7 +50,8 @@ export function queryTerms(query: string): string[] {
}
// This is a projection of existing storage, not a second graph or a historical state store.
export function readStoryGraph(db: DatabaseSync, name: StoryGraphToolName, input: unknown, projectId?: string, signal?: AbortSignal): Record<string, unknown> {
export function readStoryGraph(db: DatabaseSync, name: StoryGraphToolName, input: unknown, projectId?: string, signal?: AbortSignal,
maxFacts = STORY_GRAPH_MAX_FACTS): Record<string, unknown> {
signal?.throwIfAborted()
if (input && typeof input === 'object' && ('as_of' in input || ('mode' in input && input.mode === 'as_of'))) throw new Error('unsupported_mode: as_of is not supported')
const args = name === 'story_graph_search' ? storyGraphSearchSchema.parse(input)
@@ -139,8 +142,18 @@ export function readStoryGraph(db: DatabaseSync, name: StoryGraphToolName, input
}
const facts: Fact[] = []
const results = new Map<string, Row>()
const stories = db.prepare('SELECT * FROM story_lines ORDER BY id').all().filter(row => sameScope(row.scope_json))
// SQL prefilter on the scope kind and project membership (a superset of exact
// matches); the exact normalized comparison still runs in JS.
const stories = db.prepare(`SELECT * FROM story_lines
WHERE json_extract(scope_json, '$.kind') = ?
AND (? IS NULL OR NOT EXISTS (SELECT 1 FROM json_each(?) requested
WHERE requested.value NOT IN (SELECT value FROM json_each(story_lines.scope_json, '$.projectIds'))))
ORDER BY id`).all(scope.kind, scope.kind === 'projects' ? 1 : null, scope.kind === 'projects' ? JSON.stringify(scope.projectIds) : '[]')
.filter(row => sameScope(row.scope_json))
const add = (type: string, data: Row, result?: Row, sources?: string[]): void => {
// Bound the in-memory projection; a scope this large must be narrowed rather than truncated,
// since a partial projection would yield misleading revisions and cursors.
if (facts.length >= maxFacts) throw new Error(`scan_limit_exceeded: more than ${maxFacts} stored objects in scope; narrow the scope`)
facts.push({ object_ref: { type, id: String(data.id) }, object_revision: digest(data), result_id: result ? String(result.id) : null, scope,
event_time: type === 'event' ? timestamp(data.occurred_at) : null,
review_generated_at: result ? timestamp(result.created_at) : null,
+5 -2
View File
@@ -933,11 +933,13 @@ if (hasSingleInstanceLock) {
onExtensionStartupFailures: (extensionIds) =>
runtimeExtensionStore.markStartupFailed(extensionIds)
})
const readonlyQueryWorkerPath = join(app.getAppPath(), 'out/main/readonly-query-worker.js')
const startupKnowledgeService = new KnowledgeService({
documentResults: documentParsingService.results,
credentialCipher: secureCipher,
databasePath: join(app.getPath('userData'), 'knowledge.sqlite'),
managedRoot: join(app.getPath('userData'), 'knowledge'),
readonlyQueryWorkerPath,
extractStructured: createModelGraphExtractor(
settingsStore,
fetch,
@@ -1037,7 +1039,7 @@ if (hasSingleInstanceLock) {
storyGraphService: {
available: async ({ runtimeTarget }) => (await applicationSettingsStore.get()).heartbeatEnabled === true &&
(await capabilityService.getEnabledBuiltinMcpServerIds(runtimeTarget)).includes('story-graph'),
read: (name, input, projectId, signal) => startupAssistantDatabase.readStoryGraph(name, input, projectId, signal)
read: (name, input, projectId, signal) => startupAssistantDatabase.readStoryGraphAsync(name, input, projectId, signal)
},
magicNotesDatabase: startupAssistantDatabase,
configService: goodbuddyConfigService,
@@ -1246,6 +1248,7 @@ if (hasSingleInstanceLock) {
createConfiguredRuntime(initialResolvedSettings),
initializeAssistant: () => {
startupAssistantDatabase.initialize(defaultWorkspace)
startupAssistantDatabase.enableReadonlyWorker(readonlyQueryWorkerPath)
imageGenerationService!.initialize()
startupAssistantDatabase.ensureChannelProjects(
defaultWorkspace,
@@ -1504,7 +1507,7 @@ if (hasSingleInstanceLock) {
storyGraphService: {
available: async ({ runtimeTarget }) => (await applicationSettingsStore.get()).heartbeatEnabled === true &&
(await capabilityService.getEnabledBuiltinMcpServerIds(runtimeTarget)).includes('story-graph'),
read: (name, input, projectId, signal) => startupAssistantDatabase.readStoryGraph(name, input, projectId, signal)
read: (name, input, projectId, signal) => startupAssistantDatabase.readStoryGraphAsync(name, input, projectId, signal)
},
magicNotesDatabase: startupAssistantDatabase, configService: goodbuddyConfigService,
obsidianService, launchEnvironmentProvider: startupLocalToolEnvironmentService.launchEnvironmentProvider
+1 -1
View File
@@ -6343,7 +6343,7 @@ export function registerIpcHandlers(
registerHandler(ipcChannels.conversationsSearch, (event, input: unknown) => {
assertTrustedSender(event, window)
const { query } = conversationSearchRequestSchema.parse(input)
return assistantDatabase.searchConversations(query)
return assistantDatabase.searchConversationsAsync(query)
})
registerHandler(
+52
View File
@@ -47,6 +47,7 @@ import {
knowledgeRetrievalTerms
} from './retrieval-text'
import { normalizeEntityAlias } from './graph-extractor'
import { ReadonlyQueryReader, readWithFallback } from '../readonly-query-reader'
import type {
Chunk,
ChunkEmbeddingInput,
@@ -742,6 +743,7 @@ function mapKnowledgeTask(row: Row): KnowledgeTaskItem {
export class KnowledgeDatabase {
externalStore!: ExternalKnowledgeStore
private database?: DatabaseSync
private readonlyReader?: ReadonlyQueryReader
private readonly taskPrunedLibraries = new Set<string>()
private readonly taskPrunePending = new Set<string>()
@@ -779,7 +781,57 @@ export class KnowledgeDatabase {
}
}
/**
* Opens a read-only connection for the readonly query worker. No migration,
* no task recovery: the Main connection owns the schema and all writes.
*/
openReadOnly(): void {
if (this.database) {
throw new Error('Knowledge database is already open')
}
const database = new DatabaseSync(this.databasePath, {
readOnly: true,
timeout: 5_000
})
this.database = database
this.externalStore = new ExternalKnowledgeStore(database)
}
/**
* Routes the async search methods through a read-only worker thread. In-memory
* databases cannot be shared with a worker and keep the synchronous path.
*/
enableReadonlyWorker(workerPath: string): void {
if (this.databasePath === ':memory:' || this.readonlyReader) return
this.readonlyReader = new ReadonlyQueryReader('knowledge', this.databasePath, workerPath)
}
/** Test hook for the worker crash and cancellation paths. */
get readonlyWorkerForTest(): ReadonlyQueryReader | undefined {
return this.readonlyReader
}
searchAsync(options: SearchOptions, signal?: AbortSignal): Promise<SearchResult[]> {
this.requireDatabase()
return readWithFallback(this.readonlyReader, 'search', [options], signal,
() => this.search(options))
}
hybridSearchAsync(options: HybridSearchOptions): Promise<HybridSearchResult[]> {
return this.hybridSearchWithDiagnosticsAsync(options).then(page => page.results)
}
hybridSearchWithDiagnosticsAsync(options: HybridSearchOptions): Promise<HybridSearchResultPage> {
this.requireDatabase()
// AbortSignal is not cloneable; the reader forwards it as a shared cancel flag.
const { signal, ...cloneable } = options
return readWithFallback(this.readonlyReader, 'hybridSearchWithDiagnostics', [cloneable], signal,
() => this.hybridSearchWithDiagnostics(options))
}
close(): void {
this.readonlyReader?.close()
this.readonlyReader = undefined
if (!this.database) {
return
}
+9 -2
View File
@@ -159,6 +159,8 @@ export type KnowledgeServiceOptions = {
embeddingProvider?: EmbeddingProvider
rerankProvider?: RerankProvider
embeddingBatchSize?: number
/** Read-only query worker bundle; searches run off the Main thread when set. */
readonlyQueryWorkerPath?: string
parseDocument?: (
name: string,
buffer: Buffer,
@@ -263,9 +265,11 @@ export class KnowledgeService {
EmbeddingIndexCoordinator
>()
private readonly lifecycleController = new AbortController()
private readonly readonlyQueryWorkerPath?: string
constructor(options: KnowledgeServiceOptions) {
this.database = new KnowledgeDatabase(options.databasePath)
this.readonlyQueryWorkerPath = options.readonlyQueryWorkerPath
this.external = new ExternalKnowledgeService(this.database, options.credentialCipher, options.externalFetcher)
this.managedRoot = resolve(options.managedRoot)
this.documentResults = options.documentResults
@@ -297,6 +301,9 @@ export class KnowledgeService {
async initialize(): Promise<void> {
await mkdir(this.managedRoot, { recursive: true })
this.database.initialize()
if (this.readonlyQueryWorkerPath) {
this.database.enableReadonlyWorker(this.readonlyQueryWorkerPath)
}
await this.reconcileDocumentResults()
for (const library of this.database.listKnowledgeBases()) {
for (const source of this.database.listSourcesForSnapshot(library.id)) {
@@ -1352,7 +1359,7 @@ export class KnowledgeService {
100,
settings.topK * settings.candidateMultiplier
)
const searchPage = this.database.hybridSearchWithDiagnostics({
const searchPage = await this.database.hybridSearchWithDiagnosticsAsync({
knowledgeBaseId: library.id,
query: input.query,
limit: candidateLimit,
@@ -1892,7 +1899,7 @@ export class KnowledgeService {
const library = this.requireLibrary(knowledgeBaseId)
const settings = library.retrievalSettings
const vector = await this.embedQuery(query, signal)
return this.database.hybridSearch({
return this.database.hybridSearchAsync({
knowledgeBaseId,
query,
limit: Math.min(limit, settings.topK),
+241
View File
@@ -0,0 +1,241 @@
// @vitest-environment node
import { createHash, randomUUID } from 'node:crypto'
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join, resolve } from 'node:path'
import { build } from 'esbuild'
import { afterAll, beforeAll, describe, expect, it } from 'vitest'
import { AssistantDatabase } from './assistant/assistant-database'
import { KnowledgeDatabase } from './knowledge/knowledge-database'
import { ReadonlyQueryReader, ReadonlyWorkerUnavailableError, readWithFallback } from './readonly-query-reader'
let directory: string
let workerPath: string
beforeAll(async () => {
directory = await mkdtemp(join(tmpdir(), 'goodbuddy-readonly-reader-'))
workerPath = join(directory, 'readonly-query-worker.cjs')
await build({ entryPoints: [resolve('src/main/readonly-query-worker.ts')], outfile: workerPath,
bundle: true, platform: 'node', format: 'cjs', logLevel: 'silent' })
}, 60_000)
afterAll(async () => { await rm(directory, { recursive: true, force: true, maxRetries: 10, retryDelay: 100 }) })
const checksum = (value: string): string => createHash('sha256').update(value).digest('hex')
function seedKnowledge(path: string): { database: KnowledgeDatabase; libraryId: string; query: number[] } {
const database = new KnowledgeDatabase(path)
database.initialize()
const library = database.createKnowledgeBase({ name: 'Worker parity', storageMode: 'reference' })
let seed = 7
const random = (): number => ((seed = (seed * 1664525 + 1013904223) >>> 0) / 4294967296)
const words = 'lighthouse harbor budget latency 发布 审批 流程 跨平台 桌面 vector'.split(' ')
const sqlite = (database as unknown as { database: import('node:sqlite').DatabaseSync }).database
for (let documentIndex = 0; documentIndex < 6; documentIndex += 1) {
const source = database.upsertSource({ knowledgeBaseId: library.id, type: 'file',
location: `C:\\parity\\doc-${documentIndex}.md`, displayName: `doc-${documentIndex}.md`, status: 'ready' })
const chunks = Array.from({ length: 30 }, (_, ordinal) => ({ id: `d${documentIndex}-c${ordinal}`, ordinal,
content: Array.from({ length: 24 }, () => words[Math.floor(random() * words.length)]).join(' '), location: `line ${ordinal}` }))
const document = database.upsertDocument({ knowledgeBaseId: library.id, sourceId: source.id,
externalId: `doc-${documentIndex}`, title: `doc-${documentIndex}`, sourceLocation: source.location }, chunks)
const stored = sqlite.prepare("SELECT id, index_content FROM chunks WHERE document_id = ? AND enabled = 1 AND role <> 'parent'")
.all(document.id) as Array<{ id: string; index_content: string }>
database.replaceDocumentEmbeddings(document.id, 'p', 'm', stored.map(row => ({ chunkId: row.id,
// Duplicate vectors exercise the tie-break on chunk id.
contentChecksum: checksum(row.index_content), vector: row.id.endsWith('-c3') ? [1, 0, 0, 0] : [random() - 0.5, random() - 0.5, random() - 0.5, random() - 0.5] })))
}
return { database, libraryId: library.id, query: [0.9, 0.1, -0.2, 0.05] }
}
describe('knowledge search worker', () => {
it('returns exactly the synchronous results, order and scores', async () => {
const { database, libraryId, query } = seedKnowledge(join(directory, `knowledge-${randomUUID()}.sqlite`))
try {
const cases = [
{ knowledgeBaseId: libraryId, query: 'lighthouse budget', limit: 40, provider: 'p', model: 'm', vector: query, graphEnabled: false, candidateMultiplier: 1 },
{ knowledgeBaseId: libraryId, query: '发布审批流程', limit: 20, provider: 'p', model: 'm', vector: query, minimumVectorSimilarity: 0.1 },
{ knowledgeBaseId: libraryId, query: 'harbor', limit: 5, ftsWeight: 0 , provider: 'p', model: 'm', vector: query },
{ knowledgeBaseId: libraryId, query: 'latency', limit: 10 }
]
const expected = cases.map(options => database.hybridSearchWithDiagnostics(options))
const expectedFts = database.search({ knowledgeBaseId: libraryId, query: '跨平台 vector', limit: 30 })
database.enableReadonlyWorker(workerPath)
const actual = await Promise.all(cases.map(options => database.hybridSearchWithDiagnosticsAsync(options)))
expect(database.readonlyWorkerForTest?.pendingCount).toBe(0)
expect(actual).toEqual(expected)
expect(actual[0]!.results.length).toBeGreaterThan(10)
expect(actual[0]!.vectorScannedCount).toBe(180)
expect(await database.searchAsync({ knowledgeBaseId: libraryId, query: '跨平台 vector', limit: 30 })).toEqual(expectedFts)
// Validation errors keep their type and message across the thread boundary.
await expect(database.hybridSearchWithDiagnosticsAsync({ ...cases[0]!, limit: 500 })).rejects.toThrow(RangeError)
// A write committed on Main is visible to the next worker read.
database.updateKnowledgeBase(libraryId, { name: 'Renamed' })
const source = database.upsertSource({ knowledgeBaseId: libraryId, type: 'file', location: 'C:\\parity\\new.md', displayName: 'new.md', status: 'ready' })
database.upsertDocument({ knowledgeBaseId: libraryId, sourceId: source.id, externalId: 'new', title: 'new', sourceLocation: source.location },
[{ id: 'fresh-chunk', ordinal: 0, content: 'zebracorn unique', location: 'l' }])
expect((await database.searchAsync({ knowledgeBaseId: libraryId, query: 'zebracorn' })).map(item => item.chunk.id)).toEqual(['fresh-chunk'])
} finally { database.close() }
}, 60_000)
it('cancels through the AbortSignal and keeps serving later requests', async () => {
const { database, libraryId, query } = seedKnowledge(join(directory, `knowledge-${randomUUID()}.sqlite`))
try {
database.enableReadonlyWorker(workerPath)
const options = { knowledgeBaseId: libraryId, query: 'harbor', provider: 'p', model: 'm', vector: query }
const preAborted = new AbortController()
preAborted.abort(new Error('cancelled before'))
await expect(database.hybridSearchWithDiagnosticsAsync({ ...options, signal: preAborted.signal })).rejects.toThrow('cancelled before')
const controller = new AbortController()
const pending = database.hybridSearchWithDiagnosticsAsync({ ...options, signal: controller.signal })
controller.abort(new Error('cancelled during'))
await expect(pending).rejects.toThrow('cancelled during')
expect(database.readonlyWorkerForTest?.pendingCount).toBe(0)
expect(await database.hybridSearchWithDiagnosticsAsync(options)).toEqual(database.hybridSearchWithDiagnostics(options))
} finally { database.close() }
}, 60_000)
it('falls back to the synchronous path when the worker crashes and restarts later', async () => {
const { database, libraryId, query } = seedKnowledge(join(directory, `knowledge-${randomUUID()}.sqlite`))
try {
database.enableReadonlyWorker(workerPath)
const reader = database.readonlyWorkerForTest!
const options = { knowledgeBaseId: libraryId, query: 'budget', provider: 'p', model: 'm', vector: query }
const expected = database.hybridSearchWithDiagnostics(options)
expect(await database.hybridSearchWithDiagnosticsAsync(options)).toEqual(expected)
// In-flight request when the worker dies: rejected internally, served synchronously.
const inFlight = database.hybridSearchWithDiagnosticsAsync(options)
await reader.terminateWorkerForTest()
expect(await inFlight).toEqual(expected)
expect(reader.available).toBe(false)
// During the backoff the synchronous path serves requests.
expect(await database.hybridSearchWithDiagnosticsAsync(options)).toEqual(expected)
reader.resetBackoffForTest()
expect(reader.available).toBe(true)
expect(await reader.call('hybridSearchWithDiagnostics', [options])).toEqual(expected)
} finally { database.close() }
}, 60_000)
it('keeps the synchronous path for in-memory databases and startup failures', async () => {
const memory = new KnowledgeDatabase(':memory:')
memory.initialize()
try {
memory.enableReadonlyWorker(workerPath)
expect(memory.readonlyWorkerForTest).toBeUndefined()
const library = memory.createKnowledgeBase({ name: 'Memory', storageMode: 'reference' })
expect(await memory.searchAsync({ knowledgeBaseId: library.id, query: 'x' })).toEqual([])
} finally { memory.close() }
const missing = new ReadonlyQueryReader('knowledge', join(directory, 'missing.sqlite'), workerPath)
try {
await expect(missing.call('search', [{}])).rejects.toBeInstanceOf(ReadonlyWorkerUnavailableError)
expect(await readWithFallback(missing, 'search', [{}], undefined, () => 'sync')).toBe('sync')
} finally { missing.close() }
}, 60_000)
})
/** The previous per-conversation JS implementation, kept as the semantic reference. */
function referenceSearch(database: AssistantDatabase, query: string): string[] {
const normalized = query.trim().toLocaleLowerCase()
if (!normalized) return []
const sqlite = (database as unknown as { database: import('node:sqlite').DatabaseSync }).database
const rows = sqlite.prepare(`SELECT id, title FROM conversations WHERE status = 'active' ORDER BY pinned DESC, updated_at DESC`).all() as Array<{ id: string; title: string }>
const messages = sqlite.prepare('SELECT content FROM messages WHERE conversation_id = ?')
return rows.filter(row => row.title.toLocaleLowerCase().includes(normalized) ||
(messages.all(row.id) as Array<{ content: string }>).some(message => message.content.toLocaleLowerCase().includes(normalized)))
.map(row => row.id)
}
describe('assistant readonly worker', () => {
it('matches the reference conversation search, including Unicode case folding and CJK, on Main and in the worker', async () => {
const root = await mkdtemp(join(directory, 'assistant-'))
const database = new AssistantDatabase(join(root, 'assistant.sqlite'))
database.initialize(root)
try {
const texts = ['ÄPFEL und Birnen', 'straße STRASSE', 'İstanbul ısı', 'ΣΊΣΥΦΟΣ', '跨平台桌面应用发布审批', '100%_literal\\x', 'DŽemal Dž', 'plain body', '']
const conversations = texts.map((text, index) => ({
id: randomUUID(), title: index % 3 === 0 ? `Title ${text}` : `Conversation ${index}`, updatedAt: index + 1, pinned: index === 4,
messages: [{ id: randomUUID(), role: 'user' as const, state: 'complete' as const, content: text, createdAt: index * 10 },
{ id: randomUUID(), role: 'assistant' as const, state: 'complete' as const, content: `reply ${index}`, createdAt: index * 10 + 1 }]
}))
database.replaceConversations(conversations)
const queries = ['äpfel', 'ÄPFEL', 'Äp', 'STRASSE', 'straße', 'İstanbul', 'i̇stanbul', 'ISI', 'σίσυφος', 'ΣΊΣΥΦΟΣ', '平台桌面', '审批',
'%_', '100%', '_', '\\x', 'dž', 'DŽemal', 'reply 1', 'REPLY', 'title', 'conversation 7', ' plain ', 'missing', ' ']
const expected = queries.map(query => referenceSearch(database, query))
expect(expected.some(ids => ids.length > 1)).toBe(true)
expect(queries.map(query => database.searchConversations(query))).toEqual(expected)
database.enableReadonlyWorker(workerPath)
expect(await Promise.all(queries.map(query => database.searchConversationsAsync(query)))).toEqual(expected)
expect(database.readonlyWorkerForTest?.pendingCount).toBe(0)
} finally { database.close() }
}, 60_000)
it('lists conversations with batched queries matching per-conversation reads', async () => {
const root = await mkdtemp(join(directory, 'assistant-'))
const database = new AssistantDatabase(join(root, 'assistant.sqlite'))
database.initialize(root)
try {
const message = (content: string, createdAt: number, state: 'complete' | 'streaming' = 'complete', role: 'user' | 'assistant' = 'assistant') =>
({ id: randomUUID(), role, state, content, createdAt })
const conversations = [
{ id: randomUUID(), title: 'Empty', updatedAt: 1, messages: [] },
{ id: randomUUID(), title: 'Streaming', updatedAt: 2, messages: [message('q', 1, 'complete', 'user'), message('partial', 2, 'streaming')] },
{ id: randomUUID(), title: 'Done', updatedAt: 3, messages: [message('q', 5, 'complete', 'user'), message('a', 9), message('b', 7)] },
{ id: randomUUID(), title: 'Single', updatedAt: 0, messages: [message('x', 3)] }
]
database.replaceConversations(conversations)
const full = database.listConversations()
expect(full).toHaveLength(4)
for (const item of full) expect(item).toEqual(database.getConversation(item.id))
const summaries = database.listConversationSummaries([conversations[2]!.id])
const byId = new Map(summaries.map(item => [item.id, item]))
expect(byId.get(conversations[2]!.id)).toEqual(database.getConversation(conversations[2]!.id))
// A streaming conversation always carries its messages.
expect(byId.get(conversations[1]!.id)).toEqual(database.getConversation(conversations[1]!.id))
expect(byId.get(conversations[0]!.id)?.messageSummary).toEqual({ count: 0, firstRole: undefined, latestMessageAt: undefined })
expect(byId.get(conversations[3]!.id)?.messageSummary).toEqual({ count: 1, firstRole: 'assistant', latestMessageAt: 3 })
expect(byId.get(conversations[3]!.id)?.messages).toEqual([])
} finally { database.close() }
}, 60_000)
it('serves story graph reads from the worker with the synchronous result and errors', async () => {
const root = await mkdtemp(join(directory, 'assistant-'))
const database = new AssistantDatabase(join(root, 'assistant.sqlite'))
database.initialize(root)
try {
const project = database.listProjects()[0]!
const other = database.createProject({ name: 'Other', description: '', rootPath: root, defaultWorkMode: 'ask' })
const conversationId = randomUUID(), messageId = randomUUID()
const body = 'Choose A, then revise A to B because of measured latency.'
database.saveLocalConversations([{ header: { id: conversationId, projectId: project.id, title: 'Decision', updatedAt: 1 },
messages: [{ id: messageId, role: 'user', content: body, createdAt: Date.parse('2026-09-01T00:00:00Z'), state: 'complete' }] }])
const request = { trigger: 'manual' as const, scope: { kind: 'projects' as const, projectIds: [project.id] },
timeRange: { from: '2026-09-01T00:00:00Z', to: '2026-09-30T00:00:00Z' } }
database.saveSupervisionResult({ request, candidates: database.listSupervisionCandidates(request),
evidence: [{ id: 'quote', sourceType: 'conversation', sourceId: conversationId, title: 'Decision evidence', content: body,
occurredAt: '2026-09-01T00:00:00Z', locator: { source: `message:${messageId}`, projectId: project.id, conversationId, messageId, start: 0, end: body.length, length: body.length } }],
output: { summary: 'Decision A', changeDigest: '', openItems: [], relations: [],
entities: [{ id: 'choice', label: 'Decision', description: 'Decision A', sourceReferenceIds: ['quote'] }],
events: [{ title: 'Decision A', description: 'Decision A', occurredAt: '2026-09-01T00:00:00Z', eventType: 'decision', entityIds: ['choice'], sourceReferenceIds: ['quote'] }],
entityChanges: [{ entityId: 'choice', changeType: 'proposed', description: 'Decision A', sourceReferenceIds: ['quote'] }] } })
const withoutReadAt = (value: Record<string, unknown>): Record<string, unknown> => ({ ...value, read_at: undefined })
const inputs = [
{ query: 'Decision', page_size: 50 },
{ query: 'decision latency', page_size: 2 },
{ query: 'Decision', page_size: 50, scope: { kind: 'global' } }
]
const expected = inputs.map(input => withoutReadAt(database.readStoryGraph('story_graph_search', input, project.id)))
expect((expected[0]!.items as unknown[]).length).toBeGreaterThan(2)
database.enableReadonlyWorker(workerPath)
const actual = await Promise.all(inputs.map(input => database.readStoryGraphAsync('story_graph_search', input, project.id)))
expect(actual.map(withoutReadAt)).toEqual(expected)
const entity = (expected[0]!.items as Array<{ object_ref: { type: string; id: string } }>).find(item => item.object_ref.type === 'entity')!
const contextInput = { object_ref: entity.object_ref, mode: 'timeline', page_size: 5 }
expect(withoutReadAt(await database.readStoryGraphAsync('story_graph_get_context', contextInput, project.id)))
.toEqual(withoutReadAt(database.readStoryGraph('story_graph_get_context', contextInput, project.id)))
await expect(database.readStoryGraphAsync('story_graph_get_context', contextInput, other.id)).rejects.toThrow('scope_mismatch')
await expect(database.readStoryGraphAsync('story_graph_search', { query: 'x' })).rejects.toThrow('scope_required')
// Schema validation errors stay ZodErrors so MCP callers classify them unchanged.
await expect(database.readStoryGraphAsync('story_graph_search', { query: 1 }, project.id)).rejects.toMatchObject({ name: 'ZodError' })
const aborted = new AbortController(); aborted.abort(new Error('stop'))
await expect(database.readStoryGraphAsync('story_graph_search', inputs[0], project.id, aborted.signal)).rejects.toThrow('stop')
expect(database.readonlyWorkerForTest?.pendingCount).toBe(0)
} finally { database.close() }
}, 60_000)
})
+204
View File
@@ -0,0 +1,204 @@
import { Worker } from 'node:worker_threads'
import { z } from 'zod'
// PERF-15/16 first slice: hot read paths run in a read-only worker per database
// file so a long scan does not block the Main event loop. The worker opens its
// own read-only connection (WAL), so it observes committed data only.
export type ReadonlyQueryKind = 'assistant' | 'knowledge'
/** Worker infrastructure failure; callers fall back to the synchronous path. */
export class ReadonlyWorkerUnavailableError extends Error {
constructor(message: string, options?: { cause?: unknown }) {
super(message, options)
this.name = 'ReadonlyWorkerUnavailableError'
}
}
export type ReadonlyQueryRequest = {
type: 'query'
id: number
op: string
args: unknown[]
/** Int32Array(1) cancel flag, polled by the worker through `signal.throwIfAborted()`. */
cancel: SharedArrayBuffer
}
export type ReadonlyQueryResponse =
| { id: number; result: unknown }
| { id: number; error: { name: string; message: string; issues?: unknown } }
| { id: number; ready: true }
type Pending = {
resolve: (value: unknown) => void
reject: (error: Error) => void
cleanup: () => void
}
const MAX_PENDING = 32
const RESTART_BACKOFF_MS = 30_000
export function deserializeWorkerError(error: { name: string; message: string; issues?: unknown }): Error {
if (error.name === 'ZodError' && Array.isArray(error.issues)) {
return new z.ZodError(error.issues as z.core.$ZodIssue[])
}
const Constructor = error.name === 'RangeError' ? RangeError
: error.name === 'TypeError' ? TypeError : Error
const result = new Constructor(error.message)
if (result.name !== error.name) result.name = error.name
return result
}
export class ReadonlyQueryReader {
private worker?: Worker
private nextId = 0
private disabledUntil = 0
private closed = false
private readonly pending = new Map<number, Pending>()
constructor(
private readonly kind: ReadonlyQueryKind,
private readonly databasePath: string,
private readonly workerPath: string,
private readonly now: () => number = Date.now
) {}
/** False while backing off after a crash or startup failure. */
get available(): boolean {
return !this.closed && this.now() >= this.disabledUntil && this.pending.size < MAX_PENDING
}
get pendingCount(): number {
return this.pending.size
}
call<T>(op: string, args: unknown[], signal?: AbortSignal): Promise<T> {
if (this.closed) return Promise.reject(new ReadonlyWorkerUnavailableError('Readonly query reader closed'))
if (signal?.aborted) return Promise.reject(signal.reason as Error)
if (this.now() < this.disabledUntil) {
return Promise.reject(new ReadonlyWorkerUnavailableError('Readonly query worker is backing off'))
}
// Bound outstanding work; the caller runs the query synchronously instead.
if (this.pending.size >= MAX_PENDING) {
return Promise.reject(new ReadonlyWorkerUnavailableError('Readonly query worker is busy'))
}
let worker: Worker
try {
worker = this.ensureWorker()
} catch (error) {
this.disabledUntil = this.now() + RESTART_BACKOFF_MS
return Promise.reject(new ReadonlyWorkerUnavailableError('Readonly query worker failed to start', { cause: error }))
}
const id = ++this.nextId
const cancel = new SharedArrayBuffer(4)
const flag = new Int32Array(cancel)
return new Promise<T>((resolve, reject) => {
const onAbort = (): void => {
Atomics.store(flag, 0, 1)
const request = this.pending.get(id)
if (!request) return
this.pending.delete(id)
request.cleanup()
if (this.pending.size === 0) worker.unref()
reject(signal!.reason as Error)
}
signal?.addEventListener('abort', onAbort, { once: true })
this.pending.set(id, {
resolve: resolve as (value: unknown) => void,
reject,
cleanup: () => signal?.removeEventListener('abort', onAbort)
})
// Keep the process alive only while a request is outstanding.
worker.ref()
try {
worker.postMessage({ type: 'query', id, op, args, cancel } satisfies ReadonlyQueryRequest)
} catch (error) {
this.pending.delete(id)
signal?.removeEventListener('abort', onAbort)
if (this.pending.size === 0) worker.unref()
// DataCloneError etc. are caller errors, but stay safe and let the caller fall back.
reject(new ReadonlyWorkerUnavailableError('Readonly query could not be posted', { cause: error }))
}
})
}
private ensureWorker(): Worker {
if (this.worker) return this.worker
const worker = new Worker(this.workerPath, {
workerData: { kind: this.kind, databasePath: this.databasePath }
})
this.worker = worker
worker.on('message', (message: ReadonlyQueryResponse) => {
if ('ready' in message) return
const request = this.pending.get(message.id)
if (!request) return
this.pending.delete(message.id)
request.cleanup()
if (this.pending.size === 0) worker.unref()
if ('error' in message) request.reject(deserializeWorkerError(message.error))
else request.resolve(message.result)
})
const failed = (error: Error): void => {
if (this.worker !== worker) return
this.worker = undefined
// Restart lazily on a later request, after a short backoff.
this.disabledUntil = this.now() + RESTART_BACKOFF_MS
const pending = [...this.pending.values()]
this.pending.clear()
for (const request of pending) {
request.cleanup()
request.reject(new ReadonlyWorkerUnavailableError(error.message, { cause: error }))
}
void worker.terminate()
}
worker.once('error', failed)
worker.once('exit', (code) => failed(new Error(`Readonly query worker exited (${code})`)))
worker.unref()
return worker
}
/** Test hook: simulate a crash by terminating the worker. */
async terminateWorkerForTest(): Promise<void> {
await this.worker?.terminate()
}
/** Test hook: end the post-crash backoff. */
resetBackoffForTest(): void {
this.disabledUntil = 0
}
close(): void {
this.closed = true
const worker = this.worker
this.worker = undefined
const pending = [...this.pending.values()]
this.pending.clear()
for (const request of pending) {
request.cleanup()
request.reject(new ReadonlyWorkerUnavailableError('Readonly query reader closed'))
}
if (worker) void worker.terminate()
}
}
/**
* Runs `op` in the worker when available, otherwise synchronously. Worker
* infrastructure failures fall back to the synchronous path; query errors and
* aborts propagate unchanged.
*/
export async function readWithFallback<T>(
reader: ReadonlyQueryReader | undefined,
op: string,
args: unknown[],
signal: AbortSignal | undefined,
sync: () => T
): Promise<T> {
signal?.throwIfAborted()
if (!reader?.available) return sync()
try {
return await reader.call<T>(op, args, signal)
} catch (error) {
if (error instanceof ReadonlyWorkerUnavailableError && !signal?.aborted) return sync()
throw error
}
}
+68
View File
@@ -0,0 +1,68 @@
import { parentPort, workerData } from 'node:worker_threads'
import { z } from 'zod'
import { AssistantDatabase } from './assistant/assistant-database'
import { KnowledgeDatabase } from './knowledge/knowledge-database'
import type { ReadonlyQueryKind, ReadonlyQueryRequest } from './readonly-query-reader'
// Read-only query worker (PERF-15/16 first slice). One worker per database file.
// 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 }
type Handler = (args: unknown[], signal: AbortSignal) => unknown
let handlers: Record<string, Handler>
let close: () => void
if (kind === 'knowledge') {
const database = new KnowledgeDatabase(databasePath)
database.openReadOnly()
close = () => database.close()
handlers = {
search: ([options]) => database.search(options as Parameters<KnowledgeDatabase['search']>[0]),
hybridSearchWithDiagnostics: ([options], signal) => database.hybridSearchWithDiagnostics({
...(options as Parameters<KnowledgeDatabase['hybridSearchWithDiagnostics']>[0]), signal
})
}
} else {
const database = new AssistantDatabase(databasePath)
database.openReadOnly()
close = () => database.close()
handlers = {
readStoryGraph: ([name, input, projectId], signal) => database.readStoryGraph(
name as Parameters<AssistantDatabase['readStoryGraph']>[0], input, projectId as string | undefined, signal),
searchConversations: ([query]) => database.searchConversations(query as string)
}
}
/** A signal backed by the shared cancel flag; queries poll it via throwIfAborted(). */
function sharedSignal(cancel: SharedArrayBuffer): AbortSignal {
const flag = new Int32Array(cancel)
const reason = (): Error => Object.assign(new Error('This operation was aborted'), { name: 'AbortError' })
const signal = new AbortController().signal
Object.defineProperties(signal, {
aborted: { get: () => Atomics.load(flag, 0) === 1 },
reason: { get: () => (Atomics.load(flag, 0) === 1 ? reason() : undefined) },
throwIfAborted: { value: () => { if (Atomics.load(flag, 0) === 1) throw reason() } }
})
return signal
}
function serializeError(error: unknown): { name: string; message: string; issues?: unknown } {
if (error instanceof z.ZodError) return { name: 'ZodError', message: error.message, issues: error.issues }
if (error instanceof Error) return { name: error.name, message: error.message }
return { name: 'Error', message: String(error) }
}
parentPort!.on('message', (message: ReadonlyQueryRequest) => {
try {
const handler = Object.hasOwn(handlers, message.op) ? handlers[message.op] : undefined
if (!handler) throw new Error(`Unknown readonly query: ${message.op}`)
const signal = sharedSignal(message.cancel)
signal.throwIfAborted()
parentPort!.postMessage({ id: message.id, result: handler(message.args, signal) })
} catch (error) {
parentPort!.postMessage({ id: message.id, error: serializeError(error) })
}
})
parentPort!.on('close', () => close())