fix(realtime): reject connect() promise when WebSocket errors before …

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

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"

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<SessionBeginsEventData>((resolve) => {
+    return new Promise<SessionBeginsEventData>((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<void>,

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(

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",