f3ece2f469
Replace per-turn detached workers with a file-backed queue consumed by
a single long-running batch worker (30s interval, 0-10s jitter).
- queue.ts, batchWorker.ts: file queue with pid-based lock for worker
singleton; tasks deduped by sessionID:messageID before processing
- worker.ts: retry with backoff (5s, 10s) for 429 and network errors;
update commit state per upload so partial failures resume cleanly;
export auth, loadSessionMessages, getSessionDirectory and upload*
helpers for batchWorker reuse
- git.ts: cap commit log at 50 entries within last 7 days; pause 500ms
every 10 commits to spread load
- worker.ts: resolve session jsonl via ~/.claude/projects/{normalized}
with fallbacks, scanning for the file containing the target session
- logger.ts: file + stderr logger gated by CSC_RAW_DUMP_DEBUG, default
silent
- sessionDataUploader.ts: implement createSessionTurnUploader to pick
the last assistant message; query.ts fires uploadSessionTurn after
query_api_streaming_end (non-blocking)
Signed-off-by: 林凯90331 <90331@sangfor.com>
Co-authored-by: CoStrict <zgsm@sangfor.com.cn>
114 lines
3.4 KiB
TypeScript
114 lines
3.4 KiB
TypeScript
/**
|
||
* Raw Dump Batch Worker
|
||
* 顺序消费队列，避免并发 429
|
||
* 独立进程，通过 setInterval 定期执行
|
||
*/
|
||
|
||
import { uploadConversation, uploadSummary, uploadCommits, auth } from './worker.js'
|
||
import { readQueue, clearQueue, acquireLock, releaseLock, type QueueTask } from './queue.js'
|
||
import { readState, writeState } from './state.js'
|
||
import { getSessionDirectory, loadSessionMessages } from './worker.js'
|
||
import { createLogger } from './logger.js'
|
||
|
||
const log = createLogger('raw-dump-batch')
|
||
|
||
const BATCH_INTERVAL_MS = 30_000 // 30 秒检查一次队列
|
||
|
||
async function processTask(task: QueueTask) {
|
||
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)
|
||
|
||
if (messages.length === 0) {
|
||
log('warn', 'no messages found', { sessionDir, sessionID: task.sessionID })
|
||
}
|
||
|
||
const authData = await auth()
|
||
const state = await readState()
|
||
|
||
try {
|
||
// conversation
|
||
const conversationUploaded = await uploadConversation(
|
||
{ sessionID: task.sessionID, messageID: task.messageID, directory: task.directory, messages },
|
||
authData,
|
||
state,
|
||
)
|
||
|
||
// summary（每个 turn 都报，但内容会累积）
|
||
await uploadSummary({ sessionID: task.sessionID, directory: task.directory, messages }, authData)
|
||
|
||
// commits（限制频率，避免重复上报）
|
||
await uploadCommits({ directory: task.directory }, authData, state)
|
||
|
||
log('info', 'task completed', { sessionID: task.sessionID, conversationUploaded })
|
||
} finally {
|
||
// 无论成功或失败，都写入 state（commits 已逐条更新）
|
||
await writeState(state)
|
||
}
|
||
}
|
||
|
||
async function runBatch() {
|
||
if (!acquireLock()) {
|
||
log('debug', 'another worker is running, skip')
|
||
return
|
||
}
|
||
|
||
try {
|
||
const tasks = readQueue()
|
||
if (tasks.length === 0) {
|
||
log('debug', 'queue empty')
|
||
return
|
||
}
|
||
|
||
log('info', `processing ${tasks.length} tasks`)
|
||
|
||
// 去重：同一个 session 的多个 task，只保留最新的一个
|
||
const deduped = new Map<string, QueueTask>()
|
||
for (const task of tasks) {
|
||
const key = `${task.sessionID}:${task.messageID}`
|
||
const existing = deduped.get(key)
|
||
if (!existing || task.enqueuedAt > existing.enqueuedAt) {
|
||
deduped.set(key, task)
|
||
}
|
||
}
|
||
|
||
const uniqueTasks = Array.from(deduped.values()).sort((a, b) => a.enqueuedAt - b.enqueuedAt)
|
||
log('info', `deduped to ${uniqueTasks.length} unique tasks`)
|
||
|
||
for (const task of uniqueTasks) {
|
||
try {
|
||
await processTask(task)
|
||
} catch (err) {
|
||
log('error', 'task failed', { error: err instanceof Error ? err.message : String(err), sessionID: task.sessionID })
|
||
}
|
||
}
|
||
|
||
clearQueue()
|
||
log('info', 'batch completed')
|
||
} finally {
|
||
releaseLock()
|
||
}
|
||
}
|
||
|
||
export function startBatchWorker() {
|
||
log('info', 'batch worker started', { interval: BATCH_INTERVAL_MS })
|
||
|
||
// 立即执行一次
|
||
void runBatch()
|
||
|
||
// 定期执行，添加随机抖动避免规律性 429
|
||
const jitter = Math.floor(Math.random() * 10_000)
|
||
setTimeout(() => {
|
||
void runBatch()
|
||
setInterval(() => {
|
||
void runBatch()
|
||
}, BATCH_INTERVAL_MS)
|
||
}, jitter)
|
||
}
|
||
|
||
// 如果直接运行此文件
|
||
if (process.argv[1]?.includes('batchWorker')) {
|
||
startBatchWorker()
|
||
}
|