[BUG]: SEVERE? Seeing tasks remaining in queue when certain settings are applied

[adiberk]

I am seeing an interesting issue. Tasks are getting lost when the following settings are applied
taskiq worker app.worker:broker app.tasks --max-async-tasks=5 --max-tasks-per-child=5 --wait-tasks-timeout=600 --reload

I did this test on purpose to see how worker restarts work etc. And I am a bit worried about the results.
Some items are getting completely lost and remain in the queue

Note: I have 3 replicas running via dockerocompose replica command and each one uses --reload (1 worker per replica)

lev-cortex-worker-3  | INFO [11/30/25 15:21:36] Streamed 120 events to Redis for run-stream: 
lev-cortex-worker-3  | 434012e5-0b54-43c8-b568-22452d27980a
lev-cortex-worker-3  | INFO [11/30/25 15:21:36] No more tasks to wait for. Shutting down.
lev-cortex-worker-3  | INFO [11/30/25 15:21:36] The runner is stopped.
lev-cortex-worker-3  | WARNING [11/30/25 15:21:36] Shutting down the broker.
lev-cortex-worker-3  | [2025-11-30 15:21:37,292][taskiq.process-manager][INFO   ][MainProcess] worker-0 is dead. Scheduling reload.
lev-cortex-worker-3  | [2025-11-30 15:21:38,354][taskiq.process-manager][INFO   ][MainProcess] Process worker-0 restarted with pid 33

At some point now I see a bunch of tasks left in the queue. If I don't use max-tasks-per-child I think it all works.. except of course, I won't get the worker restarts I am testing.

Also the shutdown logs seem strange - I don't think those are the logs for the max-tasks-per-child shutdown right?

[s3rius]

That's a curious observation. I'm going to look into it tomorrow.

[adiberk]

Happy to provide more info on anything.

I am using redis streaming broker.

Really like this project.

[adiberk]

Also I sent 20 simultaneous tasks at once to ensure this issue would show up.
And when I say lost I mean that in the admin panel they remain in "queue"

[danfimov]

I think it would be helpful if you could provide some details about your setup:

• Which admin panel do you use — taskiq-admin or taskiq-dashboard? Version of this admin panel will be also helpful.
• Taskiq and taskiq-redis versions;
• If possible, the code you use for the broker setup.

This information will really help to reproduce the issue.

[adiberk]

Ok
For admin dashboard - i wrote my own but I can provide the middleware - it is pretty simple and I copied a bunch of code from the existing library. I also am sure it isn't being created because I run agents in these workers and keep audits of the agents being run, so I know some aren't getting run

taskiq version: Latest release
taskiq-redis: 1.1.0

Broker setup (local) - in production I use sentinel with a bunch of customizations. However I am experiencing the issue locally right now.

return RedisStreamBroker(
            url=settings.redis_or_sentinel.redis.url,  # type: ignore
            unacknowledged_batch_size=3,
            xread_count=2,
            maxlen=2_000,
        ).with_result_backend(
            RedisAsyncResultBackend(
                serializer=JSONSerializer(),
                redis_url=settings.redis_or_sentinel.redis.url,  # type: ignore
                result_ex_time=60 * 5,  # 5 minutes
            )
        )

Here is my code for the taskiq admin middleware which stores state. I essentially copied how the lib works but stored directly in db

class TaskiqAdminMiddleware(TaskiqMiddleware):
    """A Taskiq middleware that reports task lifecycle events to an external admin API.

    This middlewar stores task lifecycle events in the database.
    """

    def __init__(
        self,
        taskiq_broker_name: str | None = None,
    ) -> None:
        super().__init__()
        self.taskiq_broker_name = taskiq_broker_name

    @property
    def task_repository(self) -> TaskRepository:
        if not hasattr(self, "_task_repository"):
            self._task_repository = ApplicationContainer.repos.task_repository()
        return self._task_repository

    @property
    def lev_cache(self) -> LevCache:
        if not hasattr(self, "_lev_cache"):
            self._lev_cache = ApplicationContainer.services.redis()
        return self._lev_cache

    def get_task_name(self, task_name: str) -> str:
        return re.sub(r"^.*?:", "", task_name)

    async def post_send(self, message: TaskiqMessage) -> None:
        """
        This hook is executed right after the task is sent.
        """
        lock_key = f"taskiq_admin_middleware:{message.task_id}"

        try:
            async with self.lev_cache.lock(lock_key, timeout=2, blocking=False) as lock:
                # Just in case we don't raise an exception, we check if we own the lock
                if not await lock.owned():
                    return
                task = await self.task_repository.get_by_task_id(message.task_id)
                if task is None:
                    task_in = TaskModelIn(
                        task_id=message.task_id,
                        name=self.get_task_name(message.task_name),
                        args=message.args,
                        kwargs=message.kwargs,
                        queued_at=datetime.now(UTC),
                        worker=self.taskiq_broker_name,
                        state=TaskState.QUEUED,
                    )
                    await self.task_repository.upsert_task(task_in)
                else:
                    task.state = TaskState.QUEUED
                    await self.task_repository.update(
                        task.id,
                        {
                            "state": TaskState.QUEUED,
                            "queued_at": datetime.now(UTC),
                        },
                    )
        except LockError:
            # Couldn't acquire the lock (another process already holds it) – that's fine.
            # pre_execute will take care of writing/overwriting the state.
            return
        except Exception:
            logger.exception(f"Error in post_send for task {message.task_id}")

    async def pre_execute(self, message: TaskiqMessage) -> TaskiqMessage:
        """
        This hook is called before executing task.

        This is a worker-side hook, which means it
        executes in the worker process.
        """
        lock_key = f"taskiq_admin_middleware:{message.task_id}"

        try:
            # Higher priority: this *will* wait on the lock.
            async with self.lev_cache.lock(lock_key, timeout=2, blocking=True):
                task = await self.task_repository.get_by_task_id(message.task_id)
                now = datetime.now(UTC)

                if task is None:
                    task_in = TaskModelIn(
                        task_id=message.task_id,
                        name=self.get_task_name(message.task_name),
                        args=message.args,
                        kwargs=message.kwargs,
                        queued_at=now,  # if post_send didn't run, we at least set it here
                        worker=self.taskiq_broker_name,
                        state=TaskState.RUNNING,
                        started_at=now,
                    )
                    await self.task_repository.upsert_task(task_in)
                else:
                    task.state = TaskState.RUNNING
                    await self.task_repository.update(
                        task.id,
                        {
                            "state": TaskState.RUNNING,
                            "started_at": now,
                        },
                    )
        except Exception:
            logger.exception(f"Error in pre_execute for task {message.task_id}")

        return message

    async def post_execute(
        self,
        message: TaskiqMessage,
        result: TaskiqResult[Any],
    ) -> None:
        """
        This hook executes after task is complete.

        This is a worker-side hook. It's called
        in worker process.

        """
        task = await self.task_repository.get_by_task_id(message.task_id)
        if task is None:
            logger.warning(f"Task {message.task_id} not found in database")
            return
        else:
            task.state = TaskState.SUCCESS if result.error is None else TaskState.FAILURE
            await self.task_repository.update(
                task.id,
                {
                    "state": TaskState.SUCCESS if result.error is None else TaskState.FAILURE,
                    "finished_at": datetime.now(UTC),
                    "execution_time": result.execution_time,
                    "return_value": {"return_value": result.return_value},
                    "error": str(result.error) if result.error is not None else None,
                },
            )

[adiberk]

Were you able to reproduce? I haven't been able to run more tests (ie. is this an issue of just have having 1 worker per container etc.)

[s3rius]

So, first of all. Please don't use --reload flag when deploying. Because it's made only for development purposes to reload workers when you update your local files.

Second thing is that I'm observing some weird behavior when using --reload flag, but it works totally fine, if I limit what to watch by passing --reload-dir {project_dir}.

[adiberk]

So, first of all. Please don't use --reload flag when deploying. Because it's made only for development purposes to reload workers when you update your local files.

Second thing is that I'm observing some weird behavior when using --reload flag, but it works totally fine, if I limit what to watch by passing --reload-dir {project_dir}.

@s3rius
Regarding first part - Of course!! I am seeing this on my local development (I run docker locally with replicas so I can use reload but enable multiple workers). Production has more workers, resources and uses Kubernetes. I am not sure I see this in prod as I haven't released it yet

But regarding what you said about weird behavior. Are you seeing what I see given a similar setup?

[adiberk]

I'll also add that in production I often see logs like this

lev-cortex-worker-3  | INFO [11/30/25 15:21:36] No more tasks to wait for. Shutting down.
lev-cortex-worker-3  | INFO [11/30/25 15:21:36] The runner is stopped.
lev-cortex-worker-3  | WARNING [11/30/25 15:21:36] Shutting down the broker.

My understanding is that this isn't the log we are supposed to see when max-tasks is hit right? Or maybe it is and I am just getting confused?

[adiberk]

Ok -I think this log is not a problem, I may have missunderstood how and when it gets sent!

However - I could use some advice / help on a final issue slightly related to this @s3rius @danfimov

taskiq-python/taskiq-redis#118

Basically - I am not sure if my configuration are wrong, my expectations are wrong, or there is a slight bug.

1. We are laeving behind ghost consumers - i think this fix should be pretty easy with on shutdown event. (unless it is an OOM killer)
2. I am not sure what to do about consumers that have pending tasks when they are shut down - should we just release them back to the wild?

[adiberk]

^^ I edited this message a couple times, sorry in advance, but I think it is now much more succinct