fix: timeout

Author: Anton

pyproject.toml

@@ -26,7 +26,7 @@ classifiers = [
 keywords = ["taskiq", "tasks", "distributed", "async"]
 
 [tool.poetry.dependencies]
-python = "^3.7"
+python = "^3.10"
 typing-extensions = ">=3.10.0.0"
 pydantic = "^1.6.2"
 importlib-metadata = "*"

taskiq/receiver/receiver.py

@@ -351,9 +351,10 @@ async def task_idler(self, wait: float) -> None:
             self.sem.release()
 
             # Wait
-            await asyncio.sleep(wait - (time() - start_time))
+            await asyncio.sleep(max(wait - (time() - start_time), 0))
 
             # Decrease max_prefetch in runner
-            await self.queue.put_first(QUEUE_SKIP)
+            task = asyncio.create_task(self.queue.put_first(QUEUE_SKIP))
             # Decrease max_tasks
             await self.sem.acquire_first()
+            await task

tests/cli/worker/test_receiver.py

@@ -332,7 +332,7 @@ async def task_add_one(val: int) -> int:
     @broker.task
     async def task_map(vals: List[int], ctx: Context = Depends()) -> List[int]:
         tasks = [await task_add_one.kiq(val) for val in vals]
-        await ctx.task_idler(0.1)
+        await ctx.task_idler(0.5)
         resps_tasks = [asyncio.create_task(t.wait_result(timeout=1)) for t in tasks]
         resps = await asyncio.gather(*resps_tasks)
         res = [r.return_value for r in resps]