github/claude-code-best
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1. 增加统计信息并上报逻辑，用于对账；2. 上报失败超过最大重试次数的消息记录到dead letter jsonl中

Browse Source
This commit is contained in:
zbc 2026-05-27 10:49:21 +08:00
parent 455489cc89
commit 14d8e59d8c
7 changed files with 201 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
docs/config-list.md
View File

@ -194,8 +194,8 @@
| `CSC_RAW_DUMP_BASE_URL` | string | - | 自定义上报服务端地址 |
| `COSTRICT_RAW_DUMP_BASE_URL` | string | - | 兼容 opencode 的自定义上报地址 |
| `COSTRICT_BASE_URL` | string | `https://zgsm.sangfor.com` | CoStrict 服务地址 |
| `CSC_RAW_DUMP_LOCAL_MODE` | boolean | `false` | **本地留存模式**：数据只写入本地文件，不上报服务端 |
| `CSC_RAW_DUMP_LOCAL_DIR` | string | `~/.claude/raw-dump-local` | 本地留存目录 |
| `CSC_RAW_DUMP_MODE` | boolean | `false` | **本地留存模式**：数据只写入本地文件，不上报服务端 |
| `CSC_RAW_DUMP_DIR` | string | `~/.claude/raw-dump` | 本地留存目录 |
### Bash/终端

69
src/services/rawDump/batchWorker.ts
View File

@ -16,6 +16,7 @@ import {
uploadConversation,
uploadSummary,
uploadCommits,
uploadStatistics,
authWithFallback,
} from './worker.js'
import {
@ -29,8 +30,7 @@ import {
MAX_ATTEMPTS,
type QueueTask,
} from './queue.js'
import { readState, writeState } from './state.js'
import type { RawDumpError } from './types.js'
import { readState, writeState, appendDeadLetter } from './state.js'
import { getSessionDirectory, loadSessionMessages } from './worker.js'
import { getRepoInfo } from './git.js'
import { createLogger } from './logger.js'
@ -134,6 +134,25 @@ async function processTask(
const authData = await authWithFallback()
const repoInfo = await getCachedRepoInfo(task.directory)
if (task.messageID === '__statistics__' && task.statsData) {
await uploadStatistics(
{
sessionID: task.sessionID,
directory: task.directory,
sessionCount: task.statsData.sessionCount,
conversationCount: task.statsData.conversationCount,
upstreamTokens: task.statsData.upstreamTokens,
downstreamTokens: task.statsData.downstreamTokens,
startTime: task.statsData.startTime,
endTime: task.statsData.endTime,
},
authData,
state,
)
log.info('statistics task completed', { sessionID: task.sessionID })
return
}
const conversationUploaded = await uploadConversation(
{
sessionID: task.sessionID,
@ -156,7 +175,10 @@ async function processTask(
repoInfo,
})
log.info('task completed', { sessionID: task.sessionID, conversationUploaded })
log.info('task completed', {
sessionID: task.sessionID,
conversationUploaded,
})
}
/**
@ -218,13 +240,16 @@ async function runBatch() {
})
if (task.attemptCount >= MAX_ATTEMPTS) {
// 彻底失败：记录错误，从队列移除
state.errors[key] = {
message: errorMsg,
count: task.attemptCount,
lastAt: new Date().toISOString(),
// 彻底失败：写入 dead letter，从队列移除
await appendDeadLetter({
sessionID: task.sessionID,
messageID: task.messageID,
directory: task.directory,
attemptCount: task.attemptCount,
error: errorMsg,
endpoint: extractEndpointFromError(errorMsg),
} satisfies RawDumpError
failedAt: new Date().toISOString(),
})
removeTask(key)
log.warn('task permanently failed, removed from queue', {
sessionID: task.sessionID,
@ -232,12 +257,6 @@ async function runBatch() {
})
} else {
// 失败但未超限：attemptCount++ 写回文件，下次重试
state.errors[key] = {
message: errorMsg,
count: task.attemptCount,
lastAt: new Date().toISOString(),
endpoint: extractEndpointFromError(errorMsg),
} satisfies RawDumpError
await flushQueue()
}
}
@ -279,15 +298,17 @@ export function startBatchWorker() {
}
// 启动时加载队列 + 随机抖动
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),
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)
})
process.exit(1)
})
}
const scriptPath = process.argv[1] || ''
@ -296,4 +317,4 @@ if (
scriptPath.endsWith('batchWorker.js')
) {
startBatchWorker()
}
}

44
src/services/rawDump/index.ts
View File

@ -3,7 +3,10 @@
* 队列模式：主进程只 enqueue，单 batch worker 顺序消费
*/
import { getRawDumpMode, RAW_DUMP_MODE } 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'
@ -103,3 +106,42 @@ export function reportSession(sessionID: string, directory: string): void {
enqueue({ sessionID, messageID: '__summary__', directory })
ensureBatchWorker()
}
export interface StatisticsData {
sessionCount: number
conversationCount: number
upstreamTokens: number
downstreamTokens: number
startTime: number
endTime: number
}
const lastReportStatsMap = new Map<string, number>()
const STATS_DEBOUNCE_MS = 60_000 // 同一 session 1 分钟内不重复 enqueue
/**
* 上报对账统计数据（session数、conversation数、token数）
* 通过特殊 messageID '__statistics__' 标识入队
*/
export function reportStatistics(
sessionID: string,
directory: string,
data: StatisticsData,
): void {
if (!isEnabled()) return
const key = `${sessionID}:__statistics__`
const now = Date.now()
const last = lastReportStatsMap.get(key)
if (last && now - last < STATS_DEBOUNCE_MS) {
log.debug('reportStatistics debounced', { sessionID, lastMs: now - last })
return
}
lastReportStatsMap.set(key, now)
enqueue({
sessionID,
messageID: '__statistics__',
directory,
statsData: data,
})
ensureBatchWorker()
}

2
src/services/rawDump/queue.ts
View File

@ -12,6 +12,7 @@
import { promises as fs } from 'node:fs'
import os from 'node:os'
import path from 'path'
import type { StatisticsData } from './index.js'
const QUEUE_FILE = path.join(
os.homedir(),
@ -29,6 +30,7 @@ export interface QueueTask {
directory: string
enqueuedAt: number
attemptCount: number
statsData?: StatisticsData
}
// ----------- 内存中的队列 -----------

34
src/services/rawDump/state.ts
View File

@ -1,21 +1,22 @@
/**
* Raw Dump 磁盘状态管理
* 用于 conversation、summary、commits 的去重，以及错误聚合
* 用于 conversation、summary、commits 的去重
* 通过文件锁保证多进程并发读写安全
*
* state.errors：按 sessionID:messageID 键控的错误聚合表
* - 每批次累加 count，记录最新错误消息和时间戳
* - 超过最大重试次数时由 batchWorker 写入 dead letter
* 错误追踪（dead letter）：
* - 超过最大重试次数的任务追加写入 dead letter jsonl 文件
* - 路径: ~/.claude/raw-dump/csc-dead-letter.jsonl
*/
import { promises as fs, readFileSync, writeFileSync } from 'node:fs'
import os from 'node:os'
import path from 'node:path'
import type { RawDumpState } from './types.js'
import type { RawDumpState, DeadLetterEntry } from './types.js'
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')
const DEAD_LETTER_FILE = path.join(STATE_DIR, 'csc-dead-letter.jsonl')
/**
* 创建空的 RawDumpState 对象
@ -26,8 +27,6 @@ function createEmptyState(): RawDumpState {
conversation: {},
summary: {},
commits: {},
errors: {},
// summary 值为 RFC3339 格式时间戳字符串，解析时无需转换
}
}
@ -106,7 +105,6 @@ export async function readState(): Promise<RawDumpState> {
conversation: parsed.conversation ?? {},
summary: parsed.summary ?? {},
commits: parsed.commits ?? {},
errors: parsed.errors ?? {},
}
} catch {
return createEmptyState()
@ -126,3 +124,23 @@ export async function writeState(state: RawDumpState): Promise<void> {
await fs.writeFile(STATE_FILE, JSON.stringify(state, null, 2), 'utf-8')
})
}
export async function readDeadLetter(): Promise<DeadLetterEntry[]> {
try {
const text = await fs.readFile(DEAD_LETTER_FILE, 'utf-8')
return text
.split('\n')
.filter(Boolean)
.map(line => JSON.parse(line) as DeadLetterEntry)
} catch {
return []
}
}
export async function appendDeadLetter(entry: DeadLetterEntry): Promise<void> {
await fs.mkdir(STATE_DIR, { recursive: true })
await fs.writeFile(DEAD_LETTER_FILE, JSON.stringify(entry) + '\n', {
flag: 'a',
encoding: 'utf-8',
})
}

33
src/services/rawDump/types.ts
View File

@ -7,10 +7,9 @@
* - summary：会话统计（按 session 去重，5 分钟内只上报一次）
* - commits：提交记录（按 commit ID 去重）
*
* 错误追踪：
* - 上报失败的任务写入 failed queue，最多重试 MAX_RETRIES 次
* - 超过最大重试次数的任务移入 dead letter 文件
* - 所有错误聚合到 state.errors，按 sessionID:messageID 键控
* 错误追踪（dead letter）：
* - 超过最大重试次数的任务追加写入 dead letter jsonl 文件
* - 路径: ~/.claude/raw-dump/csc-dead-letter.jsonl
*/
export const RAW_DUMP_EVENT_ENV_KEY = '__CSC_RAW_DUMP_EVENT__'
@ -25,14 +24,16 @@ export interface RawDumpState {
conversation: Record<string, true>
summary: Record<string, string> // RFC3339 时间戳，如 "2024-01-01T12:00:00.000Z"
commits: Record<string, string>
errors: Record<string, RawDumpError>
}
export interface RawDumpError {
message: string
count: number
lastAt: string
export interface DeadLetterEntry {
sessionID: string
messageID: string
directory: string
attemptCount: number
error: string
endpoint?: string
failedAt: string
}
export interface JwtPayload {
@ -107,3 +108,17 @@ export interface CommitPayload {
subject: string
parent_ids: string[]
}
export interface StatisticsPayload {
task_id: string
start_time: string
end_time: string
user_id: string
user_name: string
client_id: string
client_version: string
session_count: number
conversation_count: number
upstream_tokens: number
downstream_tokens: number
}

59
src/services/rawDump/worker.ts
View File

@ -37,6 +37,7 @@ import type {
CommitPayload,
ConversationPayload,
JwtPayload,
StatisticsPayload,
SummaryPayload,
} from './types.js'
@ -1072,6 +1073,64 @@ export async function uploadCommits(
return commits.length
}
const STATS_DEDUP_WINDOW_MS = 60 * 60 * 1000 // 同一 session 1 小时内只上报一次 statistics
export async function uploadStatistics(
payload: {
sessionID: string
directory: string
sessionCount: number
conversationCount: number
upstreamTokens: number
downstreamTokens: number
startTime: number
endTime: number
},
authData: Awaited<ReturnType<typeof auth>>,
state: Awaited<ReturnType<typeof readState>>,
): Promise<void> {
const key = `stats:${payload.sessionID}`
const lastReported = state.summary[key]
if (
lastReported &&
Date.now() - new Date(lastReported).getTime() < STATS_DEDUP_WINDOW_MS
) {
log.debug('statistics skipped: reported recently', {
task_id: payload.sessionID,
lastReported,
})
return
}
const body: StatisticsPayload = {
task_id: payload.sessionID,
start_time: formatIso(payload.startTime),
end_time: formatIso(payload.endTime),
...authData.user,
client_id: authData.clientId,
client_version: authData.version,
session_count: payload.sessionCount,
conversation_count: payload.conversationCount,
upstream_tokens: payload.upstreamTokens,
downstream_tokens: payload.downstreamTokens,
}
await postJson(
authData.baseUrl,
authData.headers,
'/raw-store/statistics',
body,
)
state.summary[key] = new Date().toISOString()
log.info('statistics uploaded', {
task_id: payload.sessionID,
session_count: payload.sessionCount,
conversation_count: payload.conversationCount,
upstream_tokens: payload.upstreamTokens,
downstream_tokens: payload.downstreamTokens,
})
}
/**
* 从环境变量中解析 worker 入口传入的 payload
* 抛出错误如果环境变量不存在或 JSON 解析失败