Skip to main content

Inbox

ruststream_sqlx

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 macro or manual section 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.

§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.

§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.

§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. Statements checked at compile time shows what it covers and what it refuses.

§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. Row mode shows both. Implemented in ruststream-sqlx-macros and used through ruststream-sqlx.