| author | |
| committer | |
| log | 4d1f24ba4123279708e169fcb196f78da8f11351 |
| tree | e4cb22fc1ba08798e26abef838496892e8270a92 |
| parent | 2ea7684d563fe0ebbe9144dbfd6b0eade7eaf49a |
| signature | Signed by PGP key B5690EEEBB952194 |
Takeover of #13902.
Co-authored-by: Derek Trimm <275381468+derektrimm@users.noreply.github.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>3 files changed, 140 insertions(+), 6 deletions(-)
apps/server/src/usage/UsageService.ts+6-3| ... | ... | @@ -8,7 +8,9 @@ |
| 8 | 8 | * `(size, mtime)`. A cold 30-day scan of ~1.4 GB lands around 2-3 seconds; warm |
| 9 | 9 | * scans only reparse files that changed, and a file that merely grew resumes |
| 10 | 10 | * from its cached parse position so only the appended bytes are read. |
| 11 | * SQLite readers query live databases each scan so WAL writes remain visible. | |
| 11 | * OpenCode's SQLite reader queries the live database each scan so WAL writes | |
| 12 | * remain visible. Antigravity databases are memoised in memory while the | |
| 13 | * database and its WAL keep the same `(size, mtime, ctime)`. | |
| 12 | 14 | * |
| 13 | 15 | * @module UsageService |
| 14 | 16 | */ |
| ... | ... | @@ -52,7 +54,7 @@ import { resolveCodexHomeLayout } from "../provider/Drivers/CodexHomeLayout.ts"; |
| 52 | 54 | import { resolveAntigravityInstanceDirectories } from "../provider/antigravityAuthSupport.ts"; |
| 53 | 55 | import { mergeProviderInstanceEnvironment } from "../provider/ProviderInstanceEnvironment.ts"; |
| 54 | 56 | import { readOpenCodeUsage } from "./opencodeUsageReader.ts"; |
| 55 | import { readAntigravityUsage } from "./antigravityUsageReader.ts"; | |
| 57 | import { makeAntigravityUsageCache, readAntigravityUsage } from "./antigravityUsageReader.ts"; | |
| 56 | 58 | import { readCursorAccountUsage } from "./cursorUsageReader.ts"; |
| 57 | 59 | import { resolveModelAliases, UsageAggregator } from "./usageAggregation.ts"; |
| 58 | 60 | import { createOverrideRateTable, parseRateTable, type RateTable } from "./usagePricing.ts"; |
| ... | ... | @@ -195,6 +197,7 @@ export const make = Effect.gen(function* () { |
| 195 | 197 | const platform = yield* HostProcessPlatform; |
| 196 | 198 | |
| 197 | 199 | const fileCache: ScanCache = new Map(); |
| 200 | const antigravityCache = makeAntigravityUsageCache(); | |
| 198 | 201 | const sourceCache = new Map<string, typeof CachedSource.Type>(); |
| 199 | 202 | let cacheDirty = false; |
| 200 | 203 | const isWithinDirectory = (filePath: string, dir: string) => { |
| ... | ... | @@ -661,7 +664,7 @@ export const make = Effect.gen(function* () { |
| 661 | 664 | antigravityDirs.add(yield* fileSystem.realPath(dir).pipe(Effect.orElseSucceed(() => dir))); |
| 662 | 665 | } |
| 663 | 666 | const antigravity = yield* Effect.promise(() => |
| 664 | readAntigravityUsage([...antigravityDirs], windowStartMs), | |
| 667 | readAntigravityUsage([...antigravityDirs], windowStartMs, antigravityCache), | |
| 665 | 668 | ); |
| 666 | 669 | for (const dir of antigravityDirs) { |
| 667 | 670 | const exists = yield* fileSystem |
apps/server/src/usage/antigravityUsageReader.ts+34-2| ... | ... | @@ -257,10 +257,28 @@ async function readDatabase(path: string, fallbackTimestamp: number): Promise<Us |
| 257 | 257 | } |
| 258 | 258 | } |
| 259 | 259 | |
| 260 | /** Reads and merges aliases across every configured Antigravity store before date filtering. */ | |
| 260 | interface CachedDatabase { | |
| 261 | readonly fingerprint: string; | |
| 262 | readonly candidates: readonly UsageCandidate[]; | |
| 263 | } | |
| 264 | ||
| 265 | /** | |
| 266 | * Parsed databases keyed by canonical path. An entry is reused only while both | |
| 267 | * the database and its `-wal` sidecar keep the same size, mtime and ctime: new | |
| 268 | * rows land in the WAL without touching the main file until a checkpoint, and | |
| 269 | * ctime moves on any content write even when mtime is restored. | |
| 270 | */ | |
| 271 | export const makeAntigravityUsageCache = () => new Map<string, CachedDatabase>(); | |
| 272 | ||
| 273 | /** | |
| 274 | * Reads and merges aliases across every configured Antigravity store before date | |
| 275 | * filtering. With a cache, databases unchanged since the previous read are not | |
| 276 | * decoded again. | |
| 277 | */ | |
| 261 | 278 | export async function readAntigravityUsage( |
| 262 | 279 | conversationsDirectories: string | readonly string[], |
| 263 | 280 | sinceMs: number, |
| 281 | cache?: Map<string, CachedDatabase>, | |
| 264 | 282 | ) { |
| 265 | 283 | const roots = |
| 266 | 284 | typeof conversationsDirectories === "string" |
| ... | ... | @@ -351,7 +369,18 @@ export async function readAntigravityUsage( |
| 351 | 369 | if (visited.has(canonical)) continue; |
| 352 | 370 | visited.add(canonical); |
| 353 | 371 | const stat = await NodeFSP.stat(path); |
| 354 | const candidates = await readDatabase(path, stat.mtimeMs); | |
| 372 | const wal = await NodeFSP.stat(`${path}-wal`).catch(() => null); | |
| 373 | const fingerprint = [stat, wal] | |
| 374 | .map((file) => (file ? `${file.size}:${file.mtimeMs}:${file.ctimeMs}` : "-")) | |
| 375 | .join("/"); | |
| 376 | const cached = cache?.get(canonical); | |
| 377 | let candidates: readonly UsageCandidate[]; | |
| 378 | if (cached?.fingerprint === fingerprint) { | |
| 379 | candidates = cached.candidates; | |
| 380 | } else { | |
| 381 | candidates = await readDatabase(path, stat.mtimeMs); | |
| 382 | cache?.set(canonical, { fingerprint, candidates }); | |
| 383 | } | |
| 355 | 384 | const fileIndex = files.length; |
| 356 | 385 | files.push({ root, path, records: [] }); |
| 357 | 386 | for (const [index, candidate] of candidates.entries()) { |
| ... | ... | @@ -365,6 +394,9 @@ export async function readAntigravityUsage( |
| 365 | 394 | } |
| 366 | 395 | }; |
| 367 | 396 | for (const root of roots) await walk(root, root); |
| 397 | if (cache !== undefined) { | |
| 398 | for (const key of cache.keys()) if (!visited.has(key)) cache.delete(key); | |
| 399 | } | |
| 368 | 400 | for (const [index, group] of groups.entries()) { |
| 369 | 401 | if (group.parent === index && group.record.timestampMs >= sinceMs) { |
| 370 | 402 | files[group.fileIndex]!.records.push(group.record); |
apps/server/src/usage/usageTranscriptReader.test.ts+100-1| ... | ... | @@ -11,7 +11,7 @@ import { afterEach, assert, beforeEach, describe, it } from "@effect/vitest"; |
| 11 | 11 | import { readTranscriptRecords } from "./usageTranscriptReader.ts"; |
| 12 | 12 | import { readOpenCodeUsage } from "./opencodeUsageReader.ts"; |
| 13 | 13 | import { readCursorAccountUsage } from "./cursorUsageReader.ts"; |
| 14 | import { readAntigravityUsage } from "./antigravityUsageReader.ts"; | |
| 14 | import { makeAntigravityUsageCache, readAntigravityUsage } from "./antigravityUsageReader.ts"; | |
| 15 | 15 | |
| 16 | 16 | let dir: string; |
| 17 | 17 | |
| ... | ... | @@ -738,6 +738,105 @@ describe("SQLite usage readers", () => { |
| 738 | 738 | ); |
| 739 | 739 | }); |
| 740 | 740 | |
| 741 | const antigravityGeneration = (responseId: string) => | |
| 742 | new Uint8Array( | |
| 743 | protoBytes(1, [ | |
| 744 | ...protoBytes(4, [ | |
| 745 | ...protoNumber(2, 100), | |
| 746 | ...protoNumber(3, 40), | |
| 747 | ...protoText(11, responseId), | |
| 748 | ]), | |
| 749 | ...protoText(19, "Gemini 3 Pro"), | |
| 750 | ...protoBytes(9, protoBytes(4, protoNumber(1, 1780000000))), | |
| 751 | ]), | |
| 752 | ); | |
| 753 | ||
| 754 | it("reuses an unchanged Antigravity database instead of decoding it again", async () => { | |
| 755 | const path = NodePath.join(dir, "session-1.db"); | |
| 756 | const db = new NodeSqlite.DatabaseSync(path); | |
| 757 | try { | |
| 758 | db.exec("CREATE TABLE gen_metadata (idx INTEGER, data BLOB)"); | |
| 759 | db.prepare("INSERT INTO gen_metadata VALUES (?, ?)").run(0, antigravityGeneration("r-1")); | |
| 760 | const cache = makeAntigravityUsageCache(); | |
| 761 | assert.deepStrictEqual((await readAntigravityUsage(dir, 0, cache)).errors, []); | |
| 762 | ||
| 763 | // An exclusive lock makes a fresh read fail without touching the file, so | |
| 764 | // only a cache hit can still return the earlier records. | |
| 765 | db.exec("BEGIN EXCLUSIVE"); | |
| 766 | assert.strictEqual((await readAntigravityUsage(dir, 0)).errors.length, 1); | |
| 767 | const cached = await readAntigravityUsage(dir, 0, cache); | |
| 768 | assert.deepStrictEqual(cached.errors, []); | |
| 769 | assert.deepStrictEqual( | |
| 770 | cached.files.flatMap((file) => file.records).map((record) => record.dedupeKey), | |
| 771 | ["antigravity:11:r-1"], | |
| 772 | ); | |
| 773 | db.exec("ROLLBACK"); | |
| 774 | } finally { | |
| 775 | db.close(); | |
| 776 | } | |
| 777 | }); | |
| 778 | ||
| 779 | it("rereads an Antigravity database rewritten with its size and mtime restored", async () => { | |
| 780 | const path = NodePath.join(dir, "session-1.db"); | |
| 781 | const db = new NodeSqlite.DatabaseSync(path); | |
| 782 | try { | |
| 783 | db.exec("CREATE TABLE gen_metadata (idx INTEGER, data BLOB)"); | |
| 784 | db.prepare("INSERT INTO gen_metadata VALUES (?, ?)").run(0, antigravityGeneration("r-1")); | |
| 785 | } finally { | |
| 786 | db.close(); | |
| 787 | } | |
| 788 | await NodeFSP.utimes(path, 1780000000, 1780000000); | |
| 789 | const cache = makeAntigravityUsageCache(); | |
| 790 | assert.deepStrictEqual((await readAntigravityUsage(dir, 0, cache)).errors, []); | |
| 791 | ||
| 792 | // ctime has the kernel's timestamp granularity, which can be a few | |
| 793 | // milliseconds, so repeat the forged rewrite until it lands on a later tick | |
| 794 | // than the cached read, as any real rewrite does. | |
| 795 | const cached = await NodeFSP.stat(path); | |
| 796 | do { | |
| 797 | await NodeFSP.writeFile(path, Buffer.alloc(cached.size)); | |
| 798 | await NodeFSP.utimes(path, 1780000000, 1780000000); | |
| 799 | } while ((await NodeFSP.stat(path)).ctimeMs === cached.ctimeMs); | |
| 800 | const restored = await NodeFSP.stat(path); | |
| 801 | assert.strictEqual(restored.size, cached.size); | |
| 802 | assert.strictEqual(restored.mtimeMs, cached.mtimeMs); | |
| 803 | const next = await readAntigravityUsage(dir, 0, cache); | |
| 804 | assert.deepStrictEqual(next.errors, [path]); | |
| 805 | assert.deepStrictEqual( | |
| 806 | next.files.flatMap((file) => file.records), | |
| 807 | [], | |
| 808 | ); | |
| 809 | }); | |
| 810 | ||
| 811 | it("rereads an Antigravity database when only its WAL changed", async () => { | |
| 812 | const path = NodePath.join(dir, "session-1.db"); | |
| 813 | const db = new NodeSqlite.DatabaseSync(path); | |
| 814 | try { | |
| 815 | db.exec( | |
| 816 | "PRAGMA journal_mode = WAL; PRAGMA wal_autocheckpoint = 0; CREATE TABLE gen_metadata (idx INTEGER, data BLOB)", | |
| 817 | ); | |
| 818 | const insert = db.prepare("INSERT INTO gen_metadata VALUES (?, ?)"); | |
| 819 | insert.run(0, antigravityGeneration("r-1")); | |
| 820 | const cache = makeAntigravityUsageCache(); | |
| 821 | const first = await readAntigravityUsage(dir, 0, cache); | |
| 822 | assert.strictEqual(first.files.flatMap((file) => file.records).length, 1); | |
| 823 | ||
| 824 | const before = await NodeFSP.stat(path); | |
| 825 | insert.run(1, antigravityGeneration("r-2")); | |
| 826 | const after = await NodeFSP.stat(path); | |
| 827 | assert.strictEqual(after.size, before.size); | |
| 828 | assert.strictEqual(after.mtimeMs, before.mtimeMs); | |
| 829 | ||
| 830 | const next = await readAntigravityUsage(dir, 0, cache); | |
| 831 | assert.deepStrictEqual( | |
| 832 | next.files.flatMap((file) => file.records).map((record) => record.dedupeKey), | |
| 833 | ["antigravity:11:r-1", "antigravity:11:r-2"], | |
| 834 | ); | |
| 835 | } finally { | |
| 836 | db.close(); | |
| 837 | } | |
| 838 | }); | |
| 839 | ||
| 741 | 840 | it("reads Antigravity step-only stores and reports malformed databases", async () => { |
| 742 | 841 | const db = new NodeSqlite.DatabaseSync(NodePath.join(dir, "steps.db")); |
| 743 | 842 | try { |