diff --git a/CHANGELOG.md b/CHANGELOG.md index afb1bfb..78a4bb4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ # Changelog +## [4.42.1] + +- 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.42.0] - Add Text-to-Speech: `client.tts.synthesize(request)` returns one text's complete audio (`TextToSpeechResponse`: `audio` plus the response headers as `requestId`, `region`, `model`, `voiceId`, `encoding`, `sampleRate`, `billedCharacters`, `audioDurationMs`, ...), `client.tts.stream(request)` returns it as the API generates it (`TextToSpeechStream`, an async iterable over a web `ReadableStream`), and `client.tts.session(params)` opens a WebSocket `TextToSpeechSession` for voice agents: `connect()`, `generate()` / `flush()` / `audio()`, `speak()`, `cancel()`, `keepAlive()`, `close()`, with `on("audio" | "flushDone" | "wordBoundaries" | "cancelled" | "termination" | "error" | "close", ...)`. Requests take the API's fields (`text`, `voice_id`, `model_id`, `language`, `sample_rate`, `encoding`, `container`, `bit_rate`); failures throw `TextToSpeechError` (`status`, `errorCode`, `param`, `requestId`, `retryable`, `retryAfter`, `errors`, `closeCode`, `body`). New client option `ttsBaseUrl` (default `https://tts.us.assemblyai.com`); `client.tts.createTemporaryToken()` mints a token for browser sessions. Text sent while `connect()` is pending is held and sent the moment the socket opens, so it need not wait for `Begin` diff --git a/package.json b/package.json index f4507f2..772a97f 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "assemblyai", - "version": "4.42.0", + "version": "4.42.1", "description": "The AssemblyAI JavaScript SDK provides an easy-to-use interface for interacting with the AssemblyAI API, which supports async and real-time transcription and text-to-speech.", "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); + }); +});