perf(priority): change priority as a new state

python/bullmq/scripts.py

@@ -41,7 +41,7 @@ def __init__(self, prefix: str, queueName: str, redisConnection: RedisConnection
             "moveToDelayed": self.redisClient.register_script(self.getScript("moveToDelayed-8.lua")),
             "moveToFinished": self.redisClient.register_script(self.getScript("moveToFinished-12.lua")),
             "obliterate": self.redisClient.register_script(self.getScript("obliterate-2.lua")),
-            "pause": self.redisClient.register_script(self.getScript("pause-4.lua")),
+            "pause": self.redisClient.register_script(self.getScript("pause-5.lua")),
             "removeJob": self.redisClient.register_script(self.getScript("removeJob-1.lua")),
             "reprocessJob": self.redisClient.register_script(self.getScript("reprocessJob-6.lua")),
             "retryJob": self.redisClient.register_script(self.getScript("retryJob-8.lua")),
@@ -231,7 +231,7 @@ def pause(self, pause: bool = True):
         """
         src = "wait" if pause else "paused"
         dst = "paused" if pause else "wait"
-        keys = self.getKeys([src, dst, 'meta', 'events'])
+        keys = self.getKeys([src, dst, 'meta', 'priority', 'events'])
         return self.commands["pause"](keys, args=["paused" if pause else "resumed"])
 
     async def obliterate(self, count: int, force: bool = False):

src/classes/queue-getters.ts

@@ -82,6 +82,7 @@ export class QueueGetters<
       'delayed',
       'failed',
       'paused',
+      'priority',
       'waiting',
       'waiting-children',
     ];

src/classes/queue.ts

@@ -388,6 +388,7 @@ export class Queue<
       | 'wait'
       | 'active'
       | 'paused'
+      | 'priority'
       | 'delayed'
       | 'failed' = 'completed',
   ): Promise<string[]> {

src/classes/scripts.ts

@@ -146,7 +146,7 @@ export class Scripts {
       dst = 'wait';
     }
 
-    const keys = [src, dst, 'meta'].map((name: string) =>
+    const keys = [src, dst, 'meta', 'priority'].map((name: string) =>
       this.queue.toKey(name),
     );
 
@@ -837,7 +837,7 @@ export class Scripts {
     const args: (string | number | boolean | Buffer)[] = [
       queueKeys[''],
       Date.now(),
-      jobId,
+      jobId || '',
       pack({
         token,
         lockDuration: opts.lockDuration,

src/commands/addJob-8.lua

@@ -145,16 +145,15 @@ elseif (delayedTimestamp ~= 0) then
   local target = getTargetQueueList(KEYS[3], KEYS[1], KEYS[2])
   addDelayMarkerIfNeeded(target, KEYS[5])
 else
-  local target = getTargetQueueList(KEYS[3], KEYS[1], KEYS[2])
+  local target, paused = getTargetQueueList(KEYS[3], KEYS[1], KEYS[2])
 
   -- Standard or priority add
   if priority == 0 then
     -- LIFO or FIFO
     local pushCmd = opts['lifo'] and 'RPUSH' or 'LPUSH'
     rcall(pushCmd, target, jobId)
   else
-    -- Priority add
-    addJobWithPriority(KEYS[6], priority, target, jobId)
+    addJobWithPriority(KEYS[1], KEYS[6], priority, paused, jobId)
   end
   -- Emit waiting event
   rcall("XADD", KEYS[8], "*", "event", "waiting", "jobId", jobId)

src/commands/changePriority-4.lua

@@ -25,21 +25,25 @@ local rcall = redis.call
 --- @include "includes/getTargetQueueList"
 
 if rcall("EXISTS", jobKey) == 1 then
-  local target = getTargetQueueList(KEYS[3], KEYS[1], KEYS[2])
-
-  local numRemovedElements = rcall("LREM", target, -1, jobId)
-  if numRemovedElements > 0 then
-    rcall("ZREM", KEYS[4], jobId)
-
-    -- Standard or priority add
-    if priority == 0 then
-      -- LIFO or FIFO
-      local pushCmd = ARGV[4] == '1' and 'RPUSH' or 'LPUSH';
-      rcall(pushCmd, target, jobId)
-    else
-      -- Priority add
-      addJobWithPriority(KEYS[4], priority, target, jobId)
-    end
+  local target, paused = getTargetQueueList(KEYS[3], KEYS[1], KEYS[2])
+
+  local isPrioritized = rcall("ZREM", KEYS[4], jobId) > 0
+  if isPrioritized then
+    -- Priority add
+    addJobWithPriority(KEYS[1], KEYS[4], priority, paused, jobId)
+  else
+    local numRemovedElements = rcall("LREM", target, -1, jobId)
+    if numRemovedElements > 0 then
+      -- Standard or priority add
+      if priority == 0 then
+        -- LIFO or FIFO
+        local pushCmd = ARGV[4] == '1' and 'RPUSH' or 'LPUSH';
+        rcall(pushCmd, target, jobId)
+      else
+        -- Priority add
+        addJobWithPriority(KEYS[1], KEYS[4], priority, paused, jobId)
+      end
+    end  
   end
 
   rcall("HSET", jobKey, "priority", priority)

src/commands/cleanJobsInSet-2.lua

@@ -35,6 +35,8 @@ if ARGV[4] == "active" then
   result = cleanList(KEYS[1], ARGV[1], rangeStart, rangeEnd, ARGV[2], false)
 elseif ARGV[4] == "delayed" then
   result = cleanSet(KEYS[1], ARGV[1], rangeStart, rangeEnd, ARGV[2], limit, {"processedOn", "timestamp"})
+elseif ARGV[4] == "priority" then
+  result = cleanSet(KEYS[1], ARGV[1], rangeStart, rangeEnd, ARGV[2], limit, {"timestamp"})
 elseif ARGV[4] == "wait" or ARGV[4] == "paused" then
   result = cleanList(KEYS[1], ARGV[1], rangeStart, rangeEnd, ARGV[2], true)
 else

src/commands/drain-4.lua

@@ -22,4 +22,4 @@ if KEYS[3] ~= "" then
   removeZSetJobs(KEYS[3], true, queueBaseKey, 0) --delayed
 end
 
-rcall("DEL", KEYS[4])
+removeZSetJobs(KEYS[4], true, queueBaseKey, 0) --priority

src/commands/includes/addJobWithPriority.lua

@@ -2,15 +2,12 @@
   Function to add job considering priority.
 ]]
 
-local function addJobWithPriority(priorityKey, priority, targetKey, jobId)
-  rcall("ZADD", priorityKey, priority, jobId)
-  local count = rcall("ZCOUNT", priorityKey, 0, priority)
+-- Includes
+--- @include "addPriorityMarkerIfNeeded"
 
-  local len = rcall("LLEN", targetKey)
-  local id = rcall("LINDEX", targetKey, len - (count - 1))
-  if id then
-    rcall("LINSERT", targetKey, "BEFORE", id, jobId)
-  else
-    rcall("RPUSH", targetKey, jobId)
+local function addJobWithPriority(waitKey, priorityKey, priority, paused, jobId)
+  rcall("ZADD", priorityKey, priority, jobId)
+  if not paused then
+    addPriorityMarkerIfNeeded(waitKey)
   end
 end

src/commands/includes/addPriorityMarkerIfNeeded.lua

@@ -0,0 +1,12 @@
+--[[
+  Function priority marker to wait if needed
+  in order to wake up our workers and to respect priority
+  order as much as possible
+]]
+local function addPriorityMarkerIfNeeded(waitKey)
+  local waitLen = rcall("LLEN", waitKey)
+
+  if waitLen == 0 then
+    rcall("LPUSH", waitKey, "0:0")
+  end
+end

src/commands/includes/cleanList.lua

@@ -35,9 +35,6 @@ local function cleanList(listKey, jobKeyPrefix, rangeStart, rangeEnd,
         -- occur at the end of the script
         rcall("LSET", listKey, rangeEnd - jobIdsLen + i, deletionMarker)
         removeJob(job, true, jobKeyPrefix)
-        if isWaiting then
-          rcall("ZREM", jobKeyPrefix .. "priority", job)
-        end
         deletedCount = deletedCount + 1
         table.insert(deleted, job)
       end

src/commands/includes/getRateLimitTTL.lua

@@ -2,7 +2,7 @@ local function getRateLimitTTL(maxJobs, rateLimiterKey)
   if maxJobs then
     local pttl = rcall("PTTL", rateLimiterKey)
 
-    if pttl <= 0 then
+    if pttl == 0 then
       rcall("DEL", rateLimiterKey)
     end
 

src/commands/includes/moveJobFromWaitToActive.lua

@@ -17,39 +17,49 @@
     opts - limiter
 ]]
 
+-- Includes
+--- @include "addJobWithPriority"
+
 local function moveJobFromWaitToActive(keys, keyPrefix, targetKey, jobId, processedOn,
-    maxJobs, expireTime, opts)
+    maxJobs, expireTime, paused, opts)
+  local jobKey = keyPrefix .. jobId
   -- Check if we need to perform rate limiting.
   if maxJobs then
     local rateLimiterKey = keys[6];
-    local jobCounter = tonumber(rcall("INCR", rateLimiterKey))
-
-    if jobCounter == 1 then
-      local limiterDuration = opts['limiter'] and opts['limiter']['duration']
-      local integerDuration = math.floor(math.abs(limiterDuration))
-      rcall("PEXPIRE", rateLimiterKey, integerDuration)
-    end
 
     -- check if we passed rate limit, we need to remove the job and return expireTime
     if expireTime > 0 then
       -- remove from active queue and add back to the wait list
       rcall("LREM", keys[2], 1, jobId)
-      rcall("RPUSH", targetKey, jobId)
+
+      local priority = tonumber(rcall("HGET", jobKey, "priority")) or 0
+
+      if priority == 0 then
+        rcall("RPUSH", targetKey, jobId)
+      else
+        addJobWithPriority(keys[1], keys[3], priority, paused, jobId)
+      end
 
       -- Return when we can process more jobs
       return {0, 0, expireTime, 0}
     end
+
+    local jobCounter = tonumber(rcall("INCR", rateLimiterKey))
+
+    if jobCounter == 1 then
+      local limiterDuration = opts['limiter'] and opts['limiter']['duration']
+      local integerDuration = math.floor(math.abs(limiterDuration))
+      rcall("PEXPIRE", rateLimiterKey, integerDuration)
+    end
   end
 
-  local jobKey = keyPrefix .. jobId
   local lockKey = jobKey .. ':lock'
 
   -- get a lock
   if opts['token'] ~= "0" then
     rcall("SET", lockKey, opts['token'], "PX", opts['lockDuration'])
   end
 
-  rcall("ZREM", keys[3], jobId) -- remove from priority
   rcall("XADD", keys[4], "*", "event", "active", "jobId", jobId, "prev", "waiting")
   rcall("HSET", jobKey, "processedOn", processedOn)
   rcall("HINCRBY", jobKey, "attemptsMade", 1)

src/commands/includes/promoteDelayedJobs.lua

@@ -10,28 +10,30 @@
 --- @include "addJobWithPriority"
 
 -- Try to get as much as 1000 jobs at once
-local function promoteDelayedJobs(delayedKey, targetKey, priorityKey,
-                                  eventStreamKey, prefix, timestamp)
+local function promoteDelayedJobs(delayedKey, waitKey, targetKey, priorityKey,
+                                  eventStreamKey, prefix, timestamp, paused)
     local jobs = rcall("ZRANGEBYSCORE", delayedKey, 0, (timestamp + 1) * 0x1000, "LIMIT", 0, 1000)
 
     if (#jobs > 0) then
         rcall("ZREM", delayedKey, unpack(jobs))
 
         for _, jobId in ipairs(jobs) do
+            local jobKey = prefix .. jobId
             local priority =
-                tonumber(rcall("HGET", prefix .. jobId, "priority")) or 0
+                tonumber(rcall("HGET", jobKey, "priority")) or 0
 
             if priority == 0 then
                 -- LIFO or FIFO
                 rcall("LPUSH", targetKey, jobId)
             else
-                addJobWithPriority(priorityKey, priority, targetKey, jobId)
+                rcall("SET", "DEBUG", "DELAYED")
+                addJobWithPriority(waitKey, priorityKey, priority, paused, jobId)
             end
 
             -- Emit waiting event
             rcall("XADD", eventStreamKey, "*", "event", "waiting", "jobId",
                   jobId, "prev", "delayed")
-            rcall("HSET", prefix .. jobId, "delay", 0)
+            rcall("HSET", jobKey, "delay", 0)
         end
     end
 end

src/commands/includes/removeJobFromAnyState.lua

@@ -19,6 +19,9 @@ local function removeJobFromAnyState( prefix, jobId)
   elseif rcall("ZSCORE", prefix .. "failed", jobId) then
     rcall("ZREM", prefix .. "failed", jobId)
     return "failed"
+  elseif rcall("ZSCORE", prefix .. "priority", jobId) then
+    rcall("ZREM", prefix .. "priority", jobId)
+    return "priority"
   -- We remove only 1 element from the list, since we assume they are not added multiple times
   elseif rcall("LREM", prefix .. "wait", 1, jobId) == 1 then
     return "wait"

src/commands/includes/updateParentDepsIfNeeded.lua

@@ -14,7 +14,8 @@ local function updateParentDepsIfNeeded(parentKey, parentQueueKey, parentDepende
   local activeParent = rcall("ZSCORE", parentQueueKey .. ":waiting-children", parentId)
   if rcall("SCARD", parentDependenciesKey) == 0 and activeParent then 
     rcall("ZREM", parentQueueKey .. ":waiting-children", parentId)
-    local parentTarget = getTargetQueueList(parentQueueKey .. ":meta", parentQueueKey .. ":wait",
+    local parentWaitKey = parentQueueKey .. ":wait"
+    local parentTarget, paused = getTargetQueueList(parentQueueKey .. ":meta", parentWaitKey,
       parentQueueKey .. ":paused")
     local jobAttributes = rcall("HMGET", parentKey, "priority", "delay")
     local priority = tonumber(jobAttributes[1]) or 0
@@ -30,7 +31,7 @@ local function updateParentDepsIfNeeded(parentKey, parentQueueKey, parentDepende
     elseif priority == 0 then
       rcall("RPUSH", parentTarget, parentId)
     else
-      addJobWithPriority(parentQueueKey .. ":priority", priority, parentTarget, parentId)
+      addJobWithPriority(parentWaitKey, parentQueueKey .. ":priority", priority, paused, parentId)
     end
 
     rcall("XADD", parentQueueKey .. ":events", "*", "event", "waiting", "jobId", parentId, "prev", "waiting-children")

src/commands/moveJobFromActiveToWait-9.lua

@@ -31,14 +31,14 @@ local pttl = rcall("PTTL", KEYS[7])
 if lockToken == token and pttl > 0 then
   local removed = rcall("LREM", KEYS[1], 1, jobId)
   if (removed > 0) then
-    local target = getTargetQueueList(KEYS[6], KEYS[2], KEYS[5])
+    local target, paused = getTargetQueueList(KEYS[6], KEYS[2], KEYS[5])
 
     rcall("SREM", KEYS[3], jobId)
 
     local priority = tonumber(rcall("HGET", ARGV[3], "priority")) or 0
 
     if priority > 0 then
-      addJobWithPriority(KEYS[8], priority, target, jobId)
+      addJobWithPriority(KEYS[2], KEYS[8], priority, paused, jobId)
     else
       rcall("RPUSH", target, jobId)
     end

src/commands/moveToActive-9.lua

@@ -41,10 +41,10 @@ local rcall = redis.call
 --- @include "includes/getTargetQueueList"
 --- @include "includes/promoteDelayedJobs"
 
-local target = getTargetQueueList(KEYS[9], KEYS[1], KEYS[8])
+local target, paused = getTargetQueueList(KEYS[9], KEYS[1], KEYS[8])
 
 -- Check if there are delayed jobs that we can move to wait.
-promoteDelayedJobs(KEYS[7], target, KEYS[3], KEYS[4], ARGV[1], ARGV[2])
+promoteDelayedJobs(KEYS[7], KEYS[1], target, KEYS[3], KEYS[4], ARGV[1], ARGV[2], paused)
 
 local opts = cmsgpack.unpack(ARGV[4])
 local maxJobs = tonumber(opts['limiter'] and opts['limiter']['max'])
@@ -59,6 +59,9 @@ else
     return { 0, 0, expireTime, 0 }
   end
 
+  -- paused queue
+  if paused then return {0, 0, 0, 0} end
+
   -- no job ID, try non-blocking move from wait to active
   jobId = rcall("RPOPLPUSH", KEYS[1], KEYS[2])
 end
@@ -72,6 +75,9 @@ if jobId then
       return { 0, 0, expireTime, 0 }
     end
 
+    -- paused queue
+    if paused then return {0, 0, 0, 0} end
+
     -- Move again since we just got the marker job.
     jobId = rcall("RPOPLPUSH", KEYS[1], KEYS[2])
 
@@ -85,14 +91,28 @@ if jobId then
 
   if jobId then
     -- this script is not really moving, it is preparing the job for processing
-    return moveJobFromWaitToActive(KEYS, ARGV[1], target, jobId, ARGV[2], maxJobs, expireTime, opts)
+    return moveJobFromWaitToActive(KEYS, ARGV[1], target, jobId, ARGV[2], maxJobs, expireTime, paused, opts)
+  else
+    local prioritizedJob = rcall("ZPOPMIN", KEYS[3])
+    if #prioritizedJob > 0 then
+      jobId = prioritizedJob[1]
+      rcall("LPUSH", KEYS[2], jobId)
+      return moveJobFromWaitToActive(KEYS, ARGV[1], target, jobId, ARGV[2], maxJobs, expireTime, paused, opts)
+    end
+  end
+else
+  local prioritizedJob = rcall("ZPOPMIN", KEYS[3])
+  if #prioritizedJob > 0 then
+    jobId = prioritizedJob[1]
+    rcall("LPUSH", KEYS[2], jobId)
+    return moveJobFromWaitToActive(KEYS, ARGV[1], target, jobId, ARGV[2], maxJobs, expireTime, paused, opts)
   end
 end
 
 -- Return the timestamp for the next delayed job if any.
 local nextTimestamp = getNextDelayedTimestamp(KEYS[7])
 if (nextTimestamp ~= nil) then
-  return { 0, 0, 0, nextTimestamp}
+  return { 0, 0, 0, nextTimestamp }
 end
 
 return { 0, 0, 0, 0}