Repository: taskforcesh/bullmq
Pull request #849 (2021-10-30 03:08:20)

fix(autorun): delete autorun option
BREAKING CHANGE: no longer support autorun option, use run method


Patch:
diff --git a/src/classes/queue-events.ts b/src/classes/queue-events.ts
index 658eb25536..0e83a473b1 100644
--- a/src/classes/queue-events.ts
+++ b/src/classes/queue-events.ts
@@ -179,7 +179,7 @@ export class QueueEvents extends QueueBase {
 
   constructor(
     name: string,
-    { connection, autorun = true, ...opts }: QueueEventsOptions = {},
+    { connection, ...opts }: QueueEventsOptions = {},
     Connection?: typeof RedisConnection,
   ) {
     super(
@@ -199,10 +199,6 @@ export class QueueEvents extends QueueBase {
       },
       this.opts,
     );
-
-    if (autorun) {
-      this.run().catch(error => this.emit('error', error));
-    }
   }
 
   async run(): Promise<void> {
diff --git a/src/classes/queue-scheduler.ts b/src/classes/queue-scheduler.ts
index e318748cdc..6eb8aa9b32 100644
--- a/src/classes/queue-scheduler.ts
+++ b/src/classes/queue-scheduler.ts
@@ -42,7 +42,7 @@ export class QueueScheduler
 
   constructor(
     name: string,
-    { connection, autorun = true, ...opts }: QueueSchedulerOptions = {},
+    { connection, ...opts }: QueueSchedulerOptions = {},
   ) {
     super(name, {
       maxStalledCount: 1,
@@ -57,12 +57,6 @@ export class QueueScheduler
     if (!(this.opts as QueueSchedulerOptions).stalledInterval) {
       throw new Error('Stalled interval cannot be zero or undefined');
     }
-
-    if (autorun) {
-      this.run().catch(error => {
-        console.error(error);
-      });
-    }
   }
 
   async run(): Promise<void> {
diff --git a/src/classes/worker.ts b/src/classes/worker.ts
index 86c56d55b3..1e4849ff15 100644
--- a/src/classes/worker.ts
+++ b/src/classes/worker.ts
@@ -142,7 +142,6 @@ export class Worker<
       concurrency: 1,
       lockDuration: 30000,
       runRetryDelay: 15000,
-      autorun: true,
       ...this.opts,
     };
 
@@ -185,10 +184,6 @@ export class Worker<
         ).bind(this);
       }
       this.timerManager = new TimerManager();
-
-      if (this.opts.autorun) {
-        this.run().catch(error => this.emit('error', error));
-      }
     }
 
     this.on('error', err => console.error(err));
diff --git a/src/interfaces/queue-options.ts b/src/interfaces/queue-options.ts
index 8bdb6d708c..9eaffdf736 100644
--- a/src/interfaces/queue-options.ts
+++ b/src/interfaces/queue-options.ts
@@ -63,10 +63,6 @@ export interface QueueOptions extends QueueBaseOptions {
  * Options for QueueEvents
  */
 export interface QueueEventsOptions extends QueueBaseOptions {
-  /**
-   * Condition to start listening to events at instance creation.
-   */
-  autorun?: boolean;
   /**
    * Last event Id. If provided it is possible to continue
    * consuming events from a known Id instead of from the last
diff --git a/src/interfaces/queue-scheduler-options.ts b/src/interfaces/queue-scheduler-options.ts
index 5b7f47d27c..bd9c4951f5 100644
--- a/src/interfaces/queue-scheduler-options.ts
+++ b/src/interfaces/queue-scheduler-options.ts
@@ -7,10 +7,6 @@ import { QueueBaseOptions } from '../interfaces';
  * @see {@link https://docs.bullmq.io/guide/queuescheduler}
  */
 export interface QueueSchedulerOptions extends QueueBaseOptions {
-  /**
-   * Condition to start scheduler at instance creation.
-   */
-  autorun?: boolean;
   /**
    * Amount of times a job can be recovered from a stalled state
    * to the `wait` state. If this is exceeded, the job is moved
diff --git a/src/interfaces/worker-options.ts b/src/interfaces/worker-options.ts
index 3647852f84..046c55135a 100644
--- a/src/interfaces/worker-options.ts
+++ b/src/interfaces/worker-options.ts
@@ -10,10 +10,6 @@ export type Processor<T = any, R = any, N extends string = string> = (
 ) => Promise<R>;
 
 export interface WorkerOptions extends QueueBaseOptions {
-  /**
-   * Condition to start processor at instance creation.
-   */
-  autorun?: boolean;
   /**
    * Amount of jobs that a single worker is allowed to work on
    * in parallel.


Test patch:
diff --git a/tests/test_bulk.ts b/tests/test_bulk.ts
index f1fa8a5dff..e2907b5879 100644
--- a/tests/test_bulk.ts
+++ b/tests/test_bulk.ts
@@ -38,6 +38,8 @@ describe('bulk jobs', () => {
     const worker = new Worker(queueName, processor, { connection });
     await worker.waitUntilReady();
 
+    worker.run();
+
     const jobs = await queue.addBulk([
       { name, data: { idx: 0, foo: 'bar' } },
       { name, data: { idx: 1, foo: 'baz' } },
@@ -58,11 +60,18 @@ describe('bulk jobs', () => {
     const parentQueueName = `parent-queue-${v4()}`;
     const parentQueue = new Queue(parentQueueName, { connection });
 
-    const parentWorker = new Worker(parentQueueName, null, { connection });
-    const childrenWorker = new Worker(queueName, null, { connection });
+    const parentWorker = new Worker(parentQueueName, async job => {}, {
+      connection,
+    });
+    const childrenWorker = new Worker(queueName, async job => {}, {
+      connection,
+    });
     await parentWorker.waitUntilReady();
     await childrenWorker.waitUntilReady();
 
+    parentWorker.run();
+    childrenWorker.run();
+
     const parent = await parentQueue.add('parent', { some: 'data' });
     const jobs = await queue.addBulk([
       {
@@ -123,6 +132,8 @@ describe('bulk jobs', () => {
     const worker = new Worker(queueName, processor, { connection });
     await worker.waitUntilReady();
 
+    worker.run();
+
     const jobs = await queue.addBulk([
       { name, data: { idx: 0, foo: 'bar' }, opts: { jobId: 'test1' } },
       { name, data: { idx: 1, foo: 'baz' }, opts: { jobId: 'test2' } },
diff --git a/tests/test_clean.ts b/tests/test_clean.ts
index c06e7e4117..b4ce18ad26 100644
--- a/tests/test_clean.ts
+++ b/tests/test_clean.ts
@@ -17,6 +17,7 @@ describe('Cleaner', () => {
     queue = new Queue(queueName, { connection });
     queueEvents = new QueueEvents(queueName, { connection });
     await queueEvents.waitUntilReady();
+    queueEvents.run();
   });
 
   afterEach(async function () {
@@ -50,6 +51,8 @@ describe('Cleaner', () => {
     const worker = new Worker(queueName, async job => {}, { connection });
     await worker.waitUntilReady();
 
+    worker.run();
+
     queue.on(
       'completed',
       after(2, async () => {
@@ -65,6 +68,8 @@ describe('Cleaner', () => {
     const worker = new Worker(queueName, async job => {}, { connection });
     await worker.waitUntilReady();
 
+    worker.run();
+
     await queue.add('test', { some: 'data' });
     await queue.add('test', { some: 'data' });
     await delay(200);
@@ -106,6 +111,8 @@ describe('Cleaner', () => {
     );
     await worker.waitUntilReady();
 
+    worker.run();
+
     await queue.add('test', { some: 'data' });
     await queue.add('test', { some: 'data' });
 
@@ -159,6 +166,8 @@ describe('Cleaner', () => {
     );
     await worker.waitUntilReady();
 
+    worker.run();
+
     const client = new IORedis();
 
     await queue.add('test', { some: 'data' });
diff --git a/tests/test_connection.ts b/tests/test_connection.ts
index 23fefe5d05..b6589c2049 100644
--- a/tests/test_connection.ts
+++ b/tests/test_connection.ts
@@ -62,6 +62,8 @@ describe('connection', () => {
       // error event has to be observed or the exception will bubble up
     });
 
+    worker.run();
+
     const workerClient = await worker.client;
     const queueClient = await queue.client;
 
@@ -125,6 +127,8 @@ describe('connection', () => {
       }
     });
 
+    worker.run();
+
     await queue.waitUntilReady();
     await queue.add('test', { foo: 'bar' });
 
diff --git a/tests/test_delay.ts b/tests/test_delay.ts
index c86caa14bd..e379bb2f1d 100644
--- a/tests/test_delay.ts
+++ b/tests/test_delay.ts
@@ -58,7 +58,6 @@ describe('Delayed jobs', function () {
           const delayedJobs = await queue.getDelayed();
           expect(delayedJobs.length).to.be.equal(0);
           expect(publishHappened).to.be.eql(true);
-          await worker.close();
           resolve();
         } catch (err) {
           reject(err);
@@ -66,6 +65,10 @@ describe('Delayed jobs', function () {
       });
     });
 
+    queueEvents.run();
+    queueScheduler.run();
+    worker.run();
+
     const job = await queue.add('test', { delayed: 'foobar' }, { delay });
 
     expect(job.id).to.be.ok;
@@ -79,86 +82,28 @@ describe('Delayed jobs', function () {
     await worker.close();
   });
 
-  describe('when autorun option is provided as false', function () {
-    it('should process a delayed job only after delayed time', async function () {
+  describe('when run method is called when queue-scheduler is running', function () {
+    it('throws an error', async function () {
       const delay = 1000;
-      const queueScheduler = new QueueScheduler(queueName, {
-        connection,
-        autorun: false,
-      });
+      const maxJobs = 10;
+      const queueScheduler = new QueueScheduler(queueName);
       await queueScheduler.waitUntilReady();
-      const queueEvents = new QueueEvents(queueName, { connection });
-      await queueEvents.waitUntilReady();
-
-      const worker = new Worker(queueName, async job => {}, { connection });
-
-      const timestamp = Date.now();
-      let publishHappened = false;
-
-      const delayed = new Promise<void>((resolve, reject) => {
-        queueEvents.on('delayed', () => {
-          publishHappened = true;
-          resolve();
-        });
-      });
 
-      const completed = new Promise<void>((resolve, reject) => {
-        queueEvents.on('completed', async function () {
-          try {
-            expect(Date.now() > timestamp + delay);
-            const jobs = await queue.getWaiting();
-            expect(jobs.length).to.be.equal(0);
-
-            const delayedJobs = await queue.getDelayed();
-            expect(delayedJobs.length).to.be.equal(0);
-            expect(publishHappened).to.be.eql(true);
-            resolve();
-          } catch (err) {
-            reject(err);
-          }
-        });
-      });
+      const worker = new Worker(queueName, async job => {});
+      const running = queueScheduler.run();
 
-      const job = await queue.add('test', { delayed: 'foobar' }, { delay });
+      for (let i = 1; i <= maxJobs; i++) {
+        await queue.add('test', { foo: 'bar', num: i }, { delay });
+      }
 
-      expect(job.id).to.be.ok;
-      expect(job.data.delayed).to.be.eql('foobar');
-      expect(job.opts.delay).to.be.eql(delay);
-
-      queueScheduler.run();
+      await expect(queueScheduler.run()).to.be.rejectedWith(
+        'Queue Scheduler is already running.',
+      );
 
-      await delayed;
-      await completed;
       await queueScheduler.close();
-      await queueEvents.close();
+      await expect(running).to.have.been.fulfilled;
       await worker.close();
     });
-
-    describe('when run method is called when queue-scheduler is running', function () {
-      it('throws an error', async function () {
-        const delay = 1000;
-        const maxJobs = 10;
-        const queueScheduler = new QueueScheduler(queueName, {
-          connection,
-          autorun: false,
-        });
-        await queueScheduler.waitUntilReady();
-
-        const worker = new Worker(queueName, async job => {}, { connection });
-        queueScheduler.run();
-
-        for (let i = 1; i <= maxJobs; i++) {
-          await queue.add('test', { foo: 'bar', num: i }, { delay });
-        }
-
-        await expect(queueScheduler.run()).to.be.rejectedWith(
-          'Queue Scheduler is already running.',
-        );
-
-        await queueScheduler.close();
-        await worker.close();
-      });
-    });
   });
 
   it('should process delayed jobs in correct order', async function () {
@@ -187,6 +132,9 @@ describe('Delayed jobs', function () {
 
     worker.on('failed', function (job, err) {});
 
+    queueScheduler.run();
+    worker.run();
+
     await Promise.all([
       queue.add('test', { order: 1 }, { delay: 100 }),
       queue.add('test', { order: 6 }, { delay: 600 }),
@@ -231,6 +179,7 @@ describe('Delayed jobs', function () {
                 connection,
               });
               await secondQueueScheduler.waitUntilReady();
+              secondQueueScheduler.run();
             }
 
             if (order === 4) {
@@ -245,6 +194,9 @@ describe('Delayed jobs', function () {
       );
     });
 
+    firstQueueScheduler.run();
+    worker.run();
+
     await Promise.all([
       queue.add('test', { order: 2 }, { delay: 500 }),
       queue.add('test', { order: 4 }, { delay: 1500 }),
@@ -255,7 +207,7 @@ describe('Delayed jobs', function () {
     await processing;
 
     await queue.close();
-    worker && (await worker.close());
+    await worker.close();
     secondQueueScheduler && (await secondQueueScheduler.close());
   });
 
@@ -288,6 +240,9 @@ describe('Delayed jobs', function () {
 
     worker.on('failed', function (job, err) {});
 
+    queueScheduler.run();
+    worker.run();
+
     const now = Date.now();
     const promises = [];
     let i = 1;
@@ -305,6 +260,7 @@ describe('Delayed jobs', function () {
     }
     await Promise.all(promises);
     await processing;
+
     await queueScheduler.close();
     await worker.close();
   });
diff --git a/tests/test_events.ts b/tests/test_events.ts
index 838e744e73..6f129f10e3 100644
--- a/tests/test_events.ts
+++ b/tests/test_events.ts
@@ -27,59 +27,33 @@ describe('events', function () {
     await removeAllQueueData(new IORedis(), queueName);
   });
 
-  describe('when autorun option is provided as false', function () {
-    it('emits waiting when a job has been added', async () => {
-      const queueName2 = `test-${v4()}`;
-      const queue2 = new Queue(queueName2, { connection });
-      const queueEvents2 = new QueueEvents(queueName2, {
-        autorun: false,
-        connection,
-      });
-      await queueEvents2.waitUntilReady();
+  it('emits waiting when a job has been added', async () => {
+    const waiting = new Promise(resolve => {
+      queue.on('waiting', resolve);
+    });
 
-      const waiting = new Promise(resolve => {
-        queue2.on('waiting', resolve);
-      });
+    await queue.add('test', { foo: 'bar' });
 
-      const running = queueEvents2.run();
+    await waiting;
+  });
 
-      await queue2.add('test', { foo: 'bar' });
+  describe('when run method is called when queueEvent is running', function () {
+    it('throws error', async function () {
+      const queueEvents2 = new QueueEvents(queueName, { connection });
+      await queueEvents2.waitUntilReady();
+      const running = queueEvents2.run();
 
-      await waiting;
+      await expect(queueEvents2.run()).to.be.rejectedWith(
+        'Queue Events is already running.',
+      );
 
-      await queue2.close();
       await queueEvents2.close();
-      await expect(running).to.have.been.fulfilled;
-      await removeAllQueueData(new IORedis(), queueName2);
-    });
-
-    describe('when run method is called when queueEvent is running', function () {
-      it('throws error', async () => {
-        const queueName2 = `test-${v4()}`;
-        const queue2 = new Queue(queueName2, { connection });
-        const queueEvents2 = new QueueEvents(queueName2, {
-          autorun: false,
-          connection,
-        });
-        await queueEvents2.waitUntilReady();
 
-        const running = queueEvents2.run();
-
-        await queue2.add('test', { foo: 'bar' });
-
-        await expect(queueEvents2.run()).to.be.rejectedWith(
-          'Queue Events is already running.',
-        );
-
-        await queue2.close();
-        await queueEvents2.close();
-        await expect(running).to.have.been.fulfilled;
-        await removeAllQueueData(new IORedis(), queueName2);
-      });
+      await expect(running).to.have.been.fulfilled;
     });
   });
 
-  it('should emit waiting when a job has been added', async function () {
+  it('should emit waiting event when a job has been added', async function () {
     const waiting = new Promise<void>(resolve => {
       queue.on('waiting', job => {
         expect(job.id).to.be.string;
@@ -87,6 +61,8 @@ describe('events', function () {
       });
     });
 
+    queueEvents.run();
+
     await queue.add('test', { foo: 'bar' });
 
     await waiting;
@@ -97,6 +73,8 @@ describe('events', function () {
       queueEvents.on('waiting', resolve);
     });
 
+    queueEvents.run();
+
     await queue.add('test', { foo: 'bar' });
 
     await waiting;
@@ -116,7 +94,9 @@ describe('events', function () {
       await queue.add('test', { foo: 'bar' });
     }
 
-    await new Promise<void>(resolve => {
+    worker.run();
+
+    const cleaned = new Promise<void>(resolve => {
       queueEvents.once('cleaned', async ({ count }) => {
         expect(count).to.be.eql('50');
         const actualCount = await queue.count();
@@ -124,6 +104,12 @@ describe('events', function () {
         resolve();
       });
     });
+
+    queueEvents.run();
+
+    await cleaned;
+
+    await worker.close();
   });
 
   it('emits drained global event when all jobs have been processed', async function () {
@@ -139,6 +125,9 @@ describe('events', function () {
       });
     });
 
+    worker.run();
+    queueEvents.run();
+
     await queue.add('test', { foo: 'bar' });
     await queue.add('test', { foo: 'baz' });
 
@@ -160,6 +149,9 @@ describe('events', function () {
       worker.once('drained', resolve);
     });
 
+    worker.run();
+    queueEvents.run();
+
     await queue.add('test', { foo: 'bar' });
     await queue.add('test', { foo: 'baz' });
 
@@ -186,6 +178,9 @@ describe('events', function () {
       worker.once('error', resolve);
     });
 
+    worker.run();
+    queueEvents.run();
+
     await queue.add('test', { foo: 'bar' });
 
     await error;
@@ -211,6 +206,9 @@ describe('events', function () {
       });
     });
 
+    worker.run();
+    queueEvents.run();
+
     await queue.add(testName, { foo: 'bar' });
 
     await added;
@@ -221,8 +219,6 @@ describe('events', function () {
   it('should emit an event when a job becomes active', async () => {
     const worker = new Worker(queueName, async job => {}, { connection });
 
-    await queue.add('test', {});
-
     const completed = new Promise<void>(resolve => {
       worker.once('active', function () {
         worker.once('completed', async function () {
@@ -232,6 +228,11 @@ describe('events', function () {
       });
     });
 
+    worker.run();
+    queueEvents.run();
+
+    await queue.add('test', {});
+
     await completed;
     await worker.close();
   });
@@ -254,6 +255,9 @@ describe('events', function () {
       });
     });
 
+    worker.run();
+    queueEvents.run();
+
     const flow = new FlowProducer({ connection });
     await flow.add({
       name,
@@ -296,6 +300,9 @@ describe('events', function () {
       });
     });
 
+    worker.run();
+    queueEvents.run();
+
     await queue.add('test', {});
 
     await completed;
@@ -325,6 +332,9 @@ describe('events', function () {
       queueEvents.on('drained', resolve);
     });
 
+    worker.run();
+    queueEvents.run();
+
     await waitForCompletion;
     await worker.close();
 
@@ -350,6 +360,8 @@ describe('events', function () {
     const queueName = 'test-manual-' + v4();
     const trimmedQueue = new Queue(queueName, { connection });
 
+    queueEvents.run();
+
     await trimmedQueue.add('test', {});
     await trimmedQueue.add('test', {});
     await trimmedQueue.add('test', {});
diff --git a/tests/test_flow.ts b/tests/test_flow.ts
index cf4ae3eeb7..536e2cac1c 100644
--- a/tests/test_flow.ts
+++ b/tests/test_flow.ts
@@ -86,6 +86,9 @@ describe('flows', () => {
     await parentWorker.waitUntilReady();
     await childrenWorker.waitUntilReady();
 
+    parentWorker.run();
+    childrenWorker.run();
+
     const flow = new FlowProducer({ connection });
     const tree = await flow.add({
       name: 'parent-job',
@@ -164,6 +167,8 @@ describe('flows', () => {
       const queueScheduler = new QueueScheduler(queueName, { connection });
       await queueScheduler.waitUntilReady();
 
+      queueScheduler.run();
+
       const parentWorker = new Worker(parentQueueName, parentProcessor, {
         connection,
       });
@@ -180,6 +185,9 @@ describe('flows', () => {
       await parentWorker.waitUntilReady();
       await childrenWorker.waitUntilReady();
 
+      parentWorker.run();
+      childrenWorker.run();
+
       const flow = new FlowProducer({ connection });
       const tree = await flow.add({
         name: 'parent-job',
@@ -238,6 +246,8 @@ describe('flows', () => {
         });
       });
 
+      worker.run();
+
       const flow = new FlowProducer({ connection });
       await flow.add({
         queueName,
@@ -370,6 +380,9 @@ describe('flows', () => {
       await parentWorker.waitUntilReady();
       await childrenWorker.waitUntilReady();
 
+      parentWorker.run();
+      childrenWorker.run();
+
       const flow = new FlowProducer({ prefix: '{bull}', connection });
       const tree = await flow.add({
         name: 'parent-job',
@@ -479,6 +492,10 @@ describe('flows', () => {
       await childrenWorker.waitUntilReady();
       await grandChildrenWorker.waitUntilReady();
 
+      parentWorker.run();
+      childrenWorker.run();
+      grandChildrenWorker.run();
+
       const flow = new FlowProducer({ connection });
       const tree = await flow.add({
         name: 'parent-job',
@@ -561,6 +578,9 @@ describe('flows', () => {
     await queueScheduler.waitUntilReady();
     await queueEvents.waitUntilReady();
 
+    queueScheduler.run();
+    queueEvents.run();
+
     const name = 'child-job';
 
     const parentQueueName = `parent-queue-${v4()}`;
@@ -591,6 +611,9 @@ describe('flows', () => {
     await parentWorker.waitUntilReady();
     await childrenWorker.waitUntilReady();
 
+    parentWorker.run();
+    childrenWorker.run();
+
     const completed: { [index: string]: number[] } = {};
 
     const running = new Promise<void>((resolve, reject) => {
@@ -726,6 +749,9 @@ describe('flows', () => {
     await parentWorker.waitUntilReady();
     await childrenWorker.waitUntilReady();
 
+    parentWorker.run();
+    childrenWorker.run();
+
     const otherValues = Array.from(Array(72).keys()).map(() => ({
       name,
       data: { bar: 'something' },
@@ -959,6 +985,9 @@ describe('flows', () => {
       });
     });
 
+    parentWorker.run();
+    childrenWorker.run();
+
     const flow = new FlowProducer({ connection });
     const tree = await flow.add({
       name: 'parent-job',
@@ -1012,6 +1041,8 @@ describe('flows', () => {
       connection,
     });
 
+    parentWorker.run();
+
     const flow = new FlowProducer({ connection });
     const tree = await flow.add({
       name: 'parent-job',
@@ -1086,6 +1117,9 @@ describe('flows', () => {
       connection,
     });
 
+    parentWorker.run();
+    childrenWorker.run();
+
     const flow = new FlowProducer({ connection });
     const tree = await flow.add({
       name: 'parent-job',
@@ -1206,6 +1240,9 @@ describe('flows', () => {
       connection,
     });
 
+    parentWorker.run();
+    childrenWorker.run();
+
     const flow = new FlowProducer({ connection });
     const tree = await flow.add({
       name: 'root-job',
@@ -1276,6 +1313,8 @@ describe('flows', () => {
       connection,
     });
 
+    childrenWorker.run();
+
     const flow = new FlowProducer({ connection });
     const tree = await flow.add({
       name: 'parent-job',
@@ -1333,6 +1372,9 @@ describe('flows', () => {
       connection,
     });
 
+    parentWorker.run();
+    childrenWorker.run();
+
     const parentQueue = new Queue(parentQueueName, { connection });
     await parentQueue.pause();
 
@@ -1482,6 +1524,9 @@ describe('flows', () => {
       connection,
     });
 
+    parentWorker.run();
+    childrenWorker.run();
+
     await processingChildren;
     await childrenWorker.close();
 
diff --git a/tests/test_getters.ts b/tests/test_getters.ts
index 10ba91f30d..45135e5355 100644
--- a/tests/test_getters.ts
+++ b/tests/test_getters.ts
@@ -86,6 +86,8 @@ describe('Jobs getters', function () {
     });
     const worker = new Worker(queueName, processor, { connection });
 
+    worker.run();
+
     await queue.add('test', { foo: 'bar' });
     await processing;
 
@@ -125,6 +127,8 @@ describe('Jobs getters', function () {
       });
     });
 
+    worker.run();
+
     await queue.add('test', { foo: 'bar' });
     await queue.add('test', { baz: 'qux' });
 
@@ -156,6 +160,8 @@ describe('Jobs getters', function () {
       });
     });
 
+    worker.run();
+
     await queue.add('test', { foo: 'bar' });
     await queue.add('test', { baz: 'qux' });
 
@@ -190,6 +196,8 @@ describe('Jobs getters', function () {
       );
     });
 
+    worker.run();
+
     await Promise.all([
       queue.add('test', { foo: 'bar' }),
       queue.add('test', { baz: 'qux' }),
@@ -251,6 +259,8 @@ describe('Jobs getters', function () {
       }),
     );
 
+    worker.run();
+
     queue.add('test', { foo: 1 });
     queue.add('test', { foo: 2 });
     queue.add('test', { foo: 3 });
@@ -287,6 +297,8 @@ describe('Jobs getters', function () {
       }),
     );
 
+    worker.run();
+
     queue.add('test', { foo: 1 });
     queue.add('test', { foo: 2 });
     queue.add('test', { foo: 3 });
@@ -315,6 +327,8 @@ describe('Jobs getters', function () {
       }),
     );
 
+    worker.run();
+
     queue.add('test', { foo: 1 });
     queue.add('test', { foo: 2 });
     queue.add('test', { foo: 3 });
@@ -340,6 +354,8 @@ describe('Jobs getters', function () {
       }),
     );
 
+    worker.run();
+
     queue.add('test', { foo: 1 });
     queue.add('test', { foo: 2 });
     queue.add('test', { foo: 3 });
@@ -365,6 +381,8 @@ describe('Jobs getters', function () {
       }),
     );
 
+    worker.run();
+
     queue.add('test', { foo: 1 });
     queue.add('test', { foo: 2 });
     queue.add('test', { foo: 3 });
@@ -399,6 +417,8 @@ describe('Jobs getters', function () {
       }),
     );
 
+    worker.run();
+
     queue.add('test', { foo: 1 });
     queue.add('test', { foo: 2 });
   });
diff --git a/tests/test_job.ts b/tests/test_job.ts
index 049380586a..aee80ad75c 100644
--- a/tests/test_job.ts
+++ b/tests/test_job.ts
@@ -158,6 +158,8 @@ describe('Job', function () {
         const testJob = await newQueue.add('test', 0);
       });
 
+      worker.run();
+
       try {
         await promise;
       } finally {
@@ -410,6 +412,8 @@ describe('Job', function () {
       const queueEvents = new QueueEvents(queueName, { connection });
       await queueEvents.waitUntilReady();
 
+      queueEvents.run();
+
       const job = await Job.create(
         queue,
         'test',
@@ -516,9 +520,13 @@ describe('Job', function () {
       const queueScheduler = new QueueScheduler(queueName, { connection });
       await queueScheduler.waitUntilReady();
 
+      queueScheduler.run();
+
       const worker = new Worker(queueName, async job => {}, { connection });
       await worker.waitUntilReady();
 
+      worker.run();
+
       const startTime = new Date().getTime();
 
       const completing = new Promise<void>((resolve, reject) => {
@@ -580,10 +588,12 @@ describe('Job', function () {
     });
 
     it('should process a promoted job according to its priority', async function () {
+      this.timeout(10000);
       const queueScheduler = new QueueScheduler(queueName, { connection });
       await queueScheduler.waitUntilReady();
 
-      this.timeout(10000);
+      queueScheduler.run();
+
       const worker = new Worker(
         queueName,
         job => {
@@ -609,6 +619,8 @@ describe('Job', function () {
         worker.on('active', after(2, resolve)),
       );
 
+      worker.run();
+
       const add = (jobId: string, ms = 0) =>
         queue.add('test', {}, { jobId, delay: ms, priority: 1 });
 
@@ -840,6 +852,7 @@ describe('Job', function () {
     beforeEach(async function () {
       queueEvents = new QueueEvents(queueName, { connection });
       await queueEvents.waitUntilReady();
+      queueEvents.run();
     });
 
     afterEach(async function () {
@@ -849,6 +862,8 @@ describe('Job', function () {
     it('should resolve when the job has been completed', async function () {
       const worker = new Worker(queueName, async job => 'qux', { connection });
 
+      worker.run();
+
       const job = await queue.add('test', { foo: 'bar' });
 
       const result = await job.waitUntilFinished(queueEvents);
@@ -879,6 +894,8 @@ describe('Job', function () {
           });
         });
 
+        worker.run();
+
         const job = await queue.add(
           'test',
           { foo: 'bar' },
@@ -902,6 +919,8 @@ describe('Job', function () {
         { connection },
       );
 
+      worker.run();
+
       const job = await queue.add('test', { foo: 'bar' });
 
       const result = await job.waitUntilFinished(queueEvents);
@@ -922,6 +941,8 @@ describe('Job', function () {
         { connection },
       );
 
+      worker.run();
+
       const job = await queue.add('test', { foo: 'bar' });
       await delay(600);
 
@@ -937,6 +958,8 @@ describe('Job', function () {
         connection,
       });
 
+      worker.run();
+
       const job = await queue.add('test', { foo: 'bar' });
 
       const result = await job.waitUntilFinished(queueEvents);
@@ -956,6 +979,9 @@ describe('Job', function () {
         },
         { connection },
       );
+      await worker.waitUntilReady();
+
+      worker.run();
 
       const job = await queue.add('test', { foo: 'bar' });
 
@@ -973,6 +999,8 @@ describe('Job', function () {
         { connection },
       );
 
+      worker.run();
+
       const job = await queue.add('test', { foo: 'bar' });
 
       await delay(500);
@@ -993,6 +1021,8 @@ describe('Job', function () {
         { connection },
       );
 
+      worker.run();
+
       const job = await queue.add('test', { foo: 'bar' });
 
       await delay(500);
diff --git a/tests/test_obliterate.ts b/tests/test_obliterate.ts
index 56534295f4..6adfc3a7fe 100644
--- a/tests/test_obliterate.ts
+++ b/tests/test_obliterate.ts
@@ -17,6 +17,7 @@ describe('Obliterate', function () {
     queue = new Queue(queueName, { connection });
     queueEvents = new QueueEvents(queueName, { connection });
     await queueEvents.waitUntilReady();
+    queueEvents.run();
   });
 
   afterEach(async function () {
@@ -58,6 +59,8 @@ describe('Obliterate', function () {
     );
     await worker.waitUntilReady();
 
+    worker.run();
+
     await job.waitUntilFinished(queueEvents);
 
     await queue.obliterate();
@@ -95,6 +98,8 @@ describe('Obliterate', function () {
         worker.on('failed', resolve);
       });
 
+      worker.run();
+
       const flow = new FlowProducer({ connection });
       await flow.add({
         name: 'parent-job',
@@ -143,6 +148,8 @@ describe('Obliterate', function () {
     );
     await worker.waitUntilReady();
 
+    worker.run();
+
     await job.waitUntilFinished(queueEvents);
 
     await expect(queue.obliterate()).to.be.rejectedWith(
@@ -178,6 +185,8 @@ describe('Obliterate', function () {
     );
     await worker.waitUntilReady();
 
+    worker.run();
+
     await job.waitUntilFinished(queueEvents);
     await queue.obliterate({ force: true });
     const client = await queue.client;
@@ -223,6 +232,9 @@ describe('Obliterate', function () {
     );
     await worker.waitUntilReady();
 
+    queueEvents.run();
+    worker.run();
+
     await job.waitUntilFinished(queueEvents);
 
     await queue.obliterate({ force: true });
@@ -252,6 +264,8 @@ describe('Obliterate', function () {
     );
     await worker.waitUntilReady();
 
+    worker.run();
+
     await lastCompletedJob.waitUntilFinished(queueEvents);
 
     fail = true;
diff --git a/tests/test_pause.ts b/tests/test_pause.ts
index 75ea6f14d2..3bdcf7c9c1 100644
--- a/tests/test_pause.ts
+++ b/tests/test_pause.ts
@@ -23,6 +23,7 @@ describe('Pause', function () {
     queue = new Queue(queueName, { connection });
     queueEvents = new QueueEvents(queueName, { connection });
     await queueEvents.waitUntilReady();
+    queueEvents.run();
   });
 
   afterEach(async function () {
@@ -38,6 +39,8 @@ describe('Pause', function () {
     const queueScheduler = new QueueScheduler(queueName);
     await queueScheduler.waitUntilReady();
 
+    queueScheduler.run();
+
     let processed = false;
 
     const worker = new Worker(
@@ -49,6 +52,8 @@ describe('Pause', function () {
     );
     await worker.waitUntilReady();
 
+    worker.run();
+
     await queue.pause();
     await queue.add('test', {}, { delay: 200 });
     const counts = await queue.getJobCounts('waiting', 'delayed');
@@ -87,6 +92,8 @@ describe('Pause', function () {
     const worker = new Worker(queueName, process, { connection });
     await worker.waitUntilReady();
 
+    worker.run();
+
     await queue.pause();
     isPaused = true;
     await queue.add('test', { foo: 'paused' });
@@ -126,7 +133,7 @@ describe('Pause', function () {
       };
     });
 
-    new Worker(queueName, process, { connection });
+    const worker = new Worker(queueName, process, { connection });
 
     queueEvents.on('paused', async (args, eventId) => {
       isPaused = false;
@@ -141,6 +148,8 @@ describe('Pause', function () {
       isResumed = true;
     });
 
+    worker.run();
+
     await queue.add('test', { foo: 'paused' });
     await queue.add('test', { foo: 'paused' });
 
@@ -165,6 +174,8 @@ describe('Pause', function () {
     worker = new Worker(queueName, process, { connection });
     await worker.waitUntilReady();
 
+    worker.run();
+
     await worker.pause();
 
     // Add the worker after the queue is in paused mode since the normal behavior is to pause
@@ -195,6 +206,8 @@ describe('Pause', function () {
     const worker = new Worker(queueName, process, { connection });
     await worker.waitUntilReady();
 
+    worker.run();
+
     const jobs: Promise<Job | void>[] = [];
     for (let i = 0; i < 10; i++) {
       jobs.push(queue.add('test', i));
@@ -249,6 +262,9 @@ describe('Pause', function () {
     const worker2 = new Worker(queueName, process2, { connection });
     await worker2.waitUntilReady();
 
+    worker1.run();
+    worker2.run();
+
     await Promise.all([
       queue.add('test', 1),
       queue.add('test', 2),
@@ -280,6 +296,8 @@ describe('Pause', function () {
     const worker = new Worker(queueName, process, { connection });
     await worker.waitUntilReady();
 
+    worker.run();
+
     await queue.add('test', 1);
     await startProcessing;
     await worker.pause();
@@ -299,6 +317,8 @@ describe('Pause', function () {
     });
     await worker.waitUntilReady();
 
+    worker.run();
+
     await queue.add('test', {});
 
     return new Promise((resolve, reject) => {
diff --git a/tests/test_queue.ts b/tests/test_queue.ts
index 4f9c0b6390..c4bd6a05c7 100644
--- a/tests/test_queue.ts
+++ b/tests/test_queue.ts
@@ -122,6 +122,8 @@ describe('queues', function () {
         const queueScheduler = new QueueScheduler(queueName, { connection });
         await queueScheduler.waitUntilReady();
 
+        queueScheduler.run();
+
         for (let i = 1; i <= maxJobs; i++) {
           added.push(queue.add('test', { foo: 'bar', num: i }));
         }
@@ -153,6 +155,8 @@ describe('queues', function () {
         const queueScheduler = new QueueScheduler(queueName, { connection });
         await queueScheduler.waitUntilReady();
 
+        queueScheduler.run();
+
         for (let i = 1; i <= maxJobs; i++) {
           added.push(queue.add('test', { foo: 'bar', num: i }));
         }
diff --git a/tests/test_rate_limiter.ts b/tests/test_rate_limiter.ts
index e1e6562e10..cd508ac343 100644
--- a/tests/test_rate_limiter.ts
+++ b/tests/test_rate_limiter.ts
@@ -18,6 +18,7 @@ describe('Rate Limiter', function () {
     queue = new Queue(queueName, { connection });
     queueEvents = new QueueEvents(queueName, { connection });
     await queueEvents.waitUntilReady();
+    queueEvents.run();
   });
 
   afterEach(async function () {
@@ -36,6 +37,8 @@ describe('Rate Limiter', function () {
     });
     await worker.waitUntilReady();
 
+    worker.run();
+
     queueEvents.on('failed', ({ failedReason }) => {});
 
     await Promise.all([
@@ -98,6 +101,9 @@ describe('Rate Limiter', function () {
       });
     });
 
+    queueScheduler.run();
+    worker.run();
+
     for (let i = 0; i < numJobs; i++) {
       await queue.add('rate test', {});
     }
@@ -148,6 +154,9 @@ describe('Rate Limiter', function () {
       });
     });
 
+    queueScheduler.run();
+    worker.run();
+
     for (let i = 0; i < numJobs; i++) {
       await queue.add('rate test', {});
     }
@@ -222,6 +231,9 @@ describe('Rate Limiter', function () {
       });
     });
 
+    queueScheduler.run();
+    worker.run();
+
     for (let i = 0; i < numJobs; i++) {
       await rateLimitedQueue.add('rate test', { accountId: i % numGroups });
     }
@@ -297,6 +309,9 @@ describe('Rate Limiter', function () {
       });
     });
 
+    queueScheduler.run();
+    worker.run();
+
     for (let i = 0; i < numJobs; i++) {
       await rateLimitedQueue.add('rate test', {});
     }
@@ -384,6 +399,9 @@ describe('Rate Limiter', function () {
       );
     });
 
+    queueScheduler.run();
+    worker.run();
+
     await result;
     await worker.close();
     await queueScheduler.close();
diff --git a/tests/test_repeat.ts b/tests/test_repeat.ts
index 01f8d701bc..1fa4f57b58 100644
--- a/tests/test_repeat.ts
+++ b/tests/test_repeat.ts
@@ -42,6 +42,7 @@ describe('repeat', function () {
     repeat = new Repeat(queueName, { connection });
     queueEvents = new QueueEvents(queueName, { connection });
     await queueEvents.waitUntilReady();
+    queueEvents.run();
   });
 
   afterEach(async function () {
@@ -73,6 +74,9 @@ describe('repeat', function () {
       });
     });
 
+    queueScheduler.run();
+    worker.run();
+
     await queue.add(
       'test',
       { foo: 'bar' },
@@ -194,6 +198,9 @@ describe('repeat', function () {
     const worker = new Worker(queueName, async job => {}, { connection });
     const delayStub = sinon.stub(worker, 'delay').callsFake(async () => {});
 
+    queueScheduler.run();
+    worker.run();
+
     const date = new Date('2017-02-07 9:24:00');
     this.clock.setSystemTime(date);
 
@@ -244,6 +251,9 @@ describe('repeat', function () {
     const worker = new Worker(queueName, async job => {}, { connection });
     const delayStub = sinon.stub(worker, 'delay').callsFake(async () => {});
 
+    queueScheduler.run();
+    worker.run();
+
     await queue.add(
       'test',
       { foo: 'bar' },
@@ -295,6 +305,9 @@ describe('repeat', function () {
     const worker = new Worker(queueName, async job => {}, { connection });
     const delayStub = sinon.stub(worker, 'delay').callsFake(async () => {});
 
+    queueScheduler.run();
+    worker.run();
+
     await queue.add(
       'repeat',
       { foo: 'bar' },
@@ -353,6 +366,9 @@ describe('repeat', function () {
     const worker = new Worker(queueName, async job => {}, { connection });
     const delayStub = sinon.stub(worker, 'delay').callsFake(async () => {});
 
+    queueScheduler.run();
+    worker.run();
+
     await queue.add(
       'test',
       { foo: 'bar' },
@@ -406,6 +422,9 @@ describe('repeat', function () {
     const worker = new Worker(queueName, async () => {}, { connection });
     const delayStub = sinon.stub(worker, 'delay').callsFake(async () => {});
 
+    queueScheduler.run();
+    worker.run();
+
     await queue.add(
       'repeat',
       { foo: 'bar' },
@@ -460,6 +479,9 @@ describe('repeat', function () {
     const worker = new Worker(queueName, async job => {}, { connection });
     const delayStub = sinon.stub(worker, 'delay').callsFake(async () => {});
 
+    queueScheduler.run();
+    worker.run();
+
     await queue.add(
       'repeat',
       { foo: 'bar' },
@@ -497,7 +519,7 @@ describe('repeat', function () {
   });
 
   it('should repeat 7:th day every month at 9:25', async function () {
-    this.timeout(200000);
+    this.timeout(220000);
     const queueScheduler = new QueueScheduler(queueName, { connection });
     await queueScheduler.waitUntilReady();
 
@@ -505,8 +527,12 @@ describe('repeat', function () {
     this.clock.setSystemTime(date);
 
     const worker = new Worker(queueName, async job => {}, { connection });
+    await worker.waitUntilReady();
     const delayStub = sinon.stub(worker, 'delay').callsFake(async () => {});
 
+    queueScheduler.run();
+    worker.run();
+
     const nextTick = () => {
       const now = moment();
       const nextMonth = moment().add(1, 'months');
@@ -605,6 +631,9 @@ describe('repeat', function () {
     const worker = new Worker(queueName, processor, { connection });
     const delayStub = sinon.stub(worker, 'delay').callsFake(async () => {});
 
+    queueScheduler.run();
+    worker.run();
+
     await queue.add('remove', { foo: 'bar' }, { repeat });
     this.clock.tick(nextTick);
 
@@ -686,6 +715,9 @@ describe('repeat', function () {
       prev = job;
     });
 
+    queueScheduler.run();
+    worker.run();
+
     await processing;
     await queueScheduler.close();
     delayStub.restore();
@@ -733,6 +765,9 @@ describe('repeat', function () {
       });
     });
 
+    queueScheduler.run();
+    worker.run();
+
     await queue.add('test', { foo: 'bar' }, { repeat: repeatOpts, jobId });
 
     this.clock.tick(nextTick);
@@ -759,6 +794,9 @@ describe('repeat', function () {
     await worker.waitUntilReady();
     const delayStub = sinon.stub(worker, 'delay').callsFake(async () => {});
 
+    queueScheduler.run();
+    worker.run();
+
     await queue.add(
       'myTestJob',
       {
@@ -797,6 +835,9 @@ describe('repeat', function () {
     const worker = new Worker(queueName, NoopProc, { connection });
     const delayStub = sinon.stub(worker, 'delay').callsFake(async () => {});
 
+    queueScheduler.run();
+    worker.run();
+
     await queue.add(
       'repeat',
       { foo: 'bar' },
@@ -859,6 +900,9 @@ describe('repeat', function () {
     const worker = new Worker(queueName, processor, { connection });
     await worker.waitUntilReady();
 
+    queueScheduler.run();
+    worker.run();
+
     await processing;
 
     await worker.close();
@@ -884,6 +928,9 @@ describe('repeat', function () {
     const delayStub = sinon.stub(worker, 'delay').callsFake(async () => {});
     await worker.waitUntilReady();
 
+    queueScheduler.run();
+    worker.run();
+
     let prevType: string;
     let counter = 0;
 
@@ -945,6 +992,9 @@ describe('repeat', function () {
       });
     });
 
+    queueScheduler.run();
+    worker.run();
+
     await queue.add(
       'test',
       { foo: 'bar' },
@@ -978,6 +1028,9 @@ describe('repeat', function () {
 
     const worker = new Worker(queueName, processor, { connection });
 
+    queueScheduler.run();
+    worker.run();
+
     await processing;
     await worker.close();
     await queueScheduler.close();
diff --git a/tests/test_sandboxed_process.ts b/tests/test_sandboxed_process.ts
index 76f943b681..a71a287abe 100644
--- a/tests/test_sandboxed_process.ts
+++ b/tests/test_sandboxed_process.ts
@@ -25,6 +25,7 @@ describe('sandboxed process', () => {
     queue = new Queue(queueName, { connection });
     queueEvents = new QueueEvents(queueName, { connection });
     await queueEvents.waitUntilReady();
+    queueEvents.run();
   });
 
   afterEach(async function () {
@@ -57,6 +58,8 @@ describe('sandboxed process', () => {
       });
     });
 
+    worker.run();
+
     await queue.add('test', { foo: 'bar' });
 
     await completing;
@@ -84,6 +87,8 @@ describe('sandboxed process', () => {
       });
       const inspect = stdout.inspect();
 
+      worker.run();
+
       await queue.add('test', { foo: 'bar' });
 
       let output = '';
@@ -120,6 +125,8 @@ describe('sandboxed process', () => {
       });
       const inspect = stderr.inspect();
 
+      worker.run();
+
       await queue.add('test', { foo: 'bar' });
 
       let output = '';
@@ -159,6 +166,8 @@ describe('sandboxed process', () => {
       });
     });
 
+    worker.run();
+
     await queue.add('foobar', { foo: 'bar' });
 
     await completing;
@@ -202,6 +211,8 @@ describe('sandboxed process', () => {
       });
     });
 
+    worker.run();
+
     await completing;
     await worker.close();
   });
@@ -245,6 +256,8 @@ describe('sandboxed process', () => {
       });
     });
 
+    worker.run();
+
     await completing;
   });
 
@@ -278,6 +291,8 @@ describe('sandboxed process', () => {
       progresses.push(progress);
     });
 
+    worker.run();
+
     await queue.add('test', { foo: 'bar' });
 
     await completing;
@@ -311,6 +326,8 @@ describe('sandboxed process', () => {
         });
       });
 
+      worker.run();
+
       await queue.add('test', { foo: 'bar' });
 
       await completing;
@@ -345,6 +362,8 @@ describe('sandboxed process', () => {
       });
     });
 
+    worker.run();
+
     await queue.add('test', { foo: 'bar' });
 
     await failing;
@@ -357,7 +376,8 @@ describe('sandboxed process', () => {
     let didThrow = false;
     try {
       const missingProcessFile = __dirname + '/fixtures/missing_processor.js';
-      worker = new Worker(queueName, missingProcessFile, { connection });
+      worker = new Worker(queueName, missingProcessFile, {});
+      worker.run();
     } catch (err) {
       didThrow = true;
     }
@@ -372,11 +392,13 @@ describe('sandboxed process', () => {
   it('should fail if the process crashes', async () => {
     const processFile = __dirname + '/fixtures/fixture_processor_crash.js';
 
-    new Worker(queueName, processFile, {
+    const worker = new Worker(queueName, processFile, {
       connection,
       drainDelay: 1,
     });
 
+    worker.run();
+
     const job = await queue.add('test', {});
 
     await expect(job.waitUntilFinished(queueEvents)).to.be.rejectedWith(
@@ -387,11 +409,13 @@ describe('sandboxed process', () => {
   it('should fail if the process exits 0', async () => {
     const processFile = __dirname + '/fixtures/fixture_processor_crash.js';
 
-    new Worker(queueName, processFile, {
+    const worker = new Worker(queueName, processFile, {
       connection,
       drainDelay: 1,
     });
 
+    worker.run();
+
     const job = await queue.add('test', { exitCode: 0 });
 
     await expect(job.waitUntilFinished(queueEvents)).to.be.rejectedWith(
@@ -402,11 +426,13 @@ describe('sandboxed process', () => {
   it('should fail if the process exits non-0', async () => {
     const processFile = __dirname + '/fixtures/fixture_processor_crash.js';
 
-    new Worker(queueName, processFile, {
+    const worker = new Worker(queueName, processFile, {
       connection,
       drainDelay: 1,
     });
 
+    worker.run();
+
     const job = await queue.add('test', { exitCode: 1 });
 
     await expect(job.waitUntilFinished(queueEvents)).to.be.rejectedWith(
@@ -417,11 +443,13 @@ describe('sandboxed process', () => {
   it('should fail if the process file is broken', async () => {
     const processFile = __dirname + '/fixtures/fixture_processor_broken.js';
 
-    new Worker(queueName, processFile, {
+    const worker = new Worker(queueName, processFile, {
       connection,
       drainDelay: 1,
     });
 
+    worker.run();
+
     const job = await queue.add('test', { exitCode: 1 });
 
     await expect(job.waitUntilFinished(queueEvents)).to.be.rejectedWith(
@@ -452,6 +480,8 @@ describe('sandboxed process', () => {
       });
     });
 
+    worker.run();
+
     await queue.add('test', { foo: 'bar' });
 
     await completing;
@@ -476,6 +506,8 @@ describe('sandboxed process', () => {
       });
     });
 
+    worker.run();
+
     const jobAdd = queue.add('foo', {});
     await onJobActive;
 
diff --git a/tests/test_stalled_jobs.ts b/tests/test_stalled_jobs.ts
index cf6a10cb3b..6a445acb6a 100644
--- a/tests/test_stalled_jobs.ts
+++ b/tests/test_stalled_jobs.ts
@@ -28,6 +28,8 @@ describe('stalled jobs', function () {
     const queueEvents = new QueueEvents(queueName, { connection });
     await queueEvents.waitUntilReady();
 
+    queueEvents.run();
+
     const concurrency = 4;
 
     const worker = new Worker(
@@ -48,6 +50,8 @@ describe('stalled jobs', function () {
 
     await worker.waitUntilReady();
 
+    worker.run();
+
     await Promise.all([
       queue.add('test', { bar: 'baz' }),
       queue.add('test', { bar1: 'baz1' }),
@@ -62,6 +66,9 @@ describe('stalled jobs', function () {
       stalledInterval: 100,
     });
     await queueScheduler.waitUntilReady();
+
+    queueScheduler.run();
+
     await worker.close(true);
 
     const allStalled = new Promise<void>(resolve => {
@@ -90,6 +97,8 @@ describe('stalled jobs', function () {
       worker2.on('completed', after(concurrency, resolve));
     });
 
+    worker2.run();
+
     await allCompleted;
 
     await queueEvents.close();
@@ -103,6 +112,8 @@ describe('stalled jobs', function () {
     const queueEvents = new QueueEvents(queueName, { connection });
     await queueEvents.waitUntilReady();
 
+    queueEvents.run();
+
     const concurrency = 4;
 
     const worker = new Worker(
@@ -123,6 +134,8 @@ describe('stalled jobs', function () {
 
     await worker.waitUntilReady();
 
+    worker.run();
+
     await Promise.all([
       queue.add('test', { bar: 'baz' }),
       queue.add('test', { bar1: 'baz1' }),
@@ -139,6 +152,8 @@ describe('stalled jobs', function () {
     });
     await queueScheduler.waitUntilReady();
 
+    queueScheduler.run();
+
     await worker.close(true);
 
     const errorMessage = 'job stalled more than allowable limit';
@@ -190,6 +205,8 @@ describe('stalled jobs', function () {
       worker.on('active', after(concurrency, resolve));
     });
 
+    worker.run();
+
     await Promise.all([
       queue.add('test', { bar: 'baz' }),
       queue.add('test', { bar1: 'baz1' }),
@@ -208,6 +225,8 @@ describe('stalled jobs', function () {
       queueScheduler.on('stalled', after(concurrency, resolve)),
     );
 
+    queueScheduler.run();
+
     await delay(500); // Wait for jobs to become active
 
     const active = await queue.getActiveCount();
diff --git a/tests/test_worker.ts b/tests/test_worker.ts
index 033be92b65..59e23a212f 100644
--- a/tests/test_worker.ts
+++ b/tests/test_worker.ts
@@ -26,6 +26,7 @@ describe('workers', function () {
     queue = new Queue(queueName, { connection });
     queueEvents = new QueueEvents(queueName, { connection });
     await queueEvents.waitUntilReady();
+    queueEvents.run();
   });
 
   afterEach(async function () {
@@ -38,6 +39,7 @@ describe('workers', function () {
   it('should get all workers for this queue', async function () {
     const worker = new Worker(queueName, async job => {}, { connection });
     await worker.waitUntilReady();
+    worker.run();
     await delay(10);
 
     const workers = await queue.getWorkers();
@@ -45,6 +47,7 @@ describe('workers', function () {
 
     const worker2 = new Worker(queueName, async job => {}, { connection });
     await worker2.waitUntilReady();
+    worker2.run();
     await delay(10);
 
     const nextWorkers = await queue.getWorkers();
@@ -61,6 +64,8 @@ describe('workers', function () {
     const worker2 = new Worker(queueName2, async job => {}, { connection });
     await worker.waitUntilReady();
     await worker2.waitUntilReady();
+    worker.run();
+    worker2.run();
 
     const workers = await queue.getWorkers();
     expect(workers).to.have.length(1);
@@ -88,6 +93,7 @@ describe('workers', function () {
         { connection },
       );
       await worker.waitUntilReady();
+      worker.run();
 
       const job = await queue.add('test', { foo: 'bar' });
 
@@ -122,6 +128,7 @@ describe('workers', function () {
         { connection },
       );
       await worker.waitUntilReady();
+      worker.run();
 
       const job = await queue.add('test', { foo: 'bar' });
 
@@ -159,6 +166,8 @@ describe('workers', function () {
       );
       await worker.waitUntilReady();
 
+      worker.run();
+
       const job = await queue.add(
         'test',
         { foo: 'bar' },
@@ -204,6 +213,8 @@ describe('workers', function () {
       );
       await worker.waitUntilReady();
 
+      worker.run();
+
       const job = await newQueue.add('test', { foo: 'bar' });
       expect(job.id).to.be.ok;
       expect(job.data.foo).to.be.eql('bar');
@@ -238,6 +249,8 @@ describe('workers', function () {
       );
       await worker.waitUntilReady();
 
+      worker.run();
+
       const datas = [0, 1, 2, 3, 4, 5, 6, 7, 8];
 
       const jobIds = await Promise.all(
@@ -295,6 +308,8 @@ describe('workers', function () {
       );
       await worker.waitUntilReady();
 
+      worker.run();
+
       const datas = [0, 1, 2, 3, 4, 5, 6, 7, 8];
 
       const jobIds = await Promise.all(
@@ -342,6 +357,8 @@ describe('workers', function () {
       );
       await worker.waitUntilReady();
 
+      worker.run();
+
       const job = await queue.add(
         'test',
         { foo: 'bar' },
@@ -376,6 +393,8 @@ describe('workers', function () {
       });
       await worker.waitUntilReady();
 
+      worker.run();
+
       const newQueue = new Queue(queueName, {
         defaultJobOptions: {
           removeOnFail: true,
@@ -408,6 +427,8 @@ describe('workers', function () {
       });
       await worker.waitUntilReady();
 
+      worker.run();
+
       const datas = [0, 1, 2, 3, 4, 5, 6, 7, 8];
 
       const jobIds = await Promise.all(
@@ -451,6 +472,8 @@ describe('workers', function () {
       });
       await worker.waitUntilReady();
 
+      worker.run();
+
       const newQueue = new Queue(queueName, {
         defaultJobOptions: {
           removeOnFail: keepJobs,
@@ -517,6 +540,8 @@ describe('workers', function () {
     const worker = new Worker(queueName, processor);
     await worker.waitUntilReady();
 
+    worker.run();
+
     await queue.pause();
     // Add a series of jobs in a predictable order
     const jobs = [
@@ -582,6 +607,8 @@ describe('workers', function () {
     const worker = new Worker(queueName, processor);
     await worker.waitUntilReady();
 
+    worker.run();
+
     // wait for all jobs to enter the queue and then start processing
     await Promise.all([normalPriority, mediumPriority, highPriority]);
 
@@ -613,6 +640,8 @@ describe('workers', function () {
     const worker = new Worker(queueName, processor);
     await worker.waitUntilReady();
 
+    worker.run();
+
     for (let i = 1; i <= maxJobs; i++) {
       await queue.add('test', { foo: 'bar', num: i });
     }
@@ -652,72 +681,69 @@ describe('workers', function () {
     });
   });
 
-  describe('when autorun option is provided as false', function () {
-    it('processes several jobs serially using process option as false', async () => {
-      let counter = 1;
-      const maxJobs = 10;
+  it('processes several jobs serially', async () => {
+    let counter = 1;
+    const maxJobs = 10;
 
-      let processor;
-      const processing = new Promise<void>((resolve, reject) => {
-        processor = async (job: Job) => {
-          try {
-            expect(job.data.num).to.be.equal(counter);
-            expect(job.data.foo).to.be.equal('bar');
-            if (counter === maxJobs) {
-              resolve();
-            }
-            counter++;
-          } catch (err) {
-            reject(err);
+    let processor;
+    const processing = new Promise<void>((resolve, reject) => {
+      processor = async (job: Job) => {
+        try {
+          expect(job.data.num).to.be.equal(counter);
+          expect(job.data.foo).to.be.equal('bar');
+          if (counter === maxJobs) {
+            resolve();
           }
-        };
-      });
+          counter++;
+        } catch (err) {
+          reject(err);
+        }
+      };
+    });
 
-      const worker = new Worker(queueName, processor, { autorun: false });
-      await worker.waitUntilReady();
+    const worker = new Worker(queueName, processor);
+    await worker.waitUntilReady();
 
-      for (let i = 1; i <= maxJobs; i++) {
-        await queue.add('test', { foo: 'bar', num: i });
-      }
+    for (let i = 1; i <= maxJobs; i++) {
+      await queue.add('test', { foo: 'bar', num: i });
+    }
 
-      worker.run();
+    worker.run();
 
-      await processing;
-      await worker.close();
-    });
+    await processing;
+    await worker.close();
+  });
 
-    describe('when process function is not defined', function () {
-      it('throws error', async () => {
-        const worker = new Worker(queueName, undefined, { autorun: false });
-        await worker.waitUntilReady();
+  describe('when process function is not defined', function () {
+    it('throws error', async () => {
+      const worker = new Worker(queueName, undefined);
+      await worker.waitUntilReady();
 
-        await expect(worker.run()).to.be.rejectedWith(
-          'No process function is defined.',
-        );
+      await expect(worker.run()).to.be.rejectedWith(
+        'No process function is defined.',
+      );
 
-        await worker.close();
-      });
+      await worker.close();
     });
+  });
 
-    describe('when run method is called when worker is running', function () {
-      it('throws error', async () => {
-        const maxJobs = 10;
-        const worker = new Worker(queueName, async (job: Job) => {}, {
-          autorun: false,
-        });
-        await worker.waitUntilReady();
-        worker.run();
+  describe('when run method is called when worker is running', function () {
+    it('throws error', async () => {
+      const maxJobs = 10;
+      const worker = new Worker(queueName, async (job: Job) => {});
+      await worker.waitUntilReady();
 
-        for (let i = 1; i <= maxJobs; i++) {
-          await queue.add('test', { foo: 'bar', num: i });
-        }
+      worker.run();
 
-        await expect(worker.run()).to.be.rejectedWith(
-          'Worker is already running.',
-        );
+      for (let i = 1; i <= maxJobs; i++) {
+        await queue.add('test', { foo: 'bar', num: i });
+      }
 
-        await worker.close();
-      });
+      await expect(worker.run()).to.be.rejectedWith(
+        'Worker is already running.',
+      );
+
+      await worker.close();
     });
   });
 
@@ -748,6 +774,8 @@ describe('workers', function () {
     const worker = new Worker(queueName, processor);
     await worker.waitUntilReady();
 
+    worker.run();
+
     await processing;
 
     await worker.close();
@@ -780,6 +808,8 @@ describe('workers', function () {
     const worker = new Worker(queueName, processor);
     await worker.waitUntilReady();
 
+    worker.run();
+
     await processing;
 
     await worker.close();
@@ -798,6 +828,8 @@ describe('workers', function () {
     const worker = new Worker(queueName, async job => {});
     await worker.waitUntilReady();
 
+    worker.run();
+
     await new Promise(resolve => {
       const resolveAfterAllJobs = after(jobs.length, resolve);
       worker.on('completed', resolveAfterAllJobs);
@@ -813,6 +845,8 @@ describe('workers', function () {
     });
     await worker.waitUntilReady();
 
+    worker.run();
+
     const job = await queue.add('test', { foo: 'bar' });
     expect(job.id).to.be.ok;
     expect(job.data.foo).to.be.eql('bar');
@@ -860,6 +894,8 @@ describe('workers', function () {
       });
     });
 
+    worker.run();
+
     await queue.add('test', { testing: true });
     await waiting;
     await worker.close();
@@ -872,6 +908,8 @@ describe('workers', function () {
     });
     await worker.waitUntilReady();
 
+    worker.run();
+
     const job = await queue.add('test', { foo: 'bar' });
     expect(job.id).to.be.ok;
     expect(job.data.foo).to.be.eql('bar');
@@ -906,6 +944,8 @@ describe('workers', function () {
     });
     await worker.waitUntilReady();
 
+    worker.run();
+
     const job = await queue.add('test', { foo: 'bar' });
     expect(job.id).to.be.ok;
     expect(job.data.foo).to.be.eql('bar');
@@ -928,6 +968,8 @@ describe('workers', function () {
 
     await worker.waitUntilReady();
 
+    worker.run();
+
     const job = await queue.add('test', { foo: 'bar' });
     expect(job.id).to.be.ok;
     expect(job.data.foo).to.be.eql('bar');
@@ -957,6 +999,8 @@ describe('workers', function () {
 
     await worker.waitUntilReady();
 
+    worker.run();
+
     const addedJob = await queue.add('test', { foo: 'bar' });
 
     const anotherWorker = new Worker(queueName, async job => {
@@ -986,6 +1030,8 @@ describe('workers', function () {
     });
     await worker.waitUntilReady();
 
+    worker.run();
+
     const job = await queue.add('test', { foo: 'bar' });
     expect(job.id).to.be.ok;
     expect(job.data.foo).to.be.eql('bar');
@@ -1023,6 +1069,8 @@ describe('workers', function () {
       });
     });
 
+    worker.run();
+
     await queue.add('test', { foo: 'bar' });
 
     await waiting;
@@ -1042,6 +1090,8 @@ describe('workers', function () {
     );
     await worker.waitUntilReady();
 
+    worker.run();
+
     const job = await queue.add('test', { foo: 'bar' });
     expect(job.id).to.be.ok;
     expect(job.data.foo).to.be.eql('bar');
@@ -1102,6 +1152,8 @@ describe('workers', function () {
       });
     });
 
+    worker.run();
+
     const job = await queue.add('test', { foo: 'bar' });
     expect(job.id).to.be.ok;
     expect(job.data.foo).to.be.eql('bar');
@@ -1130,6 +1182,8 @@ describe('workers', function () {
     );
     await worker.waitUntilReady();
 
+    worker.run();
+
     const job = await queue.add('test', { foo: 'bar' });
     expect(job.id).to.be.ok;
     expect(job.data.foo).to.be.eql('bar');
@@ -1188,6 +1242,9 @@ describe('workers', function () {
     });
     await queueScheduler.waitUntilReady();
 
+    queueScheduler.run();
+    worker.run();
+
     const job = await queue.add('test', { bar: 'baz' });
 
     const errorMessage = `Missing lock for job ${job.id}. failed`;
@@ -1230,6 +1287,9 @@ describe('workers', function () {
     });
     await queueScheduler.waitUntilReady();
 
+    queueScheduler.run();
+    worker.run();
+
     await queue.add('test', { bar: 'baz' });
 
     const completed = new Promise(resolve => {
@@ -1272,6 +1332,8 @@ describe('workers', function () {
       );
       await worker.waitUntilReady();
 
+      worker.run();
+
       await queue.add('test', {});
       await queue.add('test', {});
 
@@ -1323,6 +1385,8 @@ describe('workers', function () {
         worker.on('failed', reject);
       });
 
+      worker.run();
+
       await Promise.all(times(8, () => queue.add('test', {})));
 
       await waiting;
@@ -1373,6 +1437,9 @@ describe('workers', function () {
         worker.on('failed', cb);
         worker.on('error', reject);
       });
+
+      worker.run();
+
       await Promise.all(times(8, () => queue.add('test', {})));
 
       await waiting;
@@ -1398,6 +1465,8 @@ describe('workers', function () {
 
       await worker.waitUntilReady();
 
+      worker.run();
+
       await queue.add(
         'test',
         { foo: 'bar' },
@@ -1430,6 +1499,8 @@ describe('workers', function () {
 
       await worker.waitUntilReady();
 
+      worker.run();
+
       await queue.add(
         'test',
         { foo: 'bar' },
@@ -1462,6 +1533,8 @@ describe('workers', function () {
 
       await worker.waitUntilReady();
 
+      worker.run();
+
       await queue.add(
         'test',
         { foo: 'bar' },
@@ -1502,6 +1575,9 @@ describe('workers', function () {
 
       await worker.waitUntilReady();
 
+      queueScheduler.run();
+      worker.run();
+
       const start = Date.now();
       await queue.add(
         'test',
@@ -1542,6 +1618,9 @@ describe('workers', function () {
 
       await worker.waitUntilReady();
 
+      queueScheduler.run();
+      worker.run();
+
       const start = Date.now();
       await queue.add(
         'test',
@@ -1594,6 +1673,9 @@ describe('workers', function () {
 
       await worker.waitUntilReady();
 
+      queueScheduler.run();
+      worker.run();
+
       const start = Date.now();
       await queue.add(
         'test',
@@ -1641,6 +1723,8 @@ describe('workers', function () {
         },
       );
 
+      worker.run();
+
       await queue.add(
         'test',
         { foo: 'bar' },
@@ -1695,6 +1779,9 @@ describe('workers', function () {
         },
       );
 
+      queueScheduler.run();
+      worker.run();
+
       const start = Date.now();
       await queue.add(
         'test',
@@ -1756,6 +1843,9 @@ describe('workers', function () {
         },
       );
 
+      queueScheduler.run();
+      worker.run();
+
       const start = Date.now();
       await queue.add(
         'test',
@@ -1804,6 +1894,10 @@ describe('workers', function () {
           },
         },
       );
+
+      queueScheduler.run();
+      worker.run();
+
       const start = Date.now();
       await queue.add(
         'test',
@@ -1861,6 +1955,9 @@ describe('workers', function () {
         worker.on('completed', resolve);
       });
 
+      queueScheduler.run();
+      worker.run();
+
       const retriedJob = await queue.add('test', { foo: 'bar' });
 
       await failing;
@@ -1915,6 +2012,9 @@ describe('workers', function () {
         worker.on('completed', resolve);
       });
 
+      queueScheduler.run();
+      worker.run();
+
       const retriedJob = await queue.add('test', { foo: 'bar' });
 
       await failing;
@@ -1951,6 +2051,8 @@ describe('workers', function () {
         worker.on('active', resolve);
       });
 
+      worker.run();
+
       const job = await queue.add('test', { foo: 'bar' });
 
       expect(job.data.foo).to.equal('bar');
@@ -2395,6 +2497,9 @@ describe('workers', function () {
     );
     await worker.waitUntilReady();
 
+    queueScheduler.run();
+    worker.run();
+
     await queue.add('test', { foo: 'bar' });
 
     const allStalled = new Promise<void>(resolve => {