fix(review-followup-fec): reconcile epoch rollback and retain reconnect backoff

2 files changed, 89 insertions, 5 deletions

diff --git a/src/adapters/ws/index.test.ts b/src/adapters/ws/index.test.ts
index 9d821e2..6f5e4ef 100644
--- a/src/adapters/ws/index.test.ts
+++ b/src/adapters/ws/index.test.ts
@@ -6,7 +6,7 @@ interface FakeSocket extends WebSocketLike {
   sent: string[];
   resolveOpen(): void;
   invokeMessage(data: string): void;
-  invokeClose(): void;
+  invokeClose(code?: number): void;
 }
 
 function fakeSocket(): FakeSocket {
@@ -44,8 +44,8 @@ function fakeSocket(): FakeSocket {
     invokeMessage(data: string) {
       onmessage?.({ data });
     },
-    invokeClose() {
-      onclose?.({ code: 1000, reason: "" });
+    invokeClose(code = 1000) {
+      onclose?.({ code, reason: "" });
     },
     sent,
   };
@@ -405,3 +405,74 @@ describe("createSurfaceSocket", () => {
     expect(JSON.parse(ws.sent[0] ?? "")).toEqual({ type: "chat.send", message: "queued" });
   });
 });
+
+describe("reconnect backoff after open", () => {
+  it.each([1013, 1006, 1011])("grows and caps delays across repeated early %s closes", (code) => {
+    vi.useFakeTimers();
+    const random = vi.spyOn(Math, "random").mockReturnValue(0.5);
+    const sockets: FakeSocket[] = [];
+    const handle = createSurfaceSocket({
+      url: "ws://test",
+      onMessage: vi.fn(),
+      socketFactory: () => {
+        const ws = fakeSocket();
+        sockets.push(ws);
+        return ws;
+      },
+    });
+    try {
+      for (const delay of [500, 1000, 2000, 4000, 8000, 16000, 30000, 30000]) {
+        const count = sockets.length;
+        sockets.at(-1)?.resolveOpen();
+        vi.advanceTimersByTime(100);
+        sockets.at(-1)?.invokeClose(code);
+        vi.advanceTimersByTime(delay - 1);
+        expect(sockets).toHaveLength(count);
+        vi.advanceTimersByTime(1);
+        expect(sockets).toHaveLength(count + 1);
+      }
+    } finally {
+      handle.close();
+      random.mockRestore();
+      vi.useRealTimers();
+    }
+  });
+
+  it.each([
+    "healthy",
+    "normal",
+    "going away",
+  ])("resets accumulated backoff after a %s connection", (mode) => {
+    vi.useFakeTimers();
+    const random = vi.spyOn(Math, "random").mockReturnValue(0.5);
+    const sockets: FakeSocket[] = [];
+    const handle = createSurfaceSocket({
+      url: "ws://test",
+      onMessage: vi.fn(),
+      socketFactory: () => {
+        const ws = fakeSocket();
+        sockets.push(ws);
+        return ws;
+      },
+    });
+    try {
+      sockets[0]?.resolveOpen();
+      sockets[0]?.invokeClose(1013);
+      vi.advanceTimersByTime(500);
+      sockets[1]?.resolveOpen();
+      if (mode === "healthy") vi.advanceTimersByTime(30000);
+      sockets[1]?.invokeClose(mode === "normal" ? 1000 : mode === "going away" ? 1001 : 1013);
+      vi.advanceTimersByTime(499);
+      expect(sockets).toHaveLength(2);
+      vi.advanceTimersByTime(1);
+      expect(sockets).toHaveLength(3);
+      handle.close();
+      vi.advanceTimersByTime(60000);
+      expect(sockets).toHaveLength(3);
+    } finally {
+      handle.close();
+      random.mockRestore();
+      vi.useRealTimers();
+    }
+  });
+});
diff --git a/src/adapters/ws/index.ts b/src/adapters/ws/index.ts
index 76d9e75..eeb7901 100644
--- a/src/adapters/ws/index.ts
+++ b/src/adapters/ws/index.ts
@@ -67,7 +67,12 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa
   let disposed = false;
   let reconnectAttempt = 0;
   let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
+  let healthyTimer: ReturnType<typeof setTimeout> | null = null;
   let isOpen = false;
+  function clearHealthyTimer(): void {
+    if (healthyTimer !== null) clearTimeout(healthyTimer);
+    healthyTimer = null;
+  }
   const queue: string[] = [];
 
   function connect(isReconnect: boolean): void {
@@ -77,7 +82,12 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa
     socket.onopen = () => {
       if (disposed) return;
       isOpen = true;
-      reconnectAttempt = 0;
+      // Opening is not proof of recovery: replay may immediately overflow the
+      // server's slow-consumer bound. Only a sustained open resets the streak.
+      healthyTimer = setTimeout(() => {
+        healthyTimer = null;
+        reconnectAttempt = 0;
+      }, 30_000);
       for (const raw of queue.splice(0)) {
         socket?.send(raw);
       }
@@ -112,8 +122,10 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa
       }
     };
 
-    socket.onclose = () => {
+    socket.onclose = (ev) => {
+      clearHealthyTimer();
       if (disposed) return;
+      if (isOpen && (ev.code === 1000 || ev.code === 1001)) reconnectAttempt = 0;
       isOpen = false;
       scheduleReconnect();
     };
@@ -143,6 +155,7 @@ export function createSurfaceSocket(opts: SurfaceSocketOptions): SurfaceSocketHa
     },
     close(): void {
       disposed = true;
+      clearHealthyTimer();
       if (reconnectTimer !== null) {
         clearTimeout(reconnectTimer);
         reconnectTimer = null;