diff --git a/packages/node-runtime/src/import/incremental-importer.test.ts b/packages/node-runtime/src/import/incremental-importer.test.ts index daab7088..a18355e9 100644 --- a/packages/node-runtime/src/import/incremental-importer.test.ts +++ b/packages/node-runtime/src/import/incremental-importer.test.ts @@ -93,6 +93,53 @@ function writeChatLabJson(filePath: string): void { ) } +interface DedupFixtureMessage { + timestamp?: number + type?: number + content?: string | null + replyToMessageId?: string + platformMessageId?: string +} + +function writeDedupFixture(filePath: string, messages: DedupFixtureMessage[]): void { + fs.writeFileSync( + filePath, + JSON.stringify({ + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Dedup Fixture', platform: 'wechat', type: 'private' }, + members: [{ platformId: 'wxid_alice', accountName: 'Alice' }], + messages: messages.map((message) => ({ + sender: 'wxid_alice', + accountName: 'Alice', + timestamp: message.timestamp ?? 1780330832, + type: message.type ?? 0, + content: Object.hasOwn(message, 'content') ? message.content : 'same message', + replyToMessageId: message.replyToMessageId, + platformMessageId: message.platformMessageId, + })), + }), + 'utf8' + ) +} + +function seedExistingMessage(dbPath: string, message: DedupFixtureMessage): void { + const db = openBetterSqliteDatabase(dbPath, { nativeBinding }) + db.prepare('INSERT INTO member (platform_id, account_name) VALUES (?, ?)').run('wxid_alice', 'Alice') + const member = db.prepare('SELECT id FROM member WHERE platform_id = ?').get('wxid_alice') as { id: number } + db.prepare( + `INSERT INTO message (sender_id, ts, type, content, reply_to_message_id, platform_message_id) + VALUES (?, ?, ?, ?, ?, ?)` + ).run( + member.id, + message.timestamp ?? 1780330832, + message.type ?? 0, + Object.hasOwn(message, 'content') ? (message.content ?? null) : 'same message', + message.replyToMessageId ?? null, + message.platformMessageId ?? null + ) + db.close() +} + function seedSessionDb(dbPath: string): void { const db = openBetterSqliteDatabase(dbPath, { nativeBinding }) db.exec(CHAT_DB_SCHEMA) @@ -355,3 +402,159 @@ test('deduplicates an ID-bearing copy of an existing fallback-only message', asy readonlyDb.close() assert.equal(row.count, 1) }) + +test('preserves canonical dedup semantics while checking existing database candidates', async (t) => { + const scenarios: Array<{ + name: string + existing: DedupFixtureMessage + incoming: DedupFixtureMessage[] + expectedNew: number + expectedDuplicates: number + expectedPlatformIds: Array + }> = [ + { + name: 'same platform ID wins even when content changed', + existing: { platformMessageId: 'msg-1', content: 'original' }, + incoming: [{ platformMessageId: 'msg-1', content: 'edited' }], + expectedNew: 0, + expectedDuplicates: 1, + expectedPlatformIds: ['msg-1'], + }, + { + name: 'fallback-only input matches an existing ID-bearing message', + existing: { platformMessageId: 'msg-1' }, + incoming: [{}], + expectedNew: 0, + expectedDuplicates: 1, + expectedPlatformIds: ['msg-1'], + }, + { + name: 'different stable IDs survive an identical fallback fingerprint', + existing: { platformMessageId: 'msg-1' }, + incoming: [{ platformMessageId: 'msg-2' }], + expectedNew: 1, + expectedDuplicates: 0, + expectedPlatformIds: ['msg-1', 'msg-2'], + }, + { + name: 'one fallback-only row bridges only one stable ID', + existing: {}, + incoming: [{ platformMessageId: 'msg-1' }, { platformMessageId: 'msg-2' }], + expectedNew: 1, + expectedDuplicates: 1, + expectedPlatformIds: [null, 'msg-2'], + }, + { + name: 'a bridged stable ID remains duplicate when repeated', + existing: {}, + incoming: [{ platformMessageId: 'msg-1' }, { platformMessageId: 'msg-1' }, { platformMessageId: 'msg-2' }], + expectedNew: 1, + expectedDuplicates: 2, + expectedPlatformIds: [null, 'msg-2'], + }, + { + name: 'an empty stored platform ID behaves as fallback-only', + existing: { platformMessageId: '' }, + incoming: [{ platformMessageId: 'msg-1' }], + expectedNew: 0, + expectedDuplicates: 1, + expectedPlatformIds: [''], + }, + { + name: 'empty and null content share the canonical fallback fingerprint', + existing: { content: null }, + incoming: [{ content: '' }], + expectedNew: 0, + expectedDuplicates: 1, + expectedPlatformIds: [null], + }, + { + name: 'reply target remains part of the fallback fingerprint', + existing: { replyToMessageId: 'reply-1' }, + incoming: [{ replyToMessageId: 'reply-2' }], + expectedNew: 1, + expectedDuplicates: 0, + expectedPlatformIds: [null, null], + }, + ] + + for (const scenario of scenarios) { + await t.test(scenario.name, async (st) => { + const tempDir = makeTempDir() + st.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const dbPath = path.join(tempDir, 'session.db') + const filePath = path.join(tempDir, 'incoming.json') + seedSessionDb(dbPath) + seedExistingMessage(dbPath, scenario.existing) + writeDedupFixture(filePath, scenario.incoming) + + const deps = createDeps(dbPath) + const analysis = await analyzeIncrementalImport('session', filePath, deps) + assert.equal(analysis.newMessageCount, scenario.expectedNew) + assert.equal(analysis.duplicateCount, scenario.expectedDuplicates) + + const result = await incrementalImport('session', filePath, deps) + assert.equal(result.success, true) + assert.equal(result.newMessageCount, scenario.expectedNew) + assert.equal(result.batch?.duplicateCount, scenario.expectedDuplicates) + + const db = openBetterSqliteDatabase(dbPath, { readonly: true, nativeBinding }) + const rows = db.prepare('SELECT platform_message_id FROM message ORDER BY id').all() as Array<{ + platform_message_id: string | null + }> + db.close() + assert.deepEqual( + rows.map((row) => row.platform_message_id), + scenario.expectedPlatformIds + ) + }) + } +}) + +test('deduplicates repeated input across parser message batches', async (t) => { + const tempDir = makeTempDir() + t.after(() => fs.rmSync(tempDir, { recursive: true, force: true })) + + const dbPath = path.join(tempDir, 'session.db') + const filePath = path.join(tempDir, 'cross-batch.jsonl') + seedSessionDb(dbPath) + + const lines: unknown[] = [ + { + _type: 'header', + chatlab: { version: '0.0.2', exportedAt: 1780330900 }, + meta: { name: 'Cross Batch', platform: 'wechat', type: 'private' }, + }, + { _type: 'member', platformId: 'wxid_alice', accountName: 'Alice' }, + ] + for (let index = 0; index < 5000; index++) { + lines.push({ + _type: 'message', + sender: 'wxid_alice', + timestamp: 1780330832 + index, + type: 0, + content: `message-${index}`, + platformMessageId: `msg-${index}`, + }) + } + lines.push({ + _type: 'message', + sender: 'wxid_alice', + timestamp: 1780330832, + type: 0, + content: 'message-0', + platformMessageId: 'msg-0', + }) + fs.writeFileSync(filePath, `${lines.map((line) => JSON.stringify(line)).join('\n')}\n`, 'utf8') + + const deps = createDeps(dbPath) + const analysis = await analyzeIncrementalImport('session', filePath, deps) + assert.equal(analysis.newMessageCount, 5000) + assert.equal(analysis.duplicateCount, 1) + + const result = await incrementalImport('session', filePath, deps) + assert.equal(result.success, true) + assert.equal(result.newMessageCount, 5000) + assert.equal(result.batch?.duplicateCount, 1) +}) diff --git a/packages/node-runtime/src/import/incremental-importer.ts b/packages/node-runtime/src/import/incremental-importer.ts index 9db62145..2735f816 100644 --- a/packages/node-runtime/src/import/incremental-importer.ts +++ b/packages/node-runtime/src/import/incremental-importer.ts @@ -8,10 +8,22 @@ * Callers provide DatabaseAdapter + progress callback via dependency injection. */ -import type { DatabaseAdapter } from '@openchatlab/core' +import type { DatabaseAdapter, PreparedStatement } from '@openchatlab/core' import { generateSessionIndex, generateIncrementalSessionIndex, getSessionIndexStats } from '@openchatlab/core' -import { streamParseFile, detectFormat, getFormatFeatureById, type ParseProgress } from '@openchatlab/parser' -import { generateFallbackMessageKey, registerMessageAndCheckDuplicate } from './message-deduplicator' +import { + streamParseFile, + detectFormat, + getFormatFeatureById, + type ParseProgress, + type StreamParseCallbacks, +} from '@openchatlab/parser' +import { + createMessageDedupState, + generateFallbackMessageKey, + registerMessageAndCheckDuplicate, + type DedupMessage, + type MessageDedupState, +} from './message-deduplicator' import { normalizeSystemMemberName, SYSTEM_MEMBER_NAME, type ImportProgressCallback } from './streaming-importer' // ==================== Public interfaces ==================== @@ -66,76 +78,159 @@ export interface IncrementalImportDeps { /** Open existing session DB for read-only access (analyze) or read-write (import). */ openDatabase(sessionId: string, readonly?: boolean): DatabaseAdapter onProgress: ImportProgressCallback + /** Optional parser diagnostics hook (e.g. native parser fallback monitoring). */ + onParserLog?: NonNullable /** Optional hook after incremental import (e.g. update overview cache). */ postImportHook?: (db: DatabaseAdapter, sessionId: string) => void | Promise } // ==================== Internal helpers ==================== -function loadExistingDedup(db: DatabaseAdapter): { - existingPlatformMsgIds: Set - existingKeys: Set - existingFallbackOnlyKeys: Set -} { - const existingPlatformMsgIds = new Set() - const existingKeys = new Set() - const existingFallbackOnlyKeys = new Set() +const PLATFORM_ID_QUERY_CHUNK_SIZE = 500 - const pmidRows = db - .prepare('SELECT platform_message_id FROM message WHERE platform_message_id IS NOT NULL') - .all() as Array<{ platform_message_id: string }> - for (const row of pmidRows) { - existingPlatformMsgIds.add(row.platform_message_id) +interface ExistingMessageCandidate { + type: number + content: string | null + reply_to_message_id: string | null + platform_message_id: string | null +} + +interface ExistingMessageLookup { + db: DatabaseAdapter + maxMessageId: number + hasFallbackOnlyMessages: boolean + consumedFallbackOnlyKeys: Set + bridgedPlatformMessageIds: Set + candidateStatement: PreparedStatement + platformIdStatements: Map +} + +type CandidateCache = Map + +function createExistingMessageLookup(db: DatabaseAdapter): ExistingMessageLookup { + const maxMessageId = + (db.prepare('SELECT MAX(id) AS max_id FROM message').get() as { max_id: number | null } | undefined)?.max_id ?? 0 + const hasFallbackOnlyMessages = Boolean( + db + .prepare( + `SELECT 1 FROM message + WHERE id <= ? AND (platform_message_id IS NULL OR platform_message_id = '') + LIMIT 1` + ) + .get(maxMessageId) + ) + + return { + db, + maxMessageId, + hasFallbackOnlyMessages, + consumedFallbackOnlyKeys: new Set(), + bridgedPlatformMessageIds: new Set(), + candidateStatement: db.prepare( + `SELECT msg.type, msg.content, msg.reply_to_message_id, msg.platform_message_id + FROM member m + JOIN message msg ON msg.sender_id = m.id + WHERE m.platform_id = ? AND msg.ts = ? AND msg.id <= ?` + ), + platformIdStatements: new Map(), + } +} + +function findExistingPlatformMessageIds( + lookup: ExistingMessageLookup, + platformMessageIds: Iterable +): Set { + const uniqueIds = [...new Set([...platformMessageIds].filter((id): id is string => Boolean(id)))] + const existingIds = new Set() + + for (let start = 0; start < uniqueIds.length; start += PLATFORM_ID_QUERY_CHUNK_SIZE) { + const chunk = uniqueIds.slice(start, start + PLATFORM_ID_QUERY_CHUNK_SIZE) + let statement = lookup.platformIdStatements.get(chunk.length) + if (!statement) { + const placeholders = chunk.map(() => '?').join(', ') + statement = lookup.db.prepare( + `SELECT platform_message_id + FROM message + WHERE id <= ? AND platform_message_id IN (${placeholders})` + ) + lookup.platformIdStatements.set(chunk.length, statement) + } + const rows = statement.all(lookup.maxMessageId, ...chunk) as Array<{ platform_message_id: string }> + for (const row of rows) existingIds.add(row.platform_message_id) } - const hashRows = db - .prepare( - `SELECT ts, m.platform_id as sender_platform_id, msg.type, content, reply_to_message_id, - platform_message_id - FROM message msg - JOIN member m ON msg.sender_id = m.id` + return existingIds +} + +function getExistingCandidates( + lookup: ExistingMessageLookup, + message: DedupMessage, + cache: CandidateCache +): ExistingMessageCandidate[] { + const cacheKey = JSON.stringify([message.senderPlatformId, message.timestamp]) + const cached = cache.get(cacheKey) + if (cached) return cached + + const rows = lookup.candidateStatement.all( + message.senderPlatformId, + message.timestamp, + lookup.maxMessageId + ) as unknown as ExistingMessageCandidate[] + cache.set(cacheKey, rows) + return rows +} + +function candidateFallbackKey(message: DedupMessage, candidate: ExistingMessageCandidate): string { + return generateFallbackMessageKey({ + timestamp: message.timestamp, + senderPlatformId: message.senderPlatformId, + type: candidate.type, + content: candidate.content, + replyToMessageId: candidate.reply_to_message_id ?? undefined, + }) +} + +function isExistingMessageDuplicate( + lookup: ExistingMessageLookup, + message: DedupMessage, + existingBatchPlatformIds: Set, + candidateCache: CandidateCache +): boolean { + if ( + message.platformMessageId && + (existingBatchPlatformIds.has(message.platformMessageId) || + lookup.bridgedPlatformMessageIds.has(message.platformMessageId)) + ) { + return true + } + + const fallbackKey = generateFallbackMessageKey(message) + if (message.platformMessageId) { + if (!lookup.hasFallbackOnlyMessages || lookup.consumedFallbackOnlyKeys.has(fallbackKey)) return false + const matchesFallbackOnly = getExistingCandidates(lookup, message, candidateCache).some( + (candidate) => !candidate.platform_message_id && candidateFallbackKey(message, candidate) === fallbackKey ) - .all() as Array<{ - ts: number - sender_platform_id: string - type: number - content: string | null - reply_to_message_id: string | null - platform_message_id: string | null - }> - for (const row of hashRows) { - const key = generateFallbackMessageKey({ - timestamp: row.ts, - senderPlatformId: row.sender_platform_id, - type: row.type, - content: row.content, - replyToMessageId: row.reply_to_message_id ?? undefined, - }) - existingKeys.add(key) - if (!row.platform_message_id) existingFallbackOnlyKeys.add(key) + if (matchesFallbackOnly) { + lookup.consumedFallbackOnlyKeys.add(fallbackKey) + lookup.bridgedPlatformMessageIds.add(message.platformMessageId) + } + return matchesFallbackOnly } - return { existingPlatformMsgIds, existingKeys, existingFallbackOnlyKeys } + return getExistingCandidates(lookup, message, candidateCache).some( + (candidate) => candidateFallbackKey(message, candidate) === fallbackKey + ) } function isDuplicate( - msg: { - platformMessageId?: string - timestamp: number - senderPlatformId: string - type: number - content: string | null - replyToMessageId?: string - }, - existingPlatformMsgIds: Set, - existingKeys: Set, - existingFallbackOnlyKeys: Set + lookup: ExistingMessageLookup, + incomingState: MessageDedupState, + message: DedupMessage, + existingBatchPlatformIds: Set, + candidateCache: CandidateCache ): boolean { - return registerMessageAndCheckDuplicate(msg, { - platformMessageIds: existingPlatformMsgIds, - fallbackKeys: existingKeys, - fallbackOnlyKeys: existingFallbackOnlyKeys, - }) + if (isExistingMessageDuplicate(lookup, message, existingBatchPlatformIds, candidateCache)) return true + return registerMessageAndCheckDuplicate(message, incomingState) } export function normalizeImportTimestamp(timestamp: unknown): number | null { @@ -163,41 +258,60 @@ export async function analyzeIncrementalImport( return { error: 'error.session_not_found', newMessageCount: 0, duplicateCount: 0, totalInFile: 0 } } - const { existingPlatformMsgIds, existingKeys, existingFallbackOnlyKeys } = loadExistingDedup(db) - db.close() - let totalInFile = 0 let newMessageCount = 0 let duplicateCount = 0 let platform = formatFeature.platform - await streamParseFile( - filePath, - { - formatOptions: options?.chatIndex === undefined ? undefined : { chatIndex: options.chatIndex }, - onMeta: (meta) => { - platform = meta.platform - }, - onMembers: () => {}, - onProgress: (progress: ParseProgress) => { - deps.onProgress(progress) - }, - onMessageBatch: (batch) => { - for (const msg of batch) { - totalInFile++ - const timestamp = normalizeImportTimestamp(msg.timestamp) - if (timestamp === null) continue + try { + const existingLookup = createExistingMessageLookup(db) + const incomingState = createMessageDedupState() - if (isDuplicate({ ...msg, timestamp }, existingPlatformMsgIds, existingKeys, existingFallbackOnlyKeys)) { - duplicateCount++ - } else { - newMessageCount++ + await streamParseFile( + filePath, + { + formatOptions: options?.chatIndex === undefined ? undefined : { chatIndex: options.chatIndex }, + onMeta: (meta) => { + platform = meta.platform + }, + onMembers: () => {}, + onProgress: (progress: ParseProgress) => { + deps.onProgress(progress) + }, + onLog: deps.onParserLog, + onMessageBatch: (batch) => { + const existingBatchPlatformIds = findExistingPlatformMessageIds( + existingLookup, + batch.map((message) => message.platformMessageId) + ) + const candidateCache: CandidateCache = new Map() + + for (const msg of batch) { + totalInFile++ + const timestamp = normalizeImportTimestamp(msg.timestamp) + if (timestamp === null) continue + + if ( + isDuplicate( + existingLookup, + incomingState, + { ...msg, timestamp }, + existingBatchPlatformIds, + candidateCache + ) + ) { + duplicateCount++ + } else { + newMessageCount++ + } } - } + }, }, - }, - options?.formatId - ) + options?.formatId + ) + } finally { + db.close() + } return { newMessageCount, duplicateCount, totalInFile, platform } } @@ -226,7 +340,8 @@ export async function incrementalImport( const memberUpdateMode = options?.memberUpdateMode ?? 'upsert' try { - const { existingPlatformMsgIds, existingKeys, existingFallbackOnlyKeys } = loadExistingDedup(db) + const existingLookup = createExistingMessageLookup(db) + const incomingState = createMessageDedupState() const memberIdMap = new Map() const existingMembers = db.prepare('SELECT id, platform_id FROM member').all() as Array<{ @@ -336,7 +451,14 @@ export async function incrementalImport( onProgress: (progress: ParseProgress) => { deps.onProgress(progress) }, + onLog: deps.onParserLog, onMessageBatch: (batch) => { + const existingBatchPlatformIds = findExistingPlatformMessageIds( + existingLookup, + batch.map((message) => message.platformMessageId) + ) + const candidateCache: CandidateCache = new Map() + for (const msg of batch) { processedCount++ @@ -354,7 +476,15 @@ export async function incrementalImport( continue } - if (isDuplicate({ ...msg, timestamp }, existingPlatformMsgIds, existingKeys, existingFallbackOnlyKeys)) { + if ( + isDuplicate( + existingLookup, + incomingState, + { ...msg, timestamp }, + existingBatchPlatformIds, + candidateCache + ) + ) { duplicateCount++ continue } diff --git a/scripts/bench-streaming-import.mts b/scripts/bench-streaming-import.mts index ab948555..a6477d41 100644 --- a/scripts/bench-streaming-import.mts +++ b/scripts/bench-streaming-import.mts @@ -5,7 +5,7 @@ * 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 single 100000 [runs=3] [delta|full] * 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 @@ -24,7 +24,11 @@ import { getNativeParserStatus, isNativeFormatAvailable, type NativeParserStatus 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 { + analyzeIncrementalImport, + incrementalImport, + type IncrementalImportDeps, +} from '../packages/node-runtime/src/import/incremental-importer' import { streamingImport, type ImportLogger, @@ -59,8 +63,14 @@ interface BenchmarkResult { inputMessages: number inputBytes: number durationMs: number + incrementalMode: 'delta' | 'full' | null + incrementalInputMessages: number incrementalMessages: number + incrementalDuplicateMessages: number + incrementalAnalyzeDurationMs: number + incrementalAnalyzeRssDeltaMb: number incrementalDurationMs: number + incrementalImportRssDeltaMb: number sampledPeakRssMb: number sampledRssDeltaMb: number databaseBytes: number @@ -204,7 +214,11 @@ async function generateWeflowFixture(filePath: string, count: number, startIndex await once(stream, 'finish') } -function createIncrementalDeps(dbPath: string): IncrementalImportDeps { +function createIncrementalDeps( + dbPath: string, + onSampleRss: () => void, + onParserLog: NonNullable +): IncrementalImportDeps { return { openDatabase() { const raw = new Database(dbPath, { nativeBinding }) @@ -213,8 +227,9 @@ function createIncrementalDeps(dbPath: string): IncrementalImportDeps { return new BetterSqliteAdapter(raw) }, onProgress() { - /* benchmark excludes UI transport */ + onSampleRss() }, + onParserLog, } } @@ -342,7 +357,12 @@ function measureLikeSearch(dbPath: string): BenchmarkResult['likeSearch'] { } } -async function runWorker(fileCount: number, messagesPerFile: number, concurrency: number): Promise { +async function runWorker( + fileCount: number, + messagesPerFile: number, + concurrency: number, + incrementalMode: 'delta' | 'full' | null +): Promise { if (typeof global.gc !== 'function') { throw new Error('Benchmark worker requires Node.js --expose-gc') } @@ -357,8 +377,19 @@ async function runWorker(fileCount: number, messagesPerFile: number, concurrency try { await generateWeflowFixture(fixturePath, messagesPerFile) const incrementalMessages = fileCount === 1 ? 1000 : 0 + const incrementalInputMessages = + incrementalMessages === 0 + ? 0 + : incrementalMode === 'full' + ? messagesPerFile + incrementalMessages + : incrementalMessages + const incrementalDuplicateMessages = incrementalMode === 'full' ? messagesPerFile : 0 if (incrementalMessages > 0) { - await generateWeflowFixture(incrementalFixturePath, incrementalMessages, messagesPerFile) + if (incrementalMode === 'full') { + await generateWeflowFixture(incrementalFixturePath, incrementalInputMessages) + } else { + await generateWeflowFixture(incrementalFixturePath, incrementalMessages, messagesPerFile) + } } const inputPaths = [fixturePath] for (let index = 1; index < fileCount; index++) { @@ -406,16 +437,62 @@ async function runWorker(fileCount: number, messagesPerFile: number, concurrency ) const durationMs = performance.now() - startedAt + let incrementalAnalyzeDurationMs = 0 + let incrementalAnalyzeRssDeltaMb = 0 let incrementalDurationMs = 0 + let incrementalImportRssDeltaMb = 0 if (incrementalMessages > 0) { + global.gc() + let rssStartMb = process.memoryUsage().rss / 1024 / 1024 + let rssPeakMb = rssStartMb + const sampleIncrementalRss = () => { + rssPeakMb = Math.max(rssPeakMb, process.memoryUsage().rss / 1024 / 1024) + } + const incrementalParserMonitor = createBenchmarkParserMonitor() + const incrementalDeps = createIncrementalDeps(dbPaths[0], sampleIncrementalRss, (_level, message) => { + incrementalParserMonitor.logger.info(message) + }) + + incrementalParserMonitor.logger.reset() + const analyzeStartedAt = performance.now() + const analysis = await analyzeIncrementalImport('bench-0', incrementalFixturePath, incrementalDeps, { + formatId: 'weflow', + }) + incrementalAnalyzeDurationMs = performance.now() - analyzeStartedAt + incrementalParserMonitor.assertRustNativeCompleted(0) + sampleIncrementalRss() + incrementalAnalyzeRssDeltaMb = Math.max(0, rssPeakMb - rssStartMb) + if ( + analysis.newMessageCount !== incrementalMessages || + analysis.duplicateCount !== incrementalDuplicateMessages || + analysis.totalInFile !== incrementalInputMessages + ) { + throw new Error( + `Incremental analysis mismatch: expected ${incrementalMessages} new/${incrementalDuplicateMessages} duplicate/${incrementalInputMessages} total, ` + + `got ${analysis.newMessageCount}/${analysis.duplicateCount}/${analysis.totalInFile}` + ) + } + + global.gc() + rssStartMb = process.memoryUsage().rss / 1024 / 1024 + rssPeakMb = rssStartMb + incrementalParserMonitor.logger.reset() const incrementalStartedAt = performance.now() - const result = await incrementalImport('bench-0', incrementalFixturePath, createIncrementalDeps(dbPaths[0]), { + const result = await incrementalImport('bench-0', incrementalFixturePath, incrementalDeps, { formatId: 'weflow', }) incrementalDurationMs = performance.now() - incrementalStartedAt - if (!result.success || result.newMessageCount !== incrementalMessages) { + incrementalParserMonitor.assertRustNativeCompleted(0) + sampleIncrementalRss() + incrementalImportRssDeltaMb = Math.max(0, rssPeakMb - rssStartMb) + if ( + !result.success || + result.newMessageCount !== incrementalMessages || + result.batch?.duplicateCount !== incrementalDuplicateMessages + ) { throw new Error( - `Incremental import failed: expected ${incrementalMessages}, got ${result.newMessageCount} (${result.error ?? 'unknown error'})` + `Incremental import failed: expected ${incrementalMessages} new/${incrementalDuplicateMessages} duplicate, ` + + `got ${result.newMessageCount}/${result.batch?.duplicateCount ?? 'missing'} (${result.error ?? 'unknown error'})` ) } } @@ -442,8 +519,14 @@ async function runWorker(fileCount: number, messagesPerFile: number, concurrency inputMessages: fileCount * messagesPerFile, inputBytes: statSync(fixturePath).size * fileCount, durationMs, + incrementalMode, + incrementalInputMessages, incrementalMessages, + incrementalDuplicateMessages, + incrementalAnalyzeDurationMs, + incrementalAnalyzeRssDeltaMb, incrementalDurationMs, + incrementalImportRssDeltaMb, sampledPeakRssMb, sampledRssDeltaMb: Math.max(0, sampledPeakRssMb - rssStartMb), databaseBytes, @@ -458,7 +541,12 @@ async function runWorker(fileCount: number, messagesPerFile: number, concurrency } } -function runIsolated(fileCount: number, messagesPerFile: number, concurrency: number): BenchmarkResult { +function runIsolated( + fileCount: number, + messagesPerFile: number, + concurrency: number, + incrementalMode: 'delta' | 'full' | null +): BenchmarkResult { const child = spawnSync( process.execPath, [ @@ -469,6 +557,7 @@ function runIsolated(fileCount: number, messagesPerFile: number, concurrency: nu String(fileCount), String(messagesPerFile), String(concurrency), + incrementalMode ?? 'none', ], { cwd: process.cwd(), @@ -489,10 +578,58 @@ function formatBytes(bytes: number): string { return `${(bytes / 1024 / 1024).toFixed(1)} MB` } +function medianNumber(values: number[]): number { + const sorted = [...values].sort((left, right) => left - right) + const midpoint = Math.floor(sorted.length / 2) + return sorted.length % 2 === 0 ? (sorted[midpoint - 1] + sorted[midpoint]) / 2 : sorted[midpoint] +} + +function buildMedianResult(results: BenchmarkResult[]): BenchmarkResult { + const representative = results[0] + const medianTiming = (key: keyof ImportStageTimings) => medianNumber(results.map((result) => result.timings[key])) + + return { + ...representative, + inputBytes: medianNumber(results.map((result) => result.inputBytes)), + durationMs: medianNumber(results.map((result) => result.durationMs)), + incrementalAnalyzeDurationMs: medianNumber(results.map((result) => result.incrementalAnalyzeDurationMs)), + incrementalAnalyzeRssDeltaMb: medianNumber(results.map((result) => result.incrementalAnalyzeRssDeltaMb)), + incrementalDurationMs: medianNumber(results.map((result) => result.incrementalDurationMs)), + incrementalImportRssDeltaMb: medianNumber(results.map((result) => result.incrementalImportRssDeltaMb)), + sampledPeakRssMb: medianNumber(results.map((result) => result.sampledPeakRssMb)), + sampledRssDeltaMb: medianNumber(results.map((result) => result.sampledRssDeltaMb)), + databaseBytes: medianNumber(results.map((result) => result.databaseBytes)), + likeSearch: { + ...representative.likeSearch, + coldMs: medianNumber(results.map((result) => result.likeSearch.coldMs)), + warmMedianMs: medianNumber(results.map((result) => result.likeSearch.warmMedianMs)), + }, + timings: { + detectionMs: medianTiming('detectionMs'), + preprocessingMs: medianTiming('preprocessingMs'), + databaseSetupMs: medianTiming('databaseSetupMs'), + parserMs: medianTiming('parserMs'), + metaWriteMs: medianTiming('metaWriteMs'), + memberWriteMs: medianTiming('memberWriteMs'), + messageWriteMs: medianTiming('messageWriteMs'), + nicknameHistoryMs: medianTiming('nicknameHistoryMs'), + indexCreationMs: medianTiming('indexCreationMs'), + checkpointMs: medianTiming('checkpointMs'), + sessionIndexMs: medianTiming('sessionIndexMs'), + postImportHookMs: medianTiming('postImportHookMs'), + totalMs: medianTiming('totalMs'), + }, + } +} + 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 | ` + + `incremental ${result.incrementalMode ?? 'none'} ` + + `${result.incrementalAnalyzeDurationMs.toFixed(0)} ms analyze/${result.incrementalDurationMs.toFixed(0)} ms import ` + + `(${result.incrementalMessages} new/${result.incrementalDuplicateMessages} duplicate) | ` + + `incremental RSS +${result.incrementalAnalyzeRssDeltaMb.toFixed(0)} MB analyze/` + + `+${result.incrementalImportRssDeltaMb.toFixed(0)} MB import | ` + `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 | ` + @@ -503,7 +640,8 @@ function printResult(result: BenchmarkResult, label: string): void { 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)) + const workerIncrementalMode = fourth === 'delta' || fourth === 'full' ? fourth : null + const result = await runWorker(Number(first), Number(second), Number(third), workerIncrementalMode) console.log(`BENCH_RESULT ${JSON.stringify(result)}`) return } @@ -512,18 +650,27 @@ async function main(): Promise { let messagesPerFile: number let runs: number let concurrency: number + let incrementalMode: 'delta' | 'full' | null if (mode === 'single') { fileCount = 1 messagesPerFile = Number(first ?? 100_000) runs = Number(second ?? 3) concurrency = 1 + incrementalMode = third === 'full' ? 'full' : 'delta' + if (third !== undefined && third !== 'delta' && third !== 'full') { + throw new Error('Incremental mode must be delta or full') + } } else if (mode === 'batch') { fileCount = Number(first ?? 100) messagesPerFile = Number(second ?? 10_000) runs = Number(third ?? 3) concurrency = Number(fourth ?? 1) + incrementalMode = null } else { - throw new Error('Usage: single [runs=3] | batch [runs=3] [concurrency=1]') + throw new Error( + 'Usage: single [runs=3] [incremental-mode=delta|full] | ' + + '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') @@ -534,13 +681,12 @@ async function main(): Promise { ) const results: BenchmarkResult[] = [] for (let index = 0; index < runs; index++) { - const result = runIsolated(fileCount, messagesPerFile, concurrency) + const result = runIsolated(fileCount, messagesPerFile, concurrency, incrementalMode) 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 median = buildMedianResult(results) const parserSignatures = new Set(results.map((result) => JSON.stringify(result.parser))) if (parserSignatures.size !== 1) { throw new Error('Parser implementations differ between isolated runs')