perf: keep jobs in waiting list when queue is paused

docs/gitbook/SUMMARY.md

@@ -65,7 +65,8 @@
   - [Producers](guide/nestjs/producers.md)
   - [Queue Events Listeners](guide/nestjs/queue-events-listeners.md)
 - [Going to production](guide/going-to-production.md)
-- [Migration to newer versions](guide/migration-to-newer-versions.md)
+- [Migration to newer versions](guide/migrations/migration-to-newer-versions.md)
+  - [Version 6](guide/migrations/v6.md)
 - [Troubleshooting](guide/troubleshooting.md)
 
 ## Patterns

docs/gitbook/guide/migration-to-newer-versions.md → docs/gitbook/guide/migrations/migration-to-newer-versions.md

@@ -55,4 +55,3 @@ Since BullMQ supports global pause, one possible strategy, if suitable for your
 ### Use new queues altogether
 
 This drastic solution involves discontinuing use of older queues and creating new ones. You could rename older queues (e.g., "myQueueV2"), use a new Redis host, or maintain two versions of the service—one running an older BullMQ version with old queues, and a newer one with the latest BullMQ and a different set of queues. When the older version has no more jobs to process, it can be retired, leaving only the upgraded version.
-

docs/gitbook/guide/migrations/v6.md

@@ -0,0 +1,27 @@
+---
+description: Tips and hints on how to migrate to v6.
+---
+
+# Migration to v6
+
+Make sure to pass **skipMigrationsExecution** option in any of our instances as false in order to execute all necessary changes when coming from an older version
+
+## Migration of deprecated paused key
+
+If you have paused queues after upgrading to this version. These jobs will be moved to wait state when initializing any of our instances (Worker, Queue, QueueEvents or FlowProducer).
+
+Paused key is not longer needed as this state is already represented by queue meta key. It also improve the process of pausing or resuming a queue as we don't need to rename any key.
+
+## Remove legacy markers
+
+When migrating from versions before v5.
+It's recommended to do this process:
+
+1. Pause your queues.
+2. Upgrade to v6.
+3. Instantiate any instance where migrations will be executed.
+4. Resume your queues.
+
+This way you will prevent that your workers pick a legacy marker.
+
+A second option would be to do incremental upgrades.

python/bullmq/queue.py

@@ -141,7 +141,7 @@ async def getJobLogs(self, job_id:str, start = 0, end = -1, asc = True):
             "logs": result[0],
             "count": result[1]
         }
-   
+
     async def obliterate(self, force: bool = False):
         """
         Completely destroys the queue and all of its contents irreversibly.
@@ -198,6 +198,17 @@ def removeDeprecatedPriorityKey(self):
         """
         return self.client.delete(self.toKey("priority"))
 
+    async def migrateDeprecatedPausedKey(self, maxCount: int = 1000):
+        """
+        Migrate deprecated paused key.
+
+        @param maxCount: Max quantity of jobs to be moved to wait per iteration.
+        """
+        while True:
+            cursor = await self.scripts.migrateDeprecatedPausedKey(maxCount)
+            if cursor is None or cursor == 0 or cursor == "0":
+                break
+
     async def getJobCountByTypes(self, *types):
         result = await self.getJobCounts(*types)
         sum = 0

python/bullmq/scripts.py

@@ -38,11 +38,11 @@ def __init__(self, prefix: str, queueName: str, redisConnection: RedisConnection
         self.redisConnection = redisConnection
         self.redisClient = redisConnection.conn
         self.commands = {
-            "addStandardJob": self.redisClient.register_script(self.getScript("addStandardJob-8.lua")),
+            "addStandardJob": self.redisClient.register_script(self.getScript("addStandardJob-7.lua")),
             "addDelayedJob": self.redisClient.register_script(self.getScript("addDelayedJob-6.lua")),
             "addParentJob": self.redisClient.register_script(self.getScript("addParentJob-4.lua")),
             "addPrioritizedJob": self.redisClient.register_script(self.getScript("addPrioritizedJob-8.lua")),
-            "changePriority": self.redisClient.register_script(self.getScript("changePriority-7.lua")),
+            "changePriority": self.redisClient.register_script(self.getScript("changePriority-6.lua")),
             "cleanJobsInSet": self.redisClient.register_script(self.getScript("cleanJobsInSet-3.lua")),
             "extendLock": self.redisClient.register_script(self.getScript("extendLock-2.lua")),
             "getCounts": self.redisClient.register_script(self.getScript("getCounts-1.lua")),
@@ -51,18 +51,19 @@ def __init__(self, prefix: str, queueName: str, redisConnection: RedisConnection
             "getState": self.redisClient.register_script(self.getScript("getState-8.lua")),
             "getStateV2": self.redisClient.register_script(self.getScript("getStateV2-8.lua")),
             "isJobInList": self.redisClient.register_script(self.getScript("isJobInList-1.lua")),
-            "moveStalledJobsToWait": self.redisClient.register_script(self.getScript("moveStalledJobsToWait-9.lua")),
-            "moveToActive": self.redisClient.register_script(self.getScript("moveToActive-11.lua")),
+            "moveStalledJobsToWait": self.redisClient.register_script(self.getScript("moveStalledJobsToWait-8.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-14.lua")),
+            "moveToFinished": self.redisClient.register_script(self.getScript("moveToFinished-13.lua")),
             "moveToWaitingChildren": self.redisClient.register_script(self.getScript("moveToWaitingChildren-5.lua")),
             "obliterate": self.redisClient.register_script(self.getScript("obliterate-2.lua")),
             "pause": self.redisClient.register_script(self.getScript("pause-7.lua")),
-            "promote": self.redisClient.register_script(self.getScript("promote-9.lua")),
+            "promote": self.redisClient.register_script(self.getScript("promote-8.lua")),
             "removeJob": self.redisClient.register_script(self.getScript("removeJob-2.lua")),
-            "reprocessJob": self.redisClient.register_script(self.getScript("reprocessJob-8.lua")),
-            "retryJob": self.redisClient.register_script(self.getScript("retryJob-11.lua")),
-            "moveJobsToWait": self.redisClient.register_script(self.getScript("moveJobsToWait-8.lua")),
+            "migrateDeprecatedPausedKey": self.redisClient.register_script(self.getScript("migrateDeprecatedPausedKey-2.lua")),
+            "reprocessJob": self.redisClient.register_script(self.getScript("reprocessJob-7.lua")),
+            "retryJob": self.redisClient.register_script(self.getScript("retryJob-10.lua")),
+            "moveJobsToWait": self.redisClient.register_script(self.getScript("moveJobsToWait-7.lua")),
             "saveStacktrace": self.redisClient.register_script(self.getScript("saveStacktrace-1.lua")),
             "updateData": self.redisClient.register_script(self.getScript("updateData-1.lua")),
             "updateProgress": self.redisClient.register_script(self.getScript("updateProgress-3.lua")),
@@ -131,7 +132,7 @@ def addStandardJob(self, job: Job, timestamp: int, pipe = None):
         """
         Add a standard job to the queue
         """
-        keys = self.getKeys(['wait', 'paused', 'meta', 'id',
+        keys = self.getKeys(['wait', 'meta', 'id',
                              'completed', 'active', 'events', 'marker'])
         args = self.addJobArgs(job, None)
         args.append(timestamp)
@@ -259,15 +260,15 @@ def saveStacktraceArgs(self, job_id: str, stacktrace: str, failedReason: str):
         return (keys, args)
 
     def retryJobArgs(self, job_id: str, lifo: bool, token: str, opts: dict = {}):
-        keys = self.getKeys(['active', 'wait', 'paused'])
+        keys = self.getKeys(['active', 'wait'])
         keys.append(self.toKey(job_id))
         keys.append(self.keys['meta'])
         keys.append(self.keys['events'])
         keys.append(self.keys['delayed'])
         keys.append(self.keys['prioritized'])
         keys.append(self.keys['pc'])
-        keys.append(self.keys['marker'])
         keys.append(self.keys['stalled'])
+        keys.append(self.keys['marker'])
 
         push_cmd = "RPUSH" if lifo else "LPUSH"
 
@@ -302,7 +303,6 @@ def promoteArgs(self, job_id: str):
         keys = self.getKeys(['delayed', 'wait', 'paused', 'meta', 'prioritized', 'active', 'pc', 'events', 'marker'])
         keys.append(self.toKey(job_id))
         keys.append(self.keys['events'])
-        keys.append(self.keys['paused'])
         keys.append(self.keys['meta'])
 
         args = [self.keys[''], job_id]
@@ -374,7 +374,6 @@ async def isJobInList(self, list_key: str, job_id: str):
 
     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['prioritized'],
             self.keys['active'],
@@ -408,7 +407,6 @@ async def reprocessJob(self, job: Job, state: str):
         keys.append(self.keys[state])
         keys.append(self.keys['wait'])
         keys.append(self.keys['meta'])
-        keys.append(self.keys['paused'])
         keys.append(self.keys['active'])
         keys.append(self.keys['marker'])
 
@@ -450,7 +448,7 @@ async def obliterate(self, count: int, force: bool = False):
 
     def moveJobsToWaitArgs(self, state: str, count: int, timestamp: int) -> int:
         keys = self.getKeys(
-            ['', 'events', state, 'wait', 'paused', 'meta', 'active', 'marker'])
+            ['', 'events', state, 'wait', 'meta', 'active', 'marker'])
 
         args = [count or 1000, timestamp or round(time.time()*1000), state]
         return (keys, args)
@@ -465,6 +463,15 @@ async def retryJobs(self, state: str, count: int, timestamp: int):
         result = await self.commands["moveJobsToWait"](keys=keys, args=args)
         return result
 
+    async def migrateDeprecatedPausedKey(self, maxCount: int):
+        keys = self.getKeys(
+            ['paused', 'wait'])
+
+        args = [maxCount]
+
+        result = await self.commands["migrateDeprecatedPausedKey"](keys=keys, args=args)
+        return result
+
     async def promoteJobs(self, count: int):
         """
         Promote jobs in delayed state
@@ -483,7 +490,7 @@ async def moveToActive(self, token: str, opts: dict) -> list[Any]:
         limiter = opts.get("limiter", None)
 
         keys = self.getKeys(['wait', 'active', 'prioritized', 'events',
-                            'stalled', 'limiter', 'delayed', 'paused', 'meta', 'pc', 'marker'])
+                            'stalled', 'limiter', 'delayed', 'meta', 'pc', 'marker'])
         packedOpts = msgpack.packb(
             {"token": token, "lockDuration": lockDuration, "limiter": limiter}, use_bin_type=True)
         args = [self.keys[''], timestamp, packedOpts]
@@ -516,7 +523,7 @@ def moveToFinishedArgs(self, job: Job, val: Any, propVal: str, shouldRemove, tar
         metricsKey = self.toKey('metrics:' + target)
 
         keys = self.getKeys(['wait', 'active', 'prioritized', 'events',
-                            'stalled', 'limiter', 'delayed', 'paused', 'meta', 'pc', target])
+                            'stalled', 'limiter', 'delayed', 'meta', 'pc', target])
         keys.append(self.toKey(job.id))
         keys.append(metricsKey)
         keys.append(self.keys['marker'])
@@ -580,7 +587,7 @@ def extendLock(self, jobId: str, token: str, duration: int, client: Redis = None
 
     def moveStalledJobsToWait(self, maxStalledCount: int, stalledInterval: int):
         keys = self.getKeys(['stalled', 'wait', 'active', 'failed',
-                            'stalled-check', 'meta', 'paused', 'marker', 'events'])
+                            'stalled-check', 'meta', 'marker', 'events'])
         args = [maxStalledCount, self.keys[''], round(
             time.time() * 1000), stalledInterval]
         return self.commands["moveStalledJobsToWait"](keys, args)

src/classes/job.ts

@@ -39,10 +39,11 @@ const logger = debuglog('bull');
 
 const optsDecodeMap = {
   de: 'deduplication',
-  ocf: 'onChildFailure',
   fpof: 'failParentOnFailure',
   idof: 'ignoreDependencyOnFailure',
   kl: 'keepLogs',
+  ocf: 'onChildFailure',
+  rdof: 'removeDependencyOnFailure',
 };
 
 const optsEncodeMap = invertObject(optsDecodeMap);

src/classes/migrations.ts

@@ -0,0 +1,116 @@
+import { RedisClient } from '../interfaces';
+
+export interface MigrationOptions {
+  prefix: string;
+  queueName: string;
+}
+
+export type MigrationFunction = (
+  client: RedisClient,
+  opts: MigrationOptions,
+) => Promise<void>;
+
+export const checkPendingMigrations = async (
+  client: RedisClient,
+  opts: MigrationOptions,
+) => {
+  const migrationsKey = getRedisKeyFromOpts(opts, 'migrations');
+  const existingMigrations = await client.zrange(migrationsKey, 0, -1);
+  return migrations.some(
+    migration =>
+      !existingMigrations.includes(`${migration.version}-${migration.name}`),
+  );
+};
+
+export const migrations: {
+  name: string;
+  version: string;
+  migrate: MigrationFunction;
+}[] = [
+  {
+    name: 'remove-legacy-markers',
+    version: '6.0.0',
+    migrate: async (client: RedisClient, opts: MigrationOptions) => {
+      const keys: (string | number)[] = [
+        getRedisKeyFromOpts(opts, 'wait'),
+        getRedisKeyFromOpts(opts, 'paused'),
+        getRedisKeyFromOpts(opts, 'meta'),
+        getRedisKeyFromOpts(opts, 'completed'),
+        getRedisKeyFromOpts(opts, 'failed'),
+      ];
+      const args = [getRedisKeyFromOpts(opts, '')];
+
+      await (<any>client).removeLegacyMarkers(keys.concat(args));
+    },
+  },
+  {
+    name: 'migrate-paused-jobs',
+    version: '6.0.0',
+    migrate: async (client: RedisClient, opts: MigrationOptions) => {
+      let cursor = 0;
+      do {
+        const keys: (string | number)[] = [
+          getRedisKeyFromOpts(opts, 'paused'),
+          getRedisKeyFromOpts(opts, 'wait'),
+        ];
+        const args = [1000];
+        cursor = await (<any>client).migrateDeprecatedPausedKey(
+          keys.concat(args),
+        );
+      } while (cursor);
+    },
+  },
+];
+
+/**
+ * Run Migrations.
+ *
+ * This method is used to run possibly existing migrations for the queue.
+ *
+ * Normally, if there are pending migrations, the Queue, Worker and QueueEvents instances
+ * will throw an error when they are instantiated. Use then this method to run the migrations
+ * before instantiating the instances.
+ *
+ * @param redisClient The Redis client instance
+ * @param opts The options for the migration
+ *
+ * @sa https://docs.bullmq.io/guide/migrations
+ */
+export const runMigrations = async (
+  redisClient: RedisClient,
+  opts: {
+    prefix?: string;
+    queueName: string;
+  },
+) => {
+  const prefix = opts.prefix || 'bull';
+  const migrationsKey = getRedisKeyFromOpts({ prefix, ...opts }, 'migrations');
+
+  // The migrations key is a ZSET with the migration timestamp as the score
+  for (const migration of migrations) {
+    const migrationId = `${migration.version}-${migration.name}`;
+    const pendingMigration = !!(await redisClient.zscore(
+      migrationsKey,
+      migrationId,
+    ));
+    if (pendingMigration) {
+      continue;
+    }
+    console.log(`[BULLMQ] Running migration ${migrationId}`);
+    try {
+      await migration.migrate(redisClient, {
+        prefix,
+        queueName: opts.queueName,
+      });
+      await redisClient.zadd(migrationsKey, Date.now(), migrationId);
+    } catch (err) {
+      console.error(`[BULLMQ] Migration ${migrationId} failed: ${err}`);
+      break;
+    }
+    console.log(`[BULLMQ] Migration ${migrationId} completed`);
+  }
+};
+
+function getRedisKeyFromOpts(opts: MigrationOptions, key: string): string {
+  return `${opts.prefix}:${opts.queueName}:${key}`;
+}