From 85c7543f490322107307b7f42741f3ace67dbaef Mon Sep 17 00:00:00 2001 From: Johannes Millan Date: Wed, 13 May 2026 19:48:43 +0200 Subject: [PATCH] fix(supersync): dedupe WS connections by clientId to evict stale sockets A reconnecting device kept appending alongside its stale entry instead of replacing it, so userSet would fill to MAX_CONNECTIONS_PER_USER and reject legitimate reconnects with 4008 until the 30s+10s heartbeat cycle caught up. addConnection now evicts any existing entry with the same clientId before adding the new socket; the cap still applies to genuinely distinct clientIds. --- .../services/websocket-connection.service.ts | 39 ++++++++++- .../websocket-connection.service.spec.ts | 67 +++++++++++++++++++ 2 files changed, 104 insertions(+), 2 deletions(-) diff --git a/packages/super-sync-server/src/sync/services/websocket-connection.service.ts b/packages/super-sync-server/src/sync/services/websocket-connection.service.ts index 55c98abf39..828819d66d 100644 --- a/packages/super-sync-server/src/sync/services/websocket-connection.service.ts +++ b/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`, diff --git a/packages/super-sync-server/tests/websocket-connection.service.spec.ts b/packages/super-sync-server/tests/websocket-connection.service.spec.ts index 716fa85224..5db2d341d0 100644 --- a/packages/super-sync-server/tests/websocket-connection.service.spec.ts +++ b/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', () => {