perf: parse, chunk and hash documents in a worker thread and add startup and document-parse benchmarks
This commit is contained in:
mesalogo
@@ -0,0 +1,375 @@
|
||||
import { createHash } from 'node:crypto'
|
||||
import { Worker } from 'node:worker_threads'
|
||||
import type { KnowledgeChunkingSettings } from '../shared/knowledge-contracts'
|
||||
import {
|
||||
assertDocumentBuffer,
|
||||
chunkDocumentAdvanced,
|
||||
DocumentTextUnavailableError,
|
||||
extractPdfTextPages,
|
||||
parseDocument,
|
||||
type DocumentChunk,
|
||||
type ParsedDocument,
|
||||
type PdfTextExtraction,
|
||||
type PdfTextExtractionOptions
|
||||
} from './knowledge/document-parser'
|
||||
import { extractPptxPages, type PptxPage } from './knowledge/pptx-parser'
|
||||
import type {
|
||||
DocumentParseOperation,
|
||||
DocumentParseRequest,
|
||||
DocumentParseResponse,
|
||||
SerializedDocumentParseError
|
||||
} from './document-parse-errors'
|
||||
|
||||
export type { DocumentParseOperation } from './document-parse-errors'
|
||||
|
||||
// PERF-15/16: pure document work (PDF / Office / text parsing, chunking,
|
||||
// hashing) runs in a single worker thread with a FIFO queue, so Main only
|
||||
// orchestrates. Database writes stay on the caller's side, in the same order.
|
||||
// Same fallback rules as readonly-query-reader: if the worker cannot start or
|
||||
// crashes, the call runs inline (the previous behaviour) and the worker is
|
||||
// retried after a backoff. Aborting a running task terminates the worker.
|
||||
|
||||
/** Worker infrastructure failure; callers fall back to the inline path. */
|
||||
export class DocumentParseWorkerUnavailableError extends Error {
|
||||
constructor(message: string, options?: { cause?: unknown }) {
|
||||
super(message, options)
|
||||
this.name = 'DocumentParseWorkerUnavailableError'
|
||||
}
|
||||
}
|
||||
|
||||
type Task = {
|
||||
id: number
|
||||
op: DocumentParseOperation
|
||||
args: unknown[]
|
||||
transfer: ArrayBuffer[]
|
||||
resolve: (value: unknown) => void
|
||||
reject: (error: unknown) => void
|
||||
cleanup: () => void
|
||||
}
|
||||
|
||||
const MAX_QUEUED = 32
|
||||
const RESTART_BACKOFF_MS = 30_000
|
||||
|
||||
type PdfExceptionConstructor = new (message: string) => Error
|
||||
type PdfExceptionClasses = Partial<Record<string, PdfExceptionConstructor>>
|
||||
const pdfExceptionNames = [
|
||||
'AbortException',
|
||||
'InvalidPDFException',
|
||||
'PasswordException',
|
||||
'ResponseException',
|
||||
'UnknownErrorException'
|
||||
] as const
|
||||
|
||||
let pdfExceptions: Promise<PdfExceptionClasses> | undefined
|
||||
|
||||
/** PDF.js exception classes, loaded only once a PDF error crosses the boundary. */
|
||||
function loadPdfExceptions(): Promise<PdfExceptionClasses> {
|
||||
pdfExceptions ??= import('pdfjs-dist/legacy/build/pdf.mjs').then(
|
||||
(pdfjs) => {
|
||||
const module = pdfjs as unknown as Record<string, unknown>
|
||||
return Object.fromEntries(pdfExceptionNames.flatMap((name) =>
|
||||
typeof module[name] === 'function' ? [[name, module[name] as PdfExceptionConstructor]] : []))
|
||||
},
|
||||
() => ({})
|
||||
)
|
||||
return pdfExceptions
|
||||
}
|
||||
|
||||
/**
|
||||
* Rebuilds an error with the same class, name, message, cause and primitive
|
||||
* fields as the one thrown inline (stack traces differ by nature).
|
||||
*/
|
||||
export async function deserializeDocumentParseError(error: SerializedDocumentParseError): Promise<Error> {
|
||||
let result: Error
|
||||
if (error.name === 'DocumentTextUnavailableError') {
|
||||
result = new DocumentTextUnavailableError(error.message)
|
||||
} else if ((pdfExceptionNames as readonly string[]).includes(error.name)) {
|
||||
const Constructor = (await loadPdfExceptions())[error.name]
|
||||
result = Constructor ? new Constructor(error.message) : new Error(error.message)
|
||||
} else {
|
||||
const Constructor = error.name === 'RangeError' ? RangeError
|
||||
: error.name === 'TypeError' ? TypeError : Error
|
||||
result = new Constructor(error.message)
|
||||
}
|
||||
if (result.name !== error.name) result.name = error.name
|
||||
if (error.cause) {
|
||||
Object.defineProperty(result, 'cause', {
|
||||
value: await deserializeDocumentParseError(error.cause), writable: true, configurable: true
|
||||
})
|
||||
}
|
||||
for (const [key, value] of Object.entries(error.extra ?? {})) {
|
||||
Object.defineProperty(result, key, { value, writable: true, configurable: true, enumerable: true })
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
export class DocumentParseWorkerClient {
|
||||
private worker?: Worker
|
||||
private nextId = 0
|
||||
private disabledUntil = 0
|
||||
private closed = false
|
||||
private active?: Task
|
||||
private readonly queue: Task[] = []
|
||||
|
||||
constructor(
|
||||
private readonly workerPath: string,
|
||||
private readonly now: () => number = Date.now
|
||||
) {}
|
||||
|
||||
/** False while closed, backing off after a crash, or when the queue is full. */
|
||||
get available(): boolean {
|
||||
return !this.closed && this.now() >= this.disabledUntil &&
|
||||
this.queue.length < MAX_QUEUED
|
||||
}
|
||||
|
||||
get pendingCount(): number {
|
||||
return this.queue.length + (this.active ? 1 : 0)
|
||||
}
|
||||
|
||||
call<T>(
|
||||
op: DocumentParseOperation,
|
||||
args: unknown[],
|
||||
transfer: ArrayBuffer[] = [],
|
||||
signal?: AbortSignal
|
||||
): Promise<T> {
|
||||
if (this.closed) return Promise.reject(new DocumentParseWorkerUnavailableError('Document parse worker closed'))
|
||||
if (signal?.aborted) return Promise.reject(signal.reason)
|
||||
if (this.now() < this.disabledUntil) {
|
||||
return Promise.reject(new DocumentParseWorkerUnavailableError('Document parse worker is backing off'))
|
||||
}
|
||||
if (this.queue.length >= MAX_QUEUED) {
|
||||
return Promise.reject(new DocumentParseWorkerUnavailableError('Document parse worker is busy'))
|
||||
}
|
||||
return new Promise<T>((resolve, reject) => {
|
||||
const task: Task = {
|
||||
id: ++this.nextId,
|
||||
op,
|
||||
args,
|
||||
transfer,
|
||||
resolve: resolve as (value: unknown) => void,
|
||||
reject,
|
||||
cleanup: () => signal?.removeEventListener('abort', onAbort)
|
||||
}
|
||||
const onAbort = (): void => {
|
||||
if (this.active === task) {
|
||||
// The task is CPU-bound in the worker; stop it by terminating the worker.
|
||||
this.active = undefined
|
||||
const worker = this.worker
|
||||
this.worker = undefined
|
||||
if (worker) void worker.terminate()
|
||||
} else {
|
||||
const index = this.queue.indexOf(task)
|
||||
if (index < 0) return
|
||||
this.queue.splice(index, 1)
|
||||
}
|
||||
task.cleanup()
|
||||
reject(signal!.reason)
|
||||
this.pump()
|
||||
}
|
||||
signal?.addEventListener('abort', onAbort, { once: true })
|
||||
this.queue.push(task)
|
||||
this.pump()
|
||||
})
|
||||
}
|
||||
|
||||
private pump(): void {
|
||||
if (this.active || this.closed) return
|
||||
const task = this.queue.shift()
|
||||
if (!task) {
|
||||
this.worker?.unref()
|
||||
return
|
||||
}
|
||||
let worker: Worker
|
||||
try {
|
||||
worker = this.ensureWorker()
|
||||
} catch (error) {
|
||||
this.disabledUntil = this.now() + RESTART_BACKOFF_MS
|
||||
this.failAll(task, new DocumentParseWorkerUnavailableError('Document parse worker failed to start', { cause: error }))
|
||||
return
|
||||
}
|
||||
this.active = task
|
||||
// Keep the process alive only while a task is outstanding.
|
||||
worker.ref()
|
||||
try {
|
||||
worker.postMessage(
|
||||
{ id: task.id, op: task.op, args: task.args } satisfies DocumentParseRequest,
|
||||
task.transfer
|
||||
)
|
||||
} catch (error) {
|
||||
this.active = undefined
|
||||
task.cleanup()
|
||||
task.reject(new DocumentParseWorkerUnavailableError('Document parse task could not be posted', { cause: error }))
|
||||
this.pump()
|
||||
}
|
||||
}
|
||||
|
||||
private failAll(first: Task | undefined, error: Error): void {
|
||||
const tasks = [...(first ? [first] : []), ...this.queue.splice(0)]
|
||||
for (const task of tasks) {
|
||||
task.cleanup()
|
||||
task.reject(error)
|
||||
}
|
||||
}
|
||||
|
||||
private ensureWorker(): Worker {
|
||||
if (this.worker) return this.worker
|
||||
const worker = new Worker(this.workerPath)
|
||||
this.worker = worker
|
||||
worker.on('message', (message: DocumentParseResponse) => {
|
||||
const task = this.active
|
||||
if (this.worker !== worker || !task || task.id !== message.id) return
|
||||
this.active = undefined
|
||||
task.cleanup()
|
||||
if ('error' in message) {
|
||||
void deserializeDocumentParseError(message.error).then(task.reject)
|
||||
} else {
|
||||
task.resolve(message.result)
|
||||
}
|
||||
this.pump()
|
||||
})
|
||||
const failed = (error: Error): void => {
|
||||
// Ignore exits of workers that were terminated on purpose (abort, close).
|
||||
if (this.worker !== worker) return
|
||||
this.worker = undefined
|
||||
this.disabledUntil = this.now() + RESTART_BACKOFF_MS
|
||||
const active = this.active
|
||||
this.active = undefined
|
||||
this.failAll(active, new DocumentParseWorkerUnavailableError(error.message, { cause: error }))
|
||||
void worker.terminate()
|
||||
}
|
||||
worker.once('error', failed)
|
||||
worker.once('exit', (code) => failed(new Error(`Document parse 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 active = this.active
|
||||
this.active = undefined
|
||||
this.failAll(active, new DocumentParseWorkerUnavailableError('Document parse worker closed'))
|
||||
if (worker) void worker.terminate()
|
||||
}
|
||||
}
|
||||
|
||||
let sharedClient: DocumentParseWorkerClient | undefined
|
||||
|
||||
/** Enables the parse worker for this process (Main calls it once at startup). */
|
||||
export function configureDocumentParseWorker(workerPath: string | undefined): void {
|
||||
sharedClient?.close()
|
||||
sharedClient = workerPath ? new DocumentParseWorkerClient(workerPath) : undefined
|
||||
}
|
||||
|
||||
export function closeDocumentParseWorker(): void {
|
||||
sharedClient?.close()
|
||||
sharedClient = undefined
|
||||
}
|
||||
|
||||
/** A private copy of the bytes that can be transferred without detaching the caller's buffer. */
|
||||
function transferableCopy(buffer: Uint8Array): ArrayBuffer {
|
||||
return Uint8Array.prototype.slice.call(buffer).buffer as ArrayBuffer
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs `op` in the worker when available, otherwise inline. Worker
|
||||
* infrastructure failures fall back to the inline path; operation errors and
|
||||
* aborts propagate unchanged. An already-aborted signal also runs inline, so
|
||||
* validation and abort errors surface in exactly the inline order.
|
||||
*/
|
||||
export async function runOffMain<T>(
|
||||
client: DocumentParseWorkerClient | undefined,
|
||||
op: DocumentParseOperation,
|
||||
request: () => { args: unknown[]; transfer?: ArrayBuffer[] },
|
||||
signal: AbortSignal | undefined,
|
||||
inline: () => T | Promise<T>
|
||||
): Promise<T> {
|
||||
if (!client?.available || signal?.aborted) return inline()
|
||||
const { args, transfer } = request()
|
||||
try {
|
||||
return await client.call<T>(op, args, transfer, signal)
|
||||
} catch (error) {
|
||||
if (error instanceof DocumentParseWorkerUnavailableError && !signal?.aborted) return inline()
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
export function parseDocumentOffMain(
|
||||
name: string,
|
||||
buffer: Buffer,
|
||||
signal?: AbortSignal,
|
||||
client: DocumentParseWorkerClient | undefined = sharedClient
|
||||
): Promise<ParsedDocument> {
|
||||
// Cheap size checks stay inline so oversized input is never copied.
|
||||
try {
|
||||
assertDocumentBuffer(buffer)
|
||||
} catch (error) {
|
||||
return Promise.reject(error)
|
||||
}
|
||||
return runOffMain(client, 'parseDocument', () => {
|
||||
const bytes = transferableCopy(buffer)
|
||||
return { args: [name, bytes], transfer: [bytes] }
|
||||
}, signal, () => parseDocument(name, buffer, signal))
|
||||
}
|
||||
|
||||
export function extractPdfTextPagesOffMain(
|
||||
buffer: Buffer,
|
||||
options: PdfTextExtractionOptions = {},
|
||||
client: DocumentParseWorkerClient | undefined = sharedClient
|
||||
): Promise<PdfTextExtraction> {
|
||||
const { signal, ...limits } = options
|
||||
return runOffMain(client, 'extractPdfTextPages', () => {
|
||||
const bytes = transferableCopy(buffer)
|
||||
return { args: [bytes, limits], transfer: [bytes] }
|
||||
}, signal, () => extractPdfTextPages(buffer, options))
|
||||
}
|
||||
|
||||
export function extractPptxPagesOffMain(
|
||||
buffer: Buffer,
|
||||
client: DocumentParseWorkerClient | undefined = sharedClient
|
||||
): Promise<PptxPage[]> {
|
||||
return runOffMain(client, 'extractPptxPages', () => {
|
||||
const bytes = transferableCopy(buffer)
|
||||
return { args: [bytes], transfer: [bytes] }
|
||||
}, undefined, () => extractPptxPages(buffer))
|
||||
}
|
||||
|
||||
/**
|
||||
* Not cancellable on purpose: the inline call it replaces was synchronous, so
|
||||
* an abort during chunking keeps surfacing at the caller's next checkpoint.
|
||||
*/
|
||||
export function chunkDocumentOffMain(
|
||||
document: ParsedDocument,
|
||||
settings: KnowledgeChunkingSettings,
|
||||
client: DocumentParseWorkerClient | undefined = sharedClient
|
||||
): Promise<DocumentChunk[]> {
|
||||
return runOffMain(client, 'chunkDocument', () => ({
|
||||
// Only the fields chunking reads; images (Buffers) would otherwise be cloned.
|
||||
args: [{ title: document.title, sourceFormat: document.sourceFormat, content: '', sections: document.sections, warnings: [] }, settings]
|
||||
}), undefined, () => chunkDocumentAdvanced(document, settings))
|
||||
}
|
||||
|
||||
/** Hex SHA-256 of `buffer`, computed in the worker for large inputs. */
|
||||
export function sha256OffMain(
|
||||
buffer: Uint8Array,
|
||||
client: DocumentParseWorkerClient | undefined = sharedClient
|
||||
): Promise<string> {
|
||||
const inline = (): string => createHash('sha256').update(buffer).digest('hex')
|
||||
// Small inputs hash faster than a round trip.
|
||||
if (buffer.byteLength < 1024 * 1024) return Promise.resolve().then(inline)
|
||||
return runOffMain(client, 'sha256', () => {
|
||||
const bytes = transferableCopy(buffer)
|
||||
return { args: [bytes], transfer: [bytes] }
|
||||
}, undefined, inline)
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
// Wire format shared by document-parse-client (Main) and document-parse-worker.
|
||||
|
||||
export type DocumentParseOperation =
|
||||
| 'parseDocument'
|
||||
| 'extractPdfTextPages'
|
||||
| 'extractPptxPages'
|
||||
| 'chunkDocument'
|
||||
| 'sha256'
|
||||
|
||||
export type DocumentParseRequest = {
|
||||
id: number
|
||||
op: DocumentParseOperation
|
||||
args: unknown[]
|
||||
}
|
||||
|
||||
export type SerializedDocumentParseError = {
|
||||
name: string
|
||||
message: string
|
||||
/** Primitive own properties some error types carry (PDF.js, Node). */
|
||||
extra?: Record<string, unknown>
|
||||
cause?: SerializedDocumentParseError
|
||||
}
|
||||
|
||||
export type DocumentParseResponse =
|
||||
| { id: number; result: unknown }
|
||||
| { id: number; error: SerializedDocumentParseError }
|
||||
|
||||
export const documentParseErrorExtraKeys = ['code', 'status', 'missing', 'details', 'extraDelay'] as const
|
||||
@@ -0,0 +1,284 @@
|
||||
// @vitest-environment node
|
||||
import { createHash } from 'node:crypto'
|
||||
import { isDeepStrictEqual } from 'node:util'
|
||||
import { afterAll, beforeAll, describe, expect, it, onTestFinished } from 'vitest'
|
||||
import { strToU8, zipSync } from 'fflate'
|
||||
import type { KnowledgeChunkingSettings } from '../shared/knowledge-contracts'
|
||||
import {
|
||||
buildDocumentParseWorker,
|
||||
createDocx,
|
||||
createLargePdf,
|
||||
createPptx,
|
||||
createTextPdf,
|
||||
createXlsx,
|
||||
sentenceGenerator
|
||||
} from '../../tests/support/document-parse-worker-fixture'
|
||||
import { createImagePptx } from '../../tests/support/pptx-fixture'
|
||||
import {
|
||||
chunkDocumentOffMain,
|
||||
DocumentParseWorkerClient,
|
||||
DocumentParseWorkerUnavailableError,
|
||||
extractPdfTextPagesOffMain,
|
||||
extractPptxPagesOffMain,
|
||||
parseDocumentOffMain,
|
||||
sha256OffMain
|
||||
} from './document-parse-client'
|
||||
import {
|
||||
chunkDocumentAdvanced,
|
||||
DocumentTextUnavailableError,
|
||||
extractPdfTextPages,
|
||||
parseDocument
|
||||
} from './knowledge/document-parser'
|
||||
import { extractPptxPages } from './knowledge/pptx-parser'
|
||||
|
||||
let workerPath: string
|
||||
let dispose: () => Promise<void>
|
||||
let client: DocumentParseWorkerClient
|
||||
|
||||
beforeAll(async () => {
|
||||
({ workerPath, dispose } = await buildDocumentParseWorker())
|
||||
client = new DocumentParseWorkerClient(workerPath)
|
||||
}, 60_000)
|
||||
afterAll(async () => {
|
||||
client?.close()
|
||||
await dispose?.()
|
||||
})
|
||||
|
||||
const sentence = sentenceGenerator(3)
|
||||
const markdown = [
|
||||
'# GoodBuddy',
|
||||
'知识内容。'.repeat(400),
|
||||
'## 安装',
|
||||
...Array.from({ length: 40 }, (_, index) => `Step ${index}: ${sentence(30)}.`),
|
||||
'### 配置\r\n' + sentence(200),
|
||||
'# 附录',
|
||||
sentence(600)
|
||||
].join('\n')
|
||||
|
||||
/** Fixture corpus: every format the parser accepts, plus edge cases. */
|
||||
const fixtures: Array<{ name: string; buffer: Buffer }> = [
|
||||
{ name: 'notes.md', buffer: Buffer.from(markdown) },
|
||||
{ name: 'plain.txt', buffer: Buffer.from(` ${sentence(5_000)} `) },
|
||||
{ name: 'data.json', buffer: Buffer.from(JSON.stringify({ items: Array.from({ length: 200 }, (_, id) => ({ id, text: sentence(8) })) }, null, 2)) },
|
||||
{ name: 'page.html', buffer: Buffer.from('<main><h1>安全标题</h1><p>网页正文</p><table><tr><td>a</td><td>b</td></tr></table></main><script>恶意脚本</script><style>x{}</style>') },
|
||||
{ name: 'large.txt', buffer: Buffer.from('x'.repeat(5_000_100)) },
|
||||
{ name: 'sample.pdf', buffer: createTextPdf([['PDF body text']]) },
|
||||
{ name: 'multi.pdf', buffer: createTextPdf([['first line', 'second (line)'], [], ['third page', 'ends.']]) },
|
||||
{ name: 'big.pdf', buffer: createLargePdf(40, 40) },
|
||||
{ name: 'sample.docx', buffer: createDocx(['文档正文', '第二段', ''], [[['h1', 'h2'], ['v1', 'v2']]]) },
|
||||
{ name: 'sample.xlsx', buffer: createXlsx([[['name', 'qty', 'note'], ['apple', '3', 'fresh']], [['b', '7', 'inline']]]) },
|
||||
{ name: 'shared.xlsx', buffer: Buffer.from(zipSync({ 'xl/sharedStrings.xml': strToU8('<sst><si><t>表格内容</t></si></sst>') })) },
|
||||
{ name: 'sample.pptx', buffer: createPptx(['幻灯片内容', 'second slide', 'third']) },
|
||||
{ name: 'image.pptx', buffer: createPptx(['with image', 'two'], true) }
|
||||
]
|
||||
|
||||
const failureFixtures: Array<{ name: string; buffer: Buffer }> = [
|
||||
{ name: 'archive.zip', buffer: Buffer.from('not supported') },
|
||||
{ name: 'empty.txt', buffer: Buffer.alloc(0) },
|
||||
{ name: 'large.txt', buffer: Buffer.alloc(20 * 1024 * 1024 + 1) },
|
||||
{ name: 'expanded.docx', buffer: Buffer.from(zipSync({ 'word/document.xml': new Uint8Array(11 * 1024 * 1024) })) },
|
||||
{ name: 'invalid.txt', buffer: Buffer.from([0xc3, 0x28]) },
|
||||
{ name: 'nulls.txt', buffer: Buffer.from('a\0b\0c\0d') },
|
||||
{ name: 'blank.md', buffer: Buffer.from(' \n\n ') },
|
||||
{ name: 'broken.pdf', buffer: Buffer.from('%PDF-1.4 not really') },
|
||||
{ name: 'empty.pdf', buffer: createTextPdf([[]]) },
|
||||
{ name: 'corrupt.xlsx', buffer: Buffer.from('PK not a zip') }
|
||||
]
|
||||
|
||||
const chunkingModes: KnowledgeChunkingSettings[] = [
|
||||
{ version: 1, mode: 'fixed', targetCharacters: 500, overlapCharacters: 50, parentCharacters: 4_800, childCharacters: 900, contextualIndexingEnabled: false },
|
||||
{ version: 1, mode: 'structure', targetCharacters: 800, overlapCharacters: 80, parentCharacters: 4_800, childCharacters: 900, contextualIndexingEnabled: true },
|
||||
{ version: 1, mode: 'parent-child', targetCharacters: 1_600, overlapCharacters: 100, parentCharacters: 1_600, childCharacters: 400, contextualIndexingEnabled: false }
|
||||
]
|
||||
|
||||
/** Byte-level comparison: JSON (key order) plus deep equality (undefined fields, typed arrays). */
|
||||
function expectIdentical(actual: unknown, expected: unknown): void {
|
||||
expect(JSON.stringify(actual)).toBe(JSON.stringify(expected))
|
||||
expect(isDeepStrictEqual(normalizeBytes(actual), normalizeBytes(expected))).toBe(true)
|
||||
}
|
||||
|
||||
function normalizeBytes(value: unknown): unknown {
|
||||
if (value instanceof Uint8Array) return { bytes: Buffer.from(value).toString('hex') }
|
||||
if (Array.isArray(value)) return value.map(normalizeBytes)
|
||||
if (value && typeof value === 'object') {
|
||||
return Object.fromEntries(Object.entries(value).map(([key, item]) => [key, normalizeBytes(item)]))
|
||||
}
|
||||
return value
|
||||
}
|
||||
|
||||
async function settle<T>(run: () => Promise<T>): Promise<{ ok: true; value: T } | { ok: false; error: unknown }> {
|
||||
try {
|
||||
return { ok: true, value: await run() }
|
||||
} catch (error) {
|
||||
return { ok: false, error }
|
||||
}
|
||||
}
|
||||
|
||||
function describeError(error: unknown): unknown {
|
||||
return error instanceof Error
|
||||
? { constructor: error.constructor, name: error.name, message: error.message,
|
||||
own: Object.entries(error).filter(([key]) => key !== 'stack'),
|
||||
unavailable: error instanceof DocumentTextUnavailableError,
|
||||
cause: describeError(error.cause) }
|
||||
: error
|
||||
}
|
||||
|
||||
describe('document parse worker equivalence', () => {
|
||||
it('parses every fixture to output identical to the inline parser', async () => {
|
||||
for (const fixture of fixtures) {
|
||||
const inline = await parseDocument(fixture.name, fixture.buffer)
|
||||
const before = Buffer.from(fixture.buffer)
|
||||
const offMain = await parseDocumentOffMain(fixture.name, fixture.buffer, undefined, client)
|
||||
expectIdentical(offMain, inline)
|
||||
// The caller's buffer is not detached or modified by the transfer.
|
||||
expect(fixture.buffer.equals(before)).toBe(true)
|
||||
}
|
||||
}, 60_000)
|
||||
|
||||
it('chunks every fixture identically in all chunking modes', async () => {
|
||||
// large.txt (5M chars, no break points) is excluded: splitNatural's backwards
|
||||
// boundary scan makes it take minutes on either path.
|
||||
for (const fixture of fixtures.filter((item) => item.name !== 'large.txt')) {
|
||||
const parsed = await parseDocument(fixture.name, fixture.buffer)
|
||||
for (const settings of chunkingModes) {
|
||||
const inline = await settle(async () => chunkDocumentAdvanced(parsed, settings))
|
||||
const offMain = await settle(() => chunkDocumentOffMain(parsed, settings, client))
|
||||
expect(offMain.ok).toBe(inline.ok)
|
||||
if (inline.ok && offMain.ok) expectIdentical(offMain.value, inline.value)
|
||||
else expect(describeError(offMain.ok ? undefined : offMain.error)).toEqual(describeError(inline.ok ? undefined : inline.error))
|
||||
}
|
||||
}
|
||||
}, 60_000)
|
||||
|
||||
it('extracts PDF pages and PPTX pages identically', async () => {
|
||||
// Pre-existing PDF.js behaviour on the inline path: cancelling a text stream
|
||||
// at the character limit can leave a stray "Controller is already closed".
|
||||
const strayPdfRejection = (reason: unknown): void => {
|
||||
if ((reason as { code?: string })?.code !== 'ERR_INVALID_STATE') throw reason
|
||||
}
|
||||
process.on('unhandledRejection', strayPdfRejection)
|
||||
onTestFinished(async () => {
|
||||
await new Promise((resolve) => setTimeout(resolve, 50))
|
||||
process.removeListener('unhandledRejection', strayPdfRejection)
|
||||
})
|
||||
for (const fixture of fixtures.filter((item) => item.name.endsWith('.pdf'))) {
|
||||
for (const options of [{}, { maximumCharacters: 50 }, { maximumPages: 1 }]) {
|
||||
const inline = await settle(() => extractPdfTextPages(fixture.buffer, options))
|
||||
const offMain = await settle(() => extractPdfTextPagesOffMain(fixture.buffer, options, client))
|
||||
if (inline.ok && offMain.ok) expectIdentical(offMain.value, inline.value)
|
||||
else expect(describeError(offMain.ok ? undefined : offMain.error)).toEqual(describeError(inline.ok ? undefined : inline.error))
|
||||
}
|
||||
}
|
||||
for (const buffer of [createPptx(['a', 'b'], true), createImagePptx('native', 3), createPptx(['plain'])]) {
|
||||
const inline = extractPptxPages(buffer)
|
||||
const offMain = await extractPptxPagesOffMain(buffer, client)
|
||||
expectIdentical(offMain, inline)
|
||||
for (const page of offMain) {
|
||||
for (const image of page.images) expect(image.data).toBeInstanceOf(Uint8Array)
|
||||
}
|
||||
}
|
||||
}, 60_000)
|
||||
|
||||
it('hashes identically', async () => {
|
||||
for (const buffer of [Buffer.alloc(0), Buffer.from('abc'), createLargePdf(40, 40), Buffer.alloc(3 * 1024 * 1024, 7)]) {
|
||||
expect(await sha256OffMain(buffer, client)).toBe(createHash('sha256').update(buffer).digest('hex'))
|
||||
}
|
||||
})
|
||||
|
||||
it('maps errors to the same types and messages as the inline parser', async () => {
|
||||
for (const fixture of failureFixtures) {
|
||||
const inline = await settle(() => parseDocument(fixture.name, fixture.buffer))
|
||||
const offMain = await settle(() => parseDocumentOffMain(fixture.name, fixture.buffer, undefined, client))
|
||||
expect(inline.ok, fixture.name).toBe(false)
|
||||
expect(offMain.ok, fixture.name).toBe(false)
|
||||
expect(describeError(offMain.ok ? undefined : offMain.error), fixture.name)
|
||||
.toEqual(describeError(inline.ok ? undefined : inline.error))
|
||||
}
|
||||
const tooManySections = {
|
||||
title: 'Too many', sourceFormat: '.txt', content: '', warnings: [],
|
||||
sections: Array.from({ length: 10_001 }, (_, index) => ({ locator: `s-${index}`, content: 'content' }))
|
||||
}
|
||||
const inline = await settle(async () => chunkDocumentAdvanced(tooManySections, chunkingModes[0]!))
|
||||
const offMain = await settle(() => chunkDocumentOffMain(tooManySections, chunkingModes[0]!, client))
|
||||
expect(describeError(offMain.ok ? undefined : offMain.error)).toEqual(describeError(inline.ok ? undefined : inline.error))
|
||||
const range = await settle(() => extractPdfTextPagesOffMain(createTextPdf([['x']]), { maximumPages: 0 }, client))
|
||||
expect(range.ok ? undefined : range.error).toBeInstanceOf(RangeError)
|
||||
})
|
||||
})
|
||||
|
||||
describe('document parse worker lifecycle', () => {
|
||||
it('cancels a running parse by terminating the worker and keeps serving later tasks', async () => {
|
||||
const local = new DocumentParseWorkerClient(workerPath)
|
||||
try {
|
||||
const big = createLargePdf(400, 50)
|
||||
const controller = new AbortController()
|
||||
const reason = new Error('文档解析已取消')
|
||||
const running = parseDocumentOffMain('big.pdf', big, controller.signal, local)
|
||||
const queued = parseDocumentOffMain('notes.md', Buffer.from(markdown), undefined, local)
|
||||
const queuedCancelled = new AbortController()
|
||||
const queuedAbort = expect(parseDocumentOffMain('plain.txt', Buffer.from('x'), queuedCancelled.signal, local)).rejects.toBe(reason)
|
||||
queuedCancelled.abort(reason)
|
||||
await new Promise((resolve) => setTimeout(resolve, 30))
|
||||
controller.abort(reason)
|
||||
await expect(running).rejects.toBe(reason)
|
||||
await queuedAbort
|
||||
expectIdentical(await queued, await parseDocument('notes.md', Buffer.from(markdown)))
|
||||
expect(local.pendingCount).toBe(0)
|
||||
} finally {
|
||||
local.close()
|
||||
}
|
||||
}, 60_000)
|
||||
|
||||
it('rejects with the abort reason when already aborted, without starting the worker', async () => {
|
||||
const local = new DocumentParseWorkerClient(workerPath)
|
||||
try {
|
||||
const controller = new AbortController()
|
||||
const reason = new Error('cancelled')
|
||||
controller.abort(reason)
|
||||
await expect(parseDocumentOffMain('a.txt', Buffer.from('a'), controller.signal, local)).rejects.toBe(reason)
|
||||
} finally {
|
||||
local.close()
|
||||
}
|
||||
})
|
||||
|
||||
it('falls back to the inline path when the worker crashes or cannot start', async () => {
|
||||
const local = new DocumentParseWorkerClient(workerPath)
|
||||
try {
|
||||
const running = local.call('parseDocument', ['big.pdf', Uint8Array.from(createLargePdf(200, 50)).buffer])
|
||||
await new Promise((resolve) => setTimeout(resolve, 20))
|
||||
await local.terminateWorkerForTest()
|
||||
await expect(running).rejects.toBeInstanceOf(DocumentParseWorkerUnavailableError)
|
||||
expect(local.available).toBe(false)
|
||||
// While backing off, calls run inline with identical output.
|
||||
const buffer = Buffer.from(markdown)
|
||||
expectIdentical(await parseDocumentOffMain('notes.md', buffer, undefined, local), await parseDocument('notes.md', buffer))
|
||||
local.resetBackoffForTest()
|
||||
expect(local.available).toBe(true)
|
||||
expectIdentical(await parseDocumentOffMain('notes.md', buffer, undefined, local), await parseDocument('notes.md', buffer))
|
||||
} finally {
|
||||
local.close()
|
||||
}
|
||||
|
||||
const missing = new DocumentParseWorkerClient(`${workerPath}.missing.cjs`)
|
||||
try {
|
||||
const buffer = createDocx(['fallback'])
|
||||
expectIdentical(await parseDocumentOffMain('a.docx', buffer, undefined, missing), await parseDocument('a.docx', buffer))
|
||||
expect(missing.available).toBe(false)
|
||||
} finally {
|
||||
missing.close()
|
||||
}
|
||||
}, 60_000)
|
||||
|
||||
it('runs tasks one at a time in FIFO order', async () => {
|
||||
const local = new DocumentParseWorkerClient(workerPath)
|
||||
try {
|
||||
const names = fixtures.slice(0, 6)
|
||||
const results = await Promise.all(names.map((fixture) => parseDocumentOffMain(fixture.name, fixture.buffer, undefined, local)))
|
||||
for (const [index, fixture] of names.entries()) {
|
||||
expectIdentical(results[index], await parseDocument(fixture.name, fixture.buffer))
|
||||
}
|
||||
} finally {
|
||||
local.close()
|
||||
}
|
||||
}, 60_000)
|
||||
})
|
||||
@@ -0,0 +1,101 @@
|
||||
import { createHash } from 'node:crypto'
|
||||
import { parentPort } from 'node:worker_threads'
|
||||
import type { KnowledgeChunkingSettings } from '../shared/knowledge-contracts'
|
||||
import {
|
||||
chunkDocumentAdvanced,
|
||||
extractPdfTextPages,
|
||||
parseDocument,
|
||||
type ParsedDocument
|
||||
} from './knowledge/document-parser'
|
||||
import { extractPptxPages } from './knowledge/pptx-parser'
|
||||
import {
|
||||
documentParseErrorExtraKeys,
|
||||
type DocumentParseRequest,
|
||||
type DocumentParseResponse,
|
||||
type SerializedDocumentParseError
|
||||
} from './document-parse-errors'
|
||||
|
||||
// PERF-15/16: CPU-heavy, pure document work (PDF / Office / text parsing,
|
||||
// chunking, hashing) runs here so Main only orchestrates. Every handler calls
|
||||
// the same production function as the inline path, so results are identical.
|
||||
// No database access and no file system access happen in this worker.
|
||||
|
||||
type Handler = (args: unknown[]) => Promise<unknown> | unknown
|
||||
|
||||
const asBuffer = (value: unknown): Buffer => {
|
||||
const bytes = value as ArrayBuffer
|
||||
return Buffer.from(bytes, 0, bytes.byteLength)
|
||||
}
|
||||
|
||||
const handlers: Record<string, Handler> = {
|
||||
parseDocument: ([name, bytes]) => parseDocument(name as string, asBuffer(bytes)),
|
||||
extractPdfTextPages: ([bytes, options]) => extractPdfTextPages(
|
||||
asBuffer(bytes),
|
||||
options as { maximumPages?: number; maximumCharacters?: number }
|
||||
),
|
||||
extractPptxPages: ([bytes]) => extractPptxPages(asBuffer(bytes)),
|
||||
chunkDocument: ([document, settings]) => chunkDocumentAdvanced(
|
||||
document as ParsedDocument,
|
||||
settings as KnowledgeChunkingSettings
|
||||
),
|
||||
sha256: ([bytes]) => createHash('sha256').update(asBuffer(bytes)).digest('hex')
|
||||
}
|
||||
|
||||
/** ArrayBuffers backing typed arrays in the result (PPTX images); transferred, not copied. */
|
||||
function transferables(value: unknown): ArrayBuffer[] {
|
||||
const buffers = new Set<ArrayBuffer>()
|
||||
if (Array.isArray(value)) {
|
||||
for (const page of value as Array<{ images?: Array<{ data?: unknown }> }>) {
|
||||
for (const image of page?.images ?? []) {
|
||||
const data = image?.data
|
||||
if (
|
||||
data instanceof Uint8Array &&
|
||||
data.buffer instanceof ArrayBuffer &&
|
||||
data.byteOffset === 0 &&
|
||||
data.byteLength === data.buffer.byteLength
|
||||
) {
|
||||
buffers.add(data.buffer)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return [...buffers]
|
||||
}
|
||||
|
||||
function serializeError(error: unknown, depth = 0): SerializedDocumentParseError {
|
||||
if (!(error instanceof Error)) return { name: 'Error', message: String(error) }
|
||||
const extra: Record<string, unknown> = {}
|
||||
for (const key of documentParseErrorExtraKeys) {
|
||||
const value = (error as unknown as Record<string, unknown>)[key]
|
||||
if (value !== undefined && (typeof value !== 'object' || value === null)) extra[key] = value
|
||||
}
|
||||
return {
|
||||
name: error.name,
|
||||
message: error.message,
|
||||
...(Object.keys(extra).length > 0 ? { extra } : {}),
|
||||
...(error.cause !== undefined && depth < 3 ? { cause: serializeError(error.cause, depth + 1) } : {})
|
||||
}
|
||||
}
|
||||
|
||||
// PDF.js can leave a stray rejection ("Controller is already closed") after a
|
||||
// text stream is cancelled at the character limit. Unhandled, it would kill
|
||||
// this thread and push the next tasks onto the inline fallback. Task errors are
|
||||
// awaited by the handler and still reported through the response.
|
||||
process.on('unhandledRejection', () => undefined)
|
||||
|
||||
parentPort!.on('message', (message: DocumentParseRequest) => {
|
||||
void (async () => {
|
||||
let response: DocumentParseResponse
|
||||
let transfer: ArrayBuffer[] = []
|
||||
try {
|
||||
const handler = Object.hasOwn(handlers, message.op) ? handlers[message.op] : undefined
|
||||
if (!handler) throw new Error(`Unknown document parse operation: ${message.op}`)
|
||||
const result = await handler(message.args)
|
||||
transfer = message.op === 'extractPptxPages' ? transferables(result) : []
|
||||
response = { id: message.id, result }
|
||||
} catch (error) {
|
||||
response = { id: message.id, error: serializeError(error) }
|
||||
}
|
||||
parentPort!.postMessage(response, transfer)
|
||||
})()
|
||||
})
|
||||
@@ -18,12 +18,14 @@ import { HttpDocumentOcr } from './http-document-ocr'
|
||||
import { hasExtractedDocumentText } from './document-extracted-text'
|
||||
import { renderOcrPdf, renderSelectedOcrPdf } from './render-ocr-pdf'
|
||||
import type { DocumentResultStorage } from './document-result-storage'
|
||||
import { extractPptxPages } from './knowledge/pptx-parser'
|
||||
import {
|
||||
extractPdfTextPagesOffMain,
|
||||
extractPptxPagesOffMain,
|
||||
parseDocumentOffMain
|
||||
} from './document-parse-client'
|
||||
import {
|
||||
assertDocumentBuffer,
|
||||
DocumentTextUnavailableError,
|
||||
extractPdfTextPages,
|
||||
parseDocument,
|
||||
type ParsedDocument,
|
||||
type ParsedSection,
|
||||
type PdfTextPage
|
||||
@@ -279,10 +281,10 @@ export class DocumentParsingService {
|
||||
return this.parsePptx(name, buffer, purpose, operation, signal)
|
||||
}
|
||||
if (extname(name).toLowerCase() !== '.pdf') {
|
||||
return parseDocument(name, buffer, signal)
|
||||
return parseDocumentOffMain(name, buffer, signal)
|
||||
}
|
||||
|
||||
const extracted = await extractPdfTextPages(buffer, { signal })
|
||||
const extracted = await extractPdfTextPagesOffMain(buffer, { signal })
|
||||
const { pages } = extracted
|
||||
ensureNotAborted(signal)
|
||||
const mode = effectiveOcrMode(settings, purpose)
|
||||
@@ -517,7 +519,7 @@ export class DocumentParsingService {
|
||||
const mode = effectiveOcrMode(settings, purpose)
|
||||
if (mode === 'disabled') {
|
||||
try {
|
||||
return await parseDocument(name, buffer, signal)
|
||||
return await parseDocumentOffMain(name, buffer, signal)
|
||||
} catch (error) {
|
||||
if (error instanceof DocumentTextUnavailableError) {
|
||||
throw new DocumentTextUnavailableError(
|
||||
@@ -527,7 +529,7 @@ export class DocumentParsingService {
|
||||
throw error
|
||||
}
|
||||
}
|
||||
const pages = extractPptxPages(buffer)
|
||||
const pages = await extractPptxPagesOffMain(buffer)
|
||||
ensureNotAborted(signal)
|
||||
const native: ParsedSection[] = pages
|
||||
.filter((page) => page.content)
|
||||
|
||||
+31
-15
@@ -1,3 +1,4 @@
|
||||
import { startupMark, startupSpan } from './startup-marks'
|
||||
import { localInferenceService } from './local-inference-service'
|
||||
import { NativeClientCoordinator } from './agent/native-client-coordinator'
|
||||
import { createEmbeddingUtilityTransport } from './knowledge/embedding-utility-transport'
|
||||
@@ -98,6 +99,7 @@ import { DocumentParsingSettingsStore } from './document-parsing-settings-store'
|
||||
import { DocumentOcrModelManager } from './document-ocr-model-manager'
|
||||
import { DocumentOcrBroker } from './document-ocr-broker'
|
||||
import { DocumentParsingService } from './document-parsing-service'
|
||||
import { configureDocumentParseWorker } from './document-parse-client'
|
||||
import { DocumentResultStorage } from './document-result-storage'
|
||||
import { ConversationAttachmentStorage } from './conversation-attachment-storage'
|
||||
import { ReleaseNotesService } from './release-notes-service'
|
||||
@@ -553,6 +555,7 @@ function buildTray(): Tray {
|
||||
const storageUpgradeController = new AbortController()
|
||||
let storageUpgrade: Promise<void> | undefined
|
||||
|
||||
startupMark('main:module-evaluated')
|
||||
if (hasSingleInstanceLock) {
|
||||
app.on('second-instance', () => {
|
||||
if (mainWindow) {
|
||||
@@ -561,6 +564,7 @@ if (hasSingleInstanceLock) {
|
||||
})
|
||||
|
||||
void app.whenReady().then(async () => {
|
||||
startupMark('main:when-ready')
|
||||
session.defaultSession.setPermissionRequestHandler(
|
||||
(webContents, permission, callback, details) => {
|
||||
const mediaTypes =
|
||||
@@ -582,15 +586,18 @@ if (hasSingleInstanceLock) {
|
||||
details.mediaType === 'audio'
|
||||
)
|
||||
|
||||
startupMark('main:create-window:start')
|
||||
mainWindow = createMainWindow(() => isQuitting, observeDesktopFailure)
|
||||
startupMark('main:create-window:end')
|
||||
registerDesktopNotificationActivation(mainWindow)
|
||||
tray = buildTray()
|
||||
startupMark('main:tray-built')
|
||||
storageUpgrade = prepareAssistantStorage(
|
||||
mainWindow,
|
||||
join(app.getPath('userData'), 'assistant.sqlite'),
|
||||
storageUpgradeController.signal
|
||||
)
|
||||
await storageUpgrade
|
||||
await startupSpan('main:storage-upgrade-check', () => storageUpgrade!)
|
||||
if (storageUpgradeController.signal.aborted) return
|
||||
const defaultWorkspace = process.env.GOODBUDDY_WORKSPACE ?? homedir()
|
||||
const secureCipher = {
|
||||
@@ -670,7 +677,7 @@ if (hasSingleInstanceLock) {
|
||||
},
|
||||
baseEnvironment: buildCredentialFilteredUserEnvironment()
|
||||
})
|
||||
await startupLocalToolEnvironmentService.initialize()
|
||||
await startupSpan('main:local-tool-environment', () => startupLocalToolEnvironmentService.initialize())
|
||||
localToolEnvironmentService =
|
||||
startupLocalToolEnvironmentService
|
||||
const sshHostStore = new SshHostStore(
|
||||
@@ -682,11 +689,11 @@ if (hasSingleInstanceLock) {
|
||||
resourcesPath: process.resourcesPath,
|
||||
packaged: app.isPackaged
|
||||
})
|
||||
const controlPlanePackageInstaller = await readFile(
|
||||
const controlPlanePackageInstaller = await startupSpan('main:read-control-plane-installer', () => readFile(
|
||||
resolveControlPlanePackageInstallerPath({
|
||||
appPath: app.getAppPath()
|
||||
})
|
||||
)
|
||||
))
|
||||
const agentPackageManager = new AgentPackageManager({
|
||||
userDataPath: app.getPath('userData'),
|
||||
desktopVersion: app.getVersion(),
|
||||
@@ -728,7 +735,7 @@ if (hasSingleInstanceLock) {
|
||||
return selection.provider === 'continue' ? resolved.continueModelProfile : resolved.opencodeModelProfile
|
||||
}
|
||||
})
|
||||
await startupManagedRemoteExecutionServices.initialize()
|
||||
await startupSpan('main:managed-remote-init', () => startupManagedRemoteExecutionServices.initialize())
|
||||
managedRemoteExecutionServices =
|
||||
startupManagedRemoteExecutionServices
|
||||
const beginRemoteHostInvalidation = (hostId: string): void => {
|
||||
@@ -777,10 +784,10 @@ if (hasSingleInstanceLock) {
|
||||
)
|
||||
})
|
||||
const [initialRuntimeSettings, initialResolvedSettings] =
|
||||
await Promise.all([
|
||||
await startupSpan('main:runtime-settings-load', () => Promise.all([
|
||||
settingsStore.getPublicSettings(),
|
||||
settingsStore.getResolvedSettings()
|
||||
])
|
||||
]))
|
||||
globalTlsPolicy = new GlobalTlsPolicy(app)
|
||||
globalTlsPolicy.install()
|
||||
const capabilityService = new CapabilityService(
|
||||
@@ -836,6 +843,7 @@ if (hasSingleInstanceLock) {
|
||||
(await applicationSettingsStore.get()).modelDownloadSource
|
||||
})
|
||||
documentOcrBroker = new DocumentOcrBroker(mainWindow)
|
||||
configureDocumentParseWorker(join(app.getAppPath(), 'out/main/document-parse-worker.js'))
|
||||
documentParsingService = new DocumentParsingService(
|
||||
documentParsingSettingsStore,
|
||||
documentOcrModelManager,
|
||||
@@ -1208,7 +1216,8 @@ if (hasSingleInstanceLock) {
|
||||
resolved.target
|
||||
)
|
||||
}
|
||||
const configuredRuntime = await runStartupPrerequisites({
|
||||
startupMark('main:services-constructed')
|
||||
const configuredRuntime = await startupSpan('main:prerequisites', () => runStartupPrerequisites({
|
||||
prepareDeepSeekHome: async () => {
|
||||
await mkdir(deepSeekHarnessHome, {
|
||||
recursive: true,
|
||||
@@ -1216,7 +1225,7 @@ if (hasSingleInstanceLock) {
|
||||
})
|
||||
},
|
||||
initializeKnowledgeAndGateway: async () => {
|
||||
await startupKnowledgeService.initialize()
|
||||
await startupSpan('main:knowledge-init', () => startupKnowledgeService.initialize())
|
||||
const embeddingProvider = await createEmbeddingProvider(
|
||||
initialResolvedSettings,
|
||||
startupEmbeddingModelManager
|
||||
@@ -1242,11 +1251,12 @@ if (hasSingleInstanceLock) {
|
||||
() => undefined
|
||||
)
|
||||
])
|
||||
await startupKnowledgeGateway.start()
|
||||
await startupSpan('main:knowledge-gateway-start', () => startupKnowledgeGateway.start())
|
||||
},
|
||||
hydrateConfiguredRuntime: () =>
|
||||
createConfiguredRuntime(initialResolvedSettings),
|
||||
startupSpan('main:runtime-hydrate', () => createConfiguredRuntime(initialResolvedSettings)),
|
||||
initializeAssistant: () => {
|
||||
startupMark('main:assistant-db-init:start')
|
||||
startupAssistantDatabase.initialize(defaultWorkspace)
|
||||
startupAssistantDatabase.enableReadonlyWorker(readonlyQueryWorkerPath)
|
||||
imageGenerationService!.initialize()
|
||||
@@ -1259,13 +1269,14 @@ if (hasSingleInstanceLock) {
|
||||
initialRuntimeSettings
|
||||
)
|
||||
)
|
||||
startupMark('main:assistant-db-init:end')
|
||||
}
|
||||
})
|
||||
}))
|
||||
const [initialSubagentRuntime, initialSubagentProfileRuntimes] =
|
||||
await Promise.all([
|
||||
await startupSpan('main:subagent-runtimes', () => Promise.all([
|
||||
createSubagentRuntime(initialResolvedSettings),
|
||||
createSubagentProfileRuntimes(initialResolvedSettings)
|
||||
])
|
||||
]))
|
||||
const subagentService = new SubagentService(
|
||||
initialSubagentRuntime,
|
||||
startupAssistantDatabase,
|
||||
@@ -1285,8 +1296,10 @@ if (hasSingleInstanceLock) {
|
||||
createSelectedStatusRuntime,
|
||||
(conversationId) => localRuntimeRegistry.releaseConversation(conversationId)
|
||||
)
|
||||
startupMark('main:attachments-reconcile:start')
|
||||
conversationAttachmentStorage = new ConversationAttachmentStorage(app.getPath('userData'), documentParsingService.results!)
|
||||
conversationAttachmentStorage.reconcile((conversationId, kind, ownerId) => startupAssistantDatabase.hasAttachmentOwner(conversationId, kind, ownerId), true)
|
||||
startupMark('main:attachments-reconcile:end')
|
||||
const contextManager = new ContextManager({
|
||||
parseDocument: documentParsingService.parse,
|
||||
assets: conversationAttachmentStorage,
|
||||
@@ -1308,7 +1321,7 @@ if (hasSingleInstanceLock) {
|
||||
},
|
||||
process.platform
|
||||
)
|
||||
await shortcutSettingsService.initialize()
|
||||
await startupSpan('main:shortcuts-init', () => shortcutSettingsService.initialize())
|
||||
|
||||
let runtimeReconfigurationQueue: Promise<void> = Promise.resolve()
|
||||
let runtimeReconfigurationClosing = false
|
||||
@@ -1513,6 +1526,7 @@ if (hasSingleInstanceLock) {
|
||||
obsidianService, launchEnvironmentProvider: startupLocalToolEnvironmentService.launchEnvironmentProvider
|
||||
})
|
||||
})
|
||||
startupMark('main:ipc-register:start')
|
||||
removeIpcHandlers = registerIpcHandlers(
|
||||
mainWindow,
|
||||
runtime,
|
||||
@@ -1581,7 +1595,9 @@ if (hasSingleInstanceLock) {
|
||||
mainWindow,
|
||||
startupFeedbackService
|
||||
)
|
||||
startupMark('main:ipc-register:end')
|
||||
loadMainWindow(mainWindow)
|
||||
startupMark('main:load-main-window')
|
||||
removeDeviceSharingIpc = registerDeviceSharingIpc(mainWindow,
|
||||
new DeviceSharingService(join(app.getPath('userData'), 'device-sharing-settings.json'), app.getVersion()))
|
||||
setImmediate(() => {
|
||||
|
||||
@@ -21,11 +21,14 @@ import {
|
||||
} from 'node:path'
|
||||
import {
|
||||
buildChunkContextPrefix,
|
||||
chunkDocumentAdvanced,
|
||||
parseDocument,
|
||||
supportedDocumentExtensions,
|
||||
type ParsedDocument
|
||||
} from './document-parser'
|
||||
import {
|
||||
chunkDocumentOffMain,
|
||||
parseDocumentOffMain,
|
||||
sha256OffMain
|
||||
} from '../document-parse-client'
|
||||
import type { DocumentResultStorage } from '../document-result-storage'
|
||||
import { parsedCompleteness } from '../document-result-storage'
|
||||
import {
|
||||
@@ -282,7 +285,7 @@ export class KnowledgeService {
|
||||
this.urlImporter = options.urlImporter ?? new UrlImporter()
|
||||
this.documentParser =
|
||||
options.parseDocument ??
|
||||
((name, buffer) => parseDocument(name, buffer))
|
||||
((name, buffer) => parseDocumentOffMain(name, buffer))
|
||||
this.embeddingProvider = options.embeddingProvider
|
||||
this.rerankProvider = options.rerankProvider
|
||||
const embeddingBatchSize = options.embeddingBatchSize ?? 16
|
||||
@@ -769,7 +772,7 @@ export class KnowledgeService {
|
||||
const documentId = input.id ?? randomUUID()
|
||||
const imagesOnly = parsedCompleteness(parsed) === 'images-only'
|
||||
if (imagesOnly && this.database.getDocument(documentId)?.metadata.status === 'ready') throw new Error('未提取到可索引文字，保留上次正文与索引')
|
||||
const chunks = imagesOnly ? [] : this.createDocumentChunks(parsed, library)
|
||||
const chunks = imagesOnly ? [] : await this.createDocumentChunks(parsed, library)
|
||||
let embeddingReplacement: PreparedEmbeddingReplacement | undefined
|
||||
let embeddingFailure: PreparedDocumentPublication['embeddingFailure']
|
||||
const embeddingProvider = this.embeddingProvider
|
||||
@@ -2484,7 +2487,7 @@ export class KnowledgeService {
|
||||
const prepared = await this.prepareDocumentPublication(
|
||||
{
|
||||
...document,
|
||||
checksum: createHash('sha256').update(buffer).digest('hex'),
|
||||
checksum: await sha256OffMain(buffer),
|
||||
metadata: {
|
||||
...document.metadata,
|
||||
status: 'ready',
|
||||
@@ -3137,7 +3140,7 @@ export class KnowledgeService {
|
||||
progress: 35,
|
||||
message: '正在解析文档内容'
|
||||
})
|
||||
const checksum = createHash('sha256').update(buffer).digest('hex')
|
||||
const checksum = await sha256OffMain(buffer)
|
||||
const previous = existing.find(
|
||||
(document) => document.externalId === file.relativePath
|
||||
)
|
||||
@@ -3835,11 +3838,12 @@ export class KnowledgeService {
|
||||
}
|
||||
}
|
||||
|
||||
private createDocumentChunks(
|
||||
private async createDocumentChunks(
|
||||
parsed: ParsedDocument,
|
||||
library: KnowledgeBase
|
||||
): ReplaceChunkInput[] {
|
||||
const chunks = chunkDocumentAdvanced(parsed, library.chunkingSettings)
|
||||
): Promise<ReplaceChunkInput[]> {
|
||||
// Pure CPU work runs off Main; chunk ids are still assigned here.
|
||||
const chunks = await chunkDocumentOffMain(parsed, library.chunkingSettings)
|
||||
const ids = chunks.map(() => randomUUID())
|
||||
return chunks.map((chunk) => ({
|
||||
id: ids[chunk.position],
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
import { lookup as dnsLookup } from 'node:dns/promises'
|
||||
import { request as httpRequest } from 'node:http'
|
||||
import { request as httpsRequest } from 'node:https'
|
||||
import { parseDocument, type ParsedDocument } from './document-parser'
|
||||
import type { ParsedDocument } from './document-parser'
|
||||
import { parseDocumentOffMain } from '../document-parse-client'
|
||||
|
||||
type ResolvedAddress = {
|
||||
address: string
|
||||
@@ -224,7 +225,7 @@ export class UrlImporter {
|
||||
.trim() || url.hostname
|
||||
).slice(0, 240)
|
||||
: url.pathname.split('/').filter(Boolean).at(-1) ?? url.hostname
|
||||
const document = await parseDocument(
|
||||
const document = await parseDocumentOffMain(
|
||||
isHtml ? `${title}.html` : `${title}.txt`,
|
||||
response.body
|
||||
)
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
import { performance } from 'node:perf_hooks'
|
||||
|
||||
// PERF-01 startup phase marks. Inactive unless GB_PERF_STARTUP_MARKS=1 (set by
|
||||
// build/run-startup-perf.cjs); when inactive each call is one boolean check.
|
||||
// Marks hold only fixed phase names and process-relative times, never user
|
||||
// content. The startup harness reads them from the global below.
|
||||
const enabled = process.env.GB_PERF_STARTUP_MARKS === '1'
|
||||
type StartupMark = { name: string; at: number }
|
||||
const marks: StartupMark[] | undefined = enabled ? [] : undefined
|
||||
if (marks) {
|
||||
;(globalThis as { __goodbuddyStartupMarks?: StartupMark[] }).__goodbuddyStartupMarks = marks
|
||||
}
|
||||
|
||||
export function startupMark(name: string): void {
|
||||
if (marks) marks.push({ name, at: performance.now() })
|
||||
}
|
||||
|
||||
// Takes a thunk so the span also covers the synchronous part of the operation
|
||||
// (everything before its first await).
|
||||
export function startupSpan<T>(name: string, operation: () => Promise<T>): Promise<T> {
|
||||
if (!marks) return operation()
|
||||
startupMark(`${name}:start`)
|
||||
let started: Promise<T>
|
||||
try {
|
||||
started = operation()
|
||||
} catch (error) {
|
||||
startupMark(`${name}:end`)
|
||||
throw error
|
||||
}
|
||||
startupMark(`${name}:sync-end`)
|
||||
return started.finally(() => startupMark(`${name}:end`))
|
||||
}
|
||||
Reference in New Issue
Block a user