fix(sandbox): broken processor files should fail jobs

OptimalBits · Feb 6, 2022 · dd0b853 · dd0b853
1 parent 27ca17b
commit dd0b853
Show file tree

Hide file tree

Showing 5 changed files with 49 additions and 19 deletions.
diff --git a/lib/process/child-pool.js b/lib/process/child-pool.js
@@ -58,9 +58,14 @@ ChildPool.prototype.retain = function(processFile) {
 
  child.on('exit', _this.remove.bind(_this, child));
 
- return initChild(child, child.processFile).then(() => {
- return child;
- });
+ return initChild(child, child.processFile)
+ .then(() => {
+ return child;
+ })
+ .catch(err => {
+ this.remove(child);
+ throw new Error(err);
+ });
  });
 };
 
@@ -106,12 +111,14 @@ ChildPool.prototype.getAllFree = function() {
 };
 
 async function initChild(child, processFile) {
- const onComplete = new Promise(resolve => {
+ const onComplete = new Promise((resolve, reject) => {
  const onMessageHandler = msg => {
  if (msg.cmd === 'init-complete') {
  resolve();
- child.off('message', onMessageHandler);
+ } else {
+ reject(msg.value);
  }
+ child.off('message', onMessageHandler);
  };
  child.on('message', onMessageHandler);
  });
@@ -122,11 +129,11 @@ async function initChild(child, processFile) {
  await onComplete;
 }
 function ChildPoolSingleton(isSharedChildPool = false) {
- if(isSharedChildPool === false) {
+ if (isSharedChildPool === false) {
  return new ChildPool();
- }
- else if (
- (!(this instanceof ChildPool) && ChildPoolSingleton.instance === undefined)
+ } else if (
+  !(this instanceof ChildPool) &&
+ ChildPoolSingleton.instance === undefined
  ) {
  ChildPoolSingleton.instance = new ChildPool();
  }

diff --git a/lib/process/master.js b/lib/process/master.js
@@ -52,7 +52,16 @@ process.on('SIGINT', waitForCurrentJobAndExit);
 process.on('message', msg => {
  switch (msg.cmd) {
  case 'init':
- processor = require(msg.value);
+ try {
+ processor = require(msg.value);
+ } catch (err) {
+ status = 'Errored';
+ return process.send({
+ cmd: 'error',
+ value: `Error loading process file ${msg.value}. ${err.toString()}`
+ });
+ }
+
  if (processor.default) {
  // support es2015 module.
  processor = processor.default;
@@ -119,7 +128,7 @@ process.on('uncaughtException', err => {
  cmd: 'failed',
  value: err
  });
- 
+
  // An uncaughException leaves this process in a potentially undetermined state so
  // we must exit
  process.exit(-1);

diff --git a/lib/queue.js b/lib/queue.js
@@ -217,7 +217,7 @@ const Queue = function Queue(name, url, opts) {
  retryProcessDelay: 5000,
  drainDelay: 5,
  backoffStrategies: {},
- isSharedChildPool: false,
+ isSharedChildPool: false
  });
 
  this.settings.lockRenewTime =
@@ -285,6 +285,7 @@ function redisClientGetter(queue, options, initCallback) {
  }
  const clientOptions = _.assign({}, options.redis);
  clientOptions.connectionName = this.clientName();
+
  const client = (connections[type] = createClient(type, clientOptions));
 
  const opts = client.options.redisOptions || client.options;

diff --git a/test/fixtures/fixture_processor_broken.js b/test/fixtures/fixture_processor_broken.js
@@ -0,0 +1,2 @@
+'use strict';
+throw new Error('Broken file processor');
diff --git a/test/test_sandboxed_process.js b/test/test_sandboxed_process.js
@@ -462,27 +462,38 @@ describe('sandboxed process', () => {
  ]);
 
  const processFile = __dirname + '/fixtures/fixture_processor.js';
- queueA.process(processFile)
- queueB.process(processFile)
+ queueA.process(processFile);
+ queueB.process(processFile);
 
  await Promise.all([queueA.add(), queueB.add()]);
 
-
  expect(queueA.childPool).to.be.eql(queueB.childPool);
  });
 
- it('should not share childPool across different queues if isSharedChildPool isn\'t specified', async () => {
+ it("should not share childPool across different queues if isSharedChildPool isn't specified", async () => {
  const [queueA, queueB] = await Promise.all([
  utils.newQueue('queueA', { settings: { isSharedChildPool: false } }),
  utils.newQueue('queueB')
  ]);
 
  const processFile = __dirname + '/fixtures/fixture_processor.js';
- queueA.process(processFile)
- queueB.process(processFile)
+ queueA.process(processFile);
+ queueB.process(processFile);
 
  await Promise.all([queueA.add(), queueB.add()]);
 
  expect(queueA.childPool).to.not.be.equal(queueB.childPool);
- })
+ });
+
+ it('should fail if the process file is broken', async () => {
+ const processFile = __dirname + '/fixtures/fixture_processor_broken.js';
+ queue.process(processFile);
+ await queue.add('test', {});
+
+ return new Promise(resolve => {
+ queue.on('failed', () => {
+ resolve();
+ });
+ });
+ });
 });