authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-03 13:03:09-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-03 13:03:09-07:00
log71dbaf1f94b580491d033f2e347a9b1c507fd793
treeb487f2087a351fb4e0eeb6ed27b21b8b0fcb7ff9
parent39efcd855833a1486ff7662e2efcedeeee6be0cc
signature Signed by PGP key B5690EEEBB952194

perf(usage): cut warm usage scans from seconds to milliseconds on large histories (#15149)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

9 files changed, 579 insertions(+), 131 deletions(-)

apps/server/src/usage/UsageService.test.ts+187
......@@ -1,5 +1,6 @@
11// @effect-diagnostics nodeBuiltinImport:off - the suite seeds and grows real
22// transcript trees on disk, outside the service's Effect FileSystem.
3import * as NodeChildProcess from "node:child_process";
34import * as NodeFSP from "node:fs/promises";
45import * as NodeOS from "node:os";
56import * as NodePath from "node:path";
......@@ -117,6 +118,41 @@ const serviceLayers = (input: {
117118 ),
118119 );
119120
121/** Outside `NARROW_WINDOW`, inside `WINDOW`. Seconds, as `utimes` takes them. */
122const BEFORE_NARROW_WINDOW = Date.parse("2026-08-01T10:00:00Z") / 1000;
123const NARROW_WINDOW: UsageSummaryInput = {
124 timeZone: "UTC",
125 sinceDay: UsageDay.make("2026-09-01"),
126 untilDay: UsageDay.make("2026-09-02"),
127};
128
129/**
130 * A FIFO named like a transcript. A scan's read of it waits in `open` until
131 * `openGate`, then fails at once, so a gate holds that scan's directory reads
132 * in flight. Each waiting gate holds one libuv pool thread; keep at most three.
133 */
134const makeGate = (path: string, lastWriteSeconds?: number) =>
135 Effect.promise(async () => {
136 NodeChildProcess.execFileSync("mkfifo", [path]);
137 if (lastWriteSeconds !== undefined) {
138 await NodeFSP.utimes(path, lastWriteSeconds, lastWriteSeconds);
139 }
140 });
141
142/**
143 * Returns once a scan has opened the gate. A scan opens every file of a
144 * directory at once, so it then holds the directory's transcripts open too.
145 */
146const openGate = (path: string) =>
147 Effect.promise(async () => (await NodeFSP.open(path, "w")).close());
148
149/** Replaces a file by rename, so a scan holding the old one keeps reading it. */
150const replaceFile = (path: string, content: string) =>
151 Effect.promise(async () => {
152 await NodeFSP.writeFile(path + ".next", content);
153 await NodeFSP.rename(path + ".next", path);
154 });
155
120156function totalOutputTokens(summary: { buckets: readonly { totals: { outputTokens: number } }[] }) {
121157 return summary.buckets.reduce((sum, bucket) => sum + bucket.totals.outputTokens, 0);
122158}
......@@ -972,6 +1008,157 @@ describe("UsageService", () => {
9721008 }).pipe(Effect.scoped),
9731009 );
9741010
1011 it.live("credits the same copy of a duplicate after its transcripts are deleted", () =>
1012 Effect.gen(function* () {
1013 const { transcript, settings, home } = yield* setup;
1014 const dir = NodePath.dirname(transcript);
1015 // The walk-first file is the original, padded so it finishes parsing
1016 // after the small fork copy that repeats its record under a new session.
1017 const [first = "", second = ""] = yield* Effect.promise(async () => {
1018 await NodeFSP.writeFile(NodePath.join(dir, "a.jsonl"), "");
1019 await NodeFSP.writeFile(NodePath.join(dir, "b.jsonl"), "");
1020 return (await NodeFSP.readdir(dir)).map((name) => NodePath.join(dir, name));
1021 });
1022 const forked = (line: string) => line.replace('"session-1"', '"session-2"');
1023 yield* Effect.promise(async () => {
1024 await NodeFSP.writeFile(
1025 first,
1026 claudeLine(1, 5).replace(
1027 '"message":',
1028 '"padding":' + encodeUnknownJsonString("x".repeat(9 * 1024 * 1024)) + ',"message":',
1029 ),
1030 );
1031 await NodeFSP.writeFile(second, forked(claudeLine(1, 5)) + forked(claudeLine(2, 7)));
1032 });
1033 yield* Effect.gen(function* () {
1034 const service = yield* UsageService.make;
1035 const live = yield* service.readSummary(WINDOW);
1036 assert.strictEqual(live.buckets[0]?.sessions, 2);
1037
1038 yield* Effect.promise(() => Promise.all([NodeFSP.rm(first), NodeFSP.rm(second)]));
1039 const saved = yield* service.readSummary(WINDOW);
1040 assert.deepStrictEqual(saved.buckets, live.buckets);
1041 assert.deepStrictEqual(saved.sources, live.sources);
1042 const restored = yield* (yield* UsageService.make).readSummary(WINDOW);
1043 assert.deepStrictEqual(restored.buckets, live.buckets);
1044 }).pipe(
1045 Effect.provide(serviceLayers({ prefix: "usage-service-copy-order-test", home, settings })),
1046 );
1047 }).pipe(Effect.scoped),
1048 );
1049
1050 it.live.skipIf(HostProcessPlatform.defaultValue() === "win32")(
1051 "keeps a newer cached read when a slower scan of another window finishes later",
1052 () =>
1053 Effect.gen(function* () {
1054 const { transcript, settings, home } = yield* setup;
1055 const dir = NodePath.dirname(transcript);
1056 const probe = NodePath.join(dir, "probe.jsonl");
1057 const hold = NodePath.join(dir, "hold.jsonl");
1058 yield* Effect.promise(() => NodeFSP.writeFile(transcript, claudeLine(1, 5)));
1059 yield* makeGate(probe, BEFORE_NARROW_WINDOW);
1060 yield* makeGate(hold, BEFORE_NARROW_WINDOW);
1061 yield* Effect.gen(function* () {
1062 const service = yield* UsageService.make;
1063 const wide = yield* service.readSummary(WINDOW).pipe(Effect.forkChild);
1064 yield* openGate(probe);
1065 yield* replaceFile(transcript, claudeLine(1, 5) + claudeLine(2, 7));
1066 // The narrow scan caches the newer read while the wide one waits.
1067 yield* service.readSummary(NARROW_WINDOW);
1068 yield* openGate(hold);
1069 yield* Fiber.join(wide);
1070
1071 yield* Effect.promise(() =>
1072 Promise.all([transcript, probe, hold].map((path) => NodeFSP.rm(path))),
1073 );
1074 assert.strictEqual(totalOutputTokens(yield* service.readSummary(WINDOW)), 12);
1075 }).pipe(
1076 Effect.provide(
1077 serviceLayers({ prefix: "usage-service-stale-read-test", home, settings }),
1078 ),
1079 );
1080 }).pipe(Effect.scoped),
1081 );
1082
1083 it.live.skipIf(HostProcessPlatform.defaultValue() === "win32")(
1084 "keeps the later read when a scan that read earlier finishes first",
1085 () =>
1086 Effect.gen(function* () {
1087 const { transcript, settings, home } = yield* setup;
1088 const dir = NodePath.dirname(transcript);
1089 const gate = (name: string) => NodePath.join(dir, `${name}.jsonl`);
1090 const wideProbe = gate("wide-probe");
1091 const wideHold = gate("wide-hold");
1092 const narrowProbe = gate("narrow-probe");
1093 const narrowHold = gate("narrow-hold");
1094 yield* Effect.promise(() => NodeFSP.writeFile(transcript, claudeLine(1, 5)));
1095 yield* makeGate(wideProbe, BEFORE_NARROW_WINDOW);
1096 yield* makeGate(wideHold, BEFORE_NARROW_WINDOW);
1097 yield* Effect.gen(function* () {
1098 const service = yield* UsageService.make;
1099 const wide = yield* service.readSummary(WINDOW).pipe(Effect.forkChild);
1100 yield* openGate(wideProbe);
1101 yield* replaceFile(transcript, claudeLine(1, 5) + claudeLine(2, 7));
1102 // Made after the wide scan's walk, so only the narrow scan waits on them.
1103 yield* makeGate(narrowProbe);
1104 yield* makeGate(narrowHold);
1105 const narrow = yield* service.readSummary(NARROW_WINDOW).pipe(Effect.forkChild);
1106 yield* openGate(narrowProbe);
1107 // Both scans started from an empty cache entry; the earlier read lands first.
1108 yield* openGate(wideHold);
1109 yield* Fiber.join(wide);
1110 yield* openGate(narrowHold);
1111 yield* Fiber.join(narrow);
1112
1113 yield* Effect.promise(() =>
1114 Promise.all(
1115 [transcript, wideProbe, wideHold, narrowProbe, narrowHold].map((path) =>
1116 NodeFSP.rm(path),
1117 ),
1118 ),
1119 );
1120 assert.strictEqual(totalOutputTokens(yield* service.readSummary(WINDOW)), 12);
1121 }).pipe(
1122 Effect.provide(serviceLayers({ prefix: "usage-service-late-read-test", home, settings })),
1123 );
1124 }).pipe(Effect.scoped),
1125 );
1126
1127 it.live("reports saved usage of a removed directory only for windows it reaches", () =>
1128 Effect.gen(function* () {
1129 const { transcript, settings, home } = yield* setup;
1130 yield* Effect.promise(async () => {
1131 await NodeFSP.writeFile(transcript, claudeLine(1, 5));
1132 const lastWrite = Date.parse("2026-08-01T10:00:00Z") / 1000;
1133 await NodeFSP.utimes(transcript, lastWrite, lastWrite);
1134 });
1135 const service = yield* UsageService.make.pipe(
1136 Effect.provide(
1137 serviceLayers({ prefix: "usage-service-saved-window-test", home, settings }),
1138 ),
1139 );
1140 const first = yield* service.readSummary(WINDOW);
1141 yield* Effect.promise(() =>
1142 NodeFSP.rm(NodePath.join(home, "claude", "projects"), { recursive: true }),
1143 );
1144
1145 const reached = yield* service.readSummary(WINDOW);
1146 assert.deepStrictEqual(reached.buckets, first.buckets);
1147 assert.strictEqual(reached.sources[0]?.status, "ok");
1148
1149 // A missing source cannot claim this directory from another environment
1150 // that still reads it, so it adds nothing to a window after its last write.
1151 const later = yield* service.readSummary({
1152 timeZone: "UTC",
1153 sinceDay: UsageDay.make("2026-08-10"),
1154 untilDay: UsageDay.make("2026-08-12"),
1155 });
1156 assert.deepStrictEqual(later.buckets, []);
1157 assert.strictEqual(later.sources[0]?.status, "missing");
1158 assert.strictEqual(later.sources[0]?.scannedFiles, 0);
1159 }).pipe(Effect.scoped),
1160 );
1161
9751162 it.live("does not share an in-flight scan after custom prices change", () =>
9761163 Effect.gen(function* () {
9771164 const { transcript, settings, home } = yield* setup;
apps/server/src/usage/UsageService.ts+122-47
......@@ -63,10 +63,11 @@ import {
6363import {
6464 decodeScanCache,
6565 dedupeWithinFile,
66 encodeScanCache,
6766 LEGACY_SCAN_CACHE_FILE_NAME,
67 makeScanCacheWriter,
6868 pruneScanCache,
6969 SCAN_CACHE_FILE_NAME,
70 type CachedFile,
7071 type ScanCache,
7172} from "./usageScanCache.ts";
7273import type { UsageRecord } from "./usageTranscripts.ts";
......@@ -90,6 +91,9 @@ const MAX_HOURLY_WINDOW_MS = 24 * 60 * 60 * 1000;
9091/** Longest window the UI offers, plus slack. Older entries are pruned. */
9192const CACHE_RETENTION_DAYS = 90;
9293
94/** Transcripts parsed at once. More gains little once the disk stays busy. */
95const TRANSCRIPT_READ_CONCURRENCY = 4;
96
9397const decodeCodexSettings = Schema.decodeOption(CodexSettings);
9498const decodeClaudeSettings = Schema.decodeOption(ClaudeSettings);
9599
......@@ -108,9 +112,37 @@ const encodeRatesCache = Schema.encodeEffect(
108112/** The scan cache is narrowed by hand in `usageScanCache`, so JSON is enough here. */
109113const ScanCacheJson = Schema.fromJsonString(Schema.Unknown as unknown as Schema.Codec<unknown>);
110114const decodeScanCacheFile = Schema.decodeUnknownEffect(ScanCacheJson);
111const encodeScanCacheFile = Schema.encodeEffect(ScanCacheJson);
112115const encodeUsageRecordKey = Schema.encodeSync(ScanCacheJson);
113116const CachedSource = Schema.Struct({ dir: Schema.String, volumeId: Schema.String });
117
118/** Whether `a` read a later state of its file than `b`. Transcripts only grow. */
119function isLaterRead(a: CachedFile, b: CachedFile): boolean {
120 return a.mtimeMs > b.mtimeMs || (a.mtimeMs === b.mtimeMs && a.size > b.size);
121}
122
123/**
124 * Codex sessions with records in more than one file, such as a rollout that
125 * moved after it was read. Only these need cross-file dedupe keys: within one
126 * file the occurrence count already keeps every key unique, so keying the rest
127 * would only build and hash a string for each of their records.
128 */
129function sharedCodexSessions(
130 files: readonly { readonly records: readonly UsageRecord[] }[],
131): ReadonlySet<string> {
132 const firstFile = new Map<string, number>();
133 const shared = new Set<string>();
134 for (const [index, file] of files.entries()) {
135 let previous = "";
136 for (const { provider, sessionId } of file.records) {
137 if (provider !== "codex" || sessionId === previous || sessionId.length === 0) continue;
138 previous = sessionId;
139 const first = firstFile.get(sessionId);
140 if (first === undefined) firstFile.set(sessionId, index);
141 else if (first !== index) shared.add(sessionId);
142 }
143 }
144 return shared;
145}
114146const decodeCachedSources = Schema.decodeUnknownOption(
115147 Schema.Struct({ sources: Schema.Record(Schema.String, CachedSource) }),
116148);
......@@ -386,20 +418,28 @@ export const make = Effect.gen(function* () {
386418 }),
387419 );
388420
421 const writeScanCache = makeScanCacheWriter();
422 // Scans with different windows can finish together; two writes interleaved
423 // in one file would corrupt it.
424 const persistLock = yield* Semaphore.make(1);
425
389426 const persistScanCache = Effect.fn("UsageService.persistScanCache")(function* () {
390427 if (!cacheDirty) return;
391 // Cleared only after the write lands, so a failed persist is retried on
392 // the next scan instead of leaving disk permanently stale.
393 yield* encodeScanCacheFile({
394 ...encodeScanCache(fileCache),
395 sources: Object.fromEntries(sourceCache),
396 }).pipe(
428 // Cleared before encoding, so a scan that changes the cache while this
429 // write is in flight marks it dirty again. A failed write restores the
430 // flag, so the next scan retries instead of leaving disk stale.
431 cacheDirty = false;
432 yield* Effect.sync(() =>
433 writeScanCache(fileCache, { sources: Object.fromEntries(sourceCache) }),
434 ).pipe(
397435 Effect.flatMap((serialized) => fileSystem.writeFileString(scanCachePath, serialized)),
398 Effect.map(() => {
399 cacheDirty = false;
400 }),
401436 // A cache we cannot write is a slower next start, not a failed read.
402 Effect.ignoreCause,
437 Effect.catchCause(() =>
438 Effect.sync(() => {
439 cacheDirty = true;
440 }),
441 ),
442 persistLock.withPermit,
403443 );
404444 });
405445
......@@ -410,13 +450,22 @@ export const make = Effect.gen(function* () {
410450 * written multi-hundred-megabyte rollout costs its appended bytes per scan
411451 * rather than a full re-read. The reader verifies the position's guard bytes
412452 * and silently restarts from byte 0 when they no longer match.
453 *
454 * A fresh parse comes back as `update` for the caller to cache, with the
455 * entry it was built from. Reads run concurrently, and the caller stores
456 * updates in walk order rather than completion order: saved records of
457 * deleted transcripts aggregate in cache order, where the first copy of a
458 * duplicate wins.
413459 */
414460 const readFileRecords = (
415461 filePath: string,
416462 size: number,
417463 mtimeMs: number,
418464 provider: UsageProviderKind,
419 ): Effect.Effect<readonly UsageRecord[]> =>
465 ): Effect.Effect<{
466 readonly records: readonly UsageRecord[];
467 readonly update?: { readonly entry: CachedFile; readonly replaces: CachedFile | undefined };
468 }> =>
420469 Effect.gen(function* () {
421470 const cached = fileCache.get(filePath);
422471 // Provider is part of the identity: if both providers were ever pointed
......@@ -427,9 +476,12 @@ export const make = Effect.gen(function* () {
427476 cached.mtimeMs === mtimeMs &&
428477 cached.provider === provider
429478 ) {
430 return cached.tailRecords.length === 0
431 ? cached.records
432 : [...cached.records, ...cached.tailRecords];
479 return {
480 records:
481 cached.tailRecords.length === 0
482 ? cached.records
483 : [...cached.records, ...cached.tailRecords],
484 };
433485 }
434486
435487 // Only a strictly grown file may resume. Same size with a new mtime, or
......@@ -445,7 +497,9 @@ export const make = Effect.gen(function* () {
445497 // A read failure is not an empty transcript: caching it under this
446498 // (size, mtime) would silently drop the file's usage until it changes.
447499 if (parsed === null)
448 return cached?.provider === provider ? [...cached.records, ...cached.tailRecords] : [];
500 return {
501 records: cached?.provider === provider ? [...cached.records, ...cached.tailRecords] : [],
502 };
449503
450504 // Stored already de-duplicated within the file, which is 99% of all
451505 // duplicates. The aggregator still runs the cross-file dedupe pass. One
......@@ -456,16 +510,13 @@ export const make = Effect.gen(function* () {
456510 const records = dedupeWithinFile([...base, ...parsed.records], seen);
457511 const tailRecords = dedupeWithinFile(parsed.tailRecords, seen);
458512
459 fileCache.set(filePath, {
460 size,
461 mtimeMs,
462 provider,
463 records,
464 tailRecords,
465 position: parsed.position,
466 });
467 cacheDirty = true;
468 return tailRecords.length === 0 ? records : [...records, ...tailRecords];
513 return {
514 records: tailRecords.length === 0 ? records : [...records, ...tailRecords],
515 update: {
516 entry: { size, mtimeMs, provider, records, tailRecords, position: parsed.position },
517 replaces: cached,
518 },
519 };
469520 });
470521
471522 /** One provider directory's walk and parse, before rates are involved. */
......@@ -505,11 +556,32 @@ export const make = Effect.gen(function* () {
505556 const files = yield* Effect.promise(() =>
506557 listTranscriptFiles(dir, windowStartMs, fileName === undefined ? undefined : { fileName }),
507558 );
508 const parsedFiles: { path: string; records: readonly UsageRecord[] }[] = [];
509 for (const file of files) {
510 const records = yield* readFileRecords(file.path, file.size, file.mtimeMs, provider);
511 parsedFiles.push({ path: file.path, records });
512 }
559 // A cold parse waits on disk reads, so a few files in flight read
560 // close to twice as fast. Results keep walk order.
561 const read = yield* Effect.forEach(
562 files,
563 (file) =>
564 readFileRecords(file.path, file.size, file.mtimeMs, provider).pipe(
565 Effect.map((result) => ({ path: file.path, ...result })),
566 ),
567 { concurrency: TRANSCRIPT_READ_CONCURRENCY },
568 );
569 const parsedFiles = read.map(({ path, records, update }) => {
570 if (update === undefined) return { path, records };
571 // A scan of another window may have cached its own read of this file
572 // meanwhile. Then keep whichever read saw the later file, so a slower
573 // scan never replaces newer usage with older.
574 const current = fileCache.get(path);
575 if (
576 current === update.replaces ||
577 current === undefined ||
578 !isLaterRead(current, update.entry)
579 ) {
580 fileCache.set(path, update.entry);
581 cacheDirty = true;
582 }
583 return { path, records };
584 });
513585 scanned.push({ provider, dir, volumeId, files: parsedFiles });
514586 }
515587
......@@ -753,37 +825,39 @@ export const make = Effect.gen(function* () {
753825
754826 const sources: UsageSource[] = [];
755827
756 for (const {
757 provider,
758 dir,
759 volumeId,
760 files,
761 status,
762 message,
763 action,
764 hostId: sourceHostId,
765 } of scannedDirs) {
828 // Cleanup may remove transcripts, but the usage we already saved still
829 // contributes to its source through the normal aggregation and dedupe
830 // path. Like the walk, skip files last written before the window: they
831 // cannot hold records inside it.
832 const retainedSinceMs = Math.max(windowStartMs, retentionCutoffMs);
833 const filesByDir = scannedDirs.map(({ provider, dir, files }) => {
766834 const retainedFiles = [...(files ?? [])];
767835 const livePaths = new Set(retainedFiles.map((file) => file.path));
768 // Cleanup may remove transcripts, but the usage we already saved still
769 // contributes to this source. Keep the normal aggregation and dedupe path.
770836 for (const [filePath, entry] of fileCache) {
771837 if (
772838 entry.provider !== provider ||
773 entry.mtimeMs < retentionCutoffMs ||
839 entry.mtimeMs < retainedSinceMs ||
774840 livePaths.has(filePath) ||
775841 !isWithinDirectory(filePath, dir)
776842 )
777843 continue;
778844 retainedFiles.push({ path: filePath, records: [...entry.records, ...entry.tailRecords] });
779845 }
846 return retainedFiles;
847 });
848 const sharedSessions = sharedCodexSessions(filesByDir.flat());
849
850 for (const [
851 index,
852 { provider, dir, volumeId, files, status, message, action, hostId: sourceHostId },
853 ] of scannedDirs.entries()) {
780854 let scannedFiles = 0;
781855 let skippedFiles = 0;
782856 // Distinct per directory. Buckets carry per-cell session counts, but a
783857 // session spans days and models, so clients total this figure instead.
784858 const sessionIds = new Set<string>();
785859
786 for (const file of retainedFiles) {
860 for (const file of filesByDir[index] ?? []) {
787861 if (file.records.length === 0) {
788862 skippedFiles += 1;
789863 continue;
......@@ -792,9 +866,10 @@ export const make = Effect.gen(function* () {
792866 const codexEventOccurrences = new Map<string, number>();
793867 for (const record of file.records) {
794868 let usageRecord = record;
795 if (record.provider === "codex" && record.sessionId.length > 0) {
869 if (record.provider === "codex" && sharedSessions.has(record.sessionId)) {
796870 // Match moved rollout copies without collapsing repeated equal events
797871 // within one rollout (timestamps can have only second precision).
872 // Only sessions seen in several files can have a copy to match.
798873 const key = encodeUsageRecordKey([
799874 record.provider,
800875 record.sessionId,
apps/server/src/usage/usageAggregation.test.ts+35
......@@ -128,6 +128,41 @@ describe("UsageAggregator", () => {
128128 expect(losAngeles.buckets[0]?.day).toBe("2026-08-06");
129129 });
130130
131 it("finds the day at a quarter-hour zone's midnight across interleaved buckets", () => {
132 // Kathmandu is UTC+5:45, so its midnight falls at 18:15 UTC.
133 const result = aggregate(
134 [
135 record({ timestampMs: Date.parse("2026-08-01T18:14:59.999Z") }),
136 record({ timestampMs: Date.parse("2026-08-01T18:15:00.000Z") }),
137 record({ timestampMs: Date.parse("2026-08-01T18:15:00.000Z"), model: "claude-opus-5" }),
138 record({ timestampMs: Date.parse("2026-08-01T18:16:00.000Z") }),
139 ],
140 "Asia/Kathmandu",
141 );
142
143 expect(result.buckets.map((bucket) => [bucket.day, bucket.model, bucket.records])).toEqual([
144 ["2026-08-01", "claude-fable-5", 1],
145 ["2026-08-02", "claude-fable-5", 2],
146 ["2026-08-02", "claude-opus-5", 1],
147 ]);
148 });
149
150 it("finds the day at a fixed offset's midnight between quarter hours", () => {
151 // At +00:01, midnight falls at 23:59 UTC.
152 const result = aggregate(
153 [
154 record({ timestampMs: Date.parse("2026-08-01T23:58:59.999Z") }),
155 record({ timestampMs: Date.parse("2026-08-01T23:59:00.000Z") }),
156 ],
157 "+00:01",
158 );
159
160 expect(result.buckets.map((bucket) => [bucket.day, bucket.records])).toEqual([
161 ["2026-08-01", 1],
162 ["2026-08-02", 1],
163 ]);
164 });
165
131166 it("splits an hourly request into fixed buckets anchored to its exact start", () => {
132167 const result = aggregate(
133168 [
apps/server/src/usage/usageAggregation.ts+87-29
......@@ -20,7 +20,7 @@ import type {
2020 UsageTokenTotals,
2121} from "@t3tools/contracts";
2222
23import { addTotals, EMPTY_TOTALS, type UsageRecord } from "./usageTranscripts.ts";
23import { EMPTY_TOTALS, type UsageRecord } from "./usageTranscripts.ts";
2424import { cacheSavingsUsd, priceUsage, type RateTable } from "./usagePricing.ts";
2525
2626/**
......@@ -47,13 +47,30 @@ function makeDayFormatter(timeZone: string): (timestampMs: number) => string {
4747 day: "2-digit",
4848 });
4949 }
50 return (timestampMs) => format.format(new Date(timestampMs));
50 // Formatting once per quarter-hour slot keeps `Intl` (about 2µs a call) off
51 // the per-record path. A slot whose two ends fall on one day lies wholly in
52 // that day. Named zones put midnight on a quarter hour, so their slots never
53 // split; a fixed offset such as `+00:01` can, and records in a split slot are
54 // formatted exactly. `null` marks a split slot.
55 const days = new Map<number, string | null>();
56 return (timestampMs) => {
57 const slot = Math.floor(timestampMs / QUARTER_HOUR_MS);
58 let day = days.get(slot);
59 if (day === undefined) {
60 const first = format.format(new Date(slot * QUARTER_HOUR_MS));
61 const last = format.format(new Date((slot + 1) * QUARTER_HOUR_MS - 1));
62 day = first === last ? first : null;
63 days.set(slot, day);
64 }
65 return day ?? format.format(new Date(timestampMs));
66 };
5167}
5268
69const QUARTER_HOUR_MS = 15 * 60 * 1000;
5370const HOUR_MS = 60 * 60 * 1000;
5471
5572interface MutableBucket {
56 totals: UsageTokenTotals;
73 totals: { -readonly [K in keyof UsageTokenTotals]: number };
5774 costUsd: number;
5875 cacheSavingsUsd: number;
5976 categoryCostUsd: UsageCategoryCost | null;
......@@ -98,6 +115,14 @@ export class UsageAggregator {
98115 readonly #toDay: (timestampMs: number) => string;
99116 readonly #hourlyWindow: { readonly sinceTimeMs: number; readonly untilTimeMs: number } | null;
100117 readonly #options: AggregateOptions;
118 #lastBucket: {
119 readonly day: string;
120 readonly hourIndex: number;
121 readonly provider: string;
122 readonly model: string;
123 readonly source: string;
124 readonly bucket: MutableBucket;
125 } | null = null;
101126 #duplicatesDropped = 0;
102127 #outOfWindow = 0;
103128
......@@ -149,35 +174,20 @@ export class UsageAggregator {
149174 return false;
150175 }
151176
152 const hourStart =
177 const hourIndex =
153178 this.#hourlyWindow === null
154 ? ""
155 : new Date(
156 this.#hourlyWindow.sinceTimeMs +
157 Math.floor((record.timestampMs - this.#hourlyWindow.sinceTimeMs) / HOUR_MS) * HOUR_MS,
158 ).toISOString();
159 const key = `${day}\u0000${hourStart}\u0000${record.provider}\u0000${record.model}\u0000${sourcePath ?? ""}`;
160 let bucket = this.#buckets.get(key);
161 if (bucket === undefined) {
162 bucket = {
163 totals: EMPTY_TOTALS,
164 costUsd: 0,
165 cacheSavingsUsd: 0,
166 categoryCostUsd: null,
167 fastCostUsd: 0,
168 ultrafastCostUsd: 0,
169 speedPremiumUsd: 0,
170 records: 0,
171 unpricedRecords: 0,
172 providerReportedRecords: 0,
173 sessions: new Set<string>(),
174 };
175 this.#buckets.set(key, bucket);
176 }
179 ? -1
180 : Math.floor((record.timestampMs - this.#hourlyWindow.sinceTimeMs) / HOUR_MS);
181 const bucket = this.#bucketFor(day, hourIndex, record.provider, record.model, sourcePath ?? "");
177182
178183 const priced = priceUsage(this.#options.rates, record, this.#options.priceOverrides);
179184
180 bucket.totals = addTotals(bucket.totals, record.totals);
185 const totals = bucket.totals;
186 totals.uncachedInputTokens += record.totals.uncachedInputTokens;
187 totals.cachedInputTokens += record.totals.cachedInputTokens;
188 totals.cacheCreationTokens += record.totals.cacheCreationTokens;
189 totals.outputTokens += record.totals.outputTokens;
190 totals.reasoningTokens += record.totals.reasoningTokens;
181191 bucket.costUsd += priced.costUsd;
182192 if (priced.categoryCostUsd !== null) {
183193 const sum = bucket.categoryCostUsd;
......@@ -207,6 +217,54 @@ export class UsageAggregator {
207217 return true;
208218 }
209219
220 /**
221 * Records arrive file by file in time order, so most land in the bucket the
222 * previous record used. Checking that first skips building and hashing a
223 * key string per record.
224 */
225 #bucketFor(
226 day: string,
227 hourIndex: number,
228 provider: string,
229 model: string,
230 source: string,
231 ): MutableBucket {
232 const last = this.#lastBucket;
233 if (
234 last !== null &&
235 last.day === day &&
236 last.hourIndex === hourIndex &&
237 last.provider === provider &&
238 last.model === model &&
239 last.source === source
240 ) {
241 return last.bucket;
242 }
243 const window = this.#hourlyWindow;
244 const hourStart =
245 window === null ? "" : new Date(window.sinceTimeMs + hourIndex * HOUR_MS).toISOString();
246 const key = `${day}\u0000${hourStart}\u0000${provider}\u0000${model}\u0000${source}`;
247 let bucket = this.#buckets.get(key);
248 if (bucket === undefined) {
249 bucket = {
250 totals: { ...EMPTY_TOTALS },
251 costUsd: 0,
252 cacheSavingsUsd: 0,
253 categoryCostUsd: null,
254 fastCostUsd: 0,
255 ultrafastCostUsd: 0,
256 speedPremiumUsd: 0,
257 records: 0,
258 unpricedRecords: 0,
259 providerReportedRecords: 0,
260 sessions: new Set<string>(),
261 };
262 this.#buckets.set(key, bucket);
263 }
264 this.#lastBucket = { day, hourIndex, provider, model, source, bucket };
265 return bucket;
266 }
267
210268 finish(): AggregateResult {
211269 const buckets: UsageBucket[] = [];
212270 for (const [key, bucket] of this.#buckets) {
......@@ -222,7 +280,7 @@ export class UsageAggregator {
222280 provider: provider as UsageBucket["provider"],
223281 model,
224282 ...(sourcePath === "" ? {} : { sourcePath }),
225 totals: bucket.totals,
283 totals: { ...bucket.totals },
226284 costUsd: bucket.costUsd,
227285 cacheSavingsUsd: bucket.cacheSavingsUsd,
228286 // Zero and unknown figures are omitted to keep payloads small.
apps/server/src/usage/usagePricing.ts+20
......@@ -235,7 +235,27 @@ const UNPRICEABLE_MODELS = new Set([
235235 "fable",
236236]);
237237
238/**
239 * Lookups per table, by raw model name. A scan prices every record twice
240 * against a few dozen models, and tables are never mutated once built.
241 */
242const resolvedRates = new WeakMap<RateTable, Map<string, ModelRate | null>>();
243
238244export function lookupRate(table: RateTable, model: string): ModelRate | null {
245 let resolved = resolvedRates.get(table);
246 if (resolved === undefined) {
247 resolved = new Map();
248 resolvedRates.set(table, resolved);
249 }
250 let rate = resolved.get(model);
251 if (rate === undefined) {
252 rate = resolveRate(table, model);
253 resolved.set(model, rate);
254 }
255 return rate;
256}
257
258function resolveRate(table: RateTable, model: string): ModelRate | null {
239259 const key = stripVariantSuffix(normalizeRateKey(model));
240260 const bareName = bareModelName(key);
241261 if (bareName.length === 0 || UNPRICEABLE_MODELS.has(bareName)) return null;
apps/server/src/usage/usageScanCache.test.ts+21
......@@ -4,6 +4,7 @@ import {
44 decodeScanCache,
55 dedupeWithinFile,
66 encodeScanCache,
7 makeScanCacheWriter,
78 pruneScanCache,
89 type CachedFile,
910 type ScanCache,
......@@ -149,6 +150,26 @@ describe("scan cache round trip", () => {
149150 expect(decodeScanCache(JSON.parse(JSON.stringify(previous))).size).toBe(0);
150151 });
151152
153 it("rewrites only changed entries and still restores the whole cache", () => {
154 const write = makeScanCacheWriter();
155 const cache = cacheWith([
156 ["/a.jsonl", 100, [record()]],
157 ["/b.jsonl", 200, [record({ sessionId: "session-b" })]],
158 ]);
159 const sources = { "claude\u0000/projects": { dir: "/projects", volumeId: "1:2" } };
160 expect(decodeScanCache(JSON.parse(write(cache, { sources })))).toEqual(cache);
161
162 // The replacement adds intern entries; /a's memoised indexes must hold.
163 cache.set("/b.jsonl", {
164 ...cache.get("/b.jsonl")!,
165 size: 30,
166 records: [record({ sessionId: "session-c", model: "claude-opus-5-5", dedupeKey: "msg_3:" })],
167 });
168 const document = JSON.parse(write(cache, { sources }));
169 expect(decodeScanCache(document)).toEqual(cache);
170 expect(document.sources).toEqual(sources);
171 });
172
152173 it("interns repeated model and session strings", () => {
153174 const encoded = encodeScanCache(
154175 cacheWith([["/a.jsonl", 100, [record(), record({ dedupeKey: "msg_2:" }), record()]]]),
apps/server/src/usage/usageScanCache.ts+81-30
......@@ -103,26 +103,32 @@ interface SerializedCache {
103103 readonly files: Readonly<Record<string, SerializedFile>>;
104104}
105105
106/** Serialises the cache, interning the repeated model and session strings. */
107export function encodeScanCache(cache: ScanCache): SerializedCache {
108 const models: string[] = [];
109 const sessions: string[] = [];
110 const modelIndex = new Map<string, number>();
111 const sessionIndex = new Map<string, number>();
106/** Model and session strings, each stored once and referenced by index. */
107interface InternTables {
108 readonly models: string[];
109 readonly sessions: string[];
110 readonly modelIndex: Map<string, number>;
111 readonly sessionIndex: Map<string, number>;
112}
112113
113 const intern = (table: string[], index: Map<string, number>, value: string): number => {
114 const existing = index.get(value);
115 if (existing !== undefined) return existing;
116 const next = table.length;
117 table.push(value);
118 index.set(value, next);
119 return next;
120 };
114function makeInternTables(): InternTables {
115 return { models: [], sessions: [], modelIndex: new Map(), sessionIndex: new Map() };
116}
117
118function intern(table: string[], index: Map<string, number>, value: string): number {
119 const existing = index.get(value);
120 if (existing !== undefined) return existing;
121 const next = table.length;
122 table.push(value);
123 index.set(value, next);
124 return next;
125}
121126
127function serializeFile(entry: CachedFile, tables: InternTables): SerializedFile {
122128 const serializeRecord = (record: UsageRecord): SerializedRecord => [
123129 record.timestampMs,
124 intern(models, modelIndex, record.model),
125 intern(sessions, sessionIndex, record.sessionId),
130 intern(tables.models, tables.modelIndex, record.model),
131 intern(tables.sessions, tables.sessionIndex, record.sessionId),
126132 record.totals.uncachedInputTokens,
127133 record.totals.cachedInputTokens,
128134 record.totals.cacheCreationTokens,
......@@ -132,23 +138,68 @@ export function encodeScanCache(cache: ScanCache): SerializedCache {
132138 record.reportedCostUsd,
133139 SPEEDS.indexOf(record.speed),
134140 ];
141 return {
142 s: entry.size,
143 m: entry.mtimeMs,
144 p: entry.provider,
145 r: entry.records.map(serializeRecord),
146 t: entry.tailRecords.map(serializeRecord),
147 o: entry.position.resumeOffset,
148 gl: entry.position.guardLength,
149 gh: entry.position.guardHash,
150 cs: entry.position.codexState,
151 };
152}
135153
154/** Serialises the cache, interning the repeated model and session strings. */
155export function encodeScanCache(cache: ScanCache): SerializedCache {
156 const tables = makeInternTables();
136157 const files: Record<string, SerializedFile> = {};
137 for (const [path, entry] of cache) {
138 files[path] = {
139 s: entry.size,
140 m: entry.mtimeMs,
141 p: entry.provider,
142 r: entry.records.map(serializeRecord),
143 t: entry.tailRecords.map(serializeRecord),
144 o: entry.position.resumeOffset,
145 gl: entry.position.guardLength,
146 gh: entry.position.guardHash,
147 cs: entry.position.codexState,
148 };
149 }
158 for (const [path, entry] of cache) files[path] = serializeFile(entry, tables);
159 return {
160 version: USAGE_SCAN_CACHE_VERSION,
161 models: tables.models,
162 sessions: tables.sessions,
163 files,
164 };
165}
150166
151 return { version: USAGE_SCAN_CACHE_VERSION, models, sessions, files };
167/**
168 * Returns a function that serialises the cache to JSON text, re-encoding only
169 * the entries that changed since its last call. Call it once per persist.
170 *
171 * Writes the same document as `encodeScanCache`. Most entries never change
172 * between scans, and encoding all of them made each persist cost close to a
173 * second on a large cache. Entries are replaced, never mutated, when their file
174 * changes, so an entry's JSON is memoised by identity. The intern tables only
175 * grow, so a memoised entry's indexes stay valid; a pruned entry can leave an
176 * unused string behind until the next process start.
177 */
178export function makeScanCacheWriter(): (
179 cache: ScanCache,
180 extra: Readonly<Record<string, unknown>>,
181) => string {
182 const tables = makeInternTables();
183 const fragments = new WeakMap<CachedFile, string>();
184 return (cache, extra) => {
185 const files: string[] = [];
186 for (const [path, entry] of cache) {
187 let fragment = fragments.get(entry);
188 if (fragment === undefined) {
189 fragment = JSON.stringify(serializeFile(entry, tables));
190 fragments.set(entry, fragment);
191 }
192 files.push(`${JSON.stringify(path)}:${fragment}`);
193 }
194 // Encoded after the files, which may have added to the intern tables.
195 const head = JSON.stringify({
196 ...extra,
197 version: USAGE_SCAN_CACHE_VERSION,
198 models: tables.models,
199 sessions: tables.sessions,
200 });
201 return `${head.slice(0, -1)},"files":{${files.join(",")}}}`;
202 };
152203}
153204
154205function isRecordArray(value: unknown): value is readonly unknown[] {
apps/server/src/usage/usageTranscriptReader.ts+26-15
......@@ -85,6 +85,8 @@ export const GUARD_LENGTH = 64;
8585// readers; it never discards a record because of its size.
8686const STREAMING_THRESHOLD_BYTES = 8 * 1024 * 1024;
8787const NEWLINE = 0x0a;
88/** `stat` calls one transcript walk keeps in flight. The libuv pool has 4 threads. */
89const STAT_CONCURRENCY = 32;
8890const CARRIAGE_RETURN = 0x0d;
8991
9092type SelectedFields = { readonly [key: string]: true | SelectedFields };
......@@ -156,15 +158,19 @@ function fnv1a(buffer: Buffer): number {
156158 * `fileName` restricts the walk to a single basename (Grok's `updates.jsonl`).
157159 * Grok sessions also ship multi-megabyte `chat_history` and `events` logs that
158160 * never carry usage, so the basename filter keeps a cold scan off those files.
161 *
162 * Directories are listed depth-first, one at a time, then the candidates are
163 * stat'd by a fixed pool of workers: a warm scan stats thousands of files, and
164 * one at a time each waits its own trip through the thread pool. Results keep
165 * `readdir` order, which the aggregator's first-seen dedupe relies on.
159166 */
160167export async function listTranscriptFiles(
161168 root: string,
162169 sinceMs: number,
163170 options?: { readonly fileName?: string },
164171): Promise<readonly TranscriptFile[]> {
165 const found: TranscriptFile[] = [];
166172 const fileName = options?.fileName;
167
173 const candidates: string[] = [];
168174 const walk = async (dir: string): Promise<void> => {
169175 let entries;
170176 try {
......@@ -174,28 +180,33 @@ export async function listTranscriptFiles(
174180 }
175181 for (const entry of entries) {
176182 const child = NodePath.join(dir, entry.name);
177 if (entry.isDirectory()) {
178 await walk(child);
179 continue;
180 }
181 if (fileName !== undefined) {
182 if (entry.name !== fileName) continue;
183 } else if (!entry.name.endsWith(".jsonl")) {
184 continue;
183 if (entry.isDirectory()) await walk(child);
184 else if (fileName !== undefined ? entry.name === fileName : entry.name.endsWith(".jsonl")) {
185 candidates.push(child);
185186 }
187 }
188 };
189 await walk(root);
190
191 const found: Array<TranscriptFile | undefined> = Array.from({ length: candidates.length });
192 // Each worker pulls the next candidate from one shared iterator.
193 const queue = candidates.entries();
194 const statQueued = async (): Promise<void> => {
195 for (const [index, path] of queue) {
186196 try {
187 const stats = await NodeFSP.stat(child);
197 const stats = await NodeFSP.stat(path);
188198 if (stats.mtimeMs >= sinceMs) {
189 found.push({ path: child, size: stats.size, mtimeMs: stats.mtimeMs });
199 found[index] = { path, size: stats.size, mtimeMs: stats.mtimeMs };
190200 }
191201 } catch {
192202 // Vanished between readdir and stat.
193203 }
194204 }
195205 };
196
197 await walk(root);
198 return found;
206 await Promise.all(
207 Array.from({ length: Math.min(STAT_CONCURRENCY, candidates.length) }, statQueued),
208 );
209 return found.filter((file) => file !== undefined);
199210}
200211
201212/**
apps/server/src/usage/usageTranscripts.ts-10
......@@ -54,16 +54,6 @@ function parseTimestampMs(value: unknown): number | null {
5454 return Number.isNaN(parsed) ? null : parsed;
5555}
5656
57export function addTotals(a: UsageTokenTotals, b: UsageTokenTotals): UsageTokenTotals {
58 return {
59 uncachedInputTokens: a.uncachedInputTokens + b.uncachedInputTokens,
60 cachedInputTokens: a.cachedInputTokens + b.cachedInputTokens,
61 cacheCreationTokens: a.cacheCreationTokens + b.cacheCreationTokens,
62 outputTokens: a.outputTokens + b.outputTokens,
63 reasoningTokens: a.reasoningTokens + b.reasoningTokens,
64 };
65}
66
6757export function totalTokens(totals: UsageTokenTotals): number {
6858 // reasoningTokens is a subset of outputTokens and must not be added again.
6959 return (