worker lock prolongate

README.md

@@ -202,9 +202,10 @@ Worker pause in seconds on empty queue
 
 You can import and change settings manually
 ```py
-from sharded_queue.settings import settings
 from sharded_queue import Queue, Worker
-from sharded_queue.drivers import RuntimeLock RuntimeStorage
+from sharded_queue.drivers import RuntimeLock, RuntimeStorage
+from sharded_queue.settings import settings
+
 settings.worker_acquire_delay = 5
 settings.worker_batch_size = 64
 

sharded_queue/__init__.py

@@ -228,10 +228,16 @@ async def process(self, tube: Tube, limit: Optional[int] = None) -> int:
                         break
                     await sleep(settings.worker_empty_pause)
 
-        if tube.handler is DeferredHandler:
-            await self.lock.ttl(tube.pipe, settings.deferred_retry_delay)
-        else:
-            await self.lock.release(tube.pipe)
+                if tube.handler is DeferredHandler:
+                    await self.lock.ttl(
+                        key=tube.pipe,
+                        ttl=settings.deferred_retry_delay,
+                    )
+                    return processed_counter
+                else:
+                    await self.lock.ttl(tube.pipe, settings.lock_timeout)
+
+        await self.lock.release(tube.pipe)
 
         return processed_counter
 

tests/test_lock.py

@@ -1,8 +1,41 @@
+from asyncio import gather, sleep
+
 from pytest import mark
 from redis.asyncio import Redis
 
-from sharded_queue.drivers import RedisLock, RuntimeLock
+from sharded_queue import Handler, Queue, Route, Tube, Worker
+from sharded_queue.drivers import RedisLock, RuntimeLock, RuntimeStorage
 from sharded_queue.protocols import Lock
+from sharded_queue.settings import settings
+
+
+class Request:
+    ...
+
+
+class TestHandler(Handler):
+    async def handle(self, *requests: Request) -> None:
+        ...
+
+
+@mark.asyncio
+async def test_lock_prolongate() -> None:
+    queue: Queue = Queue(RuntimeStorage())
+    redis: Redis = Redis(decode_responses=True)
+    await redis.flushall()
+    worker: Worker = Worker(RedisLock(redis), queue)
+    pipe: str = settings.lock_prefix + Tube(TestHandler, Route()).pipe
+
+    async def register_task():
+        ttl1 = await redis.ttl(pipe)
+        await sleep(1)
+        ttl2 = await redis.ttl(pipe)
+        assert ttl1 == ttl2, 'ttl is updated'
+        await queue.register(TestHandler, Request())
+
+    await queue.register(TestHandler, Request())
+    await gather(worker.loop(2), register_task())
+    await redis.close()
 
 
 @mark.asyncio