Fixed process-pool for sync tasks.

Author: s3rius

a.py

@@ -0,0 +1,19 @@
+import asyncio
+
+from taskiq_redis import ListQueueBroker
+
+broker = ListQueueBroker("redis://localhost")
+
+
+@broker.task
+async def my_task():
+    pass
+
+
+async def main():
+    async with broker:
+        await my_task.kiq()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

taskiq/receiver/receiver.py

@@ -46,15 +46,6 @@ def _execute_sync_task_in_executor(
     return target(*args, **kwargs)
 
 
-def _is_process_pool_executor(executor: Executor) -> bool:
-    """Check if the executor is a ProcessPoolExecutor.
-
-    :param executor: executor instance
-    :return: True if it's a ProcessPoolExecutor
-    """
-    return isinstance(executor, ProcessPoolExecutor)
-
-
 class Receiver:
     """Class that uses as a callback handler."""
 
@@ -96,6 +87,7 @@ def __init__(
                 "can result in undefined behavior",
             )
         self.sem_prefetch = asyncio.Semaphore(max_prefetch)
+        self.is_process_pool = isinstance(executor, ProcessPoolExecutor)
 
     async def callback(  # noqa: C901, PLR0912
         self,
@@ -274,7 +266,7 @@ async def run_task(  # noqa: C901, PLR0912, PLR0915
                 is_coroutine = False
                 # If this is a synchronous function, we
                 # run it in executor and preserve the context.
-                if _is_process_pool_executor(self.executor):
+                if self.is_process_pool:
                     # For ProcessPoolExecutor, we can't use ctx.run because it contains
                     # a reference to contextvars.Context which cannot be pickled.
                     # Instead, we call the target function directly in the executor.
@@ -284,7 +276,7 @@ async def run_task(  # noqa: C901, PLR0912, PLR0915
                         self.executor,
                         _execute_sync_task_in_executor,
                         target,
-                        message.args,
+                        tuple(message.args),
                         kwargs,
                     )
                 else: