diff --git a/apps/desktop/preload/index.d.ts b/apps/desktop/preload/index.d.ts index 2a6e0995..c3df4b63 100644 --- a/apps/desktop/preload/index.d.ts +++ b/apps/desktop/preload/index.d.ts @@ -59,8 +59,9 @@ interface ImportDiagnostics { messageBatchCount: number messageTransactionCount: number rssStartMb: number - rssPeakMb: number - rssDeltaMb: number + /** Highest RSS observed at import stage and parser batch boundaries; not a continuous process peak. */ + rssSampledPeakMb: number + rssSampledDeltaMb: number } } diff --git a/packages/node-runtime/src/import/auto-importer.test.ts b/packages/node-runtime/src/import/auto-importer.test.ts index 2f675d79..98e7c928 100644 --- a/packages/node-runtime/src/import/auto-importer.test.ts +++ b/packages/node-runtime/src/import/auto-importer.test.ts @@ -78,8 +78,8 @@ function createDeps(options?: { messageBatchCount: 0, messageTransactionCount: 0, rssStartMb: 0, - rssPeakMb: 0, - rssDeltaMb: 0, + rssSampledPeakMb: 0, + rssSampledDeltaMb: 0, }, }, } diff --git a/packages/node-runtime/src/import/streaming-importer.test.ts b/packages/node-runtime/src/import/streaming-importer.test.ts index b0222e00..b54eac05 100644 --- a/packages/node-runtime/src/import/streaming-importer.test.ts +++ b/packages/node-runtime/src/import/streaming-importer.test.ts @@ -89,8 +89,8 @@ function assertPerformanceDiagnostics(result: StreamImportResult): void { assert.ok(diagnostics.timings.totalMs >= diagnostics.timings.messageWriteMs) assert.ok(diagnostics.messageBatchCount > 0) assert.ok(diagnostics.messageTransactionCount > 0) - assert.ok(diagnostics.rssPeakMb >= diagnostics.rssStartMb) - assert.ok(diagnostics.rssDeltaMb >= 0) + assert.ok(diagnostics.rssSampledPeakMb >= diagnostics.rssStartMb) + assert.ok(diagnostics.rssSampledDeltaMb >= 0) } function writeChunkedShuakamiQqExport(root: string): string { diff --git a/packages/node-runtime/src/import/streaming-importer.ts b/packages/node-runtime/src/import/streaming-importer.ts index e4e5e22f..3cb48d44 100644 --- a/packages/node-runtime/src/import/streaming-importer.ts +++ b/packages/node-runtime/src/import/streaming-importer.ts @@ -71,8 +71,9 @@ export interface ImportPerformanceDiagnostics { messageBatchCount: number messageTransactionCount: number rssStartMb: number - rssPeakMb: number - rssDeltaMb: number + /** Highest RSS observed at import stage and parser batch boundaries; not a continuous process peak. */ + rssSampledPeakMb: number + rssSampledDeltaMb: number } export interface StreamImportResult { @@ -843,8 +844,8 @@ async function streamImportSingle( messageBatchCount: callbackStats.onMessageBatchCalls, messageTransactionCount, rssStartMb: rssStartBytes / bytesPerMegabyte, - rssPeakMb: rssPeakBytes / bytesPerMegabyte, - rssDeltaMb: (rssPeakBytes - rssStartBytes) / bytesPerMegabyte, + rssSampledPeakMb: rssPeakBytes / bytesPerMegabyte, + rssSampledDeltaMb: (rssPeakBytes - rssStartBytes) / bytesPerMegabyte, }, } diff --git a/scripts/bench-streaming-import.mts b/scripts/bench-streaming-import.mts index f1e2b4b7..d848466a 100644 --- a/scripts/bench-streaming-import.mts +++ b/scripts/bench-streaming-import.mts @@ -17,10 +17,12 @@ 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 { streamingImport, + type ImportLogger, type ImportStageTimings, type StreamImportDeps, } from '../packages/node-runtime/src/import/streaming-importer' @@ -38,14 +40,21 @@ const sampleTexts = [ '周末有人一起打球吗？地点老地方，时间下午三点，人齐就开打。', ] +interface BenchmarkParser { + formatId: 'weflow' + implementation: 'rust-native' + nativeModuleAvailable: true +} + interface BenchmarkResult { scenario: 'single' | 'batch' + parser: BenchmarkParser fileCount: number inputMessages: number inputBytes: number durationMs: number - peakRssMb: number - rssDeltaMb: number + sampledPeakRssMb: number + sampledRssDeltaMb: number databaseBytes: number messagesWritten: number membersWritten: number @@ -53,6 +62,72 @@ interface BenchmarkResult { 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, @@ -115,7 +190,7 @@ async function generateWeflowFixture(filePath: string, count: number): Promise { + 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` ) - .get() as Record - db.close() - const signature = createHash('sha256').update(JSON.stringify(row)).digest('hex').slice(0, 16) - return { messages: row.messages, members: row.members, signature } + 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` + ) + hashRows('message_fts', 'SELECT rowid FROM message_fts ORDER BY rowid') + return { messages, members, signature: hash.digest('hex') } + } finally { + db.close() + } } async function runWorker(fileCount: number, messagesPerFile: 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 cacheDir = path.join(root, 'cache') @@ -183,27 +304,29 @@ async function runWorker(fileCount: number, messagesPerFile: number): Promise { 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') } @@ -324,4 +453,9 @@ async function main(): Promise { } } -await main() +if (path.resolve(process.argv[1] ?? '') === scriptPath) { + void main().catch((error: unknown) => { + console.error(error) + process.exitCode = 1 + }) +} diff --git a/scripts/bench-streaming-import.test.ts b/scripts/bench-streaming-import.test.ts new file mode 100644 index 00000000..68d17ce2 --- /dev/null +++ b/scripts/bench-streaming-import.test.ts @@ -0,0 +1,100 @@ +import assert from 'node:assert/strict' +import fs from 'node:fs' +import path from 'node:path' +import test from 'node:test' +import Database from 'better-sqlite3' +import { CHAT_DB_TABLES } from '@openchatlab/core' +import { BetterSqliteAdapter } from '../packages/node-runtime/src/better-sqlite3-adapter' +import { buildFtsIndex } from '../packages/node-runtime/src/fts' +import { createChatLabTempDir } from './chatlab-temp.mjs' +import { createBenchmarkParserMonitor, inspectDatabase, resolveBenchmarkParser } from './bench-streaming-import.mts' + +const nativeBinding = path.resolve('apps/cli/native/better_sqlite3.node') + +test('benchmark requires and records the Rust Native WeFlow parser', () => { + assert.deepEqual(resolveBenchmarkParser({ available: true, disabled: false }, true), { + formatId: 'weflow', + implementation: 'rust-native', + nativeModuleAvailable: true, + }) + assert.throws( + () => resolveBenchmarkParser({ available: false, disabled: true }, false), + /CHATLAB_DISABLE_NATIVE_PERF=1/ + ) + assert.throws( + () => resolveBenchmarkParser({ available: true, disabled: false }, false), + /does not provide the WeFlow kernel/ + ) +}) + +test('benchmark rejects imports that do not complete with the Rust Native WeFlow parser', () => { + const missingNative = createBenchmarkParserMonitor() + assert.throws(() => missingNative.assertRustNativeCompleted(0), /did not start the Rust Native WeFlow parser/) + + const native = createBenchmarkParserMonitor() + native.logger.info('[NativeParser] Parsing WeFlow export with Rust kernel, size: 1.00 MB') + assert.doesNotThrow(() => native.assertRustNativeCompleted(1)) + + native.logger.info('[NativeParser] Rust parse failed, falling back to TS parser: invalid fixture') + assert.throws(() => native.assertRustNativeCompleted(1), /fell back to the TypeScript WeFlow parser/) +}) + +function createFixtureDatabase(dbPath: string): void { + const raw = new Database(dbPath, { nativeBinding }) + raw.exec(CHAT_DB_TABLES) + raw + .prepare( + `INSERT INTO meta + (name, platform, type, imported_at, group_id, group_avatar, owner_id) + VALUES (?, ?, ?, ?, ?, ?, ?)` + ) + .run('Signature Test', 'WeChat', 'group', 1_704_164_645, 'group-1', null, 'member-1') + const insertMember = raw.prepare( + `INSERT INTO member + (platform_id, account_name, group_nickname, aliases, avatar, roles) + VALUES (?, ?, ?, ?, ?, ?)` + ) + insertMember.run('member-1', 'Alice', 'Alice', '[]', null, '[]') + insertMember.run('member-2', 'Bob', 'Bob', '[]', null, '[]') + const insertMessage = raw.prepare( + `INSERT INTO message + (sender_id, sender_account_name, sender_group_nickname, ts, type, content) + VALUES (?, ?, ?, ?, ?, ?)` + ) + insertMessage.run(1, 'Alice', 'Alice', 1_704_164_645, 0, 'alpha') + insertMessage.run(2, 'Bob', 'Bob', 1_704_164_648, 0, 'bravo') + buildFtsIndex(new BetterSqliteAdapter(raw)) + raw.close() +} + +test('canonical database signature changes for equal-length content and sender permutations', () => { + const root = createChatLabTempDir('tests', 'bench-signature-') + const dbPath = path.join(root, 'session.db') + try { + createFixtureDatabase(dbPath) + const baseline = inspectDatabase(dbPath) + assert.equal(baseline.messages, 2) + assert.equal(baseline.members, 2) + + const raw = new Database(dbPath, { nativeBinding }) + raw.prepare('UPDATE message SET content = ? WHERE id = 1').run('omega') + raw.close() + const changedContent = inspectDatabase(dbPath) + assert.notEqual(changedContent.signature, baseline.signature) + + const senderDb = new Database(dbPath, { nativeBinding }) + senderDb.prepare('UPDATE message SET content = ? WHERE id = 1').run('alpha') + senderDb.exec( + `UPDATE message + SET sender_id = CASE id + WHEN 1 THEN 2 + WHEN 2 THEN 1 + END` + ) + senderDb.close() + const changedSenders = inspectDatabase(dbPath) + assert.notEqual(changedSenders.signature, baseline.signature) + } finally { + fs.rmSync(root, { recursive: true, force: true }) + } +}) diff --git a/src/services/import/types.ts b/src/services/import/types.ts index 4f31f48a..007e759c 100644 --- a/src/services/import/types.ts +++ b/src/services/import/types.ts @@ -88,8 +88,9 @@ export interface ImportPerformanceDiagnosticsInfo { messageBatchCount: number messageTransactionCount: number rssStartMb: number - rssPeakMb: number - rssDeltaMb: number + /** Highest RSS observed at import stage and parser batch boundaries; not a continuous process peak. */ + rssSampledPeakMb: number + rssSampledDeltaMb: number } // ==================== 格式信息 ====================