diff --git a/src/main/daemon/client.test.ts b/src/main/daemon/client.test.ts index 81333d8f3..728a0d758 100644 --- a/src/main/daemon/client.test.ts +++ b/src/main/daemon/client.test.ts @@ -335,6 +335,32 @@ describe('DaemonClient', () => { }) describe('disconnect', () => { + it('removes socket listeners when disconnecting', async () => { + await startMockDaemon() + + client = new DaemonClient({ socketPath, tokenPath }) + await client.ensureConnected() + + const connectedClient = client as unknown as { + controlSocket: Socket | null + streamSocket: Socket | null + } + const sockets = [connectedClient.controlSocket, connectedClient.streamSocket] + for (const socket of sockets) { + expect(socket?.listenerCount('data')).toBe(1) + expect(socket?.listenerCount('close')).toBe(1) + expect(socket?.listenerCount('error')).toBe(1) + } + + client.disconnect() + + for (const socket of sockets) { + expect(socket?.listenerCount('data')).toBe(0) + expect(socket?.listenerCount('close')).toBe(0) + expect(socket?.listenerCount('error')).toBe(0) + } + }) + it('emits disconnected when server destroys sockets', async () => { const serverSockets: Socket[] = [] await startMockDaemon() diff --git a/src/main/daemon/client.ts b/src/main/daemon/client.ts index 72cbea13d..757acd182 100644 --- a/src/main/daemon/client.ts +++ b/src/main/daemon/client.ts @@ -1,3 +1,4 @@ +/* eslint-disable max-lines -- Why: daemon handshake, RPC, stream events, and reconnect cleanup share one socket lifecycle. */ import { connect, type Socket } from 'net' import { readFileSync } from 'fs' import { randomUUID } from 'crypto' @@ -45,6 +46,7 @@ export class DaemonClient { private eventListeners: ((event: unknown) => void)[] = [] private disconnectedListeners: (() => void)[] = [] private requestCounter = 0 + private cleanupSocketListeners: (() => void) | null = null constructor(opts: DaemonClientOptions) { this.socketPath = opts.socketPath @@ -74,16 +76,22 @@ export class DaemonClient { private async doConnect(): Promise { const token = readFileSync(this.tokenPath, 'utf-8').trim() + const pendingListenerCleanups: (() => void)[] = [] + const cleanupPendingListeners = (): void => { + for (const cleanup of pendingListenerCleanups.splice(0)) { + cleanup() + } + } try { // Sequential: control first, then stream this.controlSocket = await this.connectSocket() await this.sendHello(this.controlSocket, token, 'control') - this.setupControlParser() + pendingListenerCleanups.push(this.setupControlParser(this.controlSocket)) this.streamSocket = await this.connectSocket() await this.sendHello(this.streamSocket, token, 'stream') - this.setupStreamParser() + pendingListenerCleanups.push(this.setupStreamParser(this.streamSocket)) this.connected = true this.disconnectArmed = true @@ -91,11 +99,21 @@ export class DaemonClient { const gen = this.connectionGeneration const handleClose = () => this.handleDisconnect(gen) - this.controlSocket.on('close', handleClose) - this.controlSocket.on('error', handleClose) - this.streamSocket.on('close', handleClose) - this.streamSocket.on('error', handleClose) + const controlSocket = this.controlSocket + const streamSocket = this.streamSocket + controlSocket.on('close', handleClose) + controlSocket.on('error', handleClose) + streamSocket.on('close', handleClose) + streamSocket.on('error', handleClose) + pendingListenerCleanups.push(() => { + controlSocket.off('close', handleClose) + controlSocket.off('error', handleClose) + streamSocket.off('close', handleClose) + streamSocket.off('error', handleClose) + }) + this.cleanupSocketListeners = cleanupPendingListeners } catch (error) { + cleanupPendingListeners() this.controlSocket?.destroy() this.streamSocket?.destroy() this.controlSocket = null @@ -163,6 +181,7 @@ export class DaemonClient { disconnect(): void { this.connected = false this.disconnectArmed = false + this.cleanupActiveSocketListeners() for (const [id, pending] of this.pendingRequests) { clearTimeout(pending.timer) @@ -275,11 +294,7 @@ export class DaemonClient { }) } - private setupControlParser(): void { - if (!this.controlSocket) { - return - } - + private setupControlParser(socket: Socket): () => void { const parser = createNdjsonParser( (msg) => { const response = msg as RpcResponse @@ -299,14 +314,12 @@ export class DaemonClient { () => {} // Ignore parse errors on control socket ) - this.controlSocket.on('data', (chunk) => parser.feed(chunk.toString())) + const onData = (chunk: Buffer) => parser.feed(chunk.toString()) + socket.on('data', onData) + return () => socket.off('data', onData) } - private setupStreamParser(): void { - if (!this.streamSocket) { - return - } - + private setupStreamParser(socket: Socket): () => void { const parser = createNdjsonParser( (msg) => { const event = msg as DaemonEvent @@ -319,7 +332,9 @@ export class DaemonClient { () => {} // Ignore parse errors on stream socket ) - this.streamSocket.on('data', (chunk) => parser.feed(chunk.toString())) + const onData = (chunk: Buffer) => parser.feed(chunk.toString()) + socket.on('data', onData) + return () => socket.off('data', onData) } private handleDisconnect(generation: number): void { @@ -328,6 +343,7 @@ export class DaemonClient { } this.disconnectArmed = false this.connected = false + this.cleanupActiveSocketListeners() for (const [id, pending] of this.pendingRequests) { clearTimeout(pending.timer) @@ -344,4 +360,10 @@ export class DaemonClient { listener() } } + + private cleanupActiveSocketListeners(): void { + const cleanup = this.cleanupSocketListeners + this.cleanupSocketListeners = null + cleanup?.() + } }