feat(worker): add discardTtl option

src/classes/scripts.ts

@@ -696,6 +696,38 @@ export class Scripts {
     ]);
   }
 
+  private discardJobArgs(jobId: string, token: string): (string | number)[] {
+    const keys: (string | number)[] = [
+      'active',
+      'wait',
+      'paused',
+      jobId,
+      'meta',
+    ].map(name => {
+      return this.queue.toKey(name);
+    });
+
+    keys.push(
+      this.queue.keys.events,
+      this.queue.keys.delayed,
+      this.queue.keys.priority,
+    );
+
+    return keys.concat([this.queue.toKey(''), Date.now(), jobId, token]);
+  }
+
+  async discardJob(jobId: string, token: string): Promise<void> {
+    const client = await this.queue.client;
+
+    const args = this.discardJobArgs(jobId, token);
+
+    const result = await (<any>client).discardJob(args);
+
+    if (result < 0) {
+      throw this.finishedErrors(result, jobId, 'addJob', 'active');
+    }
+  }
+
   retryJobArgs(
     jobId: string,
     lifo: boolean,

src/classes/worker.ts

@@ -2,6 +2,7 @@ import * as fs from 'fs';
 import { Redis } from 'ioredis';
 import * as path from 'path';
 import { v4 } from 'uuid';
+import { ErrorMessages } from '../enums';
 import {
   GetNextJobOptions,
   IoredisListener,
@@ -72,6 +73,14 @@ export interface WorkerListener<
     prev: string,
   ) => void;
 
+  /**
+   * Listen to 'discarded' event.
+   *
+   * This event is triggered when a job is picked after the discardTtl.
+   * Job is removed.
+   */
+  discarded: (job: Job<DataType, ResultType, NameType>, prev: string) => void;
+
   /**
    * Listen to 'drained' event.
    *
@@ -144,8 +153,6 @@ export interface WorkerListener<
   stalled: (jobId: string, prev: string) => void;
 }
 
-const RATE_LIMIT_ERROR = 'bullmq:rateLimitExceeded';
-
 /**
  *
  * This class represents a worker that is able to process jobs from the queue.
@@ -187,7 +194,7 @@ export class Worker<
   >>;
 
   static RateLimitError(): Error {
-    return new Error(RATE_LIMIT_ERROR);
+    return new Error(ErrorMessages.RATE_LIMIT);
   }
 
   constructor(
@@ -536,11 +543,6 @@ export class Worker<
             ? Math.ceil(blockTimeout)
             : blockTimeout;
 
-        // We restrict the maximum block timeout to 10 second to avoid
-        // blocking the connection for too long in the case of reconnections
-        // reference: https://github.com/taskforcesh/bullmq/issues/1658
-        blockTimeout = Math.min(blockTimeout, maximumBlockTimeout);
-
         const jobId = await client.brpoplpush(
           this.keys.wait,
           this.keys.active,
@@ -632,17 +634,16 @@ export class Worker<
     const handleFailed = async (err: Error) => {
       if (!this.connection.closing) {
         try {
-          if (err.message == RATE_LIMIT_ERROR) {
+          if (err.message == ErrorMessages.RATE_LIMIT) {
             this.limitUntil = await this.moveLimitedBackToWait(job, token);
             return;
           }
 
           if (
-            (err instanceof DelayedError || err.name == 'DelayedError') ||
-            (
-              err instanceof WaitingChildrenError ||
-              err.name == 'WaitingChildrenError'
-            )
+            err instanceof DelayedError ||
+            err.name == 'DelayedError' ||
+            err instanceof WaitingChildrenError ||
+            err.name == 'WaitingChildrenError'
           ) {
             return;
           }
@@ -664,8 +665,16 @@ export class Worker<
 
     try {
       jobsInProgress.add(inProgressItem);
-      const result = await this.callProcessJob(job, token);
-      return await handleCompleted(result);
+      if (this.opts.discardTtl) {
+        if (this.opts.discardTtl < Date.now() - job.timestamp) {
+          await this.discardJob(job.id, token);
+
+          this.emit('discarded', job, 'active');
+        }
+      } else {
+        const result = await this.callProcessJob(job, token);
+        return await handleCompleted(result);
+      }
     } catch (err) {
       return handleFailed(<Error>err);
     } finally {
@@ -951,4 +960,8 @@ export class Worker<
     }
     return parseInt(limitUntil as string) || 0;
   }
+
+  private async discardJob(jobId: string, token: string) {
+    await this.scripts.discardJob(jobId, token);
+  }
 }

src/commands/discardJob-8.lua

@@ -0,0 +1,58 @@
+--[[
+  Removes job from queue after ttl.
+
+    Input:
+      KEYS[1] 'active',
+      KEYS[2] 'wait'
+      KEYS[3] 'paused'
+      KEYS[4] job key
+      KEYS[5] 'meta'
+      KEYS[6] events stream
+      KEYS[7] delayed key
+      KEYS[8] priority key
+
+      ARGV[1]  key prefix
+      ARGV[2]  timestamp
+      ARGV[3]  jobId
+      ARGV[4]  token
+
+    Events:
+      'waiting'
+
+    Output:
+     0  - OK
+     -1 - Missing key
+     -2 - Missing lock
+]]
+local rcall = redis.call
+
+-- Includes
+--- @include "includes/getTargetQueueList"
+--- @include "includes/promoteDelayedJobs"
+--- @include "includes/removeJob"
+
+promoteDelayedJobs(KEYS[7], KEYS[2], KEYS[8], KEYS[3], KEYS[5], KEYS[6], ARGV[1], ARGV[2])
+
+if rcall("EXISTS", KEYS[4]) == 1 then
+
+  if ARGV[4] ~= "0" then
+    local lockKey = KEYS[4] .. ':lock'
+    if rcall("GET", lockKey) == ARGV[4] then
+      rcall("DEL", lockKey)
+    else
+      return -2
+    end
+  end
+
+  local target = getTargetQueueList(KEYS[5], KEYS[2], KEYS[3])
+
+  rcall("LREM", KEYS[1], 0, ARGV[3])
+  removeJob(ARGV[3], false, ARGV[1])
+
+  -- Emit discarded event
+  rcall("XADD", KEYS[6], "*", "event", "discarded", "jobId", ARGV[3], "prev", "active");
+
+  return 0
+else
+  return -1
+end

src/enums/error-messages.ts

@@ -0,0 +1,3 @@
+export enum ErrorMessages {
+  RATE_LIMIT = 'bullmq:rateLimitExceeded',
+}

src/enums/index.ts

@@ -1,2 +1,3 @@
-export * from './error-code.enum';
+export * from './error-code';
+export * from './error-messages';
 export * from './metrics-time';

src/interfaces/worker-options.ts

@@ -124,6 +124,11 @@ export interface WorkerOptions extends QueueBaseOptions {
    */
   runRetryDelay?: number;
 
+  /**
+   * Time for discard jobs without processing them.
+   */
+  discardTtl?: number;
+
   /**
    * More advanced options.
    */

tests/test_worker.ts

@@ -847,6 +847,45 @@ describe('workers', function () {
     });
   });
 
+  describe('when discardTtl is provided', function () {
+    it('remove jobs when time is greater than discardTtl', async () => {
+      const maxJobs = 5;
+
+      const worker = new Worker(queueName, async () => {}, {
+        autorun: false,
+        connection,
+        discardTtl: 100,
+      });
+      await worker.waitUntilReady();
+
+      const jobs = Array.from(Array(maxJobs).keys()).map(index => ({
+        name: 'test',
+        data: { index },
+      }));
+
+      await queue.addBulk(jobs);
+
+      const failing = new Promise<void>(resolve => {
+        worker.on(
+          'discarded',
+          after(maxJobs, (job: Job, prev: string) => {
+            expect(prev).to.be.eql('active');
+            expect(job.name).to.be.eql('test');
+            resolve();
+          }),
+        );
+      });
+
+      await delay(150);
+
+      worker.run();
+
+      await failing;
+
+      await worker.close();
+    });
+  });
+
   it('process a job that updates progress as number', async () => {
     let processor;