fix(realtime): preserve reconnect budget across flaps (#102348)

HOYALIM · steipete · web-flow · commit 99d82a2f21ae · 2026-07-09T12:08:50.000+01:00
Reset reconnect attempts only after a provider-ready connection stays stable, and cover both flap exhaustion and stable recovery.

Co-authored-by: Peter Steinberger &lt;steipete@gmail.com&gt;
diff --git a/src/realtime-transcription/websocket-session.test.ts b/src/realtime-transcription/websocket-session.test.ts
@@ -16,6 +16,7 @@ beforeEach(() => {
 });
 
 afterEach(async () => {
+  vi.restoreAllMocks();
   vi.useRealTimers();
   await cleanup?.();
   cleanup = undefined;
@@ -25,6 +26,7 @@ async function createRealtimeServer(params?: {
   closeOnConnection?: boolean;
   initialEvent?: unknown;
   initialText?: string;
+  onConnection?: (ws: WebSocket) => void;
   onUpgrade?: (headers: Record<string, string | string[] | undefined>) => void;
   onBinary?: (payload: Buffer) => void;
   onText?: (payload: unknown) => void;
@@ -48,6 +50,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
@@ -407,6 +410,94 @@ describe("createRealtimeTranscriptionWebSocketSession", () => {
     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();
+  });
+
+  it("refreshes the reconnect budget after a stable ready connection", async () => {
+    let now = 0;
+    vi.spyOn(Date, "now").mockImplementation(() => now);
+    const connections: WebSocket[] = [];
+    const onError = vi.fn();
+    const server = await createRealtimeServer({
+      initialEvent: { type: "session.created" },
+      onConnection: (ws) => connections.push(ws),
+    });
+    const session = createRealtimeTranscriptionWebSocketSession<{ type?: string }>({
+      providerId: "test",
+      callbacks: { onError },
+      url: server.url,
+      maxReconnectAttempts: 1,
+      reconnectDelayMs: 5,
+      onMessage: (event, transport) => {
+        if (event.type === "session.created") {
+          transport.markReady();
+        }
+      },
+      sendAudio: (audio, transport) => {
+        transport.sendBinary(audio);
+      },
+    });
+
+    await session.connect();
+    connections[0]?.close(1011, "first flap");
+    await vi.waitFor(() => expect(connections).toHaveLength(2));
+    await vi.waitFor(() => expect(session.isConnected()).toBe(true));
+
+    now = 30_000;
+    connections[1]?.close(1011, "stable disconnect");
+    await vi.waitFor(() => expect(connections).toHaveLength(3));
+    await vi.waitFor(() => expect(session.isConnected()).toBe(true));
+
+    connections[2]?.close(1011, "second flap");
+    await vi.waitFor(() => {
+      expect(onError).toHaveBeenCalledWith(
+        expect.objectContaining({
+          message: "test realtime transcription reconnect limit reached",
+        }),
+      );
+    });
+    expect(connections).toHaveLength(3);
+    session.close();
+  });
+
   it("delivers a legitimate large inbound message below the payload cap", async () => {
     // Well above any real transcript message yet far under the 16 MiB cap: proves
     // the bound does not reject legitimate large provider traffic.
diff --git a/src/realtime-transcription/websocket-session.ts b/src/realtime-transcription/websocket-session.ts
@@ -50,6 +50,9 @@ 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;
+// A raw WebSocket open is not recovery. Only a provider-ready connection
+// that survives this window earns a fresh retry budget.
+const RECONNECT_STABLE_RESET_MS = 30_000;
 // Bound inbound messages before ws buffers them for JSON parsing. The 16 MiB cap
 // matches realtime voice; ws rejects larger messages with close 1009 before
 // they reach onMessage, replacing its 100 MiB client default.
@@ -81,6 +84,7 @@ class WebSocketRealtimeTranscriptionSession<Event> implements RealtimeTranscript
   private queuedAudio: Buffer[] = [];
   private queuedBytes = 0;
   private ready = false;
+  private readySinceMs: number | undefined;
   private reconnectAttempts = 0;
   private reconnecting = false;
   private suppressReconnect = false;
@@ -111,6 +115,7 @@ class WebSocketRealtimeTranscriptionSession<Event> implements RealtimeTranscript
   async connect(): Promise<void> {
     this.closed = false;
     this.suppressReconnect = false;
+    this.readySinceMs = undefined;
     this.reconnectAttempts = 0;
     await this.doConnect();
   }
@@ -132,6 +137,7 @@ class WebSocketRealtimeTranscriptionSession<Event> implements RealtimeTranscript
     this.closed = true;
     this.connected = false;
     this.ready = false;
+    this.readySinceMs = undefined;
     this.queuedAudio = [];
     this.queuedBytes = 0;
     if (!this.ws || this.ws.readyState !== WebSocket.OPEN) {
@@ -205,6 +211,7 @@ class WebSocketRealtimeTranscriptionSession<Event> implements RealtimeTranscript
         settled = true;
         clearConnectTimeout();
         this.ready = true;
+        this.readySinceMs = Date.now();
         this.flushQueuedAudio();
         resolve();
       };
@@ -264,7 +271,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);
@@ -303,8 +309,13 @@ class WebSocketRealtimeTranscriptionSession<Event> implements RealtimeTranscript
         this.ws.on("close", (code, reasonBuffer) => {
           clearConnectTimeout();
           this.captureClose(code, reasonBuffer);
+          const readyForMs = this.readySinceMs === undefined ? 0 : Date.now() - this.readySinceMs;
           this.connected = false;
           this.ready = false;
+          this.readySinceMs = undefined;
+          if (readyForMs >= RECONNECT_STABLE_RESET_MS) {
+            this.reconnectAttempts = 0;
+          }
           if (this.closeTimer) {
             clearTimeout(this.closeTimer);
             this.closeTimer = undefined;
@@ -422,6 +433,7 @@ class WebSocketRealtimeTranscriptionSession<Event> implements RealtimeTranscript
     }
     this.connected = false;
     this.ready = false;
+    this.readySinceMs = undefined;
     if (this.ws) {
       this.ws.close(1000, "Transcription session closed");
       this.ws = null;
