Transactions

A query run through a Db handle runs on its own, in autocommit mode. When several statements must succeed or fail together, write a transaction flow: a type that implements TxFlow. A flow is a state machine whose steps run queries on an open transaction, a Tx, and decide what comes next from their results.

wakuwaku owns the protocol around the steps: it sends BEGIN, COMMIT and ROLLBACK, knows which errors ended the transaction on the server, refuses to commit an aborted transaction, and reports a COMMIT whose answer was lost. Your flow holds only the business logic.

This page assumes you know how to write a query for sqlx; see sqlx (PostgreSQL and MySQL).

The modelCopy Anchor Link

The state is the flow value itself, usually an enum: a working state for each point the transaction can be at, and a terminal state. Two functions define the machine:

• async fn step(tx: Tx<'_, Self>, state: Self) -> TxState<'_, Self>: run the queries of state on tx and return the next state. It is never called with a terminal state.
• fn finished(self) -> Either<Self, TxEnd<Self>>: Right with how the transaction ends if the state is terminal, Left(self) otherwise.

The terminal states are the TxEnds:

A flow turns a TxEnd into a state with From<TxEnd<Self>>, typically an End(TxEnd<Self>) variant, and finished must give it back: finished(end.into()) is Right(end).

Inside a step, the Tx is moved in and handed back:

• tx.run(query).await runs a query and returns (tx, result), where result is the query's own Result<Output, Error>.
• tx.with(state) goes on with state. Returning a terminal state this way is fine: the driver ends it before it would step it.
• tx.end(state).await ends the transaction now if state is terminal (otherwise it behaves like with).

Both return a TxState, the state together with its transaction. A step can only produce a TxState from the Tx it was given, so the transaction never escapes the step.

The associated items say what the flow talks about. A flow is not a query and doesn't implement Query: what it may touch is the Capability it declares with tx_effect! (see Capabilities and read-only transactions).

A complete flow: moving moneyCopy Anchor Link

The example moves amount from one account to another: debit, then credit, then commit. A debit that matches no row means not enough money; a credit that matches no row means an unknown account. Both roll back.

First the queries. Each is an ordinary standalone query with its own Output and Error; the flow runs them unchanged:

use wakuwaku::deps::sqlx::{self, PgConnection, Postgres};
use wakuwaku::entity::sqlx::backend::SqlxSource;
use wakuwaku::entity::sqlx::transaction::{Either, IsolationLevel, Tx, TxEnd, TxFlow, TxState};
use wakuwaku::entity::{Entity, Execute, Query, Read, Write, tx_effect};

pub struct Account;
impl Entity for Account {}

/// Takes `amount` from `from` if the balance allows it. Rows affected.
pub struct Debit {
    pub from: i64,
    pub amount: i64,
}

impl Query for Debit {
    type Effect = (Read<Account>, Write<Account>);
}

impl Execute<SqlxSource<Postgres>> for Debit {
    type Output = u64;
    type Error = sqlx::Error;

    async fn execute(self, conn: &mut PgConnection) -> Result<u64, sqlx::Error> {
        let done = sqlx::query(
            "UPDATE account SET balance = balance - $2 WHERE id = $1 AND balance >= $2",
        )
        .bind(self.from)
        .bind(self.amount)
        .execute(conn)
        .await?;
        Ok(done.rows_affected())
    }
}

/// Adds `amount` to `to`. Rows affected.
pub struct Credit {
    pub to: i64,
    pub amount: i64,
}

impl Query for Credit {
    type Effect = Write<Account>;
}

impl Execute<SqlxSource<Postgres>> for Credit {
    type Output = u64;
    type Error = sqlx::Error;

    async fn execute(self, conn: &mut PgConnection) -> Result<u64, sqlx::Error> {
        let done = sqlx::query("UPDATE account SET balance = balance + $2 WHERE id = $1")
            .bind(self.to)
            .bind(self.amount)
            .execute(conn)
            .await?;
        Ok(done.rows_affected())
    }
}

Then the flow. A transfer is a fixed sequence of statements followed by a commit, so it has a single working state, Start, and its one step runs the debit and the credit in order. TransferEffect declares the one entity both queries touch. The decisions are plain functions of a query result (check_debit, check_credit), and step only runs the queries and acts on the answers:

/// Published once the money has moved.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Transferred {
    pub from: i64,
    pub to: i64,
    pub amount: i64,
}

/// Why the transfer was rolled back.
#[derive(Debug)]
pub enum TransferFailed {
    InsufficientFunds,
    UnknownAccount,
    /// Serialization failure or deadlock: safe to run the transfer again.
    Conflict,
    Database(sqlx::Error),
}

impl TransferFailed {
    fn from_sqlx(error: sqlx::Error) -> Self {
        if is_conflict(&error) {
            TransferFailed::Conflict
        } else {
            TransferFailed::Database(error)
        }
    }
}

tx_effect! {
    /// Everything a transfer touches.
    pub enum TransferEffect { write Account }
}

/// The state of a transfer.
#[derive(Debug)]
pub enum Transfer {
    Start { from: i64, to: i64, amount: i64 },
    End(TxEnd<Transfer>),
}

impl Transfer {
    pub fn new(from: i64, to: i64, amount: i64) -> Self {
        Transfer::Start { from, to, amount }
    }

    fn commit(from: i64, to: i64, amount: i64) -> Self {
        Transfer::End(TxEnd::Commit {
            output: (),
            report: Some(Transferred { from, to, amount }),
        })
    }

    fn rollback(why: TransferFailed) -> Self {
        Transfer::End(TxEnd::Rollback { output: why, report: None })
    }

    /// Whether the debit took the money.
    fn check_debit(debited: Result<u64, sqlx::Error>) -> Result<(), TransferFailed> {
        match debited {
            Ok(1) => Ok(()),
            Ok(_) => Err(TransferFailed::InsufficientFunds),
            Err(error) => Err(TransferFailed::from_sqlx(error)),
        }
    }

    /// Whether the credit reached an account.
    fn check_credit(credited: Result<u64, sqlx::Error>) -> Result<(), TransferFailed> {
        match credited {
            Ok(1) => Ok(()),
            Ok(_) => Err(TransferFailed::UnknownAccount),
            Err(error) => Err(TransferFailed::from_sqlx(error)),
        }
    }
}

impl From<TxEnd<Transfer>> for Transfer {
    fn from(end: TxEnd<Transfer>) -> Self {
        Transfer::End(end)
    }
}

impl TxFlow for Transfer {
    type Capability = TransferEffect;
    type Database = Postgres;
    type Commit = ();
    type Abort = TransferFailed;
    type Report = Option<Transferred>;

    const ISOLATION: Option<IsolationLevel> = Some(IsolationLevel::Serializable);

    async fn step(tx: Tx<'_, Self>, state: Self) -> TxState<'_, Self> {
        let Transfer::Start { from, to, amount } = state else {
            // Never stepped: the driver ends a terminal state first.
            return tx.end(state).await;
        };
        let (tx, debited) = tx.run(Debit { from, amount }).await;
        if let Err(why) = Self::check_debit(debited) {
            return tx.end(Self::rollback(why)).await;
        }
        let (tx, credited) = tx.run(Credit { to, amount }).await;
        match Self::check_credit(credited) {
            Ok(()) => tx.end(Self::commit(from, to, amount)).await,
            Err(why) => tx.end(Self::rollback(why)).await,
        }
    }

    fn finished(self) -> Either<Self, TxEnd<Self>> {
        match self {
            Transfer::End(end) => Either::Right(end),
            state => Either::Left(state),
        }
    }
}

For MySQL, the flow is identical except type Database = MySql;. Each query result keeps its own type, and the code using it follows the query that produced it: no enum of queries, no enum of rows, no catch-all arm for a result that cannot happen.

TxEnd derives Debug, so a flow enum that embeds it can derive Debug too, as long as its Commit, Abort and Report types are Debug. It is deliberately not Clone: a flow value is one run of the transaction, and its terminal state is consumed once when the transaction ends, the way an awaited future is not awaited again. See Retries for what that means in practice.

The is_conflict helper recognizes the errors that mean "another transaction got in the way, run me again":

/// SQLSTATE 40001 (serialization failure) or 40P01 (deadlock detected).
pub fn is_conflict(error: &sqlx::Error) -> bool {
    error
        .as_database_error()
        .and_then(|error| error.code())
        .is_some_and(|code| code == "40001" || code == "40P01")
}

Running itCopy Anchor Link

Wrap the first state in Transaction and process it with a Db whose capability covers what the flow declares. What each outcome means is the caller's business, so map the result into the caller's own error where the transaction runs:

use crate::transfer::{Transfer, TransferFailed};
use wakuwaku::deps::kanau::processor::Processor;
use wakuwaku::deps::sqlx;
use wakuwaku::entity::sqlx::backend::PgSource;
use wakuwaku::entity::sqlx::transaction::{Transaction, TxError, TxOutput};
use wakuwaku::entity::{Db, ReadWrite};

/// Why a payment did not go through.
#[derive(Debug)]
pub enum PayError {
    /// The transfer rolled back, and this is why.
    Refused(TransferFailed),
    /// `COMMIT` was sent but its answer was lost: the money may have moved.
    OutcomeUnknown(sqlx::Error),
    /// The transaction did not end as the flow planned.
    Transaction(TxError),
}

pub async fn pay(
    db: &Db<PgSource, ReadWrite>,
    from: i64,
    to: i64,
    amount: i64,
) -> Result<(), PayError> {
    match db.process(Transaction(Transfer::new(from, to, amount))).await {
        Ok(TxOutput::Committed(())) => Ok(()),
        Ok(TxOutput::RolledBack(why)) => Err(PayError::Refused(why)),
        Err(TxError::CommitUnknown(error)) => Err(PayError::OutcomeUnknown(error)),
        Err(error) => Err(PayError::Transaction(error)),
    }
}

process returns Result<TxOutput<Commit, Abort>, TxError>:

A flow is not a queryCopy Anchor Link

Run a flow as db.process(Transaction(flow)) and handle its outcome there, as pay does. Don't hide it behind a Processor<Input> implementation for the Db, so that db.process(CreateServer { … }) runs a whole transaction and answers with one Output and one Error as if it were a single query. That shape loses three things:

• TxError::CommitUnknown is folded into an ordinary database error. A caller that retries failed statements then retries a transaction that may already have committed.
• To fit one success type, a rollback the caller should treat as success has to travel in Abort, which becomes a second success channel (see What Abort is for).
• A call site can no longer tell a transaction from a single statement, though the two fail differently.

The business function is the boundary: it runs the transaction and maps Committed, RolledBack and TxError into its own error, with CommitUnknown as a case of its own. Whoever gets PayError::OutcomeUnknown must reconcile before reporting a failure or trying again; see Retries.

Designing a flowCopy Anchor Link

States are for loops and branchesCopy Anchor Link

A state is a point the transaction can come back to: the top of a loop, or a branch that continues in a later step. It is not a slot per statement sent. A straight-line transaction, a fixed sequence of statements followed by a commit, is a single step: it runs its queries in order with Tx::run, ends with a rollback as soon as one answer rules out the rest, and otherwise finishes with tx.end(...), as the transfer does. Splitting it into one state per query (Start, Debited, Credited, …) only carries the same data from variant to variant.

Use several states when the shape of the transaction depends on what it reads:

• A loop whose length is known only at run time. PublishAll in Delivering changes from transactions steps once per post, and its state carries the posts still pending and those already published.
• A branch that comes back, such as retry-with-read: a conditional write that lost a race goes to a state that re-reads the row and tries again.

What Abort is forCopy Anchor Link

Abort is why the flow rolled back: a domain reason, such as TransferFailed::InsufficientFunds, or the statement that failed, such as TransferFailed::Database. It is never a second way to succeed. Don't make it Result<Commit, _> so that a rolled-back transaction can still hand the caller a result: if the flow got what it came for, it commits.

That includes a write that changed nothing. A conditional write that matched no row wrote nothing, so when "nothing matched" is an acceptable answer (the row was already in the wanted state, say), commit that empty result, for example as a Commit of Option<T> or of the rows affected. Roll back when the flow cannot do what it was asked to, as the transfer does when the debit finds too little money.

How the driver runs a flowCopy Anchor Link

The driver does exactly this, and nothing more:

1. Acquire a connection from the pool.
2. Send BEGIN in the flow's mode: the isolation level from ISOLATION, and READ ONLY if the flow declares no write, READ WRITE otherwise. If either fails, the result is Err(TxError::Begin) and the flow never runs.
3. Loop: if the state is terminal (finished returns Right), end the transaction (COMMIT or ROLLBACK, as Tx::end does) and stop. Otherwise call step with the open Tx and the state, and loop on the TxState it returns.

So a step can either end the transaction itself with tx.end(state) or return a terminal state with tx.with(state) and let the driver end it; the result is the same.

Capabilities and read-only transactionsCopy Anchor Link

A flow declares everything its steps may touch with tx_effect!, and names the declaration as TxFlow::Capability, as TransferEffect does for the transfer. A flow that also reads who owns the accounts would declare:

tx_effect! {
    /// Moves money between accounts, after checking who owns them.
    pub enum CheckedTransferEffect { write Account, read User }
}

write E lets the flow's queries read and write the entity E; read E lets them only read it. List each entity once: write already covers reading, and an entity listed twice does not compile. The macro generates an empty enum that is both a capability (it has the CanRead / CanWrite markers) and the matching effect (TxEffect); don't implement TxEffect by hand.

The declaration is checked at compile time, three ways:

• Each query against the flow. Tx::run requires the flow's capability to cover the query's effect. A query that touches an entity the flow did not declare, or writes one it declared read, is error[E0277].
• The flow against the handle. Db<S, C>, and AfterTransaction around it, processes Transaction<F> only if C covers the declared effect, exactly as for single queries.
• The mode. A flow that declares no write begins READ ONLY.

So a handle that may run a flow may run every query in it: a transaction can't reach an entity its handle was never checked for. The transfer writes Account, so it does not compile against a replica handle:

async fn on_replica(replica: &Db<PgSource, ReadOnly>) {
    // error[E0277]: the trait bound
    // `Db<SqlxSource<Postgres>, ReadOnly>: Processor<Transaction<Transfer>>` is not satisfied
    let _ = replica.process(Transaction(Transfer::new(1, 2, 10))).await;
}

A flow that only reads (for example a consistent report across several SELECTs, declared as read Account) begins READ ONLY and runs on a Db<_, ReadOnly> handle. In such a flow, Tx::run rejects a query that writes:

async fn step(tx: Tx<'_, Self>, state: Self) -> TxState<'_, Self> {
    // Does not compile: `Debit` writes `Account`, and this flow declares `read Account`.
    let (tx, _) = tx.run(Debit { from: 1, amount: 10 }).await;
    // …
}

Isolation levelsCopy Anchor Link

Set const ISOLATION to Some(IsolationLevel::…) to choose a level; leave it out for the server's default. The statements sent differ per database:

Errors inside a transactionCopy Anchor Link

Every query error reaches the step as the query's result, and the step decides what to do with it. What wakuwaku tracks for you is whether the error ended the transaction on the server, which differs between databases. Errors raised by sqlx on the client, after or instead of sending a statement, never change the server's state, so they never abort: RowNotFound, TypeNotFound, ColumnIndexOutOfBounds, ColumnNotFound, ColumnDecode, Encode, Decode, InvalidArgument.

Queries with their own error typeCopy Anchor Link

A query whose Error is not sqlx::Error (for example one that refuses on its own when an optimistic-lock version no longer matches) runs in a transaction if its error implements AsSqlxError. It tells wakuwaku which sqlx error, if any, is behind it. An error of the query's own (None) never aborts the transaction. See queries with their own error type for writing such a query:

use wakuwaku::deps::sqlx;
use wakuwaku::entity::sqlx::transaction::AsSqlxError;

#[derive(Debug)]
pub enum SaveError {
    /// Someone else saved first: the version did not match.
    Stale,
    Database(sqlx::Error),
}

impl From<sqlx::Error> for SaveError {
    fn from(error: sqlx::Error) -> Self {
        SaveError::Database(error)
    }
}

impl AsSqlxError for SaveError {
    fn as_sqlx_error(&self) -> Option<&sqlx::Error> {
        match self {
            SaveError::Database(error) => Some(error),
            SaveError::Stale => None,
        }
    }
}

RetriesCopy Anchor Link

wakuwaku does not retry transactions for you. A serialization failure or a deadlock reaches the step as a query error (the transfer turns it into TransferFailed::Conflict), and a SERIALIZABLE transaction can also fail at the end with TxError::CommitRejected. What to do next is your decision.

A flow value is one run and is not Clone, so retrying means building a fresh start state from the request, which is plain Copy data:

use crate::transfer::{Transfer, TransferFailed, is_conflict};
use wakuwaku::deps::kanau::processor::Processor;
use wakuwaku::entity::sqlx::transaction::{Transaction, TxError, TxOutput};

const MAX_ATTEMPTS: usize = 3;

/// Run the transfer, running it again from the start after a conflict.
pub async fn transfer_with_retry<D>(
    db: &D,
    from: i64,
    to: i64,
    amount: i64,
) -> Result<TxOutput<(), TransferFailed>, TxError>
where
    D: Processor<Transaction<Transfer>, Output = TxOutput<(), TransferFailed>, Error = TxError>,
{
    for _ in 1..MAX_ATTEMPTS {
        match db.process(Transaction(Transfer::new(from, to, amount))).await {
            // Aborted by a serialization failure or a deadlock.
            Ok(TxOutput::RolledBack(TransferFailed::Conflict)) => continue,
            // SERIALIZABLE can also reject the COMMIT itself.
            Err(TxError::CommitRejected(error)) if is_conflict(&error) => continue,
            // Never retry `CommitUnknown`: the transfer may have happened.
            outcome => return outcome,
        }
    }
    db.process(Transaction(Transfer::new(from, to, amount))).await
}

Only retry when the transaction is known not to have committed. Never retry CommitUnknown blindly: the money may already have moved. Make the operation idempotent (for example with a unique transfer id) or reconcile before trying again.

Hooking the end of a transactionCopy Anchor Link

Often something must happen after a transaction ends: publish an event, invalidate a cache, notify another service. Do not do it inside a step (the transaction is still open and may yet roll back) and do not do it with write hooks: queries inside a transaction run on the transaction's connection and never pass through BeforeWrite or AfterWrite. Use AfterTransaction instead.

AfterTransaction<D, H, EH> wraps a sqlx Db handle. After every transaction whose flow ran, it hands the flow's Report, tagged with what became of the transaction, to the hook H:

The hook is any Processor<TxFinalState<F::Report>>. Acting on a commit, a rollback, or an unknown commit is the hook's choice. If the hook fails, its error goes to an error handler EH, a Processor of the hook's error whose own error type is Infallible: it must deal with the failure itself (log it, retry, park it in an outbox). The caller's result is never changed by the hook.

use crate::transfer::{Transfer, Transferred};
use std::convert::Infallible;
use std::sync::mpsc::Sender;
use wakuwaku::deps::kanau::processor::Processor;
use wakuwaku::deps::sqlx::PgPool;
use wakuwaku::entity::sqlx::backend::{PgSource, SqlxSource};
use wakuwaku::entity::sqlx::transaction::hook::AfterTransaction;
use wakuwaku::entity::sqlx::transaction::{Transaction, TxError, TxFinalState};
use wakuwaku::entity::{Db, ReadWrite};

/// Hands each committed transfer to the event publisher.
pub struct PublishTransfers {
    events: Sender<Transferred>,
}

impl Processor<TxFinalState<Option<Transferred>>> for PublishTransfers {
    type Output = ();
    type Error = String;

    async fn process(&self, state: TxFinalState<Option<Transferred>>) -> Result<(), String> {
        match state {
            TxFinalState::Committed(Some(event)) => {
                self.events.send(event).map_err(|error| error.to_string())
            }
            TxFinalState::CommitUnknown(Some(event)) => {
                Err(format!("{event:?} may or may not have been committed"))
            }
            // Rolled back: nothing happened, nothing to publish.
            _ => Ok(()),
        }
    }
}

/// Hook errors end here; they never change what the caller gets.
pub struct LogHookError;

impl Processor<String> for LogHookError {
    type Output = ();
    type Error = Infallible;

    async fn process(&self, error: String) -> Result<(), Infallible> {
        eprintln!("transfer hook failed: {error}");
        Ok(())
    }
}

pub type Accounts = AfterTransaction<Db<PgSource, ReadWrite>, PublishTransfers, LogHookError>;

pub fn accounts(pool: PgPool, events: Sender<Transferred>) -> Accounts {
    AfterTransaction::new(
        Db::new(SqlxSource::new(pool)),
        PublishTransfers { events },
        LogHookError,
    )
}

Accounts processes Transaction<Transfer> with the same output and error as the bare Db, so the transfer_with_retry helper above works with it unchanged. A few details:

• The hook does not run when the flow never began (TxError::Begin): there is no report to hand over.
• In a real service, the hook would publish to Iggy; see Publishing events. Deciding what to publish is what Report is for: the flow puts the event in the report, and only the hook sees it.
• AfterTransaction forwards queries run on their own to the wrapped Db, and their errors are the queries' own Error types, so write hooks can wrap it: AfterWrite<AfterTransaction<Db<_, _>, _, _>, _, _>. Write hooks do not forward transactions; reach the transaction hook through AfterWrite::inner().
• AfterTransaction must wrap the Db directly.

Testing flowsCopy Anchor Link

A Tx cannot be built outside wakuwaku, so step itself only runs against a real database. Split the testing accordingly:

• Decisions without a database. Keep the logic in pure functions of query results and the request, as check_debit, check_credit, commit and rollback are, and test them together with finished. These tests need no database, no Docker, and no async runtime.
• Steps against a database. Run the whole flow through a Db in an integration test, to check the SQL and that each step runs the query you meant.
• The protocol is not yours to test. Which errors abort, refusing to commit an aborted transaction, and what a lost COMMIT means are tested inside wakuwaku.
• Hooks are processors. Call process on the hook with a TxFinalState you build yourself.

#[cfg(test)]
mod tests {
    use super::*;
    use wakuwaku::entity::sqlx::transaction::Either;

    #[test]
    fn no_row_debited_is_insufficient_funds() {
        assert!(matches!(Transfer::check_debit(Ok(0)), Err(TransferFailed::InsufficientFunds)));
    }

    #[test]
    fn no_row_credited_is_an_unknown_account() {
        assert!(matches!(Transfer::check_credit(Ok(0)), Err(TransferFailed::UnknownAccount)));
    }

    #[test]
    fn a_failed_query_is_a_database_error() {
        let lost = sqlx::Error::Io(std::io::Error::other("connection reset"));
        assert!(matches!(Transfer::check_credit(Err(lost)), Err(TransferFailed::Database(_))));
    }

    #[test]
    fn a_commit_reports_the_transfer() {
        let Either::Right(TxEnd::Commit { report, .. }) = Transfer::commit(1, 2, 10).finished() else {
            panic!("`commit` is a terminal commit");
        };
        assert_eq!(report, Some(Transferred { from: 1, to: 2, amount: 10 }));
    }

    #[test]
    fn a_rollback_reports_nothing() {
        assert!(matches!(
            Transfer::rollback(TransferFailed::InsufficientFunds).finished(),
            Either::Right(TxEnd::Rollback { output: TransferFailed::InsufficientFunds, report: None })
        ));
    }

    #[test]
    fn the_start_is_not_terminal() {
        assert!(Transfer::new(1, 2, 10).finished().is_left());
    }
}

API referenceCopy Anchor Link

• transaction module: TxFlow, TxEffect, Tx, TxState, TxEnd, AsSqlxError, Transaction, TxOutput, TxError, TxFinalState
• tx_effect!
• AfterTransaction