diff --git a/apps/server/src/factory/ledger/listRuns.test.ts b/apps/server/src/factory/ledger/listRuns.test.ts index eb563d8a4063..e1cdfdd46298 100644 --- a/apps/server/src/factory/ledger/listRuns.test.ts +++ b/apps/server/src/factory/ledger/listRuns.test.ts @@ -2,7 +2,7 @@ import * as NodeFS from "node:fs"; import * as NodePath from "node:path"; -import { afterEach, describe, expect, it } from "vite-plus/test"; +import { afterEach, describe, expect, it, vi } from "vite-plus/test"; import type { State } from "../domain/lifecycle.ts"; import { makeCause } from "./events.ts"; @@ -11,6 +11,12 @@ import { holdsSlot, listUnfinishedRuns } from "./listRuns.ts"; import { RUNS_DIR, runIdFileName, type RunId } from "./runId.ts"; import { T0, ticking, useTempDirs } from "./testSupport.ts"; +// Records every full-file read; `node:fs` namespaces are frozen, so the module itself is wrapped. +vi.mock("node:fs", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, readFileSync: vi.fn(actual.readFileSync) }; +}); + const tempDir = useTempDirs(); const handles: Array = []; @@ -163,6 +169,126 @@ describe("listUnfinishedRuns: whether a session may be running", () => { }); }); +const line = (event: string, data: unknown, ms = T0) => + `${JSON.stringify({ ts: "2026-10-04T10:13:16.123Z", ms, event, data })}\n`; +const stateLine = (to: string) => line("state_changed", { from: "ready", to }); +const finishedLine = line("finished", { outcome: "stuck" }); + +const writeRun = (dataDir: string, id: RunId, content: string) => { + NodeFS.mkdirSync(NodePath.join(dataDir, RUNS_DIR), { recursive: true }); + const path = NodePath.join(dataDir, RUNS_DIR, runIdFileName(id)); + NodeFS.writeFileSync(path, content); + return path; +}; + +const fullReads = () => vi.mocked(NodeFS.readFileSync).mock.calls.map(([path]) => String(path)); + +describe("listUnfinishedRuns: finished runs are skipped by their tail", () => { + it("never reads a large finished run whole, and still lists an unfinished sibling", () => { + const dataDir = tempDir(); + const finished = writeRun( + dataDir, + rid(1), + stateLine("triage") + + line("step_changed", { step: "x".repeat(200) }).repeat(1000) + + finishedLine, + ); + expect(NodeFS.statSync(finished).size).toBeGreaterThan(64 * 1024); + const open = writeRun(dataDir, rid(2), stateLine("building")); + vi.mocked(NodeFS.readFileSync).mockClear(); + expect(listUnfinishedRuns(dataDir)).toEqual([ + { id: rid(2), state: "building", unreadable: false, confirmed: false, session: false }, + ]); + expect(fullReads()).toContain(open); + expect(fullReads()).not.toContain(finished); + }); + + it("skips a small finished run, also after a trailing blank line or CRLF", () => { + const dataDir = tempDir(); + const small = writeRun(dataDir, rid(1), stateLine("triage") + finishedLine); + const blank = writeRun(dataDir, rid(2), stateLine("triage") + finishedLine + "\n\n"); + const crlf = writeRun(dataDir, rid(3), stateLine("triage") + finishedLine + "\r\n"); + vi.mocked(NodeFS.readFileSync).mockClear(); + expect(listUnfinishedRuns(dataDir)).toEqual([]); + for (const path of [small, blank, crlf]) expect(fullReads()).not.toContain(path); + }); + + it("reads a file whose last line is cut off in full and lists it as unfinished", () => { + const dataDir = tempDir(); + writeRun(dataDir, rid(1), stateLine("building") + finishedLine.slice(0, 40)); + expect(listUnfinishedRuns(dataDir)).toEqual([ + { id: rid(1), state: "building", unreadable: false, confirmed: false, session: false }, + ]); + }); + + it("reads a file in full when its last line names finished but is not a valid entry", () => { + const dataDir = tempDir(); + const path = writeRun( + dataDir, + rid(1), + stateLine("building") + '{"event":"finished","data":{}}\n', + ); + vi.mocked(NodeFS.readFileSync).mockClear(); + expect(listUnfinishedRuns(dataDir)).toEqual([ + { id: rid(1), state: "building", unreadable: false, confirmed: false, session: false }, + ]); + expect(fullReads()).toContain(path); + }); + + it("reads a file in full when a cut-off fragment follows the finished line", () => { + const dataDir = tempDir(); + const path = writeRun( + dataDir, + rid(1), + stateLine("triage") + finishedLine + finishedLine.slice(0, 40), + ); + vi.mocked(NodeFS.readFileSync).mockClear(); + expect(listUnfinishedRuns(dataDir)).toEqual([]); + expect(fullReads()).toContain(path); + }); + + it("reads a file whose last complete line is not JSON in full and lists it as unfinished", () => { + const dataDir = tempDir(); + writeRun(dataDir, rid(1), stateLine("building") + "garbage\n"); + expect(listUnfinishedRuns(dataDir)).toEqual([ + { id: rid(1), state: "building", unreadable: false, confirmed: false, session: false }, + ]); + }); + + it("a finished line longer than the tail is still skipped, after a full read", () => { + const dataDir = tempDir(); + const long = writeRun( + dataDir, + rid(1), + stateLine("triage") + line("finished", { outcome: "stuck", pad: "y".repeat(20_000) }), + ); + vi.mocked(NodeFS.readFileSync).mockClear(); + expect(listUnfinishedRuns(dataDir)).toEqual([]); + expect(fullReads()).toContain(long); + }); + + it("lists a run file of only blank lines as unfinished, as readRun reads it", () => { + const dataDir = tempDir(); + writeRun(dataDir, rid(1), "\n"); + writeRun(dataDir, rid(2), "\n\n"); + writeRun(dataDir, rid(3), "\n\r\n"); + const unfinished = { state: null, unreadable: false, confirmed: false, session: false }; + expect(listUnfinishedRuns(dataDir)).toEqual([ + { id: rid(1), ...unfinished }, + { id: rid(2), ...unfinished }, + { id: rid(3), ...unfinished }, + ]); + }); + + it("lists a large run file whose whole tail is blank lines, with its state", () => { + const dataDir = tempDir(); + writeRun(dataDir, rid(1), stateLine("building") + "\n".repeat(9000)); + expect(listUnfinishedRuns(dataDir)).toEqual([ + { id: rid(1), state: "building", unreadable: false, confirmed: false, session: false }, + ]); + }); +}); + describe("holdsSlot", () => { const run = (state: State | null, confirmed: boolean, unreadable = false) => ({ id: rid(1), diff --git a/apps/server/src/factory/ledger/listRuns.ts b/apps/server/src/factory/ledger/listRuns.ts index cf260579b2e6..3a0aa09124bc 100644 --- a/apps/server/src/factory/ledger/listRuns.ts +++ b/apps/server/src/factory/ledger/listRuns.ts @@ -5,8 +5,8 @@ import * as NodePath from "node:path"; import type { State } from "../domain/lifecycle.ts"; import { readProgress } from "../runner/progress.ts"; import { errnoCode, LedgerIoError } from "./errors.ts"; -import { readRun } from "./read.ts"; -import { asciiLower, RUNS_DIR, runIdFromFileName, type RunId } from "./runId.ts"; +import { lastLineIsFinished, readRun } from "./read.ts"; +import { asciiLower, RUNS_DIR, runIdFromFileName, runPath, type RunId } from "./runId.ts"; /** One unfinished run as the pass sees it: its id, its last state, and whether its file could be read. */ export interface RunInfo { @@ -37,7 +37,8 @@ export const holdsSlot = (run: RunInfo): boolean => run.unreadable || (run.confirmed && run.state !== "pr-open"); /** - * Every run under `/runs/` that is not finished, oldest first by the first line's time; with + * Every run under `/runs/` that is not finished (a file whose last line is `finished` is + * skipped by reading only its tail), oldest first by the first line's time; with * `only`, the runs of that one issue, and no other file is read. Files that are not named like a * run are ignored, and a missing runs directory is an empty list. A run file that cannot be read * is still listed, flagged `unreadable`, so one bad file does not hide the others. A run with no @@ -68,6 +69,7 @@ export const listUnfinishedRuns = ( let firstMs: number | null = null; let info: RunInfo; try { + if (lastLineIsFinished(runPath(dataDir, id))) continue; const record = readRun(dataDir, id); firstMs = record.entries[0]?.ms ?? null; const seen = readProgress(record.entries); diff --git a/apps/server/src/factory/ledger/read.ts b/apps/server/src/factory/ledger/read.ts index d92e8a0221b0..8dd0c574745b 100644 --- a/apps/server/src/factory/ledger/read.ts +++ b/apps/server/src/factory/ledger/read.ts @@ -152,3 +152,60 @@ export const readRun = (dataDir: string, id: RunId): RunRecord => { const bytes = readRunBytes(runPath(dataDir, id)); return { id, ...parseLines(bytes) }; }; + +/** How much of a file's end `lastLineIsFinished` reads; a `finished` line is a few hundred bytes. */ +const TAIL_BYTES = 8192; + +/** + * Whether the last complete line of a run file is a valid `finished` event, read from at most + * `TAIL_BYTES` at the end of the file. `finish` writes that line last, so a run for which this is + * true is over without parsing the rest. Answers false for anything it cannot be sure of: a cut-off + * last line, a last line that is not a valid entry, or one longer than the tail. Never fails on + * content; IO errors map as `readRun` maps them. + */ +export const lastLineIsFinished = (path: string): boolean => { + requireRegularFile(path); + let tail: Buffer; + let fromStart: boolean; + try { + const fd = NodeFS.openSync(path, NodeFS.constants.O_RDONLY | NodeFS.constants.O_NOFOLLOW); + try { + const size = NodeFS.fstatSync(fd).size; + const offset = Math.max(0, size - TAIL_BYTES); + tail = Buffer.alloc(size - offset); + let filled = 0; + while (filled < tail.length) { + const n = NodeFS.readSync(fd, tail, filled, tail.length - filled, offset + filled); + if (n === 0) break; + filled += n; + } + tail = tail.subarray(0, filled); + fromStart = offset === 0; + } finally { + NodeFS.closeSync(fd); + } + } catch (error) { + if (errnoCode(error) === "ENOENT") throw new LedgerNotFoundError({ path }); + throw new LedgerIoError({ path, cause: error }); + } + + // Bytes after the last newline are a write that was cut off, as `parseLines` reads them. + let end = tail.lastIndexOf(NEWLINE) + 1; + if (end === 0 || !isBlankLine(tail.subarray(end))) return false; + // Step back over blank lines to the last line that holds something. + for (;;) { + const lineEnd = end - 1; + // A negative offset would count from the end of the buffer, so the first line has no earlier one. + const start = lineEnd === 0 ? 0 : tail.lastIndexOf(NEWLINE, lineEnd - 1) + 1; + const line = tail.subarray(start, lineEnd); + if (isBlankLine(line)) { + if (start === 0) return false; + end = start; + continue; + } + // A line that starts the chunk of a longer file may have begun before it. + if (start === 0 && !fromStart) return false; + const object = decodeLine(line); + return object !== null && lineHead(object)?.event === "finished"; + } +};