fix(move-to-delayed): consider promoting delayed jobs (#1493)

Author: roggervalf

src/classes/scripts.ts

@@ -496,7 +496,11 @@ export class Scripts {
  }
 
  // Note: We have an issue here with jobs using custom job ids
- moveToDelayedArgs(jobId: string, timestamp: number, token: string): string[] {
+ moveToDelayedArgs(
+ jobId: string,
+ timestamp: number,
+ token: string,
+ ): (string | number)[] {
  //
  // Bake in the job id first 12 bits into the timestamp
  // to guarantee correct execution order of delayed jobs
@@ -510,12 +514,28 @@ export class Scripts {
  timestamp = timestamp * 0x1000 + (+jobId & 0xfff);
  }
 
- const keys = ['active', 'delayed', jobId].map(name => {
+ const keys: (string | number)[] = [
+ 'wait',
+ 'active',
+ 'priority',
+ 'delayed',
+ jobId,
+ ].map(name => {
  return this.queue.toKey(name);
  });
- keys.push.apply(keys, [this.queue.keys.events, this.queue.keys.delay]);
+ keys.push.apply(keys, [
+ this.queue.keys.events,
+ this.queue.keys.paused,
+ this.queue.keys.meta,
+ ]);
 
- return keys.concat([JSON.stringify(timestamp), jobId, token]);
+ return keys.concat([
+ this.queue.keys[''],
+ Date.now(),
+ JSON.stringify(timestamp),
+ jobId,
+ token,
+ ]);
  }
 
  moveToWaitingChildrenArgs(

src/commands/moveToDelayed-5.lua

@@ -0,0 +1,63 @@
+--[[
+ Moves job from active to delayed set.
+
+ Input:
+ KEYS[1] wait key
+ KEYS[2] active key
+ KEYS[3] priority key
+ KEYS[4] delayed key
+ KEYS[5] job key
+ KEYS[6] events stream
+ KEYS[7] paused key
+ KEYS[8] meta key
+
+ ARGV[1] key prefix
+ ARGV[2] timestamp
+ ARGV[3] delayedTimestamp
+ ARGV[4] the id of the job
+ ARGV[5] queue token
+
+ Output:
+ 0 - OK
+ -1 - Missing job.
+ -3 - Job not in active set.
+
+ Events:
+ - delayed key.
+]]
+local rcall = redis.call
+
+-- Includes
+--- @include "includes/promoteDelayedJobs"
+
+promoteDelayedJobs(KEYS[4], KEYS[1], KEYS[3], KEYS[7], KEYS[8], KEYS[6], ARGV[1], ARGV[2])
+
+if rcall("EXISTS", KEYS[5]) == 1 then
+
+ if ARGV[5] ~= "0" then
+ local lockKey = KEYS[5] .. ':lock'
+ if rcall("GET", lockKey) == ARGV[5] then
+ rcall("DEL", lockKey)
+ else
+ return -2
+ end
+ end
+
+ local jobId = ARGV[4]
+ local score = tonumber(ARGV[3])
+ local delayedTimestamp = (score / 0x1000)
+ 
+ local numRemovedElements = rcall("LREM", KEYS[2], -1, jobId)
+
+ if(numRemovedElements < 1) then
+ return -3
+ end
+
+ rcall("ZADD", KEYS[4], score, jobId)
+
+ rcall("XADD", KEYS[6], "*", "event", "delayed", "jobId", jobId, "delay", delayedTimestamp);
+
+ return 0
+else
+ return -1
+end

src/commands/moveToDelayed-8.lua

@@ -198,6 +198,60 @@ describe('Delayed jobs', function () {
  await worker2.close();
  });
 
+ describe('when delayed jobs are ready when pending jobs are moved to delayed', function () {
+ it('processes jobs without getting stuck', async () => {
+ const countJobs = 28;
+
+ for (let j = 0; j < countJobs; j++) {
+ await queue.add(
+ 'test',
+ { foo: `bar${j}` },
+ { attempts: 2, backoff: 10 },
+ );
+ }
+
+ const concurrency = 50;
+
+ let worker: Worker;
+ const processedJobs: { data: any }[] = [];
+ const processing = new Promise<void>(resolve => {
+ worker = new Worker(
+ queueName,
+ async (job: Job) => {
+ if (job.attemptsMade == 1) {
+ await delay(250);
+
+ throw new Error('error');
+ }
+
+ await delay(25);
+
+ processedJobs.push({ data: job.data });
+
+ if (processedJobs.length == countJobs) {
+ resolve();
+ }
+
+ return;
+ },
+ {
+ connection,
+ concurrency,
+ },
+ );
+ worker.on('error', err => {
+ console.error(err);
+ });
+ });
+
+ await processing;
+
+ expect(processedJobs.length).to.be.equal(countJobs);
+
+ await worker.close();
+ }).timeout(4000);
+ });
+
  it('should process delayed jobs with exact same timestamps in correct order (FIFO)', async function () {
  let order = 1;
  const numJobs = 43;