Skip to content

Commit 94cb14b

Browse files
wangmiao0668000666wangmiao0668000666
authored
fix(transcripts): close stream on parse failure
1 parent 90e31be commit 94cb14b

2 files changed

Lines changed: 127 additions & 8 deletions

File tree

‎src/transcripts/store.test.ts‎

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
// Tests TranscriptsStore stream cleanup and transcript reading behavior.
2+
import fs from "node:fs";
3+
import path from "node:path";
4+
import { afterEach, describe, expect, it } from "vitest";
5+
import { listOpenFileDescriptorsForPath } from "../../src/infra/open-file-descriptors.test-support.js";
6+
import { cleanupTempDirs, makeTempDir } from "../../test/helpers/temp-dir.js";
7+
import { TranscriptsStore } from "./store.js";
8+
9+
const tempRoots: string[] = [];
10+
11+
describe("TranscriptsStore.readUtterancesFromSessionDir", () => {
12+
afterEach(() => {
13+
cleanupTempDirs(tempRoots);
14+
});
15+
16+
it("returns an empty array when transcript.jsonl is missing", () => {
17+
const tmpDir = makeTempDir(tempRoots, "openclaw-transcript-test-");
18+
const store = new TranscriptsStore(tmpDir);
19+
const sessionDir = path.join(tmpDir, "2026-07-01", "missing");
20+
fs.mkdirSync(sessionDir, { recursive: true });
21+
22+
const result = store.readUtterancesFromSessionDir(sessionDir, { maxUtterances: 10 });
23+
24+
return expect(result).resolves.toEqual([]);
25+
});
26+
27+
it("reads utterances from transcript.jsonl", () => {
28+
const tmpDir = makeTempDir(tempRoots, "openclaw-transcript-test-");
29+
const store = new TranscriptsStore(tmpDir);
30+
const sessionDir = path.join(tmpDir, "2026-07-01", "session-1");
31+
fs.mkdirSync(sessionDir, { recursive: true });
32+
fs.writeFileSync(
33+
path.join(sessionDir, "transcript.jsonl"),
34+
[
35+
JSON.stringify({ text: "hello", sessionId: "session-1" }),
36+
JSON.stringify({ text: "world", sessionId: "session-1" }),
37+
].join("\n") + "\n",
38+
);
39+
40+
const result = store.readUtterancesFromSessionDir(sessionDir, { maxUtterances: 10 });
41+
42+
return expect(result).resolves.toEqual([
43+
expect.objectContaining({ text: "hello" }),
44+
expect.objectContaining({ text: "world" }),
45+
]);
46+
});
47+
48+
it("keeps only the tail when utterances exceed maxUtterances", () => {
49+
const tmpDir = makeTempDir(tempRoots, "openclaw-transcript-test-");
50+
const store = new TranscriptsStore(tmpDir);
51+
const sessionDir = path.join(tmpDir, "2026-07-01", "session-1");
52+
fs.mkdirSync(sessionDir, { recursive: true });
53+
const lines = Array.from({ length: 5 }, (_, i) =>
54+
JSON.stringify({ text: `line-${i}`, sessionId: "session-1" }),
55+
);
56+
fs.writeFileSync(path.join(sessionDir, "transcript.jsonl"), lines.join("\n") + "\n");
57+
58+
const result = store.readUtterancesFromSessionDir(sessionDir, { maxUtterances: 2 });
59+
60+
return expect(result).resolves.toEqual([
61+
expect.objectContaining({ text: "line-3" }),
62+
expect.objectContaining({ text: "line-4" }),
63+
]);
64+
});
65+
66+
it.runIf(process.platform === "linux")(
67+
"does not leak file descriptors when JSON.parse throws",
68+
async () => {
69+
const tmpDir = makeTempDir(tempRoots, "openclaw-transcript-test-");
70+
const store = new TranscriptsStore(tmpDir);
71+
const sessionDir = path.join(tmpDir, "2026-07-01", "session-1");
72+
fs.mkdirSync(sessionDir, { recursive: true });
73+
const transcriptPath = path.join(sessionDir, "transcript.jsonl");
74+
fs.writeFileSync(transcriptPath, "not valid json\n");
75+
76+
const fdsBefore = listOpenFileDescriptorsForPath(sessionDir);
77+
await expect(
78+
store.readUtterancesFromSessionDir(sessionDir, { maxUtterances: 10 }),
79+
).rejects.toThrow();
80+
const fdsAfter = listOpenFileDescriptorsForPath(sessionDir);
81+
82+
const leaked = fdsAfter.filter((p) => !fdsBefore.includes(p));
83+
expect(leaked).toHaveLength(0);
84+
},
85+
);
86+
87+
it.runIf(process.platform === "linux")(
88+
"does not leak file descriptors in the happy path",
89+
async () => {
90+
const tmpDir = makeTempDir(tempRoots, "openclaw-transcript-test-");
91+
const store = new TranscriptsStore(tmpDir);
92+
const sessionDir = path.join(tmpDir, "2026-07-01", "session-1");
93+
fs.mkdirSync(sessionDir, { recursive: true });
94+
const transcriptPath = path.join(sessionDir, "transcript.jsonl");
95+
fs.writeFileSync(
96+
transcriptPath,
97+
JSON.stringify({ text: "hello", sessionId: "session-1" }) + "\n",
98+
);
99+
100+
const fdsBefore = listOpenFileDescriptorsForPath(sessionDir);
101+
await store.readUtterancesFromSessionDir(sessionDir, { maxUtterances: 10 });
102+
const fdsAfter = listOpenFileDescriptorsForPath(sessionDir);
103+
104+
const leaked = fdsAfter.filter((p) => !fdsBefore.includes(p));
105+
expect(leaked).toHaveLength(0);
106+
},
107+
);
108+
});

‎src/transcripts/store.ts‎

Lines changed: 19 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -196,18 +196,29 @@ export class TranscriptsStore {
196196
if (maxUtterances !== undefined) {
197197
const utterances: TranscriptUtterance[] = [];
198198
try {
199+
const stream = createReadStream(transcriptPath, { encoding: "utf8" });
199200
const lines = createInterface({
200-
input: createReadStream(transcriptPath, { encoding: "utf8" }),
201+
input: stream,
201202
crlfDelay: Infinity,
202203
});
203-
for await (const line of lines) {
204-
if (!line) {
205-
continue;
204+
try {
205+
for await (const line of lines) {
206+
if (!line) {
207+
continue;
208+
}
209+
utterances.push(JSON.parse(line) as TranscriptUtterance);
210+
if (utterances.length > maxUtterances) {
211+
// Stream and keep only the tail so large transcripts do not require full-file memory.
212+
utterances.shift();
213+
}
206214
}
207-
utterances.push(JSON.parse(line) as TranscriptUtterance);
208-
if (utterances.length > maxUtterances) {
209-
// Stream and keep only the tail so large transcripts do not require full-file memory.
210-
utterances.shift();
215+
} finally {
216+
lines.close();
217+
stream.destroy();
218+
if (!stream.closed) {
219+
await new Promise<void>((resolve) => {
220+
stream.once("close", () => resolve());
221+
});
211222
}
212223
}
213224
} catch (err) {

0 commit comments

Comments
 (0)