diff --git a/.changeset/socket-mode-disconnect-socket-leak.md b/.changeset/socket-mode-disconnect-socket-leak.md new file mode 100644 index 000000000..dfc768344 --- /dev/null +++ b/.changeset/socket-mode-disconnect-socket-leak.md @@ -0,0 +1,9 @@ +--- +"@slack/socket-mode": patch +--- + +Fix leaked TCP sockets on disconnect ([#2709](https://github.com/slackapi/node-slack-sdk/issues/2709)). Moving from `ws` to undici's `WebSocket` in 3.0.0 dropped every force-close mechanism, so `ESTABLISHED` sockets were never torn down and accumulated toward Slack's per-app connection cap. + +- `disconnect()` now arms a 30s close-handshake timeout, so a peer that never replies to the close frame no longer leaves the socket stuck in `CLOSING` indefinitely. +- On the default dispatcher path, the underlying TCP socket is now captured and force-destroyed during cleanup. A user-supplied `dispatcher` owns its own socket and instead relies on the close-handshake timeout — this limitation is documented on the `dispatcher` option. +- The `close` handler no longer schedules a reconnect on stale or overlapping close events, so at most one reconnect is scheduled per disconnect. diff --git a/packages/socket-mode/src/SlackWebSocket.test.ts b/packages/socket-mode/src/SlackWebSocket.test.ts index 5a3563bcb..3ed500919 100644 --- a/packages/socket-mode/src/SlackWebSocket.test.ts +++ b/packages/socket-mode/src/SlackWebSocket.test.ts @@ -93,4 +93,158 @@ describe('SlackWebSocket', () => { sinon.assert.calledOnce(discStub); }); }); + + describe('disconnect() with an unresponsive peer', () => { + // Peer accepts our close frame but never replies with its own: close() moves to CLOSING + // and no 'close' event is ever dispatched. + class DeadPeerWS extends EventTarget { + static CONNECTING = 0; + static OPEN = 1; + static CLOSING = 2; + static CLOSED = 3; + readyState = 1; + closeCalls = 0; + close() { + this.closeCalls += 1; + this.readyState = DeadPeerWS.CLOSING; + } + send(_data: string) {} + } + + it('should force cleanup and emit "close" when the peer never replies with a close frame', () => { + const clock = sandbox.useFakeTimers(); + const ws = new DeadPeerWS(); + SlackWebSocket = proxyquire.load('./SlackWebSocket', { + undici: { + WebSocket: class Fake { + static CONNECTING = 0; + static OPEN = 1; + static CLOSING = 2; + static CLOSED = 3; + constructor() { + // biome-ignore lint/correctness/noConstructorReturn: for test mocking purposes + return ws; + } + }, + CloseEvent, + ErrorEvent, + MessageEvent, + ping: () => {}, + }, + }).SlackWebSocket; + const client = new EventEmitter(); + let closeEmitted = false; + client.on('close', () => { + closeEmitted = true; + }); + const sws = new SlackWebSocket({ + url: 'ws://127.0.0.1/', + client, + clientPingTimeoutMS: 60000, + serverPingTimeoutMS: 60000, + }); + sws.connect(); + + sws.disconnect(); + assert.strictEqual(ws.closeCalls, 1, 'a close frame should have been sent to the peer'); + assert.strictEqual(sws.readyState, DeadPeerWS.CLOSING, 'socket should be CLOSING after sending close frame'); + assert.strictEqual(closeEmitted, false, 'close must not fire before the handshake completes or times out'); + + clock.tick(60000); + + assert.strictEqual( + closeEmitted, + true, + 'expected disconnect() to force cleanup and emit "close" after the peer failed to reply', + ); + }); + }); + + describe('cleanup() with a captured socket', () => { + it('should destroy the captured underlying socket during cleanup', () => { + const sws = new SlackWebSocket({ + url: 'ws://127.0.0.1/', + client: new EventEmitter(), + clientPingTimeoutMS: 1, + serverPingTimeoutMS: 1, + }); + const destroy = sandbox.spy(); + (sws as unknown as { defaultSocket: { destroy: () => void; destroyed: boolean } }).defaultSocket = { + destroy, + destroyed: false, + }; + + sws.disconnect(); + + sinon.assert.calledOnce(destroy); + }); + }); + + describe('buildDefaultDispatcher() connect hook', () => { + type ConnectCb = (err: Error | null, socket: unknown) => void; + type ConnectFn = (opts: unknown, cb: ConnectCb) => void; + + function loadWith(baseConnect: ConnectFn) { + let capturedConnect!: ConnectFn; + const SWS = proxyquire.load('./SlackWebSocket', { + undici: { + WebSocket: WSMock, + CloseEvent, + ErrorEvent, + MessageEvent, + ping: () => {}, + buildConnector: () => baseConnect, + Agent: class { + constructor(o: { connect: ConnectFn }) { + capturedConnect = o.connect; + } + }, + }, + }).SlackWebSocket; + const sws = new SWS({ + url: 'ws://127.0.0.1/', + client: new EventEmitter(), + clientPingTimeoutMS: 1, + serverPingTimeoutMS: 1, + }); + sws.connect(); // no dispatcher provided => builds the default dispatcher => captures the hook + return { sws, invoke: (cb: ConnectCb) => capturedConnect({}, cb) }; + } + + it('should capture the socket and call back with (null, socket) on success', () => { + const fakeSocket = { destroy() {}, destroyed: false }; + const { sws, invoke } = loadWith((_opts, cb) => cb(null, fakeSocket)); + let cbErr: unknown = 'unset'; + let cbSocket: unknown = 'unset'; + invoke((err, socket) => { + cbErr = err; + cbSocket = socket; + }); + assert.strictEqual(cbErr, null); + assert.strictEqual(cbSocket, fakeSocket); + assert.strictEqual((sws as unknown as { defaultSocket: unknown }).defaultSocket, fakeSocket); + }); + + it('should propagate the connector error and not capture a socket', () => { + const boom = new Error('connect failed'); + const { sws, invoke } = loadWith((_opts, cb) => cb(boom, null)); + let received: unknown = 'unset'; + invoke((err) => { + received = err; + }); + assert.strictEqual(received, boom); + assert.strictEqual((sws as unknown as { defaultSocket: unknown }).defaultSocket, null); + }); + + it('should synthesize an error when the connector yields no socket and no error', () => { + const { sws, invoke } = loadWith((_opts, cb) => cb(null, null)); + let received: unknown; + invoke((err) => { + received = err; + }); + assert.ok(received instanceof Error); + assert.match((received as Error).message, /returned no socket/); + assert.strictEqual((sws as unknown as { defaultSocket: unknown }).defaultSocket, null); + }); + }); }); diff --git a/packages/socket-mode/src/SlackWebSocket.ts b/packages/socket-mode/src/SlackWebSocket.ts index 3fcb24a13..4340db39b 100644 --- a/packages/socket-mode/src/SlackWebSocket.ts +++ b/packages/socket-mode/src/SlackWebSocket.ts @@ -1,7 +1,9 @@ import { channel } from 'node:diagnostics_channel'; +import type { Socket } from 'node:net'; +import type { TLSSocket } from 'node:tls'; import type { EventEmitter } from 'eventemitter3'; -import { CloseEvent, type Dispatcher, ErrorEvent, MessageEvent, ping, WebSocket } from 'undici'; +import { Agent, buildConnector, CloseEvent, type Dispatcher, ErrorEvent, MessageEvent, ping, WebSocket } from 'undici'; import { SMWebsocketError } from './errors'; import log, { type Logger, LogLevel } from './logger'; @@ -9,6 +11,8 @@ import type { SocketModeDispatcher } from './SocketModeOptions'; export const WS_READY_STATES = ['CONNECTING', 'OPEN', 'CLOSING', 'CLOSED']; +const CLOSE_HANDSHAKE_TIMEOUT_MS = 30_000; + interface PingPongMessage { websocket: WebSocket; payload: Buffer; @@ -38,7 +42,11 @@ export interface SlackWebSocketOptions { logger?: Logger; /** @description Delay between this client sending a `ping` message, in milliseconds. */ pingInterval?: number; - /** @description An undici Dispatcher used to establish the WebSocket connection (e.g. ProxyAgent). */ + /** + * @description An undici Dispatcher used to establish the WebSocket connection (e.g. ProxyAgent). + * Overrides the default dispatcher, which force-destroys the underlying socket on cleanup; a custom + * dispatcher cannot, so a stalled close handshake falls back to a timeout instead. + */ dispatcher?: SocketModeDispatcher; /** @description Whether this WebSocket should DEBUG log ping and pong events. `false` by default. */ pingPongLoggingEnabled?: boolean; @@ -68,6 +76,8 @@ export class SlackWebSocket { private websocket: WebSocket | null; + private defaultSocket: Socket | TLSSocket | null = null; + /** * The last timetamp that this WebSocket received pong from the server */ @@ -89,6 +99,11 @@ export class SlackWebSocket { */ private clientPingTimeout: NodeJS.Timeout | undefined; + /** + * Reference to the timer that force-closes the connection if the peer never completes the close handshake + */ + private closeHandshakeTimeout: ReturnType | undefined; + private openHandler: (() => void) | null = null; private errorHandler: ((event: Event) => void) | null = null; private messageHandler: ((event: Event) => void) | null = null; @@ -136,7 +151,10 @@ export class SlackWebSocket { public connect(): void { this.logger.debug('Initiating new WebSocket connection.'); - this.websocket = new WebSocket(this.options.url, { dispatcher: this.options.dispatcher as Dispatcher }); + const dispatcher: Dispatcher = this.options.dispatcher + ? (this.options.dispatcher as Dispatcher) + : this.buildDefaultDispatcher(); + this.websocket = new WebSocket(this.options.url, { dispatcher }); this.openHandler = () => { this.logger.debug('WebSocket open event received (connection established)!'); @@ -205,6 +223,27 @@ export class SlackWebSocket { SlackWebSocket.pongChannel.subscribe(this.pongHandler); } + /** + * The `connect` hook captures the underlying socket into `this.defaultSocket` so `cleanup()` can + * force-destroy it: undici's `WebSocket` hides its socket and detaches it from the pool at upgrade, + * leaving no other way to close a stalled peer. + */ + private buildDefaultDispatcher(): Dispatcher { + const baseConnect = buildConnector({}); + return new Agent({ + connect: (opts, callback) => { + baseConnect(opts, (err, socket) => { + if (!socket) { + callback(err ?? new Error('Socket Mode connector returned no socket'), null); + return; + } + this.defaultSocket = socket; + callback(null, socket); + }); + }, + }); + } + /** * Disconnects the WebSocket connection with Slack, if connected. */ @@ -216,8 +255,6 @@ export class SlackWebSocket { this.logger.debug('Terminating WebSocket (close frame received).'); this.cleanup(); } else if (this.websocket.readyState === WebSocket.CLOSING) { - // A close frame was already sent but the peer hasn't responded. Force-terminate rather than - // waiting for the ws library's closeTimeout (~30s) while the ping monitor logs repeated warnings. this.logger.debug('Terminating WebSocket (close frame sent but no response, force-terminating).'); this.cleanup(); } else { @@ -225,6 +262,10 @@ export class SlackWebSocket { // in response. this.logger.debug('Sending close frame (status=1000).'); this.websocket.close(1000); // 1000 = Normal Closure + this.closeHandshakeTimeout = setTimeout(() => { + this.logger.warn('Peer did not complete the close handshake in time; forcing cleanup.'); + this.cleanup(); + }, CLOSE_HANDSHAKE_TIMEOUT_MS); } } else { this.logger.debug('WebSocket already disconnected, flushing remainder.'); @@ -250,9 +291,14 @@ export class SlackWebSocket { if (this.pongHandler) SlackWebSocket.pongChannel.unsubscribe(this.pongHandler); this.pingHandler = null; this.pongHandler = null; + if (this.defaultSocket && !this.defaultSocket.destroyed) { + this.defaultSocket.destroy(); + } + this.defaultSocket = null; this.websocket = null; clearTimeout(this.serverPingTimeout); clearInterval(this.clientPingTimeout); + clearTimeout(this.closeHandshakeTimeout); // Emit event back to client letting it know connection has closed (in case it needs to reconnect if // reconnecting is enabled) this.options.client.emit('close'); diff --git a/packages/socket-mode/src/SocketModeClient.test.ts b/packages/socket-mode/src/SocketModeClient.test.ts index 368644556..e68ff270f 100644 --- a/packages/socket-mode/src/SocketModeClient.test.ts +++ b/packages/socket-mode/src/SocketModeClient.test.ts @@ -5,7 +5,8 @@ import type { FetchFunction } from '@slack/web-api'; import proxyquire from 'proxyquire'; import sinon from 'sinon'; -import logModule from './logger'; +import logModule, { LogLevel } from './logger'; +import type { SlackWebSocket } from './SlackWebSocket'; import { SocketModeClient } from './SocketModeClient'; import type { SocketModeDispatcher } from './SocketModeOptions'; @@ -386,6 +387,32 @@ describe('SocketModeClient', () => { }); }); }); + + describe("reconnection on 'close' (issue #2709)", () => { + it('should not reconnect on a stale close event while the current connection is still active', () => { + const client = new SocketModeClient({ appToken: 'xapp-', logLevel: LogLevel.ERROR }); + client.websocket = { isActive: () => true } as unknown as SlackWebSocket; + const startStub = sandbox.stub(client, 'start').resolves(undefined as never); + const clock = sandbox.useFakeTimers(); + + client.emit('close'); + clock.tick(1_000_000); + + sinon.assert.notCalled(startStub); + }); + + it('should schedule at most one reconnect when multiple close events fire', () => { + const client = new SocketModeClient({ appToken: 'xapp-', logLevel: LogLevel.ERROR }); + const startStub = sandbox.stub(client, 'start').resolves(undefined as never); + const clock = sandbox.useFakeTimers(); + + client.emit('close'); + client.emit('close'); + clock.tick(1_000_000); + + sinon.assert.calledOnce(startStub); + }); + }); }); async function sleep(ms: number): Promise { diff --git a/packages/socket-mode/src/SocketModeClient.ts b/packages/socket-mode/src/SocketModeClient.ts index 9f07c58c4..93bf8c7c0 100644 --- a/packages/socket-mode/src/SocketModeClient.ts +++ b/packages/socket-mode/src/SocketModeClient.ts @@ -157,6 +157,8 @@ export class SocketModeClient extends EventEmitter { }); this.on('close', () => { // Underlying WebSocket connection was closed, possibly reconnect. + if (this.websocket?.isActive()) return; + if (this.reconnectionTimer) return; if (!this.shuttingDown && this.autoReconnectEnabled) { this.delayReconnectAttempt(this.start); } else {