diff --git a/src/services/rawDump/batchWorker.ts b/src/services/rawDump/batchWorker.ts index 561089cae..7086929e3 100644 --- a/src/services/rawDump/batchWorker.ts +++ b/src/services/rawDump/batchWorker.ts @@ -13,7 +13,24 @@ import { createLogger } from './logger.js' const log = createLogger('raw-dump-batch') -const BATCH_INTERVAL_MS = 30_000 // 每轮间隔 +type RepoInfo = Awaited> + +const BATCH_INTERVAL_MS = 120_000 // 每轮间隔（2 分钟），降低内联运行时的 CPU 影响 + +// Git repo 信息缓存，同一 directory 的多个 task 短时间内不需要重复 spawn git +const repoInfoCache = new Map() +const REPO_CACHE_TTL_MS = 60_000 + +async function getCachedRepoInfo(directory: string): Promise { + const cached = repoInfoCache.get(directory) + if (cached && Date.now() - cached.ts < REPO_CACHE_TTL_MS) { + log.debug('repo info cache hit', { directory }) + return cached.repoInfo + } + const repoInfo = await getRepoInfo(directory) + repoInfoCache.set(directory, { repoInfo, ts: Date.now() }) + return repoInfo +} // 进程内重入保护：文件锁不防同进程重入，必须用内存 flag 兜底 let isRunning = false @@ -30,21 +47,41 @@ function isParentAlive(): boolean { } } -async function processTask(task: QueueTask) { +// Session messages 缓存：同一 session 的多个 task 短时间内不需要重复读取 JSONL +const sessionMessagesCache = new Map[]; ts: number }>() +const SESSION_CACHE_TTL_MS = 60_000 + +async function getCachedSessionMessages(sessionDir: string, sessionID: string, messageID?: string) { + const cacheKey = `${sessionDir}:${sessionID}` + const cached = sessionMessagesCache.get(cacheKey) + if (cached && Date.now() - cached.ts < SESSION_CACHE_TTL_MS) { + log.debug('session messages cache hit', { sessionID, messageID }) + return cached.messages + } + const start = Date.now() + const messages = await loadSessionMessages(sessionDir, sessionID, messageID) + const elapsed = Date.now() - start + if (elapsed > 100) { + log.info('loadSessionMessages slow', { sessionID, elapsedMs: elapsed, messageCount: messages.length }) + } + sessionMessagesCache.set(cacheKey, { messages, ts: Date.now() }) + return messages +} + +async function processTask(task: QueueTask, state: Awaited>) { log.info('processing task', { sessionID: task.sessionID, messageID: task.messageID }) const sessionDir = getSessionDirectory(task.directory, task.sessionID) - const messages = await loadSessionMessages(sessionDir, task.sessionID, task.messageID) + const messages = await getCachedSessionMessages(sessionDir, task.sessionID, task.messageID) if (messages.length === 0) { log.warn('no messages found', { sessionDir, sessionID: task.sessionID }) } const authData = await authWithFallback() - const state = await readState() // 预加载 git 信息，commits 和 repo 字段共享，避免每个 task 重复 spawn git 进程 - const repoInfo = await getRepoInfo(task.directory) + const repoInfo = await getCachedRepoInfo(task.directory) try { // conversation @@ -66,9 +103,12 @@ async function processTask(task: QueueTask) { await uploadCommits({ directory: task.directory }, authData, state, { repoInfo }) log.info('task completed', { sessionID: task.sessionID, conversationUploaded }) - } finally { - // 无论成功或失败，都写入 state（commits 已逐条更新） - await writeState(state) + } catch (err) { + log.error('task failed', { + error: err instanceof Error ? err.message : String(err), + sessionID: task.sessionID, + }) + throw err } } @@ -82,13 +122,13 @@ async function runBatch() { try { // 第二道防线：跨进程文件锁 - if (!acquireLock()) { + if (!(await acquireLock())) { log.debug('another worker process holds the lock, skip') return } try { - const tasks = readQueue() + const tasks = await readQueue() if (tasks.length === 0) { log.debug('queue empty') return @@ -97,7 +137,7 @@ async function runBatch() { // 第三道防线：读完立刻清空队列 // - 处理期间新进来的任务会在下一轮处理 // - 即使有意外的并发 runBatch 拿到锁，也只会看到空队列直接返回 - clearQueue() + await clearQueue() log.info(`processing ${tasks.length} tasks`) @@ -114,9 +154,12 @@ async function runBatch() { const uniqueTasks = Array.from(deduped.values()).sort((a, b) => a.enqueuedAt - b.enqueuedAt) log.info(`deduped to ${uniqueTasks.length} unique tasks`) + // 一次性读取 state，所有 task 共享，减少文件锁竞争和重复 JSON 解析 + const state = await readState() + for (const task of uniqueTasks) { try { - await processTask(task) + await processTask(task, state) } catch (err) { log.error('task failed', { error: err instanceof Error ? err.message : String(err), @@ -125,9 +168,11 @@ async function runBatch() { } } + // 所有 task 处理完后一次性写入 state + await writeState(state) log.info('batch completed') } finally { - releaseLock() + await releaseLock() } } finally { isRunning = false diff --git a/src/services/rawDump/index.ts b/src/services/rawDump/index.ts index 85b55f571..bf559f711 100644 --- a/src/services/rawDump/index.ts +++ b/src/services/rawDump/index.ts @@ -7,9 +7,16 @@ import { isLocalDumpMode } from './localStorage.js' import { enqueue } from './queue.js' import { spawnBatchWorker } from './spawn.js' import { startBatchWorker } from './batchWorker.js' +import { createLogger } from './logger.js' + +const log = createLogger('raw-dump') let batchWorkerSpawned = false +// 调用频率限制：同一 session + messageID 5s 内不重复 enqueue +const lastEnqueueMap = new Map() +const ENQUEUE_DEBOUNCE_MS = 5_000 + function isEnabled(): boolean { // 本地调试模式自动启用 if (isLocalDumpMode()) return true @@ -24,23 +31,37 @@ function ensureBatchWorker() { batchWorkerSpawned = true const spawned = spawnBatchWorker() if (!spawned) { - // Fallback: compiled binary mode where external worker file is not available + log.warn('batch worker spawn failed, falling back to inline worker') startBatchWorker() } } +function shouldEnqueue(sessionID: string, messageID: string): boolean { + const key = `${sessionID}:${messageID}` + const now = Date.now() + const last = lastEnqueueMap.get(key) + if (last && now - last < ENQUEUE_DEBOUNCE_MS) { + log.debug('reportTurn debounced', { sessionID, messageID, lastMs: now - last }) + return false + } + lastEnqueueMap.set(key, now) + return true +} + /** * 上报一轮对话 * 只写入队列，由 batch worker 顺序消费 */ export function reportTurn(sessionID: string, messageID: string, directory: string): void { if (!isEnabled()) return + if (!shouldEnqueue(sessionID, messageID)) return enqueue({ sessionID, messageID, directory }) ensureBatchWorker() } export function reportSession(sessionID: string, directory: string): void { if (!isEnabled()) return + if (!shouldEnqueue(sessionID, '__summary__')) return enqueue({ sessionID, messageID: '__summary__', directory }) ensureBatchWorker() } diff --git a/src/services/rawDump/queue.ts b/src/services/rawDump/queue.ts index 5334cb948..31b21ef46 100644 --- a/src/services/rawDump/queue.ts +++ b/src/services/rawDump/queue.ts @@ -3,7 +3,8 @@ * 主进程只写队列，独立 batch worker 顺序消费 */ -import { appendFileSync, readFileSync, writeFileSync } from 'node:fs' +import { appendFileSync } from 'node:fs' +import { promises as fs } from 'node:fs' import os from 'node:os' import path from 'node:path' @@ -19,16 +20,13 @@ export interface QueueTask { export function enqueue(task: Omit): void { const item: QueueTask = { ...task, enqueuedAt: Date.now() } - try { - appendFileSync(QUEUE_FILE, JSON.stringify(item) + '\n', 'utf-8') - } catch { - // ignore - } + // 使用 fire-and-forget 异步写入，避免阻塞主进程 event loop + fs.appendFile(QUEUE_FILE, JSON.stringify(item) + '\n', 'utf-8').catch(() => {}) } -export function readQueue(): QueueTask[] { +export async function readQueue(): Promise { try { - const text = readFileSync(QUEUE_FILE, 'utf-8') + const text = await fs.readFile(QUEUE_FILE, 'utf-8') return text .split('\n') .filter(Boolean) @@ -45,19 +43,18 @@ export function readQueue(): QueueTask[] { } } -export function clearQueue(): void { +export async function clearQueue(): Promise { try { - writeFileSync(QUEUE_FILE, '', 'utf-8') + await fs.writeFile(QUEUE_FILE, '', 'utf-8') } catch { // ignore } } -export function acquireLock(): boolean { +export async function acquireLock(): Promise { try { - // 简单文件锁：如果 lock 文件存在且 60 秒内，认为已有 worker try { - const stat = readFileSync(LOCK_FILE, 'utf-8') + const stat = await fs.readFile(LOCK_FILE, 'utf-8') const pid = parseInt(stat, 10) if (!isNaN(pid) && pid !== process.pid) { // 检查进程是否还在运行 @@ -71,16 +68,16 @@ export function acquireLock(): boolean { } catch { // lock 文件不存在 } - writeFileSync(LOCK_FILE, String(process.pid), 'utf-8') + await fs.writeFile(LOCK_FILE, String(process.pid), 'utf-8') return true } catch { return false } } -export function releaseLock(): void { +export async function releaseLock(): Promise { try { - writeFileSync(LOCK_FILE, '', 'utf-8') + await fs.writeFile(LOCK_FILE, '', 'utf-8') } catch { // ignore } diff --git a/src/services/rawDump/spawn.ts b/src/services/rawDump/spawn.ts index 095180e53..8b2f88454 100644 --- a/src/services/rawDump/spawn.ts +++ b/src/services/rawDump/spawn.ts @@ -44,6 +44,8 @@ function resolveRuntime(): { entry: string; isBun: boolean } | null { if (typeof Bun !== 'undefined' && Bun.which) { const bun = Bun.which('bun') if (bun) return { entry: bun, isBun: true } + const node = Bun.which('node') + if (node) return { entry: node, isBun: false } } return null diff --git a/src/services/rawDump/worker.ts b/src/services/rawDump/worker.ts index aec15f59e..2bf691780 100644 --- a/src/services/rawDump/worker.ts +++ b/src/services/rawDump/worker.ts @@ -248,7 +248,16 @@ export async function loadSessionMessages(sessionDir: string, sessionId: string, const jsonlFiles = entries.filter((f) => f.endsWith('.jsonl')) log.debug('found jsonl files', { sessionDir, count: jsonlFiles.length, files: jsonlFiles.slice(0, 5) }) - for (const file of jsonlFiles) { + // 优先读取文件名包含 sessionId 的文件，减少无意义解析 + const prioritized = jsonlFiles.sort((a, b) => { + const aHas = a.includes(sessionId) + const bHas = b.includes(sessionId) + if (aHas && !bHas) return -1 + if (!aHas && bHas) return 1 + return 0 + }) + + for (const file of prioritized) { const filePath = path.join(sessionDir, file) try { const text = await fs.readFile(filePath, 'utf-8')