Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
128 changes: 127 additions & 1 deletion apps/server/src/factory/ledger/listRuns.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand All @@ -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<typeof import("node:fs")>();
return { ...actual, readFileSync: vi.fn(actual.readFileSync) };
});

const tempDir = useTempDirs();

const handles: Array<Ledger> = [];
Expand Down Expand Up @@ -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),
Expand Down
8 changes: 5 additions & 3 deletions apps/server/src/factory/ledger/listRuns.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -37,7 +37,8 @@ export const holdsSlot = (run: RunInfo): boolean =>
run.unreadable || (run.confirmed && run.state !== "pr-open");

/**
* Every run under `<dir>/runs/` that is not finished, oldest first by the first line's time; with
* Every run under `<dir>/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
Expand Down Expand Up @@ -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);
Expand Down
57 changes: 57 additions & 0 deletions apps/server/src/factory/ledger/read.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";
}
};