fix(live): cancel a web ReadableStream source when the upload stops early

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

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"

src/utils/live.ts

@@ -177,13 +177,22 @@ export async function* liveChunks(
 ): AsyncGenerator<Uint8Array> {
   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 {

tests/unit/live-chunks.test.ts

@@ -0,0 +1,64 @@
+import { liveChunks } from "../../src/utils/live";
+
+function sourceStream(
+  chunks: unknown[],
+  onCancel: jest.Mock,
+): ReadableStream<Uint8Array> {
+  let index = 0;
+  return new ReadableStream<Uint8Array>({
+    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);
+  });
+});