fix(web): handle disk errors during transcription assembly

.github/workflows/recording-reliability.yml

@@ -46,6 +46,7 @@ jobs:
             __tests__/unit/recording-storage-lifecycle.test.ts \
             __tests__/unit/desktop-recording-*.test.ts \
             __tests__/unit/desktop-segments-*.test.ts \
+            __tests__/unit/segments-audio-download.test.ts \
             __tests__/unit/finalize-desktop-recording.test.ts \
             __tests__/unit/media-server-progress.test.ts \
             __tests__/unit/media-processing-budget.test.ts \

apps/web/__tests__/unit/segments-audio-download.test.ts

@@ -0,0 +1,45 @@
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { downloadConcatenatedSegments } from "@/lib/segments-audio-download";
+
+describe("segment audio output errors", () => {
+	let directory: string;
+
+	beforeEach(async () => {
+		directory = await mkdtemp(join(tmpdir(), "cap-segments-output-"));
+	});
+
+	afterEach(async () => {
+		await rm(directory, { recursive: true, force: true });
+	});
+
+	const delayedFetch: typeof fetch = async () => {
+		await new Promise((resolve) => setTimeout(resolve, 20));
+		return new Response("audio segment");
+	};
+
+	it("rejects an output open failure while a segment is downloading", async () => {
+		await expect(
+			downloadConcatenatedSegments(
+				["https://fixture.invalid/audio.m4s"],
+				join(directory, "missing", "audio.mp4"),
+				{ fetchImpl: delayedFetch },
+			),
+		).rejects.toMatchObject({ code: "ENOENT" });
+	});
+
+	it.skipIf(process.platform !== "linux")(
+		"rejects a full output device without emitting an uncaught stream error",
+		async () => {
+			await expect(
+				downloadConcatenatedSegments(
+					["https://fixture.invalid/audio.m4s"],
+					"/dev/full",
+					{ fetchImpl: delayedFetch },
+				),
+			).rejects.toMatchObject({ code: "ENOSPC" });
+		},
+	);
+});

apps/web/lib/segments-audio-download.ts

@@ -1,4 +1,5 @@
 import { createWriteStream } from "node:fs";
+import { pipeline } from "node:stream/promises";
 
 /**
  * Byte-level segment assembly. Lives apart from the pure planning helpers in
@@ -83,15 +84,9 @@ export async function downloadConcatenatedSegments(
 	);
 	const fetchImpl = options.fetchImpl ?? fetch;
 
-	const stream = createWriteStream(outputPath);
 	let totalBytes = 0;
 
-	const write = (chunk: Buffer) =>
-		new Promise<void>((resolve, reject) => {
-			stream.write(chunk, (error) => (error ? reject(error) : resolve()));
-		});
-
-	try {
+	async function* downloadSegments() {
 		for (let start = 0; start < urls.length; start += concurrency) {
 			const batch = urls.slice(start, start + concurrency);
 			const buffers = await Promise.all(
@@ -113,12 +108,11 @@ export async function downloadConcatenatedSegments(
 						`Segment audio exceeds ${MAX_SEGMENTS_AUDIO_BYTES} bytes`,
 					);
 				}
-				await write(buffer);
+				yield buffer;
 			}
 		}
-	} finally {
-		await new Promise<void>((resolve) => stream.end(() => resolve()));
 	}
 
+	await pipeline(downloadSegments(), createWriteStream(outputPath));
 	return totalBytes;
 }