From d8660bcb0e87205a0f620daaa658a8012413c060 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 19:09:28 +0000 Subject: [PATCH 1/2] fix(server): stream the usage-scan cache to disk instead of one string Saving a large usage-scan cache built the whole document as one JSON string, kept a JSON copy of every entry between writes, and UTF-8 encoded the string in a single allocation. With hundreds of MB of cached usage this ran the backend out of heap and aborted it. The cache writer now yields the document per transcript entry and no longer memoizes serialized entries. writeFileStringAtomically accepts an iterable of strings and writes it to the temp file in ~1M-character batches, so peak memory no longer scales with the file size. Fixes pingdotgg/t3code#34018 AI-assisted; verified with the tests and tools listed above. --- apps/server/src/usage/UsageService.ts | 2 +- apps/server/src/usage/usageScanCache.test.ts | 21 ++++++-------- apps/server/src/usage/usageScanCache.ts | 26 ++++++++---------- packages/shared/src/atomicWrite.test.ts | 16 +++++++++++ packages/shared/src/atomicWrite.ts | 29 ++++++++++++++++++-- 5 files changed, 63 insertions(+), 31 deletions(-) diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index 6b6762c089ba6..b17bdcd8004bb 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -172,7 +172,7 @@ export const make = Effect.gen(function* () { const ratesCachePath = path.join(config.stateDir, "usage-model-rates.json"); const scanCachePath = path.join(config.stateDir, SCAN_CACHE_FILE_NAME); - const writeCacheFile = (filePath: string, contents: string) => + const writeCacheFile = (filePath: string, contents: string | Iterable) => writeFileStringAtomically({ filePath, contents }).pipe( Effect.provideService(FileSystem.FileSystem, fileSystem), Effect.provideService(Path.Path, path), diff --git a/apps/server/src/usage/usageScanCache.test.ts b/apps/server/src/usage/usageScanCache.test.ts index fa659c4a443b3..d0e5eaceec269 100644 --- a/apps/server/src/usage/usageScanCache.test.ts +++ b/apps/server/src/usage/usageScanCache.test.ts @@ -11,6 +11,8 @@ import type { UsageRecord } from "@t3tools/provider-core/server/usage"; import { TEST_FORMAT_MAP } from "./usageTestFormats.ts"; import { initialCodexScanState, parseCodexLine } from "../provider/Drivers/codexUsage.ts"; +const parse = (parts: Iterable) => JSON.parse([...parts].join("")); + function record(overrides: Partial = {}): UsageRecord { return { provider: "claude", @@ -152,10 +154,7 @@ describe("scan cache round trip", () => { record({ model: "", dedupeKey: "msg_tail:" }), ], }); - const restored = decodeScanCache( - JSON.parse(makeScanCacheWriter()(cached, {})), - TEST_FORMAT_MAP, - ); + const restored = decodeScanCache(parse(makeScanCacheWriter()(cached, {})), TEST_FORMAT_MAP); expect(restored.size).toBe(4); expect(restored).toEqual(original); @@ -196,7 +195,7 @@ describe("scan cache round trip", () => { malformedRecords: 0, }; original.set("/clean-codex.jsonl", cleanCodex); - const encoded = JSON.parse(makeScanCacheWriter()(original, {})); + const encoded = parse(makeScanCacheWriter()(original, {})); const entry = encoded.files["/gone.jsonl"]!; const legacy = { ...encoded, @@ -251,7 +250,7 @@ describe("scan cache round trip", () => { it("re-parses a Claude transcript cached before advisor usage was counted", () => { const original = cacheWith([["/claude.jsonl", 100, [record()]]]); - const v9 = { ...JSON.parse(makeScanCacheWriter()(original, {})), version: 9 }; + const v9 = { ...parse(makeScanCacheWriter()(original, {})), version: 9 }; const restored = decodeScanCache(JSON.parse(JSON.stringify(v9)), TEST_FORMAT_MAP); expect(restored.get("/claude.jsonl")).toEqual({ @@ -269,7 +268,7 @@ describe("scan cache round trip", () => { ]); cache.set("/b.jsonl", { ...cache.get("/b.jsonl")!, malformedRecords: 2 }); const sources = { "claude\u0000/projects": { dir: "/projects", volumeId: "1:2" } }; - expect(decodeScanCache(JSON.parse(write(cache, { sources })), TEST_FORMAT_MAP)).toEqual(cache); + expect(decodeScanCache(parse(write(cache, { sources })), TEST_FORMAT_MAP)).toEqual(cache); cache.set("/b.jsonl", { ...cache.get("/b.jsonl")!, @@ -277,7 +276,7 @@ describe("scan cache round trip", () => { malformedRecords: 4, records: [record({ sessionId: "session-c", model: "claude-opus-5-5", dedupeKey: "msg_3:" })], }); - const document = JSON.parse(write(cache, { sources })); + const document = parse(write(cache, { sources })); expect(decodeScanCache(document, TEST_FORMAT_MAP)).toEqual(cache); expect(document.sources).toEqual(sources); }); @@ -291,9 +290,7 @@ describe("scan cache round trip", () => { }); it("skips malformed file entries but keeps good ones", () => { - const encoded = JSON.parse( - makeScanCacheWriter()(cacheWith([["/good.jsonl", 100, [record()]]]), {}), - ); + const encoded = parse(makeScanCacheWriter()(cacheWith([["/good.jsonl", 100, [record()]]]), {})); const withJunk = { ...encoded, files: { ...encoded.files, "/bad.jsonl": { s: "nope", m: 1, p: "claude", r: [] } }, @@ -304,7 +301,7 @@ describe("scan cache round trip", () => { }); it("drops the whole entry when any row is corrupt, forcing a cold re-parse", () => { - const encoded = JSON.parse( + const encoded = parse( makeScanCacheWriter()( cacheWith([["/a.jsonl", 100, [record(), record({ dedupeKey: "msg_2:" })]]]), {}, diff --git a/apps/server/src/usage/usageScanCache.ts b/apps/server/src/usage/usageScanCache.ts index c9032d64c05bd..6bec44cb46df0 100644 --- a/apps/server/src/usage/usageScanCache.ts +++ b/apps/server/src/usage/usageScanCache.ts @@ -134,32 +134,28 @@ function serializeFile(entry: CachedFile, tables: InternTables): SerializedFile }; } -/** Memoizes immutable entries by identity; growing intern tables preserve indexes, retaining unused strings until restart. */ +/** Yields the document in pieces so large caches never become one string; growing intern tables preserve indexes, retaining unused strings until restart. */ export function makeScanCacheWriter(): ( cache: ScanCache, extra: Readonly>, -) => string { +) => Iterable { const tables = makeInternTables(); - const fragments = new WeakMap(); - return (cache, extra) => { - const files: string[] = []; - for (const [path, entry] of cache) { - let fragment = fragments.get(entry); - if (fragment === undefined) { - fragment = JSON.stringify(serializeFile(entry, tables)); - fragments.set(entry, fragment); - } - files.push(`${JSON.stringify(path)}:${fragment}`); + function* encode(entries: readonly (readonly [string, CachedFile])[], extra: object) { + yield '{"files":{'; + for (const [index, [path, entry]] of entries.entries()) { + yield `${index === 0 ? "" : ","}${JSON.stringify(path)}:${JSON.stringify(serializeFile(entry, tables))}`; } // Encoded after the files, which may have added to the intern tables. - const head = JSON.stringify({ + const tail = JSON.stringify({ ...extra, version: USAGE_SCAN_CACHE_VERSION, models: tables.models, sessions: tables.sessions, }); - return `${head.slice(0, -1)},"files":{${files.join(",")}}}`; - }; + yield `},${tail.slice(1)}`; + } + // Snapshot now: the encoding runs later, while the file is written. + return (cache, extra) => encode([...cache], extra); } function isRecordArray(value: unknown): value is readonly unknown[] { diff --git a/packages/shared/src/atomicWrite.test.ts b/packages/shared/src/atomicWrite.test.ts index 3b0a99ec0d2d2..6aec907c049ad 100644 --- a/packages/shared/src/atomicWrite.test.ts +++ b/packages/shared/src/atomicWrite.test.ts @@ -44,6 +44,22 @@ it.layer(NodeServices.layer)("writeFileStringAtomically", (it) => { }), ); + it.effect("streams iterable contents across write batches", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const root = yield* fs.makeTempDirectoryScoped({ prefix: "t3-atomic-write-" }); + const filePath = path.join(root, "cache.json"); + yield* fs.writeFileString(filePath, "a much longer previous document"); + const chunks = ["é".repeat(700_000), "🙂".repeat(400_000), "end"]; + + yield* writeFileStringAtomically({ filePath, contents: chunks.values() }); + + assert.strictEqual(yield* fs.readFileString(filePath), chunks.join("")); + assert.deepStrictEqual(yield* fs.readDirectory(root), ["cache.json"]); + }), + ); + it.effect("keeps a symlinked file linked and rewrites its destination", () => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; diff --git a/packages/shared/src/atomicWrite.ts b/packages/shared/src/atomicWrite.ts index e48755047398f..39904817fb673 100644 --- a/packages/shared/src/atomicWrite.ts +++ b/packages/shared/src/atomicWrite.ts @@ -16,11 +16,12 @@ export const isWindowsRenameLock = (error: PlatformError.PlatformError) => { // Bound denial delays to about two seconds because settings saves hold their write lock. const LOCKED_RENAME_RETRIES = 20; const LOCKED_RENAME_RETRY_DELAY = "100 millis"; +const CHUNK_WRITE_LENGTH = 1 << 20; -/** Resolves symlink targets before renaming so writes preserve the link. */ +/** Resolves symlink targets before renaming so writes preserve the link. Iterable contents stream in bounded batches. */ export const writeFileStringAtomically = (input: { readonly filePath: string; - readonly contents: string; + readonly contents: string | Iterable; }) => Effect.scoped( Effect.gen(function* () { @@ -41,7 +42,29 @@ export const writeFileStringAtomically = (input: { ); const tempPath = path.join(tempDirectory, "contents.tmp"); - yield* fs.writeFileString(tempPath, input.contents); + const contents = input.contents; + if (typeof contents === "string") { + yield* fs.writeFileString(tempPath, contents); + } else { + // Closed before the rename, which Windows refuses for open files. + yield* Effect.scoped( + Effect.gen(function* () { + const file = yield* fs.open(tempPath, { flag: "w" }); + const encoder = new TextEncoder(); + let batch: string[] = []; + let batchLength = 0; + for (const chunk of contents) { + batch.push(chunk); + batchLength += chunk.length; + if (batchLength < CHUNK_WRITE_LENGTH) continue; + yield* file.writeAll(encoder.encode(batch.join(""))); + batch = []; + batchLength = 0; + } + if (batchLength > 0) yield* file.writeAll(encoder.encode(batch.join(""))); + }), + ); + } if (platform !== "win32") return yield* fs.rename(tempPath, targetPath); // Stop retrying if another save changes the target; read-only targets also return permanent EPERM. From 00231afbedb1313f2fb2b16e06e7935b6a250ebe Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 19:55:01 +0000 Subject: [PATCH 2/2] fix(server): remove superseded usage-scan caches and stale temp folders Older cache versions were read but never deleted, and a write that aborted left its temp folder next to the cache for good. After the first successful save of the current cache, the usage service now deletes the legacy cache files and removes cache temp folders that have been idle for over an hour, so a write still in progress from another server is left alone. Fixes pingdotgg/t3code#34018 AI-assisted; verified with the tests and tools listed above. --- apps/server/src/usage/UsageService.test.ts | 16 ++++++---- apps/server/src/usage/UsageService.ts | 34 ++++++++++++++++++++++ apps/server/src/usage/usageScanCache.ts | 2 +- 3 files changed, 45 insertions(+), 7 deletions(-) diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index a3775c6aebe17..a1de34fac38bb 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -2587,7 +2587,7 @@ describe("UsageService", () => { ); it.live.each([4, 5, 6, 7, 8, 10])( - "upgrades a v%s cache: reparses live Codex usage, keeps deleted rollouts and the old file", + "upgrades a v%s cache: reparses live Codex usage, keeps deleted rollouts, removes the old file", (version) => Effect.gen(function* () { const { home, settings } = yield* setup; @@ -2636,7 +2636,7 @@ describe("UsageService", () => { yield* first.readSummary(WINDOW); yield* first.awaitPersisted; - const legacy = yield* Effect.promise(async () => { + yield* Effect.promise(async () => { const livePath = await NodeFSP.realpath(live); const document = decodeUnknownJsonString(await NodeFSP.readFile(cachePath, "utf8")) as { files: Record }>; @@ -2674,7 +2674,9 @@ describe("UsageService", () => { } } await NodeFSP.rm(deleted); - return text; + await NodeFSP.mkdir(`${cachePath}.stale`); + await NodeFSP.utimes(`${cachePath}.stale`, 0, 0); + await NodeFSP.mkdir(`${cachePath}.active`); }); const upgraded = yield* UsageService.make; @@ -2686,9 +2688,11 @@ describe("UsageService", () => { ); yield* upgraded.awaitPersisted; assert.strictEqual(SCAN_CACHE_FILE_NAME, "usage-scan-cache-v11.json"); - assert.strictEqual( - yield* Effect.promise(() => NodeFSP.readFile(legacyPath, "utf8")), - legacy, + assert.deepStrictEqual( + (yield* Effect.promise(() => NodeFSP.readdir(stateDir))) + .filter((name) => name.startsWith("usage-scan-cache")) + .toSorted(), + [SCAN_CACHE_FILE_NAME, `${SCAN_CACHE_FILE_NAME}.active`], ); }).pipe( Effect.provide( diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index b17bdcd8004bb..b834b078058a0 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -71,6 +71,7 @@ const MTIME_SLACK_MS = 36 * 60 * 60 * 1000; const MAX_HOURLY_WINDOW_MS = 24 * 60 * 60 * 1000; const CACHE_RETENTION_DAYS = 92; +const STALE_TEMP_DIRECTORY_MS = 60 * 60 * 1000; const TRANSCRIPT_READ_CONCURRENCY = 4; @@ -407,6 +408,32 @@ export const make = Effect.gen(function* () { // Serialize writes so concurrently finished window scans cannot overwrite newer snapshots. const persistLock = yield* Semaphore.make(1); + // The saved cache supersedes older versions. Temp folders an aborted write left behind are + // removed once idle for an hour, so a write in progress from another server survives. + let supersededFilesRemoved = false; + const removeSupersededCacheFiles = Effect.gen(function* () { + for (const fileName of LEGACY_SCAN_CACHE_FILE_NAMES) { + yield* fileSystem.remove(path.join(config.stateDir, fileName)).pipe(Effect.ignore); + } + const tempPrefixes = [SCAN_CACHE_FILE_NAME, ...LEGACY_SCAN_CACHE_FILE_NAMES].map( + (fileName) => `${fileName}.`, + ); + const staleBeforeMs = (yield* Clock.currentTimeMillis) - STALE_TEMP_DIRECTORY_MS; + for (const name of yield* fileSystem.readDirectory(config.stateDir)) { + if (!tempPrefixes.some((prefix) => name.startsWith(prefix))) continue; + const tempPath = path.join(config.stateDir, name); + yield* fileSystem.stat(tempPath).pipe( + Effect.flatMap((info) => { + const mtimeMs = Option.getOrUndefined(info.mtime)?.getTime(); + return info.type === "Directory" && mtimeMs !== undefined && mtimeMs <= staleBeforeMs + ? fileSystem.remove(tempPath, { recursive: true }) + : Effect.void; + }), + Effect.ignore, + ); + } + }).pipe(Effect.ignore); + // Clear dirty before encoding to retain concurrent changes; restore it on failure for retry. const persistCaches = Effect.gen(function* () { if (cacheDirty) { @@ -415,6 +442,13 @@ export const make = Effect.gen(function* () { writeScanCache(fileCache, { sources: Object.fromEntries(sourceCache) }), ).pipe( Effect.flatMap((contents) => writeCacheFile(scanCachePath, contents)), + Effect.andThen( + Effect.suspend(() => { + if (supersededFilesRemoved) return Effect.void; + supersededFilesRemoved = true; + return removeSupersededCacheFiles; + }), + ), Effect.catchCause(() => Effect.sync(() => { cacheDirty = true; diff --git a/apps/server/src/usage/usageScanCache.ts b/apps/server/src/usage/usageScanCache.ts index 6bec44cb46df0..c2cdc1cbf5abd 100644 --- a/apps/server/src/usage/usageScanCache.ts +++ b/apps/server/src/usage/usageScanCache.ts @@ -15,7 +15,7 @@ const SPEED_COMPATIBLE_SINCE_VERSION = 4; /** * An older server sharing that directory cannot read a newer cache and would * replace it, dropping saved usage for deleted transcripts. Separate files keep - * both. + * the newer one; older files are removed once it is saved. */ export const SCAN_CACHE_FILE_NAME = "usage-scan-cache-v11.json"; export const LEGACY_SCAN_CACHE_FILE_NAMES = [