Derive Macro Inbox

Source

#[derive(Inbox)]
{
    // Attributes available to this derive:
    #[inbox]
    #[field]
    #[sqlx]
}

Expand description

Describes a queue table with a struct: writes its InboxTable impl, the description a table described by hand writes, and the traits its roles need.

The struct is an ordinary sqlx struct: #[inbox(..)] names the table, sqlx’s own attributes name the columns, and #[field(..)] marks the columns that run the queue. The InboxSpec page shows the same table written by hand.

§Examples

use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;

// app.email_jobs: job_id BIGSERIAL PRIMARY KEY, name TEXT, attempt SMALLINT DEFAULT 1,
// payload BYTEA
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs", schema = "app")]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(payload)]
    payload: Vec<u8>,
}

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

// The subscription claims the rows of group `emails`, oldest first, and skips the rows
// another worker holds.
#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    HandlerOutcome::ack()
}

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

§The table

#[inbox(table = "..")] names the table and is required. schema = ".." places it in a schema. Each names one thing and holds no dot: the schema goes into schema, never into the table’s name. advisory_lock = "jobs-{job_id}" selects the advisory lock form, with a key built from the fields named between braces. A placeholder names a field as written in Rust, without r# ({type} for r#type), and the key reads that field’s column. In that form a group keeps its order through the key, as in advisory_lock = "jobs-{name}", so fifo = true does not apply there. The form selects its candidates with their keys itself, so custom(..) does not take claim beside it; custom(lock, unlock) hands the lock and the unlock of each key to the service (Lock, Unlock). Every built-in dialect builds the lease and advisory lock forms; Postgres and MySQL build the row lock form too.

A table in the advisory lock form holds each row by a lock on its key while the handler runs, with no transaction open:

use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;

// jobs: job_id BIGSERIAL PRIMARY KEY, attempt SMALLINT NOT NULL DEFAULT 1,
// payload BYTEA NOT NULL
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "jobs", advisory_lock = "jobs-{job_id}")]
pub struct Transcode {
    #[field(id, generated)]
    job_id: i64,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
pub struct Video {
    path: String,
}

// A video takes minutes to encode. The delivery's connection keeps the job's lock while the
// handler runs, with no transaction open. If the process dies, the database ends that session,
// and the job returns.
#[subscriber(InboxQueue::<Transcode>::new("videos"))]
async fn transcode(video: &Video) -> HandlerOutcome {
    if encode(video).await {
        return HandlerOutcome::ack();
    }
    HandlerOutcome::retry()
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("transcoder", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        b.include(transcode);
    })
}

§Isolation and mode

isolation = <level> opens the table’s transactions at an isolation level: read_uncommitted, read_committed, repeatable_read or serializable. mode = <mode> opens them in a SQLite mode instead: deferred, immediate or exclusive. A table names one of the two, or neither and opens at its database’s default (READ COMMITTED on MySQL and MariaDB). The row lock claim’s transaction opens at it, and so does the transaction a delivery lends its handler in transactional mode, in every form.

use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::{PgPool, Postgres};

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "ledger_jobs", advisory_lock = "ledger-{id}", isolation = repeatable_read)]
pub struct Posting {
    #[field(id)]
    id: i64,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
struct Entry {
    account: String,
    cents: i64,
}

// The handler's transaction opens with `BEGIN ISOLATION LEVEL REPEATABLE READ`: the balance it
// checks stays as its first statement read it.
#[subscriber(InboxQueue::<Posting>::new("postings"))]
async fn post(entry: &Entry, Ctx(mut tx): Ctx<keys::Tx<Postgres>>) -> HandlerOutcome {
    let posted = sqlx::query(
        "INSERT INTO ledger (account, cents) SELECT $1, $2 \
         WHERE (SELECT sum(cents) FROM ledger WHERE account = $1) + $2 >= 0",
    )
    .bind(&entry.account)
    .bind(entry.cents)
    .execute(&mut *tx)
    .await;
    if posted.is_ok() { HandlerOutcome::ack() } else { HandlerOutcome::retry() }
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        b.include(post.transactional());
    })
}

The broker’s dialect opens what its database keeps: Postgres read_committed, repeatable_read and serializable, MySQL and MariaDB all four levels, SQLite the three modes. A subscription to a table its dialect does not open does not compile, and the error names the levels the dialect opens. An AnyPool reaches a database named only when the broker connects, so there the subscription stops when it starts instead. On Postgres a row lock table at repeatable_read or serializable fails claims with serialization errors when claims and settlements of its rows run at once; read_committed is the practical level there. A lease table at either level on Postgres refuses transactional mode when it starts: its transaction reads every row as its first statement found it, and would not see the lease extended later.

§Roles

#[field(..)] gives a field one role:

id, required: the row’s identity. Its type is Clone: a lease subscription keeps a copy of each id in work to extend its lease.
group: the group a subscription reads; #[field(group, fifo = true)] keeps each group in order.
partition_key: the delivery’s partition key.
priority: the claim order, a smaller value first.
retry_after: the time before which a row is not claimed.
attempt: the attempt number.
locked_until: the lease’s expiry; it selects the lease form.
processed_at: acknowledgement marks the row instead of deleting it.
headers and payload: the delivery’s headers and the message bytes.

generated, alone or beside a role, marks a column the database fills in.

Each role turns on one behaviour of the queue, and a table without it keeps the default. The subscription below reads the billing group, takes urgent tickets first, keeps each customer’s tickets in order on one worker, delays a failed ticket by five minutes, and moves it to the escalated group at the fifth failure. An acknowledged ticket stays in the table with its processed_at set.

use std::time::Duration;

use chrono::{DateTime, Utc};
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;

// support_tickets: id BIGSERIAL PRIMARY KEY, queue TEXT NOT NULL, urgency SMALLINT NOT NULL,
// customer TEXT NOT NULL, retry_after TIMESTAMPTZ NOT NULL DEFAULT now(),
// attempt SMALLINT NOT NULL DEFAULT 1, processed_at TIMESTAMPTZ, payload BYTEA NOT NULL
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "support_tickets")]
pub struct Ticket {
    #[field(id, generated)]
    id: i64,
    #[field(group)]
    queue: String,
    #[field(priority)]
    urgency: i16,
    #[field(partition_key)]
    customer: String,
    #[field(retry_after, generated)]
    retry_after: DateTime<Utc>,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(processed_at, generated)]
    processed_at: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
pub struct Question {
    text: String,
}

#[subscriber(InboxQueue::<Ticket>::new("billing"))]
async fn triage(question: &Question) -> HandlerOutcome {
    if answer(question).await {
        return HandlerOutcome::ack();
    }
    HandlerOutcome::retry_after(Duration::from_secs(300))
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("support", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        b.include(triage.workers_by_key(nonzero!(4)))
            .max_attempts(nonzero!(5u32))
            .dead_letter("escalated");
    })
}

#[field(group, fifo = true)] keeps each group in order: one row of a group is in work at a time, however many brokers read it, and a claim takes the group’s head alone:

use chrono::{DateTime, Utc};
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;

// sync_jobs: id BIGSERIAL PRIMARY KEY, target TEXT NOT NULL,
// retry_after TIMESTAMPTZ NOT NULL DEFAULT now(), payload BYTEA NOT NULL
// Each group holds the changes bound for one system.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "sync_jobs")]
pub struct SyncChange {
    #[field(id, generated)]
    id: i64,
    #[field(group, fifo = true)]
    target: String,
    #[field(retry_after, generated)]
    retry_after: DateTime<Utc>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Deserialize)]
pub struct Change {
    record: String,
}

// The CRM receives its changes one at a time, in the order they were queued.
#[subscriber(InboxQueue::<SyncChange>::new("crm"))]
async fn sync(change: &Change) -> HandlerOutcome {
    if push(change).await {
        return HandlerOutcome::ack();
    }
    // At once, so the change keeps its place at the head of the group.
    HandlerOutcome::retry()
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("crm-sync", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        b.include(sync);
    })
}

§The headers layout

A #[field(headers)] field with #[sqlx(flatten)] holds a headers struct, which derives InboxHeaders and describes the queue table: its attributes, the fields that play a role, and the service’s own headers. The struct that flattens it is the message, which a handler takes itself, &Message or &[Message], as in row mode. It takes #[inbox(custom(..))] alone, and no role beside its headers field. Its other fields are its own data: the default fetch reads them from the queue table, by name, so a column the table lacks stops the subscription at startup; with custom(fetch) the service’s Fetch reads them from wherever they live. The delivery’s headers hold the headers struct’s fields without a role, built on the first read. The headers struct gets the generated insert, the message none.

use ruststream_sqlx::prelude::*;
use sqlx::PgPool;

// order_jobs: job_id BIGSERIAL PRIMARY KEY, name TEXT, attempt SMALLINT DEFAULT 1,
// tenant TEXT, note TEXT
#[derive(Debug, Clone, InboxHeaders, sqlx::FromRow)]
#[inbox(table = "order_jobs")]
pub struct OrderHeaders {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(attempt, generated)]
    attempt: i16,
    tenant: String,
}

#[derive(Debug, Clone, Inbox, sqlx::FromRow)]
pub struct OrderJob {
    #[field(headers)]
    #[sqlx(flatten)]
    headers: OrderHeaders,
    note: Option<String>,
}

// The header `tenant` comes from the headers struct, the note from the message's own column.
#[subscriber(InboxQueue::<OrderJob>::new("orders"))]
async fn ship(job: &OrderJob, ctx: &mut Context<'_>) -> HandlerOutcome {
    let tenant = ctx.headers().get_str("tenant").unwrap_or_default();
    tracing::info!(tenant, attempt = job.headers.attempt, note = ?job.note, "shipping");
    HandlerOutcome::ack()
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        b.include(ship);
    })
}

A struct with a payload field is in payload mode: its handler takes the payload, decoded by a codec. A struct without one is in row mode: its handler takes the struct itself, as the driver read it. A struct in row mode derives Clone, because the test harness keeps a copy of each value. It does not derive Deserialize: a type that deserializes rides the codec.

A generic struct keeps its parameters: the impl requires Send + Sync + 'static of the struct and Clone + Debug + Send + Sync + 'static of the id field’s type. In row mode a handler takes it where its parameters make it Clone.

§Events of the service’s own

custom(..) names the events the struct writes itself, and every other event keeps the statement its dialect builds. The struct implements each named event’s trait for its database (Claim, Fetch, Ack, Retry, RetryAfter, Discard, DeadLetter, Extend, and Lock and Unlock in the advisory lock form). A settlement of the service’s own runs in the transaction the crate commits:

use ruststream_sqlx::Ack;
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::{PgConnection, PgPool, Postgres};

// email_jobs: job_id BIGSERIAL PRIMARY KEY, name TEXT NOT NULL, payload BYTEA NOT NULL
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs", custom(ack))]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
}

// A sent email stays in its table, in the `sent` group, for an audit.
impl Ack<Postgres> for SendEmail {
    async fn ack(conn: &mut PgConnection, id: &i64) -> Result<(), sqlx::Error> {
        sqlx::query("UPDATE email_jobs SET name = 'sent' WHERE job_id = $1")
            .bind(id)
            .execute(conn)
            .await?;
        Ok(())
    }
}

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

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

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

§Column names

A column is named in one place, sqlx’s attributes. #[sqlx(rename = "..")] names a field’s column as written, #[sqlx(rename_all = "..")] recases every other field’s name, a raw identifier loses its r#, and a #[sqlx(skip)] field reads no column. Outside the headers layout, a #[sqlx(flatten)] field reads columns the derive cannot see, so the statements select *, and a dead-letter move copies the row by position: the dead-letter table has the same columns in the same order. The other options of #[sqlx(..)], such as json, try_from and default, belong to sqlx’s own derive and pass through untouched.

§Checked at compile time

#[inbox(checked, db = postgres)] (or mysql, sqlite) also puts the statements the struct determines into sqlx’s compile-time check, against DATABASE_URL or the .sqlx data of cargo sqlx prepare; the service’s sqlx needs its macros feature. In the headers layout checked goes on the headers struct. The checked mode’s page lists what it covers and what it refuses.

use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;

// email_jobs: job_id BIGSERIAL PRIMARY KEY, name TEXT NOT NULL,
// attempt SMALLINT NOT NULL DEFAULT 1, payload BYTEA NOT NULL
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs", checked, db = postgres)]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    #[field(group)]
    name: String,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(payload)]
    payload: Vec<u8>,
}

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

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

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

§Compile errors

A struct that cannot drive a queue does not compile, and the error points at the field or the name that causes it: no id, a role played twice, a column named twice, a role or generated on a field without a column, fifo outside the group role, locked_until, fifo = true or claim in custom(..) beside advisory_lock, extend in custom(..) without locked_until, lock or unlock in custom(..) without advisory_lock or without each other, locked_until on clock = DatabaseClock, a lock key naming no field, a dot in table or schema, an unknown isolation level or mode, isolation beside mode. In the headers layout: a flattened headers field whose type does not derive InboxHeaders, a role or a table attribute on the message, and a custom event the headers struct’s form does not take.

A struct in row mode without Clone does not compile. The error points at its name and suggests the derive:

use ruststream_sqlx::Inbox;

#[derive(Inbox)]
#[inbox(table = "email_jobs")]
struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    to: String,
}

A struct in row mode that derives Deserialize does not compile either. Such a type rides the codec, and rustc reports conflicting implementations of Input:

use ruststream_sqlx::Inbox;
use serde::Deserialize;

#[derive(Inbox, Clone, Deserialize)]
#[inbox(table = "email_jobs")]
struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    to: String,
}

A handler that does not fit its table’s mode meets the core’s errors at the mount site: a handler of rows on a table in payload mode, a batch handler of rows with a reply or Out slots:

A handler of rows, mounted on a table in payload mode, does not compile. The error names the subscription and the row type its deliveries do not carry:

use ruststream_sqlx::prelude::*;
use sqlx::PgPool;

#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "jobs")]
pub struct Job {
    #[field(id, generated)]
    id: i64,
    #[field(payload)]
    payload: Vec<u8>,
}

#[derive(Inbox, sqlx::FromRow, Clone)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    to: String,
}

// `jobs` holds payloads, not `SendEmail` rows.
#[subscriber(InboxQueue::<Job>::new("emails"))]
async fn send(email: &SendEmail) -> HandlerOutcome {
    tracing::info!(to = %email.to, "sending");
    HandlerOutcome::ack()
}

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

A batch handler of rows with a reply does not compile either. The error notes that such a batch mounts in the plain form only:

use ruststream_sqlx::prelude::*;
use serde::Serialize;
use sqlx::PgPool;

#[derive(Inbox, sqlx::FromRow, Clone)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
    #[field(id, generated)]
    job_id: i64,
    to: String,
}

#[derive(Serialize, Outgoing)]
#[outgoing(name = "sent")]
pub struct Sent {
    to: String,
}

#[subscriber(InboxQueue::<SendEmail>::new("emails"), reply)]
async fn send_all(emails: &[SendEmail]) -> Vec<Sent> {
    emails.iter().map(|email| Sent { to: email.to.clone() }).collect()
}

pub fn app(pool: PgPool) -> RustStream {
    RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
        b.include(send_all.batch(nonzero!(64)));
    })
}

Implemented in ruststream-sqlx-macros and used through ruststream-sqlx.

Inbox

Derive Macro Inbox Copy item path