fix(sandbox): ensure non-bullmq messages are ignored (#3704) fixes #3703

Author: lewnelson

src/classes/child.ts

@@ -175,6 +175,10 @@ export class Child extends EventEmitter {
   private async initChild() {
     const onComplete = new Promise<void>((resolve, reject) => {
       const onMessageHandler = (msg: any) => {
+        if (!Object.values(ParentCommand).includes(msg.cmd)) {
+          return;
+        }
+
         if (msg.cmd === ParentCommand.InitCompleted) {
           resolve();
         } else if (msg.cmd === ParentCommand.InitFailed) {

tests/fixtures/fixture_main_non_bullmq_messages.js

@@ -0,0 +1,65 @@
+/**
+ * Custom main file that sends non-BullMQ messages during initialization
+ * to test that the child process properly ignores them.
+ */
+'use strict';
+
+const { ChildProcessor } = require('../../dist/cjs/classes/child-processor');
+const { ParentCommand, ChildCommand } = require('../../dist/cjs/enums');
+const { errorToJSON, toString } = require('../../dist/cjs/utils');
+
+const send = async msg => {
+  if (process.send) {
+    return new Promise((resolve, reject) => {
+      process.send(msg, err => {
+        if (err) {
+          reject(err);
+        } else {
+          resolve();
+        }
+      });
+    });
+  }
+};
+
+const childProcessor = new ChildProcessor(send, process);
+
+process.on('message', async msg => {
+  try {
+    switch (msg.cmd) {
+      case ChildCommand.Init:
+        // Send non-BullMQ messages before initialization
+        await send({ type: 'debug', message: 'Starting initialization' });
+        await send({ randomKey: 'randomValue' });
+        await send({ cmd: 'INVALID_COMMAND' });
+        await send({ cmd: 999 });
+
+        // Now do the actual initialization
+        await childProcessor.init(msg.value);
+        break;
+      case ChildCommand.Start:
+        await childProcessor.start(msg.job, msg?.token);
+        break;
+      case ChildCommand.Stop:
+        break;
+    }
+  } catch (err) {
+    console.error('Error handling child message');
+  }
+});
+
+process.on('SIGTERM', () => childProcessor.waitForCurrentJobAndExit());
+process.on('SIGINT', () => childProcessor.waitForCurrentJobAndExit());
+
+process.on('uncaughtException', async err => {
+  if (typeof err !== 'object') {
+    err = new Error(toString(err));
+  }
+
+  await send({
+    cmd: ParentCommand.Failed,
+    value: errorToJSON(err),
+  });
+
+  process.exit();
+});

tests/fixtures/fixture_processor_simple.js

@@ -0,0 +1,8 @@
+/**
+ * A simple processor for testing non-BullMQ message handling.
+ */
+'use strict';
+
+module.exports = function(job) {
+  return Promise.resolve({ processed: true, data: job.data });
+};

tests/sandboxed_process.test.ts

@@ -57,6 +57,24 @@ describe('Sandboxed process using child processes', () => {
       await connection.quit();
     });
 
+    it('should handle non-BullMQ messages during child initialization', async function () {
+      const mainFile = __dirname + '/fixtures/fixture_main_non_bullmq_messages.js';
+      const processFile = __dirname + '/fixtures/fixture_processor_simple.js';
+
+      const child = new Child(mainFile, processFile, {
+        useWorkerThreads: false,
+      });
+
+      // This should complete successfully even though non-BullMQ messages
+      // are sent before the InitCompleted message
+      await child.init();
+
+      expect(child.exitCode).to.be.null;
+      expect(child.killed).to.be.false;
+
+      await child.kill();
+    });
+
     it('should allow to pass workerForkOptions', async () => {
       const processFile = __dirname + '/fixtures/fixture_processor.js';