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 6b6762c089ba6..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; @@ -172,7 +173,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), @@ -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.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..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 = [ @@ -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.