WIP: Handling 409 response from websocket server

packages/socket-mode/src/SlackWebSocket.ts

@@ -111,29 +111,28 @@
       agent: this.options.httpAgent,
     };
 
-    const ws = new WebSocket(this.options.url, options);
+    this.websocket = new WebSocket(this.options.url, options);
 
-    ws.addEventListener('open', (_event) => {
+    this.websocket.addEventListener('open', (_event) => {
       this.logger.debug('WebSocket open event received (connection established)!');
-      this.websocket = ws;
       this.monitorPingToSlack();
     });
-    ws.addEventListener('error', (event) => {
+    this.websocket.addEventListener('error', (event) => {
       this.logger.error(`WebSocket error occurred: ${event.message}`);
       this.disconnect();
       this.options.client.emit('error', websocketErrorWithOriginal(event.error));
     });
-    ws.on('message', (msg, isBinary) => {
+    this.websocket.on('message', (msg, isBinary) => {
       this.options.client.emit('ws_message', msg, isBinary);
     });
-    ws.on('close', (code: number, data: Buffer) => {
+    this.websocket.on('close', (code: number, data: Buffer) => {
       this.logger.debug(`WebSocket close frame received (code: ${code}, reason: ${data.toString()})`);
       this.closeFrameReceived = true;
       this.disconnect();
     });
 
     // Confirm WebSocket connection is still active
-    ws.on('ping', (data) => {
+    this.websocket.on('ping', (data) => {
       // Note that ws' `autoPong` option is true by default, so no need to respond to ping.
       // see https://github.com/websockets/ws/blob/2aa0405a5e96754b296fef6bd6ebdfb2f11967fc/doc/ws.md#new-websocketaddress-protocols-options
       if (this.options.pingPongLoggingEnabled) {
@@ -142,7 +141,7 @@
       this.monitorPingFromSlack();
     });
 
-    ws.on('pong', (data) => {
+    this.websocket.on('pong', (data) => {
       if (this.options.pingPongLoggingEnabled) {
         this.logger.debug(`WebSocket received pong from Slack server (data: ${data.toString()})`);
       }

packages/socket-mode/src/SocketModeClient.ts

@@ -229,9 +229,13 @@
     this.logger.debug(`Before trying to reconnect, this client will wait for ${msBeforeRetry} milliseconds`);
     return new Promise((res, _rej) => {
       setTimeout(() => {
-        this.logger.debug('Continuing with reconnect...');
-        this.emit(State.Reconnecting);
-        cb.apply(this).then(res);
+        if (this.shuttingDown) {
+          this.logger.debug('Client shutting down, will not attempt reconnect.');
+        } else {
+          this.logger.debug('Continuing with reconnect...');
+          this.emit(State.Reconnecting);
+          cb.apply(this).then(res);
+        }
       }, msBeforeRetry);
     });
   }

packages/socket-mode/src/errors.ts

@@ -61,10 +61,7 @@ function errorWithCode(error: Error, code: ErrorCode): CodedError {
  * A factory to create SMWebsocketError objects.
  */
 export function websocketErrorWithOriginal(original: Error): SMWebsocketError {
-  const error = errorWithCode(
-    new Error(`Failed to send message on websocket: ${original.message}`),
-    ErrorCode.WebsocketError,
-  ) as Partial<SMWebsocketError>;
+  const error = errorWithCode(new Error(original.message), ErrorCode.WebsocketError) as Partial<SMWebsocketError>;
   error.original = original;
   return error as SMWebsocketError;
 }

packages/socket-mode/test/integration.spec.js

@@ -42,9 +42,9 @@ describe('Integration tests with a WebSocket server', () => {
     });
   });
   afterEach(async () => {
-    server.close();
+    if (server) server.close();
     server = null;
-    wss.close();
+    if (wss) wss.close();
     wss = null;
     exposed_ws_connection = null;
     if (client) {
@@ -90,7 +90,7 @@ describe('Integration tests with a WebSocket server', () => {
       await client.disconnect();
     });
   });
-  describe('catastrophic server behaviour', () => {
+  describe('`apps.connections.open` API failure modes', () => {
     beforeEach(() => {
       client = new SocketModeClient({
         appToken: 'whatever',
@@ -130,7 +130,7 @@ describe('Integration tests with a WebSocket server', () => {
       await client.disconnect();
     });
   });
-  describe('failure modes / unexpected messages sent to client', () => {
+  describe('unexpected socket messages sent to client', () => {
     const debugLoggerSpy = sinon.stub(); // add the following to expose further logging: .callsFake(console.log);
     const noop = () => {};
     beforeEach(() => {
@@ -139,6 +139,7 @@ describe('Integration tests with a WebSocket server', () => {
         clientOptions: {
           slackApiUrl: `http://localhost:${HTTP_PORT}/`,
         },
+        logLevel: 'debug',
         logger: {
           debug: debugLoggerSpy,
           info: noop,
@@ -168,6 +169,47 @@ describe('Integration tests with a WebSocket server', () => {
       assert.isTrue(debugLoggerSpy.calledWith(sinon.match('Unable to parse an incoming WebSocket message')));
       await client.disconnect();
     });
+    it('should maintain one serial reconnection attempt if WSS server sends unexpected HTTP response during handshake, like a 409', async () => {
+      // override socket mode client instance with lower client ping timeout, which controls reconnection rate
+      client = new SocketModeClient({
+        appToken: 'whatever',
+        clientOptions: {
+          slackApiUrl: `http://localhost:${HTTP_PORT}/`,
+        },
+        clientPingTimeout: 20,
+        logLevel: 'debug',
+      });
+      // shut down the default mock WS server used in these tests as we will customize its behaviour in this test
+      wss.close();
+      wss = null;
+      // custom HTTP server that blows up during initial WS handshake
+      const badServer = createServer((_req, res) => {
+        res.writeHead(409, { 'content-type': 'application/json' });
+        res.end(
+          JSON.stringify({
+            message: 'Unexpected server response: 409',
+          }),
+        );
+      });
+      badServer.listen(WSS_PORT);
+      let closed = 0;
+      client.on('close', () => {
+        closed++;
+      });
+      // do not use await here, since `start()` won't return until the connection is established. we are intentionally testing the connection establishment failure, so that will never finish. so, let's run this in a rogue "thread"!
+      client.start();
+      await sleep(50);
+      // after 50ms, with a timeout of 20ms, we would expect 2 retries by this point.
+      assert.equal(closed, 2, 'unexpected number of times `close` event was raised during reconnection!');
+      await client.disconnect();
+      await new Promise((res, rej) => {
+        // shut down the bad server
+        badServer.close((err) => {
+          if (err) rej(err);
+          else res();
+        });
+      });
+    });
   });
   describe('lifecycle events', () => {
     beforeEach(() => {