perf: keep jobs in waiting list when queue is paused

docs/gitbook/SUMMARY.md

@@ -70,7 +70,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/bullmq-pro/changelog.md

@@ -1,9 +1,58 @@
+## [7.20.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.20.0...v7.20.1) (2024-11-10)
+
+
+### Bug Fixes
+
+* **repeatable:** only apply immediately in the first iteration ([f69cfbc](https://github.com/taskforcesh/bullmq/commit/f69cfbcbc5516a854adbbc29b259d08e65a19705))
+
+# [7.20.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.19.0...v7.20.0) (2024-11-09)
+
+
+### Bug Fixes
+
+* **scripts:** set package version by default for extension ([#2887](https://github.com/taskforcesh/bullmq/issues/2887)) ([b955340](https://github.com/taskforcesh/bullmq/commit/b955340b940e4c1e330445526cd572e0ab25daa9))
+* **worker:** allow retrieving concurrency value ([#2883](https://github.com/taskforcesh/bullmq/issues/2883)) fixes [#2880](https://github.com/taskforcesh/bullmq/issues/2880) ([52f6317](https://github.com/taskforcesh/bullmq/commit/52f6317ecd2080a5c9684a4fe384e20d86f21de4))
+* **connection:** set packageVersion as protected attribute for extension ([#2884](https://github.com/taskforcesh/bullmq/issues/2884)) ([411ccae](https://github.com/taskforcesh/bullmq/commit/411ccae9419e008d916be6cf71c4d57dd2a07b2b))
+
+
+### Features
+
+* **queue-events:** add QueueEventsProducer for publishing custom events ([#2844](https://github.com/taskforcesh/bullmq/issues/2844)) ([5eb03cd](https://github.com/taskforcesh/bullmq/commit/5eb03cd7f27027191eb4bc4ed7386755fd9be1fb))
+* **flows:** add telemetry support ([#2879](https://github.com/taskforcesh/bullmq/issues/2879)) ([5ed154b](https://github.com/taskforcesh/bullmq/commit/5ed154ba240dbe9eb5c22e27ad02e851c0f3cf69))
+
+# [7.19.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.18.0...v7.19.0) (2024-11-08)
+
+
+### Bug Fixes
+
+* **deps:** bump msgpackr to 1.1.2 to resolve ERR_BUFFER_OUT_OF_BOUNDS error ([#2882](https://github.com/taskforcesh/bullmq/issues/2882)) ref [#2747](https://github.com/taskforcesh/bullmq/issues/2747) ([4d2136c](https://github.com/taskforcesh/bullmq/commit/4d2136cc6ba340e511a539c130c9a739fe1055d0))
+
+
+### Features
+
+* **scheduler:** add getJobScheduler method ([#2877](https://github.com/taskforcesh/bullmq/issues/2877)) ref [#2875](https://github.com/taskforcesh/bullmq/issues/2875) ([956d98c](https://github.com/taskforcesh/bullmq/commit/956d98c6890484742bb080919c70692234f28c69))
+* **queue:** add a telemetry interface ([#2721](https://github.com/taskforcesh/bullmq/issues/2721)) ([273b574](https://github.com/taskforcesh/bullmq/commit/273b574e6b5628680990eb02e1930809c9cba5bb))
+
+# [7.18.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.17.2...v7.18.0) (2024-11-07)
+
+
+### Bug Fixes
+
+* proper way to get version ([b4e25c1](https://github.com/taskforcesh/bullmq/commit/b4e25c13cafc001748ee6eb590133feb8ee24d7b))
+* **scripts:** add missing wait in isJobInList ([9ef865c](https://github.com/taskforcesh/bullmq/commit/9ef865c7de6086cb3c906721fd046aeed1e0d27f))
+* **redis:** use version for naming loaded lua scripts ([fe73f6d](https://github.com/taskforcesh/bullmq/commit/fe73f6d4d776dc9f99ad3a094e5c59c5fafc96f1))
+
+
+### Features
+
+* **queue:** add option to skip metas update ([b7dd925](https://github.com/taskforcesh/bullmq/commit/b7dd925e7f2a4468c98a05f3a3ca1a476482b6c0))
+* **queue:** add queue version support ([#2822](https://github.com/taskforcesh/bullmq/issues/2822)) ([3a4781b](https://github.com/taskforcesh/bullmq/commit/3a4781bf7cadf04f6a324871654eed8f01cdadae))
+
 ## [7.17.2](https://github.com/taskforcesh/bullmq-pro/compare/v7.17.1...v7.17.2) (2024-10-23)
 
 
 ### Bug Fixes
 
-* **repeatable:** export RepeatOptions ([#261](https://github.com/taskforcesh/bullmq-pro/issues/261)) ([b620bdf](https://github.com/taskforcesh/bullmq-pro/commit/b620bdf4f7449ad20f0ffd07786880115ec77fd9))
 * **sandbox:** fix serialization of error with circular references are present ([#2815](https://github.com/taskforcesh/bullmq/issues/2815)) fix [#2813](https://github.com/taskforcesh/bullmq/issues/2813) ([a384d92](https://github.com/taskforcesh/bullmq/commit/a384d926bee15bffa84178a8fad7b94a6a08b572))
 
 ## [7.17.1](https://github.com/taskforcesh/bullmq-pro/compare/v7.17.0...v7.17.1) (2024-10-18)
@@ -16,6 +65,12 @@
 # [7.17.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.16.0...v7.17.0) (2024-10-12)
 
 
+### Bug Fixes
+
+* **repeat:** also consider startDate when using "every" ([25bbaa8](https://github.com/taskforcesh/bullmq/commit/25bbaa81af87f9944a64bc4fb7e0c76ef223ada4))
+* **sandbox:** catch exit errors ([#2800](https://github.com/taskforcesh/bullmq/issues/2800)) ([6babb9e](https://github.com/taskforcesh/bullmq/commit/6babb9e2f355feaf9bd1a8ed229c1001e6de7144))
+
+
 ### Features
 
 * **repeat:** deprecate immediately on job scheduler ([ed047f7](https://github.com/taskforcesh/bullmq/commit/ed047f7ab69ebdb445343b6cb325e90b95ee9dc5))
@@ -25,11 +80,6 @@
 * **worker-fork:** allow passing fork options ([#2795](https://github.com/taskforcesh/bullmq/issues/2795)) ([f7a4292](https://github.com/taskforcesh/bullmq/commit/f7a4292e064b41236f4489b3d7785a4c599a6435))
 * **worker-thread:** allow passing Worker options ([#2791](https://github.com/taskforcesh/bullmq/issues/2791)) ref [#1555](https://github.com/taskforcesh/bullmq/issues/1555) ([6a1f7a9](https://github.com/taskforcesh/bullmq/commit/6a1f7a9f0303561d6ec7b2005ba0227132b89e07))
 
-### Bug Fixes
-
-* **repeat:** also consider startDate when using "every" ([25bbaa8](https://github.com/taskforcesh/bullmq/commit/25bbaa81af87f9944a64bc4fb7e0c76ef223ada4))
-* **sandbox:** catch exit errors ([#2800](https://github.com/taskforcesh/bullmq/issues/2800)) ([6babb9e](https://github.com/taskforcesh/bullmq/commit/6babb9e2f355feaf9bd1a8ed229c1001e6de7144))
-
 # [7.16.0](https://github.com/taskforcesh/bullmq-pro/compare/v7.15.4...v7.16.0) (2024-09-24)
 
 

docs/gitbook/changelog.md

@@ -1,3 +1,24 @@
+## [5.26.2](https://github.com/taskforcesh/bullmq/compare/v5.26.1...v5.26.2) (2024-11-15)
+
+
+### Bug Fixes
+
+* **telemetry:** do not set span on parent context if undefined ([c417a23](https://github.com/taskforcesh/bullmq/commit/c417a23bb28d9effa42115e954b18cc41c1fc043))
+
+## [5.26.1](https://github.com/taskforcesh/bullmq/compare/v5.26.0...v5.26.1) (2024-11-14)
+
+
+### Bug Fixes
+
+* **queue:** fix generics to be able to properly be extended ([f2495e5](https://github.com/taskforcesh/bullmq/commit/f2495e5ee9ecdb26492da510dc38730718cb28c5))
+
+# [5.26.0](https://github.com/taskforcesh/bullmq/compare/v5.25.6...v5.26.0) (2024-11-14)
+
+
+### Features
+
+* improve queue getters to use generic job type ([#2905](https://github.com/taskforcesh/bullmq/issues/2905)) ([c9531ec](https://github.com/taskforcesh/bullmq/commit/c9531ec7a49126a017611eb2fd2eaea8fcb5ada5))
+
 ## [5.25.6](https://github.com/taskforcesh/bullmq/compare/v5.25.5...v5.25.6) (2024-11-11)
 
 

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,25 @@
+---
+description: Tips and hints on how to migrate to v6.
+---
+
+# Migration to v6
+
+Make sure to call **runMigrations** method from Queue class 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 inside meta key. It also improves 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 a Queue instance and execute runMigrations method where migrations will be executed.
+4. Resume your queues.
+
+This way you will prevent that your workers pick a legacy marker that is no longer used because new markers are added in a different structure.

docs/gitbook/guide/redis-tm-compatibility/README.md

@@ -1,4 +1,4 @@
 # Redis™ Compatibility
 
-There are several alternatives for Redis and even though BullMQ is full Redis™ compliant, not all the alternatives are going to work properly. In this section we present the vendors that officially support BullMQ and that we regularly test to verify they keep staying compatible.
+There are several alternatives for Redis and even though BullMQ is full Redis™ compliant with version 6.2.0 or newer, not all the alternatives are going to work properly. In this section we present the vendors that officially support BullMQ and that we regularly test to verify they keep staying compatible.
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "bullmq",
-  "version": "5.25.6",
+  "version": "5.26.2",
   "description": "Queue for messages and jobs based on Redis",
   "homepage": "https://bullmq.io/",
   "main": "./dist/cjs/index.js",

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.

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-scheduler.ts

@@ -295,13 +295,3 @@ export const defaultRepeatStrategy = (
     // Ignore error
   }
 };
-
-function removeUndefinedFields(obj: Record<string, any>) {
-  const newObj: Record<string, any> = {};
-  for (const key in obj) {
-    if (obj[key] !== undefined) {
-      newObj[key] = obj[key];
-    }
-  }
-  return newObj;
-}

src/classes/job.ts

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