Inconsistency with start order

[sakshamgoyal-01]

Taskiq version

0.12.4

Python version

Python 3.14

OS

Debian 12

What happened?

The workers seem to not pick up tasks depending on the order the broker/workers were started.

if the worker was started before the broker, then the worker will refuse to pick up any tasks till it is restarted (and the broker is already running).

Broker initialization code

from taskiq import TaskiqState
from taskiq.events import TaskiqEvents
from taskiq_dashboard import DashboardMiddleware, TaskiqDashboard
from taskiq_postgresql import PostgresqlBroker, PostgresqlResultBackend


result_backend: PostgresqlResultBackend[ParseResult] = PostgresqlResultBackend(
    dsn=POSTGRES_DSN,
    run_migrations=True,
)

broker = PostgresqlBroker(
    dsn=POSTGRES_DSN,
    run_migrations=True,
).with_result_backend(result_backend).with_middlewares(
    DashboardMiddleware(url=DASHBOARD_URL, api_token=DASHBOARD_API_TOKEN),
)

[sakshamgoyal-01]

@s3rius any update on this? It's causing a lot of headaches

[GefMar]

I reproduced the same observable symptom after restarting PostgreSQL with an established asyncpg LISTEN connection:

1. Start PostgreSQL and the worker normally.
2. Verify that a task is consumed.
3. Stop and restart PostgreSQL.
4. The worker remains alive but no longer consumes tasks.
5. Restarting only the worker restores consumption.

The likely root cause is in AsyncpgListenDriver.__aiter__(). It checks connection.is_closed() before waiting on the internal queue, but connection termination does not wake the pending queue.get(). Consequently, the iterator neither raises nor completes, and Taskiq core receives no terminal state that would allow it to restart the worker.

[PostgresqlBroker.listen() also needs an explicit terminal-listener policy. Its broad exception handling could otherwise swallow a connection failure or create a retry loop without rebuilding the listener.

This therefore appears to be primarily a taskiq-postgresql lifecycle issue rather than a missing Taskiq core retry. A generic timeout in core would be unsafe because an idle healthy listener and a disconnected listener are indistinguishable without a transport-specific signal.

The adapter should:

• wake the pending iterator when the asyncpg connection terminates;
• either reconnect with explicit backoff or propagate a typed terminal error;
• avoid swallowing terminal listener failures;
• make cleanup safe after partial startup;
• add a real PostgreSQL stop/start integration test.

Could you share the exact taskiq-postgresql version and, if possible, the PostgreSQL/container startup logs? In particular, it would help to know whether PostgreSQL briefly accepts connections and then replaces them during initialization. That would confirm whether this listener-disconnect transition is the exact cause of the reported startup-order behavior.

[GefMar]

PR: z22092/taskiq-postgresql#25

[anthonyatp]

Same root cause reaches further than a Postgres restart — an idle-evicted (blackholed) LISTEN socket produces the identical symptom with no connection_lost to recover from, so is_closed() stays False and the worker waits forever. Details, evidence, and a fix pattern (bounded wait + probe/sweep on the same connection) in z22092/taskiq-postgresql#25 (comment).