Repository navigation
TaskiqAdminMiddleware adds significant latency to task triggering #8
Description
Activity
Just wanted to add one more point: Instead of reporting task status via HTTP, a better pattern might be to publish task events to the message broker (e.g., Redis/RabbitMQ) and have the admin service consume them asynchronously. This avoids blocking tasks on network requests and removes the per-task overhead of HTTP entirely.
This is actually how Celery and Flower handle monitoring:
- Celery workers emit task events (e.g.,
task-started,task-succeeded) directly to the broker. - Flower subscribes to these events in real-time without workers needing to make HTTP calls.
Adopting a similar event-driven approach would likely scale much better—especially under heavy load or in batch-processing scenarios—while keeping the system decoupled and efficient.
- Celery workers emit task events (e.g.,
Hey! Thank you for such comprehensive description of the issue
Yes, we are aware of this problem and will fix it asapLet's then do this in 2 steps: firstly we will get rid of the overhead that middleware provides in the current version, and then we will see how can we implement event-based flower-like approach.
Reacted by nimaxin and Alexander SmirnovReacted by nimaxin, Alexander Smirnov and AthroniaethHi! Are there any updates on this issue?
Reacted by Athroniaeth and Alexander SmirnovHi! Are there any updates on this issue?
I also wanted to know if modifying the middleware to use a client as an attribute and using a
fire_and_forgetcould solve this problem? I was thinking of something like this:import asyncio from typing import Any from urllib.parse import urljoin from datetime import datetime, UTC import httpx from taskiq import TaskiqMiddleware, TaskiqResult, TaskiqMessage def fire_and_forget(coro): """Utility to run async tasks in background.""" asyncio.create_task(coro) class TaskiqAdminMiddleware(TaskiqMiddleware): def __init__( self, url: str, api_token: str, taskiq_broker_name: str | None = None, ): super().__init__() self.url = url self.client = httpx.AsyncClient() # Reuse the client self.api_token = api_token self.__ta_broker_name = taskiq_broker_name async def post_send(self, message): now = datetime.now(UTC).replace(tzinfo=None).isoformat() fire_and_forget( self.client.post( url=urljoin(self.url, f"/api/tasks/{message.task_id}/queued"), headers={"access-token": self.api_token}, json={ "args": message.args, "kwargs": message.kwargs, "taskName": message.task_name, "worker": self.__ta_broker_name, "queuedAt": now, }, ) ) return super().post_send(message) async def pre_execute(self, message: TaskiqMessage): now = datetime.now(UTC).replace(tzinfo=None).isoformat() fire_and_forget( self.client.post( url=urljoin(self.url, f"/api/tasks/{message.task_id}/started"), headers={"access-token": self.api_token}, json={ "startedAt": now, "args": message.args, "kwargs": message.kwargs, "taskName": message.task_name, "worker": self.__ta_broker_name, }, ) ) return super().pre_execute(message) async def post_execute( self, message: TaskiqMessage, result: TaskiqResult[Any], ): now = datetime.now(UTC).replace(tzinfo=None).isoformat() fire_and_forget( self.client.post( url=urljoin(self.url, f"/api/tasks/{message.task_id}/executed"), headers={"access-token": self.api_token}, json={ "finishedAt": now, "error": result.error if result.error is None else repr(result.error), "executionTime": result.execution_time, "returnValue": {"return_value": result.return_value}, }, ) ) return super().post_execute(message, result)
Reacted by Alexander SmirnovHey guys! Sorry for the late answer
Yes, the PR is on it's way, but i need to fix some linting errors, had no time to do this
You are welcome to speed it up 😄 : taskiq-python/taskiq#478
Updates: the middleware will be moved to taskiq package and aiohttp will be used instead of httpx. And yes, a proper fire and forgetIf i have some time today/tomorrow i will fix all errors in the PR and merge it
Reacted by Alexander Smirnov, Athroniaeth and nimaxinfor now, fixed for the HTTP approach
we will start implementing event-based middleware soontaskiq-python/taskiq#478 is merged, you can use it:
from taskiq.middlewares.taskiq_admin_middleware import TaskiqAdminMiddleware
Reacted by Athroniaeth, nimaxin and Alexander Smirnov
The
TaskiqAdminMiddlewaresignificantly slows down task triggering (.kiq()) because it blocks execution waiting for HTTP requests to complete. Each task operation creates a new HTTP clienthttpx.AsyncClient()instead of reusing one, adding unnecessary overhead.The middleware should:
httpx.AsyncClient()This becomes particularly noticeable when triggering many tasks rapidly (e.g., in loops or batch operations). The current implementation forces sequential HTTP requests, making bulk task triggering much slower than expected.
I believe the problem is clear, but I'm happy to provide code examples or additional details if needed.