From c91ee05f4408038db03ed9701f93d58621113696 Mon Sep 17 00:00:00 2001 From: Akshar Patel Date: Thu, 22 Oct 2026 13:04:24 -0400 Subject: [PATCH] fix(web): handle disk errors during transcription assembly --- .github/workflows/recording-reliability.yml | 1 + .../unit/segments-audio-download.test.ts | 45 +++++++++++++++++++ apps/web/lib/segments-audio-download.ts | 14 ++---- 3 files changed, 50 insertions(+), 10 deletions(-) create mode 100644 apps/web/__tests__/unit/segments-audio-download.test.ts diff --git a/.github/workflows/recording-reliability.yml b/.github/workflows/recording-reliability.yml index 492552d061..78dfe665ab 100644 --- a/.github/workflows/recording-reliability.yml +++ b/.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 \ diff --git a/apps/web/__tests__/unit/segments-audio-download.test.ts b/apps/web/__tests__/unit/segments-audio-download.test.ts new file mode 100644 index 0000000000..2cdb62d98f --- /dev/null +++ b/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" }); + }, + ); +}); diff --git a/apps/web/lib/segments-audio-download.ts b/apps/web/lib/segments-audio-download.ts index 3103e6515a..8c7342c98d 100644 --- a/apps/web/lib/segments-audio-download.ts +++ b/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((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((resolve) => stream.end(() => resolve())); } + await pipeline(downloadSegments(), createWriteStream(outputPath)); return totalBytes; }