Expand description
SQL databases for the RustStream messaging framework,
through sqlx.
The crate serves task queues kept in the service’s own tables, on Postgres, MySQL, MariaDB and
SQLite. A struct of the service’s own describes a queue table: Inbox reads the table from
#[inbox(..)], the column names from sqlx’s attributes and the role of each column from
#[field(..)]. The same description may be written by hand, through InboxTable and its typed
builder InboxSpec (macro or manual). SqlxBroker serves the queues of the service’s sqlx pool as a RustStream
broker: a subscription claims rows, a handler’s outcome settles them, and a publish writes a row
through the service’s own SQL.
The crate also serves a transactional outbox over any RustStream broker: what a service publishes is recorded in its own table until a consumer has processed it, and what was not processed is published again at startup. It is described in the transactional outbox.
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, a handler that writes its results in its delivery’s transaction. The outbox fits messages a service publishes on another broker and must not lose. A service may use both: the outbox’s middlewares wrap inbox handlers as they wrap any other.
The inbox feature turns the broker on, the outbox feature the outbox, and a service adds what
its database and its tables need:
postgres,mysql(MySQL and MariaDB) andsqlite: a built-in dialect and its sqlx driver;any: anAnyPool, served by the dialect of the database it reaches;chronoortime: the time columns;json: the headers column;testing: the broker’s in-process mode forTestApp, and the outbox’s test switch;asyncapi: what a subscription adds to the generated document.
Where things are:
Inbox: what a struct declares, the roles a column plays, the compile errors it meets.InboxTableandInboxSpec: a queue table described by hand, its settings inspec.SqlxBrokerandInboxQueue: the broker and its subscriptions, described below.- Roles and Time: what each column turns on, and where “now” comes from.
RepositoryandRouted: publishing into tables.keys: what a handler reads off a delivery.InboxSettings::transactionalandTx: transactional mode, where a handler writes through its delivery’s transaction.dialect: the SQL each database runs, and the traits a dialect of the service’s own implements.- The transactional outbox:
#[derive(Outbox)]or a record described by hand,outbox!ortrack, and the registry’s middlewares, under theoutboxfeature.
§The inbox broker
use std::time::Duration;
use chrono::{DateTime, Utc};
use ruststream::OutgoingMessage;
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::{PgConnection, PgPool, Postgres};
// email_jobs: job_id BIGSERIAL PRIMARY KEY, name TEXT, retry_after TIMESTAMPTZ DEFAULT now(),
// attempt SMALLINT DEFAULT 1, processed_at TIMESTAMPTZ, payload BYTEA
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
#[field(id, generated)]
job_id: i64,
#[field(group)]
name: 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>,
}
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)]
pub struct Email {
to: String,
}
#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
if deliver(email).await {
return HandlerOutcome::ack();
}
// The handler owns the backoff: a minute per attempt.
HandlerOutcome::retry_after(Duration::from_secs(60 * attempt.unwrap_or(1)))
}
pub 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);
},
)
}SqlxBroker serves the queues in the tables of the service’s pool, and the pool stays the
service’s. An InboxQueue subscription reads one queue: its name selects a group where the
table has a group column, and addresses the whole table where it has none. A subscription
claims rows when its stream is polled; after a claim that found fewer rows than it asked for, it
waits the poll interval, one second unless set. The bytes reach the codec lent from the row. A
table without a payload column hands its handler the row itself (see Row mode). The
same table described by hand opens macro or manual.
What a handler answers decides the row’s fate:
ack()deletes the row, or setsprocessed_atwhere the table has it;retry()returns the row to the queue at once, and its next delivery reads one more attempt;retry_after(d)hides the row untilretry_aftercomes; a table without that column returns it at once, and the runtime logs a warning;drop()finishes the row asack()does;max_attempts(n)anddead_letter(..)on the mount come together:attemptcounts the deliveries, and at the cap the row moves to another group or into a table with the same columns. Withmax_attempts(1)every failure moves the row at once. A mount that declares one without the other stops at startup.
Each #[field(..)] role turns on one behaviour of the queue (Roles). “Now” comes from
the host’s clock unless the table names another source (Time).
§An event of the service’s own
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);
})
}By hand, the table names the event in its chain and in its type:
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::spec::{Payload, own};
use ruststream_sqlx::{Ack, InboxSpec, InboxTable, PayloadRow};
use sqlx::{PgConnection, Postgres};
#[derive(sqlx::FromRow)]
pub struct SendEmail {
job_id: i64,
payload: Vec<u8>,
}
impl InboxTable for SendEmail {
type Id = i64;
type Table = InboxSpec<(Payload, own::Ack)>;
const TABLE: Self::Table = InboxSpec::new("email_jobs", Column::new("job_id").generated())
.group(Column::new("name"))
.payload(Column::new("payload"))
.own::<own::Ack>();
fn id(&self) -> &i64 {
&self.job_id
}
}
impl PayloadRow for SendEmail {
type Column = Vec<u8>;
fn payload(&self) -> &[u8] {
&self.payload
}
}
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(())
}
}An event the table names, in custom(..) on the derive or with .own::<..>() by hand, runs the
service’s own SQL in place of the statement the dialect builds, and every other event keeps the
dialect’s. The table and the broker stay as they were. Each event is a trait the struct implements
for its database: Claim, Fetch, Ack, Retry, RetryAfter, Discard,
DeadLetter, Extend, and Lock and Unlock in the advisory lock form; the markers of
spec::own name them by hand. A named event without its impl does not compile, and the error
names the trait. A settlement of the service’s own runs in the transaction the crate commits, so it
settles the row as the built-in one does. An acknowledgement takes the row out of what the claim
selects: a row it leaves claimable is delivered again.
§Macro or manual
use std::time::Duration;
use chrono::{DateTime, Utc};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::spec::{Attempt, Payload, ProcessedAt, RetryAfter};
use ruststream_sqlx::{AttemptRow, InboxSpec, InboxTable, PayloadRow};
use serde::Deserialize;
use sqlx::PgPool;
// email_jobs: job_id BIGSERIAL PRIMARY KEY, name TEXT, retry_after TIMESTAMPTZ DEFAULT now(),
// attempt SMALLINT DEFAULT 1, processed_at TIMESTAMPTZ, payload BYTEA
#[derive(sqlx::FromRow)]
pub struct SendEmail {
job_id: i64,
attempt: i16,
payload: Vec<u8>,
}
impl InboxTable for SendEmail {
type Id = i64;
type Table = InboxSpec<(
RetryAfter<DateTime<Utc>>,
Attempt,
ProcessedAt<DateTime<Utc>>,
Payload,
)>;
const TABLE: Self::Table = InboxSpec::new("email_jobs", Column::new("job_id").generated())
.group(Column::new("name"))
.retry_after(Column::new("retry_after").generated())
.attempt(Column::new("attempt").generated())
.processed_at(Column::new("processed_at").generated())
.payload(Column::new("payload"));
fn id(&self) -> &i64 {
&self.job_id
}
}
impl PayloadRow for SendEmail {
type Column = Vec<u8>;
fn payload(&self) -> &[u8] {
&self.payload
}
}
impl AttemptRow for SendEmail {
type Attempt = i16;
fn attempt(&self) -> &i16 {
&self.attempt
}
}
#[derive(Deserialize)]
pub struct Email {
to: String,
}
#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &Email, Ctx(attempt): Ctx<keys::Attempt>) -> HandlerOutcome {
if deliver(email).await {
return HandlerOutcome::ack();
}
HandlerOutcome::retry_after(Duration::from_secs(60 * attempt.unwrap_or(1)))
}
pub fn app(pool: PgPool) -> RustStream {
RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
b.include(send);
})
}#[derive(Inbox)] and #[derive(InboxHeaders)] describe a table through a public API, and a
service may write that description by hand: for a declaration a reviewer reads as it is, or for a
struct that holds fewer fields than the table has columns. The table above is the one the inbox
broker derives. Both forms build the same description, the same statements and
the same per-message code.
A table described by hand implements InboxTable. TABLE holds its description, built by
InboxSpec: new takes the table’s name and the id column, and each setter adds a setting.
Every setter is a const fn, so the description is a constant and costs nothing at run time.
type Table names the builder’s final type. Stable Rust does not infer the type of an associated
constant, so the impl writes it once, and the compiler holds the chain to it: a chain that does
not match the type is a type mismatch. The type lists the settings the crate reads as types, in
the order the chain sets them:
- the form:
Lease<Time>orAdvisory; - payload mode:
Payload; - the roles a delivery reads off a row:
Key,Attempt,Headers,HeaderFields; - the time roles with their time type:
RetryAfter<Time>,ProcessedAt<Time>; Fifo, the clockClock<Source>and the openingOpens<Level>;- the events the service writes itself, the markers of
spec::own.
The time types in the chain come from this type. The column-only setters (within, group,
priority, data, fetching, selecting_all) leave the type as it is. A setting the chain
leaves out keeps its default: the row lock form, row mode, SystemClock, the database’s own
opening and the crate’s events.
The struct holds only what the service reads. A column only the queue reads, such as name or
processed_at above, is named in the description and needs no field. A role a delivery reads off
the row is a small trait, implemented where the type names the role:
In type Table | Trait | What it lends |
|---|---|---|
Payload | PayloadRow | the message bytes, and the column’s type |
Key | KeyRow | the partition key |
Attempt | AttemptRow | the attempt count |
Headers | HeaderRow | the field that holds the headers column |
HeaderFields | HeaderFields | the header map, built with put_header |
A role in the type without its trait does not compile, and the error names the trait.
A table without .payload(..) is in row mode, and its struct says so with one more impl: impl Input for SendEmail { type Axis = SoloCarried<Self>; } from ruststream::runtime. Input is the
core’s trait, so the crate cannot write it for the service’s type, and the derive writes that line
too. The struct derives Clone, and .data(..) names the columns of its own data, which the claim
reads beside the roles (row mode shows a whole table).
An event the service writes itself is a marker in the chain, .own::<own::Ack>(), with
own::Ack in the type, where the derive takes custom(ack). The struct implements the event’s
trait, and a named event without its impl does not compile at the mount site.
dialect::insert renders the built-in insert from the description, in a constant:
use ruststream_sqlx::dialect::insert::{self, Sql};
use sqlx::PgConnection;
// INSERT INTO "email_jobs" ("name", "payload") VALUES ($1, $2)
const INSERT: Sql<128> = insert::postgres(&SendEmail::TABLE.spec());
/// Queues an email in the service's own transaction, so it commits with the order it belongs to.
pub async fn enqueue(tx: &mut PgConnection, email: &[u8]) -> Result<(), sqlx::Error> {
sqlx::query(INSERT.as_str())
.bind("emails")
.bind(email)
.execute(tx)
.await?;
Ok(())
}The insert names every column the database does not fill, and binds them in the description’s
order: the roles, then the data columns. The derive’s insert takes its text
from the same writer. A capacity N the text outgrows is a build error that names the size to
raise it to.
§Where each rule is checked
The derive checks a table while it expands, and points at the field at fault. By hand, a rule is held by a type where it can be, and otherwise by the evaluation of the constant while the service builds:
| Rule | Derive | By hand |
|---|---|---|
| an id and a table name | compile | the constructor’s arguments |
| one form, one message mode, one clock, one opening; each typed role and each own event once | compile | the setter’s bound, at compile time |
| a role in the type has its trait; an own event is implemented | compile | a trait bound, at compile time |
| the lease form on the host’s clock; no FIFO group and no own claim in the advisory lock form; an own extend only in the lease form; own lock and unlock together, only in the advisory lock form | compile | a bound on type Table, at compile time |
| a column-only role set once; a column named once | compile | the constant’s evaluation, at build time |
| an outbox name tracked once | compile (outbox!) | the evaluation of track, at build time |
an outbox name registered once through register("..") | n/a | app construction |
| a lock key names a column the table has | compile | the statement’s prepare, at startup |
A refusal at build time names the table and the rule, and points at the crate’s check rather than the service’s line. That is the one place where the manual form reads worse than the derive.
§Roles
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");
})
}Each role is a switch: a column that plays it turns on one behaviour of the queue, and a table
without it keeps the default. The subscription above 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.
| Role | By hand | With the role | Without it |
|---|---|---|---|
group | .group(..) | a subscription reads the group its name selects; a dead letter moves the row to another group | a subscription reads the whole table; a dead letter moves the row into another table |
group, fifo = true | .fifo_group(..) | one row of a group is in work at a time, in claim order (FIFO groups) | the rows of a group run side by side |
priority | .priority(..) | a smaller value is claimed first | rows are claimed by retry_after where the table has it, then by id |
partition_key | .partition_key(..) | the delivery’s partition key: under workers_by_key(n) the rows of one key run in order | the delivery has no key |
retry_after | .retry_after(..) | retry_after(d) hides the row for d | retry_after(d) returns the row at once, with a warning |
attempt | .attempt(..) | Ctx<keys::Attempt> reads the count, and max_attempts(n) caps it | Ctx<keys::Attempt> reads None, and a mount with max_attempts(n) stops at startup |
processed_at | .processed_at(..) | acknowledgement marks the row | acknowledgement deletes the row |
locked_until | .lease(..) | the lease form (the lease form) | the row lock form, or the advisory lock form under advisory_lock |
headers | .headers(..) | the delivery’s headers, kept in one column | the delivery’s headers are empty |
payload | .payload(..) | payload mode: the handler takes the decoded payload | row mode: the handler takes the row (Row mode) |
generated beside a role marks a column the database fills in, and the derive’s insert leaves it
out. Every role but id is optional, and each plays on one column at most.
By hand, each role is a setter of InboxSpec. A role a delivery reads off the row also names
its marker in type Table and implements its trait: Payload and PayloadRow, Key and
KeyRow, Attempt and AttemptRow, Headers and HeaderRow. The roles only the queue
reads need no field in the struct:
use chrono::{DateTime, Utc};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::{Attempt, Key, Payload, ProcessedAt, RetryAfter};
use ruststream_sqlx::{AttemptRow, InboxSpec, InboxTable, KeyRow, PayloadRow};
#[derive(sqlx::FromRow)]
pub struct Ticket {
id: i64,
customer: String,
attempt: i16,
payload: Vec<u8>,
}
impl InboxTable for Ticket {
type Id = i64;
type Table = InboxSpec<(
Key,
RetryAfter<DateTime<Utc>>,
Attempt,
ProcessedAt<DateTime<Utc>>,
Payload,
)>;
const TABLE: Self::Table = InboxSpec::new("support_tickets", Column::new("id").generated())
.group(Column::new("queue"))
.priority(Column::new("urgency"))
.partition_key(Column::new("customer"))
.retry_after(Column::new("retry_after").generated())
.attempt(Column::new("attempt").generated())
.processed_at(Column::new("processed_at").generated())
.payload(Column::new("payload"));
fn id(&self) -> &i64 {
&self.id
}
}
impl KeyRow for Ticket {
type Key = String;
fn partition_key(&self) -> &String {
&self.customer
}
}
impl AttemptRow for Ticket {
type Attempt = i16;
fn attempt(&self) -> &i16 {
&self.attempt
}
}
impl PayloadRow for Ticket {
type Column = Vec<u8>;
fn payload(&self) -> &[u8] {
&self.payload
}
}§Time
use std::time::Duration;
use chrono::{DateTime, Utc};
use ruststream_sqlx::DatabaseClock;
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;
// invoice_jobs: id BIGSERIAL PRIMARY KEY, retry_after TIMESTAMPTZ NOT NULL DEFAULT now(),
// processed_at TIMESTAMPTZ, payload BYTEA NOT NULL
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "invoice_jobs", clock = DatabaseClock)]
pub struct SendInvoice {
#[field(id, generated)]
id: i64,
#[field(retry_after, generated)]
retry_after: DateTime<Utc>,
#[field(processed_at, generated)]
processed_at: Option<DateTime<Utc>>,
#[field(payload)]
payload: Vec<u8>,
}
#[derive(Deserialize)]
pub struct Invoice {
number: String,
}
#[subscriber(InboxQueue::<SendInvoice>::new("invoices"))]
async fn send(invoice: &Invoice) -> HandlerOutcome {
if deliver(invoice).await {
return HandlerOutcome::ack();
}
// An hour by the database's clock, whichever host claimed the row.
HandlerOutcome::retry_after(Duration::from_secs(3600))
}
pub fn app(pool: PgPool) -> RustStream {
RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
b.include(send);
})
}A queue reads “now” when it claims a row, delays a retry and marks a row processed. By default
the host reads it: SystemClock takes the system clock, and the crate binds the time in the
column’s own type, so sqlx encodes it as the service’s own inserts do. Keep the clocks of the hosts
that serve one table in step, well within a lease.
#[inbox(clock = DatabaseClock)] makes the statements read the database’s clock instead:
statement_timestamp() on Postgres, UTC_TIMESTAMP(6) on MySQL and MariaDB, and the current time
as text on SQLite. Every host then reads one clock. The lease form reads the host’s clock, because
each settlement names the expiry its claim wrote: locked_until beside DatabaseClock does not
compile.
By hand, the table names its clock in its chain and in its type:
use chrono::{DateTime, Utc};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::{Clock, Payload, ProcessedAt, RetryAfter};
use ruststream_sqlx::{DatabaseClock, InboxSpec, InboxTable, PayloadRow};
#[derive(sqlx::FromRow)]
pub struct SendInvoice {
id: i64,
payload: Vec<u8>,
}
impl InboxTable for SendInvoice {
type Id = i64;
type Table = InboxSpec<(
Clock<DatabaseClock>,
RetryAfter<DateTime<Utc>>,
ProcessedAt<DateTime<Utc>>,
Payload,
)>;
const TABLE: Self::Table = InboxSpec::new("invoice_jobs", Column::new("id").generated())
.clock::<DatabaseClock>()
.retry_after(Column::new("retry_after").generated())
.processed_at(Column::new("processed_at").generated())
.payload(Column::new("payload"));
fn id(&self) -> &i64 {
&self.id
}
}
impl PayloadRow for SendInvoice {
type Column = Vec<u8>;
fn payload(&self) -> &[u8] {
&self.payload
}
}A clock of the service’s own implements Clock and takes the same place,
#[inbox(clock = ..)] or .clock::<..>(); the example of SystemClock shows one.
A time column holds a QueueTime: chrono::DateTime<Utc> under the chrono feature, or
time::OffsetDateTime under time. SQLite keeps times as text, where time values sort right
only across seconds (SQLite).
A test reads the real clock too: it starts the service with TestApp::start_live, and
tb.advance(by) lets that much real time pass
(Testing a service on the inbox).
§Row mode
use ruststream_sqlx::prelude::*;
use serde::{Deserialize, Serialize};
use sqlx::{PgPool, Postgres};
// email_jobs: job_id BIGSERIAL PRIMARY KEY, name TEXT, attempt SMALLINT DEFAULT 1, "to" TEXT,
// subject TEXT NULL, attachments JSONB
/// One mail to send: the table has no payload column, so a handler takes the row itself.
#[derive(Inbox, sqlx::FromRow, Clone)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
#[field(id, generated)]
job_id: i64,
#[field(group)]
name: String,
#[field(attempt, generated)]
attempt: i16,
to: String,
subject: Option<String>,
#[sqlx(json)]
attachments: Vec<Attachment>,
}
#[derive(Clone, Serialize, Deserialize)]
pub struct Attachment {
file: String,
}
#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send(email: &SendEmail) -> HandlerOutcome {
if deliver(email).await {
HandlerOutcome::ack()
} else {
HandlerOutcome::retry()
}
}
/// One SMTP session per claim of up to 64 rows.
#[subscriber(InboxQueue::<SendEmail>::new("newsletters"))]
async fn send_newsletter(emails: &[SendEmail]) -> HandlerOutcome {
tracing::info!(count = emails.len(), "sent a newsletter page");
HandlerOutcome::ack()
}
/// The audit row commits with the acknowledgement of the mail's row, or neither does.
#[subscriber(InboxQueue::<SendEmail>::new("receipts"))]
async fn send_receipt(email: &SendEmail, Ctx(mut tx): Ctx<keys::Tx<Postgres>>) -> HandlerOutcome {
let logged = sqlx::query("INSERT INTO sent_mail (job_id, recipient) VALUES ($1, $2)")
.bind(email.job_id)
.bind(&email.to)
.execute(&mut *tx)
.await;
if logged.is_ok() && deliver(email).await {
HandlerOutcome::ack()
} else {
HandlerOutcome::retry()
}
}
pub fn app(pool: PgPool) -> RustStream {
RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
b.include(send);
b.include(send_newsletter.batch(nonzero!(64)));
b.include(send_receipt.transactional());
})
}By hand, the table leaves .payload(..) out, names its own columns with .data(..), and writes the
core’s carried lane for its row. Its Insert writes a task with the text dialect::insert
renders from the description, as the derive’s does:
use ruststream::runtime::{Input, SoloCarried};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::dialect::insert::{self, Sql};
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::spec::Attempt;
use ruststream_sqlx::{AttemptRow, InboxSpec, InboxTable};
use serde::{Deserialize, Serialize};
use sqlx::{PgConnection, PgPool};
#[derive(sqlx::FromRow, Clone)]
pub struct SendEmail {
job_id: i64,
name: String,
attempt: i16,
to: String,
subject: Option<String>,
#[sqlx(json)]
attachments: Vec<Attachment>,
}
#[derive(Clone, Serialize, Deserialize)]
pub struct Attachment {
file: String,
}
impl InboxTable for SendEmail {
type Id = i64;
type Table = InboxSpec<(Attempt,)>;
const TABLE: Self::Table = InboxSpec::new("email_jobs", Column::new("job_id").generated())
.group(Column::new("name"))
.attempt(Column::new("attempt").generated())
.data(&[Column::new("to"), Column::new("subject"), Column::new("attachments")]);
fn id(&self) -> &i64 {
&self.job_id
}
}
// Row mode: a handler takes the row itself.
impl Input for SendEmail {
type Axis = SoloCarried<Self>;
}
impl AttemptRow for SendEmail {
type Attempt = i16;
fn attempt(&self) -> &i16 {
&self.attempt
}
}
// INSERT INTO "email_jobs" ("name", "to", "subject", "attachments") VALUES ($1, $2, $3, $4)
const INSERT: Sql<128> = insert::postgres(&SendEmail::TABLE.spec());
impl Insert<PgConnection> for SendEmail {
async fn insert(&self, conn: &mut PgConnection) -> Result<(), sqlx::Error> {
sqlx::query(INSERT.as_str())
.bind(&self.name)
.bind(&self.to)
.bind(&self.subject)
.bind(sqlx::types::Json(&self.attachments))
.execute(conn)
.await?;
Ok(())
}
}A table whose struct has no #[field(payload)] field, or whose description sets no .payload(..),
is in row mode. Its handler takes the struct itself, &SendEmail, as the driver read it. The struct
derives Clone, because the test harness keeps a copy of each value. It does not derive
Deserialize: a type that deserializes rides the codec, and Inbox shows the error. Serde
appears only where sqlx needs it, such as the #[sqlx(json)] column above.
A delivery lends its handler the row the driver read, with no codec and no copy. A batch handler
takes &[SendEmail]: the rows of one claim, lent as one slice in claim order (RowBatch). A
batch handler mounts in the plain form, with no reply and no Out slots. A handler of one row
mounts in every form the core offers, a reply and Out slots included.
The column types are the service’s to get right, as in payload mode. A row the driver cannot read
settles by the subscription’s decode policy, on_failure(decode = ..), and the driver’s error
goes to the log. A claimed id whose row is gone settles the same way. A nullable column takes
Option<T>: SQLite’s driver reads a NULL text column into a String as an empty string.
Every form serves row mode: the row lock, the lease and the advisory lock. Transactional mode
serves it too, and the handler takes &SendEmail beside Ctx<keys::Tx<DB>>.
The headers column becomes the delivery’s headers, which middleware reads. The row a handler
reads holds that column empty. A message assembled from a headers struct builds its delivery’s
headers from that struct instead (the headers layout).
A task is written by insert, in the service’s own transaction or in the
handler’s: the derive generates it, and a table described by hand writes it over the text
dialect::insert renders from its description. The columns bind in the description’s order: the
roles, then the data columns. Repository writes a row-mode table through a Publish of the
service’s own. Routes and by-name subscriptions read payload-mode tables: they choose a table by a
name at run time, and a row type is chosen at compile time.
A handler that decodes a payload, mounted on a row-mode table, fails each delivery by the decode policy. The subscription logs one warning that names the table, row mode and the row type the handler should take.
In a test, with_value(&row) compares the row a handler took, and
received_values::<SendEmail>() returns every row the subscription lent. The generated
AsyncAPI document describes the row through its JsonSchema derive
(ruststream::schemars::JsonSchema), as it describes any value a broker lends.
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)));
})
}§The headers layout
use ruststream_sqlx::Fetch;
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::{PgConnection, PgPool, Postgres};
// order_jobs: job_id BIGSERIAL PRIMARY KEY, name TEXT, attempt SMALLINT DEFAULT 1, tenant TEXT,
// trace TEXT NULL, order_id BIGINT, note TEXT NULL
// orders: id BIGINT PRIMARY KEY, customer TEXT, total BIGINT
/// The queue table: the columns that run the queue, and the headers `tenant`, `trace` and
/// `order_id`.
#[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,
trace: Option<String>,
order_id: i64,
}
/// A job and its note, read from the queue table by the default fetch.
#[derive(Debug, Clone, Inbox, sqlx::FromRow)]
pub struct OrderJob {
#[field(headers)]
#[sqlx(flatten)]
headers: OrderHeaders,
note: Option<String>,
}
/// A job and the order it is for, read by the service's own fetch over a join.
#[derive(Debug, Clone, Inbox, sqlx::FromRow)]
#[inbox(custom(fetch))]
pub struct OrderMail {
#[field(headers)]
#[sqlx(flatten)]
headers: OrderHeaders,
customer: String,
total: i64,
}
impl Fetch<Postgres> for OrderMail {
async fn fetch(conn: &mut PgConnection, ids: &[i64]) -> Result<Vec<Self>, sqlx::Error> {
sqlx::query_as(
"SELECT j.job_id, j.name, j.attempt, j.tenant, j.trace, j.order_id, \
o.customer, o.total \
FROM order_jobs j JOIN orders o ON o.id = j.order_id \
WHERE j.job_id = ANY($1)",
)
.bind(ids)
.fetch_all(conn)
.await
}
}
#[derive(Deserialize)]
pub struct Tenant {
tenant: String,
trace: Option<String>,
}
#[subscriber(InboxQueue::<OrderJob>::new("packing"))]
async fn pack(job: &OrderJob) -> HandlerOutcome {
tracing::info!(attempt = job.headers.attempt, note = ?job.note, "packing");
HandlerOutcome::ack()
}
// The header map is built from `OrderHeaders` on its first read, here by `Headers<Tenant>`.
#[subscriber(InboxQueue::<OrderMail>::new("receipts"))]
async fn send_receipt(mail: &OrderMail, Headers(tenant): Headers<Tenant>) -> HandlerOutcome {
tracing::info!(
tenant = %tenant.tenant,
trace = ?tenant.trace,
customer = %mail.customer,
total = mail.total,
"sending the receipt"
);
HandlerOutcome::ack()
}
pub fn app(pool: PgPool) -> RustStream {
RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
b.include(pack);
// A job whose order is gone fails to decode: the default policy drops it.
b.include(send_receipt);
})
}By hand, one description on the message holds the whole table. The headers struct is a plain sqlx
struct, .data(..) names the header columns, .fetching(..) the message’s own, and
.header_fields() builds the header map from the fields through HeaderFields:
use ruststream::HeaderMap;
use ruststream::runtime::{Input, SoloCarried};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::spec::{self, Attempt};
use ruststream_sqlx::{AttemptRow, HeaderFields, InboxSpec, InboxTable, put_header};
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct OrderHeaders {
job_id: i64,
attempt: i16,
tenant: String,
trace: Option<String>,
order_id: i64,
}
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct OrderJob {
#[sqlx(flatten)]
headers: OrderHeaders,
note: Option<String>,
}
impl InboxTable for OrderJob {
type Id = i64;
type Table = InboxSpec<(Attempt, spec::HeaderFields)>;
const TABLE: Self::Table = InboxSpec::new("order_jobs", Column::new("job_id").generated())
.group(Column::new("name"))
.attempt(Column::new("attempt").generated())
.data(&[Column::new("tenant"), Column::new("trace"), Column::new("order_id")])
.fetching(&[Column::new("note")])
.header_fields();
fn id(&self) -> &i64 {
&self.headers.job_id
}
}
impl Input for OrderJob {
type Axis = SoloCarried<Self>;
}
impl AttemptRow for OrderJob {
type Attempt = i16;
fn attempt(&self) -> &i16 {
&self.headers.attempt
}
}
impl HeaderFields for OrderJob {
const NAMES: &'static [&'static str] = &["tenant", "trace", "order_id"];
fn header_map(&self) -> HeaderMap {
let mut headers = HeaderMap::with_capacity(Self::NAMES.len());
put_header(&mut headers, "tenant", &self.headers.tenant);
put_header(&mut headers, "trace", &self.headers.trace);
put_header(&mut headers, "order_id", &self.headers.order_id);
headers
}
}A message read by a fetch of its own adds .own::<spec::own::Fetch>() to the chain, with
spec::own::Fetch in the type, and implements Fetch as OrderMail does above.
In the headers layout of the derives, two structs describe one queue. The headers struct derives
InboxHeaders and describes the queue table. The message struct derives
Inbox and is what a handler takes. It holds the headers struct in a field marked
#[field(headers)] and #[sqlx(flatten)], beside data of its own. Without #[sqlx(flatten)], a
headers field is a header column, as in a flat struct.
The headers layout is row mode. A handler takes the message struct itself, &OrderJob, as the
driver read it, with no codec. A batch handler takes &[OrderJob], the rows of one claim. The
message struct derives Clone and does not derive Deserialize, as in row mode.
Several message structs may flatten one headers struct, each read by its own subscriptions.
§Where the attributes go
The headers struct carries everything that describes the table: #[inbox(table, schema, advisory_lock, clock, isolation, mode)] and the fields that play a role (id, group,
attempt, locked_until and the rest). So the headers struct selects the form: a
#[field(locked_until)] field gives the lease form, advisory_lock = ".." the advisory lock
form, with a key built from the headers struct’s fields. The message’s deliveries hold the lease
or the lock as a flat struct’s do.
The message struct carries #[inbox(custom(..))] alone, and no role but headers. A role or a
table attribute on the message does not compile, and the error names the headers struct as the
place for it. A custom event the headers struct’s form does not take does not compile either:
claim in the advisory lock form, extend outside the lease form, lock and unlock outside
the advisory lock form.
The headers struct gets the generated insert, which writes the table’s columns.
The message struct gets none: the service writes the message’s own data, or writes the whole row
through a Publish of its own. A table described by hand writes its insert itself, over the text
dialect::insert renders (row mode shows one).
§The default fetch
The claim selects the ids of the rows it takes, and a fetch reads their rows. A message struct
without custom(fetch) is read by the default fetch. It names the headers struct’s columns and
the message’s own fields, and reads them from the queue table. A subscription prepares it when it
opens: a message field whose column the table lacks stops the subscription at startup with
SqlxBrokerError::Schema, which names the table and the statement.
§The header values
Every field of the headers struct without a role is a header of the delivery. Its name is the
column’s name, as sqlx’s rename and rename_all set it. Its value comes from HeaderField:
- strings and byte vectors give their bytes;
- integers and
boolgive their decimal text; chrono::DateTime<Utc>(featurechrono) andtime::OffsetDateTime(featuretime) give their RFC 3339 text;Option<T>gives its value, andNoneleaves the header out.
A service implements HeaderField for a type of its own. A field of a type without it does not
compile, and the error offers the impl or a role.
The header map is built on the first read of the delivery’s headers, by middleware or an
extractor such as Headers<T>. A delivery whose headers nobody reads never builds it. The fields
that play a role are not headers: the handler reads them on the message.
A message published into the table through a Repository carries only the headers the headers
struct names. A publish with another header fails with SqlxBrokerError::Header, so no header
is lost on the way into the table.
§A fetch over a join
A message whose data lives in other tables lists fetch in #[inbox(custom(..))] and implements
Fetch for its database. The fetch takes the claimed ids and returns the rows it found. Its
SELECT returns the headers struct’s columns under their names, beside the message’s own, and
the crate matches the rows to the claimed ids by the headers struct’s id field. The fetch
selects every column of the headers struct, locked_until included in the lease form.
The fetch runs on the claim’s connection: in the claim’s transaction in the row lock form, and once per row, with that row’s id alone, in the advisory lock form. A subscription to such a message does not check the message’s own columns at startup, because the fetch reads them from wherever they live.
§A claimed id without a row
A claimed id that the fetch returned no row for, such as a job whose order the join does not find,
settles by the subscription’s decode policy, on_failure(decode = ..), before the handler runs.
The default policy drops the row. The log names the id. A batch lends its handler the rows the
fetch found, and the policy settles the others. The headers of such a delivery are empty.
§Row locks, leases or advisory locks
use std::time::Duration;
use chrono::{DateTime, Utc};
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;
// report_jobs: id BIGSERIAL PRIMARY KEY, attempt SMALLINT NOT NULL DEFAULT 1,
// locked_until TIMESTAMPTZ, payload BYTEA NOT NULL
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "report_jobs")]
pub struct RenderReport {
#[field(id, generated)]
id: i64,
#[field(attempt, generated)]
attempt: i16,
#[field(locked_until)]
locked_until: Option<DateTime<Utc>>,
#[field(payload)]
payload: Vec<u8>,
}
#[derive(Deserialize)]
pub struct Report {
id: u64,
}
// A report takes minutes to render. The claim has committed, and the subscription extends the
// lease while the handler runs; after a crash the row waits out its minute, then returns.
#[subscriber(InboxQueue::<RenderReport>::new("reports").lease(Duration::from_secs(60)))]
async fn handle(report: &Report) -> HandlerOutcome {
render(report).await;
HandlerOutcome::ack()
}
pub fn app(pool: PgPool) -> RustStream {
RustStream::new(AppInfo::new("reports", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
b.include(handle);
})
}By hand, .lease(..) selects the form, and Lease<Time> in the type gives the lease’s time type:
use chrono::{DateTime, Utc};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::{Attempt, Lease, Payload};
use ruststream_sqlx::{AttemptRow, InboxSpec, InboxTable, PayloadRow};
#[derive(sqlx::FromRow)]
pub struct RenderReport {
id: i64,
attempt: i16,
payload: Vec<u8>,
}
impl InboxTable for RenderReport {
type Id = i64;
type Table = InboxSpec<(Attempt, Lease<DateTime<Utc>>, Payload)>;
const TABLE: Self::Table = InboxSpec::new("report_jobs", Column::new("id").generated())
.attempt(Column::new("attempt").generated())
.lease(Column::new("locked_until"))
.payload(Column::new("payload"));
fn id(&self) -> &i64 {
&self.id
}
}
impl AttemptRow for RenderReport {
type Attempt = i16;
fn attempt(&self) -> &i16 {
&self.attempt
}
}
impl PayloadRow for RenderReport {
type Column = Vec<u8>;
fn payload(&self) -> &[u8] {
&self.payload
}
}A table takes its rows in one of three forms, and its description decides which.
A table with no form set takes its rows by row lock: no locked_until or advisory_lock on the
derive, no .lease(..) or .advisory(..) by hand. A claim locks each row with FOR UPDATE SKIP LOCKED in a transaction that stays open while the handler runs, and the settlement’s statement
commits that transaction. A row and its outcome change together, and the rows of a crashed process
return at once, when the database rolls its transactions back. The price is a connection of the pool
per message in work, held for the whole handler.
A #[field(locked_until)] field, or .lease(..) by hand, selects the lease form (LeaseRow).
The claim writes a lease into each row and commits at once, so the handler runs with no transaction
open and no connection held. Declare it for handlers that run long and for a pool that cannot spare
a connection per message in work.
advisory_lock = ".." in #[inbox(..)], or .advisory(..) by hand, selects the advisory lock
form. A lock on each row’s key holds the row. The connection that holds the delivery keeps that
lock, and no transaction stays open. Rows that share a key go into work one at a time. The locks of
a crashed process end with its connections, and its rows return. The price is a connection of the
pool per message in work, as in the row lock form.
SQLite has no row locks, so a SQLite table takes the lease or the advisory lock form.
A subscription mounted with workers(n) keeps a claim in flight for each free worker, so its
workers take rows at once instead of one after another. It never holds more than the pool’s size
less one connection, so the pool always keeps a connection for the handlers’ own queries, their
publishes and the settlements that take one. A claim beside another one starts only while the
pool has a connection to spare, so subscriptions that share a pool, and handlers that hold its
connections, leave each other room. A pool short of room runs fewer deliveries at once.
A subscription keeps one claim in flight where claims that run at once would cost more than they
give. On SQLite, which takes one writer at a time, they only collide. On a table whose groups keep
their order, or with a partition_key column for workers_by_key(n), they could finish out of
claim order. On MySQL and MariaDB a lease claim stamps its rows in a transaction of several
statements, which locks the rows it reads, leases in work included, until it commits.
What a subscription holds depends on the form. In the row lock form each delivery in work and
each claim in flight holds a connection: with workers(n) up to n. In the lease form a claim and
a settlement each take a connection only for their statements, and each subscription takes one
each half lease to extend the leases in work. In the advisory lock form each delivery in work
holds a connection of its own until it settles, in a batch too. In
transactional mode each delivery in work holds a connection for the
transaction its handler writes through, in every form.
Size the pool for every subscription that holds connections: n for each workers(n), one more for
the pool, and the connections the handlers take for their own queries and inserts. A publish takes
a connection of its own for its insert. In the lease form a pool of n + 1 lets every free worker
claim at once.
A handler that computes runs on dedicated threads with
on_threads, the inbox’s spelling of threads(n):
b.include(scan.on_threads(InboxThreads::new(nonzero!(4)).connections(nonzero!(2))));The subscription claims on the app’s runtime and runs each handler on one of its threads.
Claims, settlements, transactions and publishes take the service’s pool, and their connections
open on the app’s runtime. A handler’s Ctx<keys::Pool<DB>> is the thread’s own pool, with the
service pool’s options. It opens one connection at most, or the number
connections sets, on the thread, and they close when the
thread ends. The service then opens the pool’s connections and threads times connections for each
such subscription. connection_limit holds that sum to
what the database allows: a subscription that would pass it does not start. The framework’s own
threads(n) gives the threads pools of one connection too, and the limit does not count them.
A subscription’s threads hold deliveries in their rings beside the handlers, and in the row lock
form each one holds a connection of the service’s pool, so the pool’s bound applies as with
workers(n).
§The lease form
- The claim writes the lease’s expiry into
locked_until, counts the attempt and commits. - The claim reads “now” once: the rows it finds due, the leases it finds ended and the expiry it writes start from that instant.
- The expiry is the delivery’s ownership token: a settlement takes effect only while the row still holds it.
- The subscription extends the lease of every delivery in work each half lease, so a handler may run longer than its lease. Each extension moves the token forward.
- A lease is how long a crashed process’s row stays out of the queue: thirty seconds, unless
SqlxBroker::leasesets another for the broker orInboxQueue::leasefor one subscription. It is whole seconds, at least one, and a shorter one is rounded up. - Ownership ends with the lease, not with its holder. A lease runs out under a running handler
only when its extensions fail or stop: the database out of reach, the subscription closed, the
broker shut down. The next claim then takes the row while the first handler still runs, and the
token stops the late one from settling: its settlement fails with
SqlxBrokerError::LeaseLost. - A delivery dropped unsettled releases its row at once. A settlement whose statement fails keeps the row until the lease runs out, as a crash does.
- The claim counts the attempt, so a crash spends one too. A delivery reports the attempt as it stood before its claim, so the first delivery reads 1, as in the row lock form.
- The crate computes the expiry on the host, because every settlement names the expiry its claim
wrote: a struct with
locked_untilonclock = DatabaseClockdoes not compile, and neither does a description withLease<Time>besideClock<DatabaseClock>. - After the broker shuts down, a delivery in work keeps its lease and settles as before; the lease is no longer extended.
§The advisory lock form
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);
})
}By hand, .advisory(..) takes the key as its parts, text and columns:
use ruststream_sqlx::dialect::{Column, KeyPart};
use ruststream_sqlx::spec::{Advisory, Attempt, Payload};
use ruststream_sqlx::{AttemptRow, InboxSpec, InboxTable, PayloadRow};
#[derive(sqlx::FromRow)]
pub struct Transcode {
job_id: i64,
attempt: i16,
payload: Vec<u8>,
}
impl InboxTable for Transcode {
type Id = i64;
type Table = InboxSpec<(Advisory, Attempt, Payload)>;
const TABLE: Self::Table = InboxSpec::new("jobs", Column::new("job_id").generated())
.advisory(&[KeyPart::Literal("jobs-"), KeyPart::Column("job_id")])
.attempt(Column::new("attempt").generated())
.payload(Column::new("payload"));
fn id(&self) -> &i64 {
&self.job_id
}
}
impl AttemptRow for Transcode {
type Attempt = i16;
fn attempt(&self) -> &i16 {
&self.attempt
}
}
impl PayloadRow for Transcode {
type Column = Vec<u8>;
fn payload(&self) -> &[u8] {
&self.payload
}
}advisory_lock = "jobs-{job_id}" names each row’s lock key. A placeholder names a field, the key
reads its column, and the database renders the key as text. By hand, KeyPart::Column names the
column itself, and a column the table lacks stops the subscription at startup, when its statements
are prepared. Rows whose keys match go into work one at a time. A key on the row’s id holds each row
alone. A key on another column, as in advisory_lock = "accounts-{account}", keeps the rows of an
account in work one at a time. A key belongs to the database, not to its table. Two tables whose
keys match wait for each other, and a key that starts with its table’s name keeps them apart.
A claim selects candidates: as many due rows of the queue as it may take, in claim order, each with its key. It locks each candidate’s key on a connection of its own, without waiting, and passes over a key held elsewhere. Then it takes the row while the row is still claimable: the take counts the attempt and reads the row. A row that another holder settled between the select and the lock is passed over, and its lock released. The take commits the count, so a crash spends an attempt, as in the lease form. A delivery reports the attempt as it stood before its claim.
A settlement runs its statement on the delivery’s connection, where it commits on its own. Then
the lock is released, and the connection goes back to the pool. retry() runs no statement: the
claim counted the attempt, and the release returns the row. A settlement whose statement fails
still releases the lock, and the row returns as it was. A release that the database does not
confirm closes the connection instead, which ends the session and its locks.
A delivery dropped unsettled releases its lock and closes its connection, in a task on the runtime the broker connected on, and its row returns. One timeout of five seconds bounds the release and the close. Past it the connection drops, and the server ends the session and its lock all the same.
shutdown releases the lock of every delivery in work. Each connection releases its key and goes
back to the pool, or closes where the database does not confirm the release. shutdown waits for
each settlement in flight and each connection still closing, and returns once the broker holds no
lock. A delivery whose lock it released settles no more: its settlement fails with
SqlxBrokerError::Closed, and its row is back in the queue. The ClosedSqlxBroker it
returns counts the two outcomes. locks_released counts the
locks released with their connections back in the pool, and
connections_closed the connections that held a lock and
closed instead. In a service the runtime drains the handlers before it shuts the broker down. A
handler aborted at the drain timeout drops its delivery, and shutdown waits for that connection
to close.
A delivery holds a connection of its own because its lock lives in that connection’s session: a
delivery dropped unsettled closes its own connection and leaves the other rows alone. A
subscription with workers(n) holds up to n connections for its deliveries and its claims, and a
batch of n rows holds n. A batch takes the connections the pool gives at once: idle ones first, then
new ones while the pool has room. It ends where the pool is full, so a batch larger than the pool
shrinks instead of waiting. Handlers that publish need room for their inserts on top.
Each database keeps the locks in its own way, and keeps the locks of two databases apart:
- Postgres locks a 64-bit hash of the key (
hashtextextended, Postgres 11 or later), and a claim leaves out the keys other sessions hold. Two keys with one hash wait for each other: a delay, never a double delivery. A lock lives in the database session that took it, so the form needs a direct connection orPgBouncerin session pooling. In transaction pooling the session that took a lock serves other clients between statements, and the lock goes with it. - MySQL and MariaDB lock a name with
GET_LOCK: the table’s database in lower case, a dot and the key, as inapp.jobs-7. A name longer than the 64 characters the server takes is locked by its SHA-256. A claim leaves out the names in use. - SQLite has no locks that a session holds, so the process keeps the keys in work, by database
and key. Two keys with one hash wait for each other here too. A claim’s select cannot see the
keys in work: it reads the first due rows alone and takes nothing while they are in work. A
handler of single messages therefore has one delivery in work at a time, and with
workers(n)each next row waits up to a poll interval. The registry serves one process per database file. Two processes on one file do not see each other’s keys, and a row may be delivered twice.
§A lock of the service’s own
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::{Lock, Unlock};
use serde::Deserialize;
use sqlx::{PgConnection, PgPool, Postgres};
// payouts: id BIGSERIAL PRIMARY KEY, account_id BIGINT NOT NULL, payload BYTEA NOT NULL
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "payouts", advisory_lock = "{account_id}", custom(lock, unlock))]
pub struct Payout {
#[field(id, generated)]
id: i64,
account_id: i64,
#[field(payload)]
payload: Vec<u8>,
}
// The billing system holds `pg_advisory_lock(account_id)` while it changes an account. The queue
// takes the same lock, so a payout never runs beside a billing run of its account.
impl Lock<Postgres> for Payout {
async fn lock(conn: &mut PgConnection, key: &str) -> Result<bool, sqlx::Error> {
sqlx::query_scalar("SELECT pg_try_advisory_lock($1::bigint)")
.bind(key)
.fetch_one(conn)
.await
}
}
impl Unlock<Postgres> for Payout {
async fn unlock(conn: &mut PgConnection, key: &str) -> Result<bool, sqlx::Error> {
sqlx::query_scalar("SELECT pg_advisory_unlock($1::bigint)")
.bind(key)
.fetch_one(conn)
.await
}
}
#[derive(Deserialize)]
pub struct Transfer {
cents: i64,
}
#[subscriber(InboxQueue::<Payout>::new("payouts"))]
async fn pay(transfer: &Transfer) -> HandlerOutcome {
transfer_to_bank(transfer).await;
HandlerOutcome::ack()
}
pub fn app(pool: PgPool) -> RustStream {
RustStream::new(AppInfo::new("payouts", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
b.include(pay);
})
}custom(lock, unlock) in #[inbox(..)] hands the lock and the release of each key to the service:
the struct implements Lock and Unlock for its database. By hand, the chain adds
.own::<own::Lock>() and .own::<own::Unlock>(), and the type lists own::Lock and own::Unlock
beside Advisory. The two come together, and only in the advisory lock form; a description that
breaks either rule does not compile. The dialect still selects the candidates and takes each row.
The lock tries without waiting and answers whether it took the key. The release answers whether the
session held the key, and one that answers false or fails closes the connection. The process keeps
no registry of keys for such a table, on SQLite too. The dialect’s select leaves out only the keys
its own locks hold, so a claim reads the first due rows alone and takes nothing while their keys are
held. A handler of single messages therefore has one delivery in work at a time, and with
workers(n) each next row waits up to a poll interval.
A database the crate builds no dialect for takes the form through a dialect of the service’s own
that implements Advisory: the claim of candidates, the lock, the release
and the take. On SQL Server its lock runs sp_getapplock with the session as the owner and no
wait, and its release runs sp_releaseapplock. Each of the two answers one 64-bit integer,
nonzero where it took or released the lock.
§Transactional mode
use std::error::Error;
use ruststream_sqlx::prelude::*;
use serde::{Deserialize, Serialize};
use sqlx::{PgConnection, PgPool, Postgres};
// signup_jobs: id BIGSERIAL PRIMARY KEY, attempt SMALLINT NOT NULL DEFAULT 1,
// payload BYTEA NOT NULL
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "signup_jobs")]
pub struct OpenAccount {
#[field(id, generated)]
id: i64,
#[field(attempt, generated)]
attempt: i16,
#[field(payload)]
payload: Vec<u8>,
}
// welcome_jobs, the mailer's queue: id BIGSERIAL PRIMARY KEY, payload BYTEA NOT NULL
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "welcome_jobs")]
pub struct SendWelcome {
#[field(id, generated)]
id: i64,
#[field(payload)]
payload: Vec<u8>,
}
#[derive(Deserialize)]
pub struct Signup {
email: String,
}
#[derive(Serialize)]
pub struct Welcome<'a> {
to: &'a str,
}
// The account and its welcome email, written on the connection of the delivery's transaction.
async fn open(
conn: &mut PgConnection,
signup: &Signup,
) -> Result<(), Box<dyn Error + Send + Sync>> {
sqlx::query("INSERT INTO accounts (email) VALUES ($1)")
.bind(&signup.email)
.execute(&mut *conn)
.await?;
let welcome = SendWelcome {
id: 0,
payload: serde_json::to_vec(&Welcome { to: &signup.email })?,
};
welcome.insert(conn).await?;
Ok(())
}
#[subscriber(InboxQueue::<OpenAccount>::new("signups"))]
async fn open_account(
signup: &Signup,
Ctx(mut tx): Ctx<keys::Tx<Postgres>>,
Ctx(pool): Ctx<keys::Pool<Postgres>>,
Ctx(attempt): Ctx<keys::Attempt>,
) -> HandlerOutcome {
if let Err(error) = open(&mut *tx, signup).await {
// Through the pool: the note stays when the retry rolls the transaction back.
let _ = sqlx::query("INSERT INTO signup_errors (email, error) VALUES ($1, $2)")
.bind(&signup.email)
.bind(format!("attempt {}: {error}", attempt.unwrap_or(1)))
.execute(&pool)
.await;
return HandlerOutcome::retry();
}
// The commit keeps the account and its welcome email, and finishes the signup.
HandlerOutcome::ack()
}
pub fn app(pool: PgPool) -> RustStream {
RustStream::new(AppInfo::new("accounts", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
b.include(open_account.transactional());
})
}.transactional() at the mount site switches an InboxQueue subscription to transactional
mode, where the handler writes through its delivery’s transaction. The handler takes that
transaction as its first Ctx parameter, Ctx(mut tx): Ctx<keys::Tx<Postgres>>, and runs its
statements through &mut *tx. A handler that takes keys::Tx on a subscription without the step
does not compile, and the error names the step.
Acknowledgement commits the handler’s writes together with the row’s settlement. Every other
outcome rolls them back first, then settles the row as it would without the step. A handler that
panics while it holds tx commits none of its writes, even where its panic policy acknowledges
the delivery. A delivery dropped unsettled closes its connection, and the server rolls its
transaction back.
A task the handler writes through tx into a queue table, with Insert or its own SQL, commits
with the acknowledgement too. A write through the pool that Ctx<keys::Pool<DB>> gives commits on
its own, whatever the outcome. So does a publish through the broker.
A handler’s first Ctx decides which of the inbox’s keys compile after it: the other two after
keys::Tx, only keys::Attempt after keys::Pool, none after keys::Attempt. The order
keys::Tx, keys::Pool, keys::Attempt fits every handler, with the keys it does not need left
out.
What the transaction is depends on the form:
- In the row lock form it is the claim’s transaction. The subscription sets a savepoint right after each claim, one more statement per delivery. A settlement other than acknowledgement rolls back to it, so a retry discards the handler’s writes and still counts the attempt.
- In the lease form the crate opens a transaction for each delivery once its claim has committed.
Acknowledgement runs in it and takes effect only while the row still holds the delivery’s lease.
A lost lease rolls the whole transaction back, and the settlement returns
SqlxBrokerError::LeaseLost. - In the advisory lock form the crate opens the transaction on the connection that holds the row’s key, after the take. The settlement ends the transaction before it releases the lock.
Each of these transactions opens at the table’s isolation level or SQLite mode
(Isolation and mode). On SQLite a delivery’s transaction takes the
database’s write lock at its first write, or at its start under mode = immediate or
exclusive, and holds it until the delivery settles. It keeps every other writer of the database
out meanwhile, claims included.
Each delivery in work holds one connection of the pool for its transaction, in every form. A
handler that takes a second connection, through keys::Pool or a publish, needs a pool larger
than its workers(n). The subscription leaves the pool’s last connection free, so such a handler
still gets one when the pool is short, and the subscription runs fewer deliveries at once.
Transactional mode serves single deliveries: a batch handler mounted with .transactional() does
not compile.
The transaction goes back to the delivery when the handler’s Tx drops. A settlement that finds
it still out, in a task the handler moved it into, returns SqlxBrokerError::TransactionHeld,
and the transaction rolls back when that Tx drops.
On MySQL and MariaDB the settlement after a handler that cut a statement through tx short, as a
select! around a query does, can read that statement’s reply as its own, a bug of sqlx-mysql
0.9.0.
§Isolation and mode
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;
// payout_jobs: id BIGSERIAL PRIMARY KEY, payload BYTEA NOT NULL
// The database's transactions default to SERIALIZABLE; the claims of this table open at
// READ COMMITTED.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "payout_jobs", isolation = read_committed)]
pub struct SendPayout {
#[field(id, generated)]
id: i64,
#[field(payload)]
payload: Vec<u8>,
}
#[derive(Deserialize)]
pub struct Payout {
account: String,
}
#[subscriber(InboxQueue::<SendPayout>::new("payouts"))]
async fn pay(payout: &Payout) -> HandlerOutcome {
transfer(payout).await;
HandlerOutcome::ack()
}
pub fn app(pool: PgPool) -> RustStream {
RustStream::new(AppInfo::new("payouts", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
b.include(pay);
})
}By hand, .opens::<Level>() takes a marker of dialect::level, and Opens<Level> goes into
the type:
use ruststream_sqlx::dialect::{Column, level};
use ruststream_sqlx::spec::{Opens, Payload};
use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};
#[derive(sqlx::FromRow)]
pub struct SendPayout {
id: i64,
payload: Vec<u8>,
}
impl InboxTable for SendPayout {
type Id = i64;
type Table = InboxSpec<(Opens<level::ReadCommitted>, Payload)>;
const TABLE: Self::Table = InboxSpec::new("payout_jobs", Column::new("id").generated())
.opens::<level::ReadCommitted>()
.payload(Column::new("payload"));
fn id(&self) -> &i64 {
&self.id
}
}
impl PayloadRow for SendPayout {
type Column = Vec<u8>;
fn payload(&self) -> &[u8] {
&self.payload
}
}isolation = <level> in #[inbox(..)] declares an isolation level: read_uncommitted,
read_committed, repeatable_read or serializable. A SQLite table declares a mode instead, as in
#[inbox(mode = immediate)]: deferred, immediate or exclusive. SQLite runs every transaction
serializable. Its mode decides when a transaction takes the write lock (Mode). By
hand, the levels and the modes are the markers of dialect::level, and a table opens at one of
them at most.
The declaration governs the transactions the crate opens for a delivery’s work. In the row lock
form that is the claim’s transaction: it holds the rows while their handler runs, and their
settlement commits it. In transactional mode it is also the transaction the
handler writes through, in every form. A table that declares neither opens them with a plain
BEGIN on Postgres and SQLite, at the database’s default, and at READ COMMITTED on MySQL and
MariaDB.
Each database takes the levels it keeps: Postgres read_committed, repeatable_read and
serializable; MySQL and MariaDB all four; SQLite the three modes. Postgres runs READ UNCOMMITTED
as READ COMMITTED, so its dialect does not open it.
A subscription to a table at a level or mode its broker’s dialect does not open does not compile,
and neither does a route into that table. The error names the dialect and the level, and lists
what each database opens. An AnyPool names its database only when the broker connects, so there
such a subscription stops when it starts, with SqlxBrokerError::Dialect. A dialect of the
service’s own opens a level by implementing Opens for it, and its
Dialect::begin returns the statement that opens it.
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 for the row lock form there.
On MySQL and MariaDB a row lock claim at repeatable_read or serializable keeps a lock on every
row it reads until its settlement. Without an index on the group’s column it reads rows of other
groups as well, and a claim held for a handler keeps those rows from their own claims. An index on
that column lets a claim read its own group alone.
§FIFO groups
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);
})
}By hand, .fifo_group(..) names the group’s column, and Fifo goes into the type:
use chrono::{DateTime, Utc};
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::spec::{Fifo, Payload, RetryAfter};
use ruststream_sqlx::{InboxSpec, InboxTable, PayloadRow};
#[derive(sqlx::FromRow)]
pub struct SyncChange {
id: i64,
payload: Vec<u8>,
}
impl InboxTable for SyncChange {
type Id = i64;
type Table = InboxSpec<(Fifo, RetryAfter<DateTime<Utc>>, Payload)>;
const TABLE: Self::Table = InboxSpec::new("sync_jobs", Column::new("id").generated())
.fifo_group(Column::new("target"))
.retry_after(Column::new("retry_after").generated())
.payload(Column::new("payload"));
fn id(&self) -> &i64 {
&self.id
}
}
impl PayloadRow for SyncChange {
type Column = Vec<u8>;
fn payload(&self) -> &[u8] {
&self.payload
}
}#[field(group, fifo = true)] keeps each group of the table in order. A group has at most one
row in work at a time, however many brokers read it. Its rows go out in claim order: a smaller
priority first, then an earlier retry_after, then the smaller id, each where the table has the
column. The group’s first unfinished row in that order is its head. A claim takes the head alone.
The rows behind the head wait while it is in work or not yet due.
retry() keeps the row at the head. retry_after(d) gives the row a later retry_after, and it
moves behind the rows that come due before it. On a table with priority it moves only among the
rows of its own priority. A head that keeps failing with retry() holds its group back.
max_attempts(n) with dead_letter(..) moves such a row out of the group after n attempts. In
the lease form a crash with a row in work holds its group until that row’s lease runs out.
workers(n) gives no parallelism inside a group, so one worker serves a FIFO subscription. A
batch of a FIFO group holds one row, its head. Groups run side by side, each through a
subscription of its own.
FIFO groups serve the row lock and lease forms. In the advisory lock form a key on the group’s field
keeps one row of the group in work, as advisory_lock = "sync_jobs-{target}" would here. fifo = true beside advisory_lock does not compile, and the error says to put the group’s field into the
key. By hand, Fifo beside Advisory in the type does not compile either.
§Databases
use chrono::{DateTime, Utc};
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::SqlitePool;
// thumbnail_jobs: id INTEGER PRIMARY KEY, attempt INTEGER NOT NULL DEFAULT 1,
// locked_until TEXT, payload BLOB NOT NULL
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "thumbnail_jobs")]
pub struct MakeThumbnail {
#[field(id, generated)]
id: i64,
#[field(attempt, generated)]
attempt: i16,
#[field(locked_until)]
locked_until: Option<DateTime<Utc>>,
#[field(payload)]
payload: Vec<u8>,
}
#[derive(Deserialize)]
pub struct Image {
path: String,
}
#[subscriber(InboxQueue::<MakeThumbnail>::new("thumbnails"))]
async fn thumbnail(image: &Image) -> HandlerOutcome {
if resize(image).await {
HandlerOutcome::ack()
} else {
HandlerOutcome::retry()
}
}
// The same struct serves a `PgPool` and a `MySqlPool` too: the broker takes the dialect of its
// pool, which builds its statements from the table's description.
pub fn app(pool: SqlitePool) -> RustStream {
RustStream::new(AppInfo::new("thumbnails", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
b.include(thumbnail);
})
}A built-in dialect builds a subscription’s statements once, when it opens, from the table’s
description. The insert of every enabled dialect is rendered at compile time, by the derive or by
dialect::insert for a table described by hand. A database whose sqlx driver lives outside sqlx
is served by a dialect of the service’s own (SqlxBroker::with_dialect): a type that implements
Dialect, RowLock, Lease and
Advisory for the forms it serves, and ByName for subscriptions by name.
The connections a subscription holds depend on its form and on transactional mode, alike on every database (Row locks, leases or advisory locks).
§Postgres
postgres serves every form. The row lock and lease claims skip locked rows with 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.
A subscription on Postgres wakes on rows other processes write when the broker listens for
pg_notify (LISTEN/NOTIFY on Postgres). The advisory lock form
needs a direct connection or PgBouncer in session pooling
(The advisory lock form). A row lock table works at read_committed;
the stricter levels fail claims with serialization errors (Isolation and mode).
§MySQL and MariaDB
mysql serves MySQL 8.0.1 and MariaDB 10.6 or later, in every form. Claims skip locked rows in
the row lock and lease forms, which those versions added, and an advisory table shares their
floor: a subscription reads the server’s version when it opens, and an older server stops it with
SqlxBrokerError::ServerTooOld. A claim’s transaction runs at READ COMMITTED unless a row lock
table declares another level (Isolation and mode). At READ COMMITTED a
claim held for a handler locks no gaps between rows and holds back no insert into its table. A
server whose binary log records statements (binlog_format = STATEMENT) refuses writes at that
level; the row and mixed formats accept them.
A lease claim selects its rows, stamps each one with its lease and commits. The expiry is a whole
second, so a DATETIME column without fractions holds the token exactly. A dead letter into a
table is two statements in one transaction. A claim of the service’s own comes with a Fetch
of its own, because the crate reads rows by a list of ids on Postgres alone; a subscription
without one stops at startup.
Index a table’s claim order: (group, priority, retry_after, id), for the columns the table has.
A row lock or lease claim locks every row of its group that it reads before sorting them, so
without that index a claim held for a handler keeps the rest of its group from other claims. In the
row lock form the next claim passes over those rows to later ones, and the rows of a partition key
go into work out of order. In a table with FIFO groups a claim first locks the unfinished rows of
its group until its transaction ends, which in the row lock form is when the delivery settles. The same index keeps that read to
the group’s own rows. A row lock table with FIFO groups that declares serializable stops its
subscription when it opens, with SqlxBrokerError::Dialect: at that level InnoDB locks every
row a read touches, and a claim would wait for the group’s row in work.
§SQLite
sqlite serves the lease and advisory lock forms. SQLite locks the whole database for a writer,
so no claim can hold rows for a handler: its dialect implements no
RowLock, and a subscription to a table without locked_until or
advisory_lock does not compile. A lease claim is one UPDATE .. RETURNING that leases its rows,
and one writer at a time keeps two claims apart. The rows of one lease claim come in no
particular order. A claim of the service’s own opens its transaction with BEGIN IMMEDIATE and
comes with a Fetch of its own. The advisory lock form keeps its locks in the process
(The advisory lock form).
SQLite keeps times as text and compares them as text. chrono times sort exactly. time values
sort right only across seconds, so with them a lease may end up to a second late, and a delayed
retry may come back up to a second early or late (QueueTime). The generated document
describes the server by its protocol alone, with no host.
§An AnyPool
any serves an AnyPool. The broker picks the built-in dialect of the database the pool
reaches when it connects, among the dialects whose features are on; another backend stops
connect with SqlxBrokerError::Backend. A row holds the types sqlx::Any carries: integers,
text and bytes, with no time and no JSON. An AnyPool therefore serves the row lock form on its
Postgres and MySQL backends and the advisory lock form on all three, without retry_after,
processed_at or headers. A lease table is out of its reach, and on a SQLite backend a row
lock table stops its subscription at startup.
§A dialect of the service’s own
use std::num::NonZeroUsize;
use ruststream_sqlx::dialect::{
ClaimShape, Dialect, Param, Role, RowLock, Statement, StatementError, TableSpec,
};
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;
// email_jobs: job_id INT8 PRIMARY KEY DEFAULT unique_rowid(), name STRING NOT NULL,
// payload BYTES NOT NULL
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
#[field(id, generated)]
job_id: i64,
#[field(group)]
name: String,
#[field(payload)]
payload: Vec<u8>,
}
/// CockroachDB, reached through sqlx's Postgres driver.
#[derive(Debug)]
pub struct Cockroach;
impl Dialect for Cockroach {
fn name(&self) -> &'static str {
"cockroach"
}
fn quote_into(&self, ident: &str, out: &mut String) {
out.push('"');
out.push_str(&ident.replace('"', "\"\""));
out.push('"');
}
fn placeholder_into(&self, index: NonZeroUsize, out: &mut String) {
out.push('$');
out.push_str(&index.to_string());
}
fn ack(&self, spec: &TableSpec<'_>) -> Result<Statement, StatementError> {
let mut sql = String::from("DELETE FROM ");
self.quote_into(spec.table(), &mut sql);
sql.push_str(" WHERE ");
self.quote_into(spec.id().name(), &mut sql);
sql.push_str(" = $1");
Ok(Statement::new(sql, [Param::Id]))
}
}
// The row lock form, which `SendEmail` takes: the due rows of the subscription's group in claim
// order, locked for the claim's transaction, with the rows other claims hold passed over.
impl RowLock for Cockroach {
fn lock_claim(
&self,
spec: &TableSpec<'_>,
shape: ClaimShape,
) -> Result<Statement, StatementError> {
if spec.is_fifo() {
return Err(StatementError::UnsupportedFifo { dialect: self.name() });
}
let mut sql = String::from("SELECT ");
match shape {
ClaimShape::Ids => self.quote_into(spec.id().name(), &mut sql),
ClaimShape::Rows if spec.selects_all() => sql.push('*'),
ClaimShape::Rows => {
for (i, column) in spec.columns().enumerate() {
if i > 0 {
sql.push_str(", ");
}
self.quote_into(column.name(), &mut sql);
}
}
ClaimShape::Roles => {
let roles = [
Role::Id,
Role::PartitionKey,
Role::Attempt,
Role::Headers,
Role::Payload,
];
let played = roles
.into_iter()
.filter_map(|role| Some((role, spec.column(role)?)));
for (i, (role, column)) in played.enumerate() {
if i > 0 {
sql.push_str(", ");
}
self.quote_into(column.name(), &mut sql);
sql.push_str(" AS ");
self.quote_into(role.attribute(), &mut sql);
}
}
}
sql.push_str(" FROM ");
self.quote_into(spec.table(), &mut sql);
let mut params = Vec::new();
let mut conditions = Vec::new();
if let Some(group) = spec.column(Role::Group) {
params.push(Param::Group);
conditions.push((group, " = "));
}
if let Some(due) = spec.column(Role::RetryAfter) {
params.push(Param::Now);
conditions.push((due, " <= "));
}
for (i, (column, operator)) in conditions.into_iter().enumerate() {
sql.push_str(if i == 0 { " WHERE " } else { " AND " });
self.quote_into(column.name(), &mut sql);
sql.push_str(operator);
self.placeholder_into(NonZeroUsize::MIN.saturating_add(i), &mut sql);
}
if let Some(processed) = spec.column(Role::ProcessedAt) {
sql.push_str(if params.is_empty() { " WHERE " } else { " AND " });
self.quote_into(processed.name(), &mut sql);
sql.push_str(" IS NULL");
}
sql.push_str(" ORDER BY ");
for role in [Role::Priority, Role::RetryAfter] {
if let Some(column) = spec.column(role) {
self.quote_into(column.name(), &mut sql);
sql.push_str(", ");
}
}
self.quote_into(spec.id().name(), &mut sql);
sql.push_str(" LIMIT ");
self.placeholder_into(NonZeroUsize::MIN.saturating_add(params.len()), &mut sql);
params.push(Param::Limit);
sql.push_str(" FOR UPDATE SKIP LOCKED");
Ok(Statement::new(sql, params))
}
}
#[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 {
// A `SqlxBroker<Postgres, Cockroach>`: the dialect is part of the broker's type.
RustStream::new(AppInfo::new("mailer", "0.1.0"))
.with_broker(SqlxBroker::with_dialect(pool, Cockroach), |b| {
b.include(send);
})
}A dialect builds the statements of a broker’s tables. SqlxBroker::new takes the one built
into the crate for its database, BuiltIn; SqlxBroker::with_dialect takes the service’s
own, for a database without a built-in dialect. The dialect is the broker’s second type
parameter, and each thing its tables can do is a trait it implements:
Dialect: names, placeholders, the statement that opens a transaction at a table’s level, the savepoint a transactional handler’s writes start after, and the statements every form runs to settle a row, move a dead letter, fetch rows and insert one;RowLock: the row lock form and its claim;Lease: the lease form, its claim, the extension of a lease in work and the stamp of a row a claim only selected;Advisory: the advisory lock form, its claim of candidates with their keys, the lock and the release of a key, and the take of a row whose key the session holds;Opens<Level>: an isolation level or SQLite mode a table may declare;ByName<DB>: subscriptions by name, the JSON headers and the times their rows hold.
A table in a form whose trait the dialect lacks does not compile, and neither does a by-name
mount on a dialect without ByName; the error names the trait. A dialect refuses with a
StatementError what it does not build, such as FIFO groups above,
and the subscription of such a table stops at startup. The provided methods of
Dialect and Lease state what their defaults mean for a database
without the feature. The statements are built once, when a subscription opens, so a message
costs the same as through a built-in dialect. A test addresses the broker by its type,
tb.broker::<SqlxBroker<Postgres, Cockroach>>().
One statement of a built-in dialect written the service’s own way is an event of the table instead (see An event of the service’s own).
§Statements checked at compile time
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);
})
}checked puts the statements the derive generates into sqlx’s compile-time check. The derive builds
them with the dialect db names, the code the broker runs when a subscription opens, and wraps each
in sqlx::query! (query_scalar! for a lock’s outcome) inside a function nothing calls. The
service’s build then checks each statement against its database: a column the struct names and the
table lacks fails the build, and cargo sqlx prepare --check fails in CI. The subscription runs the
statements it runs without checked, so a message costs the same.
§What it needs
- The service’s own sqlx with its
macrosfeature and the database’s driver, under the namesqlx:sqlx = { version = "0.9", features = ["macros", "postgres"] }. DATABASE_URLat build time, or the.sqlxdata thatcargo sqlx prepare(sqlx-cli0.9) writes, committed beside the crate.SQLX_OFFLINE=truemakes a build read that data even whereDATABASE_URLis set.- The feature of the dialect
dbnames onruststream-sqlx:postgres,mysqlorsqlite.
Each statement meets sqlx’s own rules. A parameter takes the Rust type sqlx picks for its column,
so with both chrono and time on, sqlx.toml (sqlx’s sqlx-toml feature) names the preferred
crate; a column of a type sqlx does not know takes a type override there.
§What it covers
The statements the struct determines: the claim, and the claim by role a by-name subscription runs
for a struct that writes no event itself; the guard of a FIFO group; the fetch after a claim of the
service’s own; the acknowledgement, the retry, the delayed retry and the discard; the dead letter of
a group; the lease’s extension and stamp; the advisory lock form’s lock, unlock and take; the
generated insert. A statement the service writes itself is its own code, which sqlx::query!
checks the same way.
The startup check still runs: each statement is prepared when its subscription opens, and the server’s version is read where the dialect depends on it. It also holds the statement the struct does not determine: the move into a dead-letter table, which the registration names.
§The headers layout, checked
use chrono::{DateTime, Utc};
use ruststream_sqlx::Fetch;
use ruststream_sqlx::prelude::*;
use sqlx::{Error, FromRow, PgConnection, Postgres};
// order_jobs: job_id BIGSERIAL PRIMARY KEY, name TEXT NOT NULL, processed_at TIMESTAMPTZ,
// trace TEXT, total BIGINT NOT NULL
#[derive(Debug, Clone, InboxHeaders, FromRow)]
#[inbox(table = "order_jobs", checked, db = postgres)]
pub struct OrderHeaders {
#[field(id, generated)]
pub job_id: i64,
#[field(group)]
pub name: String,
#[field(processed_at)]
pub processed_at: Option<DateTime<Utc>>,
pub trace: Option<String>,
}
#[derive(Debug, Clone, Inbox, FromRow)]
#[inbox(custom(fetch))]
pub struct Order {
#[field(headers)]
#[sqlx(flatten)]
pub headers: OrderHeaders,
pub total: i64,
}
impl Fetch<Postgres> for Order {
async fn fetch(conn: &mut PgConnection, ids: &[i64]) -> Result<Vec<Self>, Error> {
let rows = sqlx::query!(
"SELECT job_id, name, processed_at, trace, total FROM order_jobs \
WHERE job_id = ANY($1)",
ids
)
.fetch_all(conn)
.await?;
Ok(rows
.into_iter()
.map(|row| Self {
headers: OrderHeaders {
job_id: row.job_id,
name: row.name,
processed_at: row.processed_at,
trace: row.trace,
},
total: row.total,
})
.collect())
}
}checked goes on the headers struct, which describes the table. Its statements claim ids alone,
so the message lists fetch in custom(..) and reads its rows in a Fetch of the service’s own,
where sqlx::query! checks that statement too. A message that leaves the fetch to the crate does
not build. The headers struct cannot see the message’s custom(..), so it checks the crate’s
statement of every other event, one the message writes itself included.
§What it refuses
A struct that checked cannot cover does not compile, and the error points at the attribute or
the field: checked without db or db without checked; a db that names no built-in dialect,
since a procedural macro cannot run a dialect of the service’s own (its tables keep the startup
check); a db whose feature is off; a #[sqlx(flatten)] field, whose struct’s columns the derive
cannot see; a clock that is a parameter of the struct; the row lock form on SQLite; checked on a
message assembled from a headers struct.
§Startup checks and rows that do not decode
use std::time::Duration;
use chrono::{DateTime, Utc};
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
#[field(id, generated)]
job_id: i64,
#[field(group)]
name: String,
#[field(retry_after, generated)]
retry_after: DateTime<Utc>,
#[field(payload)]
payload: Vec<u8>,
}
#[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 {
// A row that does not decode stays in the table and comes back every ten minutes, until a
// release that reads it ships.
let keep = FailurePolicy::RetryAfter(Duration::from_secs(600));
RustStream::new(AppInfo::new("mailer", "0.1.0")).with_broker(SqlxBroker::new(pool), |b| {
b.include(send.on_failure(FailurePolicies::default().with_decode(keep)));
})
}A mistake stops the service as early as it can be seen. A struct that cannot drive a queue does not
compile, and Inbox lists the errors; macro or manual shows where each rule
of a table described by hand is checked. A subscription builds its statements when it opens and
prepares each one on the server: a table or a column the description names and the database lacks
stops it, and SqlxBrokerError::Schema names the table and the statement. Preparing checks the
names, not the column types: the types are the service’s to get right. A retry declaration the table
cannot carry stops it too (SqlxBrokerError::Declaration): max_attempts(..) and
dead_letter(..) come together, and the cap needs an attempt column. On MySQL and MariaDB the
subscription also reads the server’s version.
A claimed row whose columns do not decode into the struct reaches the subscription’s
on_failure(decode = ..) policy, which settles it, and the subscription goes on with the next
row. The default policy drops the row: it is deleted, or marked where the table has
processed_at. Such a row still reports its attempt: the claim reads the attempt column alone,
as the struct reads it. So max_attempts(..) spends it like any other row, and a policy that
retries it keeps it only up to the cap. When the attempt column itself does not decode, the row
reports no attempt, and a policy that retries it keeps it in the queue until it decodes. The
policy settles such a row before the handler runs, whatever the handler takes, a Deserialized
type included. It settles the same way a claimed id that the service’s own fetch returned no row
for. A row whose id does not decode fails the claim, and the error names the subscription and the
table; every claim that reaches the row fails the same way until the row is fixed.
§Batches
use ruststream_sqlx::prelude::*;
use serde::Deserialize;
use sqlx::PgPool;
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "email_jobs")]
pub struct SendEmail {
#[field(id, generated)]
job_id: i64,
#[field(group)]
name: String,
#[field(payload)]
payload: Vec<u8>,
}
#[derive(Deserialize)]
pub struct Email {
to: String,
}
// One claim is the batch, and each email settles its own row.
#[subscriber(InboxQueue::<SendEmail>::new("emails"))]
async fn send_all(emails: &[Email]) -> Vec<HandlerOutcome> {
let mut outcomes = Vec::with_capacity(emails.len());
for email in emails {
outcomes.push(if deliver(email).await {
HandlerOutcome::ack()
} else {
HandlerOutcome::retry()
});
}
outcomes
}
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!(50)));
})
}A batch is one claim: up to batch(n) rows of the queue. A row that does not decode is settled by
the decode policy, and the batch keeps its other rows. On SQLite the rows of a batch come in no
particular order.
In the row lock form the rows of a batch share the claim’s transaction, which holds one
connection for the whole batch. Each delivery settles its own row in that transaction, and the
settlements take effect together, when the last delivery finishes. A settlement whose statement
fails rolls the whole batch back, its rows return, and the later settlements fail with
SqlxBrokerError::BatchRolledBack. A batch that holds a row whose statement always fails
therefore rolls back and returns all of its rows on every attempt, until the service’s SQL or
schema is fixed.
In the lease form each delivery settles on its own, on a connection it takes for the settlement.
In the advisory lock form each delivery of a batch holds a connection of its own and settles on it, on its own. A batch holds one row per key, and it ends where the pool has no connection to spare.
§Routes and by-name subscriptions
use ruststream::OutgoingMessage;
use ruststream_sqlx::prelude::*;
use serde::{Deserialize, Serialize};
use sqlx::{PgConnection, PgPool, Postgres};
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "jobs")]
pub struct Job {
#[field(id, generated)]
id: i64,
#[field(group)]
name: String,
#[field(payload)]
payload: Vec<u8>,
}
impl Publish<Postgres> for Job {
async fn publish(
conn: &mut PgConnection,
message: &OutgoingMessage<'_>,
) -> Result<(), sqlx::Error> {
sqlx::query("INSERT INTO jobs (name, payload) VALUES ($1, $2)")
.bind(message.name())
.bind(message.payload())
.execute(conn)
.await?;
Ok(())
}
}
#[derive(Deserialize)]
pub struct Signup {
email: String,
}
#[derive(Serialize, Outgoing)]
#[outgoing(name = "welcome")]
pub struct Welcome {
email: String,
}
#[subscriber("signups", reply)]
async fn greet(signup: &Signup) -> Welcome {
Welcome {
email: signup.email.clone(),
}
}
pub fn app(pool: PgPool) -> RustStream {
// Both names lead into `jobs`: `signups` is read by name, and the replies become rows of the
// `welcome` group.
let broker = SqlxBroker::new(pool)
.route::<Job>("signups")
.route::<Job>("welcome");
RustStream::new(AppInfo::new("signup", "0.1.0")).with_broker(broker, |b| {
b.include(greet);
})
}SqlxBroker::route leads a name into the table of a row type, through the row’s Publish:
the service’s own insert, which lays the name, the bytes and the headers out in its columns. A
route serves both directions. A publish to the name writes a row, and #[subscriber("signups")]
opens a subscription by that name on the same table. A name ending in * leads every name that
starts with what precedes it, and an exact name wins over a prefix.
A Repository names its table at compile time, so a publish through it is a static call.
Routed, the broker’s default for replies and Out slots, looks each message’s name up: one
hash lookup for an exact name, the prefixes scanned only when none matches, then one dynamic call
into the row’s Publish. That call’s future stays in a 1024-byte slot, so a routed publish
allocates what a repository publish does; a row whose Publish future is larger is boxed, one
allocation per publish on its route. A name no route leads anywhere fails the publish with
SqlxBrokerError::NoRoute.
A by-name subscription takes the broker’s poll interval and lease. Where the route’s row leaves
every event to the crate and holds column types the crate reads itself, listed on
NamedSubscriber, the subscription reads the rows by those columns: no box and no dynamic call
per message, in every form, as through an InboxQueue. Any other row runs its own code, at one
boxed delivery and one boxed settlement future per message. By-name subscriptions run on a
dialect that implements ByName for its database, as every built-in dialect does, and refuse
max_attempts(..) and dead_letter(..) at startup: an InboxQueue takes those.
§Waking a subscription
use std::time::Duration;
use ruststream::OutgoingMessage;
use ruststream_sqlx::prelude::*;
use serde::{Deserialize, Serialize};
use sqlx::{PgConnection, PgPool, Postgres};
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "jobs")]
pub struct Job {
#[field(id, generated)]
id: i64,
#[field(group)]
name: String,
#[field(payload)]
payload: Vec<u8>,
}
impl Publish<Postgres> for Job {
async fn publish(
conn: &mut PgConnection,
message: &OutgoingMessage<'_>,
) -> Result<(), sqlx::Error> {
sqlx::query("INSERT INTO jobs (name, payload) VALUES ($1, $2)")
.bind(message.name())
.bind(message.payload())
.execute(conn)
.await?;
Ok(())
}
}
#[derive(Deserialize)]
pub struct Signup {
email: String,
}
#[derive(Serialize, Deserialize, Outgoing)]
#[outgoing(name = "welcome")]
pub struct Welcome {
email: String,
}
// Another service writes a signup and announces it: `SELECT pg_notify('jobs', 'signups')`.
#[subscriber(InboxQueue::<Job>::new("signups"), reply)]
async fn greet(signup: &Signup) -> Welcome {
Welcome {
email: signup.email.clone(),
}
}
// The reply's row wakes this subscription in the same process: no poll interval in between.
#[subscriber(InboxQueue::<Job>::new("welcome"))]
async fn send(welcome: &Welcome) -> HandlerOutcome {
tracing::info!(to = %welcome.email, "sending the welcome");
HandlerOutcome::ack()
}
pub fn app(pool: PgPool) -> RustStream {
// The poll each minute reaches a row whose notification was lost.
let broker = SqlxBroker::new(pool)
.poll_interval(Duration::from_secs(60))
.listen_notify();
RustStream::new(AppInfo::new("signup", "0.1.0")).with_broker(broker, |b| {
b.include(greet).out_reply(Repository::<Job>::default());
b.include(send);
})
}A subscription claims rows as long as its claims come back full. After a claim that found fewer
rows than it asked for, it waits the poll interval: one second unless the broker’s
poll_interval or the subscription’s own
poll_interval sets another. Two things end the wait early: a
publish from the same process, and a Postgres notification.
§A publish from the same process
A publish through the broker wakes the subscriptions of its table on the same broker. That covers
a Repository, a route, a reply or an Out slot published through either of them, and a
publish in a test through the harness. In a table with groups, a publish wakes the subscriptions
of its own group alone, and the others keep their interval. In a table without groups it wakes
every subscription of the table. A subscription busy with a claim keeps the wake-up for later, so
its next wait returns at once.
A row the service writes itself wakes nobody: an insert, a statement of the
service’s own, a write through a handler’s Tx. Such a row waits for the poll interval, or for a
notification the service sends.
The wake-up costs a publish one atomic operation per subscription it wakes. A subscription pays nothing per message for it.
§LISTEN/NOTIFY on Postgres
listen_notify turns notifications on for a broker on Postgres. The
broker listens on one connection of the pool, taken at connect and kept until shutdown. Each
subscription listens on its table’s channel before its first claim. The channel is the table’s
name, qualified with its schema where the table has one (app.jobs). The payload is the
group. A notification with an empty payload wakes every subscription of the table.
Each publish of the broker announces its row: SELECT pg_notify(channel, group) runs on the
connection that wrote the row, right after the write. A subscription in another process therefore
claims the row once it is visible. A notification that fails is logged as a warning, and the
publish still succeeds: the row is written, and the poll interval reaches it.
A writer outside RustStream sends the same notification itself, in its transaction or from a trigger:
CREATE FUNCTION announce_job() RETURNS trigger AS $$
BEGIN
PERFORM pg_notify(TG_TABLE_NAME, NEW.name);
RETURN NULL;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER jobs_announced AFTER INSERT ON jobs
FOR EACH ROW EXECUTE FUNCTION announce_job();Postgres delivers a notification once its transaction commits. A notification sent while the listening connection was lost is gone: the broker connects again and wakes every subscription once. The poll interval stays the floor under every wake-up.
Notifications have a price, and the switch is off unless set:
- the broker keeps one connection of the pool for its whole life, so the pool needs one more;
- each publish runs one more statement,
pg_notify, on its connection; - a transaction that notifies takes one lock, global to the server, at commit, so commits of notifying transactions run one after another, which limits many concurrent writers.
Postgres keeps channel names up to 63 bytes. A subscription to a table whose qualified name is
longer stops at startup with SqlxBrokerError::Declaration, which names the table. On another
database listen_notify does not compile: Notifies names the fix.
§Testing a service on the inbox
use std::error::Error;
use ruststream::testing::TestApp;
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use sqlx::{Connection, Sqlite, SqliteConnection};
const SCHEMA: &str = "CREATE TABLE thumbnail_jobs (id INTEGER PRIMARY KEY, \
attempt INTEGER NOT NULL DEFAULT 1, locked_until TEXT, payload BLOB NOT NULL)";
pub async fn a_thumbnail_is_made() -> Result<(), Box<dyn Error + Send + Sync>> {
// An in-memory database that every connection naming it shares, alive while one of them
// is open: the test holds one beside the pool.
let options: SqliteConnectOptions =
"sqlite:file:thumbnails?mode=memory&cache=shared".parse()?;
let keeper = SqliteConnection::connect_with(&options).await?;
let pool = SqlitePoolOptions::new().connect_with(options).await?;
sqlx::raw_sql(SCHEMA).execute(&pool).await?;
let tb = TestApp::start_live(app(pool)).await?;
tb.broker::<SqlxBroker<Sqlite>>()
.message(&Image { path: "cat.png".to_owned() })
.to("thumbnails")
.publish()
.await?;
tb.broker::<SqlxBroker<Sqlite>>()
.subscriber("thumbnails")
.assert_called_once();
tb.shutdown().await?;
keeper.close().await?;
Ok(())
}A test runs the service against the database it runs on, because the service’s SQL is part of
what the test checks. The message goes in through the route and the service’s Publish, as in
production. On Postgres, MySQL and MariaDB a test takes a database of its own on a server, with
the service’s migrations applied. SQLite needs no server: a named in-memory database serves every
connection that names it and lives while one of them is open. The clock is real as well: a paused
tokio clock jumps to the next timer while a database reply is in flight, so a test starts with
TestApp::start_live, and tb.advance(by) lets that much real time pass.
A test build of a service that also tracks its publishes with the outbox leaves the outbox off,
so such a test needs no outbox table. RUSTSTREAM_SQLX_OUTBOX=on in the environment turns it on
for the whole test process.
§The transactional outbox
use std::convert::Infallible;
use std::io;
use chrono::{DateTime, Utc};
use ruststream::OutgoingMessage;
use ruststream::memory::prelude::*;
use ruststream_sqlx::{Outbox, outbox};
use serde::{Deserialize, Serialize};
use sqlx::postgres::{PgConnection, PgPool, Postgres};
// outbox: id BIGSERIAL PRIMARY KEY, name TEXT NOT NULL, payload BYTEA NOT NULL,
// processed_at TIMESTAMPTZ
#[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>>,
}
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,
}
#[subscriber("checkout", reply)]
async fn place(cmd: &PlaceOrder) -> OrderPlaced {
OrderPlaced { id: cmd.id }
}
#[subscriber("orders")]
async fn fulfil(order: &OrderPlaced) -> HandlerOutcome {
println!("fulfilling order {}", order.id);
HandlerOutcome::ack()
}
/// What `#[ruststream::app]` runs.
pub fn app() -> impl App {
let tracking = outbox! {
"orders" => OrderOutbox,
};
let registry = tracking.clone();
RustStream::new(AppInfo::new("orders", "0.1.0"))
.on_startup(async move |()| {
let pool = PgPool::connect("postgres://localhost/orders")
.await
.map_err(io::Error::other)?;
registry.set_pool(pool.clone()).map_err(io::Error::other)?;
Ok::<_, io::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);
b.after_startup(Publish, tracking.republish());
})
}The outbox keeps what a service publishes in a table of its own until a consumer has processed
it. The publish middleware writes a record of each message published under a registered name,
then sends the message with the record’s id in the
x-ruststream-outbox-id header. The subscription middleware takes a
delivery that carries an id into work, runs the handler, and marks the record processed when the
handler acknowledges. At startup the republish sends every unprocessed record again. The outbox
works over any RustStream broker and needs no broker of its own: the example above runs on
MemoryBroker, and examples/outbox in the repository runs the same service over Redis Pub/Sub.
The outbox feature turns it on, beside a driver feature (postgres, mysql, sqlite or
any). It is independent of the inbox; a service may enable both.
§The guarantee
A message is delivered at least once. The record is written before the message is sent. A send that fails, a consumer that stops before its handler finishes, or a handler that asks for the message again leaves the record unprocessed, and the next startup sends it again. The consumer then takes the record by its id, so a copy whose record is processed already is acknowledged without running the handler. Two instances of a service that start together may both send the same record again; a consumer that must not process a message twice checks that itself.
§The record and the registry
#[derive(Outbox)] describes the table: id, name and payload are required
roles, headers and processed_at are optional. The service writes the record itself, in
outbox::Publish, because its statement decides how the name, the payload and the headers are
laid out. The registry runs the other events from statements it builds when the record type is
registered, from the table’s description and the database’s built-in dialect.
The same record may be described by hand, and registered under a name that is a type:
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::outbox::spec::ProcessedAt;
use ruststream_sqlx::outbox::{self, Outbox, OutboxSpec, OutboxTable, TrackedName};
#[derive(sqlx::FromRow)]
struct OrderOutbox {
id: i64,
name: String,
payload: Vec<u8>,
}
impl OutboxTable for OrderOutbox {
type Id = i64;
type Table = OutboxSpec<(ProcessedAt,)>;
const TABLE: Self::Table = OutboxSpec::new(
"outbox",
Column::new("id"),
Column::new("name"),
Column::new("payload"),
)
.processed_at(Column::new("processed_at"));
fn id(&self) -> &i64 {
&self.id
}
fn name(&self) -> &str {
&self.name
}
fn payload(&self) -> &[u8] {
&self.payload
}
}
/// The name `OrderPlaced` is published under.
struct Orders;
impl TrackedName for Orders {
const NAME: &'static str = "orders";
}
pub fn app(pool: PgPool) -> impl App {
let tracking = Outbox::new(pool).track::<OrderOutbox, Orders>();
RustStream::new(AppInfo::new("orders", "0.1.0"))
.layer(tracking.layer())
.publish_layer(tracking.publish_layer())
.with_broker(MemoryBroker::new(), |b| {
b.include(place).out_reply(Publish);
b.include(fulfil);
b.after_startup(Publish, tracking.republish());
})
}A record described by hand implements OutboxTable. OutboxSpec::new takes the table and
its three required columns, and type Table lists the typed settings in the order the chain sets
them: Headers, ProcessedAt and the
events of outbox::spec::own. A record with Headers implements HeaderRow for the field
that holds the column. processed_at takes no time type, because the outbox writes it from the
database’s clock.
outbox! registers record types under the names they track, and a name written twice does not
compile. Outbox::track registers one under a name that is a type, a
TrackedName, and a name tracked twice stops the build. Outbox::register
takes the name as a string, for a name read from configuration, and panics on a repeated name when
the app is built. One record type may track several names; each registration republishes only its
own records.
§The two middlewares
Outbox::layer is mounted with RustStream::layer and Outbox::publish_layer with
RustStream::publish_layer. A message whose name is not registered passes both after a
comparison of names. A handler’s outcome settles a tracked delivery’s record:
| Outcome | Event | Default statement |
|---|---|---|
| acknowledged | outbox::Ack | marks the record processed |
asked again (retry, retry_after) | outbox::Retry | none: the record waits for the next startup |
| dropped | outbox::Discard | marks the record processed |
A record whose fetch finds it processed already acknowledges the delivery without the handler. A fetch that fails retries the delivery without the handler. A settlement that fails logs a warning and keeps the handler’s outcome; the record stays for the next startup.
§The republish
b.after_startup(policy, tracking.republish()) publishes every unprocessed record through the
broker’s publisher once the subscriptions are open, each with its id.
Outbox::republish_names republishes only the names it lists, for a service whose names live on
several brokers. A record the republish cannot read or send fails startup.
§Publishing outside a handler
#[derive(Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders")]
struct OrderPlaced {
id: u64,
}
pub async fn serve(pool: PgPool) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
let broker = MemoryBroker::new().bindable();
let egress = broker.bind(Publish);
let app = RustStream::new(AppInfo::new("orders", "0.1.0"))
.publish_layer(tracking.publish_layer())
.with_broker(broker, |_b| {});
let running = app.start().await?;
// An HTTP endpoint publishes through this one: the record is written before the send.
let publisher = tracking.wrap(running.publisher(egress).await?);
publisher.message(&OrderPlaced { id: 7 }).publish().await?;
running.shutdown().await?;
Ok(())
}The publish pipeline reaches what handlers publish. A publisher from RunningApp::publisher or an
after_startup hook runs outside it, so Outbox::wrap gives it the same tracking.
§The pool
Inside #[ruststream::app] the app is built before the Tokio runtime exists, and sqlx builds a
pool only inside one. There the registry starts without a pool (outbox! without pool:, or
Outbox::deferred), and on_startup builds the pool and hands it over with
Outbox::set_pool, as the first example does. An app built inside a runtime passes the pool
directly: outbox! { pool: pool, .. } or Outbox::new. Every handle the registry gave out
reads the same pool. The service closes the pool in after_shutdown, which runs once the
handlers have finished and the brokers have stopped.
§Events of the service’s own
A record names the events it writes itself: #[outbox(custom(fetch, ack, retry, discard, recover))] on the derive, or .own::<own::Ack>() and the others by hand, each with its marker of
outbox::spec::own in the type. It implements each named event’s trait, and the registry runs
it in place of the default. The example service in examples/outbox takes its records with a
fetch of its own that also stamps taken_at. A record type that names an event and lacks its
trait does not register, and the error names the missing trait.
§Beside the inbox
The outbox’s middlewares wrap the handlers of an inbox subscription as they wrap any other. A
tracked publish takes a connection of its own for the record’s insert, so a record written while a
transactional handler runs commits on its own, whatever the delivery’s
outcome. A message tracked into an inbox table carries its record id where the table keeps
headers: a headers column that the service’s Publish impl writes.
§Testing
A test build of the service (the testing feature) leaves the outbox off: the middlewares pass
every message untouched and the republish sends nothing, so a test needs no outbox table and no
database for it. RUSTSTREAM_SQLX_OUTBOX=on in the environment turns it on for the whole test
process; the variable is read once, the first time the outbox runs. A production build always
tracks, and the switch is compiled out.
§Costs
A message whose name is not registered costs one comparison per registered name and nothing else:
no allocation and no database call. A tracked publish takes a connection for the record’s insert
and allocates the id header’s value. A tracked delivery takes a connection for the fetch, returns
it before the handler runs, so the handler can publish tracked messages of its own, and takes one
again for the mark. Under sqlx’s default test_before_acquire, each of these acquires pings the
connection first, a round trip a pool built with test_before_acquire(false) skips. The default
Retry takes no connection. The republish allocates its body once per startup.
Re-exports§
pub use outbox::Outbox;pub use outbox::OutboxDatabase;pub use outbox::OutboxRow;pub use outbox::OutboxSpec;pub use outbox::OutboxTable;pub use ruststream_sqlx_dialect as dialect;
Modules§
- keys
- What a handler reads off the delivery it handles, through
Ctx<Key>. What a handler reads off the delivery it handles, throughCtx<Key>. - outbox
- The transactional outbox: what a service publishes is recorded in its own table until a consumer has processed it, and what was not processed is published again at startup.
- prelude
- The one glob a service on the inbox imports: the framework’s prelude and the inbox’s own names.
- spec
- The settings of a queue table described by hand, as types.
Macros§
- outbox
- Registers outbox records under the names they track, and returns the registry.
Structs§
- BuiltIn
- The dialect built into the crate for the database
DB: whatSqlxBroker::newbuilds its statements with. - Closed
Sqlx Broker - The broker after
shutdown: what it did to the advisory locks its deliveries held. - Connected
Sqlx Broker - The connected inbox broker: the typed witness that the database answered.
- Database
Clock - The database’s own clock:
#[inbox(clock = DatabaseClock)]makes the statements read it (statement_timestamp()on Postgres,UTC_TIMESTAMP(6)on MySQL) instead of binding the host’s time. - Inbox
Delivery - One claimed row in a handler’s hands.
- Inbox
Queue - A subscription to a queue table: the rows of
Rowthat the name addresses. - Inbox
Spec - The description of a queue table, with its settings as types.
- Inbox
Subscriber - The subscriber an
InboxQueueopens: a stream of deliveries claimed from the table. - Inbox
Threads - The dedicated threads of an
InboxQueuesubscription, and the connections each one opens for its handlers’ queries: whatInboxSettings::on_threadsmounts. - Named
Delivery - A delivery of a by-name subscription: a row of the table its name’s route leads to.
- Named
Subscriber - The subscriber a by-name subscription opens: the claim loop of the table its name’s route leads to.
- Plain
- A subscription whose handler leaves the delivery’s transaction alone: the default mode.
- Repository
- The typed publish policy of a row with
Publish: a message published through it becomes a row ofRow’s table. - Repository
Publisher - The live form of a
Repository: publishes intoRow’s table through itsPublish. - Routed
- The broker’s default publish policy: a publish goes where the route table leads its name.
- Routed
Publisher - The live form of
Routed: publishes into the table the route table leads each name to. - RowBatch
- The rows one claim of a table in row mode took, which a batch handler borrows as one slice,
&[Row], and the claims that settle them. - Sqlx
Broker - The inbox broker: task queues in the service’s own tables, served through a sqlx pool.
- System
Clock - The host’s system clock: what a table reads unless it names another.
- Transactional
- A subscription whose handler writes through the delivery’s transaction, which acknowledgement commits.
- Tx
- The delivery’s transaction, lent to the handler: it dereferences to the connection the transaction runs on, and goes back to the delivery when the handler ends.
Enums§
- Named
Time - A time a by-name subscription binds, in the type its column holds: the current time, a delayed retry’s, or in the lease form a lease’s expiry.
- Sqlx
Broker Error - What can go wrong in the inbox broker.
Traits§
- Ack
- Acknowledges a row: by default it deletes the row, or sets
processed_at. - Attempt
Column - The type of a field playing
attempt: the delivery’s redelivery count. - Attempt
Row - The row’s attempt, which a table described by hand reads off its row where its description
sets
InboxSpec::attempt. - Built
InDialect - A database whose dialect is built into this crate, so
SqlxBroker::newneeds no dialect of the service’s own. - ByName
- Subscriptions by name on the database
DB, as a feature of a dialect: the headers and the times their rows hold, which no struct of the service’s own names. - Claim
- Claims up to
limitrows of the queuequeueand returns their ids, inside the claim’s transaction. - Clock
- A clock on the host: what “now” is for the statements that bind it.
- Dead
Letter - Moves a row whose attempts are spent to
destination: by default it moves the row to that group (on a table with agroupcolumn) or into that table. - Discard
- Drops a row: by default it deletes the row, or sets
processed_at. - Extend
- Extends the lease on a row: writes
untilintolocked_untilwhile the row still holdsheld, and says whether it did. - Fetch
- Reads the rows of claimed ids, on the claim’s connection.
- Header
Column - The type of a field playing
headers: what the delivery’s header map holds. - Header
Field - A value a field of a headers struct holds, as the header the delivery’s header map carries.
- Header
Fields - Header fields of the row: where its description sets
InboxSpec::header_fields, the delivery builds its header map from them on its first read. - Header
Row - A record or a row whose headers live in one column: the field holding the column, which the
table’s description names with its
headerssetter. - Inbox
Headers - A struct that describes a queue table whose message is assembled from it: the mechanics a
subscription runs the queue by, and the service’s own headers.
#[derive(InboxHeaders)]implements it. - Inbox
Row - A struct that describes a queue table;
#[derive(Inbox)]implements it. - Inbox
Settings - The inbox’s own mount-site steps, on a
#[subscriber]definition and on its settings builder. - Inbox
Table - A queue table described by hand.
- Insert
- Inserts the row: every field except the
generatedones, the queue’s mechanics included, so a delayed task is a row whoseretry_afterlies ahead. - KeyColumn
- The type of a field playing
partition_key: the bytes the delivery’s key lends. - KeyRow
- The row’s partition key, which a table described by hand reads off its row where its
description sets
InboxSpec::partition_key. - Lease
Row - A queue row taken by lease: its struct has a
#[field(locked_until)]field. - Lock
- Takes the advisory lock on a row’s key for the session that will hold its delivery, without
waiting: the
lockevent of the advisory lock form. - Notifies
- A database whose sessions notify each other: a broker on it may
listen_notify. - Payload
Row - A queue row that carries its message as bytes: payload mode.
- Publish
- Writes a published message into the table: the name, the bytes and the headers reach the service’s SQL, which lays them out in its columns.
- Queue
Database - A sqlx database the inbox runs on: one that binds and reads the values the queue’s own statements use, runs statements on a connection, and reports the rows a statement changed.
- Queue
Time - A time a queue column holds:
retry_after,processed_atorlocked_until. - Retry
- Releases a row for another attempt at once: by default it counts the attempt, or leaves the row for the rollback to release; in the lease form it clears the lease.
- Retry
After - Releases a row for another attempt after
delay: by default it setsretry_after. - Time
Column - The type of a field playing a time role: a
QueueTimeor anOptionof one. - Time
Source - Where a table reads “now”: a
Clockon the host, orDatabaseClock. - Unlock
- Releases the advisory lock on a delivery’s key: the
unlockevent of the advisory lock form, which a table names as its own besideLock.
Functions§
- put_
header - Puts
fieldintoheadersundername, the way a field of a struct derivingInboxHeadersbecomes a header; a field without a value leaves the header out.
Derive Macros§
- Inbox
- Describes a queue table with a struct: writes its
InboxTableimpl, the description a table described by hand writes, and the traits its roles need. - Inbox
Headers - Describes the queue table of a message assembled from it, and implements
InboxHeadersfor the struct. - Outbox
- Describes a service’s outbox table with a struct: implements
OutboxTable, the description a record by hand writes, andHeaderRowfor aheadersfield.