Skip to content

SQL databases

ruststream-sqlx brings SQL databases into a RustStream service through sqlx. It has two components:

  • the inbox: task queues in the service's own Postgres, MySQL/MariaDB and SQLite tables, served as a RustStream broker;
  • the transactional outbox: what a service publishes on any RustStream broker stays recorded in a table of its own until a consumer has processed it.
ruststream = { version = "0.7", features = ["macros"] }
ruststream-sqlx = { version = "0.7", features = ["inbox", "outbox", "postgres"] }
sqlx = { version = "0.9", features = ["runtime-tokio", "postgres", "derive", "chrono"] }
chrono = "0.4"
serde = { version = "1", features = ["derive"] }

The inbox and outbox features turn the components on, each independently of the other. A driver feature picks the database: postgres, mysql (MySQL and MariaDB), sqlite, or any for an AnyPool.

Task queues in the service's tables

crates/ruststream-sqlx/examples/inbox.rs
use std::error::Error;

use ruststream::OutgoingMessage;
use ruststream_sqlx::prelude::*;
use serde::{Deserialize, Serialize};
use sqlx::postgres::{PgConnectOptions, PgConnection, PgPool, Postgres};

/// One row of `email_jobs`: the struct names the table and the role of each column.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
}

/// A publish to `emails` writes a row through the service's own statement.
impl Publish<Postgres> for SendEmail {
    async fn publish(
        conn: &mut PgConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query("INSERT INTO email_jobs (name, payload) VALUES ($1, $2)")
            .bind(message.name())
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

#[derive(Deserialize, Serialize, Outgoing)]
struct Email {
    to: String,
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    println!("sending to {}", email.to);
    HandlerOutcome::ack()
}

fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(
        SqlxBroker::new(pool).route::<SendEmail>("emails"),
        |b| {
            b.include(send);
        },
    )
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    // sqlx builds a pool only inside a Tokio runtime, so the service builds it here, before the
    // app.
    let pool = PgPool::connect_with(PgConnectOptions::new()).await?;
    app(pool.clone()).run().await?;
    pool.close().await;
    Ok(())
}

#[derive(Inbox)] describes the queue table: its name, and the role each column plays. SqlxBroker serves the queues of the service's pool. A subscription claims rows, and the handler's outcome settles them: an acknowledgement deletes the row, a retry returns it to the queue. A publish writes a row through the service's own statement.

A subscription claims its rows in one of three forms: by row lock, in a transaction open while the handler runs; by lease, with the claim committed at once; or by an advisory lock on the row's key. A handler mounted with .transactional() writes through its delivery's transaction, and the acknowledgement commits its writes together with the row. Each role a column plays turns on one behaviour: groups, the order rows are taken in, delayed retries, an attempt cap with a dead letter, a mark that keeps a processed row.

The manual API is the other path to the same table, beside the derive: the struct implements a trait, and a typed builder holds every setting. The compiler checks it as strictly as the derive, and both give the same statements at the same cost.

A team that checks its SQL at compile time adds checked to the derive, and cargo sqlx prepare then checks the derive's statements together with the service's own queries. A test runs the service's own app in TestApp against a real database, since the service's SQL is part of what it checks; SQLite needs no server for it.

The transactional outbox

crates/ruststream-sqlx/examples/outbox.rs
use std::convert::Infallible;

use chrono::{DateTime, Utc};
use ruststream::OutgoingMessage;
use ruststream::memory::prelude::*;
use ruststream_sqlx::{Outbox, outbox};
use serde::{Deserialize, Serialize};
use sqlx::postgres::{PgConnectOptions, PgConnection, PgPool, Postgres};

/// One record of the `outbox` table.
#[derive(Outbox, sqlx::FromRow)]
#[outbox(table = "outbox")]
struct OrderOutbox {
    #[field(id)]
    id: i64,
    #[field(name)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
}

/// The service writes the record; the outbox sends its id with the message.
impl outbox::Publish<Postgres> for OrderOutbox {
    async fn publish(conn: &mut PgConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
        sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES ($1, $2) RETURNING id")
            .bind(msg.name())
            .bind(msg.payload())
            .fetch_one(conn)
            .await
    }
}

#[derive(Deserialize)]
struct PlaceOrder {
    id: u64,
}

#[derive(Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders")]
struct OrderPlaced {
    id: u64,
}

/// The reply to `orders` is recorded before it is sent.
#[subscriber("checkout", reply)]
async fn place(cmd: &PlaceOrder) -> OrderPlaced {
    OrderPlaced { id: cmd.id }
}

/// After the ack, the record is marked processed.
#[subscriber("orders")]
async fn fulfil(order: &OrderPlaced) -> HandlerOutcome {
    println!("fulfilling order {}", order.id);
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let tracking = outbox! {
        "orders" => OrderOutbox,
    };
    let registry = tracking.clone();

    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // sqlx builds a pool only inside the Tokio runtime, which starts after this builder.
        .on_startup(async move |()| {
            let pool = PgPool::connect_with(PgConnectOptions::new()).await?;
            registry
                .set_pool(pool.clone())
                .expect("the outbox is deferred and gets one pool");
            Ok::<_, sqlx::Error>(pool)
        })
        .layer(tracking.layer())
        .publish_layer(tracking.publish_layer())
        .after_shutdown(async move |pool| {
            pool.close().await;
            Ok::<_, Infallible>(())
        })
        .with_broker(MemoryBroker::new(), |b| {
            b.include(place).out_reply(Publish);
            b.include(fulfil);
            // Once the subscriptions are open, the unprocessed records are sent again.
            b.after_startup(Publish, tracking.republish());
        })
}

#[derive(Outbox)] describes the service's outbox table, and outbox! registers it under the names it tracks. The publish middleware records each tracked message before it is sent, with the record's id in a header. The subscription middleware takes the record into work by that id and marks it processed when the handler acknowledges. At startup the republish sends every unprocessed record again.

A message is delivered at least once. Two instances of a service that start together may both send the same record again, so a consumer that must not process a message twice checks that itself.

Inside #[ruststream::app] the app is built before the Tokio runtime starts, and sqlx builds a pool only inside the runtime. The registry therefore starts without a pool, and on_startup builds one and hands it over with set_pool. A test build leaves the outbox off unless the environment sets RUSTSTREAM_SQLX_OUTBOX=on.

The outbox's middlewares wrap inbox handlers as they wrap any other. A record is written on a connection of its own, so it commits even when the delivery's transaction rolls back. A message tracked into an inbox table keeps its record id where the table has a headers column that the service's publish statement writes. The rest is in the transactional outbox.

Which one to use

The inbox fits work that belongs to the service's data: a task written in the same transaction as the data it serves, a queue in the database the service already runs. The outbox fits messages a service publishes on another broker and must not lose. A service may use both.

Where the rest is

The reference on docs.rs opens with the crate's own textbook, one section per topic:

What the crate costs per message and how many messages it moves per second, beside a raw sqlx loop doing the same work, is on the benchmarks page.

Handlers, routers, codecs and middleware come from the framework, whose own entry pages start at the RustStream site.