Skip to main content

listen_notify

On Postgres, listen_notify wakes subscriptions on pg_notify from any process: each publish announces its row, and a trigger can announce others.

The broker listens on one connection of the pool, taken at connect and kept until shutdown. Each subscription listens on its table's channel before its first claim. The channel is the table's name, qualified with its schema where the table has one (app.jobs). The payload is the group, and a notification with an empty payload wakes every subscription of the table.

Each publish of the broker announces its row: SELECT pg_notify(channel, group) runs on the connection that wrote the row, right after the write. A subscription in another process therefore claims the row once it is visible. A notification that fails is logged as a warning, and the publish still succeeds: the row is written, and the poll interval reaches it. A writer outside RustStream sends the same notification itself, in its transaction or from a trigger.

Postgres delivers a notification when its transaction commits (NOTIFY). A notification sent while the listening connection was lost is gone: the broker connects again and wakes every subscription once. The poll interval stays the floor under every wake-up.

Notifications have a price, and the switch is off unless set:

  • the broker keeps one connection of the pool for its whole life, so the pool needs one more;
  • each publish runs one more statement, pg_notify, on its connection;
  • a transaction that notifies takes one lock, global to the server, at commit, so commits of notifying transactions run one after another, which limits many concurrent writers.

Postgres keeps channel names up to 63 bytes. A subscription to a table whose qualified name is longer stops at startup with SqlxBrokerError::Declaration, which names the table. On MySQL, MariaDB and SQLite listen_notify does not compile, and the error names the fix: Notifies is Postgres's alone.

Example​

Another program inserts signups, and a trigger announces each row with its group. The subscription claims the row as soon as the insert commits, though its poll waits a minute.

examples/sqlx_wake_notify.rs
#[subscriber(InboxQueue::<SignupJob>::new("signups"))]
async fn welcome(signup: &Signup) -> HandlerOutcome {
    tracing::info!(to = %signup.email, "sending the welcome");
    HandlerOutcome::ack()
}

/// The service on the database `options` reach.
fn service(options: PgConnectOptions) -> RustStream {
    let schema = options.clone();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_connect_options(options, PgPoolOptions::new());
    // A notification wakes the subscription; the poll each minute reaches a row whose
    // notification was lost.
    let broker = broker
        .poll_interval(Duration::from_secs(60))
        .listen_notify()
        .route::<SignupJob>("signups");
    RustStream::new(AppInfo::new("signup", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = PgConnection::connect_with(&schema).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            sqlx::raw_sql(ANNOUNCE).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker, |b| {
            b.include(welcome);
        })
}

/// The trigger announces every row another program inserts: the channel is the table, the
/// payload its group.
const ANNOUNCE: &str = "
CREATE OR REPLACE FUNCTION announce_signup() RETURNS trigger AS $$
BEGIN
    PERFORM pg_notify(TG_TABLE_NAME, NEW.name);
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

CREATE OR REPLACE TRIGGER signup_announced AFTER INSERT ON signup_jobs
    FOR EACH ROW EXECUTE FUNCTION announce_signup();
";