feat(worker): add discardTtl option

src/classes/worker.ts

@@ -139,6 +139,8 @@ export interface WorkerListener<
 
 const RATE_LIMIT_ERROR = 'bullmq:rateLimitExceeded';
 
+const DISCARD_TTL_ERROR = 'bullmq:discardTtlExceeded';
+
 /**
  *
  * This class represents a worker that is able to process jobs from the queue.
@@ -527,7 +529,6 @@ export class Worker<
             ? Math.ceil(blockTimeout)
             : blockTimeout;
 
-        const now = Date.now();
         const jobId = await client.brpoplpush(
           this.keys.wait,
           this.keys.active,
@@ -669,6 +670,12 @@ export class Worker<
     lockExtender();
 
     try {
+      if (this.opts.discardTtl) {
+        if (this.opts.discardTtl < Date.now() - job.timestamp) {
+          return handleFailed(new Error(DISCARD_TTL_ERROR));
+        }
+      }
+
       const result = await this.callProcessJob(job, token);
       return await handleCompleted(result);
     } catch (err) {

src/interfaces/worker-options.ts

@@ -53,7 +53,16 @@ export interface WorkerOptions extends QueueBaseOptions {
   lockDuration?: number;
   lockRenewTime?: number;
   runRetryDelay?: number;
-  settings?: AdvancedOptions; // FIXME only backoffStrategies is used
+
+  /**
+   * Settings for advanced options.
+   */
+  settings?: AdvancedOptions;
+
+  /**
+   * Time for discard jobs without processing them.
+   */
+  discardTtl?: number;
 }
 
 export interface GetNextJobOptions {

tests/test_worker.ts

@@ -757,6 +757,53 @@ describe('workers', function () {
     });
   });
 
+  describe('when discardTtl is provided', function () {
+    it('moves jobs to failed when time is greater than discardTtl', async () => {
+      const maxJobs = 5;
+
+      let processor;
+      const processing = new Promise<void>(resolve => {
+        processor = async () => {
+          await delay(200);
+          resolve();
+        };
+      });
+
+      const worker = new Worker(queueName, processor, {
+        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(
+          'failed',
+          after(4, (job: Job, error) => {
+            expect(error.name).to.be.eql('Error');
+            expect(error.message).to.be.eql('bullmq:discardTtlExceeded');
+            expect(job.failedReason).to.be.eql('bullmq:discardTtlExceeded');
+            resolve();
+          }),
+        );
+      });
+
+      worker.run();
+
+      await processing;
+
+      await failing;
+
+      await worker.close();
+    });
+  });
+
   it('process a job that updates progress as number', async () => {
     let processor;