diff --git a/CHANGELOG.md b/CHANGELOG.md index f96d1b6..337b37d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ # Changelog +## [4.36.8] + +- Fix an uncaught exception that killed the process when a still-connecting socket was torn down — on a connection-attempt timeout or a `close()` racing `connect()`, `ws` emits `error` on the next tick after the SDK had removed all listeners. Teardown now keeps an error sink attached, and a timed-out `connect()` rejects into the caller's `catch` instead of crashing first + ## [4.36.7] - Fix `StreamingTranscriber.close()` and `RealtimeTranscriber.close()` hanging forever when the socket closes without a `Termination` message — `onclose` now releases the pending wait, and the wait is bounded by a new optional `terminationTimeout` parameter on `close()` (5000ms default; `0` waits indefinitely). The socket is closed either way diff --git a/package.json b/package.json index 80d630f..e5aef75 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "assemblyai", - "version": "4.36.7", + "version": "4.36.8", "description": "The AssemblyAI JavaScript SDK provides an easy-to-use interface for interacting with the AssemblyAI API, which supports async and real-time transcription, as well as the latest LeMUR models.", "engines": { "node": ">=18" diff --git a/src/services/realtime/service.ts b/src/services/realtime/service.ts index 739fa9e..8be0de1 100644 --- a/src/services/realtime/service.ts +++ b/src/services/realtime/service.ts @@ -388,7 +388,13 @@ Learn more at https://github.com/AssemblyAI/assemblyai-node-sdk/blob/main/docs/c this.socket.send(terminateSessionMessage); } } - if (this.socket?.removeAllListeners) this.socket.removeAllListeners(); + if (this.socket?.removeAllListeners) { + this.socket.removeAllListeners(); + // Closing a still-CONNECTING socket (close() racing connect()) makes + // `ws` emit `error` on the next tick; keep a sink attached so that + // emit cannot become an uncaughtException. + this.socket.onerror = () => {}; + } this.socket.close(); } diff --git a/src/services/streaming/service.ts b/src/services/streaming/service.ts index 2fff85e..db7d63c 100644 --- a/src/services/streaming/service.ts +++ b/src/services/streaming/service.ts @@ -749,7 +749,15 @@ Learn more at https://github.com/AssemblyAI/assemblyai-node-sdk/blob/main/docs/c private discardPendingSocket(): void { if (!this.socket) return; try { - if (this.socket.removeAllListeners) this.socket.removeAllListeners(); + 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. @@ -1116,7 +1124,13 @@ Learn more at https://github.com/AssemblyAI/assemblyai-node-sdk/blob/main/docs/c this.socket.send(terminateSessionMessage); } } - if (this.socket?.removeAllListeners) this.socket.removeAllListeners(); + if (this.socket?.removeAllListeners) { + this.socket.removeAllListeners(); + // Closing a still-CONNECTING socket (close() racing connect()) makes + // `ws` emit `error` on the next tick; keep a sink attached so that + // emit cannot become an uncaughtException. + this.socket.onerror = () => {}; + } this.socket.close(); } diff --git a/tests/unit/streaming-connecting-teardown.test.ts b/tests/unit/streaming-connecting-teardown.test.ts new file mode 100644 index 0000000..1af6b9a --- /dev/null +++ b/tests/unit/streaming-connecting-teardown.test.ts @@ -0,0 +1,112 @@ +import { EventEmitter } from "events"; +import { StreamingTranscriber, RealtimeTranscriber } from "../../src"; + +// Regression tests for +// https://github.com/AssemblyAI/assemblyai-node-sdk/issues/170. +// +// When a socket in the CONNECTING state is closed, `ws` aborts the handshake +// and emits `error` on the next tick. Node's EventEmitter turns an `error` +// emit with zero listeners into a thrown error — an uncaughtException that +// kills the process, since the emit happens outside any caller's try/catch. +// The SDK's teardown paths call `removeAllListeners()` before `close()`, so +// they must leave an error sink attached for that deferred emit to land on. +// +// The shared `tests/unit/mocks/ws.ts` mock can't catch this: its +// `removeAllListeners()` replaces handlers with no-ops instead of removing +// them, and mock-socket never does the deferred emit. This fake reproduces +// the relevant `ws` semantics: EventEmitter-backed listeners and an +// `onerror` attribute that registers a real listener. +class FakeConnectingWsSocket extends EventEmitter { + CONNECTING = 0 as const; + OPEN = 1 as const; + CLOSING = 2 as const; + CLOSED = 3 as const; + readyState = 0; + binaryType = "arraybuffer"; + closeCalled = false; + + private errorAttributeHandler: ((event: unknown) => void) | null = null; + + set onerror(handler: ((event: unknown) => void) | null) { + if (this.errorAttributeHandler) { + this.removeListener("error", this.errorAttributeHandler); + } + this.errorAttributeHandler = handler; + if (handler) this.on("error", handler); + } + + get onerror(): ((event: unknown) => void) | null { + return this.errorAttributeHandler; + } + + send(): void {} + + close(): void { + this.closeCalled = true; + } + + // ws's deferred `emitErrorAndClose`, run synchronously so a missing + // listener fails this test instead of crashing the jest worker. + emitDeferredHandshakeAbort(): void { + this.emit( + "error", + new Error("WebSocket was closed before the connection was established"), + ); + } +} + +function injectSocket( + transcriber: StreamingTranscriber | RealtimeTranscriber, + socket: FakeConnectingWsSocket, +): void { + (transcriber as unknown as { socket: unknown }).socket = socket; +} + +describe("tearing down a CONNECTING socket", () => { + it("discardPendingSocket() leaves an error sink attached", () => { + const rt = new StreamingTranscriber({ + 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("StreamingTranscriber.close() leaves an error sink attached", async () => { + const rt = new StreamingTranscriber({ + apiKey: "123", + sampleRate: 16_000, + }); + const socket = new FakeConnectingWsSocket(); + injectSocket(rt, socket); + + // close() racing an unresolved connect(): the socket is not OPEN, so the + // Termination handshake is skipped and teardown runs immediately. + await rt.close(); + + 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", + sampleRate: 16_000, + }); + const socket = new FakeConnectingWsSocket(); + injectSocket(rt, socket); + + await rt.close(); + + expect(socket.closeCalled).toBe(true); + expect(socket.listenerCount("error")).toBeGreaterThan(0); + expect(() => socket.emitDeferredHandshakeAbort()).not.toThrow(); + }); +});