Forms

A table takes rows by row lock unless it names another form: FOR UPDATE SKIP LOCKED in a transaction held while the handler runs.

A table takes its rows in one of three forms, and its description decides which:

Form	Selected by	What holds a row in work	After a crash	Costs
row lock	nothing: the default	the claim's transaction	the row returns at once	a connection per message in work
lease	`#[field(locked_until)]`	an expiry written into the row	the row returns when the lease runs out	statements to claim, extend and settle
advisory lock	`advisory_lock = ".."`	a lock on the row's key	the row returns at once	a connection per message in work

A row lock claim locks each row with FOR UPDATE SKIP LOCKED in a transaction that stays open while the handler runs. Rows that another claim holds are passed over, so claims of several workers and processes never wait for each other. The settlement's statement, a delete or the processed_at mark, commits that transaction, so a row and its outcome change together. 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 (Connections). Postgres takes the form from 9.5, MySQL from 8.0.1 and MariaDB from 10.6; SQLite has no row locks, so a subscription to a SQLite table that names no other form does not compile. On Postgres read_committed is the practical isolation level for the form: a claim at a stricter level fails with a serialization error when claims and settlements of its rows run at once. By hand, a description that calls neither .lease(..) nor .advisory(..) takes the row lock form.

Example

Four workers issue invoices on Postgres; each holds its row's lock and a connection until its acknowledgement commits.

Macros
Manual

examples/sqlx_forms_row_lock.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::postgres::PgPoolOptions;

/// One row of `invoice_jobs`. No `locked_until` and no `advisory_lock`: the row lock form.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "invoice_jobs")]
struct IssueInvoice {
    #[field(id, generated)]
    id: i64,
    #[field(payload)]
    payload: Vec<u8>,
}

/// The claim's transaction holds the row while the handler runs; the acknowledgement deletes
/// the row and commits. A crash rolls the transaction back, and the row returns at once.
#[subscriber(InboxQueue::<IssueInvoice>::new("invoices"))]
async fn issue(invoice: &Invoice) -> HandlerOutcome {
    tracing::info!(invoice.number, "issuing an invoice");
    HandlerOutcome::ack()
}

/// The service on the database `options` reach.
fn service(options: PgConnectOptions) -> RustStream {
    let schema = options.clone();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_connect_options(options, PgPoolOptions::new());
    RustStream::new(AppInfo::new("billing", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = PgConnection::connect_with(&schema).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<IssueInvoice>("invoices"), |b| {
            // Each of the four workers holds a connection while its invoice is in work.
            b.include(issue.workers(nonzero!(4)));
        })
}

examples/manual/sqlx_forms_row_lock.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::postgres::PgPoolOptions;

/// One row of `invoice_jobs`. No `.lease(..)` and no `.advisory(..)`: the row lock form.
#[derive(sqlx::FromRow)]
struct IssueInvoice {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for IssueInvoice {
    type Id = i64;
    type Table = InboxSpec<(Payload,)>;
    const TABLE: Self::Table = InboxSpec::new("invoice_jobs", Column::new("id").generated())
        .payload(Column::new("payload"));

    fn id(&self) -> &i64 {
        &self.id
    }
}

impl PayloadRow for IssueInvoice {
    type Column = Vec<u8>;

    fn payload(&self) -> &[u8] {
        &self.payload
    }
}

/// The claim's transaction holds the row while the handler runs; the acknowledgement deletes
/// the row and commits. A crash rolls the transaction back, and the row returns at once.
struct Issue;

impl Handle<Invoice> for Issue {
    fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(invoice.number, "issuing an invoice");
        ready(Ok(()))
    }
}

/// The service on the database `options` reach.
fn service(options: PgConnectOptions) -> RustStream {
    let schema = options.clone();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_connect_options(options, PgPoolOptions::new());
    RustStream::new(AppInfo::new("billing", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = PgConnection::connect_with(&schema).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<IssueInvoice>("invoices"), |b| {
            // Each of the four workers holds a connection while its invoice is in work.
            let issue = subscriber("invoices", Issue)
                .map_source(|_| InboxQueue::<IssueInvoice>::new("invoices"))
                .workers(nonzero!(4));
            b.include(issue.build());
        })
}

Example​

Example