fix(streaming,realtime): cancel a pending connect() when close() is called

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## [4.41.6]
+
+- Fix `StreamingTranscriber.close()` while `connect()` is still pending. Previously the pending attempt was not cancelled, so its retry loop opened a new socket after `close()` returned, and the `connect()` promise could later resolve on a closed transcriber. `close()` now cancels the pending attempt, and `connect()` rejects with a "Connection closed before the session started" error. Calling `connect()` again after `close()` still works
+- Fix `RealtimeTranscriber.close()` while `connect()` is still pending. The `connect()` promise never settled because `close()` removed the socket listeners; it now rejects with the same error
+
 ## [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/services/realtime/service.ts

@@ -420,6 +420,13 @@ Learn more at https://github.com/AssemblyAI/assemblyai-node-sdk/blob/main/docs/c
     waitForSessionTermination = true,
     terminationTimeout = DEFAULT_TERMINATION_TIMEOUT_MS,
   ) {
+    // close() removes the socket listeners below, so nothing else would ever
+    // settle a connect() that is still waiting for `SessionBegins`.
+    const rejectConnect = this.connectReject;
+    this.connectReject = undefined;
+    rejectConnect?.(
+      new RealtimeError("Connection closed before the session started"),
+    );
     if (this.socket) {
       if (this.socket.readyState === this.socket.OPEN) {
         if (waitForSessionTermination) {

src/services/streaming/service.ts

@@ -141,6 +141,10 @@ export class StreamingTranscriber {
   private socket?: PolyfillWebSocket;
   private listeners: StreamingListeners = {};
   private sessionTerminatedResolve?: () => void;
+  // Incremented by `close()` so a `connect()` that is still retrying can tell
+  // the transcriber was closed and stop instead of opening a new socket.
+  private connectEpoch = 0;
+  private cancelPendingConnect?: (error: Error) => void;
 
   // Dual-channel mode state (allocated only when params.channels is set).
   private isDualChannel = false;
@@ -558,8 +562,13 @@ export class StreamingTranscriber {
     const retryDelay =
       this.params.connectionRetryDelay ?? DEFAULT_CONNECTION_RETRY_DELAY_MS;
 
+    const epoch = this.connectEpoch;
+    const closedError = () =>
+      new StreamingError("Connection closed before the session started");
+
     let lastError: Error | undefined;
     for (let attempt = 0; attempt <= maxRetries; attempt++) {
+      if (epoch !== this.connectEpoch) throw closedError();
       try {
         return await this.connectOnce();
       } catch (err) {
@@ -574,6 +583,7 @@ export class StreamingTranscriber {
         if (retryDelay > 0) {
           await new Promise((resolve) => setTimeout(resolve, retryDelay));
         }
+        if (epoch !== this.connectEpoch) throw closedError();
       }
     }
     // The loop above always returns or throws; this only satisfies the type
@@ -597,6 +607,7 @@ export class StreamingTranscriber {
       const failAttempt = (error: ConnectionAttemptError) => {
         if (settled) return;
         settled = true;
+        this.cancelPendingConnect = undefined;
         if (timer) clearTimeout(timer);
         this.discardPendingSocket();
         reject(error);
@@ -605,10 +616,16 @@ export class StreamingTranscriber {
       const succeed = (begin: BeginEvent) => {
         if (settled) return;
         settled = true;
+        this.cancelPendingConnect = undefined;
         if (timer) clearTimeout(timer);
         resolve(begin);
       };
 
+      this.cancelPendingConnect = (error) => {
+        (error as ConnectionAttemptError).retryable = false;
+        failAttempt(error as ConnectionAttemptError);
+      };
+
       if (timeoutMs > 0) {
         timer = setTimeout(() => {
           const err = new StreamingError(
@@ -1120,6 +1137,10 @@ Learn more at https://github.com/AssemblyAI/assemblyai-node-sdk/blob/main/docs/c
     waitForSessionTermination = true,
     terminationTimeout = DEFAULT_TERMINATION_TIMEOUT_MS,
   ) {
+    this.connectEpoch++;
+    this.cancelPendingConnect?.(
+      new StreamingError("Connection closed before the session started"),
+    );
     if (this.flushTimer) {
       clearInterval(this.flushTimer);
       this.flushTimer = undefined;

tests/unit/realtime-close-during-connect.test.ts

@@ -0,0 +1,36 @@
+jest.mock("ws", () => require("./mocks/ws"));
+
+import WS from "jest-websocket-mock";
+import { RealtimeTranscriber } from "../../src";
+
+const realtimeUrl = "wss://localhost:1234";
+
+describe("realtime close() while connect() is pending", () => {
+  let server: WS;
+
+  beforeEach(() => {
+    WS.clean();
+    server = new WS(realtimeUrl);
+  });
+
+  afterEach(() => {
+    WS.clean();
+  });
+
+  it("rejects the pending connect() promise", async () => {
+    const rt = new RealtimeTranscriber({
+      realtimeUrl,
+      token: "123",
+      sampleRate: 16_000,
+    });
+
+    const outcome = rt.connect().then(
+      () => "resolved",
+      (error: Error) => error.message,
+    );
+    await server.connected;
+    await rt.close(false);
+
+    await expect(outcome).resolves.toMatch(/closed/i);
+  });
+});

tests/unit/streaming-close-during-connect.test.ts

@@ -0,0 +1,125 @@
+jest.mock("ws", () => require("./mocks/ws"));
+
+import WS from "jest-websocket-mock";
+import { StreamingTranscriber } from "../../src";
+
+const websocketBaseUrl = "wss://localhost:1234/v3/ws";
+const sessionBeginsMessage = {
+  type: "Begin",
+  id: "123",
+  expires_at: 123456789,
+};
+
+function wait(ms: number) {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+// Counts connections and never answers with `Begin`, so `connect()` stays
+// pending until the attempt times out.
+function countConnections(server: WS, onConnection?: () => void) {
+  const state = { count: 0 };
+  const mockServer = (
+    server as unknown as {
+      server: { on: (event: "connection", cb: () => void) => void };
+    }
+  ).server;
+  mockServer.on("connection", () => {
+    state.count++;
+    onConnection?.();
+  });
+  return state;
+}
+
+describe("streaming close() while connect() is pending", () => {
+  let server: WS;
+
+  beforeEach(() => {
+    WS.clean();
+    server = new WS(websocketBaseUrl);
+  });
+
+  afterEach(() => {
+    WS.clean();
+  });
+
+  it("does not open a new connection after close()", async () => {
+    const connections = countConnections(server);
+    const rt = new StreamingTranscriber({
+      websocketBaseUrl,
+      token: "123",
+      sampleRate: 16_000,
+      connectTimeout: 50,
+      maxConnectionRetries: 2,
+      connectionRetryDelay: 0,
+    });
+
+    const connectPromise = rt.connect();
+    connectPromise.catch(() => {});
+    await server.connected;
+    expect(connections.count).toBe(1);
+
+    await rt.close();
+    // Longer than several connect timeouts: a retry would show up by now.
+    await wait(300);
+
+    expect(connections.count).toBe(1);
+    await expect(connectPromise).rejects.toThrow(/closed/i);
+  });
+
+  it("does not open a new connection when close() runs during the retry delay", async () => {
+    const connections = countConnections(server);
+    const rt = new StreamingTranscriber({
+      websocketBaseUrl,
+      token: "123",
+      sampleRate: 16_000,
+      connectTimeout: 30,
+      maxConnectionRetries: 2,
+      connectionRetryDelay: 200,
+    });
+
+    const connectPromise = rt.connect();
+    connectPromise.catch(() => {});
+    await server.connected;
+    // The first attempt times out at ~30ms and the retry delay begins.
+    await wait(80);
+    await rt.close();
+    await wait(400);
+
+    expect(connections.count).toBe(1);
+    await expect(connectPromise).rejects.toThrow(/closed/i);
+  });
+
+  it("can connect again after close()", async () => {
+    const mockServer = (
+      server as unknown as {
+        server: {
+          on: (
+            event: "connection",
+            cb: (socket: { send: (data: string) => void }) => void,
+          ) => void;
+        };
+      }
+    ).server;
+    let connectionCount = 0;
+    mockServer.on("connection", (socket) => {
+      // Only the second connection gets a `Begin`.
+      if (++connectionCount === 2) {
+        socket.send(JSON.stringify(sessionBeginsMessage));
+      }
+    });
+    const rt = new StreamingTranscriber({
+      websocketBaseUrl,
+      token: "123",
+      sampleRate: 16_000,
+      connectTimeout: 1000,
+    });
+    const first = rt.connect();
+    first.catch(() => {});
+    await server.connected;
+    await rt.close();
+    await expect(first).rejects.toThrow(/closed/i);
+
+    await expect(rt.connect()).resolves.toMatchObject({ type: "Begin" });
+    await rt.close(false);
+  });
+});