Postgres

On Postgres every form runs: SKIP LOCKED claims, a one-statement lease claim, advisory locks on 64-bit hashes, statement_timestamp and LISTEN/NOTIFY.

The postgres feature adds the dialect and sqlx's Postgres driver. The row lock and lease claims pass over rows other claims hold with FOR UPDATE SKIP LOCKED, which Postgres 9.5 added. A lease claim is one statement: it locks the claimable rows, writes their lease and returns them. A claim of the service's own may leave the rows to the crate's fetch, which reads them by a list of ids. A table on the database's clock reads statement_timestamp().

The advisory lock form and the claims of FIFO groups lock 64-bit hashes made by hashtextextended, which Postgres 11 added. An older server refuses their statements, so their subscriptions stop when they open. An advisory lock belongs to the session that took it, so the advisory lock form needs a direct connection or PgBouncer in session pooling.

A table that declares no level opens its transactions with a plain BEGIN, at the server's default. A row lock table works at read_committed; the stricter levels fail claims with serialization errors when claims and settlements of its rows run at once. Postgres runs READ UNCOMMITTED as READ COMMITTED, so its dialect does not open it, and a table that declares it does not compile. With listen_notify a subscription wakes on rows other processes write and announce. The AsyncAPI document names the server by the pool's host and port, never its credentials.

Example

Invoices in the row lock form: each claim holds its row in a transaction while the handler runs, and four claims run at once.

Macros
Manual

examples/sqlx_databases_postgres.rs
/// One row of `invoice_jobs`. No lease and no advisory lock: the table takes its rows by row lock.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "invoice_jobs")]
struct SendInvoice {
    #[field(id, generated)]
    id: i64,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber(InboxQueue::<SendInvoice>::new("invoices"))]
async fn send(invoice: &Invoice) -> HandlerOutcome {
    tracing::info!(number = %invoice.number, "sending the invoice");
    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());
    RustStream::new(AppInfo::new("billing", "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?;
            conn.close().await
        })
        .with_broker(broker.route::<SendInvoice>("invoices"), |b| {
            // Four claims in flight: each holds a connection of the pool while its handler runs.
            b.include(send.workers(nonzero!(4)));
        })
}

examples/manual/sqlx_databases_postgres.rs
/// One row of `invoice_jobs`. No `.lease(..)` and no `.advisory(..)`: the table takes its rows by
/// row lock.
#[derive(sqlx::FromRow)]
struct SendInvoice {
    id: i64,
    attempt: i16,
    payload: Vec<u8>,
}

impl InboxTable for SendInvoice {
    type Id = i64;
    type Table = InboxSpec<(Attempt, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("invoice_jobs", Column::new("id").generated())
        .attempt(Column::new("attempt").generated())
        .payload(Column::new("payload"));

    fn id(&self) -> &i64 {
        &self.id
    }
}

impl AttemptRow for SendInvoice {
    type Attempt = i16;

    fn attempt(&self) -> &i16 {
        &self.attempt
    }
}

impl PayloadRow for SendInvoice {
    type Column = Vec<u8>;

    fn payload(&self) -> &[u8] {
        &self.payload
    }
}

struct SendInvoices;

impl Handle<Invoice> for SendInvoices {
    fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(number = %invoice.number, "sending the invoice");
        ready(Ok(()))
    }
}

/// 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());
    RustStream::new(AppInfo::new("billing", "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?;
            conn.close().await
        })
        .with_broker(broker.route::<SendInvoice>("invoices"), |b| {
            // Four claims in flight: each holds a connection of the pool while its handler runs.
            let send = subscriber("invoices", SendInvoices)
                .map_source(|_| InboxQueue::<SendInvoice>::new("invoices"));
            b.include(send.workers(nonzero!(4)).build());
        })
}

Example​

Example