From 455489cc895a91c5ed7d880951db9d48854c7b63 Mon Sep 17 00:00:00 2001 From: zbc Date: Wed, 27 May 2026 09:48:38 +0800 Subject: [PATCH] =?UTF-8?q?1.=20=E4=BF=AE=E6=94=B9=E4=B8=8A=E6=8A=A5?= =?UTF-8?q?=E9=98=9F=E5=88=97=E6=9C=BA=E5=88=B6=EF=BC=8C=E7=A1=AE=E4=BF=9D?= =?UTF-8?q?=E4=B8=8A=E6=8A=A5=E5=A4=B1=E8=B4=A5=E6=9C=89=E5=8F=AF=E9=9D=A0?= =?UTF-8?q?=E7=9A=84=E9=87=8D=E8=AF=95=E6=9C=BA=E5=88=B6=EF=BC=8C=E4=B8=8D?= =?UTF-8?q?=E4=BC=9A=E9=9D=99=E9=BB=98=E5=A4=B1=E8=B4=A5=EF=BC=9B2.?= =?UTF-8?q?=E4=BF=AE=E6=94=B9conversation=E4=B8=8A=E6=8A=A5=E9=81=97?= =?UTF-8?q?=E6=BC=8F=E7=9A=84=E9=97=AE=E9=A2=98=EF=BC=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/services/rawDump/README.md | 41 ++++-- src/services/rawDump/batchWorker.ts | 197 ++++++++++++++------------- src/services/rawDump/index.ts | 12 +- src/services/rawDump/localStorage.ts | 46 +++++-- src/services/rawDump/logger.ts | 6 +- src/services/rawDump/queue.ts | 94 ++++++++----- src/services/rawDump/state.ts | 15 +- src/services/rawDump/types.ts | 20 ++- src/services/rawDump/worker.ts | 46 +++++-- 9 files changed, 295 insertions(+), 182 deletions(-) diff --git a/src/services/rawDump/README.md b/src/services/rawDump/README.md index c83ac6324..1cdd77aef 100644 --- a/src/services/rawDump/README.md +++ b/src/services/rawDump/README.md @@ -225,7 +225,7 @@ if (!existing || task.enqueuedAt > existing.enqueuedAt) { ### 2. Conversation 去重（磁盘） ```typescript -// ~/.claude/csc-raw-dump-state.json +// ~/.claude/raw-dump/csc-state.json { "conversation": { "taskID:requestID": true @@ -301,21 +301,36 @@ import { refreshCoStrictToken } from '../../costrict/provider/token.js' | `CSC_RAW_DUMP_BASE_URL` | 自定义上报 base URL | 从凭证读取 | | `COSTRICT_RAW_DUMP_BASE_URL` | 兼容 opencode 的自定义 URL | 从凭证读取 | | `COSTRICT_BASE_URL` | CoStrict 服务地址 | `https://zgsm.sangfor.com` | -| `CSC_RAW_DUMP_LOCAL_MODE` | 本地留存模式，数据只写入本地文件不上报服务端 | `false` | -| `CSC_RAW_DUMP_LOCAL_DIR` | 本地留存目录 | `~/.claude/raw-dump-local` | +| `CSC_RAW_DUMP_MODE` | 上报模式：`0`=禁用, `1`=remote, `2`=local, `3`=both | `1`（Remote） | +| `CSC_RAW_DUMP_DIR` | 本地留存目录（mode=2/3 时使用） | `~/.claude/raw-dump` | + +--- + +## 上报模式（`CSC_RAW_DUMP_MODE`） + +`CSC_RAW_DUMP_MODE` 环境变量控制数据输出行为，支持 4 种模式： + +| 值 | 模式 | 说明 | +|----|------|------| +| `0` | 禁用（Disabled） | 完全跳过，不写入队列、不触发任何 HTTP 请求 | +| `1` | 远程上报（Remote） | 默认值，数据 POST 到 CoStrict 服务端 | +| `2` | 本地留存（Local） | 数据只写入本地 JSON 文件，不上报服务端 | +| `3` | 双写（Both） | 同时写入本地文件和上报服务端 | + +默认值为 `1`（Remote），即正常运行时数据上报到 CoStrict 服务端。 --- ## 本地留存模式（调试排障） -通过环境变量开启，开启后数据**不上报服务端**，仅写入本地 JSON 文件： +通过设置 `CSC_RAW_DUMP_MODE=2` 或 `3` 开启，开启后数据写入本地 JSON 文件： ```bash # 开启本地留存模式 -export CSC_RAW_DUMP_LOCAL_MODE=1 +export CSC_RAW_DUMP_MODE=2 -# 可选：自定义留存目录（默认 ~/.claude/raw-dump-local） -export CSC_RAW_DUMP_LOCAL_DIR=/tmp/raw-dump-debug +# 可选：自定义留存目录(默认 ~/.claude/raw-dump) +export CSC_RAW_DUMP_DIR=/tmp/raw-dump-debug ``` 留存文件结构： @@ -352,7 +367,7 @@ export CSC_RAW_DUMP_LOCAL_DIR=/tmp/raw-dump-debug ### 状态文件 ``` -~/.claude/csc-raw-dump-state.json +~/.claude/raw-dump/csc-state.json ``` 内容格式： @@ -373,7 +388,7 @@ export CSC_RAW_DUMP_LOCAL_DIR=/tmp/raw-dump-debug ### 日志文件 ``` -~/.claude/csc-raw-dump.log +~/.claude/raw-dump/csc-raw-dump.log ``` 主进程和 batch worker 的日志都追加写入该文件。由于 worker 是 detached 进程（`stdio: 'ignore'`），日志只能通过文件查看。 @@ -431,7 +446,7 @@ export CSC_RAW_DUMP_LOCAL_DIR=/tmp/raw-dump-debug export CSC_RAW_DUMP_DEBUG=1 # 查看日志 -tail -f ~/.claude/csc-raw-dump.log +tail -f ~/.claude/raw-dump/csc-raw-dump.log ``` 关键日志标识： @@ -443,11 +458,11 @@ tail -f ~/.claude/csc-raw-dump.log 常用排查命令： ```bash # 查看 state 文件 -cat ~/.claude/csc-raw-dump-state.json +cat ~/.claude/raw-dump/csc-state.json # 查看队列文件 -cat ~/.claude/csc-raw-dump-queue.jsonl +cat ~/.claude/raw-dump/csc-work-queue.json # 查看是否有 worker 在运行（锁文件） -cat ~/.claude/csc-raw-dump.lock +cat ~/.claude/raw-dump/csc-work-queue.lock ``` diff --git a/src/services/rawDump/batchWorker.ts b/src/services/rawDump/batchWorker.ts index 8d0f283c5..9ff99b660 100644 --- a/src/services/rawDump/batchWorker.ts +++ b/src/services/rawDump/batchWorker.ts @@ -2,6 +2,14 @@ * Raw Dump Batch Worker * 顺序消费队列，避免并发 429 * 独立进程，通过自循环 setTimeout 严格串行执行 + * + * 单队列设计： + * - 队列常驻内存，进程启动时一次性加载 + * - 入队/移除时写文件，其他操作仅更新内存 + * - 任务成功或 attemptCount >= MAX_ATTEMPTS 时从队列移除 + * - 任务失败但未超限：attemptCount++ 写回文件，下次 batch 重试 + * + * 每轮 batch 三道防线：进程内重入保护 → 跨进程文件锁 → 消费后清空文件 */ import { @@ -11,54 +19,56 @@ import { authWithFallback, } from './worker.js' import { - readQueue, - clearQueue, + loadQueue, acquireLock, releaseLock, + peekTask, + removeTask, + getQueue, + flushQueue, + MAX_ATTEMPTS, type QueueTask, } from './queue.js' import { readState, writeState } from './state.js' +import type { RawDumpError } from './types.js' import { getSessionDirectory, loadSessionMessages } from './worker.js' import { getRepoInfo } from './git.js' import { createLogger } from './logger.js' const log = createLogger('raw-dump-batch') +/** + * 从错误信息中提取 API endpoint 路径 + * 错误格式："/raw-store/task-conversation failed: 401 ..." + */ +function extractEndpointFromError(errorMsg: string): string | undefined { + const match = errorMsg.match(/\/(raw-store\/[^\s]+)/) + return match?.[1] +} + type RepoInfo = Awaited> -const BATCH_INTERVAL_MS = 120_000 // 每轮间隔（2 分钟），降低内联运行时的 CPU 影响 +const BATCH_INTERVAL_MS = 120_000 // Git repo 信息缓存，同一 directory 的多个 task 短时间内不需要重复 spawn git const repoInfoCache = new Map() const REPO_CACHE_TTL_MS = 60_000 -/** - * 批量获取 Git 仓库信息（带缓存） - * 同一 directory 在 60 秒内不重复调用 git 命令 - * @returns 缓存的 RepoInfo 或新获取的信息 - */ 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 const PARENT_PID = process.ppid const IS_WORKER_PROCESS = process.argv[1]?.includes('batchWorker') || false -/** - * 检测父进程是否还在运行 - * - 仅在独立 worker 进程中生效（IS_WORKER_PROCESS = true） - * - 使用 kill(pid, 0) 检测进程是否存在 - * - 父进程退出时 worker 应停止，避免孤儿进程 - */ function isParentAlive(): boolean { if (!IS_WORKER_PROCESS) return true try { @@ -69,18 +79,13 @@ function isParentAlive(): boolean { } } -// Session messages 缓存：同一 session 的多个 task 短时间内不需要重复读取 JSONL +// Session messages 缓存 const sessionMessagesCache = new Map< string, { messages: Record[]; ts: number } >() const SESSION_CACHE_TTL_MS = 60_000 -/** - * 批量获取 session 消息列表（带缓存） - * 同一 sessionDir + sessionID 在 60 秒内不重复读取 JSONL 文件 - * 记录加载耗时，超过 100ms 时输出 info 日志（便于发现性能问题） - */ async function getCachedSessionMessages( sessionDir: string, sessionID: string, @@ -89,18 +94,13 @@ async function getCachedSessionMessages( 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, - }) + log.info('loadSessionMessages slow', { sessionID, elapsedMs: elapsed }) } sessionMessagesCache.set(cacheKey, { messages, ts: Date.now() }) return messages @@ -108,11 +108,7 @@ async function getCachedSessionMessages( /** * 处理单个上报任务 - * - 加载 session 消息（可能为空） - * - 依次调用 uploadConversation、uploadSummary、uploadCommits - * - 任何任务失败都抛异常，由上层 runBatch 统一处理 - * @param task 队列任务 - * @param state 当前 batch 的 state 快照（内存引用，函数内修改会直接反映到外层） + * 成功完成后由调用方从队列中移除 */ async function processTask( task: QueueTask, @@ -121,6 +117,7 @@ async function processTask( log.info('processing task', { sessionID: task.sessionID, messageID: task.messageID, + attempt: task.attemptCount, }) const sessionDir = getSessionDirectory(task.directory, task.sessionID) @@ -135,56 +132,38 @@ async function processTask( } const authData = await authWithFallback() - - // 预加载 git 信息，commits 和 repo 字段共享，避免每个 task 重复 spawn git 进程 const repoInfo = await getCachedRepoInfo(task.directory) - try { - // conversation - const conversationUploaded = await uploadConversation( - { - sessionID: task.sessionID, - messageID: task.messageID, - directory: task.directory, - messages, - }, - authData, - state, - { repoInfo }, - ) - - // summary（5 分钟内同一 session 只上报一次） - await uploadSummary( - { sessionID: task.sessionID, directory: task.directory, messages }, - authData, - state, - ) - - // commits（限制频率，避免重复上报） - await uploadCommits({ directory: task.directory }, authData, state, { - repoInfo, - }) - - log.info('task completed', { + const conversationUploaded = await uploadConversation( + { sessionID: task.sessionID, - conversationUploaded, - }) - } catch (err) { - log.error('task failed', { - error: err instanceof Error ? err.message : String(err), - sessionID: task.sessionID, - }) - throw err - } + messageID: task.messageID, + directory: task.directory, + messages, + }, + authData, + state, + { repoInfo }, + ) + + await uploadSummary( + { sessionID: task.sessionID, directory: task.directory, messages }, + authData, + state, + ) + + await uploadCommits({ directory: task.directory }, authData, state, { + repoInfo, + }) + + log.info('task completed', { sessionID: task.sessionID, conversationUploaded }) } /** * 执行一轮 batch 处理 - * 三道防线：进程内重入保护 → 跨进程文件锁 → 读取后立即清空队列 - * 流程：读取队列 → 清空队列 → 去重 → 逐个处理任务 → 一次性写入 state + * 流程：获取锁 → 读队列内存 → 去重 → 逐个处理 → 写 state */ async function runBatch() { - // 第一道防线：同进程重入保护 if (isRunning) { log.debug('runBatch already running in-process, skip') return @@ -192,27 +171,23 @@ async function runBatch() { isRunning = true try { - // 第二道防线：跨进程文件锁 if (!(await acquireLock())) { log.debug('another worker process holds the lock, skip') return } try { - const tasks = await readQueue() + const state = await readState() + const tasks = getQueue() + if (tasks.length === 0) { log.debug('queue empty') return } - // 第三道防线：读完立刻清空队列 - // - 处理期间新进来的任务会在下一轮处理 - // - 即使有意外的并发 runBatch 拿到锁，也只会看到空队列直接返回 - await clearQueue() - log.info(`processing ${tasks.length} tasks`) - // 去重：同一个 session 的多个 task，只保留最新的一个 + // 去重：同一 session:messageID 只保留最新的 const deduped = new Map() for (const task of tasks) { const key = `${task.sessionID}:${task.messageID}` @@ -225,23 +200,49 @@ 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) { + const key = `${task.sessionID}:${task.messageID}` try { await processTask(task, state) + removeTask(key) } catch (err) { + const errorMsg = err instanceof Error ? err.message : String(err) + task.attemptCount++ log.error('task failed', { - error: err instanceof Error ? err.message : String(err), + error: errorMsg, sessionID: task.sessionID, + attemptCount: task.attemptCount, }) + + if (task.attemptCount >= MAX_ATTEMPTS) { + // 彻底失败：记录错误，从队列移除 + state.errors[key] = { + message: errorMsg, + count: task.attemptCount, + lastAt: new Date().toISOString(), + endpoint: extractEndpointFromError(errorMsg), + } satisfies RawDumpError + removeTask(key) + log.warn('task permanently failed, removed from queue', { + sessionID: task.sessionID, + attemptCount: task.attemptCount, + }) + } else { + // 失败但未超限：attemptCount++ 写回文件，下次重试 + state.errors[key] = { + message: errorMsg, + count: task.attemptCount, + lastAt: new Date().toISOString(), + endpoint: extractEndpointFromError(errorMsg), + } satisfies RawDumpError + await flushQueue() + } } } - // 所有 task 处理完后一次性写入 state await writeState(state) log.info('batch completed') } finally { @@ -254,16 +255,11 @@ async function runBatch() { /** * 启动 Batch Worker - * 通过 setTimeout 自循环实现严格串行执行，避免并发触发 API 限流 - * 启动时随机抖动 0~10 秒，避免多个 csc 实例同时启动时撞 API - * 每次 batch 结束后等待 2 分钟（加 0~5 秒随机抖动）再执行下一轮 - * 父进程退出时自动停止 + * 进程启动时加载队列到内存，此后队列常驻内存运行 */ export function startBatchWorker() { log.info('batch worker started', { interval: BATCH_INTERVAL_MS }) - // 自循环 setTimeout：上一轮跑完才安排下一轮，从源头消除并发 - // 即便 runBatch 抛错也确保下一轮被排上，避免 worker 卡死 const scheduleNext = (delay: number) => { setTimeout(async () => { if (!isParentAlive()) { @@ -282,15 +278,22 @@ export function startBatchWorker() { }, delay) } - // 启动时随机抖动 0~10s，避免多个 csc 实例同时起 worker 撞 API - scheduleNext(Math.floor(Math.random() * 10_000)) + // 启动时加载队列 + 随机抖动 + loadQueue().then(() => { + log.info('queue loaded', { count: getQueue().length }) + scheduleNext(Math.floor(Math.random() * 10_000)) + }).catch(err => { + log.error('failed to load queue', { + error: err instanceof Error ? err.message : String(err), + }) + process.exit(1) + }) } -// 如果直接运行此文件 const scriptPath = process.argv[1] || '' if ( scriptPath.endsWith('batchWorker.ts') || scriptPath.endsWith('batchWorker.js') ) { startBatchWorker() -} +} \ No newline at end of file diff --git a/src/services/rawDump/index.ts b/src/services/rawDump/index.ts index 12241c4b5..b1008e71c 100644 --- a/src/services/rawDump/index.ts +++ b/src/services/rawDump/index.ts @@ -3,7 +3,7 @@ * 队列模式：主进程只 enqueue，单 batch worker 顺序消费 */ -import { isLocalDumpMode } from './localStorage.js' +import { getRawDumpMode, RAW_DUMP_MODE } from './localStorage.js' import { enqueue } from './queue.js' import { spawnBatchWorker } from './spawn.js' import { startBatchWorker } from './batchWorker.js' @@ -19,13 +19,13 @@ const ENQUEUE_DEBOUNCE_MS = 5_000 /** * 判断 Raw Dump 是否启用 - * - 本地调试模式（isLocalDumpMode）自动启用 + * - mode=0 时不输出（完全禁用） + * - mode=1/2/3 时启用 * - 环境变量 CSC_DISABLE_RAW_DUMP 或 COSTRICT_DISABLE_RAW_DUMP 为 '1'/'true' 时禁用 - * - 默认启用 + * - 默认 mode=1（remote only） */ function isEnabled(): boolean { - // 本地调试模式自动启用 - if (isLocalDumpMode()) return true + if (getRawDumpMode() === RAW_DUMP_MODE.DISABLED) return false // 显式禁用 if ( process.env.CSC_DISABLE_RAW_DUMP === '1' || @@ -37,7 +37,7 @@ function isEnabled(): boolean { process.env.COSTRICT_DISABLE_RAW_DUMP === 'true' ) return false - // 默认启用 raw dump + // 默认启用 return true } diff --git a/src/services/rawDump/localStorage.ts b/src/services/rawDump/localStorage.ts index c6efbbdb4..3bfc7d780 100644 --- a/src/services/rawDump/localStorage.ts +++ b/src/services/rawDump/localStorage.ts @@ -1,32 +1,54 @@ /** * Raw Dump 本地存储模式 * 开启后上报数据不落服务端，仅写入本地 JSON 文件，用于排障和调试 + * + * CSC_RAW_DUMP_MODE 语义： + * - 0: 不输出 + * - 1: 输出到 remote（POST） + * - 2: 输出到本地（writeLocalDump） + * - 3: 同时输出到 remote 和 local */ import { promises as fs } from 'node:fs' import os from 'node:os' import path from 'node:path' -const DEFAULT_LOCAL_DIR = path.join(os.homedir(), '.claude', 'raw-dump-local') +const DEFAULT_LOCAL_DIR = path.join(os.homedir(), '.claude', 'raw-dump') + +export type RawDumpMode = 0 | 1 | 2 | 3 + +export const RAW_DUMP_MODE = { + DISABLED: 0, + REMOTE: 1, + LOCAL: 2, + BOTH: 3, +} as const /** * 获取本地存储目录路径 - * 支持环境变量 CSC_RAW_DUMP_LOCAL_DIR 覆盖 + * 支持环境变量 CSC_RAW_DUMP_DIR 覆盖 */ export function getLocalDumpDir(): string { - return (process.env.CSC_RAW_DUMP_LOCAL_DIR || DEFAULT_LOCAL_DIR).replace( + return (process.env.CSC_RAW_DUMP_DIR || DEFAULT_LOCAL_DIR).replace( /\/$/, '', ) } /** - * 判断是否启用本地存储模式 - * 环境变量 CSC_RAW_DUMP_LOCAL_MODE 为 '1' 或 'true' 时启用 + * 获取 Raw Dump 输出模式 + * - 0: 不输出（跳过所有上报） + * - 1: 输出到 remote（POST） + * - 2: 输出到本地（writeLocalDump） + * - 3: 同时输出到 remote 和 local + * 默认 1（remote only） */ -export function isLocalDumpMode(): boolean { - const mode = process.env.CSC_RAW_DUMP_LOCAL_MODE - return mode === '1' || mode === 'true' +export function getRawDumpMode(): RawDumpMode { + const v = process.env.CSC_RAW_DUMP_MODE + if (v === '0') return 0 + if (v === '2') return 2 + if (v === '3') return 3 + return 1 } /** @@ -40,14 +62,14 @@ export async function writeLocalDump( body: Record, ): Promise { const dir = getLocalDumpDir() - const taskId = (body.task_id as string) || 'unknown' - const taskDir = path.join(dir, taskId) + const taskId = (body.task_id as string) || (body.commit_id as string)|| 'unknown' + const taskDir = path.join(dir, type, taskId) await fs.mkdir(taskDir, { recursive: true }) const timestamp = new Date().toISOString().replace(/[:.]/g, '-').slice(0, -5) const requestId = - (body.request_id as string) || (body.commit_id as string) || 'unknown' - const filename = `${timestamp}-${type}-${requestId}.json` + (body.request_id as string) || (body.commit_id as string) || (body.task_id as string) || 'unknown' + const filename = `${timestamp}-${requestId}.json` const filePath = path.join(taskDir, filename) const payload = { diff --git a/src/services/rawDump/logger.ts b/src/services/rawDump/logger.ts index ec789f918..7c042aab2 100644 --- a/src/services/rawDump/logger.ts +++ b/src/services/rawDump/logger.ts @@ -7,7 +7,7 @@ import { appendFileSync } from 'node:fs' import os from 'node:os' import path from 'node:path' -const LOG_FILE = path.join(os.homedir(), '.claude', 'csc-raw-dump.log') +const LOG_FILE = path.join(os.homedir(), '.claude', 'raw-dump', 'csc-raw-dump.log') function isDebugEnabled(): boolean { const v = process.env.CSC_RAW_DUMP_DEBUG @@ -18,7 +18,9 @@ export function createLogger(prefix: string) { const enabled = isDebugEnabled() function write(level: string, msg: string, meta?: Record) { - if (!enabled) return + // error 和 warn 级别始终输出，不受 DEBUG 开关控制 + const alwaysWrite = level === 'error' || level === 'warn' + if (!alwaysWrite && !enabled) return const timestamp = new Date().toISOString() const metaStr = meta ? ` ${JSON.stringify(meta)}` : '' const line = `[${timestamp}] [${prefix}:${level}] ${msg}${metaStr}\n` diff --git a/src/services/rawDump/queue.ts b/src/services/rawDump/queue.ts index a3fd61e3e..1b49341b4 100644 --- a/src/services/rawDump/queue.ts +++ b/src/services/rawDump/queue.ts @@ -1,49 +1,48 @@ /** * Raw Dump 任务队列 * 主进程只写队列，独立 batch worker 顺序消费 + * + * 单一队列设计（csc-workq-queue.jsonl）： + * - 队列常驻内存，进程启动时一次性加载，后续操作均在内存中进行 + * - 任务入队/移除时写文件（其他操作仅更新内存） + * - 成功则移除；彻底失败（attemptCount >= MAX_ATTEMPTS）也移除 + * - 失败但未超限：attemptCount++ 后写回文件，下次 batch 重试 */ -import { appendFileSync } from 'node:fs' import { promises as fs } from 'node:fs' import os from 'node:os' -import path from 'node:path' +import path from 'path' const QUEUE_FILE = path.join( os.homedir(), '.claude', - 'csc-raw-dump-queue.jsonl', + 'raw-dump', + 'csc-work-queue.jsonl', ) -const LOCK_FILE = path.join(os.homedir(), '.claude', 'csc-raw-dump.lock') +const LOCK_FILE = path.join(os.homedir(), '.claude', 'raw-dump', 'csc-work-queue.lock') + +export const MAX_ATTEMPTS = 4 // 最多尝试次数 export interface QueueTask { sessionID: string messageID: string directory: string enqueuedAt: number + attemptCount: number } -/** - * 将任务追加到队列文件（JSONL 格式） - * 使用 fire-and-forget 异步写入，避免阻塞主进程 event loop - * 写入失败被静默吞掉，不影响主流程 - */ -export function enqueue(task: Omit): void { - const item: QueueTask = { ...task, enqueuedAt: Date.now() } - // 使用 fire-and-forget 异步写入，避免阻塞主进程 event loop - fs.appendFile(QUEUE_FILE, JSON.stringify(item) + '\n', 'utf-8').catch( - () => {}, - ) -} +// ----------- 内存中的队列 ----------- +let queue: QueueTask[] = [] +let queueLoaded = false /** - * 从队列文件读取所有待处理的 task - * - 按行解析 JSONL，每行一个 task - * - 解析失败或空文件返回空数组 + * 从磁盘加载队列到内存，只在进程启动时调用一次 */ -export async function readQueue(): Promise { +export async function loadQueue(): Promise { + if (queueLoaded) return try { const text = await fs.readFile(QUEUE_FILE, 'utf-8') - return text + queue = text .split('\n') .filter(Boolean) .map(line => { @@ -55,20 +54,52 @@ export async function readQueue(): Promise { }) .filter((t): t is QueueTask => t !== null) } catch { - return [] + queue = [] } + queueLoaded = true } /** - * 清空队列文件（truncate 为空） - * 在 batch worker 读取队列后立即调用，避免处理期间积压的任务在下一轮重复消费 + * 将内存中的队列同步写回磁盘（仅在入队/移除时调用） */ -export async function clearQueue(): Promise { - try { - await fs.writeFile(QUEUE_FILE, '', 'utf-8') - } catch { - // ignore - } +export async function flushQueue(): Promise { + const lines = queue.map(t => JSON.stringify(t)).join('\n') + '\n' + await fs.writeFile(QUEUE_FILE, lines, 'utf-8') +} + +/** + * 将任务追加到队列（内存 + 文件） + * 写入失败被静默吞掉，不影响主流程 + */ +export function enqueue(task: Omit): void { + const item: QueueTask = { ...task, enqueuedAt: Date.now(), attemptCount: 0 } + queue.push(item) + // 同步写文件，不阻塞主进程 event loop + fs.writeFile(QUEUE_FILE, JSON.stringify(item) + '\n', { flag: 'a', encoding: 'utf-8' }).catch(() => {}) +} + +/** + * 从内存队列中移除任务（成功或彻底失败时调用） + * @param key sessionID:messageID + */ +export function removeTask(key: string): void { + queue = queue.filter(t => `${t.sessionID}:${t.messageID}` !== key) + flushQueue().catch(() => {}) +} + +/** + * 消费队首任务（不移除，仅返回引用） + * 用于在处理前 peek，失败时自行更新 attemptCount 再写回 + */ +export function peekTask(): QueueTask | undefined { + return queue[0] +} + +/** + * 获取当前内存队列快照 + */ +export function getQueue(): QueueTask[] { + return [...queue] } /** @@ -82,10 +113,9 @@ export async function acquireLock(): Promise { const stat = await fs.readFile(LOCK_FILE, 'utf-8') const pid = parseInt(stat, 10) if (!isNaN(pid) && pid !== process.pid) { - // 检查进程是否还在运行 try { process.kill(pid, 0) - return false // 已有 worker 在运行 + return false } catch { // 进程已退出，可以抢占锁 } diff --git a/src/services/rawDump/state.ts b/src/services/rawDump/state.ts index 49802853d..2e2217da2 100644 --- a/src/services/rawDump/state.ts +++ b/src/services/rawDump/state.ts @@ -1,7 +1,11 @@ /** * Raw Dump 磁盘状态管理 - * 用于 conversation、summary 和 commits 的去重 + * 用于 conversation、summary、commits 的去重，以及错误聚合 * 通过文件锁保证多进程并发读写安全 + * + * state.errors：按 sessionID:messageID 键控的错误聚合表 + * - 每批次累加 count，记录最新错误消息和时间戳 + * - 超过最大重试次数时由 batchWorker 写入 dead letter */ import { promises as fs, readFileSync, writeFileSync } from 'node:fs' @@ -9,9 +13,9 @@ import os from 'node:os' import path from 'node:path' import type { RawDumpState } from './types.js' -const STATE_DIR = path.join(os.homedir(), '.claude') -const STATE_FILE = path.join(STATE_DIR, 'csc-raw-dump-state.json') -const STATE_LOCK_FILE = path.join(STATE_DIR, 'csc-raw-dump-state.lock') +const STATE_DIR = path.join(os.homedir(), '.claude', 'raw-dump') +const STATE_FILE = path.join(STATE_DIR, 'csc-state.json') +const STATE_LOCK_FILE = path.join(STATE_DIR, 'csc-state.lock') /** * 创建空的 RawDumpState 对象 @@ -22,6 +26,8 @@ function createEmptyState(): RawDumpState { conversation: {}, summary: {}, commits: {}, + errors: {}, + // summary 值为 RFC3339 格式时间戳字符串，解析时无需转换 } } @@ -100,6 +106,7 @@ export async function readState(): Promise { conversation: parsed.conversation ?? {}, summary: parsed.summary ?? {}, commits: parsed.commits ?? {}, + errors: parsed.errors ?? {}, } } catch { return createEmptyState() diff --git a/src/services/rawDump/types.ts b/src/services/rawDump/types.ts index 403d91a95..c6ca33479 100644 --- a/src/services/rawDump/types.ts +++ b/src/services/rawDump/types.ts @@ -1,6 +1,16 @@ /** * Raw Dump 上报类型定义 * 与框架解耦，不依赖任何 UI 或特定运行时 + * + * 三类上报任务： + * - conversation：对话明细（request/response/diff） + * - summary：会话统计（按 session 去重，5 分钟内只上报一次） + * - commits：提交记录（按 commit ID 去重） + * + * 错误追踪： + * - 上报失败的任务写入 failed queue，最多重试 MAX_RETRIES 次 + * - 超过最大重试次数的任务移入 dead letter 文件 + * - 所有错误聚合到 state.errors，按 sessionID:messageID 键控 */ export const RAW_DUMP_EVENT_ENV_KEY = '__CSC_RAW_DUMP_EVENT__' @@ -13,8 +23,16 @@ export interface RawDumpEventPayload { export interface RawDumpState { conversation: Record - summary: Record + summary: Record // RFC3339 时间戳，如 "2024-01-01T12:00:00.000Z" commits: Record + errors: Record +} + +export interface RawDumpError { + message: string + count: number + lastAt: string + endpoint?: string } export interface JwtPayload { diff --git a/src/services/rawDump/worker.ts b/src/services/rawDump/worker.ts index 083774b46..d14d40a9e 100644 --- a/src/services/rawDump/worker.ts +++ b/src/services/rawDump/worker.ts @@ -30,7 +30,7 @@ import { toCommitComment, } from './git.js' import { createLogger } from './logger.js' -import { isLocalDumpMode, writeLocalDump } from './localStorage.js' +import { getRawDumpMode, RAW_DUMP_MODE, writeLocalDump } from './localStorage.js' import { readState, writeState } from './state.js' import { RAW_DUMP_EVENT_ENV_KEY, type RawDumpEventPayload } from './types.js' import type { @@ -116,13 +116,21 @@ async function postJson( endpoint: string, body: object, ): Promise { - if (isLocalDumpMode()) { - const type = - endpoint === '/raw-store/task-conversation' - ? 'conversation' - : endpoint === '/raw-store/task-summary' - ? 'summary' - : 'commit' + const mode = getRawDumpMode() + + if (mode === RAW_DUMP_MODE.DISABLED) { + log.debug(`dump disabled, skipping ${endpoint}`) + return + } + + const type = + endpoint === '/raw-store/task-conversation' + ? 'conversation' + : endpoint === '/raw-store/task-summary' + ? 'summary' + : 'commit' + + if (mode === RAW_DUMP_MODE.LOCAL || mode === RAW_DUMP_MODE.BOTH) { await writeLocalDump(type, body as Record) const b = body as Record log.info(`local dump: ${type} saved`, { @@ -130,9 +138,10 @@ async function postJson( request_id: b.request_id, commit_id: b.commit_id, }) - return } + // REMOTE / BOTH 模式下继续执行 remote 上报逻辑 + const isAnonymous = !headers.get('Authorization') const url = getRawDumpUrl(baseUrl, endpoint, isAnonymous) log.debug(`POST ${endpoint}`, { url, isAnonymous }) @@ -935,7 +944,12 @@ export async function uploadSummary( }) const lastReported = state.summary[payload.sessionID] - if (lastReported && Date.now() - lastReported < SUMMARY_DEDUP_WINDOW_MS) { + if ( + lastReported && + Date.now() - + new Date(lastReported).getTime() < + SUMMARY_DEDUP_WINDOW_MS + ) { log.info('summary skipped: reported recently', { task_id: payload.sessionID, lastReported, @@ -966,7 +980,7 @@ export async function uploadSummary( '/raw-store/task-summary', body, ) - state.summary[payload.sessionID] = Date.now() + state.summary[payload.sessionID] = new Date().toISOString() log.info('summary uploaded', { task_id: payload.sessionID }) } @@ -1081,8 +1095,10 @@ export function getClaudeConfigHomeDir(): string { * 如 /Users/linkai/code/csc → -Users-linkai-code-csc */ function normalizeProjectPath(dir: string): string { - // 将 /Users/linkai/code/csc 转换为 -Users-linkai-code-csc - return dir.replace(/\//g, '-') + // 将路径中的路径分隔符替换为 -，统一处理 / 和 \ (Windows) + // 如 /Users/linkai/code/csc → -Users-linkai-code-csc + // 如 D:\shenma\zgsm-ai\csc → D--shenma-zgsm-ai-csc (drive letter 后的 \ 也转为 -) + return dir.replace(/:/, '-').replace(/[\/\\]/g, '-') } /** @@ -1210,7 +1226,7 @@ export async function runRawDumpWorker() { /** * 认证兜底逻辑，优先尝试正常认证，失败后根据模式降级： - * - 本地模式（isLocalDumpMode）：使用假的匿名凭证，允许本地存储模式运行 + * - 本地模式（mode >= 2）：使用假的匿名凭证，允许本地存储模式运行 * - 非本地模式：降级为匿名接口（无 Authorization header） * 确保即使认证失败，上报流程仍可继续 */ @@ -1220,7 +1236,7 @@ export async function authWithFallback(): Promise< try { return await auth() } catch (err) { - if (isLocalDumpMode()) { + if (getRawDumpMode() >= RAW_DUMP_MODE.LOCAL) { log.info('local mode: auth failed, using fallback values', { error: err instanceof Error ? err.message : String(err), })