Merge branch 'master' of https://github.com/taskforcesh/bullmq into f…

…ix/force-timeout-for-bzpopmin

docs/gitbook/changelog.md

@@ -1,3 +1,15 @@
+## [5.7.6](https://github.com/taskforcesh/bullmq/compare/v5.7.5...v5.7.6) (2024-04-27)
+
+
+### Bug Fixes
+
+* **redis-connection:** increase redis retry strategy backoff ([#2546](https://github.com/taskforcesh/bullmq/issues/2546)) [python] ([6cf7712](https://github.com/taskforcesh/bullmq/commit/6cf77122da845e5b0afa1607348cf06602679329))
+
+
+### Performance Improvements
+
+* **worker:** do not call bzpopmin when blockDelay is lower or equal 0 ([#2544](https://github.com/taskforcesh/bullmq/issues/2544)) ref [#2466](https://github.com/taskforcesh/bullmq/issues/2466) ([9760b85](https://github.com/taskforcesh/bullmq/commit/9760b85dfbcc9b3c744f616961ef939e8951321d))
+
 ## [5.7.5](https://github.com/taskforcesh/bullmq/compare/v5.7.4...v5.7.5) (2024-04-24)
 
 

package.json

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

python/bullmq/__init__.py

@@ -3,7 +3,7 @@
 
 A background job processor and message queue for Python based on Redis.
 """
-__version__ = "2.7.3"
+__version__ = "2.7.5"
 __author__ = 'Taskforce.sh Inc.'
 __credits__ = 'Taskforce.sh Inc.'
 

python/bullmq/redis_connection.py

@@ -24,7 +24,7 @@ class RedisConnection:
 
     def __init__(self, redisOpts: dict | str | redis.Redis = {}):
         self.version = None
-        retry = Retry(ExponentialBackoff(), 3)
+        retry = Retry(ExponentialBackoff(cap=20, base=1), 20)
         retry_errors = [BusyLoadingError, ConnectionError, TimeoutError]
 
         if isinstance(redisOpts, redis.Redis):

python/bullmq/worker.py

@@ -243,13 +243,18 @@ async def runStalledJobsCheck(self):
 
     async def close(self, force: bool = False):
         """
-        Close the worker
+        Closes the worker and related redis connections.
+
+        This method waits for current jobs to finalize before returning.
         """
         self.closing = True
         if force:
             self.forceClosing = True
             self.cancelProcessing()
 
+        if not force and len(self.processing) > 0:
+            await asyncio.wait(self.processing, return_when=asyncio.ALL_COMPLETED)
+
         await self.blockingRedisConnection.close()
         await self.redisConnection.close()
 

python/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "bullmq"
-version = "2.7.3"
+version = "2.7.5"
 description='BullMQ for Python'
 readme="README.md"
 authors = [

src/classes/worker.ts

@@ -621,35 +621,32 @@ will never work with more accuracy than 1ms. */
       if (!this.closing) {
         let blockTimeout = this.getBlockTimeout(blockUntil);
 
-        blockTimeout = this.blockingConnection.capabilities.canDoubleTimeout
-          ? blockTimeout
-          : Math.ceil(blockTimeout);
-
-        // We restrict the maximum block timeout to 10 second to avoid
-        // blocking the connection for too long in the case of reconnections
-        // reference: https://github.com/taskforcesh/bullmq/issues/1658
-        blockTimeout = Math.min(blockTimeout, maximumBlockTimeout);
-
-        // We cannot trust that the blocking connection stays blocking forever
-        // due to issues in Redis and IORedis, so we will reconnect if we
-        // don't get a response in the expected time.
-        const timeout = setTimeout(async () => {
-          await this.blockingConnection.disconnect();
-          await this.blockingConnection.reconnect();
-        }, blockTimeout * 1000 + 1000);
-
-        this.updateDelays(); // reset delays to avoid reusing same values in next iteration
-
-        // Markers should only be used for un-blocking, so we will handle them in this
-        // function only.
-        const result = await bclient.bzpopmin(this.keys.marker, blockTimeout);
-        clearTimeout(timeout);
-
-        if (result) {
-          const [_key, member, score] = result;
-
-          if (member) {
-            return parseInt(score);
+        if (blockTimeout > 0) {
+          blockTimeout = this.blockingConnection.capabilities.canDoubleTimeout
+            ? blockTimeout
+            : Math.ceil(blockTimeout);
+
+          // We cannot trust that the blocking connection stays blocking forever
+          // due to issues in Redis and IORedis, so we will reconnect if we
+          // don't get a response in the expected time.
+          const timeout = setTimeout(async () => {
+            await this.blockingConnection.disconnect();
+            await this.blockingConnection.reconnect();
+          }, blockTimeout * 1000 + 1000);
+
+          this.updateDelays(); // reset delays to avoid reusing same values in next iteration
+
+          // Markers should only be used for un-blocking, so we will handle them in this
+          // function only.
+          const result = await bclient.bzpopmin(this.keys.marker, blockTimeout);
+          clearTimeout(timeout);
+
+          if (result) {
+            const [_key, member, score] = result;
+
+            if (member) {
+              return parseInt(score);
+            }
           }
         }
 
@@ -673,7 +670,9 @@ will never work with more accuracy than 1ms. */
     if (blockUntil) {
       const blockDelay = blockUntil - Date.now();
       // when we reach the time to get new jobs
-      if (blockDelay < this.minimumBlockTimeout * 1000) {
+      if (blockDelay <= 0) {
+        return blockDelay;
+      } else if (blockDelay < this.minimumBlockTimeout * 1000) {
         return this.minimumBlockTimeout;
       } else {
         // We restrict the maximum block timeout to 10 second to avoid

tests/test_worker.ts

@@ -949,23 +949,17 @@ describe('workers', function () {
 
     describe('when blockUntil is greater than 0', () => {
       describe('when blockUntil is lower than date now value', () => {
-        it('returns minimumBlockTimeout', async () => {
+        it('returns blockDelay value lower or equal 0', async () => {
           const worker = new Worker(queueName, async () => {}, {
             connection,
             prefix,
             autorun: false,
           });
           await worker.waitUntilReady();
 
-          if (isRedisVersionLowerThan(worker.redisVersion, '7.0.8')) {
-            expect(worker['getBlockTimeout'](Date.now() - 1)).to.be.equal(
-              0.002,
-            );
-          } else {
-            expect(worker['getBlockTimeout'](Date.now() - 1)).to.be.equal(
-              0.001,
-            );
-          }
+          expect(
+            worker['getBlockTimeout'](Date.now() - 1),
+          ).to.be.lessThanOrEqual(0);
           await worker.close();
         });
       });