From 992d46dd1552935a180e0482aee43e74087080ae Mon Sep 17 00:00:00 2001 From: Aqil Aziz Date: Sun, 17 May 2026 10:44:36 +0700 Subject: [PATCH] fix(socket): preserve queued once listeners (#1978) --- .../services/__tests__/socketService.test.ts | 54 +++++++++++++++++++ app/src/services/socketService.ts | 20 +++++-- 2 files changed, 69 insertions(+), 5 deletions(-) diff --git a/app/src/services/__tests__/socketService.test.ts b/app/src/services/__tests__/socketService.test.ts index 0073ea14a..bbc231245 100644 --- a/app/src/services/__tests__/socketService.test.ts +++ b/app/src/services/__tests__/socketService.test.ts @@ -160,6 +160,60 @@ describe('socketService — resolveCoreSocketBaseUrl uses getCoreRpcUrl', () => expect(connectedUrl).toBe('http://custom-core-host:9000'); } }); + + it('preserves queued once listeners when the socket is created later', async () => { + const { io } = await import('socket.io-client'); + const ioMock = vi.mocked(io); + ioMock.mockClear(); + + hoisted.getCoreRpcUrlMock.mockResolvedValue('http://127.0.0.1:7788/rpc'); + + const { socketService } = await import('../socketService'); + socketService.disconnect(); + + const callback = vi.fn(); + socketService.once('queued-once-event', callback); + socketService.connect('mock-jwt-queued-once'); + + await pollUntil(() => expect(ioMock).toHaveBeenCalled()); + + const latestSocket = ioMock.mock.results[ioMock.mock.results.length - 1].value as { + on: ReturnType; + once: ReturnType; + }; + + await pollUntil(() => + expect(latestSocket.once).toHaveBeenCalledWith('queued-once-event', expect.any(Function)) + ); + expect(latestSocket.on).not.toHaveBeenCalledWith('queued-once-event', expect.any(Function)); + }); + + it('preserves queued on listeners when the socket is created later', async () => { + const { io } = await import('socket.io-client'); + const ioMock = vi.mocked(io); + ioMock.mockClear(); + + hoisted.getCoreRpcUrlMock.mockResolvedValue('http://127.0.0.1:7788/rpc'); + + const { socketService } = await import('../socketService'); + socketService.disconnect(); + + const callback = vi.fn(); + socketService.on('queued-on-event', callback); + socketService.connect('mock-jwt-queued-on'); + + await pollUntil(() => expect(ioMock).toHaveBeenCalled()); + + const latestSocket = ioMock.mock.results[ioMock.mock.results.length - 1].value as { + on: ReturnType; + once: ReturnType; + }; + + await pollUntil(() => + expect(latestSocket.on).toHaveBeenCalledWith('queued-on-event', expect.any(Function)) + ); + expect(latestSocket.once).not.toHaveBeenCalledWith('queued-on-event', expect.any(Function)); + }); }); describe('socketService — connectivity dispatch on socket events (lines 164, 212, 230, 237, 240)', () => { diff --git a/app/src/services/socketService.ts b/app/src/services/socketService.ts index 7ac28e6ba..d257cf831 100644 --- a/app/src/services/socketService.ts +++ b/app/src/services/socketService.ts @@ -53,6 +53,12 @@ interface ChannelConnectionUpdatedEvent { capabilities?: string[]; } +interface PendingSocketListener { + event: string; + callback: (...args: unknown[]) => void; + once: boolean; +} + function normalizeChannelConnectionUpdatePayload( value: unknown ): ChannelConnectionUpdatedEvent | null { @@ -115,7 +121,7 @@ class SocketService { private socket: Socket | null = null; private token: string | null = null; private mcpTransport: SocketIOMCPTransportImpl | null = null; - private pendingListeners: Array<{ event: string; callback: (...args: unknown[]) => void }> = []; + private pendingListeners: PendingSocketListener[] = []; // Maps original caller callbacks → wrapped callbacks so off() can locate the // exact function references that were registered with socket.io, scoped by event. private listenerMap = new Map< @@ -185,8 +191,12 @@ class SocketService { // Flush any listeners that were registered before the socket existed. if (this.pendingListeners.length > 0) { socketLog('Flushing pending listeners', { count: this.pendingListeners.length }); - for (const { event, callback } of this.pendingListeners) { - this.socket.on(event, callback); + for (const { event, callback, once } of this.pendingListeners) { + if (once) { + this.socket.once(event, callback); + } else { + this.socket.on(event, callback); + } } this.pendingListeners = []; } @@ -372,7 +382,7 @@ class SocketService { this.socket.on(event, wrappedCallback); } else { socketLog('Socket not ready, queuing listener', { event }); - this.pendingListeners.push({ event, callback: wrappedCallback }); + this.pendingListeners.push({ event, callback: wrappedCallback, once: false }); } } @@ -441,7 +451,7 @@ class SocketService { this.socket.once(event, wrappedCallback); } else { socketLog('Socket not ready, queuing once listener', { event }); - this.pendingListeners.push({ event, callback: wrappedCallback }); + this.pendingListeners.push({ event, callback: wrappedCallback, once: true }); } } }