Skip to main content

Exactly once inside the database: transactional handlers

This guide commits a handler's writes together with its job's settlement, so a job whose effects are rows in the same database takes effect exactly once.

A queue delivers each job at least once. A handler that writes its result and then crashes before the job settles runs again, and writes the result twice. When the result is rows in the database that holds the queue, one transaction removes that duplicate: the writes and the settlement commit together, or neither does.

The example is a refunds service. A refund writes its record and a receipt job for the mailer, and calls the card processor.

The tables​

src/main.rs
const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS refunds (order_id INTEGER PRIMARY KEY, amount_cents INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS refund_jobs (
    id INTEGER PRIMARY KEY,
    attempt INTEGER NOT NULL DEFAULT 1,
    locked_until TEXT,
    payload BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS receipt_jobs (
    id INTEGER PRIMARY KEY,
    locked_until TEXT,
    payload BLOB NOT NULL
);
";
src/main.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Sqlite, SqliteConnection};

/// One row of `refund_jobs`, the refunds the shop's support queues.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "refund_jobs")]
struct IssueRefund {
    #[field(id, generated)]
    id: i64,
    #[field(attempt, generated)]
    attempt: i16,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// One row of `receipt_jobs`, the mailer's queue.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "receipt_jobs")]
struct SendReceipt {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

refund_jobs is the queue this service reads. receipt_jobs is the mailer's queue: this service only writes into it, with the derive's insert; in the manual form, with a statement of its own.

The handler​

src/main.rs
/// The refund's record and its receipt, written on the connection of the delivery's transaction.
async fn record(
    conn: &mut SqliteConnection,
    refund: &Refund,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    sqlx::query("INSERT INTO refunds (order_id, amount_cents) VALUES (?, ?)")
        .bind(refund.order)
        .bind(refund.amount_cents)
        .execute(&mut *conn)
        .await?;
    let receipt = SendReceipt {
        id: 0,
        locked_until: None,
        payload: serde_json::to_vec(&Receipt {
            order: refund.order,
        })?,
    };
    receipt.insert(conn).await?;
    Ok(())
}

#[subscriber(InboxQueue::<IssueRefund>::new("refunds"))]
async fn issue(
    refund: &Refund,
    Ctx(mut tx): Ctx<keys::Tx<Sqlite>>,
    Ctx(attempt): Ctx<keys::Attempt>,
) -> HandlerOutcome {
    if let Err(error) = record(&mut tx, refund).await {
        tracing::warn!(subscription = "refunds", refund.order, %error, "refund not recorded");
        return HandlerOutcome::drop();
    }
    // The processor takes the order as its idempotency key: a call repeated after a retry
    // refunds the card once.
    if !card_refund(refund, attempt).await {
        // The retry rolls the record and the receipt back first.
        return HandlerOutcome::retry();
    }
    // The commit keeps the record and the receipt, and finishes the refund's row.
    HandlerOutcome::ack()
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    RustStream::new(AppInfo::new("refunds", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<IssueRefund>("refunds"), |b| {
            b.include(issue.transactional());
        })
}

.transactional() at the mount lends the handler its delivery's transaction. The handler takes it as its first Ctx parameter, Ctx<keys::Tx<Sqlite>>, and writes through &mut *tx. A handler that takes keys::Tx on a subscription without the step does not compile.

The acknowledgement commits the record, the receipt job and the refund row's settlement in one transaction. Every other outcome rolls the writes back first, then settles the row; a retry still counts the attempt. A panic commits nothing. A second refund of the same order fails on the primary key of refunds and is dropped, so a duplicate that reaches the queue changes nothing.

The receipt job is a chain: the mailer sees it only once the refund committed, and never for a refund that rolled back. A write through Ctx<keys::Pool> commits on its own, whatever the outcome, and so does a publish through the broker; neither joins the transaction.

What stays outside​

The card processor is outside the database, so no transaction covers it. A retry after the processor refunded the card calls it again, so the call carries an idempotency key, the order here, and the processor refunds once. Every effect outside the database stays at least once in the same way: an email, an HTTP call, a publish to another broker. Relaying events to another broker and the outbox deal with publishes.

On SQLite a delivery's transaction takes the database's write lock at its first write and holds it until the delivery settles, keeping every other writer out, claims included. Here the processor's call runs after the two inserts, so the database stays locked for other writers while it runs. A server locks only the rows the transaction wrote. A handler that publishes a message the outbox tracks publishes it before its first write through tx on SQLite: the record is written on a connection of its own, which needs the delivery's write lock. A tracked publish after that write fails with OutboxError::WriteLockHeld, and the message is not sent.

Isolation​

Each delivery's transaction opens at the table's isolation level, #[inbox(isolation = read_committed)] on a server, or at its SQLite mode, #[inbox(mode = immediate)]. A table that declares neither takes the database's default, and READ COMMITTED on MySQL and MariaDB. Each delivery in work holds a connection for its transaction, so a handler that takes a second one, through keys::Pool or a publish, needs a pool larger than its workers(n). Transactional mode serves single deliveries; a batch handler mounted with it does not compile. On MySQL and MariaDB, a handler that cuts a statement through tx short, as a select! around a query does, can confuse the settlement that follows, a bug of sqlx-mysql 0.9.0: let each statement finish.

Testing​

The tests in the example file run the service with TestApp::start on SQLite in memory and read the tables after each publish. A refund commits one record and one receipt. A refund whose first call to the processor times out is retried, and the second attempt leaves one record and one receipt, not two. The same refund published twice is dropped the second time.