Public: anyone can read this; the crew can change it.
Appending a cost-ledger entry reads only the file's tail, so the finish path stays fast at month's end #12280
Check · Review · Land · Live done — Landed and live in production.
torque/forge:torque/12272-appending-a-cost-ledger-entry-reads-only → main 4 commits fork 1 thread · 1 unresolved
2 files
changed +96
−16
src/core/ledger.ts +59 −15
| @@ -9,7 +9,7 @@ | ||
| 9 | 9 | */ |
| 10 | 10 | |
| 11 | 11 | import { createHash } from "node:crypto"; |
| 12 | import { appendFileSync, existsSync, mkdirSync, readdirSync, readFileSync, statSync } from "node:fs"; | |
| 12 | import { appendFileSync, closeSync, existsSync, mkdirSync, openSync, readdirSync, readFileSync, readSync, statSync } from "node:fs"; | |
| 13 | 13 | import { join } from "node:path"; |
| 14 | 14 | import type { Run } from "./runs.ts"; |
| 15 | 15 | |
| @@ -101,17 +101,39 @@ function parseEntry(line: string): LedgerEntry | undefined { | ||
| 101 | 101 | } |
| 102 | 102 | } |
| 103 | 103 | |
| 104 | /** | |
| 105 | * The hash of the last readable entry on disk, or the genesis when there is | |
| 106 | * none. Read from the file on every append, so a second process on the same | |
| 107 | * directory chains from what is really there. A corrupt line is skipped, with | |
| 108 | * one loud log: the ledger keeps recording, and `ledger-verify` still names | |
| 109 | * the corrupt line. | |
| 110 | */ | |
| 111 | function lastHash(d: string): string { | |
| 112 | if (!existsSync(d)) return GENESIS; | |
| 113 | for (const file of months(d).reverse()) { | |
| 114 | const lines = readFileSync(join(d, file), "utf8").split("\n"); | |
| 104 | /** How much of a month file end the tip lookup reads first; it doubles if no entry fits. */ | |
| 105 | export const TAIL_BYTES = 64 * 1024; | |
| 106 | ||
| 107 | /** Bytes the append path has read from ledger files; tests assert on it. */ | |
| 108 | export const readStats = { bytes: 0 }; | |
| 109 | ||
| 110 | /** Read `[from, to)` of a file. */ | |
| 111 | function readRange(path: string, from: number, to: number): Buffer { | |
| 112 | const fd = openSync(path, "r"); | |
| 113 | try { | |
| 114 | const buf = Buffer.alloc(to - from); | |
| 115 | let got = 0; | |
| 116 | while (got < buf.length) { | |
| 117 | const n = readSync(fd, buf, got, buf.length - got, from + got); | |
| 118 | if (n === 0) break; | |
| 119 | got += n; | |
| 120 | } | |
| 121 | readStats.bytes += got; | |
| 122 | return buf.subarray(0, got); | |
| 123 | } finally { | |
| 124 | closeSync(fd); | |
| 125 | } | |
| 126 | } | |
| 127 | ||
| 128 | /** The hash of the last entry in one month file, reading its tail and widening only when needed. */ | |
| 129 | function tipOfFile(path: string, file: string): string | undefined { | |
| 130 | const size = statSync(path).size; | |
| 131 | for (let window = TAIL_BYTES; ; window *= 2) { | |
| 132 | const from = Math.max(0, size - window); | |
| 133 | const buf = readRange(path, from, size); | |
| 134 | // A window that starts mid-file begins inside some line: drop that fragment. | |
| 135 | const start = from === 0 ? 0 : buf.indexOf(0x0a) + 1; | |
| 136 | const lines = start === 0 && from > 0 ? [] : buf.subarray(start).toString("utf8").split("\n"); | |
| 115 | 137 | for (let i = lines.length - 1; i >= 0; i--) { |
| 116 | 138 | const line = lines[i] ?? ""; |
| 117 | 139 | if (line === "") continue; |
| @@ -119,13 +141,36 @@ function lastHash(d: string): string { | ||
| 119 | 141 | if (entry !== undefined) return entry.hash; |
| 120 | 142 | if (!warned) { |
| 121 | 143 | warned = true; |
| 122 | console.error(`ledger: ${file}:${i + 1} is not an entry; chaining from the last readable one`); | |
| 144 | console.error(`ledger: ${file} has a line that is not an entry; chaining from the last readable one`); | |
| 123 | 145 | } |
| 124 | 146 | } |
| 147 | if (from === 0) return undefined; | |
| 148 | } | |
| 149 | } | |
| 150 | ||
| 151 | /** | |
| 152 | * The hash of the last readable entry on disk, or the genesis when there is | |
| 153 | * none. Read from the file's tail on every append (never cached), so a second | |
| 154 | * process on the same directory chains from what is really there. A corrupt | |
| 155 | * line is skipped, with one loud log: the ledger keeps recording, and | |
| 156 | * `ledger-verify` still names the corrupt line. | |
| 157 | */ | |
| 158 | function lastHash(d: string): string { | |
| 159 | if (!existsSync(d)) return GENESIS; | |
| 160 | for (const file of months(d).reverse()) { | |
| 161 | const tip = tipOfFile(join(d, file), file); | |
| 162 | if (tip !== undefined) return tip; | |
| 125 | 163 | } |
| 126 | 164 | return GENESIS; |
| 127 | 165 | } |
| 128 | 166 | |
| 167 | /** Whether the file ends mid-line: reads one byte. */ | |
| 168 | function isTorn(path: string): boolean { | |
| 169 | if (!existsSync(path)) return false; | |
| 170 | const size = statSync(path).size; | |
| 171 | return size > 0 && readRange(path, size - 1, size)[0] !== 0x0a; | |
| 172 | } | |
| 173 | ||
| 129 | 174 | function appendBody(make: (prev: string) => LedgerEntry, what: string): void { |
| 130 | 175 | if (dir === undefined) return; |
| 131 | 176 | try { |
| @@ -133,8 +178,7 @@ function appendBody(make: (prev: string) => LedgerEntry, what: string): void { | ||
| 133 | 178 | const entry = make(lastHash(dir)); |
| 134 | 179 | const file = join(dir, `${entry.endedAt.slice(0, 7)}.jsonl`); |
| 135 | 180 | // A torn last line (no newline) must not swallow this entry. |
| 136 | const torn = existsSync(file) && !readFileSync(file, "utf8").endsWith("\n") && statSync(file).size > 0; | |
| 137 | appendFileSync(file, `${torn ? "\n" : ""}${JSON.stringify(entry)}\n`); | |
| 181 | appendFileSync(file, `${isTorn(file) ? "\n" : ""}${JSON.stringify(entry)}\n`); | |
| 138 | 182 | } catch (error) { |
| 139 | 183 | console.error(`ledger: could not append ${what}: ${(error as Error).message}`); |
| 140 | 184 | } |
test/cost-ledger.test.ts +37 −1
| @@ -2,7 +2,7 @@ import { afterEach, beforeEach, expect, spyOn, test } from "bun:test"; | ||
| 2 | 2 | import { appendFileSync, existsSync, mkdirSync, mkdtempSync, readFileSync, readdirSync, writeFileSync } from "node:fs"; |
| 3 | 3 | import { tmpdir } from "node:os"; |
| 4 | 4 | import { join } from "node:path"; |
| 5 | import { GENESIS, appendRun, hashOf, setLedgerDir, verifyDir } from "../src/core/ledger.ts"; | |
| 5 | import { GENESIS, TAIL_BYTES, appendRun, hashOf, readStats, setLedgerDir, verifyDir } from "../src/core/ledger.ts"; | |
| 6 | 6 | import { type Run, cancelRun, claimRun, createRun, finishRun } from "../src/core/runs.ts"; |
| 7 | 7 | import { Database } from "bun:sqlite"; |
| 8 | 8 | import { createApp } from "../src/app.ts"; |
| @@ -153,6 +153,42 @@ test("a second writer on the same directory chains from what is on disk", () => | ||
| 153 | 153 | expect(JSON.parse(ls[2] ?? "").prev).toBe(other.hash); |
| 154 | 154 | }); |
| 155 | 155 | |
| 156 | test("one append reads only the tail window of a 20 000-entry month", () => { | |
| 157 | const file = join(dir, "2026-11.jsonl"); | |
| 158 | mkdirSync(dir, { recursive: true }); | |
| 159 | let prev = GENESIS; | |
| 160 | const out: string[] = []; | |
| 161 | for (let i = 0; i < 20000; i++) { | |
| 162 | const body = { run: `r${i}`, kind: "check", repo: "r", endedAt: "2026-11-01T00:00:00.000Z", prev }; | |
| 163 | const hash = hashOf(body); | |
| 164 | out.push(JSON.stringify({ ...body, hash })); | |
| 165 | prev = hash; | |
| 166 | } | |
| 167 | writeFileSync(file, `${out.join("\n")}\n`); | |
| 168 | expect(readFileSync(file).length).toBeGreaterThan(TAIL_BYTES * 20); | |
| 169 | readStats.bytes = 0; | |
| 170 | appendRun(fake("z", "2026-11-02T00:00:00.000Z")); | |
| 171 | expect(readStats.bytes).toBeLessThanOrEqual(TAIL_BYTES + 1); | |
| 172 | const last = readFileSync(file, "utf8").trimEnd().split("\n").pop() ?? ""; | |
| 173 | expect(JSON.parse(last).prev).toBe(prev); | |
| 174 | expect(verifyDir(dir)).toEqual({ ok: true, entries: 20001 }); | |
| 175 | }); | |
| 176 | ||
| 177 | test("the tail window widens when it holds no whole entry, and a torn tail is one byte", () => { | |
| 178 | appendRun(fake("a", "2026-11-01T00:00:00.000Z")); | |
| 179 | const file = join(dir, "2026-11.jsonl"); | |
| 180 | const a = JSON.parse(readFileSync(file, "utf8").trim()); | |
| 181 | // A corrupt last line longer than the window pushes the tip out of it. | |
| 182 | writeFileSync(file, `${readFileSync(file, "utf8")}{"junk":"${"x".repeat(TAIL_BYTES * 2)}`); | |
| 183 | const err = spyOn(console, "error").mockImplementation(() => {}); | |
| 184 | readStats.bytes = 0; | |
| 185 | appendRun(fake("b", "2026-11-01T00:00:01.000Z")); | |
| 186 | err.mockRestore(); | |
| 187 | expect(readStats.bytes).toBeGreaterThan(TAIL_BYTES); | |
| 188 | const ls = readFileSync(file, "utf8").split("\n").filter(Boolean); | |
| 189 | expect(JSON.parse(ls[2] ?? "").prev).toBe(a.hash); | |
| 190 | }); | |
| 191 | ||
| 156 | 192 | test("an agent run that finishes with usage writes both token fields; a check's entry has neither", async () => { |
| 157 | 193 | const { db, repoId } = await openForTest(); |
| 158 | 194 | end(db, repoId); |
new above
offline
Non-blocking: readStats is a mutable test hook in the finish path. Prefer a narrower seam.