Rust and PostgreSQL Listen Notify for Lightweight Events

Sometimes a service only needs to know that something changed. PostgreSQL has a built-in way to say so, and sqlx exposes it. The limits are the whole story, so they come first. Everything compiles on Rust 1.90 with edition 2024, sqlx 0.8 (features postgres, runtime-tokio), tokio 1 and PostgreSQL 17.

What it guarantees

• A notification is delivered to every session currently listening on the channel, after the sender's transaction commits.
• If nobody is listening, it vanishes.
• If a listener is disconnected, it misses everything sent meanwhile.
• The payload is a text string below 8000 bytes.

That makes it a doorbell, not a mailbox. The pattern that works: write the work to a table, NOTIFY as a hint, and have the worker also scan the table on startup and on a timer.

Sending

NOTIFY order_created, '{"id": 42}';
SELECT pg_notify('order_created', '{"id": 42}');

pg_notify takes parameters, so use it from application code. A trigger can send one automatically:

CREATE FUNCTION notify_order() RETURNS trigger AS $$
BEGIN
    PERFORM pg_notify('order_created', NEW.id::text);
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER orders_notify AFTER INSERT ON orders
FOR EACH ROW EXECUTE FUNCTION notify_order();

From Rust, in the same transaction as the insert:

use sqlx::PgPool;

pub async fn create_order(pool: &PgPool, total: i64) -> sqlx::Result<i64> {
    let mut tx = pool.begin().await?;
    let id: i64 = sqlx::query_scalar("INSERT INTO orders (total) VALUES ($1) RETURNING id")
        .bind(total)
        .fetch_one(&mut *tx)
        .await?;
    sqlx::query("SELECT pg_notify('order_created', $1)")
        .bind(id.to_string())
        .execute(&mut *tx)
        .await?;
    tx.commit().await?;
    Ok(id)
}

If the commit fails, no notification goes out, which is exactly what you want.

Listening

use sqlx::postgres::PgListener;
use sqlx::PgPool;

pub async fn run_listener(pool: PgPool) -> sqlx::Result<()> {
    let mut listener = PgListener::connect_with(&pool).await?;
    listener.listen("order_created").await?;

    drain_pending(&pool).await?;

    loop {
        let note = listener.recv().await?;
        let id: i64 = note.payload().parse().unwrap_or_default();
        println!("order {id} created");
        handle_order(&pool, id).await?;
    }
}

async fn drain_pending(_pool: &PgPool) -> sqlx::Result<()> { Ok(()) }
async fn handle_order(_pool: &PgPool, _id: i64) -> sqlx::Result<()> { Ok(()) }

PgListener::connect_with opens a connection separate from the pool's checked-out ones. recv waits for the next notification; if the connection drops, recv reconnects and re-listens before returning, and notifications sent during the gap are gone. That is why drain_pending exists: after (re)subscribing, query the table for unprocessed rows, so a missed hint cannot strand work. Run the loop in its own task with tokio::spawn and handle its error by restarting it.

Fan out inside the process

One listener per process is enough. Forward notifications into a broadcast channel and let request handlers or SSE streams subscribe (server sent events, channels compared). Opening a listener per client uses a database connection each and falls over quickly.

Pitfalls

• A pooler in transaction mode (PgBouncer) breaks LISTEN, because the session changes between statements. Connect the listener directly to the database or use session mode.
• High volume channels are a poor fit: every notification takes a lock at commit time, and heavy use can serialize committers.
• Do not put secrets in payloads; any session with LISTEN permission can read them.
• Batch work: a burst of 1000 notifications should trigger one scan, not 1000 handlers.

When you need durable delivery, a queue table with FOR UPDATE SKIP LOCKED is the next step, discussed in background job queues. Pool concerns are in pool sizing.