feat(sandbox): emulate moveToDelayed method (#2122) ref #2118

src/classes/child-processor.ts

@@ -59,7 +59,7 @@ export class ChildProcessor {
     });
   }
 
-  public async start(jobJson: JobJson): Promise<void> {
+  public async start(jobJson: JobJson, token?: string): Promise<void> {
     if (this.status !== ChildStatus.Idle) {
       return this.send({
         cmd: ParentCommand.Error,
@@ -70,7 +70,8 @@ export class ChildProcessor {
     this.currentJobPromise = (async () => {
       try {
         const job = wrapJob(jobJson, this.send);
-        const result = (await this.processor(job)) || {};
+        //console.log('el token', token, job.id)
+        const result = (await this.processor(job, token)) || {};
         await this.send({
           cmd: ParentCommand.Completed,
           value: result,
@@ -144,7 +145,16 @@ function wrapJob(
       });
     },
     /*
-     * Emulate the real job `update` function.
+     * Emulate the real job `moveToDelayed` function.
+     */
+    moveToDelayed: async (timestamp: number, token?: string) => {
+      send({
+        cmd: ParentCommand.MoveToDelayed,
+        value: { timestamp, token },
+      });
+    },
+    /*
+     * Emulate the real job `updateData` function.
      */
     updateData: async (data: any) => {
       send({

src/classes/main-base.ts

@@ -20,7 +20,7 @@ export default (
           await childProcessor.init(msg.value);
           break;
         case ChildCommand.Start:
-          await childProcessor.start(msg.job);
+          await childProcessor.start(msg.job, msg?.token);
           break;
         case ChildCommand.Stop:
           break;

src/classes/sandbox.ts

@@ -6,14 +6,15 @@ const sandbox = <T, R, N extends string>(
   processFile: any,
   childPool: ChildPool,
 ) => {
-  return async function process(job: Job<T, R, N>): Promise<R> {
+  return async function process(job: Job<T, R, N>, token?: string): Promise<R> {
     const child = await childPool.retain(processFile);
     let msgHandler: any;
     let exitHandler: any;
 
     await child.send({
       cmd: ChildCommand.Start,
       job: job.asJSONSandbox(),
+      token,
     });
 
     const done: Promise<R> = new Promise((resolve, reject) => {
@@ -35,6 +36,9 @@ const sandbox = <T, R, N extends string>(
           case ParentCommand.Log:
             await job.log(msg.value);
             break;
+          case ParentCommand.MoveToDelayed:
+            await job.moveToDelayed(msg.value?.timestamp, msg.value?.token);
+            break;
           case ParentCommand.Update:
             await job.updateData(msg.value);
             break;

src/classes/worker.ts

@@ -651,7 +651,7 @@ export class Worker<
 
           if (
             err instanceof DelayedError ||
-            err.name == 'DelayedError' ||
+            err.message == 'DelayedError' ||
             err instanceof WaitingChildrenError ||
             err.name == 'WaitingChildrenError'
           ) {

src/interfaces/parent-command.ts

@@ -5,6 +5,7 @@ export enum ParentCommand {
   InitFailed,
   InitCompleted,
   Log,
+  MoveToDelayed,
   Progress,
   Update,
 }

src/interfaces/sandboxed-job.ts

@@ -8,8 +8,9 @@ export interface SandboxedJob<T = any, R = any>
   extends Omit<JobJson, 'data' | 'opts' | 'progress' | 'returnValue'> {
   data: T;
   opts: JobsOptions;
-  updateProgress: (value: object | number) => Promise<void>;
+  moveToDelayed: (timestamp: number, token?: string) => Promise<void>;
   log: (row: any) => void;
   updateData: (data: any) => Promise<void>;
+  updateProgress: (value: object | number) => Promise<void>;
   returnValue: R;
 }

tests/fixtures/fixture_processor_move_to_delayed.js

@@ -0,0 +1,21 @@
+/**
+ * A processor file to be used in tests.
+ *
+ */
+'use strict';
+
+const { DelayedError } = require('../../dist/cjs/classes/delayed-error');
+const delay = require('./delay');
+
+module.exports = function (job, token) {
+  if (job.attemptsMade == 1) {
+    return delay(250)
+      .then(() => {
+        job.moveToDelayed(2500, token);
+        return delay(500);
+      })
+      .then(() => {
+        throw new DelayedError();
+      });
+  }
+};

tests/fixtures/fixture_processor_with_extra_param.js

@@ -6,7 +6,7 @@
 
 const delay = require('./delay');
 
-module.exports = function (job, extraParam) {
+module.exports = function (job, token, extraParam) {
   return delay(500).then(() => {
     return 42;
   });

tests/test_sandboxed_process.ts

@@ -73,7 +73,7 @@ function sandboxProcessTests(
       await worker.close();
     });
 
-    describe('when processor has more than 1 param', () => {
+    describe('when processor has more than 2 params', () => {
       it('should ignore extra params, process and complete', async () => {
         const processFile =
           __dirname + '/fixtures/fixture_processor_with_extra_param.js';
@@ -429,7 +429,8 @@ function sandboxProcessTests(
     });
 
     it('should process and update data', async () => {
-      const processFile = __dirname + '/fixtures/fixture_processor_update.js';
+      const processFile =
+        __dirname + '/fixtures/fixture_processor_update_data.js';
 
       const worker = new Worker(queueName, processFile, {
         connection,
@@ -459,6 +460,50 @@ function sandboxProcessTests(
       await worker.close();
     });
 
+    it('should process and move to delayed', async () => {
+      const processFile =
+        __dirname + '/fixtures/fixture_processor_move_to_delayed.js';
+
+      const worker = new Worker(queueName, processFile, {
+        connection,
+        drainDelay: 1,
+        useWorkerThreads,
+      });
+
+      const delaying = new Promise<void>((resolve, reject) => {
+        queueEvents.on('delayed', async ({ delay }) => {
+          try {
+            expect(Number(delay)).to.be.greaterThanOrEqual(2500);
+            expect(Object.keys(worker['childPool'].retained)).to.have.lengthOf(
+              1,
+            );
+            expect(worker['childPool'].getAllFree()).to.have.lengthOf(0);
+            resolve();
+          } catch (err) {
+            reject(err);
+          }
+        });
+      });
+
+      const completing = new Promise<void>((resolve, reject) => {
+        worker.on('completed', async (job: Job) => {
+          expect(job.data.bar).to.be.equal('foo');
+          resolve();
+        });
+      });
+
+      const job = await queue.add('test', { bar: 'foo' });
+
+      await delaying;
+
+      const state = await queue.getJobState(job.id!);
+
+      expect(state).to.be.equal('delayed');
+
+      await completing;
+      await worker.close();
+    });
+
     describe('when env variables are provided', () => {
       it('shares env variables', async () => {
         const processFile = __dirname + '/fixtures/fixture_processor_env.js';