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,17 @@ 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 }),
+			(rmErr, tempPath) =>
+				this.logger.warn("Failed to delete temp file", tempPath, rmErr),
+		);
 	}
 }

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/range.ts

@@ -102,15 +102,21 @@ export function validateUtcDateInput(value: string): string | undefined {
 		: "Enter a valid calendar date.";
 }
 
+/** Parses a telemetry ISO timestamp to epoch ms, throwing on unparseable input. */
+export function parseTelemetryTimestampMs(timestamp: string): number {
+	const ms = Date.parse(timestamp);
+	if (!Number.isFinite(ms)) {
+		throw new Error(`Invalid telemetry timestamp '${timestamp}'.`);
+	}
+	return ms;
+}
+
 /** True if the ISO `timestamp` falls inside the range. */
 export function isTimestampInRange(
 	timestamp: string,
 	range: TelemetryDateRange,
 ): boolean {
-	const ms = Date.parse(timestamp);
-	if (!Number.isFinite(ms)) {
-		throw new Error(`Invalid telemetry timestamp '${timestamp}'.`);
-	}
+	const ms = parseTelemetryTimestampMs(timestamp);
 	return (
 		(range.startMs === undefined || ms >= range.startMs) &&
 		(range.endMs === undefined || ms < range.endMs)

src/telemetry/export/writers/json.ts

@@ -0,0 +1,42 @@
+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";
+
+/**
+ * Streams `events` as a JSON array to `outputPath` via a temp file and
+ * atomic rename. Returns the number of events written. `onCleanupError`
+ * is invoked if removing the temp file after a failed write itself fails
+ * (typically a Windows lock); callers are expected to log it.
+ */
+export async function writeJsonArrayExport(
+	outputPath: string,
+	events: AsyncIterable<TelemetryEvent>,
+	onCleanupError: (err: unknown, tempPath: string) => void,
+): 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" }),
+			);
+		},
+		onCleanupError,
+	);
+	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,84 @@
+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)}`;
+}
+
+/**
+ * Atomically writes to `outputPath` via a sibling temp file and rename.
+ * The parent directory must already exist. On failure the destination is
+ * left untouched, the temp file is best-effort removed, and the writer
+ * error is always rethrown. `onCleanupError` receives any error from the
+ * cleanup attempt; its own throws are swallowed.
+ */
+export async function writeAtomically<T>(
+	outputPath: string,
+	write: (tempPath: string) => Promise<T>,
+	onCleanupError: (err: unknown, tempPath: string) => void,
+): Promise<T> {
+	const tempPath = tempFilePath(outputPath, "temp");
+	try {
+		const result = await write(tempPath);
+		await renameWithRetry(fs.rename, tempPath, outputPath);
+		return result;
+	} catch (err) {
+		try {
+			await fs.rm(tempPath, { force: true }).catch((rmErr) => {
+				onCleanupError(rmErr, tempPath);
+			});
+		} catch {
+			// onCleanupError threw; the writer error below takes precedence.
+		}
+		throw err;
+	}
+}

test/mocks/asyncIterable.ts

@@ -0,0 +1,12 @@
+/**
+ * Wraps a sync array as an `AsyncIterable` that yields one item per microtask,
+ * so consumers exercise the same async iteration path they would in production.
+ */
+export async function* asyncIterable<T>(
+	values: readonly T[],
+): AsyncGenerator<T> {
+	for (const value of values) {
+		await Promise.resolve();
+		yield value;
+	}
+}

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/remote/sshProcess.test.ts

@@ -445,7 +445,10 @@ describe("SshProcessMonitor", () => {
 			});
 			vi.mocked(find)
 				.mockResolvedValueOnce([{ pid: 999, ppid: 1, name: "ssh", cmd: "ssh" }])
-				.mockResolvedValue([{ pid: 888, ppid: 1, name: "ssh", cmd: "ssh" }]);
+				.mockResolvedValueOnce([{ pid: 888, ppid: 1, name: "ssh", cmd: "ssh" }])
+				// No rediscovery after takeover: a same-PID rediscovery would
+				// emit ssh.process.recovered and break the negative assertion.
+				.mockResolvedValue([]);
 
 			const monitor = createMonitor({
 				networkInfoPath: "/network",
@@ -456,8 +459,6 @@ describe("SshProcessMonitor", () => {
 			await waitUntil(
 				() => sink.eventsNamed("ssh.process.replaced").length > 0,
 			);
-			// Halt monitoring before the negative assertion so the 888 loop
-			// can't race ahead and emit its own lost/recovered cycle.
 			monitor.dispose();
 
 			const replaced = sink.eventsNamed("ssh.process.replaced");

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

@@ -11,19 +11,10 @@ import { serializeTelemetryEventLine } 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;
-});
+vi.mock("node:fs", async () => (await import("memfs")).fs);
+vi.mock("node:fs/promises", async () => (await import("memfs")).fs.promises);
 
 const DIR = "/telemetry";