Merge branch 'feat/production-sync-server-is-full-of-these-b7ab49'

* feat/production-sync-server-is-full-of-these-b7ab49:
  fix(supersync): dedupe WS connections by clientId to evict stale sockets

packages/super-sync-server/src/sync/services/websocket-connection.service.ts

@@ -27,6 +27,8 @@ export class WebSocketConnectionService {
   private static readonly NOTIFY_DEBOUNCE_MS = 100;
   /** Max WebSocket connections per user to prevent resource exhaustion */
   private static readonly MAX_CONNECTIONS_PER_USER = 10;
+  /** Close code sent to the stale socket when a new one from the same clientId replaces it */
+  private static readonly REPLACED_CLOSE_CODE = 4009;
 
   private pendingNotifications = new Map<
     number,
@@ -38,10 +40,35 @@ export class WebSocketConnectionService {
   >();
 
   addConnection(userId: number, clientId: string, ws: WebSocket): void {
+    // A new socket from the same clientId means the device is reconnecting — the
+    // server's old entry is by definition stale (network blip, proxy idle close,
+    // OS sleep). Evict it eagerly instead of waiting up to ~40s for the heartbeat
+    // cycle; otherwise stale entries pile up to MAX_CONNECTIONS_PER_USER and
+    // legitimate reconnects get rejected with 4008.
+    const existingSet = this.connections.get(userId);
+    if (existingSet) {
+      for (const existing of existingSet) {
+        if (existing.clientId === clientId) {
+          Logger.info(
+            `[ws:user:${userId}:${clientId}] Replacing stale connection from same client`,
+          );
+          // `removeConnection` may drop the userId entry from `this.connections`
+          // entirely if the set becomes empty; the `!this.connections.has(userId)`
+          // check below re-creates it before we add the new client.
+          this.removeConnection(userId, existing, {
+            code: WebSocketConnectionService.REPLACED_CLOSE_CODE,
+            reason: 'Replaced by newer connection',
+          });
+          break;
+        }
+      }
+    }
+
     if (!this.connections.has(userId)) {
       this.connections.set(userId, new Set());
     }
     const userSet = this.connections.get(userId)!;
+
     if (userSet.size >= WebSocketConnectionService.MAX_CONNECTIONS_PER_USER) {
       Logger.warn(
         `[ws:user:${userId}] Connection rejected: max connections per user reached`,
@@ -94,7 +121,11 @@ export class WebSocketConnectionService {
     );
   }
 
-  removeConnection(userId: number, client: ConnectedClient): void {
+  removeConnection(
+    userId: number,
+    client: ConnectedClient,
+    closeFrame?: { code: number; reason: string },
+  ): void {
     const userSet = this.connections.get(userId);
     if (userSet) {
       userSet.delete(client);
@@ -108,7 +139,11 @@ export class WebSocketConnectionService {
       client.ws.readyState === WebSocket.CONNECTING
     ) {
       try {
-        client.ws.close();
+        if (closeFrame) {
+          client.ws.close(closeFrame.code, closeFrame.reason);
+        } else {
+          client.ws.close();
+        }
       } catch (err) {
         Logger.debug(
           `[ws:user:${userId}:${client.clientId}] Error closing connection`,

packages/super-sync-server/tests/websocket-connection.service.spec.ts

@@ -109,6 +109,73 @@ describe('WebSocketConnectionService', () => {
 
       expect(service.getConnectionCount()).toBe(0);
     });
+
+    it('should replace a stale connection from the same clientId', () => {
+      const stale = createMockWs();
+      service.addConnection(1, 'client-a', stale as any);
+      expect(service.getConnectionCount()).toBe(1);
+
+      const fresh = createMockWs();
+      service.addConnection(1, 'client-a', fresh as any);
+
+      // Stale socket was evicted with the "replaced" close code.
+      expect(stale.close).toHaveBeenCalledWith(4009, 'Replaced by newer connection');
+      // Only the fresh entry remains.
+      expect(service.getConnectionCount()).toBe(1);
+      // The new connection was accepted, not rejected with 4008.
+      expect(fresh.close).not.toHaveBeenCalled();
+
+      // Real `ws` fires the 'close' event asynchronously after `ws.close()`.
+      // The stale socket's late close event must not disturb the fresh entry.
+      stale._emitClose();
+      expect(service.getConnectionCount()).toBe(1);
+      expect(fresh.close).not.toHaveBeenCalled();
+    });
+
+    it('should not call close() on the evicted socket when it is already CLOSED', () => {
+      const stale = createMockWs();
+      service.addConnection(1, 'client-a', stale as any);
+
+      // Simulate the stale socket already being closed at the OS level
+      // (e.g. removeConnection's gate must skip ws.close to avoid double-close).
+      stale.readyState = WS_CLOSED;
+      const closeCallsBefore = stale.close.mock.calls.length;
+
+      const fresh = createMockWs();
+      service.addConnection(1, 'client-a', fresh as any);
+
+      // Dedup took effect (only the fresh entry remains) but ws.close was not
+      // re-invoked on the already-closed stale socket.
+      expect(service.getConnectionCount()).toBe(1);
+      expect(stale.close.mock.calls.length).toBe(closeCallsBefore);
+    });
+
+    it('should not exceed the per-user cap when same clientId reconnects repeatedly', () => {
+      // 9 unique clientIds use 9 slots; clientId 'A' reconnects 5 times in the
+      // remaining slot. Without dedup, attempts 2-5 would be rejected with 4008.
+      for (let i = 0; i < 9; i++) {
+        service.addConnection(1, `unique-${i}`, createMockWs() as any);
+      }
+      for (let i = 0; i < 5; i++) {
+        const ws = createMockWs();
+        service.addConnection(1, 'A', ws as any);
+        // The latest 'A' socket is the survivor — it was not rejected/closed.
+        expect(ws.close).not.toHaveBeenCalled();
+      }
+      expect(service.getConnectionCount()).toBe(10);
+    });
+
+    it('should still enforce the cap across distinct clientIds', () => {
+      for (let i = 0; i < 10; i++) {
+        service.addConnection(1, `client-${i}`, createMockWs() as any);
+      }
+
+      const eleventh = createMockWs();
+      service.addConnection(1, 'client-10', eleventh as any);
+
+      expect(eleventh.close).toHaveBeenCalledWith(4008, 'Too many connections');
+      expect(service.getConnectionCount()).toBe(10);
+    });
   });
 
   describe('removeConnection', () => {