feat(telemetry): add streaming JSON export writer

src/core/cliCredentialManager.ts

@@ -9,7 +9,8 @@ import { isAbortError } from "../error/errorUtils";
 import { featureSetForVersion } from "../featureSet";
 import { isKeyringEnabled } from "../settings/cli";
 import { getHeaderArgs } from "../settings/headers";
-import { renameWithRetry, tempFilePath, toSafeHost } from "../util";
+import { toSafeHost } from "../util";
+import { writeAtomically } from "../util/fs";
 
 import { version } from "./cliExec";
 
@@ -259,23 +260,14 @@ export class CliCredentialManager {
 		}
 	}
 
-	/**
-	 * Atomically write content to a file via temp-file + rename.
-	 */
+	/** Atomically write content to a file. */
 	private async atomicWriteFile(
 		filePath: string,
 		content: string,
 	): Promise<void> {
 		await fs.mkdir(path.dirname(filePath), { recursive: true });
-		const tempPath = tempFilePath(filePath, "temp");
-		try {
-			await fs.writeFile(tempPath, content, { mode: 0o600 });
-			await renameWithRetry(fs.rename, tempPath, filePath);
-		} catch (err) {
-			await fs.rm(tempPath, { force: true }).catch((rmErr) => {
-				this.logger.warn("Failed to delete temp file", tempPath, rmErr);
-			});
-			throw err;
-		}
+		await writeAtomically(filePath, (tempPath) =>
+			fs.writeFile(tempPath, content, { mode: 0o600 }),
+		);
 	}
 }

src/core/cliManager.ts

@@ -13,7 +13,8 @@ import { errToStr } from "../api/api-helper";
 import * as pgp from "../pgp";
 import { withCancellableProgress, withOptionalProgress } from "../progress";
 import { isKeyringEnabled } from "../settings/cli";
-import { tempFilePath, toSafeHost } from "../util";
+import { toSafeHost } from "../util";
+import { tempFilePath } from "../util/fs";
 import { vscodeProposed } from "../vscodeProposed";
 
 import { BinaryLock } from "./binaryLock";

src/core/supportBundleLogs.ts

@@ -4,7 +4,7 @@ import * as path from "node:path";
 import { promisify } from "node:util";
 
 import { type Logger } from "../logging/logger";
-import { renameWithRetry } from "../util";
+import { renameWithRetry } from "../util/fs";
 
 export interface LogSources {
 	remoteSshLogPath?: string;

src/remote/sshConfig.ts

@@ -8,7 +8,8 @@ import {
 } from "node:fs/promises";
 import path from "node:path";
 
-import { countSubstring, renameWithRetry, tempFilePath } from "../util";
+import { countSubstring } from "../util";
+import { renameWithRetry, tempFilePath } from "../util/fs";
 
 import type { Logger } from "../logging/logger";
 

src/telemetry/export/writers.ts

@@ -0,0 +1,37 @@
+import { createWriteStream } from "node:fs";
+import { Readable } from "node:stream";
+import { pipeline } from "node:stream/promises";
+
+import { writeAtomically } from "../../util/fs";
+import { serializeTelemetryEvent } from "../wireFormat";
+
+import type { TelemetryEvent } from "../event";
+
+/**
+ * Writes `events` as a JSON array to `outputPath` via a temp file + atomic
+ * rename, so a partial write never replaces the destination. Streams chunks
+ * with backpressure so memory stays flat even for large exports.
+ * Returns the number of events written.
+ */
+export async function writeJsonArrayExport(
+	outputPath: string,
+	events: AsyncIterable<TelemetryEvent>,
+): Promise<number> {
+	let count = 0;
+	async function* chunks(): AsyncGenerator<string> {
+		yield "[";
+		for await (const event of events) {
+			yield (count === 0 ? "\n" : ",\n") +
+				JSON.stringify(serializeTelemetryEvent(event));
+			count += 1;
+		}
+		yield count === 0 ? "]\n" : "\n]\n";
+	}
+	await writeAtomically(outputPath, async (tempPath) => {
+		await pipeline(
+			Readable.from(chunks()),
+			createWriteStream(tempPath, { encoding: "utf8" }),
+		);
+	});
+	return count;
+}

src/util.ts

@@ -164,51 +164,6 @@ export function countSubstring(needle: string, haystack: string): number {
 	return count;
 }
 
-const transientRenameCodes: ReadonlySet<string> = new Set([
-	"EPERM",
-	"EACCES",
-	"EBUSY",
-]);
-
-/**
- * Rename with retry for transient Windows filesystem errors (EPERM, EACCES,
- * EBUSY). On Windows, antivirus, Search Indexer, cloud sync, or concurrent
- * processes can briefly lock files causing renames to fail.
- *
- * On non-Windows platforms, calls renameFn directly with no retry.
- *
- * Matches the strategy used by VS Code (pfs.ts) and graceful-fs: 60s
- * wall-clock timeout with linear backoff (10ms increments) capped at 100ms.
- */
-export async function renameWithRetry(
-	renameFn: (src: string, dest: string) => Promise<void>,
-	source: string,
-	destination: string,
-	timeoutMs = 60_000,
-	delayCapMs = 100,
-): Promise<void> {
-	if (process.platform !== "win32") {
-		return renameFn(source, destination);
-	}
-	const startTime = Date.now();
-	for (let attempt = 1; ; attempt++) {
-		try {
-			return await renameFn(source, destination);
-		} catch (err) {
-			const code = (err as NodeJS.ErrnoException).code;
-			if (
-				!code ||
-				!transientRenameCodes.has(code) ||
-				Date.now() - startTime >= timeoutMs
-			) {
-				throw err;
-			}
-			const delay = Math.min(delayCapMs, attempt * 10);
-			await new Promise((resolve) => setTimeout(resolve, delay));
-		}
-	}
-}
-
 /**
  * Wraps `arg` in `"..."` unless every character is in the shell-safe
  * whitelist (matching Python `shlex.quote`'s set: alphanumerics plus
@@ -240,12 +195,3 @@ export function escapeShellArg(arg: string): string {
 	}
 	return `'${arg.replace(/'/g, "'\\''")}'`;
 }
-
-/**
- * Generate a temporary file path by appending a suffix with a random component.
- * The suffix describes the purpose of the temp file (e.g. "temp", "old").
- * Example: tempFilePath("/a/b", "temp") → "/a/b.temp-k7x3f9qw"
- */
-export function tempFilePath(basePath: string, suffix: string): string {
-	return `${basePath}.${suffix}-${crypto.randomUUID().substring(0, 8)}`;
-}

src/util/fs.ts

@@ -0,0 +1,77 @@
+import * as fs from "node:fs/promises";
+
+const transientRenameCodes: ReadonlySet<string> = new Set([
+	"EPERM",
+	"EACCES",
+	"EBUSY",
+]);
+
+/**
+ * Rename with retry for transient Windows filesystem errors (EPERM, EACCES,
+ * EBUSY). On Windows, antivirus, Search Indexer, cloud sync, or concurrent
+ * processes can briefly lock files causing renames to fail.
+ *
+ * On non-Windows platforms, calls renameFn directly with no retry.
+ *
+ * Matches the strategy used by VS Code (pfs.ts) and graceful-fs: 60s
+ * wall-clock timeout with linear backoff (10ms increments) capped at 100ms.
+ */
+export async function renameWithRetry(
+	renameFn: (src: string, dest: string) => Promise<void>,
+	source: string,
+	destination: string,
+	timeoutMs = 60_000,
+	delayCapMs = 100,
+): Promise<void> {
+	if (process.platform !== "win32") {
+		return renameFn(source, destination);
+	}
+	const startTime = Date.now();
+	for (let attempt = 1; ; attempt++) {
+		try {
+			return await renameFn(source, destination);
+		} catch (err) {
+			const code = (err as NodeJS.ErrnoException).code;
+			if (
+				!code ||
+				!transientRenameCodes.has(code) ||
+				Date.now() - startTime >= timeoutMs
+			) {
+				throw err;
+			}
+			const delay = Math.min(delayCapMs, attempt * 10);
+			await new Promise((resolve) => setTimeout(resolve, delay));
+		}
+	}
+}
+
+/**
+ * Generate a temporary file path by appending a suffix with a random component.
+ * The suffix describes the purpose of the temp file (e.g. "temp", "old").
+ * Example: tempFilePath("/a/b", "temp") → "/a/b.temp-k7x3f9qw"
+ */
+export function tempFilePath(basePath: string, suffix: string): string {
+	return `${basePath}.${suffix}-${crypto.randomUUID().substring(0, 8)}`;
+}
+
+/**
+ * Writes `outputPath` atomically: `write` builds the content at a sibling
+ * temp path, then we rename onto the destination. A failure mid-write never
+ * touches the destination; the temp file is best-effort cleaned up.
+ */
+export async function writeAtomically<T>(
+	outputPath: string,
+	write: (tempPath: string) => Promise<T>,
+): Promise<T> {
+	const tempPath = tempFilePath(outputPath, "tmp");
+	try {
+		const result = await write(tempPath);
+		await renameWithRetry(fs.rename, tempPath, outputPath);
+		return result;
+	} catch (err) {
+		await fs.rm(tempPath, { force: true }).catch(() => {
+			// Surface the original failure, not the cleanup failure.
+		});
+		throw err;
+	}
+}

test/unit/core/supportBundleLogs.test.ts

@@ -5,14 +5,14 @@ import * as path from "node:path";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 
 import { appendVsCodeLogs } from "@/core/supportBundleLogs";
-import { renameWithRetry } from "@/util";
+import { renameWithRetry } from "@/util/fs";
 
 import { createMockLogger } from "../../mocks/testHelpers";
 
 // Wrap renameWithRetry so individual tests can override it via
 // mockRejectedValueOnce; by default it calls through to the real impl.
-vi.mock("@/util", async () => {
-	const actual = await vi.importActual<typeof import("@/util")>("@/util");
+vi.mock("@/util/fs", async () => {
+	const actual = await vi.importActual<typeof import("@/util/fs")>("@/util/fs");
 	return { ...actual, renameWithRetry: vi.fn(actual.renameWithRetry) };
 });
 

test/unit/telemetry/export/writers.test.ts

@@ -0,0 +1,104 @@
+import { vol } from "memfs";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+import { writeJsonArrayExport } from "@/telemetry/export/writers";
+import { serializeTelemetryEvent } from "@/telemetry/wireFormat";
+
+import { createTelemetryEventFactory } from "../../../mocks/telemetry";
+
+import type * as fs from "node:fs";
+
+import type { TelemetryEvent } from "@/telemetry/event";
+
+vi.mock("node:fs/promises", async () => {
+	const memfs: { fs: typeof fs } = await vi.importActual("memfs");
+	return memfs.fs.promises;
+});
+
+vi.mock("node:fs", async () => {
+	const memfs: { fs: typeof fs } = await vi.importActual("memfs");
+	return memfs.fs;
+});
+
+const DIR = "/exports";
+
+let makeEvent: ReturnType<typeof createTelemetryEventFactory>;
+
+beforeEach(() => {
+	vol.reset();
+	vol.mkdirSync(DIR, { recursive: true });
+	makeEvent = createTelemetryEventFactory();
+});
+
+afterEach(() => {
+	vol.reset();
+});
+
+describe("writeJsonArrayExport", () => {
+	it("writes events as a JSON array in wire format", async () => {
+		const events = [
+			makeEvent({
+				eventName: "first",
+				properties: { result: "success" },
+				measurements: { durationMs: 12 },
+				traceId: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+			}),
+			makeEvent({
+				eventName: "second",
+				parentEventId: "id-0",
+				error: { message: "boom", type: "Error" },
+			}),
+		];
+
+		const count = await writeJsonArrayExport(
+			`${DIR}/telemetry.json`,
+			asyncIterable(events),
+		);
+
+		expect(count).toBe(2);
+		expect(readJson(`${DIR}/telemetry.json`)).toEqual(
+			events.map(serializeTelemetryEvent),
+		);
+	});
+
+	it("writes a valid empty JSON array when there are no events", async () => {
+		const count = await writeJsonArrayExport(
+			`${DIR}/empty.json`,
+			asyncIterable([]),
+		);
+
+		expect(count).toBe(0);
+		expect(readJson(`${DIR}/empty.json`)).toEqual([]);
+	});
+
+	it("leaves the destination untouched when writing fails midway", async () => {
+		const outputPath = `${DIR}/telemetry.json`;
+		vol.writeFileSync(outputPath, "previous content");
+
+		const events = (async function* () {
+			yield makeEvent();
+			await Promise.resolve();
+			throw new Error("boom");
+		})();
+
+		await expect(writeJsonArrayExport(outputPath, events)).rejects.toThrow(
+			/boom/,
+		);
+
+		expect(vol.readFileSync(outputPath, "utf8")).toBe("previous content");
+		expect(vol.readdirSync(DIR)).toEqual(["telemetry.json"]);
+	});
+});
+
+async function* asyncIterable(
+	events: readonly TelemetryEvent[],
+): AsyncGenerator<TelemetryEvent> {
+	for (const event of events) {
+		await Promise.resolve();
+		yield event;
+	}
+}
+
+function readJson(filePath: string): unknown {
+	return JSON.parse(vol.readFileSync(filePath, "utf8") as string);
+}