perf(priority): change priority as a new state

docs/gitbook/guide/jobs/job-data.md

@@ -34,15 +34,15 @@ job.data # { color: 'red' }
 
 ## Update data
 
-If you want to change the data after inserting a job, just use the **update** method. For example:
+If you want to change the data after inserting a job, just use the **updateData** method. For example:
 
 {% tabs %}
 {% tab title="TypeScript" %}
 
 ```typescript
 const job = await Job.create(queue, 'wall', { color: 'red' });
 
-await job.update({
+await job.updateData({
   color: 'blue',
 });
 
@@ -69,4 +69,4 @@ job.data # { color: 'blue' }
 
 ## Read more:
 
-- 💡 [Update API Reference](https://api.docs.bullmq.io/classes/Job.html#update)
+- 💡 [Update API Reference](https://api.docs.bullmq.io/classes/Job.html#updateData)

docs/gitbook/patterns/process-step-jobs.md

@@ -17,15 +17,15 @@ const worker = new Worker(
       switch (step) {
         case Step.Initial: {
           await doInitialStepStuff();
-          await job.update({
+          await job.updateData({
             step: Step.Second,
           });
           step = Step.Second;
           break;
         }
         case Step.Second: {
           await doSecondStepStuff();
-          await job.update({
+          await job.updateData({
             step: Step.Finish,
           });
           step = Step.Finish;
@@ -67,15 +67,15 @@ const worker = new Worker(
         case Step.Initial: {
           await doInitialStepStuff();
           await job.moveToDelayed(Date.now() + 200, token);
-          await job.update({
+          await job.updateData({
             step: Step.Second,
           });
           step = Step.Second;
           break;
         }
         case Step.Second: {
           await doSecondStepStuff();
-          await job.update({
+          await job.updateData({
             step: Step.Finish,
           });
           throw new DelayedError();
@@ -124,7 +124,7 @@ const worker = new Worker(
               },
             },
           );
-          await job.update({
+          await job.updateData({
             step: Step.Second,
           });
           step = Step.Second;
@@ -142,7 +142,7 @@ const worker = new Worker(
               },
             },
           );
-          await job.update({
+          await job.updateData({
             step: Step.Third,
           });
           step = Step.Third;
@@ -151,7 +151,7 @@ const worker = new Worker(
         case Step.Third: {
           const shouldWait = await job.moveToWaitingChildren(token);
           if (!shouldWait) {
-            await job.update({
+            await job.updateData({
               step: Step.Finish,
             });
             step = Step.Finish;
@@ -223,15 +223,15 @@ const worker = new Worker(
             },
           });
 
-          await job.update({
+          await job.updateData({
             step: Step.Second,
           });
           step = Step.Second;
           break;
         }
         case Step.Second: {
           await doSecondStepStuff();
-          await job.update({
+          await job.updateData({
             step: Step.Third,
           });
           step = Step.Third;
@@ -240,7 +240,7 @@ const worker = new Worker(
         case Step.Third: {
           const shouldWait = await job.moveToWaitingChildren(token);
           if (!shouldWait) {
-            await job.update({
+            await job.updateData({
               step: Step.Finish,
             });
             step = Step.Finish;

python/bullmq/queue.py

@@ -103,6 +103,12 @@ def trimEvents(self, maxLength: int):
         """
         return self.client.xtrim(f"{self.prefix}:{self.name}:events", maxlen = maxLength, approximate = "~")
 
+    def removeDeprecatedPriorityKey(self):
+        """
+        Delete old priority helper key.
+        """
+        return self.client.delete(f"{self.prefix}:{self.name}:priority")
+
     async def getJobCounts(self, *types):
         """
         Returns the job counts for each type specified or every list/set in the queue by default.

python/bullmq/scripts.py

@@ -30,21 +30,21 @@ def __init__(self, prefix: str, queueName: str, redisConnection: RedisConnection
         self.redisConnection = redisConnection
         self.redisClient = redisConnection.conn
         self.commands = {
-            "addJob": self.redisClient.register_script(self.getScript("addJob-8.lua")),
-            "changePriority": self.redisClient.register_script(self.getScript("changePriority-4.lua")),
+            "addJob": self.redisClient.register_script(self.getScript("addJob-9.lua")),
+            "changePriority": self.redisClient.register_script(self.getScript("changePriority-5.lua")),
             "extendLock": self.redisClient.register_script(self.getScript("extendLock-2.lua")),
             "getCounts": self.redisClient.register_script(self.getScript("getCounts-1.lua")),
-            "getState": self.redisClient.register_script(self.getScript("getState-7.lua")),
-            "getStateV2": self.redisClient.register_script(self.getScript("getStateV2-7.lua")),
+            "getState": self.redisClient.register_script(self.getScript("getState-8.lua")),
+            "getStateV2": self.redisClient.register_script(self.getScript("getStateV2-8.lua")),
             "moveStalledJobsToWait": self.redisClient.register_script(self.getScript("moveStalledJobsToWait-8.lua")),
-            "moveToActive": self.redisClient.register_script(self.getScript("moveToActive-9.lua")),
+            "moveToActive": self.redisClient.register_script(self.getScript("moveToActive-10.lua")),
             "moveToDelayed": self.redisClient.register_script(self.getScript("moveToDelayed-8.lua")),
-            "moveToFinished": self.redisClient.register_script(self.getScript("moveToFinished-12.lua")),
+            "moveToFinished": self.redisClient.register_script(self.getScript("moveToFinished-13.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")),
+            "retryJob": self.redisClient.register_script(self.getScript("retryJob-9.lua")),
             "retryJobs": self.redisClient.register_script(self.getScript("retryJobs-6.lua")),
             "saveStacktrace": self.redisClient.register_script(self.getScript("saveStacktrace-1.lua")),
             "updateData": self.redisClient.register_script(self.getScript("updateData-1.lua")),
@@ -53,7 +53,7 @@ def __init__(self, prefix: str, queueName: str, redisConnection: RedisConnection
 
         # loop all the names and add them to the keys object
         names = ["", "active", "wait", "paused", "completed", "failed", "delayed",
-                 "stalled", "limiter", "priority", "id", "stalled-check", "meta", "events", "waiting-children"]
+                 "stalled", "limiter", "prioritized", "id", "stalled-check", "meta", "pc", "events", "waiting-children"]
         for name in names:
             self.keys[name] = self.toKey(name)
 
@@ -96,7 +96,7 @@ def addJob(self, job: Job):
         packedOpts = msgpack.packb(job.opts)
 
         keys = self.getKeys(['wait', 'paused', 'meta', 'id',
-                            'delayed', 'priority', 'completed', 'events'])
+                            'delayed', 'prioritized', 'completed', 'events', 'pc'])
 
         return self.commands["addJob"](keys=keys, args=[packedArgs, jsonData, packedOpts])
 
@@ -112,7 +112,8 @@ def retryJobArgs(self, job_id: str, lifo: bool, token: str):
         keys.append(self.keys['meta'])
         keys.append(self.keys['events'])
         keys.append(self.keys['delayed'])
-        keys.append(self.keys['priority'])
+        keys.append(self.keys['prioritized'])
+        keys.append(self.keys['pc'])
 
         push_cmd = "R" if lifo else "L"
 
@@ -127,7 +128,7 @@ def moveToDelayedArgs(self, job_id: str, timestamp: int, token: str):
         if timestamp > 0:
             max_timestamp = max_timestamp * 0x1000 + (convert_to_int(job_id) & 0xfff)
 
-        keys = self.getKeys(['wait', 'active', 'priority', 'delayed'])
+        keys = self.getKeys(['wait', 'active', 'prioritized', 'delayed'])
         keys.append(self.toKey(job_id))
         keys.append(self.keys['events'])
         keys.append(self.keys['paused'])
@@ -163,9 +164,9 @@ def getCounts(self, types):
 
     async def getState(self, job_id):
         keys = self.getKeys(['completed', 'failed', 'delayed', 'active', 'wait',
-                'paused', 'waiting-children'])
+                'paused', 'waiting-children', 'prioritized'])
 
-        args = [job_id]
+        args = [job_id, self.toKey(job_id)]
 
         redis_version = await self.redisConnection.getRedisVersion()
 
@@ -180,7 +181,8 @@ async def changePriority(self, job_id: str, priority:int = 0, lifo:bool = False)
         keys = [self.keys['wait'],
             self.keys['paused'],
             self.keys['meta'],
-            self.keys['priority']]
+            self.keys['prioritized'],
+            self.keys['pc']]
 
         args = [priority, self.toKey(job_id), job_id, 1 if lifo else 0]
 
@@ -231,7 +233,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', 'prioritized', 'events'])
         return self.commands["pause"](keys, args=["paused" if pause else "resumed"])
 
     async def obliterate(self, count: int, force: bool = False):
@@ -265,8 +267,8 @@ async def moveToActive(self, token: str, opts: dict, jobId: str = "") -> list[An
         lockDuration = opts.get("lockDuration", 0)
         limiter = opts.get("limiter", None)
 
-        keys = self.getKeys(['wait', 'active', 'priority', 'events',
-                            'stalled', 'limiter', 'delayed', 'paused', 'meta'])
+        keys = self.getKeys(['wait', 'active', 'prioritized', 'events',
+                            'stalled', 'limiter', 'delayed', 'paused', 'meta', 'pc'])
         packedOpts = msgpack.packb(
             {"token": token, "lockDuration": lockDuration, "limiter": limiter}, use_bin_type=True)
         args = [self.keys[''], timestamp, jobId or "", packedOpts]
@@ -296,10 +298,9 @@ def moveToFinishedArgs(self, job: Job, val: Any, propVal: str, shouldRemove, tar
         timestamp = round(time.time() * 1000)
         metricsKey = self.toKey('metrics:' + target)
 
-        keys = self.getKeys(['wait', 'active', 'priority', 'events',
-                            'stalled', 'limiter', 'delayed', 'paused', target])
+        keys = self.getKeys(['wait', 'active', 'prioritized', 'events',
+                            'stalled', 'limiter', 'delayed', 'paused', 'meta', 'pc', target])
         keys.append(self.toKey(job.id))
-        keys.append(self.keys['meta'])
         keys.append(metricsKey)
 
         def getKeepJobs(shouldRemove: bool | dict | int | None):

src/classes/child-processor.ts

@@ -150,7 +150,7 @@ function wrapJob(
     /*
      * Emulate the real job `update` function.
      */
-    update: async (data: any) => {
+    updateData: async (data: any) => {
       send({
         cmd: ParentCommand.Update,
         value: data,

src/classes/job.ts

@@ -429,7 +429,7 @@ export class Job<
    *
    * @param data - the data that will replace the current jobs data.
    */
-  update(data: DataType): Promise<void> {
+  updateData(data: DataType): Promise<void> {
     this.data = data;
 
     return this.scripts.updateData<DataType, ReturnType, NameType>(this, data);
@@ -551,7 +551,7 @@ export class Job<
     );
 
     const result = await this.scripts.moveToFinished(this.id, args);
-    this.finishedOn = args[13] as number;
+    this.finishedOn = args[14] as number;
 
     return result;
   }
@@ -632,7 +632,7 @@ export class Job<
         fetchNext,
       );
       (<any>multi).moveToFinished(args);
-      finishedOn = args[13];
+      finishedOn = args[14];
       command = 'failed';
     }
 
@@ -1085,6 +1085,17 @@ export class Job<
       );
     }
 
+    if (this.opts.priority) {
+      if (Math.trunc(this.opts.priority) !== this.opts.priority) {
+        throw new Error(`Priority should not be float`);
+      }
+
+      const priorityLimit = 2 ** 21;
+      if (this.opts.priority > 2 ** 21) {
+        throw new Error(`Priority should be between 0 and ${priorityLimit}`);
+      }
+    }
+
     return this.scripts.addJob(
       client,
       jobData,

src/classes/queue-getters.ts

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

src/classes/queue-keys.ts

@@ -14,7 +14,7 @@ export class QueueKeys {
       'resumed',
       'id',
       'delayed',
-      'priority',
+      'prioritized',
       'stalled-check',
       'completed',
       'failed',
@@ -26,6 +26,7 @@ export class QueueKeys {
       'meta',
       'events',
       'delay',
+      'pc',
     ].forEach(key => {
       keys[key] = this.toKey(name, key);
     });

src/classes/queue.ts

@@ -388,6 +388,7 @@ export class Queue<
       | 'wait'
       | 'active'
       | 'paused'
+      | 'prioritized'
       | 'delayed'
       | 'failed' = 'completed',
   ): Promise<string[]> {
@@ -454,4 +455,12 @@ export class Queue<
     const client = await this.client;
     return client.xtrim(this.keys.events, 'MAXLEN', '~', maxLength);
   }
+
+  /**
+   * Delete old priority helper key.
+   */
+  async removeDeprecatedPriorityKey(): Promise<number> {
+    const client = await this.client;
+    return client.del(this.toKey('priority'));
+  }
 }

src/classes/sandbox.ts

@@ -36,7 +36,7 @@ const sandbox = <T, R, N extends string>(
             await job.log(msg.value);
             break;
           case ParentCommand.Update:
-            await job.update(msg.value);
+            await job.updateData(msg.value);
             break;
         }
       };