fix(claude-usage): review findings of the #52/#51/#53 stack #56

Merged
maximus merged 4 commits from issue-54-claude-usage-review-findings into main 2026-10-08 00:54:24 +00:00
5 changed files with 338 additions and 22 deletions
Showing only changes of commit 123b3c3c8c - Show all commits

View file

@ -644,6 +644,75 @@ describe("frozen days", () => {
expect(r.projects).toEqual([{ name: "alpha", work: 0, cacheRead: 1, requests: 1 }]); expect(r.projects).toEqual([{ name: "alpha", work: 0, cacheRead: 1, requests: 1 }]);
expect(Object.keys(r.openDays)).toHaveLength(29); expect(Object.keys(r.openDays)).toHaveLength(29);
}); });
// #54: a request whose lines straddle midnight. The day freezes at D+2, so
// the beat that froze it saw every line of the request and gave it to the
// day of its best line. Later beats must make the same choice, not count
// the request again on the open day once the frozen line is filtered out.
// Toronto is UTC-4 here: 03:59Z is Oct 1 23:59, 04:01Z is Oct 2 00:01.
it("a request straddling midnight whose best line sits on a frozen day is not counted again on the open day", async () => {
writeTranscript(
"a/straddle.jsonl",
[
line({ ts: "2026-10-02T03:59:00Z", id: "msg_mid", req: "req_mid", output: 5 }),
line({ ts: "2026-10-02T04:01:00Z", id: "msg_mid", req: "req_mid", output: 5 }),
],
"2026-10-02T12:00:00Z",
);
const r = await aggregate({ root: tmpDir, today: "2026-10-04", homedir: HOME, frozenDays: FROZEN });
expect(dayOf(r, "2026-10-02").models).toEqual({});
expect(r.openDays["2026-10-02"].models).toEqual({});
// The frozen day is served as it was counted, untouched.
expect(dayOf(r, "2026-10-01").models["claude-opus-5-5"].requests).toBe(7);
});
it("a request straddling midnight whose best line sits on the open day is counted once there", async () => {
writeTranscript(
"a/straddle.jsonl",
[
line({ ts: "2026-10-02T03:59:00Z", id: "msg_mid", req: "req_mid", output: 1 }),
line({ ts: "2026-10-02T04:01:00Z", id: "msg_mid", req: "req_mid", output: 5 }),
],
"2026-10-02T12:00:00Z",
);
const r = await aggregate({ root: tmpDir, today: "2026-10-04", homedir: HOME, frozenDays: FROZEN });
expect(dayOf(r, "2026-10-02").models["claude-opus-5-5"]).toEqual({
input: 1,
output: 5,
cacheCreation: 0,
cacheRead: 0,
requests: 1,
});
});
it("on equal output the earliest line wins, whatever the order the files are read in", async () => {
// Sorted read order puts the later line first. The beat that freezes a day
// and the beats after it do not read the same files (the threshold moves),
// so the winner must not depend on the read order.
const later = writeTranscript(
"a/a-first.jsonl",
[line({ ts: "2026-10-02T04:01:00Z", id: "msg_tie", req: "req_tie", output: 5 })],
"2026-10-02T12:00:00Z",
);
const earlier = writeTranscript(
"a/b-second.jsonl",
[line({ ts: "2026-10-02T03:59:00Z", id: "msg_tie", req: "req_tie", output: 5 })],
"2026-10-02T12:00:00Z",
);
const opened = [];
const r = await aggregate({
root: tmpDir,
today: "2026-10-04",
homedir: HOME,
openFile: (f, o) => {
opened.push(f);
return fs.createReadStream(f, o);
},
});
expect(opened).toEqual([later, earlier]);
expect(dayOf(r, "2026-10-01").models["claude-opus-5-5"].requests).toBe(1);
expect(dayOf(r, "2026-10-02").models).toEqual({});
});
}); });
// --- Robustness ----------------------------------------------------------------- // --- Robustness -----------------------------------------------------------------
@ -684,6 +753,70 @@ describe("robustness", () => {
}); });
}); });
// #54: an unreadable transcript, directory or stat is counted, so that the
// agent freezes nothing on that beat.
it("counts a stream error as a read error", async () => {
writeTranscript("p/good.jsonl", [line({ ts: "2026-10-03T15:00:00Z" })], "2026-10-04T12:00:00Z");
const bad = writeTranscript("p/bad.jsonl", [line({ ts: "2026-10-03T15:00:00Z" })], "2026-10-04T12:00:00Z");
const r = await aggregate({
root: tmpDir,
today: "2026-10-04",
homedir: HOME,
openFile: (f, o) => fs.createReadStream(f === bad ? path.join(tmpDir, "does-not-exist") : f, o),
});
expect(r.readErrors).toBe(1);
});
it("counts an opener that throws as a read error", async () => {
writeTranscript("p/bad.jsonl", [line({ ts: "2026-10-03T15:00:00Z" })], "2026-10-04T12:00:00Z");
const r = await aggregate({
root: tmpDir,
today: "2026-10-04",
homedir: HOME,
openFile: () => {
throw Object.assign(new Error("denied"), { code: "EACCES" });
},
});
expect(r.readErrors).toBe(1);
});
it("counts an unreadable subdirectory as a read error, and skips its files", async () => {
writeTranscript("p/good.jsonl", [line({ ts: "2026-10-03T15:00:00Z", output: 4 })], "2026-10-04T12:00:00Z");
writeTranscript("q/hidden.jsonl", [line({ ts: "2026-10-03T15:00:00Z", output: 9 })], "2026-10-04T12:00:00Z");
const badDir = path.join(tmpDir, "q");
// A spy rather than chmod 000: the CI runs as root, which reads anyway.
const realReaddir = fs.promises.readdir;
vi.spyOn(fs.promises, "readdir").mockImplementation((dir, opts) =>
dir === badDir
? Promise.reject(Object.assign(new Error("denied"), { code: "EACCES" }))
: realReaddir.call(fs.promises, dir, opts),
);
const r = await aggregate({ root: tmpDir, today: "2026-10-04", homedir: HOME });
expect(r.readErrors).toBe(1);
expect(dayOf(r, "2026-10-03").models["claude-opus-5-5"].output).toBe(4);
});
it("counts a failed stat as a read error, ENOENT included", async () => {
writeTranscript("p/good.jsonl", [line({ ts: "2026-10-03T15:00:00Z" })], "2026-10-04T12:00:00Z");
const gone = writeTranscript("p/gone.jsonl", [line({ ts: "2026-10-03T15:00:00Z" })], "2026-10-04T12:00:00Z");
const realStat = fs.promises.stat;
vi.spyOn(fs.promises, "stat").mockImplementation((file, opts) =>
file === gone
? Promise.reject(Object.assign(new Error("gone"), { code: "ENOENT" }))
: realStat.call(fs.promises, file, opts),
);
const r = await aggregate({ root: tmpDir, today: "2026-10-04", homedir: HOME });
expect(r.readErrors).toBe(1);
});
it("a missing root is an empty scan, not a read error; a clean scan has none", async () => {
const missing = await aggregate({ root: path.join(tmpDir, "absent"), today: "2026-10-04", homedir: HOME });
expect(missing.readErrors).toBe(0);
writeTranscript("p/good.jsonl", [line({ ts: "2026-10-03T15:00:00Z" })], "2026-10-04T12:00:00Z");
const clean = await aggregate({ root: tmpDir, today: "2026-10-04", homedir: HOME });
expect(clean.readErrors).toBe(0);
});
it("folds a model name outside the server key pattern into `other`", async () => { it("folds a model name outside the server key pattern into `other`", async () => {
writeTranscript( writeTranscript(
"p/s.jsonl", "p/s.jsonl",
@ -995,6 +1128,61 @@ describe("limits — the endpoint's answer", () => {
}); });
}); });
describe("limits — the response size is counted in bytes, while reading (#54)", () => {
const MAX = 256 * 1024;
const limitsOpts = (fetch) => ({
env: {},
credentialsPath: "/unused",
readFile: () => JSON.stringify(goodCredentials()),
claudeCodeVersion: "2.1.283",
now: NOW,
fetch,
});
// A valid answer padded to exactly `bytes` UTF-8 bytes with `pad`.
function padded(bytes, pad) {
const base = JSON.stringify({ ...usageFixture(), pad: "" });
let room = bytes - Buffer.byteLength(base);
const unit = Buffer.byteLength(pad);
let fill = "a".repeat(room % unit);
fill += pad.repeat(Math.floor(room / unit));
const body = JSON.stringify({ ...usageFixture(), pad: fill });
expect(Buffer.byteLength(body)).toBe(bytes);
return body;
}
it("accepts a body of exactly 256 KiB", async () => {
const body = padded(MAX, "a");
const r = await fetchLimits(limitsOpts(makeFetch(() => new Response(body, { status: 200 }))));
expect(r.status).toBe("ok");
});
it("refuses 256 KiB + 1 byte of two-byte characters, fewer than 256 Ki characters", async () => {
const body = padded(MAX + 1, "\u00e9");
expect(body.length).toBeLessThan(MAX);
const r = await fetchLimits(limitsOpts(makeFetch(() => new Response(body, { status: 200 }))));
expect(r).toEqual({ status: "unavailable", detail: DETAILS.BAD_RESPONSE, fetchedAt: null, items: [] });
});
it("stops reading an endless body and cancels it", async () => {
let pulls = 0;
let cancelled = false;
const chunk = new Uint8Array(64 * 1024).fill(0x20);
const stream = new ReadableStream({
pull(controller) {
pulls += 1;
controller.enqueue(chunk);
},
cancel() {
cancelled = true;
},
});
const r = await fetchLimits(limitsOpts(makeFetch(() => new Response(stream, { status: 200 }))));
expect(r).toEqual({ status: "unavailable", detail: DETAILS.BAD_RESPONSE, fetchedAt: null, items: [] });
expect(cancelled).toBe(true);
expect(pulls).toBeLessThan(10); // 5 chunks cross the ceiling
});
});
// --- limits: when no call is made ---------------------------------------------- // --- limits: when no call is made ----------------------------------------------
describe("limits — cases that never call the endpoint", () => { describe("limits — cases that never call the endpoint", () => {
@ -1252,6 +1440,24 @@ describe("state — local cache", () => {
expect(Object.keys(s.frozenDays).sort()).toEqual(["2026-08-30", "2026-09-20"]); expect(Object.keys(s.frozenDays).sort()).toEqual(["2026-08-30", "2026-09-20"]);
}); });
it("with freeze: false, keeps the frozen days (pruned) and freezes no new one", () => {
const state = {
...emptyState(),
frozenDays: {
"2026-08-29": { models: {}, projects: {} }, // pruned all the same
"2026-09-20": { models: {}, projects: {} },
},
};
const openDays = {
"2026-10-01": { models: { m: { input: 1 } }, projects: {} },
"2026-10-03": { models: {}, projects: {} },
};
const ok = { status: "ok", fetchedAt: "2026-10-04T16:00:00.000Z", items: [{ kind: "session" }] };
const s = nextState(state, { openDays, today: "2026-10-04", limits: ok, freeze: false });
expect(Object.keys(s.frozenDays)).toEqual(["2026-09-20"]);
expect(s.lastGoodLimits).toEqual({ fetchedAt: ok.fetchedAt, items: ok.items });
});
it("updates lastGoodLimits on `ok` only", () => { it("updates lastGoodLimits on `ok` only", () => {
const ok = { status: "ok", fetchedAt: "2026-10-04T16:00:00.000Z", items: [{ kind: "session" }] }; const ok = { status: "ok", fetchedAt: "2026-10-04T16:00:00.000Z", items: [{ kind: "session" }] };
const s1 = nextState(emptyState(), { openDays: {}, today: "2026-10-04", limits: ok }); const s1 = nextState(emptyState(), { openDays: {}, today: "2026-10-04", limits: ok });
@ -1292,6 +1498,19 @@ describe("state — local cache", () => {
expect(loadState(file).frozenDays).toEqual({ "2026-10-01": { models: {}, projects: {} } }); expect(loadState(file).frozenDays).toEqual({ "2026-10-01": { models: {}, projects: {} } });
}); });
it("a temp file left at mode 644 by a killed run still ends in a mode 600 cache", () => {
// writeFileSync only applies `mode` when it creates the file (#54).
const file = path.join(tmpDir, "state", "state.json");
fs.mkdirSync(path.dirname(file), { recursive: true });
const tmp = `${file}.${process.pid}.tmp`;
fs.writeFileSync(tmp, "leftover");
fs.chmodSync(tmp, 0o644); // explicit: the umask could already give 600
expect(fs.statSync(tmp).mode & 0o777).toBe(0o644);
saveState(file, emptyState());
expect(fs.existsSync(tmp)).toBe(false);
expect(fs.statSync(file).mode & 0o777).toBe(0o600);
});
it("the default state dir is ~/.local/state/maximus-claude-usage", () => { it("the default state dir is ~/.local/state/maximus-claude-usage", () => {
expect(stateFilePath(defaultStateDir("/home/u"))).toBe("/home/u/.local/state/maximus-claude-usage/state.json"); expect(stateFilePath(defaultStateDir("/home/u"))).toBe("/home/u/.local/state/maximus-claude-usage/state.json");
}); });
@ -1353,6 +1572,42 @@ describe("push-claude-usage — one beat", () => {
expect(snap2.projects).toEqual(snap1.projects); expect(snap2.projects).toEqual(snap1.projects);
}); });
it("a beat with a read error freezes no day, still pushes, logs a count only; the next clean beat freezes", async () => {
const { home, stateFile } = makeHome({ credentials: undefined });
const projectDir = path.join(home, ".claude", "projects", "-home-tester-claude-code-alpha");
fs.mkdirSync(projectDir, { recursive: true });
const write = (name, lines) => {
const file = path.join(projectDir, name);
fs.writeFileSync(file, `${lines.map((l) => JSON.stringify(l)).join("\n")}\n`);
setMtime(file, "2026-10-04T15:30:00Z");
return file;
};
write("good.jsonl", [line({ ts: "2026-10-01T15:00:00Z", output: 3 })]);
const bad = write("secret-name.jsonl", [line({ ts: "2026-10-01T15:00:00Z", output: 8 })]);
const env = agentEnv({ CLAUDE_USAGE_LIMITS: "off" });
const fetch1 = routedFetch();
const r1 = await runAgent({
home, fetch: fetch1, env,
openFile: (f, o) => fs.createReadStream(f === bad ? path.join(projectDir, "missing.jsonl") : f, o),
});
expect(r1.code).toBe(EXIT_OK);
expect(pushCalls(fetch1)).toHaveLength(1);
expect(loadState(stateFile).frozenDays).toEqual({});
expect(r1.stderr).toContain("read_errors=1");
expect(r1.stderr).not.toContain("secret-name");
expect(r1.stderr).not.toContain(home);
expect(r1.stdout).not.toContain("secret-name");
const fetch2 = routedFetch();
const r2 = await runAgent({ home, fetch: fetch2, env, now: NOW + 15 * 60 * 1000 });
expect(r2.code).toBe(EXIT_OK);
expect(r2.stderr).not.toContain("read_errors");
const frozen = loadState(stateFile).frozenDays;
expect(Object.keys(frozen)).toHaveLength(28);
expect(frozen["2026-10-01"].models["claude-opus-5-5"].output).toBe(11);
});
it("a corrupt state.json gives a full beat without failing, and is rewritten", async () => { it("a corrupt state.json gives a full beat without failing, and is rewritten", async () => {
const { home, stateFile } = makeHome({ fixtures: true }); const { home, stateFile } = makeHome({ fixtures: true });
fs.mkdirSync(path.dirname(stateFile), { recursive: true }); fs.mkdirSync(path.dirname(stateFile), { recursive: true });

View file

@ -16,7 +16,19 @@
// times (one line per content block, and again in subagent transcripts), // times (one line per content block, and again in subagent transcripts),
// with identical input/cache counters and a growing output_tokens. Lines // with identical input/cache counters and a growing output_tokens. Lines
// are deduplicated on (message.id, requestId) across ALL files, keeping the // are deduplicated on (message.id, requestId) across ALL files, keeping the
// line with the largest output_tokens. // line with the largest output_tokens; on equal output, the earliest line,
// so the choice never depends on the order the files are read in.
// * FROZEN DAYS ARE DROPPED AFTER DEDUPLICATION, NOT BEFORE. A request whose
// lines straddle midnight belongs to the day of its best line. The beat
// that froze a day saw all its lines (a day freezes at D+2) and made that
// choice; filtering the frozen day's lines first would leave the later
// line alone and count the request a second time on the open day (#54).
// Known limit: the window start is still filtered per line, so a request
// straddling the window's first midnight lands on the window's first day.
// * READ ERRORS ARE COUNTED. An unreadable directory, a failed stat or a
// stream error skips that input but is reported in `readErrors`: the
// agent then freezes nothing on that beat, rather than freezing a day
// with lines missing (#54). A missing root is an empty scan, not an error.
// * THE SCAN THRESHOLD. Frozen days are never recomputed. A file whose mtime // * THE SCAN THRESHOLD. Frozen days are never recomputed. A file whose mtime
// (Toronto day) is before the oldest open day of the window cannot hold a // (Toronto day) is before the oldest open day of the window cannot hold a
// line on an open day, so it is not even opened: this is what keeps a run // line on an open day, so it is not even opened: this is what keeps a run
@ -225,8 +237,9 @@ function normalizeDay(day) {
// --- Transcript scan -------------------------------------------------------- // --- Transcript scan --------------------------------------------------------
// Every *.jsonl under root, recursively (subagents/ included). Symlinks are // Every *.jsonl under root, recursively (subagents/ included). Symlinks are
// neither followed nor read. A missing root is an empty scan, not an error. // neither followed nor read. A missing root is an empty scan, not an error;
async function listTranscripts(root) { // any other directory that cannot be listed is reported through onError.
async function listTranscripts(root, onError = () => {}) {
const out = []; const out = [];
const stack = [root]; const stack = [root];
while (stack.length) { while (stack.length) {
@ -234,7 +247,8 @@ async function listTranscripts(root) {
let entries; let entries;
try { try {
entries = await fs.promises.readdir(dir, { withFileTypes: true }); entries = await fs.promises.readdir(dir, { withFileTypes: true });
} catch { } catch (err) {
if (!(dir === root && err && err.code === "ENOENT")) onError();
continue; continue;
} }
for (const e of entries) { for (const e of entries) {
@ -247,33 +261,41 @@ async function listTranscripts(root) {
return out; return out;
} }
// Calls onLine for each line of the file. Resolves once the file is consumed; // Calls onLine for each line of the file. Resolves once the file is consumed,
// a read error (file removed mid-run, permission) resolves too: one unreadable // with true when it was read to the end; a read error (file removed mid-run,
// transcript must not sink the whole snapshot. // permission) resolves with false: one unreadable transcript must not sink
// the whole snapshot, but the caller must know the scan is incomplete.
function readLines(file, openFile, onLine) { function readLines(file, openFile, onLine) {
return new Promise((resolve) => { return new Promise((resolve) => {
let stream; let stream;
try { try {
stream = openFile(file, { encoding: "utf-8" }); stream = openFile(file, { encoding: "utf-8" });
} catch { } catch {
resolve(); resolve(false);
return; return;
} }
let done = false; let done = false;
let rl = null; let rl = null;
// The flag is set by the error handlers only, before finish(): neither a
// 'close' after an error nor the destroy() below can turn it back.
let failed = false;
const finish = () => { const finish = () => {
if (done) return; if (done) return;
done = true; done = true;
if (rl) rl.close(); if (rl) rl.close();
if (typeof stream.destroy === "function") stream.destroy(); if (typeof stream.destroy === "function") stream.destroy();
resolve(); resolve(!failed);
};
const fail = () => {
failed = true;
finish();
}; };
// Listen on both: readline forwards the input's error to the interface, // Listen on both: readline forwards the input's error to the interface,
// and an unlistened 'error' there is an uncaught exception that kills the // and an unlistened 'error' there is an uncaught exception that kills the
// process (and the interface then never emits 'close'). // process (and the interface then never emits 'close').
stream.on("error", finish); stream.on("error", fail);
rl = readline.createInterface({ input: stream, crlfDelay: Infinity }); rl = readline.createInterface({ input: stream, crlfDelay: Infinity });
rl.on("error", finish); rl.on("error", fail);
rl.on("line", onLine); rl.on("line", onLine);
rl.on("close", finish); rl.on("close", finish);
}); });
@ -294,6 +316,7 @@ function readLines(file, openFile, onLine) {
* days: Array<{ date: string, models: Object<string, {input, output, cacheCreation, cacheRead, requests}> }>, * days: Array<{ date: string, models: Object<string, {input, output, cacheCreation, cacheRead, requests}> }>,
* projects: Array<{ name: string, work: number, cacheRead: number, requests: number }>, * projects: Array<{ name: string, work: number, cacheRead: number, requests: number }>,
* claudeCodeVersion: string | null, * claudeCodeVersion: string | null,
* readErrors: number, unreadable directories, stats and transcripts; 0 on a complete scan
* }>} * }>}
*/ */
async function aggregate({ async function aggregate({
@ -357,9 +380,11 @@ async function aggregate({
const day = torontoDay(ts); const day = torontoDay(ts);
if (day < windowStart || day > today) return; if (day < windowStart || day > today) return;
if (Object.prototype.hasOwnProperty.call(frozen, day)) return; // Frozen days are NOT filtered here: see FROZEN DAYS ARE DROPPED AFTER
// DEDUPLICATION in the header.
const entry = { const entry = {
ts,
day, day,
model: typeof model === "string" && MODEL_KEY_RE.test(model) ? model : OTHER_MODEL, model: typeof model === "string" && MODEL_KEY_RE.test(model) ? model : OTHER_MODEL,
project: mapProject(obj.cwd, homedir), project: mapProject(obj.cwd, homedir),
@ -375,18 +400,27 @@ async function aggregate({
// it counts as its own request rather than being dropped. // it counts as its own request rather than being dropped.
const key = id && reqId ? `${id}\u0000${reqId}` : `\u0001${anonymous++}`; const key = id && reqId ? `${id}\u0000${reqId}` : `\u0001${anonymous++}`;
const prev = requests.get(key); const prev = requests.get(key);
if (!prev || entry.output > prev.output) requests.set(key, entry); if (!prev || entry.output > prev.output || (entry.output === prev.output && entry.ts < prev.ts)) {
requests.set(key, entry);
}
}; };
for (const file of await listTranscripts(root)) { let readErrors = 0;
const onError = () => {
readErrors += 1;
};
for (const file of await listTranscripts(root, onError)) {
let stat; let stat;
try { try {
stat = await fs.promises.stat(file); stat = await fs.promises.stat(file);
} catch { } catch {
// ENOENT included: a transcript removed between the walk and the stat
// may have held lines of a day about to be frozen.
onError();
continue; continue;
} }
if (torontoDay(stat.mtimeMs) < threshold) continue; if (torontoDay(stat.mtimeMs) < threshold) continue;
await readLines(file, openFile, onLine); if (!(await readLines(file, openFile, onLine))) onError();
} }
// Open days: every non-frozen day of the window, empty ones included, so // Open days: every non-frozen day of the window, empty ones included, so
@ -395,6 +429,8 @@ async function aggregate({
const openDays = dict(); const openDays = dict();
for (const day of openDayList) openDays[day] = { models: dict(), projects: dict() }; for (const day of openDayList) openDays[day] = { models: dict(), projects: dict() };
for (const r of requests.values()) { for (const r of requests.values()) {
// A request whose best line sits on a frozen day was counted there.
if (Object.prototype.hasOwnProperty.call(frozen, r.day)) continue;
const bucket = openDays[r.day]; const bucket = openDays[r.day];
if (!bucket.models[r.model]) bucket.models[r.model] = emptyModel(); if (!bucket.models[r.model]) bucket.models[r.model] = emptyModel();
addModel(bucket.models[r.model], { ...r, requests: 1 }); addModel(bucket.models[r.model], { ...r, requests: 1 });
@ -425,6 +461,7 @@ async function aggregate({
days, days,
projects: topProjects(projectTotals), projects: topProjects(projectTotals),
claudeCodeVersion: latest ? latest.version : null, claudeCodeVersion: latest ? latest.version : null,
readErrors,
}; };
} }

View file

@ -29,7 +29,8 @@ const path = require("node:path");
const USAGE_URL = "https://api.anthropic.com/api/oauth/usage"; const USAGE_URL = "https://api.anthropic.com/api/oauth/usage";
const ANTHROPIC_BETA = "oauth-2025-04-20"; const ANTHROPIC_BETA = "oauth-2025-04-20";
const LIMITS_TIMEOUT_MS = 10000; const LIMITS_TIMEOUT_MS = 10000;
// Far beyond the ~2 KB the endpoint answers; past it the body is not parsed. // Far beyond the ~2 KB the endpoint answers. Counted in bytes while reading:
// past it the stream is cancelled, neither read further nor parsed (#54).
const MAX_RESPONSE_BYTES = 256 * 1024; const MAX_RESPONSE_BYTES = 256 * 1024;
const MAX_ITEMS = 12; const MAX_ITEMS = 12;
const MAX_STRING = 128; const MAX_STRING = 128;
@ -175,9 +176,22 @@ function readCredentials(credentialsPath, readFile) {
} }
async function readBody(res) { async function readBody(res) {
const text = await res.text(); if (!res.body || typeof res.body.getReader !== "function") throw new Error("no body");
if (text.length > MAX_RESPONSE_BYTES) throw new Error("too large"); const reader = res.body.getReader();
return JSON.parse(text); const chunks = [];
let total = 0;
for (;;) {
// A timeout surfaces here as a TimeoutError/AbortError, mapped by the caller.
const { done, value } = await reader.read();
if (done) break;
total += value.byteLength;
if (total > MAX_RESPONSE_BYTES) {
reader.cancel().catch(() => {});
throw new Error("too large");
}
chunks.push(value);
}
return JSON.parse(Buffer.concat(chunks).toString("utf-8"));
} }
function discardBody(res) { function discardBody(res) {

View file

@ -177,7 +177,13 @@ async function main(argv = process.argv.slice(2), env = process.env, deps = {})
now, now,
}); });
const updated = nextState(state, { openDays: aggregated.openDays, today, limits }); // An incomplete scan freezes nothing. Only a count is printed: a path
// would put a project name in syslog for good.
const complete = aggregated.readErrors === 0;
if (!complete) {
logError(`${PREFIX} warning: read_errors=${aggregated.readErrors}, no day frozen this beat`);
}
const updated = nextState(state, { openDays: aggregated.openDays, today, limits, freeze: complete });
// Cache first. A dry run is read-only: it shows what would be sent and // Cache first. A dry run is read-only: it shows what would be sent and
// leaves the cache exactly as the cron left it. // leaves the cache exactly as the cron left it.
if (!dryRun) { if (!dryRun) {

View file

@ -92,8 +92,9 @@ function loadState(file, { readFile = (f) => fs.readFileSync(f, "utf-8") } = {})
* @param {object} opts.openDays aggregate().openDays * @param {object} opts.openDays aggregate().openDays
* @param {string} opts.today Toronto "YYYY-MM-DD" * @param {string} opts.today Toronto "YYYY-MM-DD"
* @param {object} opts.limits fetchLimits() result * @param {object} opts.limits fetchLimits() result
* @param {boolean} [opts.freeze] false: freeze no new day (incomplete scan)
*/ */
function nextState(state, { openDays = {}, today, limits = null }) { function nextState(state, { openDays = {}, today, limits = null, freeze = true }) {
const yesterday = shiftDay(today, -1); const yesterday = shiftDay(today, -1);
const oldest = shiftDay(today, -RETENTION_DAYS); const oldest = shiftDay(today, -RETENTION_DAYS);
@ -101,7 +102,10 @@ function nextState(state, { openDays = {}, today, limits = null }) {
for (const [day, value] of Object.entries(state.frozenDays || {})) { for (const [day, value] of Object.entries(state.frozenDays || {})) {
if (day >= oldest) frozenDays[day] = value; if (day >= oldest) frozenDays[day] = value;
} }
for (const [day, value] of Object.entries(openDays || {})) { // freeze: false after a scan that hit a read error (#54): a day frozen
// with lines missing would stay wrong for RETENTION_DAYS. The next clean
// beat freezes it instead.
for (const [day, value] of Object.entries(freeze ? openDays || {} : {})) {
if (!isDay(day) || day >= yesterday || day < oldest) continue; if (!isDay(day) || day >= yesterday || day < oldest) continue;
frozenDays[day] = { models: value.models || {}, projects: value.projects || {} }; frozenDays[day] = { models: value.models || {}, projects: value.projects || {} };
} }