fix(sandbox): ignore extra params on processor (#2142)

.github/workflows/test.yml

@@ -66,7 +66,7 @@ jobs:
     strategy:
       matrix:
         node-version: [lts/*, lts/-1, lts/-2, current]
-        redis-version: [7-alpine]
+        redis-version: [7.0-alpine]
         include:
           - node-version: 'lts/*'
             redis-version: 6-alpine

src/classes/child-processor.ts

@@ -1,4 +1,3 @@
-import { promisify } from 'util';
 import { JobJson, ParentCommand, SandboxedJob } from '../interfaces';
 import { errorToJSON } from '../utils';
 
@@ -44,18 +43,15 @@ export class ChildProcessor {
       });
     }
 
-    if (processor.length > 1) {
-      processor = promisify(processor);
-    } else {
-      const origProcessor = processor;
-      processor = function (...args: any[]) {
-        try {
-          return Promise.resolve(origProcessor(...args));
-        } catch (err) {
-          return Promise.reject(err);
-        }
-      };
-    }
+    const origProcessor = processor;
+    processor = function (job: SandboxedJob, token?: string) {
+      try {
+        return Promise.resolve(origProcessor(job, token));
+      } catch (err) {
+        return Promise.reject(err);
+      }
+    };
+
     this.processor = processor;
     this.status = ChildStatus.Idle;
     await this.send({

tests/fixtures/fixture_processor_with_extra_param.js

@@ -0,0 +1,13 @@
+/**
+ * A processor file to be used in tests.
+ *
+ */
+'use strict';
+
+const delay = require('./delay');
+
+module.exports = function (job, extraParam) {
+  return delay(500).then(() => {
+    return 42;
+  });
+};

tests/test_sandboxed_process.ts

@@ -73,6 +73,43 @@ function sandboxProcessTests(
       await worker.close();
     });
 
+    describe('when processor has more than 1 param', () => {
+      it('should ignore extra params, process and complete', async () => {
+        const processFile =
+          __dirname + '/fixtures/fixture_processor_with_extra_param.js';
+
+        const worker = new Worker(queueName, processFile, {
+          connection,
+          drainDelay: 1,
+          useWorkerThreads,
+        });
+
+        const completing = new Promise<void>((resolve, reject) => {
+          worker.on('completed', async (job: Job, value: any) => {
+            try {
+              expect(job.data).to.be.eql({ foo: 'bar' });
+              expect(value).to.be.eql(42);
+              expect(
+                Object.keys(worker['childPool'].retained),
+              ).to.have.lengthOf(0);
+              expect(worker['childPool'].free[processFile]).to.have.lengthOf(1);
+              await worker.close();
+              resolve();
+            } catch (err) {
+              await worker.close();
+              reject(err);
+            }
+          });
+        });
+
+        await queue.add('test', { foo: 'bar' });
+
+        await completing;
+
+        await worker.close();
+      });
+    });
+
     describe('when processor file is .cjs (CommonJS)', () => {
       it('processes and completes', async () => {
         const processFile = __dirname + '/fixtures/fixture_processor.cjs';