Skip to content

Commit 88acffc

Browse files
committed
fix(session): publish partial records without a torn line
appendFile creates partial.jsonl before the JSON line is written, so a reader can observe empty or truncated JSON. Rewrite through a temp file and rename, and serialize overlapping flushes so one publish cannot drop the other.
1 parent 4198732 commit 88acffc

2 files changed

Lines changed: 105 additions & 6 deletions

File tree

‎src/session/stream-journal.test.ts‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -124,6 +124,60 @@ describe("createCycleTextRecorder", () => {
124124
expect(records[0]?.error?.message).toBe("aborted");
125125
});
126126

127+
test("a reader never observes a torn partial.jsonl record", async () => {
128+
// appendFile creates the file before the line lands, so a poller can read
129+
// empty or truncated JSON. Every observed byte must already parse.
130+
for (let trial = 0; trial < 30; trial++) {
131+
const temp = createTempDir("stream-journal-race-");
132+
try {
133+
const recorder = createCycleTextRecorder(() => temp.dir);
134+
recorder.handleEvent({
135+
type: "inference.error",
136+
data: {
137+
error: {
138+
category: "rate_limit",
139+
message: "429 too many requests",
140+
statusCode: 429,
141+
},
142+
},
143+
} as unknown as ReactorEmittedEvent);
144+
145+
const started = Date.now();
146+
for (;;) {
147+
let raw: string;
148+
try {
149+
raw = await readFile(join(temp.dir, PARTIAL_FILE), "utf8");
150+
} catch (err) {
151+
if (Date.now() - started >= 2_000) throw err;
152+
continue;
153+
}
154+
const records = parsePartialRaw(raw);
155+
expect(records).toHaveLength(1);
156+
expect(records[0]?.text).toBe("");
157+
expect(records[0]?.error?.statusCode).toBe(429);
158+
break;
159+
}
160+
} finally {
161+
temp.cleanup();
162+
}
163+
}
164+
});
165+
166+
test("two flushes append both records", async () => {
167+
const recorder = createCycleTextRecorder(() => dir);
168+
recorder.handleEvent(delta("one"));
169+
await recorder.flush("cancelled");
170+
recorder.handleEvent(delta("two"));
171+
await recorder.flush("interrupted");
172+
173+
const records = await readPartialRecords();
174+
expect(records.map((record) => record.text)).toEqual(["one", "two"]);
175+
expect(records.map((record) => record.reason)).toEqual([
176+
"cancelled",
177+
"interrupted",
178+
]);
179+
});
180+
127181
test("inference.error with empty cycle text still writes a partial with the error payload", async () => {
128182
// Live failures can emit inference.error with no streamed text; the
129183
// partial must still land so category/message survive.

‎src/session/stream-journal.ts‎

Lines changed: 51 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { appendFile } from "node:fs/promises";
1+
import { readFile, rename, rm, writeFile } from "node:fs/promises";
22
import { join } from "node:path";
33

44
import type { ReactorEmittedEvent } from "@intx/inference";
@@ -21,6 +21,44 @@ export const PARTIAL_FILE = "partial.jsonl";
2121
/** Cap on retained cycle text; older text is dropped from the front. */
2222
export const CYCLE_TEXT_CAP_CHARS = 262_144;
2323

24+
function isEnoent(err: unknown): boolean {
25+
return (
26+
typeof err === "object" &&
27+
err !== null &&
28+
"code" in err &&
29+
(err as { code?: unknown }).code === "ENOENT"
30+
);
31+
}
32+
33+
let partialTmpCounter = 0;
34+
35+
/**
36+
* Append one JSONL line so readers never observe a torn record. `appendFile`
37+
* creates the destination before the bytes land; a poller then reads empty or
38+
* truncated JSON. Rewrite the whole file through a temp sibling and `rename`
39+
* so the path appears only with complete lines.
40+
*/
41+
async function appendPartialLine(dir: string, line: string): Promise<void> {
42+
const target = join(dir, PARTIAL_FILE);
43+
let prior = "";
44+
try {
45+
prior = await readFile(target, "utf8");
46+
} catch (err) {
47+
if (!isEnoent(err)) throw err;
48+
}
49+
const tmp = join(
50+
dir,
51+
`.${PARTIAL_FILE}.${process.pid}.${(partialTmpCounter += 1)}.tmp`,
52+
);
53+
try {
54+
await writeFile(tmp, prior + line, "utf8");
55+
await rename(tmp, target);
56+
} catch (err) {
57+
await rm(tmp, { force: true }).catch(() => undefined);
58+
throw err;
59+
}
60+
}
61+
2462
/** Append a streamed token to the cycle buffer, keeping only the tail. */
2563
export function appendCycleText(
2664
text: string,
@@ -105,6 +143,9 @@ export function createCycleTextRecorder(
105143
let cycleText = "";
106144
let cycleThinkingText = "";
107145
let closed = false;
146+
// Overlapping flushes (inference.error is not awaited) must not both read
147+
// the same prior bytes and rename away the other line.
148+
let publishChain: Promise<void> = Promise.resolve();
108149

109150
const writeRecord = async (
110151
reason: PartialFlushReason,
@@ -132,12 +173,16 @@ export function createCycleTextRecorder(
132173
record.thinkingText = thinkingText;
133174
}
134175
if (error !== undefined) record.error = error;
176+
const line = `${JSON.stringify(record)}\n`;
177+
const run = publishChain.then(() =>
178+
appendPartialLine(resolveContextDir(), line),
179+
);
180+
publishChain = run.then(
181+
() => undefined,
182+
() => undefined,
183+
);
135184
try {
136-
await appendFile(
137-
join(resolveContextDir(), PARTIAL_FILE),
138-
`${JSON.stringify(record)}\n`,
139-
"utf8",
140-
);
185+
await run;
141186
} catch (err) {
142187
getLogger([LOG_NAMESPACE_ROOT, "session", "partial"]).warn(
143188
"failed to write partial stream output: {error}",

0 commit comments

Comments
 (0)