Transactional

.transactional() lends the handler its delivery's transaction: acknowledgement commits the handler's writes with the row's settlement, or neither.

The handler takes the transaction as its first Ctx parameter, Ctx(mut tx): Ctx<keys::Tx<DB>>, 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 retry still counts the attempt. 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 keys::Pool commits on its own, whatever the outcome, and so does a publish through the broker. What the transaction is depends on the form:

Form	The handler's transaction
row lock	the claim's transaction, with a savepoint right after the claim that a settlement other than acknowledgement rolls back to
lease	a transaction of the delivery's own, opened once the claim committed; a lost lease rolls it back and the settlement fails with LeaseLost
advisory lock	a transaction on the session that holds the row's key, ended before the key's release

Each of these transactions opens at the table's isolation level or SQLite mode. 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. 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) (Connections). 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, fails with TransactionHeld.

Example

A signup opens its account and queues its welcome email through the delivery's transaction; when the CRM times out, the retry discards both, and the next attempt commits them.

Macros

examples/sqlx_transactions.rs

use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Sqlite, SqliteConnection};

/// One row of `signup_jobs`.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "signup_jobs")]
struct OpenAccount {
    #[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 `welcome_jobs`, the mailer's queue.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "welcome_jobs")]
struct SendWelcome {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// The account and its welcome email, written on the connection of the delivery's transaction.
async fn open(
    conn: &mut SqliteConnection,
    signup: &Signup,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    sqlx::query("INSERT INTO accounts (email) VALUES (?)")
        .bind(&signup.email)
        .execute(&mut *conn)
        .await?;
    let welcome = SendWelcome {
        id: 0,
        locked_until: None,
        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<Sqlite>>,
    Ctx(attempt): Ctx<keys::Attempt>,
) -> HandlerOutcome {
    if open(&mut tx, signup).await.is_err() {
        return HandlerOutcome::retry();
    }
    if !register(signup, attempt).await {
        // The retry rolls the account and its welcome email back first.
        return HandlerOutcome::retry();
    }
    // The commit keeps the account and its welcome email, and finishes the signup.
    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("accounts", "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::<OpenAccount>("signups"), |b| {
            b.include(open_account.transactional());
        })
}

Manual

examples/manual/sqlx_transactions.rs

use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Sqlite, SqliteConnection};

/// One row of `signup_jobs`.
#[derive(sqlx::FromRow)]
struct OpenAccount {
    id: i64,
    attempt: i16,
    payload: Vec<u8>,
}

impl InboxTable for OpenAccount {
    type Id = i64;
    type Table = InboxSpec<(Attempt, Lease<DateTime<Utc>>, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("signup_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 OpenAccount {
    type Attempt = i16;

    fn attempt(&self) -> &i16 {
        &self.attempt
    }
}

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

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

/// The account and its welcome email, written on the connection of the delivery's transaction.
async fn open(
    conn: &mut SqliteConnection,
    signup: &Signup,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    sqlx::query("INSERT INTO accounts (email) VALUES (?)")
        .bind(&signup.email)
        .execute(&mut *conn)
        .await?;
    sqlx::query("INSERT INTO welcome_jobs (payload) VALUES (?)")
        .bind(serde_json::to_vec(&Welcome { to: &signup.email })?)
        .execute(conn)
        .await?;
    Ok(())
}

struct OpenAccountHandler;

impl Handle<Signup, (), (), TxContext<Sqlite>> for OpenAccountHandler {
    async fn handle(
        &self,
        signup: &Signup,
        _outs: &(),
        ctx: &mut Context<'_, TxContext<Sqlite>>,
    ) -> Result<(), HandlerOutcome> {
        let Ctx(mut tx) = Ctx::<keys::Tx<Sqlite>>::from_context(ctx).await?;
        let Ctx(attempt) = Ctx::<keys::Attempt>::from_context(ctx).await?;
        if open(&mut tx, signup).await.is_err() {
            return Err(HandlerOutcome::retry());
        }
        if !register(signup, attempt).await {
            // The retry rolls the account and its welcome email back first.
            return Err(HandlerOutcome::retry());
        }
        // The commit keeps the account and its welcome email, and finishes the signup.
        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("accounts", "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::<OpenAccount>("signups"), |b| {
            let open_account = subscriber("signups", OpenAccountHandler)
                .map_source(|_| InboxQueue::<OpenAccount>::new("signups"))
                .transactional();
            b.include(open_account.build());
        })
}