Isolation

isolation = <level> opens a table's transactions at an isolation level, mode = <mode> at a SQLite mode; a level the dialect does not open does not compile.

isolation in #[inbox(..)] takes read_uncommitted, read_committed, repeatable_read or serializable. A SQLite table declares mode instead: deferred, immediate or exclusive. SQLite runs every transaction serializable, and its mode decides when a transaction takes the write lock (Mode). A table declares one of the two at most. The declaration governs the transactions the crate opens for a delivery's work: in the row lock form the claim's transaction, and in transactional mode the one 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 opens the levels it keeps:

Database	Opens
Postgres	read_committed, repeatable_read, serializable; it runs READ UNCOMMITTED as READ COMMITTED, so its dialect does not open it
MySQL and MariaDB	all four levels
SQLite	the three modes

A subscription to a table at a level 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. An AnyPool names its database only when the broker connects, so there such a subscription stops at startup with SqlxBrokerError::Dialect. 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, so read_committed is the practical level there. A lease table on Postgres at a strict level refuses transactional mode at startup: its transaction would not see the lease extended after its first statement. On MySQL and MariaDB a row lock claim at repeatable_read or serializable locks every row it reads until its settlement; an index on the group's column keeps a claim to its own group. By hand, .opens::<Level>() takes a marker of dialect::level, and Opens<Level> goes into the description's type.

Example

A withdrawal reads the balance and writes it back in one transaction; mode = immediate takes the write lock when the transaction begins, so no other writer changes the balance between the read and the write.

Macros

examples/sqlx_transactions_isolation.rs

use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::Sqlite;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `withdrawal_jobs`; its transactions open with `BEGIN IMMEDIATE`.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "withdrawal_jobs", mode = immediate)]
struct Withdraw {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

#[subscriber(InboxQueue::<Withdraw>::new("withdrawals"))]
async fn withdraw(order: &Withdrawal, Ctx(mut tx): Ctx<keys::Tx<Sqlite>>) -> HandlerOutcome {
    let balance: Result<i64, _> =
        sqlx::query_scalar("SELECT cents FROM balances WHERE account = ?")
            .bind(&order.account)
            .fetch_one(&mut *tx)
            .await;
    let Ok(balance) = balance else {
        return HandlerOutcome::retry();
    };
    if balance < order.cents {
        tracing::warn!(account = %order.account, "insufficient funds");
        return HandlerOutcome::drop();
    }
    let paid = sqlx::query("UPDATE balances SET cents = ? WHERE account = ?")
        .bind(balance - order.cents)
        .bind(&order.account)
        .execute(&mut *tx)
        .await;
    if paid.is_ok() {
        HandlerOutcome::ack()
    } else {
        HandlerOutcome::retry()
    }
}

/// 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("wallet", "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::<Withdraw>("withdrawals"), |b| {
            b.include(withdraw.transactional());
        })
}

Manual

examples/manual/sqlx_transactions_isolation.rs

use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::Sqlite;
use sqlx::sqlite::SqlitePoolOptions;

/// One row of `withdrawal_jobs`; its transactions open with `BEGIN IMMEDIATE`.
#[derive(sqlx::FromRow)]
struct Withdraw {
    id: i64,
    payload: Vec<u8>,
}

impl InboxTable for Withdraw {
    type Id = i64;
    type Table = InboxSpec<(Opens<level::Immediate>, Lease<DateTime<Utc>>, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("withdrawal_jobs", Column::new("id").generated())
        .opens::<level::Immediate>()
        .lease(Column::new("locked_until"))
        .payload(Column::new("payload"));

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

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

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

struct WithdrawHandler;

impl Handle<Withdrawal, (), (), TxContext<Sqlite>> for WithdrawHandler {
    async fn handle(
        &self,
        order: &Withdrawal,
        _outs: &(),
        ctx: &mut Context<'_, TxContext<Sqlite>>,
    ) -> Result<(), HandlerOutcome> {
        let Ctx(mut tx) = Ctx::<keys::Tx<Sqlite>>::from_context(ctx).await?;
        let balance: i64 = sqlx::query_scalar("SELECT cents FROM balances WHERE account = ?")
            .bind(&order.account)
            .fetch_one(&mut *tx)
            .await
            .map_err(|_| HandlerOutcome::retry())?;
        if balance < order.cents {
            tracing::warn!(account = %order.account, "insufficient funds");
            return Err(HandlerOutcome::drop());
        }
        sqlx::query("UPDATE balances SET cents = ? WHERE account = ?")
            .bind(balance - order.cents)
            .bind(&order.account)
            .execute(&mut *tx)
            .await
            .map_err(|_| HandlerOutcome::retry())?;
        Ok(())
    }
}

/// 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("wallet", "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::<Withdraw>("withdrawals"), |b| {
            let withdraw = subscriber("withdrawals", WithdrawHandler)
                .map_source(|_| InboxQueue::<Withdraw>::new("withdrawals"))
                .transactional();
            b.include(withdraw.build());
        })
}