Skip to content

Commit 5c189f0

Browse files
EhabYEhabY
committed
fix: address telemetry export review feedback
Close opened session iterators on early termination, simplify cursor payloads, avoid repeated timestamp parsing, and restore range-filtering coverage.
1 parent 9c0145b commit 5c189f0

2 files changed

Lines changed: 172 additions & 55 deletions

File tree

‎src/telemetry/export/files.ts‎

Lines changed: 93 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -26,15 +26,14 @@ interface TelemetryLogFile {
2626
readonly part: number;
2727
}
2828

29-
interface TelemetryEventEntry {
29+
interface QueuedTelemetryEvent {
3030
readonly event: TelemetryEvent;
31-
readonly file: TelemetryLogFile;
32-
readonly lineNumber: number;
31+
readonly timestampMs: number;
3332
}
3433

3534
interface EventCursor {
36-
readonly entry: TelemetryEventEntry;
37-
readonly iterator: AsyncIterator<TelemetryEventEntry>;
35+
readonly queued: QueuedTelemetryEvent;
36+
readonly iterator: AsyncIterator<TelemetryEvent>;
3837
}
3938

4039
/** Log files whose dates could overlap `range`. */
@@ -62,41 +61,33 @@ export async function listTelemetryFilesForRange(
6261
.map(({ path: filePath }) => filePath);
6362
}
6463

65-
/** Merge per-session append streams by timestamp, buffering one event per session. */
64+
/**
65+
* Merge range-filtered, per-session event streams by timestamp. This stays
66+
* globally sorted while each session's timestamps are monotonic; if a session's
67+
* clock moves backward, that session's append order is preserved.
68+
*/
6669
export async function* streamTelemetryEventsSorted(
6770
filePaths: readonly string[],
6871
range: TelemetryDateRange,
6972
): AsyncIterable<TelemetryEvent> {
73+
const iterators: Array<AsyncIterator<TelemetryEvent>> = [];
7074
const frontier: EventCursor[] = [];
71-
for (const files of sessionFileGroups(filePaths)) {
72-
const iterator = streamTelemetryEventEntries(files, range)[
73-
Symbol.asyncIterator
74-
]();
75-
const next = await iterator.next();
76-
if (!next.done) {
77-
frontier.push({ entry: next.value, iterator });
78-
}
79-
}
80-
81-
while (frontier.length > 0) {
82-
frontier.sort((a, b) => compareEventEntries(a.entry, b.entry));
83-
const cursor = frontier.shift();
84-
if (!cursor) {
85-
return;
86-
}
87-
yield cursor.entry.event;
88-
89-
const next = await cursor.iterator.next();
90-
if (!next.done) {
91-
frontier.push({ entry: next.value, iterator: cursor.iterator });
75+
try {
76+
await seedFrontier(frontier, iterators, filePaths, range);
77+
while (frontier.length > 0) {
78+
const cursor = takeNextCursor(frontier);
79+
yield cursor.queued.event;
80+
await advanceCursor(frontier, cursor);
9281
}
82+
} finally {
83+
await closeIterators(iterators);
9384
}
9485
}
9586

96-
async function* streamTelemetryEventEntries(
87+
async function* streamTelemetryEventsFromFiles(
9788
files: readonly TelemetryLogFile[],
9889
range: TelemetryDateRange,
99-
): AsyncIterable<TelemetryEventEntry> {
90+
): AsyncIterable<TelemetryEvent> {
10091
for (const file of files) {
10192
const name = path.basename(file.path);
10293
const stream = createReadStream(file.path, { encoding: "utf8" });
@@ -113,7 +104,7 @@ async function* streamTelemetryEventEntries(
113104
}
114105
const event = parseTelemetryEventLine(line, name, lineNumber);
115106
if (isTimestampInRange(event.timestamp, range)) {
116-
yield { event, file, lineNumber };
107+
yield event;
117108
}
118109
}
119110
} catch (err) {
@@ -135,23 +126,22 @@ async function* streamTelemetryEventEntries(
135126
}
136127
}
137128

138-
function sessionFileGroups(filePaths: readonly string[]): TelemetryLogFile[][] {
139-
const groups = new Map<string, TelemetryLogFile[]>();
140-
for (const file of parseLogFilePaths(filePaths).sort(compareLogFiles)) {
141-
const group = groups.get(file.session);
142-
if (group) {
143-
group.push(file);
144-
} else {
145-
groups.set(file.session, [file]);
146-
}
147-
}
148-
return [...groups.values()];
129+
function groupFilesBySession(
130+
filePaths: readonly string[],
131+
): TelemetryLogFile[][] {
132+
const files = parseLogFilePaths(filePaths).sort(compareLogFiles);
133+
return [...Map.groupBy(files, (file) => file.session).values()];
149134
}
150135

151136
function parseLogFilePaths(filePaths: readonly string[]): TelemetryLogFile[] {
152-
return filePaths.flatMap((filePath) => {
137+
return filePaths.map((filePath) => {
153138
const file = parseLogFilePath(filePath);
154-
return file ? [file] : [];
139+
if (!file) {
140+
throw new Error(
141+
`Invalid telemetry file path: ${path.basename(filePath)}`,
142+
);
143+
}
144+
return file;
155145
});
156146
}
157147

@@ -168,15 +158,68 @@ function compareLogFiles(a: TelemetryLogFile, b: TelemetryLogFile): number {
168158
);
169159
}
170160

171-
function compareEventEntries(
172-
a: TelemetryEventEntry,
173-
b: TelemetryEventEntry,
161+
async function seedFrontier(
162+
frontier: EventCursor[],
163+
iterators: Array<AsyncIterator<TelemetryEvent>>,
164+
filePaths: readonly string[],
165+
range: TelemetryDateRange,
166+
): Promise<void> {
167+
for (const files of groupFilesBySession(filePaths)) {
168+
const iterator = streamTelemetryEventsFromFiles(files, range)[
169+
Symbol.asyncIterator
170+
]();
171+
iterators.push(iterator);
172+
await advanceCursor(frontier, { iterator });
173+
}
174+
}
175+
176+
async function advanceCursor(
177+
frontier: EventCursor[],
178+
cursor: Pick<EventCursor, "iterator">,
179+
): Promise<void> {
180+
const next = await cursor.iterator.next();
181+
if (!next.done) {
182+
frontier.push({
183+
queued: queueEvent(next.value),
184+
iterator: cursor.iterator,
185+
});
186+
}
187+
}
188+
189+
async function closeIterators(
190+
iterators: ReadonlyArray<AsyncIterator<TelemetryEvent>>,
191+
): Promise<void> {
192+
await Promise.allSettled(
193+
iterators.map(async (iterator) => {
194+
await iterator.return?.();
195+
}),
196+
);
197+
}
198+
199+
function queueEvent(event: TelemetryEvent): QueuedTelemetryEvent {
200+
return { event, timestampMs: parseTelemetryTimestampMs(event.timestamp) };
201+
}
202+
203+
function takeNextCursor(frontier: EventCursor[]): EventCursor {
204+
let nextIndex = 0;
205+
for (let i = 1; i < frontier.length; i += 1) {
206+
if (
207+
compareQueuedEvents(frontier[i].queued, frontier[nextIndex].queued) < 0
208+
) {
209+
nextIndex = i;
210+
}
211+
}
212+
const cursor = frontier[nextIndex];
213+
frontier.splice(nextIndex, 1);
214+
return cursor;
215+
}
216+
217+
function compareQueuedEvents(
218+
a: QueuedTelemetryEvent,
219+
b: QueuedTelemetryEvent,
174220
): number {
175-
const timestamp =
176-
parseTelemetryTimestampMs(a.event.timestamp) -
177-
parseTelemetryTimestampMs(b.event.timestamp);
178221
return (
179-
timestamp ||
222+
a.timestampMs - b.timestampMs ||
180223
a.event.context.sessionId.localeCompare(b.event.context.sessionId)
181224
);
182225
}

‎test/unit/telemetry/export/files.test.ts‎

Lines changed: 79 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -13,14 +13,31 @@ import { createTelemetryEventFactory } from "../../../mocks/telemetry";
1313

1414
import type { TelemetryEvent } from "@/telemetry/event";
1515

16-
vi.mock("node:fs", async () => (await import("memfs")).fs);
16+
const streamTracker = vi.hoisted(() => ({ destroyed: 0 }));
17+
18+
vi.mock("node:fs", async () => {
19+
const { fs } = await import("memfs");
20+
return {
21+
...fs,
22+
createReadStream: (...args: Parameters<typeof fs.createReadStream>) => {
23+
const stream = fs.createReadStream(...args);
24+
const destroy = stream.destroy.bind(stream);
25+
stream.destroy = (err?: Error) => {
26+
streamTracker.destroyed += 1;
27+
return destroy(err);
28+
};
29+
return stream;
30+
},
31+
};
32+
});
1733
vi.mock("node:fs/promises", async () => (await import("memfs")).fs.promises);
1834

1935
const DIR = "/telemetry";
2036

2137
let makeEvent: ReturnType<typeof createTelemetryEventFactory>;
2238

2339
beforeEach(() => {
40+
streamTracker.destroyed = 0;
2441
vol.reset();
2542
vol.mkdirSync(DIR, { recursive: true });
2643
makeEvent = createTelemetryEventFactory();
@@ -99,25 +116,67 @@ describe("streamTelemetryEventsSorted", () => {
99116
]);
100117
});
101118

102-
it("preserves events when a session timestamp moves backward", async () => {
119+
it("filters by range and preserves backward session timestamps", async () => {
103120
writeFiles({
104121
"telemetry-2026-05-12-aaaaaaaa.jsonl":
105122
serializeTelemetryEventLine(
106-
makeSessionEvent("session-a", 0, "2026-05-12T10:02:00.000Z"),
123+
makeSessionEvent("session-a", 0, "2026-05-11T23:59:59.999Z"),
107124
) +
108125
serializeTelemetryEventLine(
109-
makeSessionEvent("session-a", 1, "2026-05-12T10:00:00.000Z"),
126+
makeSessionEvent("session-a", 1, "2026-05-12T10:02:00.000Z"),
127+
) +
128+
serializeTelemetryEventLine(
129+
makeSessionEvent("session-a", 2, "2026-05-12T10:00:00.000Z"),
110130
),
131+
"telemetry-2026-05-12-bbbbbbbb.jsonl": serializeTelemetryEventLine(
132+
makeSessionEvent("session-b", 0, "2026-05-12T10:01:00.000Z"),
133+
),
111134
});
112135

113-
const events = await collectSorted(["telemetry-2026-05-12-aaaaaaaa.jsonl"]);
136+
const events = await collectSorted([
137+
"telemetry-2026-05-12-aaaaaaaa.jsonl",
138+
"telemetry-2026-05-12-bbbbbbbb.jsonl",
139+
]);
114140

115141
expect(events.map((event) => event.timestamp)).toEqual([
142+
"2026-05-12T10:01:00.000Z",
116143
"2026-05-12T10:02:00.000Z",
117144
"2026-05-12T10:00:00.000Z",
118145
]);
119146
});
120147

148+
it("closes opened session streams when abandoned before exhaustion", async () => {
149+
writeFiles({
150+
"telemetry-2026-05-12-aaaaaaaa.jsonl":
151+
serializeTelemetryEventLine(
152+
makeSessionEvent("session-a", 0, "2026-05-12T10:00:00.000Z"),
153+
) +
154+
serializeTelemetryEventLine(
155+
makeSessionEvent("session-a", 1, "2026-05-12T10:02:00.000Z"),
156+
),
157+
"telemetry-2026-05-12-bbbbbbbb.jsonl":
158+
serializeTelemetryEventLine(
159+
makeSessionEvent("session-b", 0, "2026-05-12T10:01:00.000Z"),
160+
) +
161+
serializeTelemetryEventLine(
162+
makeSessionEvent("session-b", 1, "2026-05-12T10:03:00.000Z"),
163+
),
164+
});
165+
const iterator = streamTelemetryEventsSorted(
166+
[
167+
`${DIR}/telemetry-2026-05-12-aaaaaaaa.jsonl`,
168+
`${DIR}/telemetry-2026-05-12-bbbbbbbb.jsonl`,
169+
],
170+
createCustomDateRange("2026-05-12", "2026-05-12"),
171+
)[Symbol.asyncIterator]();
172+
173+
await expect(iterator.next()).resolves.toMatchObject({ done: false });
174+
expect(streamTracker.destroyed).toBe(0);
175+
await iterator.return?.();
176+
177+
expect(streamTracker.destroyed).toBe(2);
178+
});
179+
121180
it("surfaces parse errors with file:line context", async () => {
122181
writeFiles({
123182
"telemetry-2026-05-12-aaaaaaaa.jsonl": "{not-json}\n",
@@ -129,6 +188,21 @@ describe("streamTelemetryEventsSorted", () => {
129188
/Failed to parse telemetry file telemetry-2026-05-12-aaaaaaaa\.jsonl:1/,
130189
);
131190
});
191+
192+
it("rejects invalid telemetry file paths instead of skipping them", async () => {
193+
writeFiles({
194+
"telemetry-2026-05-12-aaaaaaaa.jsonl": serializeTelemetryEventLine(
195+
makeSessionEvent("session-a", 0, "2026-05-12T10:00:00.000Z"),
196+
),
197+
});
198+
199+
await expect(
200+
collectSorted([
201+
"telemetry-2026-05-12-aaaaaaaa.jsonl",
202+
"notes-2026-05-12.jsonl",
203+
]),
204+
).rejects.toThrow("Invalid telemetry file path: notes-2026-05-12.jsonl");
205+
});
132206
});
133207

134208
function writeFiles(files: Record<string, string>): void {

0 commit comments

Comments
 (0)