fix/matching: Add graceful process exit for main process

Signed-off-by: SeeuSim <[email protected]>

Author: SeeuSim

backend/matching/src/index.ts

@@ -5,7 +5,7 @@ import { logger } from '@/lib/utils';
 import server, { io } from '@/server';
 import { initWorker } from '@/workers';
 
-const workers: { controller: AbortController; worker: ChildProcess }[] = [];
+const workers: ChildProcess[] = [];
 
 const port = Number.parseInt(EXPRESS_PORT || '8001');
 
@@ -18,15 +18,17 @@ server.listen(port, () => {
 });
 
 const shutdown = () => {
-  workers.forEach(({ controller, worker }) => {
-    controller.abort('Main process shutdown'); // Force an abort error to kill child processes
-    worker.kill();
-  });
-  io.close(() => {
-    logger.info('WS Server shut down');
-  });
   server.close(() => {
-    logger.info('App shut down');
+    workers.forEach((worker) => {
+      worker.kill();
+    });
+    void io
+      .close(() => {
+        logger.info('WS Server shut down');
+      })
+      .then(() => {
+        logger.info('App shut down');
+      });
   });
 };
 

backend/matching/src/workers/cleaner.js

@@ -1,5 +1,4 @@
-
 const path = require('path');
 
 require('ts-node').register();
-require(path.resolve(__dirname, './cleaner.ts'));
+require(path.resolve(__dirname, './cleaner.ts'));

backend/matching/src/workers/cleaner.ts

@@ -19,15 +19,14 @@ const cancel = () => {
 };
 const shutdown = () => {
   cancel();
-  client
-    .disconnect()
-    .then(() => client.quit())
-    .then(process.exit(0));
+  client.disconnect().then(() => {
+    process.exit(0);
+  });
 };
 
-process.on('exit', shutdown);
 process.on('SIGINT', shutdown);
 process.on('SIGTERM', shutdown);
+process.on('exit', shutdown);
 
 async function clean() {
   const redisClient = await connectClient(client);

backend/matching/src/workers/common.ts

@@ -1,3 +1,5 @@
+// CHILD PROCESS UTIL LIB
+
 import { client } from '@/lib/db';
 import { logger } from '@/lib/utils';
 import type { IChildProcessMessage, IMatchEvent } from '@/ws';

backend/matching/src/workers/index.ts

@@ -1,3 +1,4 @@
+// MAIN PROCESS
 import { fork } from 'child_process';
 import path from 'path';
 
@@ -6,10 +7,11 @@ import type { Server } from 'socket.io';
 import { logger } from '@/lib/utils';
 import { MATCH_SVC_EVENT, type IChildProcessMessage } from '@/ws';
 
+let nWorkers = 0; // For tracking graceful exit of main process
 export const initWorker = (name: string, io: Server) => {
-  const controller = new AbortController();
   const lCaseName = name.toLowerCase();
-  const worker = fork(path.join(__dirname, `${lCaseName}.js`), { signal: controller.signal });
+  const worker = fork(path.join(__dirname, `${lCaseName}.js`));
+  nWorkers += 1;
   const upperCaseName = name.replace(/^[A-Za-z]/, (c) => c.toUpperCase());
   worker.on('message', (message) => {
     if (typeof message.valueOf() === 'string') {
@@ -27,6 +29,11 @@ export const initWorker = (name: string, io: Server) => {
   });
   worker.on('exit', (code) => {
     logger.error(`${upperCaseName} exited with code ${code}.`);
+    nWorkers -= 1;
+    if (nWorkers === 0) {
+      logger.info('Main Process exiting.');
+      process.exit(0);
+    }
   });
-  return { worker, controller };
+  return worker;
 };

backend/matching/src/workers/matcher.js

@@ -1,5 +1,4 @@
-
 const path = require('path');
 
 require('ts-node').register();
-require(path.resolve(__dirname, './matcher.ts'));
+require(path.resolve(__dirname, './matcher.ts'));

backend/matching/src/workers/matcher.ts

@@ -19,15 +19,14 @@ const cancel = () => {
 };
 const shutdown = () => {
   cancel();
-  client
-    .disconnect()
-    .then(() => client.quit())
-    .then(process.exit(0));
+  client.disconnect().then(() => {
+    process.exit(0);
+  });
 };
 
-process.on('exit', shutdown);
 process.on('SIGINT', shutdown);
 process.on('SIGTERM', shutdown);
+process.on('exit', shutdown);
 
 type RequestorParams = {
   requestorUserId: string;
@@ -42,32 +41,36 @@ async function processMatch(
   searchIdentifier?: string
 ) {
   if (matches.total > 0) {
-    const matched = matches.documents[0];
-    const {
-      userId: matchedUserId,
-      timestamp, // We use timestamp as the Stream ID
-      socketPort: matchedSocketPort,
-    } = decodePoolTicket(matched);
-
-    // To block cancellation
-    sendNotif([matchedSocketPort], MATCH_SVC_EVENT.MATCHING);
-
-    const matchedStreamId = getStreamId(timestamp);
-
-    logger.info(`Found match: ${JSON.stringify(matched)}`);
-
-    await Promise.all([
-      // Remove other from pool
-      redisClient.del([getPoolKey(requestorUserId), getPoolKey(matchedUserId)]),
-      // Remove other from queue
-      redisClient.xDel(STREAM_NAME, [requestorStreamId, matchedStreamId]),
-    ]);
-
-    // Notify both sockets
-    const { ...matchItems } = getMatchItems();
-    sendNotif([requestorSocketPort, matchedSocketPort], MATCH_SVC_EVENT.SUCCESS, matchItems);
-    sendNotif([requestorSocketPort, matchedSocketPort], MATCH_SVC_EVENT.DISCONNECT);
-    return true;
+    for (const matched of matches.documents) {
+      const {
+        userId: matchedUserId,
+        timestamp, // We use timestamp as the Stream ID
+        socketPort: matchedSocketPort,
+      } = decodePoolTicket(matched);
+      if (matchedUserId === requestorUserId) {
+        continue;
+      }
+
+      // To block cancellation
+      sendNotif([matchedSocketPort], MATCH_SVC_EVENT.MATCHING);
+
+      const matchedStreamId = getStreamId(timestamp);
+
+      logger.info(`Found match: ${JSON.stringify(matched)}`);
+
+      await Promise.all([
+        // Remove other from pool
+        redisClient.del([getPoolKey(requestorUserId), getPoolKey(matchedUserId)]),
+        // Remove other from queue
+        redisClient.xDel(STREAM_NAME, [requestorStreamId, matchedStreamId]),
+      ]);
+
+      // Notify both sockets
+      const { ...matchItems } = getMatchItems();
+      sendNotif([requestorSocketPort, matchedSocketPort], MATCH_SVC_EVENT.SUCCESS, matchItems);
+      sendNotif([requestorSocketPort, matchedSocketPort], MATCH_SVC_EVENT.DISCONNECT);
+      return true;
+    }
   }
 
   logger.info(`Found no matches` + (searchIdentifier ? ` for ${searchIdentifier}` : ''));