diff --git a/CHANGELOG.md b/CHANGELOG.md index 6106b84..5245e7e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ # Changelog +## [4.41.6] + +- Fix `transcribeLive()` on the sync and dictation services leaving a web `ReadableStream` audio source open when the upload ends early (a rejected request, an abort, a timeout, or a chunk that is not bytes). The stream is now cancelled, matching how Node streams and other async iterables were already released + ## [4.41.5] - Add `speakerLabelsRevisionIntervalMs` streaming param (sent as `speaker_labels_revision_interval_ms`) — the cadence, in ms of audio time, at which the server emits mid-stream `speakerRevision` events when `speakerLabels` is enabled. Unset or `0` sends only the end-of-stream revision; non-zero values are clamped server-side to 120 000–300 000 ms. The first mid-stream revision needs ~120 s of streamed speech, and one is sent only when some earlier turn's label changed diff --git a/package.json b/package.json index 8266741..944ab0a 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "assemblyai", - "version": "4.41.5", + "version": "4.41.6", "description": "The AssemblyAI JavaScript SDK provides an easy-to-use interface for interacting with the AssemblyAI API, which supports async and real-time transcription.", "engines": { "node": ">=18" diff --git a/src/utils/live.ts b/src/utils/live.ts index 08e78f5..b9e9b69 100644 --- a/src/utils/live.ts +++ b/src/utils/live.ts @@ -177,13 +177,22 @@ export async function* liveChunks( ): AsyncGenerator { if (isWebReadableStream(input)) { const reader = input.getReader(); + let exhausted = false; try { for (;;) { const { done, value } = await reader.read(); - if (done) return; + if (done) { + exhausted = true; + return; + } yield asBytes(value); } } finally { + // Stopping early (the upload was rejected, aborted or timed out, or a + // chunk was not bytes) must release the source, as `for await` does for + // Node streams and other async iterables. A cancel that fails because + // the stream already errored has nothing left to release. + if (!exhausted) reader.cancel().catch(() => undefined); reader.releaseLock(); } } else { diff --git a/tests/unit/live-chunks.test.ts b/tests/unit/live-chunks.test.ts new file mode 100644 index 0000000..658a27d --- /dev/null +++ b/tests/unit/live-chunks.test.ts @@ -0,0 +1,64 @@ +import { liveChunks } from "../../src/utils/live"; + +function sourceStream( + chunks: unknown[], + onCancel: jest.Mock, +): ReadableStream { + let index = 0; + return new ReadableStream({ + pull(controller) { + if (index < chunks.length) { + controller.enqueue(chunks[index++] as Uint8Array); + } else { + controller.close(); + } + }, + cancel: onCancel, + }); +} + +describe("liveChunks with a web ReadableStream", () => { + it("yields every chunk and leaves a fully read stream alone", async () => { + const onCancel = jest.fn(); + const received: number[] = []; + for await (const chunk of liveChunks( + sourceStream([new Uint8Array([1, 2]), new Uint8Array([3])], onCancel), + )) { + received.push(...chunk); + } + + expect(received).toEqual([1, 2, 3]); + expect(onCancel).not.toHaveBeenCalled(); + }); + + it("cancels the source when the consumer stops early", async () => { + const onCancel = jest.fn(); + const chunks = liveChunks( + sourceStream( + [new Uint8Array([1]), new Uint8Array([2]), new Uint8Array([3])], + onCancel, + ), + ); + + const first = await chunks.next(); + expect(first.value).toEqual(new Uint8Array([1])); + await chunks.return(undefined); + + expect(onCancel).toHaveBeenCalledTimes(1); + }); + + it("cancels the source when a chunk is not bytes", async () => { + const onCancel = jest.fn(); + const chunks = liveChunks( + sourceStream( + [new Uint8Array([1]), "text", new Uint8Array([3])], + onCancel, + ), + ); + + await chunks.next(); + await expect(chunks.next()).rejects.toThrow(TypeError); + + expect(onCancel).toHaveBeenCalledTimes(1); + }); +});