Skip to content

Using LISTEN/NOTIFY with asyncpg

A service that polls SELECT * FROM jobs WHERE status = 'pending' every second pays for that second in latency and pays the database for the query whether or not there is work. PostgreSQL's LISTEN/NOTIFY removes both costs: the writer says NOTIFY jobs, and every connection currently listening on that channel gets a callback in under a millisecond. asyncpg exposes it as Connection.add_listener(channel, callback), which is a three-line change to a service — and a much longer change to the assumptions it can safely make, because notifications are not persistent, not counted, and not delivered to anyone who happens to be reconnecting. This guide establishes what the mechanism actually guarantees, measured against a live PostgreSQL 18 server, then builds a listener that survives a dropped connection without losing work.

Prerequisites

  • Python 3.11+ with asyncpg (pip install asyncpg) and a PostgreSQL server you can open two connections to.
  • Connection lifecycle from Async Database Drivers, because a listener needs a connection that is never returned to a pool.
  • Bounded queues from Async Queue Management, which is where every notification should land.
The path of one notification 5 stages from NOTIFY to bounded queue. The path of one notification NOTIFY inside a transaction COMMIT now it is sent listening backend if subscribed now asyncpg callback runs on the loop bounded queue workers consume Nothing is stored: a listener that is not connected at commit time never sees it.

1. Subscribe and receive

add_listener registers a callback that asyncpg invokes with four arguments: the connection, the notifying backend's process id, the channel and the payload string.

import asyncio

import asyncpg


async def demo(dsn: str) -> None:
    listener = await asyncpg.connect(dsn)
    publisher = await asyncpg.connect(dsn)
    seen: list[str] = []

    def on_notify(conn, pid, channel, payload):        # plain function: runs inline on the loop
        seen.append(payload)

    await listener.add_listener("jobs", on_notify)
    await publisher.execute("NOTIFY jobs, 'hello'")
    await asyncio.sleep(0.2)
    print(seen)                                        # ['hello']
    await listener.close()
    await publisher.close()

The callback may also be a coroutine function; asyncpg schedules it as a task. Two consequences follow immediately, and both matter in production. A plain function runs inline on the event loop, so any blocking work in it stalls every other task — see offloading blocking calls. A coroutine function is scheduled without any limit: 200 notifications in a burst create 200 concurrent tasks. Neither shape gives you back-pressure, which is why step 3 hands off to a queue instead.

Use pg_notify('channel', payload) rather than the NOTIFY statement whenever the payload is dynamic — it takes parameters, so there is no string interpolation into SQL:

await publisher.execute("SELECT pg_notify('jobs', $1)", json.dumps({"id": job_id}))

Verify: the payload appears in the listener within a few hundred milliseconds of the publisher's call.

2. Learn the four delivery rules before designing around them

Each of these was measured against a live server, and each one invalidates a design that looks reasonable on paper.

Notifications are delivered at commit, not at NOTIFY. Inside an open transaction the listener sees nothing; 300 ms after the NOTIFY the count of received notifications was still 0, and it became 1 immediately after the commit. A transaction that rolls back delivers nothing at all — which is the behaviour you want, since it makes the notification atomic with the row it is announcing.

Identical payloads inside one transaction are folded. Five NOTIFY jobs, 'same' statements in a single transaction produced exactly one delivery. Any logic of the form "increment a counter per notification" is therefore wrong; a notification means at least one thing changed.

Payloads are limited to just under 8,000 bytes. A 7,999-byte payload arrived intact; 8,000 bytes raised:

asyncpg.exceptions.InvalidParameterValueError: payload string too long

Send an identifier and let the consumer read the row. That also keeps the notification correct when the row changes again before the consumer gets to it.

Nothing is stored for a listener that is not connected. A NOTIFY issued while the listener was reconnecting was never delivered; the next one, sent after the new connection had subscribed, arrived normally. This is the rule that decides the architecture, and step 4 deals with it.

Verify: run a NOTIFY inside an uncommitted transaction and confirm the listener stays silent until the commit.

What LISTEN/NOTIFY does and does not promise A grid of 4 rows by 2 columns. What LISTEN/NOTIFY does and does not promise behaviour observed consequence sent at commit nothing during the transaction rollback delivers nothing no persistence notification while down is lost you need a catch-up query duplicates folded 5 identical NOTIFYs arrive as 1 never count notifications payload under 8,000 bytes 7,999 fine, 8,000 raises send an id, not a document Treat a notification as a hint that work exists, never as the work itself.

3. Hand every notification to a bounded queue

The callback's only job is to move the payload somewhere it can be processed with back-pressure. put_nowait on a bounded queue does that without awaiting, so it is safe in a plain function, and the QueueFull branch makes overload visible instead of invisible:

class Notifications:
    def __init__(self, channel: str, *, maxsize: int = 1000):
        self.channel = channel
        self.queue: asyncio.Queue[str] = asyncio.Queue(maxsize=maxsize)
        self.dropped = 0

    def _on_notify(self, conn, pid, channel, payload):
        try:
            self.queue.put_nowait(payload)             # never awaits, never blocks the loop
        except asyncio.QueueFull:
            self.dropped += 1                          # a metric, not a silent loss

Pushing 200 notifications through this path queued all 200 in 0.53 s with zero drops, and the connection stayed free to receive the next one throughout. Workers then consume the queue at whatever rate they can sustain, exactly as in building a worker pool with TaskGroup. Because notifications are folded and droppable anyway, a full queue is not a correctness problem as long as the catch-up query in the next step exists.

Verify: the drop counter stays at zero under normal load and rises under a burst, while the event loop's latency does not.

4. Reconnect, and always catch up on connect

A listening connection is long-lived, which means it will eventually be closed by a failover, a deploy, a network blip or an administrator. add_termination_listener gives you a callback when that happens, which turns "detect the failure" into "await an event":

async def run(self, dsn: str, on_reconnect) -> None:
    delay = 0.1
    while True:
        conn = None
        try:
            conn = await asyncpg.connect(dsn)
            await conn.add_listener(self.channel, self._on_notify)
            await on_reconnect()                       # catch-up query, every time
            delay = 0.1                                # reset only after a good connect
            closed = asyncio.Event()
            conn.add_termination_listener(lambda c: closed.set())
            await closed.wait()
        except (OSError, asyncpg.PostgresConnectionError):
            pass
        finally:
            if conn is not None and not conn.is_closed():
                await conn.close()
        await asyncio.sleep(delay + random.random() * delay)
        delay = min(delay * 2, 5.0)                    # jittered backoff, capped

The order inside the try is deliberate: subscribe before the catch-up query, so anything that arrives while the query runs is queued rather than missed. The catch-up query is whatever "find work I have not done" means for your schema — typically SELECT id FROM jobs WHERE status = 'pending' AND claimed_at IS NULL. It runs on the first connect too, which is what makes a restart safe.

Terminating the listening backend with pg_terminate_backend() mid-run exercises all of it: the run above recorded 1 reconnect, 2 catch-up runs and 0 drops, and both the notification sent before the kill and the one sent after the reconnect arrived. The backoff policy is the standard one from retry and backoff strategies — jitter included, because a failover disconnects every listener at once.

Verify: kill the listening backend and confirm the queue receives notifications published after the reconnect, with the catch-up counter incremented.

A reconnect that loses no work 5 ordered steps. A reconnect that loses no work connect a dedicated connection, never pooled add_listener(channel) subscription is per connection run the catch-up query claims anything missed while down await the termination event set by add_termination_listener sleep with jittered backoff then reconnect from the top The catch-up query runs on every connect, including the first one.

5. Decide whether you needed notifications at all

LISTEN/NOTIFY is a latency optimisation on top of a durable mechanism, not a replacement for one. The durable mechanism is a table you can query. If a missed notification would mean a job never runs, the system needs a periodic sweep regardless — and once the sweep exists, notifications simply make the common case fast.

A useful default for a job runner: notify on insert, sweep every 30 seconds, and claim rows with UPDATE ... WHERE id = $1 AND claimed_at IS NULL RETURNING id so a notification and a sweep racing each other cannot double-process. A durable job queue on Postgres builds that end to end.

Verify: disable notifications entirely and confirm the service still completes all work, only later.

Notifications or polling? 2 columns contrasting LISTEN/NOTIFY, Polling a table. Notifications or polling? LISTEN/NOTIFY push, sub-millisecond latency is one round trip idle costs nothing at-most-once delivery needs a dedicated connection silent loss on disconnect Polling a table pull, one interval of delay latency is half the interval constant query load exactly-once with claiming shares the pool recovers by itself Most durable systems use both: NOTIFY for latency, a poll loop as the safety net.

Verification

A listener is production-ready when:

  • The connection is dedicated: it is never acquired from or returned to a pool, and nothing else runs queries on it.
  • The callback never blocks: it does one put_nowait and returns; no I/O, no await on external services.
  • Reconnects are exercised: killing the backend with pg_terminate_backend() produces a reconnect and a catch-up run, not a stalled service.
  • Catch-up is unconditional: the query runs on every successful connect, including the first.
  • Loss is survivable: dropping every notification for a minute delays work but never loses it.

Pitfalls & edge cases

  • Listening on a pooled connection. pool.acquire() returns the connection to the pool at the end of the block, and asyncpg resets it — the subscription goes with it. Listeners get their own connection.
  • Channel names are identifiers. LISTEN takes an identifier, not a parameter, so a dynamic channel must be quoted with asyncpg.utils._quote_ident or validated against an allowlist. pg_notify takes the channel as a normal parameter and avoids the question.
  • Counting notifications. Folding makes counts meaningless. Derive state from the table, not from the number of callbacks.
  • A slow callback under a burst. Coroutine callbacks are scheduled as tasks with no limit; 10,000 notifications become 10,000 tasks. The queue handoff is the fix, not a semaphore inside the callback.
  • Notifications inside long transactions. The notification is released at commit, so a NOTIFY at the start of a five-minute transaction is delivered five minutes late.
  • Connection poolers in transaction mode. PgBouncer in transaction pooling mode does not support LISTEN at all; the listener needs a direct connection or a session-mode pool.

Frequently Asked Questions

How do I listen for PostgreSQL notifications in asyncio?

Open a dedicated asyncpg connection and call await conn.add_listener(channel, callback). The callback receives (connection, pid, channel, payload) whenever a NOTIFY on that channel commits. Keep the callback trivial — put the payload on a bounded asyncio.Queue and let worker tasks do the work.

Why is my asyncpg listener missing notifications?

Most likely it was not connected at the moment the notification committed — PostgreSQL stores nothing for absent listeners. The other common causes are listening on a pooled connection, which loses the subscription on release, and identical payloads inside one transaction, which PostgreSQL folds into a single delivery. Add a catch-up query that runs on every connect.

How large can a NOTIFY payload be?

Just under 8,000 bytes. A 7,999-byte payload is delivered; 8,000 raises InvalidParameterValueError: payload string too long. Send a row id and let the consumer read the current state, which is also more correct when the row changes again before processing.

Should I use LISTEN/NOTIFY or poll the table?

Use both. Notifications cut latency from half a poll interval to a round trip, but they are at-most-once and vanish during a disconnect. A periodic sweep of unclaimed rows makes the system durable; notifications make it fast. If missed work is acceptable, notifications alone are fine.

How do I detect that an asyncpg listening connection has died?

Register conn.add_termination_listener(callback) and set an asyncio.Event from it, then await that event in your supervisor loop. When it fires, close the connection, sleep with jittered backoff, reconnect, resubscribe and re-run the catch-up query.