fix(realtime): preserve reconnect backoff across flaps

HOYALIM · HOYALIM · commit c37203fd812c · 2026-07-08T18:56:12.000-07:00
Signed-off-by: Ho Lim &lt;subhoya@gmail.com&gt;
diff --git a/src/realtime-transcription/websocket-session.test.ts b/src/realtime-transcription/websocket-session.test.ts
@@ -22,6 +22,7 @@ async function createRealtimeServer(params?: {
   closeOnConnection?: boolean;
   initialEvent?: unknown;
   initialText?: string;
+  onConnection?: (client: WebSocket) => void;
   onUpgrade?: (headers: Record<string, string | string[] | undefined>) => void;
   onBinary?: (payload: Buffer) => void;
   onText?: (payload: unknown) => void;
@@ -45,6 +46,7 @@ async function createRealtimeServer(params?: {
       if (params?.initialText) {
         ws.send(params.initialText);
       }
+      params?.onConnection?.(ws);
       ws.on("message", (data, isBinary) => {
         const buffer = Buffer.isBuffer(data)
           ? data
@@ -403,4 +405,46 @@ describe("createRealtimeTranscriptionWebSocketSession", () => {
     expect(closeError).toBeInstanceOf(Error);
     expect(closeError.message).toBe("test realtime transcription connection closed before ready");
   });
+
+  it("stops reconnecting after repeated ready-then-close flaps", async () => {
+    const onError = vi.fn();
+    let openCount = 0;
+    const server = await createRealtimeServer({
+      initialEvent: { type: "session.created" },
+      onConnection: (ws) => {
+        openCount += 1;
+        setTimeout(() => ws.close(1011, "flap"), 1);
+      },
+    });
+    const session = createRealtimeTranscriptionWebSocketSession<{ type?: string }>({
+      providerId: "test",
+      callbacks: { onError },
+      url: server.url,
+      maxReconnectAttempts: 3,
+      reconnectDelayMs: 5,
+      onMessage: (event, transport) => {
+        if (event.type === "session.created") {
+          transport.markReady();
+        }
+      },
+      sendAudio: (audio, transport) => {
+        transport.sendBinary(audio);
+      },
+    });
+
+    await session.connect();
+
+    await vi.waitFor(
+      () => {
+        expect(onError).toHaveBeenCalledWith(
+          expect.objectContaining({
+            message: "test realtime transcription reconnect limit reached",
+          }),
+        );
+      },
+      { timeout: 1000 },
+    );
+    expect(openCount).toBe(4);
+    session.close();
+  });
 });
diff --git a/src/realtime-transcription/websocket-session.ts b/src/realtime-transcription/websocket-session.ts
@@ -50,6 +50,7 @@ const DEFAULT_CLOSE_TIMEOUT_MS = 5_000;
 const DEFAULT_MAX_RECONNECT_ATTEMPTS = 5;
 const DEFAULT_RECONNECT_DELAY_MS = 1000;
 const DEFAULT_MAX_QUEUED_BYTES = 2 * 1024 * 1024;
+const RECONNECT_STABLE_RESET_MS = 30_000;
 
 function rawWsDataToBuffer(data: RawData): Buffer {
   if (Buffer.isBuffer(data)) {
@@ -78,6 +79,7 @@ class WebSocketRealtimeTranscriptionSession<Event> implements RealtimeTranscript
   private queuedBytes = 0;
   private ready = false;
   private reconnectAttempts = 0;
+  private reconnectStableTimer: ReturnType<typeof setTimeout> | undefined;
   private reconnecting = false;
   private suppressReconnect = false;
   private ws: WebSocket | null = null;
@@ -108,6 +110,7 @@ class WebSocketRealtimeTranscriptionSession<Event> implements RealtimeTranscript
     this.closed = false;
     this.suppressReconnect = false;
     this.reconnectAttempts = 0;
+    this.clearReconnectStableReset();
     await this.doConnect();
   }
 
@@ -128,6 +131,7 @@ class WebSocketRealtimeTranscriptionSession<Event> implements RealtimeTranscript
     this.closed = true;
     this.connected = false;
     this.ready = false;
+    this.clearReconnectStableReset();
     this.queuedAudio = [];
     this.queuedBytes = 0;
     if (!this.ws || this.ws.readyState !== WebSocket.OPEN) {
@@ -201,6 +205,7 @@ class WebSocketRealtimeTranscriptionSession<Event> implements RealtimeTranscript
         settled = true;
         clearConnectTimeout();
         this.ready = true;
+        this.scheduleReconnectStableReset();
         this.flushQueuedAudio();
         resolve();
       };
@@ -259,7 +264,6 @@ class WebSocketRealtimeTranscriptionSession<Event> implements RealtimeTranscript
         this.ws.on("open", () => {
           opened = true;
           this.connected = true;
-          this.reconnectAttempts = 0;
           this.captureLocalOpen();
           try {
             this.options.onOpen?.(this.transport);
@@ -300,6 +304,7 @@ class WebSocketRealtimeTranscriptionSession<Event> implements RealtimeTranscript
           this.captureClose(code, reasonBuffer);
           this.connected = false;
           this.ready = false;
+          this.clearReconnectStableReset();
           if (this.closeTimer) {
             clearTimeout(this.closeTimer);
             this.closeTimer = undefined;
@@ -415,6 +420,7 @@ class WebSocketRealtimeTranscriptionSession<Event> implements RealtimeTranscript
       clearTimeout(this.closeTimer);
       this.closeTimer = undefined;
     }
+    this.clearReconnectStableReset();
     this.connected = false;
     this.ready = false;
     if (this.ws) {
@@ -423,6 +429,27 @@ class WebSocketRealtimeTranscriptionSession<Event> implements RealtimeTranscript
     }
   }
 
+  private scheduleReconnectStableReset(): void {
+    this.clearReconnectStableReset();
+    if (this.reconnectAttempts <= 0) {
+      return;
+    }
+    this.reconnectStableTimer = setTimeout(() => {
+      this.reconnectStableTimer = undefined;
+      if (this.connected && this.ready) {
+        this.reconnectAttempts = 0;
+      }
+    }, RECONNECT_STABLE_RESET_MS);
+  }
+
+  private clearReconnectStableReset(): void {
+    if (!this.reconnectStableTimer) {
+      return;
+    }
+    clearTimeout(this.reconnectStableTimer);
+    this.reconnectStableTimer = undefined;
+  }
+
   private emitError(error: unknown): void {
     const normalized = error instanceof Error ? error : new Error(String(error));
     try {
