From c2cec0da398e74c72a01ff00dda9f9c94ed09347 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Mon, 2 Nov 2026 15:49:41 -0800 Subject: [PATCH 1/2] perf(server): warm the usage scan cache after startup The first usage summary after a restart on a fresh or stale cache paid for parsing every changed transcript. After activation the server now parses the default 30-day window into the persisted scan cache in the background; a summary arriving meanwhile waits for that warm-up instead of parsing the same files again. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/server.ts | 1 + apps/server/src/usage/UsageService.test.ts | 71 ++++++++++++++++++++++ apps/server/src/usage/UsageService.ts | 55 ++++++++++++++++- 3 files changed, 126 insertions(+), 1 deletion(-) diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 69dc339be3e23..7d580b7272c17 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -1173,6 +1173,7 @@ const layerMakeServer = Layer.unwrap( layerRuntimeState.pipe(Layer.provide(layerLauncher)), layerTailscaleServe, layerCloudDesiredLinkReconcile, + UsageService.layerWarmup, HeapSnapshot.layer, ); diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index 89f136eacf210..79c09305183a5 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -34,6 +34,7 @@ import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as Scheduler from "effect/Scheduler"; +import * as Tracer from "effect/Tracer"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; @@ -53,6 +54,7 @@ import * as AntigravityUsage from "../provider/Drivers/AntigravityUsage.ts"; import * as ProviderHostLive from "../provider/ProviderHostLive.ts"; import { totalTokens, type UsageRecord } from "@t3tools/provider-core/server/usage"; import * as UsageService from "./UsageService.ts"; +import * as ServerActivation from "../serverActivation.ts"; import * as UsageTranscriptParser from "./UsageTranscriptParser.ts"; import { TRANSCRIPT_USAGE_FORMATS } from "./usageTranscriptFormats.ts"; import { BUILT_IN_USAGE_DRIVERS } from "../provider/builtInDrivers.ts"; @@ -3255,6 +3257,75 @@ describe("UsageService", () => { }).pipe(Effect.scoped), ); + it.live("warms the scan cache after activation and shares it with summaries meanwhile", () => + Effect.gen(function* () { + const { home, transcript, settings } = yield* setup; + yield* Effect.promise(() => NodeFSP.writeFile(transcript, claudeLine(1, 5))); + const activation = yield* Deferred.make(); + const parseStarted = yield* Deferred.make(); + const releaseParse = yield* Deferred.make(); + const summaryWaiting = yield* Deferred.make(); + const secondParse = yield* Deferred.make(); + const parsedPaths: string[] = []; + const gatedParser = Layer.effect( + UsageTranscriptParser.UsageTranscriptParser, + Effect.gen(function* () { + const inProcess = yield* UsageTranscriptParser.UsageTranscriptParser; + return UsageTranscriptParser.UsageTranscriptParser.of({ + parse: (requests) => + requests.length === 0 + ? inProcess.parse(requests) + : Effect.gen(function* () { + parsedPaths.push(...requests.map((request) => request.path)); + yield* Deferred.succeed( + parsedPaths.length > 1 ? secondParse : parseStarted, + undefined, + ); + yield* Deferred.await(releaseParse); + return yield* inProcess.parse(requests); + }), + }); + }), + ).pipe(Layer.provide(UsageTranscriptParser.layerInProcess)); + const tracer = Tracer.make({ + span: (options) => { + if (options.name === "UsageService.awaitWarmup") { + Deferred.doneUnsafe(summaryWaiting, Effect.void); + } + return new Tracer.NativeSpan(options); + }, + }); + + yield* Effect.gen(function* () { + const service = yield* UsageService.make; + // Building the warm-up layer returns before activation, so startup never waits on it. + yield* Layer.build(UsageService.layerWarmup).pipe( + Effect.provideService(UsageService.UsageService, service), + Effect.provideService(ServerActivation.ServerActivation, Deferred.await(activation)), + ); + assert.deepStrictEqual(parsedPaths, []); + + yield* Deferred.succeed(activation, undefined); + yield* Deferred.await(parseStarted); + const summary = yield* service + .readSummary(WINDOW) + .pipe(Effect.withTracer(tracer), Effect.forkChild); + // Either the summary waits for the warm-up, or it starts its own parse of the same file. + yield* Effect.raceFirst(Deferred.await(summaryWaiting), Deferred.await(secondParse)); + assert.deepStrictEqual(parsedPaths, [transcript]); + yield* Deferred.succeed(releaseParse, undefined); + assert.strictEqual(totalOutputTokens(yield* Fiber.join(summary)), 5); + assert.deepStrictEqual(parsedPaths, [transcript]); + yield* service.awaitPersisted; + }).pipe( + Effect.scoped, + Effect.provide( + layerService({ prefix: "usage-warmup", home, settings, transcriptParser: gatedParser }), + ), + ); + }).pipe(Effect.scoped), + ); + it.live("does not orphan an in-flight scan when its first caller is interrupted", () => Effect.gen(function* () { const { settings, home } = yield* setup; diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index b851c42c4a920..773427db56165 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -31,6 +31,7 @@ import { writeFileStringAtomically } from "@t3tools/shared/atomicWrite"; import { formatMalformedUsageWarning } from "@t3tools/shared/usageFormat"; import * as ServerConfig from "../config.ts"; import * as ServerSettings from "../serverSettings.ts"; +import { forkParked } from "../serverActivation.ts"; import { BUILT_IN_USAGE_DRIVERS, type BuiltInUsageReadersEnv } from "../provider/builtInDrivers.ts"; import type { ProviderDriver } from "@t3tools/provider-core/server/driver"; import { mergeProviderInstanceEnvironment } from "@t3tools/provider-core/server/instanceEnvironment"; @@ -68,6 +69,9 @@ const MTIME_SLACK_MS = 36 * 60 * 60 * 1000; const MAX_HOURLY_WINDOW_MS = 24 * 60 * 60 * 1000; const CACHE_RETENTION_DAYS = 92; +const DAY_MS = 24 * 60 * 60 * 1000; +// The usage page opens on 30 days; a day more covers the client's local-midnight start. +const WARM_WINDOW_DAYS = 31; const transcriptReaders = BUILT_IN_USAGE_DRIVERS.flatMap((driver) => driver.usage?.kind === "transcripts" ? [{ driver, reader: driver.usage }] : [], @@ -138,6 +142,8 @@ export class UsageService extends Context.Service< { readonly readSummary: (input: UsageSummaryInput) => Effect.Effect; readonly refreshRates: Effect.Effect; + /** Parses transcripts of the default window into the scan cache; concurrent summaries await it. */ + readonly warmScanCache: Effect.Effect; } >()("t3/usage/UsageService") {} @@ -677,6 +683,13 @@ export const make = Effect.gen(function* () { return scanned; }); + let warming: Deferred.Deferred | undefined; + const awaitWarmup = Effect.suspend(() => + warming === undefined + ? Effect.void + : Deferred.await(warming).pipe(Effect.withSpan("UsageService.awaitWarmup")), + ); + const scanSummary = Effect.fn("UsageService.scanSummary")(function* ( input: UsageSummaryInput, settings: ServerSettingsValue, @@ -714,6 +727,8 @@ export const make = Effect.gen(function* () { const startedAtMs = yield* Clock.currentTimeMillis; yield* ensureScanCacheLoaded; + // Wait rather than parse the same cold files a second time. + yield* awaitWarmup; const windowStart = DateTime.make(`${input.sinceDay}T00:00:00Z`); if (Option.isNone(windowStart)) { @@ -955,7 +970,45 @@ export const make = Effect.gen(function* () { return yield* Deferred.await(deferred); }); - return { readSummary, refreshRates, awaitPersisted } as const; + const warmScanCache = Effect.gen(function* () { + // A summary already in flight warms the cache itself. + if (warming !== undefined || inflightScans.size > 0) return; + const done = Deferred.makeUnsafe(); + warming = done; + yield* Effect.gen(function* () { + yield* ensureScanCacheLoaded; + const settings = yield* readSettings; + const now = yield* Clock.currentTimeMillis; + const dirs = yield* resolveTranscriptDirs(settings, now - CACHE_RETENTION_DAYS * DAY_MS).pipe( + Effect.provideContext(readerContext), + ); + const windowStartMs = now - WARM_WINDOW_DAYS * DAY_MS - MTIME_SLACK_MS; + yield* Effect.forEach(dirs, (dir) => scanTranscriptDir(dir, windowStartMs), { + discard: true, + }); + yield* schedulePersist; + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Usage scan cache warm-up failed.", { cause }), + ), + Effect.ensuring( + Effect.suspend(() => { + warming = undefined; + return Deferred.succeed(done, undefined); + }), + ), + ); + }).pipe(Effect.withSpan("UsageService.warmScanCache")); + + return { readSummary, refreshRates, warmScanCache, awaitPersisted } as const; }); export const layer = Layer.effect(UsageService, make); + +/** Warms after activation so a cold cache never delays readiness or the first summary's parse. */ +export const layerWarmup = Layer.effectDiscard( + Effect.gen(function* () { + const usage = yield* UsageService; + yield* forkParked(usage.warmScanCache); + }), +); From b581f660b4e1a01f31c7d91c17c4160736b90376 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Mon, 2 Nov 2026 16:10:01 -0800 Subject: [PATCH 2/2] fix(server): warm-up waits for shorter in-flight summaries; fetch prices while waiting A summary in flight at activation used to make the warm-up skip for the server's lifetime, even when it covered only an hour or a day. The warm-up now waits for in-flight summaries and then warms the rest of the window from cache. A summary that waits on the warm-up now fetches prices in the meantime instead of afterwards. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/usage/UsageService.test.ts | 61 ++++++++++++++++++++++ apps/server/src/usage/UsageService.ts | 18 +++++-- 2 files changed, 75 insertions(+), 4 deletions(-) diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index 79c09305183a5..c08d9d7e8e645 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -23,6 +23,7 @@ import { } from "@t3tools/contracts"; import * as Duration from "effect/Duration"; import * as Clock from "effect/Clock"; +import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -3326,6 +3327,66 @@ describe("UsageService", () => { }).pipe(Effect.scoped), ); + it.live("warms the rest of the window after a shorter in-flight summary", () => + Effect.gen(function* () { + const { home, transcript, settings } = yield* setup; + const older = NodePath.join(NodePath.dirname(transcript), "older.jsonl"); + const nowMs = yield* Clock.currentTimeMillis; + yield* Effect.promise(async () => { + await NodeFSP.writeFile(transcript, claudeLine(1, 5)); + await NodeFSP.writeFile(older, claudeLine(2, 7, undefined, "session-2")); + const tenDaysAgoSeconds = (nowMs - 10 * 24 * 60 * 60 * 1000) / 1000; + await NodeFSP.utimes(older, tenDaysAgoSeconds, tenDaysAgoSeconds); + }); + const parseStarted = yield* Deferred.make(); + const releaseParse = yield* Deferred.make(); + const parsedPaths: string[] = []; + const gatedParser = Layer.effect( + UsageTranscriptParser.UsageTranscriptParser, + Effect.gen(function* () { + const inProcess = yield* UsageTranscriptParser.UsageTranscriptParser; + return UsageTranscriptParser.UsageTranscriptParser.of({ + parse: (requests) => + requests.length === 0 + ? inProcess.parse(requests) + : Effect.gen(function* () { + parsedPaths.push(...requests.map((request) => request.path)); + yield* Deferred.succeed(parseStarted, undefined); + yield* Deferred.await(releaseParse); + return yield* inProcess.parse(requests); + }), + }); + }), + ).pipe(Layer.provide(UsageTranscriptParser.layerInProcess)); + const today = UsageDay.make(DateTime.formatIsoDateUtc(DateTime.makeUnsafe(nowMs))); + + yield* Effect.gen(function* () { + const service = yield* UsageService.make; + // A one-day summary only scans files written since yesterday, not `older`. + const summary = yield* service + .readSummary({ timeZone: "UTC", sinceDay: today, untilDay: today }) + .pipe(Effect.forkChild); + yield* Deferred.await(parseStarted); + const warmup = yield* service.warmScanCache.pipe(Effect.forkChild); + yield* Deferred.succeed(releaseParse, undefined); + yield* Fiber.join(summary); + yield* Fiber.join(warmup); + assert.include(parsedPaths, older); + yield* service.awaitPersisted; + }).pipe( + Effect.scoped, + Effect.provide( + layerService({ + prefix: "usage-warm-after-summary", + home, + settings, + transcriptParser: gatedParser, + }), + ), + ); + }).pipe(Effect.scoped), + ); + it.live("does not orphan an in-flight scan when its first caller is interrupted", () => Effect.gen(function* () { const { settings, home } = yield* setup; diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index 773427db56165..0ff304bcd5199 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -727,7 +727,8 @@ export const make = Effect.gen(function* () { const startedAtMs = yield* Clock.currentTimeMillis; yield* ensureScanCacheLoaded; - // Wait rather than parse the same cold files a second time. + // Fetch prices while waiting rather than parse the same cold files a second time. + const ratesFiber = yield* Effect.forkChild(ensureRates(false)); yield* awaitWarmup; const windowStart = DateTime.make(`${input.sinceDay}T00:00:00Z`); @@ -744,7 +745,7 @@ export const make = Effect.gen(function* () { const [, scannedDirs] = yield* Effect.all( [ - ensureRates(false), + Fiber.join(ratesFiber), collectDirs(windowStartMs, settings, retentionCutoffMs, input.awaitRefresh === true), ], { concurrency: 2 }, @@ -971,8 +972,17 @@ export const make = Effect.gen(function* () { }); const warmScanCache = Effect.gen(function* () { - // A summary already in flight warms the cache itself. - if (warming !== undefined || inflightScans.size > 0) return; + if (warming !== undefined) return; + // In-flight summaries may cover a shorter window; let them finish, then warm the rest from cache. + // Waiting before `warming` is set keeps a summary not yet at awaitWarmup from waiting on us. + yield* Effect.forEach( + [...inflightScans.values()], + (scan) => Effect.exit(Deferred.await(scan)), + { + discard: true, + }, + ); + if (warming !== undefined) return; const done = Deferred.makeUnsafe(); warming = done; yield* Effect.gen(function* () {