`TaskiqAdminMiddleware` adds significant latency to task triggering

[nimaxin]

The TaskiqAdminMiddleware significantly slows down task triggering (.kiq()) because it blocks execution waiting for HTTP requests to complete. Each task operation creates a new HTTP client httpx.AsyncClient() instead of reusing one, adding unnecessary overhead.

The middleware should:

1. Avoid blocking task operations while reporting status
2. Reuse HTTP connections where possible httpx.AsyncClient()
3. Handle reporting asynchronously without impacting task throughput

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.

[nimaxin]

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.

[spikeninja]

Hey! Thank you for such comprehensive description of the issue
Yes, we are aware of this problem and will fix it asap

Let'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.

[smalyu]

Hi! Are there any updates on this issue?

[Athroniaeth]

Hi! 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_forget could 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)

[spikeninja]

Hey 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 forget

If i have some time today/tomorrow i will fix all errors in the PR and merge it