Skip to main content

Running in production

This guide takes a queue to production on Postgres: the schema and its index, the checks that catch a wrong schema, the pool, the database versions and poolers, and shutdown.

The service is a mailer with eight workers, a cap of five attempts and a dead-letter group.

The schema​

src/main.rs
/// The service's first migration; in a service crate, `migrations/0001_email_jobs.sql`, run by
/// `sqlx::migrate!`.
const MIGRATION: &str = "
CREATE TABLE IF NOT EXISTS email_jobs (
    job_id  BIGSERIAL PRIMARY KEY,
    name    TEXT NOT NULL,
    attempt SMALLINT NOT NULL DEFAULT 1,
    payload BYTEA NOT NULL
);
CREATE INDEX IF NOT EXISTS email_jobs_claim ON email_jobs (name, job_id);
";

The crate creates no table: the schema is the service's, and its migrations create it, as sqlx::migrate! runs them in examples/outbox. The index follows the claim order, (group, priority, retry_after, id) for the columns the table has, (name, job_id) here. A claim locks every row of its group it reads before it sorts them, so without that index a claim held for a handler keeps the rest of its group from other claims. In a table with FIFO groups the same index keeps the claim's read of the group to the group's own rows.

Checked statements​

src/main.rs
use ruststream_sqlx::prelude::*;
use sqlx::Postgres;

/// One row of `email_jobs`: every statement it determines goes through sqlx's compile-time check.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs", checked, db = postgres)]
struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(payload)]
    payload: Vec<u8>,
}

checked, db = postgres hands every statement the struct determines to sqlx's compile-time check: the claim, the settlements, the dead letter of a group and the insert. A column the struct names and the table lacks fails the build. The build checks against DATABASE_URL, or against the .sqlx data that cargo sqlx prepare writes, committed beside the crate; cargo sqlx prepare --check in CI fails a build whose data is stale. The subscription runs the same statements it runs without checked, so a message costs the same. The check is the derive's: a table described by hand keeps the startup check alone.

Every subscription also prepares its statements at startup. A table or column the database lacks stops the service with SqlxBrokerError::Schema, which names the subscription, the table and the statement. A mount that declares a cap without its dead letter, or the reverse, stops it too. So on_startup runs the migrations before the subscriptions open, and the checks see the schema the release reads.

The pool​

src/main.rs
use std::num::NonZeroUsize;
use std::time::Duration;

use sqlx::Connection;
use sqlx::postgres::{PgConnectOptions, PgPoolOptions};

/// The emails in work at once; each holds its row's lock and a connection until it settles.
const WORKERS: NonZeroUsize = nonzero!(8);

/// A connection for each worker, and one the subscription leaves to settlements and publishes.
fn pool() -> PgPoolOptions {
    let size = u32::try_from(WORKERS.get() + 1).unwrap_or(u32::MAX);
    PgPoolOptions::new()
        .max_connections(size)
        .acquire_timeout(Duration::from_secs(5))
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
    tracing::info!(to = %email.to, ?attempt, "sending");
    HandlerOutcome::ack()
}

/// The service on the database `options` reach.
fn service(options: PgConnectOptions) -> RustStream {
    let migrate = options.clone();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_connect_options(options, pool()).route::<SendEmail>("emails");
    RustStream::new(AppInfo::new("mailer", "0.1.0"))
        // The schema the release reads exists before any subscription prepares its statements.
        .on_startup(async move |()| {
            let mut conn = PgConnection::connect_with(&migrate).await?;
            sqlx::raw_sql(MIGRATION).execute(&mut conn).await?;
            conn.close().await
        })
        // The label names the server in the generated AsyncAPI document.
        .with_broker_labeled("postgres", broker, |b| {
            b.include(send.workers(WORKERS))
                .max_attempts(nonzero!(5u32))
                .dead_letter("emails.dead");
        })
}

The table names no form, so it claims by row lock: each email in work holds its row's lock and a connection until it settles. A subscription never holds more than the pool's size less one, and leaves that one to the handlers' own queries, their publishes and the settlements. So the pool is sized for every subscription of the service: n for each workers(n), one more, and the connections the handlers take for their own work (Connections). A lease table holds a connection only for each statement, and suits a pool that cannot spare one per job. On Postgres, notifications keep one more connection for the broker's life. pool returns the pool's options, and the broker builds the pool from them when the service starts.

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    // The connection's settings come from the `PG*` environment variables.
    service(PgConnectOptions::new())
}

Databases and poolers​

DatabaseNeeds
Postgres9.5 for SKIP LOCKED; 11 for the advisory lock form and FIFO groups, which lock 64-bit hashes
MySQL, MariaDBMySQL 8.0.1 or MariaDB 10.6, read at startup; an older server stops it with ServerTooOld
SQLitethe lease or the advisory lock form; one process per database file in the advisory form

On MySQL and MariaDB a claim runs at READ COMMITTED unless a row lock table declares another level. A server that logs statements, binlog_format = STATEMENT, refuses writes at that level; the row and mixed formats accept them. On Postgres read_committed is the practical isolation level of a row lock table: a stricter level fails claims with serialization errors when claims and settlements run at once. The advisory lock form keeps its lock in the database session, so it needs a direct connection or PgBouncer in session pooling; transaction pooling hands the session to another client between statements. A broker with listen_notify listens on a session too.

Shutdown​

At shutdown the runtime stops the subscriptions, lets the handlers in work finish within its shutdown timeout, then shuts the broker down. A row lock delivery holds its transaction until it settles; one still unsettled when the process exits rolls back with its connection, and its row returns. A lease delivery keeps its lease and settles if it can, or its row returns when the lease runs out. The broker releases the advisory locks of the deliveries in work. The broker closes the pool it built last; a connection a delivery in work still holds closes when it comes back.

The AsyncAPI document​

With the asyncapi feature, the generated AsyncAPI document describes the server and each subscription's table and group. The label of with_broker_labeled names the server, and the server entry holds the database's host and port, never its user or password: the document is published and shared.

Testing​

One test in the example file builds the document of the service on a URL with a password, with no I/O, and finds the host and port in it and the password nowhere. Two more run against a Postgres database of their own: a table of an older release, without a column the struct names, stops the startup with an error that names it, and the checked statements serve the queue. They read the server's address from POSTGRES_TEST_URL and skip without it (Testing a service on SQL).