From 883fa0896de05ef41cd6f4402059d4bf6d6a66db Mon Sep 17 00:00:00 2001 From: digua Date: Mon, 3 Aug 2026 16:37:39 +0800 Subject: [PATCH] docs(search): record FTS removal benchmarks --- docs/cn/standard/chatlab-import.md | 2 +- docs/cn/standard/chatlab-pull.md | 2 +- docs/en/standard/chatlab-import.md | 2 +- docs/en/standard/chatlab-pull.md | 2 +- .../src/semantic-index/retrieval/evidence.ts | 2 +- packages/tools/src/tool-metadata.ts | 2 +- scripts/bench-streaming-import.mts | 579 ++++++++++++++++++ 7 files changed, 585 insertions(+), 6 deletions(-) create mode 100644 scripts/bench-streaming-import.mts diff --git a/docs/cn/standard/chatlab-import.md b/docs/cn/standard/chatlab-import.md index 8912fa78..e989f024 100644 --- a/docs/cn/standard/chatlab-import.md +++ b/docs/cn/standard/chatlab-import.md @@ -13,7 +13,7 @@ outline: deep - **Push 模式**（本文档）：外部系统主动将数据推送到 ChatLab 的导入接口。适用于脚本集成、一次性文件导入等场景。 - **[Pull 模式](./chatlab-pull.md)**：第三方暴露标准 HTTP 端点，ChatLab 主动拉取数据。**推荐的第三方集成方式。** -两种模式底层共用同一套导入逻辑（去重、meta/members 更新、FTS 索引），数据格式统一为 [ChatLab Format](./chatlab-format.md)。 +两种模式底层共用同一套导入逻辑（去重、meta/members 更新、会话索引维护），数据格式统一为 [ChatLab Format](./chatlab-format.md)。 ::: diff --git a/docs/cn/standard/chatlab-pull.md b/docs/cn/standard/chatlab-pull.md index 61b44f96..8876a759 100644 --- a/docs/cn/standard/chatlab-pull.md +++ b/docs/cn/standard/chatlab-pull.md @@ -239,7 +239,7 @@ ChatLab 按用户配置的间隔，定期对已订阅的对话执行增量拉取 GET {baseUrl}/sessions/{sessionId}/messages?format=chatlab&since={lastPullAt} ``` -远程数据源返回 `since` 之后的增量消息。ChatLab 通过内部导入管道处理（去重、meta/members 更新、FTS 索引等全部复用 Push 模式逻辑）。 +远程数据源返回 `since` 之后的增量消息。ChatLab 通过内部导入管道处理（去重、meta/members 更新、会话索引维护等全部复用 Push 模式逻辑）。 --- diff --git a/docs/en/standard/chatlab-import.md b/docs/en/standard/chatlab-import.md index befb07e0..4a0c01dd 100644 --- a/docs/en/standard/chatlab-import.md +++ b/docs/en/standard/chatlab-import.md @@ -13,7 +13,7 @@ This document defines the standard import protocol for external data sources to - **Push mode** (this document): The external system actively pushes data to ChatLab's import endpoint. Suitable for script integrations and one-time file imports. - **[Pull mode](./chatlab-pull.md)**: A third-party exposes standard HTTP endpoints and ChatLab pulls data on demand. **The recommended integration approach for third-party tools.** -Both modes share the same underlying import pipeline (deduplication, meta/members update, FTS indexing). Data format is unified as [ChatLab Format](./chatlab-format.md). +Both modes share the same underlying import pipeline (deduplication, meta/members updates, and session-index maintenance). Data format is unified as [ChatLab Format](./chatlab-format.md). ::: diff --git a/docs/en/standard/chatlab-pull.md b/docs/en/standard/chatlab-pull.md index 8a1cba12..b7cae55a 100644 --- a/docs/en/standard/chatlab-pull.md +++ b/docs/en/standard/chatlab-pull.md @@ -239,7 +239,7 @@ ChatLab periodically performs incremental pulls on subscribed conversations at t GET {baseUrl}/sessions/{sessionId}/messages?format=chatlab&since={lastPullAt} ``` -The remote data source returns incremental messages since `lastPullAt`. ChatLab processes them through the internal import pipeline (deduplication, meta/members update, FTS indexing — all the same as Push mode). +The remote data source returns incremental messages since `lastPullAt`. ChatLab processes them through the internal import pipeline (deduplication, meta/members updates, and session-index maintenance — all the same as Push mode). --- diff --git a/packages/node-runtime/src/semantic-index/retrieval/evidence.ts b/packages/node-runtime/src/semantic-index/retrieval/evidence.ts index 85a3aa5a..51381a34 100644 --- a/packages/node-runtime/src/semantic-index/retrieval/evidence.ts +++ b/packages/node-runtime/src/semantic-index/retrieval/evidence.ts @@ -2,7 +2,7 @@ * 证据块组装 * * chunking-decision-final.md 第 14 节： - * - RRF 后取 topK 命中 chunk，每个命中在 parent 内前后各扩展若干条消息，不跨 parent。 + * - 语义检索后取 topK 命中 chunk，每个命中在 parent 内前后各扩展若干条消息，不跨 parent。 * - 相邻命中可合并，但合并后超过单块 soft cap 则拆开注入。 * - 总证据预算固定 1500 token：不够时优先减少前后扩展，再减少 chunk 数。 * diff --git a/packages/tools/src/tool-metadata.ts b/packages/tools/src/tool-metadata.ts index cd49a09e..b321e447 100644 --- a/packages/tools/src/tool-metadata.ts +++ b/packages/tools/src/tool-metadata.ts @@ -25,7 +25,7 @@ const ENGLISH_TOOL_METADATA: Record = { }, deep_search_messages: { description: - 'Search messages using slower substring matching. Use this for exact full-text matches or when regular search misses results.', + 'Search messages using slower substring matching. Use this for exact phrase matches or when regular search misses results.', properties: { keywords: 'Keywords to search for.', sender_id: 'Filter by sender ID obtained from get_members.', diff --git a/scripts/bench-streaming-import.mts b/scripts/bench-streaming-import.mts new file mode 100644 index 00000000..ab948555 --- /dev/null +++ b/scripts/bench-streaming-import.mts @@ -0,0 +1,579 @@ +/** + * End-to-end streaming import benchmark. + * + * Each measured run executes in a fresh process and writes fresh databases. + * Fixture generation and result verification are excluded from the timed range. + * + * Usage: + * pnpm exec tsx scripts/bench-streaming-import.mts single 100000 [runs=3] + * pnpm exec tsx scripts/bench-streaming-import.mts batch 100 10000 [runs=3] [concurrency=1] + * + * Single-file runs also append 1,000 messages incrementally and measure a + * representative LIKE query against the final database. + */ + +import { createHash } from 'node:crypto' +import { once } from 'node:events' +import { copyFileSync, createWriteStream, linkSync, mkdirSync, rmSync, statSync, unlinkSync } from 'node:fs' +import path from 'node:path' +import { fileURLToPath } from 'node:url' +import { spawnSync } from 'node:child_process' +import Database from 'better-sqlite3' +import { CHAT_DB_TABLES } from '@openchatlab/core' +import { getNativeParserStatus, isNativeFormatAvailable, type NativeParserStatus } from '@openchatlab/parser' +import { BetterSqliteAdapter } from '../packages/node-runtime/src/better-sqlite3-adapter' +import { computeAndSetOverviewCache } from '../packages/node-runtime/src/cache/session-cache' +import { runKeyedBatch } from '../packages/node-runtime/src/import/batch-coordinator' +import { incrementalImport, type IncrementalImportDeps } from '../packages/node-runtime/src/import/incremental-importer' +import { + streamingImport, + type ImportLogger, + type ImportStageTimings, + type StreamImportDeps, +} from '../packages/node-runtime/src/import/streaming-importer' +import { createChatLabTempDir } from './chatlab-temp.mjs' + +const scriptPath = fileURLToPath(import.meta.url) +const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') +const memberCount = 200 +const sampleTexts = [ + '今天天气不错，我们出去玩吧！', + '哈哈哈哈哈哈这也太好笑了', + '[图片]', + '好的，收到，明天见。', + 'This is a mixed language message with some English words 和中文混排。', + '周末有人一起打球吗？地点老地方，时间下午三点，人齐就开打。', +] + +interface BenchmarkParser { + formatId: 'weflow' + implementation: 'rust-native' + nativeModuleAvailable: true +} + +interface BenchmarkResult { + scenario: 'single' | 'batch' + parser: BenchmarkParser + fileCount: number + concurrency: number + inputMessages: number + inputBytes: number + durationMs: number + incrementalMessages: number + incrementalDurationMs: number + sampledPeakRssMb: number + sampledRssDeltaMb: number + databaseBytes: number + messagesWritten: number + membersWritten: number + outputSignature: string + likeSearch: { + coldMs: number + warmMedianMs: number + resultCount: number + resultSignature: string + } + timings: ImportStageTimings +} + +export interface BenchmarkParserMonitor { + logger: ImportLogger + assertRustNativeCompleted(importIndex: number): void +} + +export function resolveBenchmarkParser(status: NativeParserStatus, nativeFormatAvailable: boolean): BenchmarkParser { + if (!status.available || !nativeFormatAvailable) { + const reason = status.disabled + ? 'CHATLAB_DISABLE_NATIVE_PERF=1 disables the native parser' + : status.error + ? `native parser failed to load: ${status.error}` + : 'the native parser does not provide the WeFlow kernel' + throw new Error( + `Streaming import benchmark requires the Rust Native WeFlow parser; ${reason}. ` + + 'Run pnpm build:native before collecting a baseline.' + ) + } + + return { + formatId: 'weflow', + implementation: 'rust-native', + nativeModuleAvailable: true, + } +} + +export function createBenchmarkParserMonitor(): BenchmarkParserMonitor { + let nativeStarted = false + let fallbackDetected = false + const ignoreNonParserLog = () => undefined + + const inspectMessage = (message: string) => { + if (message.includes('[NativeParser] Parsing WeFlow export with Rust kernel')) { + nativeStarted = true + } + if (message.includes('[NativeParser]') && message.includes('falling back to TS parser')) { + fallbackDetected = true + } + } + + return { + logger: { + info: inspectMessage, + error: inspectMessage, + perf: ignoreNonParserLog, + perfDetail: ignoreNonParserLog, + summary: ignoreNonParserLog, + reset() { + nativeStarted = false + fallbackDetected = false + }, + init: ignoreNonParserLog, + getCurrentLogFile() { + return null + }, + }, + assertRustNativeCompleted(importIndex: number) { + if (fallbackDetected) { + throw new Error(`Import ${importIndex} fell back to the TypeScript WeFlow parser; benchmark result rejected`) + } + if (!nativeStarted) { + throw new Error(`Import ${importIndex} did not start the Rust Native WeFlow parser; benchmark result rejected`) + } + }, + } +} + +const zeroTimings = (): ImportStageTimings => ({ + detectionMs: 0, + preprocessingMs: 0, + databaseSetupMs: 0, + parserMs: 0, + metaWriteMs: 0, + memberWriteMs: 0, + messageWriteMs: 0, + nicknameHistoryMs: 0, + indexCreationMs: 0, + checkpointMs: 0, + sessionIndexMs: 0, + postImportHookMs: 0, + totalMs: 0, +}) + +function addTimings(target: ImportStageTimings, source: ImportStageTimings): void { + for (const key of Object.keys(target) as Array) target[key] += source[key] +} + +async function generateWeflowFixture(filePath: string, count: number, startIndex = 0): Promise { + const stream = createWriteStream(filePath, { encoding: 'utf-8' }) + const write = async (chunk: string) => { + if (!stream.write(chunk)) await once(stream, 'drain') + } + + await write( + `{"weflow":{"version":"1.0.0","exportedAt":1704164645},` + + `"session":{"wxid":"bench@chatroom","nickname":"性能测试群","displayName":"性能测试群","type":"群聊"},` + + '"avatars":{},"messages":[' + ) + + const batch: string[] = [] + for (let index = 0; index < count; index++) { + const absoluteIndex = startIndex + index + const member = absoluteIndex % memberCount + batch.push( + JSON.stringify({ + localId: absoluteIndex + 1, + createTime: 1_704_164_645 + absoluteIndex * 3, + formattedTime: '2024-01-02 03:04:05', + type: '文本消息', + localType: 1, + content: `${sampleTexts[absoluteIndex % sampleTexts.length]} #${absoluteIndex}`, + isSend: member === 0 ? 1 : 0, + senderUsername: `wxid_member_${member}`, + senderDisplayName: `成员${member}号`, + senderAvatarKey: `wxid_member_${member}`, + source: '', + }) + ) + if (batch.length >= 5000) { + await write((index + 1 > batch.length ? ',' : '') + batch.join(',')) + batch.length = 0 + } + } + if (batch.length > 0) await write((count > batch.length ? ',' : '') + batch.join(',')) + await write(']}') + stream.end() + await once(stream, 'finish') +} + +function createIncrementalDeps(dbPath: string): IncrementalImportDeps { + return { + openDatabase() { + const raw = new Database(dbPath, { nativeBinding }) + raw.pragma('journal_mode = WAL') + raw.pragma('synchronous = NORMAL') + return new BetterSqliteAdapter(raw) + }, + onProgress() { + /* benchmark excludes UI transport */ + }, + } +} + +function createImportDeps(dbPath: string, cacheDir: string, logger: ImportLogger): StreamImportDeps { + return { + openDatabase() { + const raw = new Database(dbPath, { nativeBinding }) + raw.pragma('journal_mode = WAL') + raw.pragma('synchronous = NORMAL') + raw.exec(CHAT_DB_TABLES) + return new BetterSqliteAdapter(raw) + }, + deleteDatabase() { + for (const suffix of ['', '-wal', '-shm']) { + try { + unlinkSync(dbPath + suffix) + } catch { + /* ignore cleanup failures in a disposable benchmark workspace */ + } + } + }, + onProgress() { + /* benchmark excludes UI transport */ + }, + logger, + postImportHook(db, sessionId) { + computeAndSetOverviewCache(db, sessionId, cacheDir) + }, + } +} + +export function inspectDatabase(dbPath: string): { + messages: number + members: number + signature: string +} { + const db = new Database(dbPath, { readonly: true, nativeBinding }) + const hash = createHash('sha256') + const hashRows = (label: string, query: string): number => { + hash.update(`table:${label}\n`) + let count = 0 + let chunk = '' + for (const row of db.prepare(query).raw().iterate() as Iterable) { + chunk += `${JSON.stringify(row)}\n` + count++ + if (chunk.length >= 1024 * 1024) { + hash.update(chunk) + chunk = '' + } + } + if (chunk) hash.update(chunk) + return count + } + + try { + hashRows( + 'meta', + `SELECT name, platform, type, group_id, group_avatar, owner_id, schema_version, session_gap_threshold + FROM meta ORDER BY rowid` + ) + const members = hashRows( + 'member', + `SELECT id, platform_id, account_name, group_nickname, aliases, avatar, roles + FROM member ORDER BY id` + ) + hashRows( + 'member_name_history', + `SELECT id, member_id, name_type, name, start_ts, end_ts + FROM member_name_history ORDER BY id` + ) + const messages = hashRows( + 'message', + `SELECT id, sender_id, sender_account_name, sender_group_nickname, ts, type, content, + reply_to_message_id, platform_message_id + FROM message ORDER BY id` + ) + hashRows( + 'segment', + `SELECT id, start_ts, end_ts, message_count, is_manual, summary + FROM segment ORDER BY id` + ) + hashRows( + 'message_context', + `SELECT message_id, segment_id, topic_id + FROM message_context ORDER BY message_id` + ) + const legacyFts = db.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'message_fts'").get() + if (legacyFts) throw new Error('Benchmark database unexpectedly contains message_fts') + return { messages, members, signature: hash.digest('hex') } + } finally { + db.close() + } +} + +function measureLikeSearch(dbPath: string): BenchmarkResult['likeSearch'] { + const db = new Database(dbPath, { readonly: true, nativeBinding }) + const statement = db.prepare( + `SELECT id, content + FROM message + WHERE content LIKE ? ESCAPE '\\' + ORDER BY ts DESC, id DESC + LIMIT 50` + ) + // Match only the oldest synthetic row so ORDER BY cannot stop after finding + // a few recent matches. This measures the existing LIKE path's full-scan cost. + const params = ['%#0'] + const run = () => { + const startedAt = performance.now() + const rows = statement.all(...params) as Array<{ id: number; content: string }> + return { durationMs: performance.now() - startedAt, rows } + } + + try { + const cold = run() + const warmRuns = Array.from({ length: 5 }, run).sort((left, right) => left.durationMs - right.durationMs) + const warmMedian = warmRuns[Math.floor(warmRuns.length / 2)] + return { + coldMs: cold.durationMs, + warmMedianMs: warmMedian.durationMs, + resultCount: cold.rows.length, + resultSignature: createHash('sha256').update(JSON.stringify(cold.rows)).digest('hex').slice(0, 16), + } + } finally { + db.close() + } +} + +async function runWorker(fileCount: number, messagesPerFile: number, concurrency: number): Promise { + if (typeof global.gc !== 'function') { + throw new Error('Benchmark worker requires Node.js --expose-gc') + } + + const parser = resolveBenchmarkParser(getNativeParserStatus(), isNativeFormatAvailable('weflow')) + const root = createChatLabTempDir('bench', 'streaming-import-') + const fixturePath = path.join(root, 'fixture.json') + const incrementalFixturePath = path.join(root, 'incremental.json') + const cacheDir = path.join(root, 'cache') + mkdirSync(cacheDir, { recursive: true }) + + try { + await generateWeflowFixture(fixturePath, messagesPerFile) + const incrementalMessages = fileCount === 1 ? 1000 : 0 + if (incrementalMessages > 0) { + await generateWeflowFixture(incrementalFixturePath, incrementalMessages, messagesPerFile) + } + const inputPaths = [fixturePath] + for (let index = 1; index < fileCount; index++) { + const linkedPath = path.join(root, `fixture-${index}.json`) + try { + linkSync(fixturePath, linkedPath) + } catch { + copyFileSync(fixturePath, linkedPath) + } + inputPaths.push(linkedPath) + } + + global.gc() + const rssStartMb = process.memoryUsage().rss / 1024 / 1024 + const timings = zeroTimings() + let sampledPeakRssMb = rssStartMb + const dbPaths: string[] = [] + const startedAt = performance.now() + + await runKeyedBatch( + inputPaths.map((inputPath, index) => ({ + value: { inputPath, index }, + key: `session:${index}`, + })), + { + concurrency, + async run({ inputPath, index }) { + const dbPath = path.join(root, `session-${index}.db`) + dbPaths[index] = dbPath + const parserMonitor = createBenchmarkParserMonitor() + const result = await streamingImport( + inputPath, + createImportDeps(dbPath, cacheDir, parserMonitor.logger), + { formatId: 'weflow' }, + `bench-${index}` + ) + if (!result.success || !result.diagnostics?.performance) { + throw new Error(`Import ${index} failed: ${result.error ?? 'missing performance diagnostics'}`) + } + parserMonitor.assertRustNativeCompleted(index) + addTimings(timings, result.diagnostics.performance.timings) + sampledPeakRssMb = Math.max(sampledPeakRssMb, result.diagnostics.performance.rssSampledPeakMb) + }, + } + ) + const durationMs = performance.now() - startedAt + + let incrementalDurationMs = 0 + if (incrementalMessages > 0) { + const incrementalStartedAt = performance.now() + const result = await incrementalImport('bench-0', incrementalFixturePath, createIncrementalDeps(dbPaths[0]), { + formatId: 'weflow', + }) + incrementalDurationMs = performance.now() - incrementalStartedAt + if (!result.success || result.newMessageCount !== incrementalMessages) { + throw new Error( + `Incremental import failed: expected ${incrementalMessages}, got ${result.newMessageCount} (${result.error ?? 'unknown error'})` + ) + } + } + + let messagesWritten = 0 + let membersWritten = 0 + let databaseBytes = 0 + const signatures: string[] = [] + let likeSearch: BenchmarkResult['likeSearch'] | undefined + for (const dbPath of dbPaths) { + const inspected = inspectDatabase(dbPath) + messagesWritten += inspected.messages + membersWritten += inspected.members + signatures.push(inspected.signature) + databaseBytes += statSync(dbPath).size + likeSearch ??= measureLikeSearch(dbPath) + } + + return { + scenario: fileCount === 1 ? 'single' : 'batch', + parser, + fileCount, + concurrency, + inputMessages: fileCount * messagesPerFile, + inputBytes: statSync(fixturePath).size * fileCount, + durationMs, + incrementalMessages, + incrementalDurationMs, + sampledPeakRssMb, + sampledRssDeltaMb: Math.max(0, sampledPeakRssMb - rssStartMb), + databaseBytes, + messagesWritten, + membersWritten, + outputSignature: createHash('sha256').update(signatures.sort().join(':')).digest('hex'), + likeSearch: likeSearch!, + timings, + } + } finally { + rmSync(root, { recursive: true, force: true }) + } +} + +function runIsolated(fileCount: number, messagesPerFile: number, concurrency: number): BenchmarkResult { + const child = spawnSync( + process.execPath, + [ + '--expose-gc', + ...process.execArgv, + scriptPath, + '--worker', + String(fileCount), + String(messagesPerFile), + String(concurrency), + ], + { + cwd: process.cwd(), + encoding: 'utf-8', + maxBuffer: 10 * 1024 * 1024, + env: { ...process.env }, + } + ) + if (child.status !== 0) { + throw new Error(`Benchmark worker failed (${child.status}):\n${child.stdout}\n${child.stderr}`) + } + const resultLine = child.stdout.split('\n').find((line) => line.startsWith('BENCH_RESULT ')) + if (!resultLine) throw new Error(`Benchmark worker returned no result:\n${child.stdout}\n${child.stderr}`) + return JSON.parse(resultLine.slice('BENCH_RESULT '.length)) as BenchmarkResult +} + +function formatBytes(bytes: number): string { + return `${(bytes / 1024 / 1024).toFixed(1)} MB` +} + +function printResult(result: BenchmarkResult, label: string): void { + console.log( + `${label}: parser ${result.parser.implementation} | ${result.durationMs.toFixed(0)} ms | ` + + `incremental ${result.incrementalDurationMs.toFixed(0)} ms/${result.incrementalMessages} rows | ` + + `sampled peak RSS ${result.sampledPeakRssMb.toFixed(0)} MB ` + + `(+${result.sampledRssDeltaMb.toFixed(0)} MB) | DB ${formatBytes(result.databaseBytes)} | ` + + `LIKE ${result.likeSearch.coldMs.toFixed(1)} ms cold/${result.likeSearch.warmMedianMs.toFixed(1)} ms warm | ` + + `signature ${result.outputSignature}` + ) +} + +async function main(): Promise { + const [mode, first, second, third, fourth] = process.argv.slice(2) + if (mode === '--worker') { + const result = await runWorker(Number(first), Number(second), Number(third)) + console.log(`BENCH_RESULT ${JSON.stringify(result)}`) + return + } + + let fileCount: number + let messagesPerFile: number + let runs: number + let concurrency: number + if (mode === 'single') { + fileCount = 1 + messagesPerFile = Number(first ?? 100_000) + runs = Number(second ?? 3) + concurrency = 1 + } else if (mode === 'batch') { + fileCount = Number(first ?? 100) + messagesPerFile = Number(second ?? 10_000) + runs = Number(third ?? 3) + concurrency = Number(fourth ?? 1) + } else { + throw new Error('Usage: single [runs=3] | batch [runs=3] [concurrency=1]') + } + if (![fileCount, messagesPerFile, runs, concurrency].every((value) => Number.isInteger(value) && value > 0)) { + throw new Error('All benchmark counts must be positive integers') + } + + console.log( + `Benchmarking ${fileCount} file(s), ${messagesPerFile.toLocaleString()} messages/file, concurrency ${concurrency}, ${runs} isolated run(s)` + ) + const results: BenchmarkResult[] = [] + for (let index = 0; index < runs; index++) { + const result = runIsolated(fileCount, messagesPerFile, concurrency) + results.push(result) + printResult(result, `run ${index + 1}`) + } + + const sorted = [...results].sort((left, right) => left.durationMs - right.durationMs) + const median = sorted[Math.floor(sorted.length / 2)] + const parserSignatures = new Set(results.map((result) => JSON.stringify(result.parser))) + if (parserSignatures.size !== 1) { + throw new Error('Parser implementations differ between isolated runs') + } + if (results.some((result) => result.outputSignature !== median.outputSignature)) { + throw new Error('Output signatures differ between isolated runs') + } + if ( + results.some( + (result) => + result.likeSearch.resultCount !== 1 || result.likeSearch.resultSignature !== median.likeSearch.resultSignature + ) + ) { + throw new Error('LIKE search results differ between isolated runs or do not match the expected fixture row') + } + const expectedMessages = fileCount * messagesPerFile + median.incrementalMessages + if (median.messagesWritten !== expectedMessages) { + throw new Error(`Expected ${expectedMessages} messages, got ${median.messagesWritten}`) + } + + printResult(median, 'median') + const stageRows = Object.entries(median.timings) + .filter(([stage]) => stage !== 'totalMs') + .sort(([, left], [, right]) => right - left) + console.log('Median phase totals:') + for (const [stage, durationMs] of stageRows) { + console.log(` ${stage.padEnd(22)} ${durationMs.toFixed(1).padStart(10)} ms`) + } +} + +if (path.resolve(process.argv[1] ?? '') === scriptPath) { + void main().catch((error: unknown) => { + console.error(error) + process.exitCode = 1 + }) +}