diff --git a/CHANGELOG.md b/CHANGELOG.md index 6106b84..6952372 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/package.json b/package.json index 8266741..944ab0a 100644 --- a/package.json +++ b/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" diff --git a/src/services/realtime/service.ts b/src/services/realtime/service.ts index fff8273..aed3133 100644 --- a/src/services/realtime/service.ts +++ b/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) { diff --git a/src/services/streaming/service.ts b/src/services/streaming/service.ts index 60ff025..1663532 100644 --- a/src/services/streaming/service.ts +++ b/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; diff --git a/tests/unit/realtime-close-during-connect.test.ts b/tests/unit/realtime-close-during-connect.test.ts new file mode 100644 index 0000000..6bd7162 --- /dev/null +++ b/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); + }); +}); diff --git a/tests/unit/streaming-close-during-connect.test.ts b/tests/unit/streaming-close-during-connect.test.ts new file mode 100644 index 0000000..66282a1 --- /dev/null +++ b/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); + }); +});