diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..26941ea --- /dev/null +++ b/.gitattributes @@ -0,0 +1,21 @@ +# Force textual diffs for source, regardless of content sniffing. +# +# Git decides binary-vs-text by looking for a NUL in the first 8000 bytes. +# jsonl-parser.ts at HEAD held 4 raw NUL bytes in an 8052-byte blob: two are +# the Map-key separators themselves (offsets 7260, 7365) and two are in the +# comment that describes them (5708, 5745). The comment's — not the code's — +# is the one inside the sniff window, so git reported "Binary files differ" +# and suppressed the diff, making the file unreviewable in `git diff` and on +# GitHub. This commit rewrites all four as `\0` escapes. +# +# The `diff` attribute alone fixes that. `text` is deliberately NOT set: it +# would also impose EOL normalisation on checkin, which this change does not +# need and which would rewrite line endings for any future contributor working +# on CRLF. (HEAD currently has zero CRLF-committed files, so `text` would be a +# no-op today — but it is a standing policy, not a fix, and does not belong in +# a memory-bug PR.) +*.ts diff +*.tsx diff +*.js diff +*.json diff +*.md diff diff --git a/src/usage/__tests__/jsonl-parser.test.ts b/src/usage/__tests__/jsonl-parser.test.ts index 4edad66..27e7899 100644 --- a/src/usage/__tests__/jsonl-parser.test.ts +++ b/src/usage/__tests__/jsonl-parser.test.ts @@ -7,10 +7,16 @@ */ import { describe, test, expect, beforeEach, afterEach } from "bun:test"; -import { mkdtempSync, mkdirSync, writeFileSync, rmSync } from "fs"; +import { + mkdtempSync, + mkdirSync, + writeFileSync, + rmSync, + readFileSync, +} from "fs"; import { tmpdir } from "os"; import { join } from "path"; -import { aggregateUsage, resolveProjectDirs } from "../jsonl-parser"; +import { aggregateUsage, readLines, resolveProjectDirs } from "../jsonl-parser"; let root: string; let env: Record; @@ -34,6 +40,11 @@ function line(over: Record = {}): string { }); } +/** Drain readLines fully (`for...of` semantics, so `finally` runs). */ +function drain(file: string, buf: Buffer = Buffer.alloc(64)): string[] { + return [...readLines(file, buf)]; +} + function writeSession(project: string, file: string, lines: string[]): void { const dir = join(root, ".claude", "projects", project); mkdirSync(dir, { recursive: true }); @@ -189,3 +200,194 @@ describe("resolveProjectDirs", () => { expect(dirs[0].endsWith(join(".claude", "projects"))).toBe(true); }); }); + +/** + * Streaming-reader regression suite. + * + * `readLines` replaced `readFileSync(f,"utf-8").split("\n")`, so that expression + * IS the contract — every case below asserts the new reader against it rather + * than against a hand-written expectation. That makes *behavioural equivalence* + * a checked claim. + * + * Two things it deliberately does NOT check, so nobody reads more into a green + * run than it earns: + * + * - **Memory.** No assertion here observes heap or RSS. The ~795 MB -> ~276 MB + * figure in the PR was measured out-of-band on a frozen 401-file corpus; this + * suite would stay green if the bound regressed. + * - **The refactor itself.** Because `readFileSync().split()` is the oracle, + * reverting `aggregateUsage` to it passes every test below. These guard the + * reader's behaviour, not its continued existence. + * + * The chunk-boundary cases guard a NARROWER failure than "wrong totals", and + * the distinction is why they are written against the oracle rather than against + * parsed output: a codepoint decoded in halves yields U+FFFD *inside a JSON + * string value*, so `JSON.parse` still succeeds and token counts are unaffected. + * What regresses is equivalence with `readFileSync`. The one case where that + * turns into a data defect is a non-ASCII `message.id`/`requestId`, which would + * corrupt the dedup key and double-count; both are ASCII today. + */ +describe("readLines (streaming)", () => { + let dir: string; + beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), "cmf-stream-")); + }); + afterEach(() => { + rmSync(dir, { recursive: true, force: true }); + }); + + /** Assert the reader reproduces readFileSync().split("\n") exactly. */ + function expectMatchesSplit(content: string, chunkBytes: number): void { + const f = join(dir, "s.jsonl"); + writeFileSync(f, content); + const got = [...readLines(f, Buffer.alloc(chunkBytes))]; + expect(got).toEqual(readFileSync(f, "utf-8").split("\n")); + } + + test("utf-8 codepoint straddling a chunk boundary (Thai, 3 bytes)", () => { + // Sweep the pad length so the boundary lands on every byte of "ก" (E0 B8 81). + for (let pad = 0; pad < 8; pad++) { + expectMatchesSplit(`${"a".repeat(pad)}ก-tail\nsecond\n`, 8); + } + }); + + test("utf-8 codepoint straddling a chunk boundary (emoji, 4 bytes)", () => { + for (let pad = 0; pad < 8; pad++) { + expectMatchesSplit(`${"a".repeat(pad)}🔥-tail\nsecond\n`, 8); + } + }); + + test("a single line longer than the chunk", () => { + expectMatchesSplit(`${"x".repeat(500)}\nshort\n`, 16); + }); + + test("file with no trailing newline", () => { + expectMatchesSplit("alpha\nbeta", 4); + }); + + test("BOM is preserved, matching readFileSync (ignoreBOM:true)", () => { + // TextDecoder's default (ignoreBOM:false) STRIPS the BOM; readFileSync does + // not. Split at 4 so the flag, not luck, is what makes them agree. + expectMatchesSplit("\uFEFFalpha\nbeta", 4); + }); + + test("CRLF keeps its carriage return, as split() left it", () => { + expectMatchesSplit("alpha\r\nbeta\r\n", 4); + }); + + test("empty file", () => { + expectMatchesSplit("", 8); + }); + + test("chunk boundary exactly on the newline", () => { + expectMatchesSplit("abc\ndef\n", 4); + }); + + test("unreadable file yields nothing rather than throwing", () => { + const got = [...readLines(join(dir, "does-not-exist.jsonl"), Buffer.alloc(8))]; + expect(got).toEqual([]); + }); +}); + +describe("aggregateUsage under a tiny read buffer", () => { + beforeEach(() => { + root = mkdtempSync(join(tmpdir(), "cmf-jsonl-")); + env = { CLAUDE_CONFIG_DIR: join(root, ".claude") }; + }); + afterEach(() => { + rmSync(root, { recursive: true, force: true }); + }); + + test("multi-byte content aggregates identically at chunkBytes=8", () => { + // A Thai cwd string is realistic: session paths carry non-ASCII in the wild. + const lines = [ + line({ __seq: 1, cwd: "/บ้าน/โครงการ" }), + line({ __seq: 2, cwd: "/บ้าน/โครงการ" }), + ]; + writeSession("proj-utf8", "s1.jsonl", lines); + const tiny = aggregateUsage({ env, chunkBytes: 8 }); + const normal = aggregateUsage({ env }); + expect(tiny.rows).toEqual(normal.rows); + expect(tiny.linesSkipped).toBe(normal.linesSkipped); + expect(tiny.rows[0].inputTokens).toBe(200); + }); +}); + +/** + * Hostile values for the two knobs the streaming change introduced. + * + * `chunkBytes: 0` is the one that matters: readSync into a zero-length buffer + * returns 0, readLines cannot tell that from EOF, so before validation every + * file read as empty and aggregateUsage returned `rows: []` with no error and + * no linesSkipped — the "false success" shape this repo's rules ban. + * + * They match the guard's MESSAGE, not just `RangeError`, and that is deliberate: + * `Buffer.alloc` already throws RangeError for -1/NaN/Infinity/oversized on its + * own, so a bare `toThrow(RangeError)` passed with the guard deleted for 5 of + * these 6 values — it would have asserted Node's behaviour, not ours. Measured + * on this file by replacing assertChunkBytes with a bare `Math.floor`: + * 28 pass / 6 fail without the guard, 34 / 0 with it. + */ +describe("chunkBytes / buffer validation", () => { + beforeEach(() => { + root = mkdtempSync(join(tmpdir(), "cmf-jsonl-")); + env = { CLAUDE_CONFIG_DIR: join(root, ".claude") }; + writeSession("proj-v", "s1.jsonl", [line({ __seq: 1 })]); + }); + afterEach(() => { + rmSync(root, { recursive: true, force: true }); + }); + + test("baseline: the fixture really does produce a row", () => { + expect(aggregateUsage({ env }).rows.length).toBe(1); + }); + + for (const bad of [0, -1, 0.5, NaN, Infinity, 8 * 1024 * 1024 + 1]) { + test(`chunkBytes=${bad} throws instead of silently returning no rows`, () => { + expect(() => aggregateUsage({ env, chunkBytes: bad })).toThrow( + /chunkBytes must be/, + ); + }); + } + + test("a fractional but >=1 chunkBytes is floored, not rejected", () => { + expect(aggregateUsage({ env, chunkBytes: 4.9 }).rows.length).toBe(1); + }); + + test("readLines rejects a zero-length buffer", () => { + // Path need not exist — the guard is ordered before openSync on purpose, so + // that a bad buffer cannot leak an fd. Uses a temp path rather than a real + // machine file so the test states no dependency it does not have. + expect(() => + drain(join(root, "nonexistent.jsonl"), Buffer.alloc(0)), + ).toThrow(/buf must be non-empty/); + }); +}); + +/** + * What a read failure looks like to the caller. + * + * `readFileSync` threw on any read error and `aggregateUsage`'s + * `catch { continue }` dropped the whole file. Streaming cannot preserve that + * atomicity: bytes already decoded have already been yielded. These pin the + * resulting shape so the divergence is a recorded property, not a surprise. + */ +describe("readLines read failures", () => { + let dir: string; + beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), "cmf-readerr-")); + }); + afterEach(() => { + rmSync(dir, { recursive: true, force: true }); + }); + + test("open failure yields nothing and does not throw to the caller", () => { + expect(drain(join(dir, "absent.jsonl"))).toEqual([]); + }); + + test("a read error after open yields nothing and does not throw", () => { + // A directory opens successfully, then readSync fails with EISDIR — a real + // errno path, no mocking. Stands in for EIO/EBADF mid-scan. + expect(drain(dir)).toEqual([]); + }); +}); diff --git a/src/usage/jsonl-parser.ts b/src/usage/jsonl-parser.ts index 8e58dff..5216cc3 100644 --- a/src/usage/jsonl-parser.ts +++ b/src/usage/jsonl-parser.ts @@ -18,7 +18,7 @@ * no watermark / byte-offset state needed (ADR-003 §Idempotency over watermarks). */ -import { readdirSync, readFileSync, statSync } from "fs"; +import { closeSync, openSync, readdirSync, readSync, statSync } from "fs"; import { homedir } from "os"; import { join } from "path"; @@ -116,6 +116,133 @@ function collectJsonlFiles(dir: string): string[] { return out; } +/** Read buffer size. 64 KiB is a page-friendly default; override only in tests. */ +const DEFAULT_CHUNK_BYTES = 64 * 1024; + +/** + * Upper bound on a caller-supplied `chunkBytes`. Past this the buffer is itself + * the memory problem this module exists to avoid, so an over-large value is + * rejected rather than honoured. + */ +const MAX_CHUNK_BYTES = 8 * 1024 * 1024; + +/** + * Reject a read size that cannot do the job, loudly. + * + * `chunkBytes: 0` is the case worth the noise: `readSync` into a zero-length + * buffer returns 0, which `readLines` cannot distinguish from end-of-file, so + * EVERY file would read as empty and `aggregateUsage` would return `rows: []` + * with no error and no `linesSkipped` — a wrong answer wearing the shape of a + * right one. Throwing is safe for the production path: the only caller, + * `usage-sync.ts:49`, passes no `chunkBytes` at all. + */ +function assertChunkBytes(n: number): number { + const size = Math.floor(n); + if (!Number.isFinite(n) || size < 1 || size > MAX_CHUNK_BYTES) { + throw new RangeError( + `aggregateUsage: chunkBytes must be a finite number in [1, ${MAX_CHUNK_BYTES}] ` + + `(fractional values are floored), got ${n}`, + ); + } + return size; +} + +/** + * Yield a file's lines without materialising it. + * + * On a file that is read start-to-finish without error, emits exactly the + * sequence `readFileSync(file,"utf-8").split("\n")` would, so per-line callers + * need no change — but live bytes are bounded by `buf.length` plus the longest + * single line rather than by the file size. The previous `readFileSync` + + * `split("\n")` held BOTH full copies simultaneously, which on this author's + * 130 MB transcript is ~260 MB of live heap for one file. + * + * Two documented divergences from that equivalence, both in the error direction: + * a mid-read failure yields a PREFIX where `readFileSync` would have thrown and + * yielded nothing (see the caller's note), and a file above `readFileSync`'s + * string cap is readable here and was not before. + * + * Three details that are load-bearing, not incidental: + * + * 1. **`{stream:true}` across chunk boundaries.** A multi-byte codepoint split + * by a chunk edge must be carried, not decoded in halves. What per-chunk + * decoding actually costs is narrower than it looks, and worth stating + * precisely so the guard is not later removed as ceremony: the U+FFFD lands + * inside a JSON *string value* (bytes >= 0x80 cannot occur elsewhere in valid + * JSONL, and 0x0A never appears as a UTF-8 continuation byte, so line + * splitting is unaffected). `JSON.parse` therefore still SUCCEEDS and token + * totals are unchanged — measured across chunk sizes 1..90 on a + * Thai-plus-emoji line: 61 sizes produced U+FFFD, 0 parse failures, 0 wrong + * totals. What breaks is the equivalence above: the emitted line is not what + * `readFileSync` produced. The one path where that becomes a data defect is + * the dedup key — `(message.id, requestId)` — which corrupts under 16 of + * those 90 sizes IF either field is non-ASCII, double-counting the row. + * Today both are ASCII (`msg_01…`, `req_…`), so this is a guard against a + * format change, not a bug being fixed. Covered by "utf-8 codepoint + * straddling a chunk boundary". + * 2. **`ignoreBOM: true`.** Despite the name, this is what makes a BOM appear in + * the output — the default (`false`) silently strips it. `readFileSync` does + * not strip it, so `true` is what preserves the old behaviour. + * 3. **Splits on "\n" only.** A CRLF file keeps its trailing "\r", exactly as + * `split("\n")` left it. + * + * The buffer may be shared across concurrent generators, but the decoder may + * not: `buf` is fully drained into `pending` before any `yield`, so a resumed + * generator only ever touches string state — whereas a hoisted decoder would + * carry a truncated codepoint from one file into the next file's first line, + * which is the very bug this function exists to prevent, resurrected at file + * granularity. Hence one decoder per call, one buffer per scan. + * + * @param buf caller-owned scratch buffer, reused across files — allocating one + * per file would trade a size problem for a churn problem. Must be non-empty. + * @internal exported for tests. Consume it fully or via `for...of`, which runs + * the `finally` on `break`. A manually `.next()`-driven generator that is + * abandoned never runs `finally` and leaks its fd; enough of those yields + * EMFILE, which surfaces here as an unreadable file — i.e. a silent undercount. + */ +export function* readLines(file: string, buf: Buffer): Generator { + // Same trap as chunkBytes: readSync into an empty buffer returns 0, which is + // indistinguishable from EOF, so the file would silently yield one empty line. + if (buf.length === 0) { + throw new RangeError("readLines: buf must be non-empty"); + } + let fd: number; + try { + fd = openSync(file, "r"); + } catch { + return; // unreadable file — skip (permission, race) + } + const decoder = new TextDecoder("utf-8", { ignoreBOM: true }); + let pending = ""; + try { + for (;;) { + let n: number; + try { + n = readSync(fd, buf, 0, buf.length, null); + } catch { + return; // read error mid-file — keep what was already yielded + } + if (n === 0) break; + pending += decoder.decode(buf.subarray(0, n), { stream: true }); + let start = 0; + let nl: number; + while ((nl = pending.indexOf("\n", start)) !== -1) { + yield pending.slice(start, nl); + start = nl + 1; + } + if (start > 0) pending = pending.slice(start); + } + pending += decoder.decode(); // flush a truncated trailing sequence + yield pending; // final segment — split("\n") always produces one + } finally { + try { + closeSync(fd); + } catch { + /* already gone; nothing left to release */ + } + } +} + /** * Derive a YYYY-MM-DD key from an ISO timestamp. Returns null if unparseable. * @@ -159,13 +286,24 @@ function dateKey(ts: string | undefined, tz?: string): string | null { * @returns sorted array of DailyUsageRow (by date, then model) */ export function aggregateUsage( - opts: { since?: string; env?: EnvLike; tz?: string } = {}, + opts: { + since?: string; + env?: EnvLike; + tz?: string; + /** Read-buffer size. Tuning/testing knob; the default suits real corpora. */ + chunkBytes?: number; + } = {}, ): { rows: DailyUsageRow[]; filesScanned: number; linesSkipped: number; } { - const { since, tz, env = process.env } = opts; + const { + since, + tz, + env = process.env, + chunkBytes = DEFAULT_CHUNK_BYTES, + } = opts; const files = resolveProjectDirs(env).flatMap(collectJsonlFiles); // Aggregate keyed `${date}\0${model}`; dedup keyed `${messageId}\0${requestId}`. @@ -173,15 +311,27 @@ export function aggregateUsage( const seen = new Set(); let linesSkipped = 0; - for (const file of files) { - let content: string; - try { - content = readFileSync(file, "utf-8"); - } catch { - continue; // unreadable file — skip (permission, race) - } + // One scratch buffer for the whole scan — see readLines' @param note. + const readBuf = Buffer.alloc(assertChunkBytes(chunkBytes)); - for (const line of content.split("\n")) { + for (const file of files) { + // readLines yields NOTHING if the file cannot be opened, and yields a + // PREFIX if a read fails partway through. The prefix case is new: the + // previous readFileSync path was per-file atomic — a file was either fully + // counted or fully dropped — and streaming cannot preserve that, because + // the earlier lines are already merged into `agg` by the time the read + // fails. Neither case is counted or reported anywhere; `filesScanned` + // below is `files.length`, fixed at enumeration time, so a scan that read + // one file of forty still reports forty. That undercount then goes to a + // replace-semantics endpoint (`usage-sync.ts:6`), which overwrites the + // server's correct row. + // + // NOT TRACKED YET — no issue is filed for this as of this commit; do not + // read the paragraph above as an accepted-and-scheduled risk. It predates + // streaming (readFileSync had the same unreported-drop shape, minus the + // prefix) and fixing it needs a return-shape change plus a push policy — + // "did this scan see everything?" has to reach usage-sync before it POSTs. + for (const line of readLines(file, readBuf)) { // Hot-path prefilter: most lines have no usage block. if (line.indexOf('"usage":{') === -1) continue; // Corruption guard: skip lines with a null in any sensitive field.