diff --git a/.changeset/socket-mode-refresh-overlap.md b/.changeset/socket-mode-refresh-overlap.md new file mode 100644 index 000000000..565a2a2eb --- /dev/null +++ b/.changeset/socket-mode-refresh-overlap.md @@ -0,0 +1,7 @@ +--- +"@slack/socket-mode": patch +--- + +Fix a 5 to 12 second gap without a connection each time Slack refreshes a Socket Mode connection ([#2767](https://github.com/slackapi/node-slack-sdk/issues/2767)). On a `disconnect` message the client now opens the replacement connection right away, without a reconnect backoff, and leaves the old connection open until Slack sends it a `disconnect` message too, so payloads Slack still sends on it during the switch are received instead of dropped. Each payload is acknowledged on the connection it arrived on. While the replacement is being opened, the current connection keeps serving, and overlapping `disconnect` messages open one replacement. A replacement that fails for good rejects with its cause, as a failed reconnect does. A `link_disabled` message still closes the connection first, since a replacement can't connect while Socket Mode is off. + +**Note:** during each refresh, a client now holds two connections for about 60 seconds, until Slack sends the old one its own `disconnect` (`refresh_requested`, about a minute after the `warning`). Both count toward Slack's limit of 10 open Socket Mode connections per app, so apps that run many instances at once should leave room for one extra connection per instance while they refresh. diff --git a/packages/socket-mode/README.md b/packages/socket-mode/README.md index 6463d578b..42bc71073 100644 --- a/packages/socket-mode/README.md +++ b/packages/socket-mode/README.md @@ -127,13 +127,13 @@ The client's connection to Slack has a lifecycle. This means the client can be s In the table below, the client's states are listed, which are also the names of the events you can use to observe the transition to that state. The table also includes descriptions for the states and arguments that a listener would receive. -| Event Name | Arguments | Description | -| --------------- | --------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `connecting` | | The client is in the process of connecting to the platform. | -| `connected` | | The client is connected to the platform and incoming events will start being emitted. | -| `disconnecting` | | The client is no longer connected to the platform and cleaning up its resources. It will soon transition to `disconnected`. | -| `reconnecting` | | The client is no longer connected to the platform and cleaning up its resources. It will soon transition to `connecting`. | -| `disconnected` | `(error)` | The client is not connected to the platform. This is a steady state - no attempt to connect is occurring. The `error` argument will be `undefined` when the client initiated the disconnect (normal). | +| Event Name | Arguments | Description | +| --------------- | --------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `connecting` | | The client is in the process of connecting to the platform. | +| `connected` | | The client is connected to the platform and incoming events will start being emitted. | +| `disconnecting` | | The client is no longer connected to the platform and cleaning up its resources. It will soon transition to `disconnected`. | +| `reconnecting` | | The client is replacing its connection. On a Slack refresh, the current connection keeps serving after the new one is `connected`, until Slack sends it `disconnect`. It will soon transition to `connecting`. | +| `disconnected` | `(error)` | The client is not connected to the platform. This is a steady state - no attempt to connect is occurring. The `error` argument will be `undefined` when the client initiated the disconnect (normal). | The client also emits events that are part of its lifecycle, but aren't states. Instead, they represent specific moments that might be helpful to your app. The following table lists these events, their description, and includes the arguments that a listener would receive. diff --git a/packages/socket-mode/src/SlackWebSocket.test.ts b/packages/socket-mode/src/SlackWebSocket.test.ts index fb620fd62..ceae6a8c9 100644 --- a/packages/socket-mode/src/SlackWebSocket.test.ts +++ b/packages/socket-mode/src/SlackWebSocket.test.ts @@ -93,6 +93,36 @@ describe('SlackWebSocket', () => { ws.dispatchEvent(new ErrorEvent('error', { error: new Error('boom'), message: 'boom' })); sinon.assert.calledOnce(discStub); }); + + it('should emit "ws_message" with the receiving SlackWebSocket, so the client can tell connections apart', () => { + const ws = new WSMock(); + SlackWebSocket = proxyquire.load('./SlackWebSocket', { + undici: { + WebSocket: class Fake { + constructor() { + // biome-ignore lint/correctness/noConstructorReturn: for test mocking purposes + return ws; + } + }, + CloseEvent, + ErrorEvent, + MessageEvent, + ping: () => {}, + }, + }).SlackWebSocket; + const client = new EventEmitter(); + const received = sandbox.spy(); + client.on('ws_message', received); + const sws = new SlackWebSocket({ + url: 'whatevs', + client, + clientPingTimeoutMS: 1, + serverPingTimeoutMS: 1, + }); + sws.connect(); + ws.dispatchEvent(new MessageEvent('message', { data: '{"type":"hello"}' })); + sinon.assert.calledOnceWithExactly(received, '{"type":"hello"}', false, sws); + }); }); describe('disconnect() with an unresponsive peer', () => { diff --git a/packages/socket-mode/src/SlackWebSocket.ts b/packages/socket-mode/src/SlackWebSocket.ts index 0544f87be..8e40452b2 100644 --- a/packages/socket-mode/src/SlackWebSocket.ts +++ b/packages/socket-mode/src/SlackWebSocket.ts @@ -179,7 +179,7 @@ export class SlackWebSocket { return; } const isBinary = typeof event.data !== 'string'; - this.options.client.emit('ws_message', event.data, isBinary); + this.options.client.emit('ws_message', event.data, isBinary, this); }; this.websocket.addEventListener('message', this.messageHandler); diff --git a/packages/socket-mode/src/SocketModeClient.test.ts b/packages/socket-mode/src/SocketModeClient.test.ts index b1a2d4b78..8588346d8 100644 --- a/packages/socket-mode/src/SocketModeClient.test.ts +++ b/packages/socket-mode/src/SocketModeClient.test.ts @@ -311,7 +311,7 @@ describe('SocketModeClient', () => { assert.strictEqual(args.type, 'events_api'); assert.deepEqual(args.body, body); await args.ack({}); - sinon.assert.calledOnceWithExactly(send, 'malformed', {}); + sinon.assert.calledOnceWithExactly(send, 'malformed', {}, undefined); } }); @@ -789,6 +789,293 @@ describe('SocketModeClient', () => { await assert.rejects(starting, { name: 'SMConnectAbortedError' }); }); }); + + describe('server-requested disconnect (issue #2767)', () => { + const refreshRequested = { type: 'disconnect', reason: 'refresh_requested' }; + const envelope = { type: 'interactive', envelope_id: 'e-1', payload: {} }; + + it('should open the replacement right away and leave the old connection open after it says hello', async () => { + const { client, error } = createClient(); + const old = await connect(client); + const disconnected = sandbox.spy(); + client.on('disconnected', disconnected); + + old.receive(refreshRequested); + await sleep(0); + + assert.strictEqual(sockets.length, 2, 'the replacement should open without waiting for a backoff'); + assert.strictEqual(old.disconnectCalls, 0); + + sockets[1].open().hello(); + await sleep(10); + + assert.strictEqual(old.disconnectCalls, 0, 'Slack may still deliver on the old connection'); + assert.strictEqual(client.websocket, sockets[1]); + old.drop(); + await sleep(10); + assert.strictEqual(sockets.length, 2, "the old connection's close must not trigger another reconnect"); + sinon.assert.notCalled(disconnected); + sinon.assert.notCalled(error); + }); + + it('should acknowledge each envelope on the connection it arrived on', async () => { + const { client } = createClient(); + const old = await connect(client); + + old.receive(refreshRequested); + await sleep(0); + old.receive(envelope); + sockets[1].open(); + old.receive(envelope); + await sleep(0); + + assert.strictEqual(old.sent.length, 2); + assert.strictEqual(sockets[1].sent.length, 0); + + sockets[1].hello(); + sockets[1].receive(envelope); + old.receive(envelope); + await sleep(0); + + assert.strictEqual(sockets[1].sent.length, 1); + assert.strictEqual(old.sent.length, 3); + }); + + it('should acknowledge on the current connection when the arrival connection has closed', async () => { + const { client } = createClient(); + const old = await connect(client); + const acks: Array<() => Promise> = []; + client.on('slash_commands', ({ ack }) => acks.push(ack)); + + old.receive(refreshRequested); + await sleep(0); + sockets[1].open().hello(); + old.receive({ type: 'slash_commands', envelope_id: 'e-2', payload: {} }); + old.drop(); + await acks[0](); + + assert.strictEqual(old.sent.length, 0); + assert.strictEqual(sockets[1].sent.length, 1); + }); + + it('should close the old connection at the next refresh or on disconnect()', async () => { + const { client } = createClient(); + const old = await connect(client); + const disconnected = sandbox.spy(); + client.on('disconnected', disconnected); + + old.receive(refreshRequested); + await sleep(0); + sockets[1].open().hello(); + assert.strictEqual(old.disconnectCalls, 0); + sockets[1].receive(refreshRequested); + await sleep(0); + + assert.strictEqual(old.disconnectCalls, 1); + assert.strictEqual(sockets[1].disconnectCalls, 0); + + await client.disconnect(); + + assert.strictEqual(sockets[1].disconnectCalls, 1); + assert.strictEqual(sockets[2].disconnectCalls, 1); + sinon.assert.calledOnce(disconnected); + }); + + it('should not close an old connection again after Slack closed it', async () => { + const { client } = createClient(); + const old = await connect(client); + + old.receive(refreshRequested); + await sleep(0); + sockets[1].open().hello(); + old.drop(); + await client.disconnect(); + + assert.strictEqual(old.disconnectCalls, 0); + }); + + it('should back off and keep the old connection serving when the replacement fails before hello', async () => { + const { client } = createClient({ clientPingTimeout: 1 }); + const old = await connect(client); + + old.receive(refreshRequested); + await sleep(0); + sockets[1].drop(); + await sleep(10); + + assert.strictEqual(sockets.length, 3, 'the failed replacement should be retried with the usual backoff'); + assert.strictEqual(old.disconnectCalls, 0); + old.receive(envelope); + await sleep(0); + assert.strictEqual(old.sent.length, 1); + + sockets[2].open().hello(); + await sleep(0); + + assert.strictEqual(old.disconnectCalls, 0); + assert.strictEqual(client.websocket, sockets[2]); + }); + + it('should open only the replacement when the old connection closes while its URL is fetched', async () => { + const { client, holdNextURL } = createClient({ clientPingTimeout: 1 }); + const old = await connect(client); + const releaseURL = holdNextURL(); + const events = { connected: sandbox.spy(), reconnecting: sandbox.spy() }; + for (const [name, spy] of Object.entries(events)) client.on(name, spy); + + old.receive(refreshRequested); + await sleep(0); + old.drop(); + await sleep(10); + releaseURL(); + await sleep(0); + sockets[1].open().hello(); + await sleep(10); + + assert.deepStrictEqual( + sockets.map((socket) => socket.isActive()), + [false, true], + ); + assert.strictEqual(client.websocket, sockets[1]); + sinon.assert.calledOnce(events.connected); + sinon.assert.calledOnce(events.reconnecting); + }); + + it('should open one replacement when disconnect messages arrive back to back', async () => { + const { client } = createClient(); + const old = await connect(client); + const reconnecting = sandbox.spy(); + client.on('reconnecting', reconnecting); + + old.receive({ type: 'disconnect', reason: 'warning' }); + old.receive(refreshRequested); + await sleep(0); + sockets[1].open().hello(); + await sleep(10); + + assert.strictEqual(sockets.length, 2); + assert.strictEqual(client.websocket, sockets[1]); + sinon.assert.calledOnce(reconnecting); + }); + + it('should close the retiring connection, not open another, on its disconnect message while the replacement handshakes', async () => { + const { client } = createClient(); + const old = await connect(client); + + old.receive({ type: 'disconnect', reason: 'warning' }); + await sleep(0); + const replacement = sockets[1].open(); + old.receive(refreshRequested); + await sleep(0); + + assert.strictEqual(sockets.length, 2); + assert.strictEqual(old.disconnectCalls, 1); + + replacement.hello(); + replacement.receive(envelope); + await sleep(0); + + assert.strictEqual(replacement.disconnectCalls, 0); + assert.strictEqual(replacement.sent.length, 1); + assert.strictEqual(client.websocket, replacement); + }); + + it('should reject with the cause when the replacement fails for good', async () => { + const { client, failNextURL } = createClient(); + const old = await connect(client); + const cause = new Error('invalid_auth'); + failNextURL(cause); + const [onMessage] = client.listeners('ws_message'); + + await assert.rejects(async () => onMessage(JSON.stringify(refreshRequested), false, old), cause); + }); + + it('should close a replacement that never opened when start() replaces it', async () => { + const { client } = createClient(); + const old = await connect(client); + + old.receive(refreshRequested); + await sleep(0); + const restarted = client.start(); + await sleep(0); + sockets[2].open().hello(); + await restarted; + + assert.strictEqual(sockets[1].disconnectCalls, 1); + assert.strictEqual(old.disconnectCalls, 0); + assert.strictEqual(client.websocket, sockets[2]); + }); + + it('should close both connections and emit disconnected once when disconnect() runs mid-refresh', async () => { + const { client, error } = createClient(); + const old = await connect(client); + const disconnected = sandbox.spy(); + client.on('disconnected', disconnected); + + old.receive(refreshRequested); + await sleep(0); + await client.disconnect(); + await sleep(10); + + assert.strictEqual(old.disconnectCalls, 1); + assert.strictEqual(sockets[1].disconnectCalls, 1); + sinon.assert.calledOnce(disconnected); + sinon.assert.notCalled(error); + }); + + it('should close without reconnecting when auto-reconnect is disabled', async () => { + const { client } = createClient({ autoReconnectEnabled: false }); + const old = await connect(client); + const disconnected = sandbox.spy(); + client.on('disconnected', disconnected); + + old.receive(refreshRequested); + await sleep(10); + + assert.strictEqual(old.disconnectCalls, 1); + assert.strictEqual(sockets.length, 1); + sinon.assert.calledOnce(disconnected); + }); + + it('should close the connection instead of opening a replacement when the link is disabled', async () => { + const { client } = createClient(); + const old = await connect(client); + + old.receive({ type: 'disconnect', reason: 'link_disabled' }); + await sleep(0); + + assert.strictEqual(sockets.length, 1); + assert.strictEqual(old.disconnectCalls, 1); + await client.disconnect(); + }); + + it('should ignore a disconnect message that arrives after disconnect()', async () => { + const { client } = createClient(); + const old = await connect(client); + + await client.disconnect(); + old.receive(refreshRequested); + await sleep(10); + + assert.strictEqual(sockets.length, 1); + }); + + it('should ignore a late disconnect message from a connection a later refresh closed', async () => { + const { client } = createClient(); + const old = await connect(client); + + old.receive(refreshRequested); + await sleep(0); + sockets[1].open().hello(); + sockets[1].receive(refreshRequested); + await sleep(0); + sockets[2].open().hello(); + old.receive(refreshRequested); + await sleep(10); + + assert.strictEqual(sockets.length, 3); + }); + }); }); async function sleep(ms: number): Promise { diff --git a/packages/socket-mode/src/SocketModeClient.ts b/packages/socket-mode/src/SocketModeClient.ts index 46de335ee..d28eb53c7 100644 --- a/packages/socket-mode/src/SocketModeClient.ts +++ b/packages/socket-mode/src/SocketModeClient.ts @@ -77,6 +77,8 @@ export class SocketModeClient extends EventEmitter { public websocket?: SlackWebSocket; + private retiringWebsocket?: SlackWebSocket; + /** * Enables ping-pong detailed logging if true */ @@ -164,6 +166,7 @@ export class SocketModeClient extends EventEmitter { }); this.on('close', async (socket: SlackWebSocket) => { // Underlying WebSocket connection was closed, possibly reconnect. + if (socket === this.retiringWebsocket) this.retiringWebsocket = undefined; if (socket !== this.websocket) return; if (this.shuttingDown || !this.autoReconnectEnabled) { // If reconnect is disabled or user explicitly called `disconnect()`, emit a disconnected state. @@ -202,6 +205,7 @@ export class SocketModeClient extends EventEmitter { if (signal.aborted) { throw new SMConnectAbortedError(); } + const previous = this.websocket; // create a socket connection using SlackWebSocket this.websocket = new SlackWebSocket({ url, @@ -215,6 +219,7 @@ export class SocketModeClient extends EventEmitter { serverPingTimeoutMS: this.serverPingTimeoutMS, pingPongLoggingEnabled: this.pingPongLoggingEnabled, }); + this.retire(previous); // Return a promise that resolves with the connection information return new Promise((resolve, reject) => { @@ -243,6 +248,7 @@ export class SocketModeClient extends EventEmitter { */ public disconnect(): Promise { this.lifecycle.abort(); + this.closeRetiringWebsocket(); this.logger.debug('Manually disconnecting this Socket Mode client'); this.emit(State.Disconnecting); return new Promise((resolve, _reject) => { @@ -259,6 +265,33 @@ export class SocketModeClient extends EventEmitter { // PRIVATE/PROTECTED METHODS + private retire(previous: SlackWebSocket | undefined): void { + if (previous?.isActive()) { + this.closeRetiringWebsocket(); + this.retiringWebsocket = previous; + } else if (previous?.readyState !== undefined) { + previous.disconnect(); + } + } + + private closeRetiringWebsocket(): void { + this.retiringWebsocket?.disconnect(); + this.retiringWebsocket = undefined; + } + + private async openReplacement(reason: string): Promise { + if (this.connectOperations.size > 0) return; + this.logger.debug(`Received "disconnect" (${reason}) message - opening a replacement connection.`); + this.emit(State.Reconnecting); + try { + await this.start(); + } catch (error) { + if (!(error instanceof SMConnectAbortedError)) { + throw error; + } + } + } + private async withConnectOperation(work: Promise): Promise { this.connectOperations.add(work); try { @@ -340,7 +373,11 @@ export class SocketModeClient extends EventEmitter { * - raising the State.Connected event (when Slack sends a type:hello message) * - disconnecting the underlying socket (when Slack sends a type:disconnect message) */ - protected async onWebSocketMessage(data: string | ArrayBuffer, isBinary: boolean): Promise { + protected async onWebSocketMessage( + data: string | ArrayBuffer, + isBinary: boolean, + socket?: SlackWebSocket, + ): Promise { if (isBinary) { this.logger.debug('Unexpected binary message received, ignoring.'); return; @@ -377,10 +414,17 @@ export class SocketModeClient extends EventEmitter { // Slack is recycling the pod handling the connection (or otherwise requires the client to reconnect) if (event.type === 'disconnect') { - this.logger.debug( - `Received "${event.type}" (${event.reason}) message - disconnecting.${this.autoReconnectEnabled ? ' Will reconnect.' : ''}`, - ); - this.websocket?.disconnect(); + if (socket && socket === this.retiringWebsocket) { + this.closeRetiringWebsocket(); + return; + } + if (socket && socket !== this.websocket) return; + if (this.autoReconnectEnabled && !this.shuttingDown && event.reason !== 'link_disabled') { + await this.openReplacement(event.reason); + } else { + this.logger.debug(`Received "${event.type}" (${event.reason}) message - disconnecting.`); + this.websocket?.disconnect(); + } return; } @@ -391,7 +435,7 @@ export class SocketModeClient extends EventEmitter { `Calling ack() - type: ${event.type}, envelope_id: ${event.envelope_id}, data: ${JSON.stringify(response)}`, ); } - await this.send(event.envelope_id, response); + await this.send(event.envelope_id, response, socket); }; // For events_api messages, expose the type of the event @@ -441,19 +485,20 @@ export class SocketModeClient extends EventEmitter { * @param id the envelope id * @param body the message body or string text */ - private send(id: string, body = {}): Promise { + private send(id: string, body = {}, socket?: SlackWebSocket): Promise { const _body = typeof body === 'string' ? { text: body } : body; const message = { envelope_id: id, payload: { ..._body } }; return new Promise((resolve, reject) => { - const wsState = this.websocket?.readyState; + const websocket = socket?.isActive() ? socket : this.websocket; + const wsState = websocket?.readyState; this.logger.debug( `send() method was called (WebSocket state: ${wsState ? WS_READY_STATES[wsState] : 'uninitialized'})`, ); - if (this.websocket === undefined) { + if (websocket === undefined) { this.logger.error('Failed to send a message as the client is not connected'); reject(new SMSendWhileDisconnectedError()); - } else if (!this.websocket.isActive()) { + } else if (!websocket.isActive()) { this.logger.error('Failed to send a message as the client has no active connection'); reject(new SMSendWhileNotReadyError()); } else { @@ -461,7 +506,7 @@ export class SocketModeClient extends EventEmitter { const flatMessage = JSON.stringify(message); this.logger.debug(`Sending a WebSocket message: ${flatMessage}`); - this.websocket.send(flatMessage, (error) => { + websocket.send(flatMessage, (error) => { if (error) { this.logger.error(`Failed to send a WebSocket message (error: ${error})`); return reject(new SMWebsocketError(error)); diff --git a/packages/socket-mode/test/integrations/integration.test.js b/packages/socket-mode/test/integrations/integration.test.js index e80f6cc6f..2f490977d 100644 --- a/packages/socket-mode/test/integrations/integration.test.js +++ b/packages/socket-mode/test/integrations/integration.test.js @@ -119,6 +119,89 @@ describe('Integration tests with a WebSocket server', { timeout: 30000 }, () => await client.disconnect(); }); }); + describe('overlapping refreshes (issue #2767)', () => { + beforeEach(() => { + client = new SocketModeClient({ + appToken: 'whatever', + logLevel: LogLevel.ERROR, + clientOptions: { + slackApiUrl: `http://localhost:${HTTP_PORT}/`, + }, + clientPingTimeout: 100, + }); + }); + it('opens only the replacement when the old connection closes while its URL is fetched', async () => { + await client.start(); + const old = exposed_ws_connection; + server.close(); + server = createServer((_req, res) => { + setTimeout(() => { + res.writeHead(200, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ ok: true, url: `ws://localhost:${WSS_PORT}/` })); + }, 300); + }); + server.listen(HTTP_PORT); + let connections = 0; + const replaced = new Promise((resolve) => { + wss.on('connection', () => { + connections += 1; + resolve(); + }); + }); + + old.send(JSON.stringify({ type: 'disconnect', reason: 'refresh_requested' })); + await sleep(50); + old.terminate(); + await replaced; + await sleep(300); + + assert.strictEqual(connections, 1); + assert.strictEqual(wss.clients.size, 1); + await client.disconnect(); + }); + it('opens one replacement when refresh_requested arrives on the old connection before the new one says hello', async () => { + await client.start(); + const old = exposed_ws_connection; + let connections = 0; + wss.removeAllListeners('connection'); + wss.on('connection', (ws) => { + connections += 1; + exposed_ws_connection = ws; + setTimeout(() => ws.send(JSON.stringify({ type: 'hello' })), 300); + }); + + old.send(JSON.stringify({ type: 'disconnect', reason: 'warning' })); + await sleep(100); + old.send(JSON.stringify({ type: 'disconnect', reason: 'refresh_requested' })); + await sleep(500); + + assert.strictEqual(connections, 1); + assert.strictEqual(wss.clients.size, 1); + await client.disconnect(); + }); + it('receives and acknowledges an envelope sent on the old connection after the new one says hello', async () => { + await client.start(); + const old = exposed_ws_connection; + const acks = []; + old.on('message', (data) => acks.push(JSON.parse(data).envelope_id)); + const connected = new Promise((resolve) => client.once('connected', resolve)); + const received = new Promise((resolve) => + client.once('interactive', async ({ ack, envelope_id }) => { + await ack(); + resolve(envelope_id); + }), + ); + + old.send(JSON.stringify({ type: 'disconnect', reason: 'warning' })); + await connected; + old.send(JSON.stringify({ type: 'interactive', envelope_id: 'e-old', payload: {} })); + + assert.strictEqual(await Promise.race([received, sleep(500)]), 'e-old'); + await sleep(50); + assert.deepStrictEqual(acks, ['e-old']); + await client.disconnect(); + }); + }); describe('`apps.connections.open` API failure modes', () => { beforeEach(() => { client = new SocketModeClient({