From 9ce1de47ac57f022d42af83dd6a4f5eb8295995b Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 12 Nov 2026 09:16:49 +0800 Subject: [PATCH] fix(core): isolate usage queries from the engine loop Large ledger reads and cost aggregation ran synchronously inside WebUI requests, blocking event delivery and other HTTP routes. Run scans in a serialized, cancellable worker queue, coalesce concurrent identical queries, and show byte size without rescanning in storage listings. Preserve accounting, pending records and corrupt-line diagnostics. Full suite: 5211 passed, 4 skipped; typecheck and packed-package query passed. --- docs/webui.md | 2 + scripts/publish-package.ts | 1 + src/bot.ts | 5 +- src/core/core.ts | 1 + src/core/usage-log.ts | 36 ++++++++++++++- src/core/usage-query-worker.mjs | 22 +++++++++ src/core/usage-query.ts | 35 ++++++++++++++ tests/core/usage-query.test.ts | 82 +++++++++++++++++++++++++++++++++ tests/web/persona-usage.test.ts | 3 +- 9 files changed, 180 insertions(+), 7 deletions(-) create mode 100644 src/core/usage-query-worker.mjs create mode 100644 src/core/usage-query.ts create mode 100644 tests/core/usage-query.test.ts diff --git a/docs/webui.md b/docs/webui.md index 4b7441a96..70056377a 100644 --- a/docs/webui.md +++ b/docs/webui.md @@ -64,6 +64,8 @@ World 留在原位照常运行，名称旁换成重启进程的入口;「待生 根 URL 不带 hash 时跳转到 `live`；未知路由不显示页面内容，侧栏仍可导航。 存储项按归属分页:Core 的在运行诊断的「数据」子页,一键清空也在那里;World、Persona 与 Memory 页 各有「数据」页签,列本页声明的项。配置组同理:Core 的在运行诊断的「配置」子页,其余在声明方自己的页。 +用量账本在存储清单中显示文件大小。完整条数与费用按需在工作线程聚合，相同的并发查询共用一次读取， +不同查询串行读取；引擎停止时取消未结束的查询。损坏行保留在账本中，并通过 `ledger.error` 报告。 工具表是装配好的整份,只挂在 Persona 页的「工具表」页签上。 World 页与 Persona 页的 manifest 带 `undeclared`:运行中出现、不在定义自描述里的工具名与事件 type。 diff --git a/scripts/publish-package.ts b/scripts/publish-package.ts index 074a8fcaa..9838273c3 100644 --- a/scripts/publish-package.ts +++ b/scripts/publish-package.ts @@ -27,6 +27,7 @@ export const PUBLISHED_DEPENDENCIES = [ 'react-dom', 'react-textarea-autosize', 'tar-stream', + 'tsx', 'ws', ] as const; diff --git a/src/bot.ts b/src/bot.ts index 22a77943d..f7beae159 100644 --- a/src/bot.ts +++ b/src/bot.ts @@ -33,7 +33,6 @@ import { RESERVED_FRAME_NAMES } from './core/loop.ts'; import type { ResponseClient } from './core/generation.ts'; import { acquireInstanceLock, type InstanceLock } from './core/instance-lock.ts'; import { assembleSystemSegments, envPromptOverridePath, envPromptTemplateSource, renderWorldEnvPrompt, type EnvPromptDirs, type EnvPromptOrigin } from './core/prefix.ts'; -import { aggregateUsage } from './core/cost.ts'; import { createUsageRates } from './core/usage-rates.ts'; import { DeadlineError, nowIso, withDeadline, readTextFile } from './core/util.ts'; import { closeRun } from './core/run.ts'; @@ -250,7 +249,7 @@ function deriveStorage(core: Core, dataDir: string, lan kind: 'disk', location: 'data/usage.jsonl', note: s.usage.note, - stat: () => s.usage.stat(core.usageLog.count(), size(dataDir, 'usage.jsonl')), + stat: () => size(dataDir, 'usage.jsonl'), clear: () => s.usage.cleared(core.usageLog.clear()), }, { @@ -1043,7 +1042,7 @@ export function createBot( usage: { aggregate: async (opts) => { const { rates, sources } = await liveRates.resolve(); - return { ...aggregateUsage(core.usageLog.readAll(), { ...opts, timezone: core.config.timezone, rates }), rateSources: sources }; + return { ...await core.usageLog.aggregate({ ...opts, timezone: core.config.timezone, rates }), rateSources: sources }; }, status: () => core.usageLog.status(), }, diff --git a/src/core/core.ts b/src/core/core.ts index a8d3cb7e3..ac905c9c0 100644 --- a/src/core/core.ts +++ b/src/core/core.ts @@ -765,6 +765,7 @@ export class Core { failures.push(...moduleFailures); // 运行中卸载的 World 各自的 stop() 有期限,结果已记日志 await Promise.all(this.failedStops); + this.usageLog.stopQueries(); this.usageLog.flush(); const ledger = this.usageLog.status(); if (ledger.pending) failures.push({ worldId: 'core.usage', detail: `${ledger.pending} usage records remain unwritten: ${ledger.error}` }); diff --git a/src/core/usage-log.ts b/src/core/usage-log.ts index 96446c182..9db171378 100644 --- a/src/core/usage-log.ts +++ b/src/core/usage-log.ts @@ -2,6 +2,8 @@ import { appendFileSync, closeSync, existsSync, mkdirSync, openSync, readSync, writeFileSync } from 'node:fs'; import { dirname } from 'node:path'; import type { UsageRecord } from './types.ts'; +import type { UsageAggregate, UsageAggregateOptions } from './cost.ts'; +import { queryUsage } from './usage-query.ts'; /** The ledger is read in chunks so its size is not bounded by V8's maximum string length. */ const READ_CHUNK_BYTES = 4 * 1024 * 1024; @@ -31,6 +33,9 @@ export class UsageLog { private pending: UsageRecord[] = []; private error: string | null = null; private reconcile = false; + private queryAbort = new AbortController(); + private queryQueue: Promise = Promise.resolve(); + private queries = new Map>(); constructor(private readonly file: string, private readonly report: (message: string) => void = console.error) {} append(rec: UsageRecord): void { @@ -83,15 +88,42 @@ export class UsageLog { return out; } - readAll(): UsageRecord[] { + readAll(options: { strict?: boolean } = {}): UsageRecord[] { let records: UsageRecord[] = []; - try { records = this.readDisk(); } catch (error) { this.note(error); } + try { records = this.readDisk(); } catch (error) { + this.note(error); + if (options.strict) throw error; + } const ids = new Set(records.map(record => record.recordId ?? record.attempt?.id)); return [...records, ...this.pending.filter(record => !ids.has(record.recordId))]; } count(): number { return this.readAll().length; } + /** Concurrent identical queries share one scan; different queries scan serially. */ + aggregate(options: UsageAggregateOptions): Promise { + const key = JSON.stringify(options); + const existing = this.queries.get(key); + if (existing) return existing; + const query = this.queryQueue.then(async () => { + const result = await queryUsage({ file: this.file, pending: structuredClone(this.pending), options }, this.queryAbort.signal); + if (result.error) { + if (result.error !== this.error) this.report(result.error); + this.error = result.error; + } + return result.aggregate; + }).catch(error => { + if (!this.queryAbort.signal.aborted) this.note(error); + throw error; + }); + this.queries.set(key, query); + this.queryQueue = query.then(() => { this.queries.delete(key); }, () => { this.queries.delete(key); }); + return query; + } + + /** Pending WebUI scans must not delay engine shutdown. The ledger remains intact. */ + stopQueries(): void { this.queryAbort.abort(new Error('Usage queries stopped')); } + clear(): number { const count = this.count(); mkdirSync(dirname(this.file), { recursive: true }); diff --git a/src/core/usage-query-worker.mjs b/src/core/usage-query-worker.mjs new file mode 100644 index 000000000..be304c3a3 --- /dev/null +++ b/src/core/usage-query-worker.mjs @@ -0,0 +1,22 @@ +import { parentPort, workerData } from 'node:worker_threads'; +import { tsImport } from 'tsx/esm/api'; + +try { + const { UsageLog } = await tsImport('./usage-log.ts', import.meta.url); + const { aggregateUsage } = await tsImport('./cost.ts', import.meta.url); + const errors = []; + const log = new UsageLog(workerData.file, error => errors.push(error)); + const records = log.readAll({ strict: true }); + const ids = new Set(records.map(record => record.recordId ?? record.attempt?.id)); + for (const record of workerData.pending) { + const id = record.recordId ?? record.attempt?.id; + if (id && ids.has(id)) continue; + if (id) ids.add(id); + records.push(record); + } + parentPort.postMessage({ aggregate: aggregateUsage(records, workerData.options), error: errors.at(-1) ?? null }); +} catch (error) { + parentPort.postMessage({ failure: String(error) }); +} finally { + parentPort.close(); +} diff --git a/src/core/usage-query.ts b/src/core/usage-query.ts new file mode 100644 index 000000000..49bd4a9f1 --- /dev/null +++ b/src/core/usage-query.ts @@ -0,0 +1,35 @@ +/** Usage scans and aggregation run outside the engine event loop. */ +import { Worker } from 'node:worker_threads'; +import type { UsageRecord } from './types.ts'; +import type { UsageAggregate, UsageAggregateOptions } from './cost.ts'; + +export interface UsageQueryInput { + file: string; + pending: UsageRecord[]; + options: UsageAggregateOptions; +} + +export interface UsageQueryResult { + aggregate: UsageAggregate; + error: string | null; +} + +export function queryUsage(input: UsageQueryInput, signal: AbortSignal): Promise { + if (signal.aborted) return Promise.reject(signal.reason); + return new Promise((resolve, reject) => { + // The bootstrap installs its own TS loader; it must not inherit host/test preloads. + const worker = new Worker(new URL('./usage-query-worker.mjs', import.meta.url), { workerData: input, execArgv: [] }); + const abort = () => { void worker.terminate(); reject(signal.reason); }; + signal.addEventListener('abort', abort, { once: true }); + worker.once('message', (message: UsageQueryResult | { failure: string }) => { + if ('failure' in message) reject(new Error(message.failure)); + else resolve(message); + }); + worker.once('error', reject); + worker.once('messageerror', reject); + worker.once('exit', code => { + signal.removeEventListener('abort', abort); + reject(new Error(`Usage query worker exited before returning a result (${code})`)); + }); + }); +} diff --git a/tests/core/usage-query.test.ts b/tests/core/usage-query.test.ts new file mode 100644 index 000000000..d57281532 --- /dev/null +++ b/tests/core/usage-query.test.ts @@ -0,0 +1,82 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { appendFileSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { UsageLog } from '../../src/core/usage-log.ts'; +import { aggregateUsage } from '../../src/core/cost.ts'; +import type { UsageRecord } from '../../src/core/types.ts'; + +let dir: string; +let file: string; +const logs: UsageLog[] = []; +const row = (id: string): UsageRecord => ({ version: 2, recordId: id, ts: '2026-01-02T12:00:00Z', + sessionId: 'test', role: 'main', label: '测试', model: 'test-model', promptTokens: 100, + completionTokens: 10, cacheHitTokens: 50, cacheMissTokens: 50, reasoningTokens: 3 }); +const options = { bucket: 'day' as const, now: new Date('2026-01-02T13:00:00Z') }; +function log(path = file, report = (_message: string) => {}) { + const value = new UsageLog(path, report); logs.push(value); return value; +} +beforeEach(() => { dir = mkdtempSync(join(tmpdir(), 'usage-query-')); file = join(dir, 'usage.jsonl'); }); +afterEach(() => { for (const value of logs.splice(0)) value.stopQueries(); rmSync(dir, { recursive: true, force: true }); }); + +describe('UsageLog.aggregate', () => { + it('matches the existing accounting, preserves corrupt lines and deduplicates record ids', async () => { + const records = [row('a'), row('b')]; + const raw = records.map(record => JSON.stringify(record)).join('\n') + '\nbroken\n' + JSON.stringify(records[0]) + '\n'; + writeFileSync(file, raw); + const errors: string[] = []; + const usage = log(file, error => errors.push(error)); + expect(await usage.aggregate(options)).toEqual(aggregateUsage(records, options)); + expect(errors).toHaveLength(1); + expect(errors[0]).toContain('1 damaged ledger lines'); + expect(readFileSync(file, 'utf8')).toBe(raw); + }); + + it('leaves the main event loop available while a large query is pending', async () => { + const record = JSON.stringify({ ...row('large'), label: '资料'.repeat(1200) }); + writeFileSync(file, (record + '\n').repeat(2200)); + const usage = log(); + let settled = false; + const query = usage.aggregate(options).finally(() => { settled = true; }); + await new Promise(resolve => setImmediate(resolve)); + expect(settled).toBe(false); + usage.append(row('new')); + expect((await query).totals.calls).toBeGreaterThanOrEqual(1); + expect((await usage.aggregate(options)).totals.calls).toBe(2); + }); + + it('shares simultaneous identical scans and refreshes after an append', async () => { + const usage = log(); usage.append(row('a')); + const first = usage.aggregate(options); + expect(usage.aggregate(options)).toBe(first); + expect((await first).totals.calls).toBe(1); + usage.append(row('b')); + expect((await usage.aggregate(options)).totals.calls).toBe(2); + }); + + it('includes queued appends and does not count an already persisted retry twice', async () => { + const parent = join(dir, 'blocked'); writeFileSync(parent, 'occupied'); + const target = join(parent, 'usage.jsonl'); + const usage = log(target); usage.append(row('a')); usage.append(row('b')); + rmSync(parent); mkdirSync(parent); + appendFileSync(target, JSON.stringify(row('a')) + '\n'); + expect((await usage.aggregate(options)).totals.calls).toBe(2); + expect(usage.status().pending).toBe(2); + }); + + it('reports read failures instead of returning an empty successful report', async () => { + mkdirSync(file); + await expect(log().aggregate(options)).rejects.toThrow(); + }); + + it('cancels both active and queued scans when the engine stops', async () => { + const usage = log(); usage.append(row('a')); + const first = usage.aggregate(options); + const second = usage.aggregate({ ...options, bucket: 'hour' }); + const results = Promise.allSettled([first, second]); + await new Promise(resolve => setImmediate(resolve)); + usage.stopQueries(); + expect((await results).map(result => result.status)).toEqual(['rejected', 'rejected']); + expect(readFileSync(file, 'utf8')).toContain('"recordId":"a"'); + }); +}); diff --git a/tests/web/persona-usage.test.ts b/tests/web/persona-usage.test.ts index a67192d48..a9fbab7ee 100644 --- a/tests/web/persona-usage.test.ts +++ b/tests/web/persona-usage.test.ts @@ -4,7 +4,6 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { WebApp } from '../../src/web/server.ts'; import { UsageLog } from '../../src/core/usage-log.ts'; -import { aggregateUsage } from '../../src/core/cost.ts'; import { nullLogger } from '../../src/core/util.ts'; import type { UsageRecord } from '../../src/core/types.ts'; import type { BotConfig } from '../../bots/corti-soulmate/assemble.ts'; @@ -54,7 +53,7 @@ beforeAll(async () => { memoryDir, dataDir, getStatus: () => ({}), - usage: { aggregate: (opts) => aggregateUsage(usage.readAll(), opts) }, + usage: { aggregate: (opts) => usage.aggregate(opts) }, config: { groups: () => configGroups.map((group) => ({ group, values: readGroupValues(cfg, group) })), set: (groupId, values) => {