fix(socket-mode): tear down leaked sockets on disconnect (#2709)

.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.

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);
+    });
+  });
 });

packages/socket-mode/src/SlackWebSocket.ts

@@ -1,14 +1,18 @@
 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';
 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<typeof setTimeout> | 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,15 +255,17 @@ 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 {
         // If we haven't received a close frame yet, then we send one to the peer, expecting to receive a close frame
         // 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');

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<void> {

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 {