codeloopback
Guide contents

Lesson 29 of 30

Sagas and Distributed Transactions

A local transaction covers one database. A saga replaces a cross-service transaction with a sequence of local transactions, plus a compensating action for each step that can be undone. This lesson builds the Storefront checkout saga in the orders service: a state machine, an outbox relay, an inbox, and timeouts.

Project setup

Start PostgreSQL:

Terminal
docker run --name saga-pg -e POSTGRES_PASSWORD=pw -e POSTGRES_DB=orders -p 5432:5432 -d postgres:16

Cargo.toml

TOML
[package]
name = "checkout-saga"
version = "0.1.0"
edition = "2021"

[dependencies]
axum = "0.8"
tokio = { version = "1", features = ["full"] }
sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio", "postgres", "json"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"

migrations/0001_saga.sql

SQL
CREATE TABLE IF NOT EXISTS orders (
    id          BIGSERIAL PRIMARY KEY,
    customer_id BIGINT NOT NULL,
    status      TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS checkout_sagas (
    order_id    BIGINT PRIMARY KEY REFERENCES orders (id),
    state       TEXT NOT NULL,
    deadline_at TIMESTAMPTZ NOT NULL,
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS outbox (
    id           BIGSERIAL PRIMARY KEY,
    topic        TEXT NOT NULL,
    msg_key      TEXT NOT NULL,
    payload      JSONB NOT NULL,
    published_at TIMESTAMPTZ
);

CREATE TABLE IF NOT EXISTS processed_messages (
    id TEXT PRIMARY KEY
);

Example 1: Why a plain sequence of calls is unsafe

Checkout has three steps: create the order, reserve stock, charge the card. Each step commits in its own database, so a failure at step three cannot roll back steps one and two.

src/main.rs

Rust
async fn create_order() -> Result<u32, String> {
    println!("order 1 created");
    Ok(1)
}

async fn reserve_stock(order_id: u32) -> Result<(), String> {
    println!("stock reserved for order {order_id}");
    Ok(())
}

async fn charge_card(order_id: u32) -> Result<(), String> {
    Err(format!("card declined for order {order_id}"))
}

#[tokio::main]
async fn main() {
    let result = async {
        let id = create_order().await?;
        reserve_stock(id).await?;
        charge_card(id).await
    }
    .await;
    println!("checkout result: {result:?}");
}
Output
order 1 created
stock reserved for order 1
checkout result: Err("card declined for order 1")

Stock stays reserved for an order that cannot ship. A crash between steps leaves no record of where the checkout stopped.

Two-phase commit and sagas

Two-phase commit (2PC) uses a coordinator. Each participant first prepares (makes changes durable but invisible) and votes. The coordinator then sends commit if all voted yes, or abort.

  • Participants hold locks between prepare and commit, so a slow coordinator blocks other requests.
  • If the coordinator crashes after prepare, participants wait for its recovery.
  • Every participant must support the protocol (for example XA). Brokers and third-party HTTP APIs do not.
  • Availability is the product of all participants' availability.

A saga gives up isolation instead: intermediate states are visible to other requests, and consistency arrives when the saga finishes.

A compensating action is a new business operation that reverses an earlier step. It does not roll back a database.

Step Local transaction Compensation
Create order Insert with status pending Set status cancelled
Reserve stock Decrease available, add reservation Delete reservation, restore available
Charge payment Create charge Refund the charge
Send email Send None; run this step last
  • Each compensation is idempotent, because it can be delivered more than once.
  • A compensation retries until it succeeds. Repeated failure escalates to a person.
  • Steps that cannot be undone run after all steps that can fail.

Choreography and orchestration

In choreography, no component drives the flow. Each service listens for events and publishes new ones (OrderPlaced to inventory, StockReserved to payments, PaymentCaptured to orders).

  • No coordinator, so no single point of failure.
  • The flow is spread over subscriptions and is hard to read as a whole.
  • Fits two to four steps in a stable flow.

In orchestration, one component holds the flow as a state machine, sends commands, receives replies, and persists its state after each transition.

  • One place shows the flow and the state of each saga.
  • The orchestrator must be durable. It can run as several instances over one saga table.
  • Fits five or more steps, branching, timeouts, and status queries.

The Storefront uses orchestration in the orders service, because orders owns the order.

Example 2: The saga state machine

The state machine is pure code with no I/O, so it can be read and tested on its own. This example adds two files.

src/saga.rs

Rust
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum State {
    Started,
    StockReserved,
    Completed,
    Releasing,
    Cancelled,
}

impl State {
    pub fn as_str(self) -> &'static str {
        match self {
            State::Started => "started",
            State::StockReserved => "stock_reserved",
            State::Completed => "completed",
            State::Releasing => "releasing",
            State::Cancelled => "cancelled",
        }
    }

    pub fn parse(s: &str) -> Option<State> {
        Some(match s {
            "started" => State::Started,
            "stock_reserved" => State::StockReserved,
            "completed" => State::Completed,
            "releasing" => State::Releasing,
            "cancelled" => State::Cancelled,
            _ => return None,
        })
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Reply {
    StockReserved,
    StockRejected,
    PaymentCaptured,
    PaymentFailed,
    StockReleased,
}

impl Reply {
    pub fn parse(s: &str) -> Option<Reply> {
        Some(match s {
            "StockReserved" => Reply::StockReserved,
            "StockRejected" => Reply::StockRejected,
            "PaymentCaptured" => Reply::PaymentCaptured,
            "PaymentFailed" => Reply::PaymentFailed,
            "StockReleased" => Reply::StockReleased,
            _ => return None,
        })
    }
}

/// Returns the next state. A reply that does not apply in the current
/// state (late or duplicate) leaves the state unchanged.
pub fn next(state: State, reply: Reply) -> State {
    match (state, reply) {
        (State::Started, Reply::StockReserved) => State::StockReserved,
        (State::Started, Reply::StockRejected) => State::Cancelled,
        (State::StockReserved, Reply::PaymentCaptured) => State::Completed,
        (State::StockReserved, Reply::PaymentFailed) => State::Releasing,
        (State::Releasing, Reply::StockReleased) => State::Cancelled,
        (s, _) => s,
    }
}

/// The command to queue when the saga enters a state.
pub fn command_for(state: State) -> Option<(&'static str, &'static str)> {
    match state {
        State::StockReserved => Some(("payments.commands", "ChargePayment")),
        State::Releasing => Some(("inventory.commands", "ReleaseStock")),
        _ => None,
    }
}

src/main.rs

Rust
mod saga;

use saga::{next, Reply, State};

fn main() {
    let mut state = State::Started;
    for reply in [Reply::StockReserved, Reply::PaymentFailed, Reply::PaymentCaptured, Reply::StockReleased] {
        state = next(state, reply);
        println!("{:?} -> {}", reply, state.as_str());
    }
}
Output
StockReserved -> stock_reserved
PaymentFailed -> releasing
PaymentCaptured -> releasing
StockReleased -> cancelled

The third reply, a late PaymentCaptured, does not change the state. The saga ignores it here. A production system also refunds the unexpected capture.

Example 3: Starting the saga with an outbox row

The dual-write problem: a service writes to its database and then publishes a message. A crash between the two loses the message. The outbox pattern inserts the message into an outbox table in the same transaction as the state change. A relay publishes it later.

This example adds src/error.rs and src/handlers.rs, and a main.rs that serves them.

src/error.rs

Rust
use axum::{http::StatusCode, response::{IntoResponse, Response}, Json};
use serde_json::json;

pub struct AppError(pub StatusCode, pub String);

impl From<sqlx::Error> for AppError {
    fn from(e: sqlx::Error) -> Self {
        eprintln!("database error: {e}");
        AppError(StatusCode::INTERNAL_SERVER_ERROR, "internal error".to_string())
    }
}

impl IntoResponse for AppError {
    fn into_response(self) -> Response {
        (self.0, Json(json!({ "error": self.1 }))).into_response()
    }
}

src/handlers.rs

Rust
use axum::{
    extract::{Path, State},
    http::StatusCode,
    Json,
};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use sqlx::PgPool;

use crate::error::AppError;

#[derive(Deserialize)]
pub struct PlaceOrder {
    pub customer_id: i64,
    pub lines: Vec<Line>,
}

#[derive(Deserialize, Serialize)]
pub struct Line {
    pub sku: String,
    pub qty: i32,
}

pub async fn place_order(
    State(pool): State<PgPool>,
    Json(req): Json<PlaceOrder>,
) -> Result<(StatusCode, Json<Value>), AppError> {
    let mut tx = pool.begin().await?;

    let order_id: i64 = sqlx::query_scalar(
        "INSERT INTO orders (customer_id, status) VALUES ($1, 'pending') RETURNING id",
    )
    .bind(req.customer_id)
    .fetch_one(&mut *tx)
    .await?;

    sqlx::query(
        "INSERT INTO checkout_sagas (order_id, state, deadline_at)
         VALUES ($1, 'started', now() + interval '2 minutes')",
    )
    .bind(order_id)
    .execute(&mut *tx)
    .await?;

    let command = json!({
        "type": "ReserveStock",
        "order_id": order_id,
        "lines": req.lines,
    });
    sqlx::query("INSERT INTO outbox (topic, msg_key, payload) VALUES ('inventory.commands', $1, $2)")
        .bind(order_id.to_string())
        .bind(command)
        .execute(&mut *tx)
        .await?;

    tx.commit().await?;
    Ok((StatusCode::ACCEPTED, Json(json!({ "order_id": order_id, "status": "pending" }))))
}

pub async fn get_order(
    State(pool): State<PgPool>,
    Path(id): Path<i64>,
) -> Result<Json<Value>, AppError> {
    let row: Option<(String, String)> = sqlx::query_as(
        "SELECT o.status, s.state FROM orders o
         JOIN checkout_sagas s ON s.order_id = o.id WHERE o.id = $1",
    )
    .bind(id)
    .fetch_optional(&pool)
    .await?;

    match row {
        Some((status, saga_state)) => Ok(Json(json!({
            "order_id": id,
            "status": status,
            "saga_state": saga_state,
        }))),
        None => Err(AppError(StatusCode::NOT_FOUND, "order not found".to_string())),
    }
}

src/main.rs

Rust
mod error;
mod handlers;
mod saga;

use axum::{routing::{get, post}, Router};
use sqlx::postgres::PgPoolOptions;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect("postgres://postgres:pw@localhost:5432/orders")
        .await?;
    sqlx::raw_sql(include_str!("../migrations/0001_saga.sql")).execute(&pool).await?;

    let app = Router::new()
        .route("/orders", post(handlers::place_order))
        .route("/orders/{id}", get(handlers::get_order))
        .with_state(pool);

    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await?;
    axum::serve(listener, app).await?;
    Ok(())
}
  • The handler returns 202 Accepted. The order is not confirmed yet, so the client polls GET /orders/{id}.
  • The order, the saga row, and the first command commit together or not at all.
  • mod saga; stays because the next examples use it.

Example 4: The relay and the timeout sweep

The relay reads unpublished outbox rows and publishes them. FOR UPDATE SKIP LOCKED lets several relay instances run without taking the same rows. A crash after publishing and before the commit republishes a row, so delivery is at-least-once. The outbox row id travels as the message id for the consumer's inbox.

The timeout sweep finds sagas past deadline_at and moves them to compensation. Deadlines live in the database, so a restart does not lose them.

This example publishes by printing. A production relay calls the Kafka producer from Messaging and Kafka at that point.

src/relay.rs

Rust
use std::time::Duration;

use serde_json::{json, Value};
use sqlx::PgPool;

pub async fn relay_once(pool: &PgPool) -> Result<usize, sqlx::Error> {
    let mut tx = pool.begin().await?;

    let rows: Vec<(i64, String, String, Value)> = sqlx::query_as(
        "SELECT id, topic, msg_key, payload FROM outbox
         WHERE published_at IS NULL
         ORDER BY id LIMIT 100
         FOR UPDATE SKIP LOCKED",
    )
    .fetch_all(&mut *tx)
    .await?;

    for (id, topic, key, payload) in &rows {
        println!("publish id={id} topic={topic} key={key} payload={payload}");
        sqlx::query("UPDATE outbox SET published_at = now() WHERE id = $1")
            .bind(id)
            .execute(&mut *tx)
            .await?;
    }

    tx.commit().await?;
    Ok(rows.len())
}

pub async fn sweep_timeouts(pool: &PgPool) -> Result<usize, sqlx::Error> {
    let mut tx = pool.begin().await?;

    let expired: Vec<i64> = sqlx::query_scalar(
        "SELECT order_id FROM checkout_sagas
         WHERE state IN ('started', 'stock_reserved') AND deadline_at < now()
         FOR UPDATE SKIP LOCKED",
    )
    .fetch_all(&mut *tx)
    .await?;

    for order_id in &expired {
        sqlx::query("UPDATE checkout_sagas SET state = 'releasing', updated_at = now() WHERE order_id = $1")
            .bind(order_id)
            .execute(&mut *tx)
            .await?;
        sqlx::query("INSERT INTO outbox (topic, msg_key, payload) VALUES ('inventory.commands', $1, $2)")
            .bind(order_id.to_string())
            .bind(json!({ "type": "ReleaseStock", "order_id": order_id }))
            .execute(&mut *tx)
            .await?;
    }

    tx.commit().await?;
    Ok(expired.len())
}

pub async fn run(pool: PgPool) {
    let mut tick = 0u64;
    loop {
        if let Err(e) = relay_once(&pool).await {
            eprintln!("relay error: {e}");
        }
        if tick % 10 == 0 {
            if let Err(e) = sweep_timeouts(&pool).await {
                eprintln!("sweep error: {e}");
            }
        }
        tick += 1;
        tokio::time::sleep(Duration::from_millis(500)).await;
    }
}

The sweep runs every ten relay ticks (5 seconds). ReleaseStock is idempotent, so it is safe even when stock was never reserved.

Example 5: The inbox and reply handling

The inbox is the consumer-side counterpart of the outbox. The handler inserts the message id into processed_messages in the same transaction as its effect. A repeated id inserts zero rows, so the effect is skipped. With both, a message is stored once, delivered one or more times, and applied once.

In this service, replies arrive on POST /internal/replies, which stands in for a broker consumer. The new file is src/replies.rs; main.rs registers it in the complete example below.

src/replies.rs

Rust
use axum::{extract::State, http::StatusCode, Json};
use serde::Deserialize;
use serde_json::{json, Value};
use sqlx::PgPool;

use crate::error::AppError;
use crate::saga::{command_for, next, Reply, State as SagaState};

#[derive(Deserialize)]
pub struct ReplyMsg {
    pub message_id: String,
    pub order_id: i64,
    #[serde(rename = "type")]
    pub kind: String,
}

pub async fn handle_reply(
    State(pool): State<PgPool>,
    Json(msg): Json<ReplyMsg>,
) -> Result<Json<Value>, AppError> {
    let reply = Reply::parse(&msg.kind).ok_or_else(|| {
        AppError(StatusCode::UNPROCESSABLE_ENTITY, format!("unknown reply type {}", msg.kind))
    })?;

    let mut tx = pool.begin().await?;

    let fresh = sqlx::query("INSERT INTO processed_messages (id) VALUES ($1) ON CONFLICT DO NOTHING")
        .bind(&msg.message_id)
        .execute(&mut *tx)
        .await?
        .rows_affected()
        == 1;
    if !fresh {
        return Ok(Json(json!({ "applied": false, "reason": "duplicate" })));
    }

    let current: Option<String> =
        sqlx::query_scalar("SELECT state FROM checkout_sagas WHERE order_id = $1 FOR UPDATE")
            .bind(msg.order_id)
            .fetch_optional(&mut *tx)
            .await?;
    let current = current
        .and_then(|s| SagaState::parse(&s))
        .ok_or_else(|| AppError(StatusCode::NOT_FOUND, "saga not found".to_string()))?;

    let new_state = next(current, reply);
    if new_state != current {
        sqlx::query("UPDATE checkout_sagas SET state = $2, updated_at = now() WHERE order_id = $1")
            .bind(msg.order_id)
            .bind(new_state.as_str())
            .execute(&mut *tx)
            .await?;

        let order_status = match new_state {
            SagaState::Completed => Some("confirmed"),
            SagaState::Cancelled => Some("cancelled"),
            _ => None,
        };
        if let Some(status) = order_status {
            sqlx::query("UPDATE orders SET status = $2 WHERE id = $1")
                .bind(msg.order_id)
                .bind(status)
                .execute(&mut *tx)
                .await?;
        }

        if let Some((topic, kind)) = command_for(new_state) {
            sqlx::query("INSERT INTO outbox (topic, msg_key, payload) VALUES ($1, $2, $3)")
                .bind(topic)
                .bind(msg.order_id.to_string())
                .bind(json!({ "type": kind, "order_id": msg.order_id }))
                .execute(&mut *tx)
                .await?;
        }
    }

    tx.commit().await?;
    Ok(Json(json!({ "applied": new_state != current, "state": new_state.as_str() })))
}
  • The state change, the order status, and the next outbox command commit in one transaction with the inbox row.
  • FOR UPDATE on the saga row serializes two replies for the same order.
  • A reply for an unknown order returns 404 and the transaction rolls back, so the inbox row is not kept.

Complete example

File tree:

Text
checkout-saga/
  Cargo.toml
  migrations/0001_saga.sql
  src/main.rs
  src/error.rs
  src/saga.rs
  src/handlers.rs
  src/replies.rs
  src/relay.rs

Cargo.toml

TOML
[package]
name = "checkout-saga"
version = "0.1.0"
edition = "2021"

[dependencies]
axum = "0.8"
tokio = { version = "1", features = ["full"] }
sqlx = { version = "0.8", default-features = false, features = ["runtime-tokio", "postgres", "json"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"

migrations/0001_saga.sql

SQL
CREATE TABLE IF NOT EXISTS orders (
    id          BIGSERIAL PRIMARY KEY,
    customer_id BIGINT NOT NULL,
    status      TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS checkout_sagas (
    order_id    BIGINT PRIMARY KEY REFERENCES orders (id),
    state       TEXT NOT NULL,
    deadline_at TIMESTAMPTZ NOT NULL,
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE IF NOT EXISTS outbox (
    id           BIGSERIAL PRIMARY KEY,
    topic        TEXT NOT NULL,
    msg_key      TEXT NOT NULL,
    payload      JSONB NOT NULL,
    published_at TIMESTAMPTZ
);

CREATE TABLE IF NOT EXISTS processed_messages (
    id TEXT PRIMARY KEY
);

src/main.rs

Rust
mod error;
mod handlers;
mod relay;
mod replies;
mod saga;

use axum::{routing::{get, post}, Router};
use sqlx::postgres::PgPoolOptions;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = PgPoolOptions::new()
        .max_connections(5)
        .connect("postgres://postgres:pw@localhost:5432/orders")
        .await?;
    sqlx::raw_sql(include_str!("../migrations/0001_saga.sql")).execute(&pool).await?;

    tokio::spawn(relay::run(pool.clone()));

    let app = Router::new()
        .route("/orders", post(handlers::place_order))
        .route("/orders/{id}", get(handlers::get_order))
        .route("/internal/replies", post(replies::handle_reply))
        .with_state(pool);

    let listener = tokio::net::TcpListener::bind("127.0.0.1:3000").await?;
    axum::serve(listener, app).await?;
    Ok(())
}

src/error.rs

Rust
use axum::{http::StatusCode, response::{IntoResponse, Response}, Json};
use serde_json::json;

pub struct AppError(pub StatusCode, pub String);

impl From<sqlx::Error> for AppError {
    fn from(e: sqlx::Error) -> Self {
        eprintln!("database error: {e}");
        AppError(StatusCode::INTERNAL_SERVER_ERROR, "internal error".to_string())
    }
}

impl IntoResponse for AppError {
    fn into_response(self) -> Response {
        (self.0, Json(json!({ "error": self.1 }))).into_response()
    }
}

src/saga.rs

Rust
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum State {
    Started,
    StockReserved,
    Completed,
    Releasing,
    Cancelled,
}

impl State {
    pub fn as_str(self) -> &'static str {
        match self {
            State::Started => "started",
            State::StockReserved => "stock_reserved",
            State::Completed => "completed",
            State::Releasing => "releasing",
            State::Cancelled => "cancelled",
        }
    }

    pub fn parse(s: &str) -> Option<State> {
        Some(match s {
            "started" => State::Started,
            "stock_reserved" => State::StockReserved,
            "completed" => State::Completed,
            "releasing" => State::Releasing,
            "cancelled" => State::Cancelled,
            _ => return None,
        })
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Reply {
    StockReserved,
    StockRejected,
    PaymentCaptured,
    PaymentFailed,
    StockReleased,
}

impl Reply {
    pub fn parse(s: &str) -> Option<Reply> {
        Some(match s {
            "StockReserved" => Reply::StockReserved,
            "StockRejected" => Reply::StockRejected,
            "PaymentCaptured" => Reply::PaymentCaptured,
            "PaymentFailed" => Reply::PaymentFailed,
            "StockReleased" => Reply::StockReleased,
            _ => return None,
        })
    }
}

/// Returns the next state. A reply that does not apply in the current
/// state (late or duplicate) leaves the state unchanged.
pub fn next(state: State, reply: Reply) -> State {
    match (state, reply) {
        (State::Started, Reply::StockReserved) => State::StockReserved,
        (State::Started, Reply::StockRejected) => State::Cancelled,
        (State::StockReserved, Reply::PaymentCaptured) => State::Completed,
        (State::StockReserved, Reply::PaymentFailed) => State::Releasing,
        (State::Releasing, Reply::StockReleased) => State::Cancelled,
        (s, _) => s,
    }
}

/// The command to queue when the saga enters a state.
pub fn command_for(state: State) -> Option<(&'static str, &'static str)> {
    match state {
        State::StockReserved => Some(("payments.commands", "ChargePayment")),
        State::Releasing => Some(("inventory.commands", "ReleaseStock")),
        _ => None,
    }
}

src/handlers.rs

Rust
use axum::{
    extract::{Path, State},
    http::StatusCode,
    Json,
};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use sqlx::PgPool;

use crate::error::AppError;

#[derive(Deserialize)]
pub struct PlaceOrder {
    pub customer_id: i64,
    pub lines: Vec<Line>,
}

#[derive(Deserialize, Serialize)]
pub struct Line {
    pub sku: String,
    pub qty: i32,
}

pub async fn place_order(
    State(pool): State<PgPool>,
    Json(req): Json<PlaceOrder>,
) -> Result<(StatusCode, Json<Value>), AppError> {
    let mut tx = pool.begin().await?;

    let order_id: i64 = sqlx::query_scalar(
        "INSERT INTO orders (customer_id, status) VALUES ($1, 'pending') RETURNING id",
    )
    .bind(req.customer_id)
    .fetch_one(&mut *tx)
    .await?;

    sqlx::query(
        "INSERT INTO checkout_sagas (order_id, state, deadline_at)
         VALUES ($1, 'started', now() + interval '2 minutes')",
    )
    .bind(order_id)
    .execute(&mut *tx)
    .await?;

    let command = json!({
        "type": "ReserveStock",
        "order_id": order_id,
        "lines": req.lines,
    });
    sqlx::query("INSERT INTO outbox (topic, msg_key, payload) VALUES ('inventory.commands', $1, $2)")
        .bind(order_id.to_string())
        .bind(command)
        .execute(&mut *tx)
        .await?;

    tx.commit().await?;
    Ok((StatusCode::ACCEPTED, Json(json!({ "order_id": order_id, "status": "pending" }))))
}

pub async fn get_order(
    State(pool): State<PgPool>,
    Path(id): Path<i64>,
) -> Result<Json<Value>, AppError> {
    let row: Option<(String, String)> = sqlx::query_as(
        "SELECT o.status, s.state FROM orders o
         JOIN checkout_sagas s ON s.order_id = o.id WHERE o.id = $1",
    )
    .bind(id)
    .fetch_optional(&pool)
    .await?;

    match row {
        Some((status, saga_state)) => Ok(Json(json!({
            "order_id": id,
            "status": status,
            "saga_state": saga_state,
        }))),
        None => Err(AppError(StatusCode::NOT_FOUND, "order not found".to_string())),
    }
}

src/replies.rs

Rust
use axum::{extract::State, http::StatusCode, Json};
use serde::Deserialize;
use serde_json::{json, Value};
use sqlx::PgPool;

use crate::error::AppError;
use crate::saga::{command_for, next, Reply, State as SagaState};

#[derive(Deserialize)]
pub struct ReplyMsg {
    pub message_id: String,
    pub order_id: i64,
    #[serde(rename = "type")]
    pub kind: String,
}

pub async fn handle_reply(
    State(pool): State<PgPool>,
    Json(msg): Json<ReplyMsg>,
) -> Result<Json<Value>, AppError> {
    let reply = Reply::parse(&msg.kind).ok_or_else(|| {
        AppError(StatusCode::UNPROCESSABLE_ENTITY, format!("unknown reply type {}", msg.kind))
    })?;

    let mut tx = pool.begin().await?;

    let fresh = sqlx::query("INSERT INTO processed_messages (id) VALUES ($1) ON CONFLICT DO NOTHING")
        .bind(&msg.message_id)
        .execute(&mut *tx)
        .await?
        .rows_affected()
        == 1;
    if !fresh {
        return Ok(Json(json!({ "applied": false, "reason": "duplicate" })));
    }

    let current: Option<String> =
        sqlx::query_scalar("SELECT state FROM checkout_sagas WHERE order_id = $1 FOR UPDATE")
            .bind(msg.order_id)
            .fetch_optional(&mut *tx)
            .await?;
    let current = current
        .and_then(|s| SagaState::parse(&s))
        .ok_or_else(|| AppError(StatusCode::NOT_FOUND, "saga not found".to_string()))?;

    let new_state = next(current, reply);
    if new_state != current {
        sqlx::query("UPDATE checkout_sagas SET state = $2, updated_at = now() WHERE order_id = $1")
            .bind(msg.order_id)
            .bind(new_state.as_str())
            .execute(&mut *tx)
            .await?;

        let order_status = match new_state {
            SagaState::Completed => Some("confirmed"),
            SagaState::Cancelled => Some("cancelled"),
            _ => None,
        };
        if let Some(status) = order_status {
            sqlx::query("UPDATE orders SET status = $2 WHERE id = $1")
                .bind(msg.order_id)
                .bind(status)
                .execute(&mut *tx)
                .await?;
        }

        if let Some((topic, kind)) = command_for(new_state) {
            sqlx::query("INSERT INTO outbox (topic, msg_key, payload) VALUES ($1, $2, $3)")
                .bind(topic)
                .bind(msg.order_id.to_string())
                .bind(json!({ "type": kind, "order_id": msg.order_id }))
                .execute(&mut *tx)
                .await?;
        }
    }

    tx.commit().await?;
    Ok(Json(json!({ "applied": new_state != current, "state": new_state.as_str() })))
}

src/relay.rs

Rust
use std::time::Duration;

use serde_json::{json, Value};
use sqlx::PgPool;

pub async fn relay_once(pool: &PgPool) -> Result<usize, sqlx::Error> {
    let mut tx = pool.begin().await?;

    let rows: Vec<(i64, String, String, Value)> = sqlx::query_as(
        "SELECT id, topic, msg_key, payload FROM outbox
         WHERE published_at IS NULL
         ORDER BY id LIMIT 100
         FOR UPDATE SKIP LOCKED",
    )
    .fetch_all(&mut *tx)
    .await?;

    for (id, topic, key, payload) in &rows {
        println!("publish id={id} topic={topic} key={key} payload={payload}");
        sqlx::query("UPDATE outbox SET published_at = now() WHERE id = $1")
            .bind(id)
            .execute(&mut *tx)
            .await?;
    }

    tx.commit().await?;
    Ok(rows.len())
}

pub async fn sweep_timeouts(pool: &PgPool) -> Result<usize, sqlx::Error> {
    let mut tx = pool.begin().await?;

    let expired: Vec<i64> = sqlx::query_scalar(
        "SELECT order_id FROM checkout_sagas
         WHERE state IN ('started', 'stock_reserved') AND deadline_at < now()
         FOR UPDATE SKIP LOCKED",
    )
    .fetch_all(&mut *tx)
    .await?;

    for order_id in &expired {
        sqlx::query("UPDATE checkout_sagas SET state = 'releasing', updated_at = now() WHERE order_id = $1")
            .bind(order_id)
            .execute(&mut *tx)
            .await?;
        sqlx::query("INSERT INTO outbox (topic, msg_key, payload) VALUES ('inventory.commands', $1, $2)")
            .bind(order_id.to_string())
            .bind(json!({ "type": "ReleaseStock", "order_id": order_id }))
            .execute(&mut *tx)
            .await?;
    }

    tx.commit().await?;
    Ok(expired.len())
}

pub async fn run(pool: PgPool) {
    let mut tick = 0u64;
    loop {
        if let Err(e) = relay_once(&pool).await {
            eprintln!("relay error: {e}");
        }
        if tick % 10 == 0 {
            if let Err(e) = sweep_timeouts(&pool).await {
                eprintln!("sweep error: {e}");
            }
        }
        tick += 1;
        tokio::time::sleep(Duration::from_millis(500)).await;
    }
}

Start PostgreSQL with the docker run line from Project setup, then run cargo run. On an empty database, the requests below give these responses.

Terminal
curl -s -X POST localhost:3000/orders -H 'content-type: application/json' \
  -d '{"customer_id":7,"lines":[{"sku":"notebook","qty":2}]}'
Output
{"order_id":1,"status":"pending"}

The service log shows publish id=1 topic=inventory.commands key=1 payload={"lines":[{"qty":2,"sku":"notebook"}],"order_id":1,"type":"ReserveStock"}.

Terminal
curl -s -X POST localhost:3000/internal/replies -H 'content-type: application/json' \
  -d '{"message_id":"m-1","order_id":1,"type":"StockReserved"}'
Output
{"applied":true,"state":"stock_reserved"}

The log shows a ChargePayment command on payments.commands. Sending the same request again returns:

Output
{"applied":false,"reason":"duplicate"}
Terminal
curl -s -X POST localhost:3000/internal/replies -H 'content-type: application/json' \
  -d '{"message_id":"m-2","order_id":1,"type":"PaymentCaptured"}'
curl -s localhost:3000/orders/1
Output
{"applied":true,"state":"completed"}
{"order_id":1,"saga_state":"completed","status":"confirmed"}

Failure path for a second order (order_id 2): after StockReserved (m-3), a PaymentFailed reply (m-4) returns {"applied":true,"state":"releasing"} and queues ReleaseStock. A StockReleased reply (m-5) returns {"applied":true,"state":"cancelled"}, and the order status becomes cancelled.

Timeout path: create order 3, then move its deadline into the past. Within 5 seconds the sweep sets it to releasing, and the log shows a ReleaseStock publish.

Terminal
docker exec saga-pg psql -U postgres -d orders \
  -c "UPDATE checkout_sagas SET deadline_at = now() - interval '1 second' WHERE order_id = 3"

Summary

  • A saga is a chain of local transactions plus compensations. Compensations are idempotent business actions retried until they succeed.
  • Two-phase commit blocks on coordinator failure and needs participant support, so it is rare between services.
  • Choreography has no coordinator. Orchestration keeps a persisted state machine.
  • The state machine ignores replies that do not apply, so late and duplicate replies are harmless.
  • The outbox removes the gap between a database write and a publish. Delivery is at-least-once.
  • The inbox turns repeated delivery into one effect.
  • Deadlines stored in the database drive timeouts. FOR UPDATE SKIP LOCKED lets several instances share work.