diff --git a/CHANGELOG.md b/CHANGELOG.md index 141449a..de78dbd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ # Changelog +## [4.41.4] + +- Fix `RealtimeTranscriber.connect()` promise hanging indefinitely when a socket error or unexpected close occurs before `SessionBegins`. The promise now rejects, and the socket reference is discarded so that connections can be safely retried. + ## [4.41.3] - Export the helper types (`LiteralUnion`) from the package root, so a downstream TypeScript package can export a value typed by `Transcript.language_code`, `TranscriptParams.language_code` or `SyncSpeechModel` without its declaration emit failing with TS2742 (the internal path `assemblyai/dist/types/helpers` is not in the `exports` map). No runtime change diff --git a/package.json b/package.json index dc2df75..cff4429 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "assemblyai", - "version": "4.41.3", + "version": "4.41.4", "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/services/realtime/service.ts b/src/services/realtime/service.ts index 8be0de1..fff8273 100644 --- a/src/services/realtime/service.ts +++ b/src/services/realtime/service.ts @@ -65,6 +65,7 @@ export class RealtimeTranscriber { private socket?: PolyfillWebSocket; private listeners: RealtimeListeners = {}; private sessionTerminatedResolve?: () => void; + private connectReject?: (reason: Error) => void; /** * Create a new RealtimeTranscriber. @@ -187,11 +188,13 @@ export class RealtimeTranscriber { * @returns A promise that resolves when the connection is established and the session begins. */ connect() { - return new Promise((resolve) => { + return new Promise((resolve, reject) => { if (this.socket) { throw new Error("Already connected"); } + this.connectReject = reject; + const url = this.connectionUrl(); if (this.token) { @@ -231,12 +234,33 @@ Learn more at https://github.com/AssemblyAI/assemblyai-node-sdk/blob/main/docs/c // The socket is gone, so no `SessionTerminated` message is coming. // Release a `close()` that is waiting for one. this.resolveSessionTermination(); + const rejectConnect = this.connectReject; + if (rejectConnect) { + this.connectReject = undefined; + this.discardPendingSocket(); + rejectConnect( + new Error( + `Realtime connection closed before session started: ${code} ${reason ?? ""}`.trim(), + ), + ); + } this.listeners.close?.(code, reason); }; this.socket!.onerror = (event: ErrorEvent) => { - if (event.error) this.listeners.error?.(event.error as Error); - else this.listeners.error?.(new Error(event.message)); + const error = event.error + ? (event.error as Error) + : new Error(event.message); + // Reject the connect() promise if SessionBegins has not yet resolved it. + // This prevents the promise from hanging when the socket fails during + // the handshake. + const rejectConnect = this.connectReject; + if (rejectConnect) { + this.connectReject = undefined; + this.discardPendingSocket(); + rejectConnect(error); + } + this.listeners.error?.(error); }; this.socket!.onmessage = ({ data }: MessageEvent) => { @@ -251,6 +275,9 @@ Learn more at https://github.com/AssemblyAI/assemblyai-node-sdk/blob/main/docs/c sessionId: message.session_id, expiresAt: new Date(message.expires_at), }; + // Session started successfully — clear the rejection handler so + // subsequent onerror calls (after handshake) only fire listeners. + this.connectReject = undefined; resolve(openObject); this.listeners.open?.(openObject); break; @@ -338,6 +365,26 @@ Learn more at https://github.com/AssemblyAI/assemblyai-node-sdk/blob/main/docs/c resolve?.(); } + /** Tear down a half-open socket from a failed connection attempt. */ + private discardPendingSocket(): void { + if (!this.socket) return; + try { + if (this.socket.removeAllListeners) { + this.socket.removeAllListeners(); + // `ws` aborts a still-CONNECTING handshake by emitting `error` on the + // next tick, and an `error` emit with no listener crashes the process + // as an uncaughtException — outside this try/catch and any caller's. + // Keep a sink attached; the failure is already reported through the + // rejected connect() promise. + this.socket.onerror = () => {}; + } + this.socket.close(); + } catch { + // Best-effort cleanup; a half-open socket may throw on close. + } + this.socket = undefined; + } + /** Awaits the session-termination message, bounded by `timeoutMs`. */ private async waitForTermination( sessionTerminated: Promise, diff --git a/tests/unit/realtime.test.ts b/tests/unit/realtime.test.ts index 9bb28a0..b4c4215 100644 --- a/tests/unit/realtime.test.ts +++ b/tests/unit/realtime.test.ts @@ -84,6 +84,63 @@ describe("realtime", () => { ); }); + // Regression: connect() must reject when a WebSocket error + // fires before SessionBegins instead of hanging indefinitely. + it("rejects connect() when WebSocket errors before SessionBegins", async () => { + const errorUrl = "wss://localhost:9991"; + const errorServer = new WS(errorUrl); + const rtNew = new RealtimeTranscriber({ realtimeUrl: errorUrl, apiKey }); + + const connectPromise = rtNew.connect(); + await errorServer.connected; + + // Simulate a network error before the server sends SessionBegins. + errorServer.error({ + code: 0, + reason: "Connection refused", + wasClean: false, + }); + + await expect(connectPromise).rejects.toThrow(); + WS.clean(); + }); + + // After a failed connect(), this.socket must be dropped so retrying does + // not throw "Already connected". + it("allows reconnecting after connect() fails", async () => { + const errorUrl = "wss://localhost:9992"; + const errorServer = new WS(errorUrl); + const rtNew = new RealtimeTranscriber({ realtimeUrl: errorUrl, apiKey }); + + const firstConnect = rtNew.connect(); + await errorServer.connected; + errorServer.error({ + code: 0, + reason: "Connection refused", + wasClean: false, + }); + await expect(firstConnect).rejects.toThrow(); + WS.clean(); + + // Reconnecting should not throw "Already connected" + const retryServer = new WS(errorUrl); + await connect(rtNew, retryServer); + await close(rtNew, retryServer); + WS.clean(); + }); + + // After SessionBegins, onerror must only fire the error listener — the + // already-resolved connect() promise must not be double-rejected. + it("does not reject an already-resolved connect() on subsequent errors", async () => { + // `rt` is already connected by beforeEach; further errors fire listener only. + const onError = jest.fn(); + rt.on("error", onError); + + server.error({ code: 0, reason: "Post-handshake error", wasClean: false }); + + expect(onError).toHaveBeenCalledWith(expect.any(Error)); + }); + it("fails to send audio with closed websocket", async () => { await close(rt, server); expect(() => rt.sendAudio(new ArrayBuffer(8))).toThrow( diff --git a/tests/unit/streaming-connecting-teardown.test.ts b/tests/unit/streaming-connecting-teardown.test.ts index 1af6b9a..d85cd0c 100644 --- a/tests/unit/streaming-connecting-teardown.test.ts +++ b/tests/unit/streaming-connecting-teardown.test.ts @@ -95,6 +95,21 @@ describe("tearing down a CONNECTING socket", () => { expect(() => socket.emitDeferredHandshakeAbort()).not.toThrow(); }); + it("RealtimeTranscriber.discardPendingSocket() leaves an error sink attached", () => { + const rt = new RealtimeTranscriber({ + apiKey: "123", + sampleRate: 16_000, + }); + const socket = new FakeConnectingWsSocket(); + injectSocket(rt, socket); + + (rt as unknown as { discardPendingSocket(): void }).discardPendingSocket(); + + expect(socket.closeCalled).toBe(true); + expect(socket.listenerCount("error")).toBeGreaterThan(0); + expect(() => socket.emitDeferredHandshakeAbort()).not.toThrow(); + }); + it("RealtimeTranscriber.close() leaves an error sink attached", async () => { const rt = new RealtimeTranscriber({ apiKey: "123",