fix(web sockets): server detects disconnected clients and closes them

.github/workflows/pullRequest.yml

@@ -4,28 +4,29 @@ on:
   pull_request:
 
 jobs:
-  unit-tests:
-    strategy:
-      matrix:
-        project:
-          ['admin-web', 'core-server', 'core-client', 'utils-generate-guid']
-    name: Jest Unit Test
-    runs-on: ubuntu-latest
-    timeout-minutes: 5
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v1
-      - name: Setup Node
-        uses: actions/setup-node@v1
-        with:
-          node-version: 16
-      - name: Npm install
-        run: npm install
-      - name: install nx
-        run: npm i -g nx
-      - name: Unit test
-        run: nx test ${{ matrix.project }}
-        # run: npm run testAll
+  # Should be covered by testCoverage.yml
+  # unit-tests:
+  #   strategy:
+  #     matrix:
+  #       project:
+  #         ['admin-web', 'core-server', 'core-client', 'utils-generate-guid']
+  #   name: Jest Unit Test
+  #   runs-on: ubuntu-latest
+  #   timeout-minutes: 5
+  #   steps:
+  #     - name: Checkout
+  #       uses: actions/checkout@v1
+  #     - name: Setup Node
+  #       uses: actions/setup-node@v1
+  #       with:
+  #         node-version: 16
+  #     - name: Npm install
+  #       run: npm install
+  #     - name: install nx
+  #       run: npm i -g nx
+  #     - name: Unit test
+  #       run: nx test ${{ matrix.project }}
+  #       # run: npm run testAll
   linting:
     name: Linting
     runs-on: ubuntu-latest

apps/admin-web/src/app/hooks/useRecordingClient.ts

@@ -73,10 +73,9 @@ export default function useRecordingClient() {
 
   useEffect(() => {
     const onCommand = (data: any) => {
-      log.debug('[useRecordingClinet] on command', data);
+      log.debug('[useRecordingClient] received', data);
       if (data.type === 'api.response') {
         const message: RecordedItem = data;
-        log.debug('[RecordScreen] onCommand api.response', message);
         dispatch({ type: 'add', payload: message });
       }
     };
@@ -108,7 +107,7 @@ export default function useRecordingClient() {
 
     return () => {
       log.info('[useRecordingClient] unmount, calling close on socket');
-      mezzoClient.current?.send('close');
+      mezzoClient.current?.send('close'); // server will detect disconnect, but this way client can also signal it's done
       mezzoClient.current?.close();
       mezzoClient.current = null;
     };

libs/core-client-server-tests/src/lib/testPorts.ts

@@ -2,7 +2,7 @@ export const adminEndpointsPort = 3050;
 export const adminEndpointsProfilesPort = 3100;
 export const websocketPort = 3200;
 export const clientServerIntegrationPort = 3250;
-export const corePort = 3300;
+// export const corePort = 3300;
 export const globalVariantPort = 3350;
 export const redirectPort = 3400;
 export const routeExpressPort = 3450;

libs/core-client/src/lib/plugins/webSocketClient.ts

@@ -146,12 +146,9 @@ export function webSocketClient(options: IWebSocketClientOptions) {
       log.info('[mezzo-core-client.onClose] Closing socket continued');
       isReady = false;
       _readyState = WebSocket.CLOSING;
-      // trigger our disconnect handler
+      connectDebounced?.cancel(); // if for some reason a connection attempt debounce is in flight but we disconnect, cancel (unit test was not stopping properly on close w/out this)
       onDisconnect?.();
       _readyState = WebSocket.CLOSED;
-
-      // as well as the plugin's onDisconnect
-      // plugins.forEach((p) => p.onDisconnect && p.onDisconnect());
     };
 
     // fires when we receive a command, just forward it off
@@ -167,8 +164,6 @@ export function webSocketClient(options: IWebSocketClientOptions) {
       if (command.type === 'setClientId') {
         setClientId?.(command.payload);
       } else if (command.type === 'ping') {
-        log.debug('[heartbeat]');
-        // reply with pong
         send('pong');
       }
     };
@@ -218,10 +213,8 @@ export function webSocketClient(options: IWebSocketClientOptions) {
     };
 
     const serializedMessage = JSON.stringify(fullMessage);
-    // const serializedMessage = fullMessage;
 
     if (isReady) {
-      log.debug('[core-client-send] attempting to send as markred as isReady');
       try {
         socket.send(serializedMessage);
       } catch {

libs/core-server/src/lib/__tests__/core-1.spec.ts

@@ -22,7 +22,7 @@ describe('route-express', () => {
 
   beforeEach(async () => {
     process.env.LOG_LEVEL = 'warn';
-    const port = routeExpressPort;
+    const port = routeExpressPort + Number(process.env.JEST_WORKER_ID);
     request = SuperTestRequest(`http://localhost:${port}`);
     await mezzo.start({
       port,

libs/core-server/src/lib/__tests__/core-2.spec.ts

@@ -25,7 +25,7 @@ describe('route-state', () => {
   const sessionId = '123';
   beforeEach(async () => {
     process.env.LOG_LEVEL = 'warn';
-    const port = routeStatePort;
+    const port = routeStatePort + Number(process.env.JEST_WORKER_ID);
     request = SuperTestRequest(`http://localhost:${port}`);
     mockedDirectory = path.join(resourcesPath, 'some-custom-mocked-data');
     await mezzo.start({

libs/core-server/src/lib/plugins/__tests__/profile-endpoints.spec.ts

@@ -6,7 +6,7 @@ import { adminEndpointsProfilesPort } from '@mezzo/core-client-server-tests';
 
 describe('profile-endpoints', () => {
   let request: SuperTestRequest.SuperTest<SuperTestRequest.Test>;
-  const port = adminEndpointsProfilesPort;
+  const port = adminEndpointsProfilesPort + Number(process.env.JEST_WORKER_ID);
   let client;
   beforeAll(() => {
     client = MezzoClient({ port });

libs/core-server/src/lib/plugins/__tests__/record-REST-endpoints.spec.ts

@@ -26,7 +26,6 @@ const movies = [
 ];
 describe('recording REST endpoints', () => {
   let request: SuperTestRequest.SuperTest<SuperTestRequest.Test>;
-  // const port = recordingServerPort + Jes;
   const port = recordingServerPort + Number(process.env.JEST_WORKER_ID);
 
   beforeAll(() => {

libs/core-server/src/lib/plugins/__tests__/record-ws.spec.ts

@@ -1,9 +1,7 @@
 import mezzo from '../../core';
 import { recordingWSPort } from '@mezzo/core-client-server-tests';
 import SuperTestRequest from 'supertest';
-import logger from '@caribou-crew/mezzo-utils-logger';
 import { WebSocket } from 'ws';
-// import { webSocketClient } from '@caribou-crew/mezzo-core-client';
 import mezzoClient from '@caribou-crew/mezzo-core-client';
 import { IClientOptions } from '@caribou-crew/mezzo-interfaces';
 import { MEZZO_API_GET_RECORDING_CLIENTS } from '@caribou-crew/mezzo-constants';
@@ -56,7 +54,6 @@ function timeout(ms: number) {
 
 describe('recordingServer', () => {
   const port = recordingWSPort + Number(process.env.JEST_WORKER_ID);
-  // let ws: WebSocket;
   let client: ReturnType<typeof mezzoClient>;
   let messages = [];
   let request: SuperTestRequest.SuperTest<SuperTestRequest.Test>;
@@ -98,20 +95,19 @@ describe('recordingServer', () => {
         res.json(movies);
       },
     });
-    // ws = new WebSocket(`ws:localhost:${port}`);
   });
   afterEach(async () => {
-    // ws.close();
     client?.close();
     await mezzo.stop();
-    await timeout(DEFAULT_DELAY);
   });
 
   it('client sending response capture should receive api.response type message', async () => {
     await connectionHelper({
       onCommand: (data) => {
         console.log('Got message: ', data);
-        messages.push(data);
+        if (data.type !== 'ping') {
+          messages.push(data);
+        }
       },
       createSocket,
       port,
@@ -140,7 +136,9 @@ describe('recordingServer', () => {
     await connectionHelper({
       onCommand: (data) => {
         console.log('Clinet 1 Got message: ', data);
-        messages.push(data);
+        if (data.type !== 'ping') {
+          messages.push(data);
+        }
       },
       createSocket,
       port,
@@ -149,8 +147,9 @@ describe('recordingServer', () => {
     // Client 2
     const otherClient = mezzoClient({
       onCommand: (data) => {
-        console.log('Client 2 got message: ', data);
-        client2Messages.push(data);
+        if (data.type !== 'ping') {
+          client2Messages.push(data);
+        }
       },
       createSocket,
       port,

libs/core-server/src/lib/plugins/__tests__/routes-endpoints.spec.ts

@@ -20,7 +20,7 @@ describe('admin-endpoints', () => {
   let client;
   beforeEach(async () => {
     process.env.LOG_LEVEL = 'warn';
-    const port = adminEndpointsPort;
+    const port = adminEndpointsPort + Number(process.env.JEST_WORKER_ID);
     client = mezzoClient({ port });
     request = SuperTestRequest(`http://localhost:${port}`);
     await mezzo.start({

libs/core-server/src/lib/plugins/record-endpoints.ts

@@ -5,21 +5,25 @@ import {
   MEZZO_API_GET_RECORDING_CLIENTS,
 } from '@caribou-crew/mezzo-constants';
 import {
+  IMezzoServerPlugin,
   RecordedItem,
   SocketRequestResponseMessage,
 } from '@caribou-crew/mezzo-interfaces';
 import { Mezzo } from '../core';
 import logger from '@caribou-crew/mezzo-utils-logger';
 import { generateGuid } from '@caribou-crew/mezzo-utils-generate-guid';
 
-// TODO type this object in interfaces so that it can also be used via RecordingScreen.tsx
-interface Clients {
+interface Client {
   id: number;
   ws: WebSocket;
+  pingTimeout?: NodeJS.Timeout;
+  disconnectTimeout?: NodeJS.Timeout;
 }
-const clients: Clients[] = [];
+const clients: Client[] = [];
 let id = 0;
 
+const keepAliveInterval = 5000;
+const disconnectAfterSilence = keepAliveInterval * 3.5;
 const recordedItems: RecordedItem[] = [];
 
 function setupRestApi(app: express.Express) {
@@ -45,26 +49,28 @@ function setupWebSocketServer(mezzo: Mezzo) {
   const websocketServer = new WebSocket.Server({ server: mezzo.server });
   websocketServer.on('connection', (ws: WebSocket) => {
     id += 1;
-    clients.push({
+    const client = {
       id,
       ws,
-    });
+    };
+    clients.push(client);
     logger.info(
       'Client connected to recording socket server, total clients: ',
       clients.map((i) => i.id)
     );
 
     ws.on('message', (message: string) => {
       // Process message from the client
-      processMessage(message, ws);
+      processMessage(message, client);
     });
     // If it is desired for server to send message to client on connect, add ws.send(...) here
+    ping(client);
   });
 
   mezzo.websocketServer = websocketServer;
 }
 
-function notifyAllClientsJSON(message: any, type?: string) {
+function broadcastJSON(message: any, type?: string) {
   logger.debug(`Sending message to all ${clients.length} clients`);
   clients.forEach(({ ws }) => {
     ws.send(JSON.stringify({ type, ...message }));
@@ -85,47 +91,73 @@ function processRequestResponseMessage(message: SocketRequestResponseMessage) {
       duration: message.payload.duration,
     };
     recordedItems.push(item);
-    notifyAllClientsJSON(item, 'api.response');
+    broadcastJSON(item, 'api.response');
   } else {
     logger.warn(
       "Received message but didn't add to data itmes as type was null"
     );
   }
 }
+function ping(client: Client) {
+  const { ws } = client;
+  if (ws.readyState === WebSocket.OPEN) {
+    ws.send(JSON.stringify({ type: 'ping' }));
+    // set disconnect timeout, if pong not received by then disconnect user
+    client.disconnectTimeout = setTimeout(() => {
+      removeClient(client);
+    }, disconnectAfterSilence);
+  } else {
+    removeClient(client);
+  }
+}
 
-function closeClient(ws: WebSocket) {
-  logger.debug('Removing client: ', id);
-  const idx = clients.findIndex((i) => i.id === id);
+function removeClient(client: Client) {
+  logger.info(`Client ${client.id} disconnected`);
+  clearTimeout(client.pingTimeout);
+  clearTimeout(client.disconnectTimeout);
+  const idx = clients.findIndex((i) => i.id === client.id);
   clients.splice(idx, 1);
-  ws.close();
+  client.ws.close();
 }
 
-function processMessage(message, ws: WebSocket) {
-  // attempt to parse
+function processMessage(message, client: Client) {
   try {
     const data = JSON.parse(message);
     logger.debug('Message received', data?.type);
     if (data?.type === 'api.response') {
       logger.debug('Received api response');
       processRequestResponseMessage(data);
-    } else if (data?.type === 'ping') {
-      ws.send(JSON.stringify({ type: 'pong' }));
+    } else if (data?.type === 'pong') {
+      clearTimeout(client.disconnectTimeout);
+      client.pingTimeout = setTimeout(() => {
+        ping(client);
+      }, keepAliveInterval);
     } else if (data?.type === 'close') {
-      closeClient(ws);
+      removeClient(client);
     }
   } catch (e) {
     logger.error('Error parsings ws message', e);
   }
 }
 
-export default () => (mezzo: Mezzo) => {
-  setupRestApi(mezzo.app);
-  setupWebSocketServer(mezzo);
-  return {
-    name: 'recording-endpoints-plugin',
-    initialize: () => {
-      clients.length = 0;
-      recordedItems.length = 0;
-    },
+export default () =>
+  (mezzo: Mezzo): IMezzoServerPlugin => {
+    setupRestApi(mezzo.app);
+    setupWebSocketServer(mezzo);
+    return {
+      name: 'recording-endpoints-plugin',
+      initialize: () => {
+        clients.length = 0;
+        recordedItems.length = 0;
+      },
+      onStop: () => {
+        clients.forEach((client) => {
+          clearTimeout(client.disconnectTimeout);
+          clearTimeout(client.pingTimeout);
+        });
+        recordedItems.length = 0;
+        clients.length = 0;
+        id = 0;
+      },
+    };
   };
-};

libs/core-server/src/lib/utils/__tests__/redirect.spec.ts

@@ -9,7 +9,7 @@ describe('redirects', () => {
   let request: SuperTestRequest.SuperTest<SuperTestRequest.Test>;
 
   beforeEach(async () => {
-    const port = redirectPort;
+    const port = redirectPort + Number(process.env.JEST_WORKER_ID);
     request = SuperTestRequest(`http://localhost:${port}`);
     await mezzo.start({
       port,

libs/core-server/src/lib/utils/__tests__/respondWithFile.spec.ts

@@ -52,7 +52,7 @@ describe('route-file-io', () => {
   `;
   beforeEach(async () => {
     process.env.LOG_LEVEL = 'warn';
-    const port = fileIOPort;
+    const port = fileIOPort + Number(process.env.JEST_WORKER_ID);
     client = mezzoClient({ port });
     request = SuperTestRequest(`http://localhost:${port}`);
     mockedDirectory = path.join(resourcesPath, 'some-custom-mocked-data');