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..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"; @@ -34,6 +35,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 +55,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 +3258,135 @@ 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("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 b851c42c4a920..0ff304bcd5199 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,9 @@ export const make = Effect.gen(function* () { const startedAtMs = yield* Clock.currentTimeMillis; yield* ensureScanCacheLoaded; + // 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`); if (Option.isNone(windowStart)) { @@ -729,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 }, @@ -955,7 +971,54 @@ export const make = Effect.gen(function* () { return yield* Deferred.await(deferred); }); - return { readSummary, refreshRates, awaitPersisted } as const; + const warmScanCache = Effect.gen(function* () { + 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* () { + 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); + }), +);