codeloopback
Guide contents

Lesson 8 of 30

Transactions, Locking, Idempotency, and the Outbox

A checkout changes several rows, two requests can compete for the last unit of stock, and a client can retry a request after a timeout. This lesson covers the database tools for those problems: transactions, locks, versions, idempotency keys, and an outbox that links database changes to messages. It assumes PostgreSQL with SQLx. Theory is in Transactions and ACID, Isolation levels, and Locking and deadlocks.

Project setup

Terminal
docker run --name storefront-pg -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=storefront -p 5432:5432 -d postgres:16
export DATABASE_URL=postgres://postgres:postgres@localhost:5432/storefront

Each demo program expects a fresh database. Remove the container with docker rm -f storefront-pg, start it again, and wait a few seconds until PostgreSQL accepts connections before each run.

Cargo.toml

TOML
[package]
name = "storefront-transactions"
version = "0.1.0"
edition = "2021"

[dependencies]
axum = "0.8.4"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
sqlx = { version = "0.8", features = ["runtime-tokio", "postgres", "migrate", "macros", "derive", "json"] }
thiserror = "2"
tokio = { version = "1.44", features = ["full"] }

The json feature maps JSONB to serde_json::Value.

migrations/20260929120000_storefront.sql

SQL
CREATE TABLE products (
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    sku         TEXT    NOT NULL UNIQUE,
    name        TEXT    NOT NULL,
    price_cents INTEGER NOT NULL CHECK (price_cents > 0),
    version     BIGINT  NOT NULL DEFAULT 1
);

CREATE TABLE inventory (
    product_id BIGINT  PRIMARY KEY REFERENCES products (id),
    on_hand    INTEGER NOT NULL CHECK (on_hand >= 0)
);

CREATE TABLE orders (
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    customer_id BIGINT      NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE order_lines (
    order_id   BIGINT  NOT NULL REFERENCES orders (id),
    product_id BIGINT  NOT NULL REFERENCES products (id),
    quantity   INTEGER NOT NULL CHECK (quantity > 0),
    PRIMARY KEY (order_id, product_id)
);

CREATE TABLE idempotency_keys (
    key           TEXT        PRIMARY KEY,
    fingerprint   TEXT        NOT NULL,
    response_body JSONB,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE outbox (
    id           BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    topic        TEXT        NOT NULL,
    payload      JSONB       NOT NULL,
    created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
    published_at TIMESTAMPTZ
);

CREATE INDEX outbox_unpublished_idx ON outbox (id) WHERE published_at IS NULL;

INSERT INTO products (sku, name, price_cents) VALUES ('NB-1', 'Notebook', 450), ('PN-1', 'Pen', 150);
INSERT INTO inventory (product_id, on_hand) VALUES (1, 5), (2, 10);

The migration seeds two products: the Notebook (id 1, stock 5) and the Pen (id 2, stock 10). The CHECK (on_hand >= 0) constraint is a second line of defense against overselling.

Example 1: A transaction and a rollback

A transaction groups statements so that all changes become visible at COMMIT or none do. pool.begin() takes a connection from the pool, starts a transaction, and returns a Transaction. If the value is dropped without commit(), the transaction rolls back, so an early return with ? undoes the work.

place_order_tx takes &mut PgConnection, so the caller decides the transaction scope. A query run on pool instead of the transaction uses another connection, does not see uncommitted changes, and can exhaust a small pool. The stock check is one atomic UPDATE: its WHERE on_hand >= $2 clause decides, and rows_affected() == 0 means there was not enough stock.

src/shop.rs

Rust
use sqlx::{PgConnection, PgPool};
use thiserror::Error;

#[derive(Debug, Error)]
pub enum ShopError {
    #[error("out of stock")]
    OutOfStock,
    #[error("database error")]
    Db(#[from] sqlx::Error),
}

/// Subtracts stock with one atomic statement. The WHERE clause is the stock check.
pub async fn reserve_stock(conn: &mut PgConnection, product_id: i64, quantity: i32) -> Result<(), ShopError> {
    let result = sqlx::query(
        "UPDATE inventory SET on_hand = on_hand - $2 WHERE product_id = $1 AND on_hand >= $2",
    )
    .bind(product_id)
    .bind(quantity)
    .execute(conn)
    .await?;

    if result.rows_affected() == 0 {
        return Err(ShopError::OutOfStock);
    }
    Ok(())
}

/// Runs on a connection that the caller owns, so the caller decides where the transaction starts and ends.
pub async fn place_order_tx(
    conn: &mut PgConnection,
    customer_id: i64,
    product_id: i64,
    quantity: i32,
) -> Result<i64, ShopError> {
    let order_id: i64 = sqlx::query_scalar("INSERT INTO orders (customer_id) VALUES ($1) RETURNING id")
        .bind(customer_id)
        .fetch_one(&mut *conn)
        .await?;

    sqlx::query("INSERT INTO order_lines (order_id, product_id, quantity) VALUES ($1, $2, $3)")
        .bind(order_id)
        .bind(product_id)
        .bind(quantity)
        .execute(&mut *conn)
        .await?;

    reserve_stock(&mut *conn, product_id, quantity).await?;
    Ok(order_id)
}

pub async fn place_order(pool: &PgPool, customer_id: i64, product_id: i64, quantity: i32) -> Result<i64, ShopError> {
    let mut tx = pool.begin().await?;
    let order_id = place_order_tx(&mut *tx, customer_id, product_id, quantity).await?;
    tx.commit().await?;
    Ok(order_id)
}

src/main.rs

Rust
mod shop;

use shop::{place_order, ShopError};
use sqlx::postgres::PgPoolOptions;

fn label(result: &Result<i64, ShopError>) -> &'static str {
    match result {
        Ok(_) => "ok",
        Err(ShopError::OutOfStock) => "out of stock",
        Err(ShopError::Db(_)) => "database error",
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = PgPoolOptions::new()
        .max_connections(10)
        .connect(&std::env::var("DATABASE_URL")?)
        .await?;
    sqlx::migrate!("./migrations").run(&pool).await?;

    let first = place_order(&pool, 7, 1, 3).await;
    println!("first order: {}", label(&first));
    let second = place_order(&pool, 7, 1, 3).await;
    println!("second order: {}", label(&second));

    let orders: i64 = sqlx::query_scalar("SELECT count(*) FROM orders").fetch_one(&pool).await?;
    let on_hand: i32 = sqlx::query_scalar("SELECT on_hand FROM inventory WHERE product_id = 1")
        .fetch_one(&pool)
        .await?;
    println!("orders stored: {orders}");
    println!("on hand: {on_hand}");
    Ok(())
}
Output
first order: ok
second order: out of stock
orders stored: 1
on hand: 2

The second call inserted an order row before the stock check failed. Dropping the transaction rolled that row back, so only one order exists.

Keep transactions short. A transaction holds its connection and its locks until it ends, so it contains no HTTP calls or other waits on remote services.

Example 2: Concurrent orders

Reading the stock, comparing it in Rust, and writing the new value loses updates. Two requests read 1, both decide one unit is available, and both write 0. The atomic UPDATE avoids this. An UPDATE locks the row it changes, so a second concurrent UPDATE on that row waits for the first transaction to end and then re-checks its WHERE clause against the new value.

This program starts 12 concurrent orders of one unit each against a stock of 5. Replace src/main.rs.

src/main.rs

Rust
mod shop;

use shop::{place_order, ShopError};
use sqlx::postgres::PgPoolOptions;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = PgPoolOptions::new()
        .max_connections(10)
        .connect(&std::env::var("DATABASE_URL")?)
        .await?;
    sqlx::migrate!("./migrations").run(&pool).await?;

    let mut handles = Vec::new();
    for customer in 1..=12 {
        let pool = pool.clone();
        handles.push(tokio::spawn(async move { place_order(&pool, customer, 1, 1).await }));
    }

    let mut placed = 0;
    let mut out_of_stock = 0;
    for handle in handles {
        match handle.await? {
            Ok(_) => placed += 1,
            Err(ShopError::OutOfStock) => out_of_stock += 1,
            Err(other) => return Err(other.into()),
        }
    }

    let orders: i64 = sqlx::query_scalar("SELECT count(*) FROM orders").fetch_one(&pool).await?;
    let on_hand: i32 = sqlx::query_scalar("SELECT on_hand FROM inventory WHERE product_id = 1")
        .fetch_one(&pool)
        .await?;
    println!("placed: {placed}");
    println!("out of stock: {out_of_stock}");
    println!("orders stored: {orders}");
    println!("on hand: {on_hand}");
    Ok(())
}
Output
placed: 5
out of stock: 7
orders stored: 5
on hand: 0

The counts are the same on every run, although the order in which tasks finish is not.

When a decision needs several reads, SELECT ... FOR UPDATE locks the rows first:

  • FOR UPDATE takes an exclusive row lock and blocks other writers and lockers.
  • FOR SHARE blocks writers and allows other FOR SHARE locks.
  • FOR UPDATE NOWAIT fails at once when the row is locked.
  • FOR UPDATE SKIP LOCKED skips locked rows. Work queues and the outbox relay use it.

Isolation levels set what a transaction sees while others run. The level is set with SET TRANSACTION ISOLATION LEVEL ... as the first statement of the transaction.

Level A statement sees Consequence
Read Committed (default) Rows committed before the statement started Non-repeatable reads and read-modify-write races are possible
Repeatable Read A snapshot from the first statement Conflicting updates abort with SQLSTATE 40001
Serializable Behaves as if transactions ran one at a time Conflicts abort with 40001

PostgreSQL treats READ UNCOMMITTED as READ COMMITTED. Higher levels turn wrong data into failed transactions, and the caller retries the whole transaction. Storefront code stays at Read Committed and uses atomic statements and explicit locks where a race matters.

Example 3: Deadlocks and retries

A deadlock is a cycle of lock waits. Transaction A locks product 1 and wants product 2, while transaction B locks product 2 and wants product 1. PostgreSQL detects the cycle after deadlock_timeout (1 second by default), aborts one transaction, and reports SQLSTATE 40P01.

A fixed lock order prevents the cycle. reserve_all sorts lines by product id, so overlapping orders both lock the lower id first and one waits for the other. Two more rules apply: keep transactions short, and set lock_timeout (SET LOCAL lock_timeout = '2s') on statements that must not wait long, so a stuck lock returns an error and does not exhaust the pool.

src/multi.rs

Rust
use sqlx::PgPool;

use crate::shop::{reserve_stock, ShopError};

/// Reserves several products in one transaction. Rows are locked in product id order,
/// so two callers with overlapping products never wait on each other in a cycle.
pub async fn reserve_all(pool: &PgPool, lines: &[(i64, i32)]) -> Result<(), ShopError> {
    let mut sorted = lines.to_vec();
    sorted.sort_by_key(|(product_id, _)| *product_id);

    let mut tx = pool.begin().await?;
    for (product_id, quantity) in sorted {
        reserve_stock(&mut *tx, product_id, quantity).await?;
    }
    tx.commit().await?;
    Ok(())
}

Deadlocks and serialization failures still happen under load. Both mean PostgreSQL rolled the transaction back, so running it again is safe. The retry helper runs an operation again on 40001 or 40P01 and returns every other result immediately. The operation contains only database work, because a retried side effect would repeat.

src/retry.rs

Rust
use std::future::Future;

use crate::shop::ShopError;

/// SQLSTATE 40001 is a serialization failure. 40P01 is a detected deadlock.
/// PostgreSQL rolled the transaction back in both cases, so running it again is safe.
fn is_retryable(err: &ShopError) -> bool {
    match err {
        ShopError::Db(sqlx::Error::Database(db)) => {
            matches!(db.code().as_deref(), Some("40001") | Some("40P01"))
        }
        _ => false,
    }
}

pub async fn with_retry<T, F, Fut>(max_attempts: u32, mut operation: F) -> Result<T, ShopError>
where
    F: FnMut() -> Fut,
    Fut: Future<Output = Result<T, ShopError>>,
{
    let mut attempt = 1;
    loop {
        match operation().await {
            Err(err) if is_retryable(&err) && attempt < max_attempts => attempt += 1,
            other => return other,
        }
    }
}

Both files are used by the routes in the complete example.

Example 4: Optimistic concurrency

A row lock cannot span two HTTP requests. A client reads a product, a person edits a form, and the update arrives minutes later. Optimistic concurrency assumes conflicts are rare and checks at write time. The version column increments on every update, and the update succeeds only when the stored version equals the version the client read.

update_price returns the new version through RETURNING. Zero changed rows means either a missing product or a stale version, and a second query tells which.

src/versioned.rs

Rust
use sqlx::PgPool;

pub enum Update {
    Updated(i64),
    NotFound,
    Conflict,
}

/// Changes the price only when the stored version equals the version the client read.
pub async fn update_price(
    pool: &PgPool,
    id: i64,
    expected_version: i64,
    price_cents: i32,
) -> Result<Update, sqlx::Error> {
    let new_version: Option<i64> = sqlx::query_scalar(
        "UPDATE products SET price_cents = $3, version = version + 1
         WHERE id = $1 AND version = $2
         RETURNING version",
    )
    .bind(id)
    .bind(expected_version)
    .bind(price_cents)
    .fetch_optional(pool)
    .await?;

    if let Some(version) = new_version {
        return Ok(Update::Updated(version));
    }

    // Zero rows changed: either the product is missing or its version moved on.
    let exists: bool = sqlx::query_scalar("SELECT EXISTS (SELECT 1 FROM products WHERE id = $1)")
        .bind(id)
        .fetch_one(pool)
        .await?;
    Ok(if exists { Update::Conflict } else { Update::NotFound })
}

HTTP carries the version as an ETag header on GET and an If-Match header on PUT. A mismatch returns 412 (Precondition Failed), and the client reads again, merges, and retries.

  • Pessimistic locking (FOR UPDATE) fits short server-side critical sections and high contention. Waiters queue, and deadlocks are possible.
  • Optimistic checks fit long client-side edits and low contention. The later writer fails and retries.

Example 5: Idempotency keys and the outbox

A client that sends POST /orders and hits a timeout cannot tell whether the order exists. It retries, and a retry can create a second order. An idempotency key makes the retry safe. The client generates one value per logical operation and sends it in an Idempotency-Key header. The server stores the result under that key and returns it for repeats.

The key claim is an INSERT ... ON CONFLICT (key) DO NOTHING in the same transaction as the work. The primary key makes the claim atomic. A concurrent request with the same key waits on that row until the first transaction ends. If the first commits, the second finds the stored response. If the first rolls back, the second inserts the key and does the work itself. A stored fingerprint of the request rejects a key reused with a different body.

The outbox handles a second problem. A handler that commits and then publishes to a broker can crash between the two steps, so the order exists and no message is sent. Publishing before the commit has the opposite failure. The database and the broker cannot share one transaction. The outbox pattern writes the message to an outbox table in the same transaction as the business change. A relay reads unpublished rows and publishes them.

enqueue runs inside the caller's transaction. relay_once claims rows with FOR UPDATE SKIP LOCKED, so several relays run without taking the same rows, and it prints each message in place of a broker client.

src/outbox.rs

Rust
use serde_json::Value;
use sqlx::{PgConnection, PgPool};

/// Stores a message in the caller's transaction. It becomes visible only if that transaction commits.
pub async fn enqueue(conn: &mut PgConnection, topic: &str, payload: &Value) -> Result<(), sqlx::Error> {
    sqlx::query("INSERT INTO outbox (topic, payload) VALUES ($1, $2)")
        .bind(topic)
        .bind(payload)
        .execute(conn)
        .await?;
    Ok(())
}

/// Claims up to `batch` unpublished rows, publishes them, and marks them published.
/// SKIP LOCKED lets several relays run at once without taking the same rows.
pub async fn relay_once(pool: &PgPool, batch: i64) -> Result<usize, sqlx::Error> {
    let mut tx = pool.begin().await?;

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

    for (id, topic, payload) in &rows {
        // Stand-in for a broker client. A failure here rolls the batch back.
        println!("publish {topic} {payload}");
        sqlx::query("UPDATE outbox SET published_at = now() WHERE id = $1")
            .bind(id)
            .execute(&mut *tx)
            .await?;
    }

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

src/idempotent.rs

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

use crate::outbox;
use crate::shop::{place_order_tx, ShopError};

pub enum Outcome {
    Created(Value),
    Replayed(Value),
    KeyReused,
}

/// Creates an order once per idempotency key. The key claim, the order, the outbox message,
/// and the stored response commit together or not at all.
pub async fn create_order(
    pool: &PgPool,
    key: &str,
    fingerprint: &str,
    customer_id: i64,
    product_id: i64,
    quantity: i32,
) -> Result<Outcome, ShopError> {
    let mut tx = pool.begin().await?;

    let claimed = sqlx::query(
        "INSERT INTO idempotency_keys (key, fingerprint) VALUES ($1, $2) ON CONFLICT (key) DO NOTHING",
    )
    .bind(key)
    .bind(fingerprint)
    .execute(&mut *tx)
    .await?
    .rows_affected()
        == 1;

    if !claimed {
        let stored: (String, Option<Value>) =
            sqlx::query_as("SELECT fingerprint, response_body FROM idempotency_keys WHERE key = $1")
                .bind(key)
                .fetch_one(&mut *tx)
                .await?;
        return Ok(if stored.0 != fingerprint {
            Outcome::KeyReused
        } else {
            Outcome::Replayed(stored.1.unwrap_or(Value::Null))
        });
    }

    let order_id = place_order_tx(&mut *tx, customer_id, product_id, quantity).await?;
    let body = json!({ "order_id": order_id });

    outbox::enqueue(
        &mut *tx,
        "order.placed",
        &json!({ "order_id": order_id, "customer_id": customer_id }),
    )
    .await?;

    sqlx::query("UPDATE idempotency_keys SET response_body = $2 WHERE key = $1")
        .bind(key)
        .bind(&body)
        .execute(&mut *tx)
        .await?;

    tx.commit().await?;
    Ok(Outcome::Created(body))
}
  • The relay holds a connection and row locks while it publishes, so batches stay small and the publisher needs its own timeout. An alternative claims rows in one short transaction, publishes outside it, and marks them published in a second one.
  • A crash after publishing and before commit sends the message again. The outbox gives at-least-once delivery, so consumers deduplicate by message id.
  • Ordering holds within one relay batch but not across concurrent relays. Consumers that need order per entity use a message key, as in the messaging lesson. The saga lesson builds on the outbox.
  • Keys are deleted after a retention period, such as 24 hours.
  • A unique constraint on a natural key, such as (customer_id, cart_id), is another form of idempotency.

Complete example

The service exposes idempotent order creation, versioned price updates, multi-product reservations with retries, and a route that runs one outbox relay batch.

Text
storefront-transactions/
  Cargo.toml
  migrations/
    20260929120000_storefront.sql
  src/
    main.rs
    shop.rs
    retry.rs
    multi.rs
    versioned.rs
    outbox.rs
    idempotent.rs

Cargo.toml

TOML
[package]
name = "storefront-transactions"
version = "0.1.0"
edition = "2021"

[dependencies]
axum = "0.8.4"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
sqlx = { version = "0.8", features = ["runtime-tokio", "postgres", "migrate", "macros", "derive", "json"] }
thiserror = "2"
tokio = { version = "1.44", features = ["full"] }

migrations/20260929120000_storefront.sql

SQL
CREATE TABLE products (
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    sku         TEXT    NOT NULL UNIQUE,
    name        TEXT    NOT NULL,
    price_cents INTEGER NOT NULL CHECK (price_cents > 0),
    version     BIGINT  NOT NULL DEFAULT 1
);

CREATE TABLE inventory (
    product_id BIGINT  PRIMARY KEY REFERENCES products (id),
    on_hand    INTEGER NOT NULL CHECK (on_hand >= 0)
);

CREATE TABLE orders (
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    customer_id BIGINT      NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE order_lines (
    order_id   BIGINT  NOT NULL REFERENCES orders (id),
    product_id BIGINT  NOT NULL REFERENCES products (id),
    quantity   INTEGER NOT NULL CHECK (quantity > 0),
    PRIMARY KEY (order_id, product_id)
);

CREATE TABLE idempotency_keys (
    key           TEXT        PRIMARY KEY,
    fingerprint   TEXT        NOT NULL,
    response_body JSONB,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE outbox (
    id           BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    topic        TEXT        NOT NULL,
    payload      JSONB       NOT NULL,
    created_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
    published_at TIMESTAMPTZ
);

CREATE INDEX outbox_unpublished_idx ON outbox (id) WHERE published_at IS NULL;

INSERT INTO products (sku, name, price_cents) VALUES ('NB-1', 'Notebook', 450), ('PN-1', 'Pen', 150);
INSERT INTO inventory (product_id, on_hand) VALUES (1, 5), (2, 10);

src/shop.rs

Rust
use sqlx::{PgConnection, PgPool};
use thiserror::Error;

#[derive(Debug, Error)]
pub enum ShopError {
    #[error("out of stock")]
    OutOfStock,
    #[error("database error")]
    Db(#[from] sqlx::Error),
}

/// Subtracts stock with one atomic statement. The WHERE clause is the stock check.
pub async fn reserve_stock(conn: &mut PgConnection, product_id: i64, quantity: i32) -> Result<(), ShopError> {
    let result = sqlx::query(
        "UPDATE inventory SET on_hand = on_hand - $2 WHERE product_id = $1 AND on_hand >= $2",
    )
    .bind(product_id)
    .bind(quantity)
    .execute(conn)
    .await?;

    if result.rows_affected() == 0 {
        return Err(ShopError::OutOfStock);
    }
    Ok(())
}

/// Runs on a connection that the caller owns, so the caller decides where the transaction starts and ends.
pub async fn place_order_tx(
    conn: &mut PgConnection,
    customer_id: i64,
    product_id: i64,
    quantity: i32,
) -> Result<i64, ShopError> {
    let order_id: i64 = sqlx::query_scalar("INSERT INTO orders (customer_id) VALUES ($1) RETURNING id")
        .bind(customer_id)
        .fetch_one(&mut *conn)
        .await?;

    sqlx::query("INSERT INTO order_lines (order_id, product_id, quantity) VALUES ($1, $2, $3)")
        .bind(order_id)
        .bind(product_id)
        .bind(quantity)
        .execute(&mut *conn)
        .await?;

    reserve_stock(&mut *conn, product_id, quantity).await?;
    Ok(order_id)
}

pub async fn place_order(pool: &PgPool, customer_id: i64, product_id: i64, quantity: i32) -> Result<i64, ShopError> {
    let mut tx = pool.begin().await?;
    let order_id = place_order_tx(&mut *tx, customer_id, product_id, quantity).await?;
    tx.commit().await?;
    Ok(order_id)
}

src/retry.rs

Rust
use std::future::Future;

use crate::shop::ShopError;

/// SQLSTATE 40001 is a serialization failure. 40P01 is a detected deadlock.
/// PostgreSQL rolled the transaction back in both cases, so running it again is safe.
fn is_retryable(err: &ShopError) -> bool {
    match err {
        ShopError::Db(sqlx::Error::Database(db)) => {
            matches!(db.code().as_deref(), Some("40001") | Some("40P01"))
        }
        _ => false,
    }
}

pub async fn with_retry<T, F, Fut>(max_attempts: u32, mut operation: F) -> Result<T, ShopError>
where
    F: FnMut() -> Fut,
    Fut: Future<Output = Result<T, ShopError>>,
{
    let mut attempt = 1;
    loop {
        match operation().await {
            Err(err) if is_retryable(&err) && attempt < max_attempts => attempt += 1,
            other => return other,
        }
    }
}

src/multi.rs

Rust
use sqlx::PgPool;

use crate::shop::{reserve_stock, ShopError};

/// Reserves several products in one transaction. Rows are locked in product id order,
/// so two callers with overlapping products never wait on each other in a cycle.
pub async fn reserve_all(pool: &PgPool, lines: &[(i64, i32)]) -> Result<(), ShopError> {
    let mut sorted = lines.to_vec();
    sorted.sort_by_key(|(product_id, _)| *product_id);

    let mut tx = pool.begin().await?;
    for (product_id, quantity) in sorted {
        reserve_stock(&mut *tx, product_id, quantity).await?;
    }
    tx.commit().await?;
    Ok(())
}

src/versioned.rs

Rust
use sqlx::PgPool;

pub enum Update {
    Updated(i64),
    NotFound,
    Conflict,
}

/// Changes the price only when the stored version equals the version the client read.
pub async fn update_price(
    pool: &PgPool,
    id: i64,
    expected_version: i64,
    price_cents: i32,
) -> Result<Update, sqlx::Error> {
    let new_version: Option<i64> = sqlx::query_scalar(
        "UPDATE products SET price_cents = $3, version = version + 1
         WHERE id = $1 AND version = $2
         RETURNING version",
    )
    .bind(id)
    .bind(expected_version)
    .bind(price_cents)
    .fetch_optional(pool)
    .await?;

    if let Some(version) = new_version {
        return Ok(Update::Updated(version));
    }

    // Zero rows changed: either the product is missing or its version moved on.
    let exists: bool = sqlx::query_scalar("SELECT EXISTS (SELECT 1 FROM products WHERE id = $1)")
        .bind(id)
        .fetch_one(pool)
        .await?;
    Ok(if exists { Update::Conflict } else { Update::NotFound })
}

src/outbox.rs

Rust
use serde_json::Value;
use sqlx::{PgConnection, PgPool};

/// Stores a message in the caller's transaction. It becomes visible only if that transaction commits.
pub async fn enqueue(conn: &mut PgConnection, topic: &str, payload: &Value) -> Result<(), sqlx::Error> {
    sqlx::query("INSERT INTO outbox (topic, payload) VALUES ($1, $2)")
        .bind(topic)
        .bind(payload)
        .execute(conn)
        .await?;
    Ok(())
}

/// Claims up to `batch` unpublished rows, publishes them, and marks them published.
/// SKIP LOCKED lets several relays run at once without taking the same rows.
pub async fn relay_once(pool: &PgPool, batch: i64) -> Result<usize, sqlx::Error> {
    let mut tx = pool.begin().await?;

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

    for (id, topic, payload) in &rows {
        // Stand-in for a broker client. A failure here rolls the batch back.
        println!("publish {topic} {payload}");
        sqlx::query("UPDATE outbox SET published_at = now() WHERE id = $1")
            .bind(id)
            .execute(&mut *tx)
            .await?;
    }

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

src/idempotent.rs

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

use crate::outbox;
use crate::shop::{place_order_tx, ShopError};

pub enum Outcome {
    Created(Value),
    Replayed(Value),
    KeyReused,
}

/// Creates an order once per idempotency key. The key claim, the order, the outbox message,
/// and the stored response commit together or not at all.
pub async fn create_order(
    pool: &PgPool,
    key: &str,
    fingerprint: &str,
    customer_id: i64,
    product_id: i64,
    quantity: i32,
) -> Result<Outcome, ShopError> {
    let mut tx = pool.begin().await?;

    let claimed = sqlx::query(
        "INSERT INTO idempotency_keys (key, fingerprint) VALUES ($1, $2) ON CONFLICT (key) DO NOTHING",
    )
    .bind(key)
    .bind(fingerprint)
    .execute(&mut *tx)
    .await?
    .rows_affected()
        == 1;

    if !claimed {
        let stored: (String, Option<Value>) =
            sqlx::query_as("SELECT fingerprint, response_body FROM idempotency_keys WHERE key = $1")
                .bind(key)
                .fetch_one(&mut *tx)
                .await?;
        return Ok(if stored.0 != fingerprint {
            Outcome::KeyReused
        } else {
            Outcome::Replayed(stored.1.unwrap_or(Value::Null))
        });
    }

    let order_id = place_order_tx(&mut *tx, customer_id, product_id, quantity).await?;
    let body = json!({ "order_id": order_id });

    outbox::enqueue(
        &mut *tx,
        "order.placed",
        &json!({ "order_id": order_id, "customer_id": customer_id }),
    )
    .await?;

    sqlx::query("UPDATE idempotency_keys SET response_body = $2 WHERE key = $1")
        .bind(key)
        .bind(&body)
        .execute(&mut *tx)
        .await?;

    tx.commit().await?;
    Ok(Outcome::Created(body))
}

src/main.rs

Rust
mod idempotent;
mod multi;
mod outbox;
mod retry;
mod shop;
mod versioned;

use axum::{
    extract::{Path, State},
    http::{header, HeaderMap, StatusCode},
    response::{IntoResponse, Response},
    routing::{get, post},
    Json, Router,
};
use idempotent::Outcome;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use shop::ShopError;
use sqlx::{postgres::PgPoolOptions, PgPool};
use versioned::Update;

#[derive(Clone)]
struct AppState {
    pool: PgPool,
}

struct ApiError(StatusCode, &'static str);

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

impl From<ShopError> for ApiError {
    fn from(err: ShopError) -> Self {
        match err {
            ShopError::OutOfStock => ApiError(StatusCode::CONFLICT, "out of stock"),
            ShopError::Db(e) => {
                eprintln!("database error: {e}");
                ApiError(StatusCode::INTERNAL_SERVER_ERROR, "internal error")
            }
        }
    }
}

impl From<sqlx::Error> for ApiError {
    fn from(err: sqlx::Error) -> Self {
        ShopError::Db(err).into()
    }
}

#[derive(Deserialize)]
struct NewOrder {
    customer_id: i64,
    product_id: i64,
    quantity: i32,
}

async fn create_order(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(input): Json<NewOrder>,
) -> Result<Response, ApiError> {
    let key = headers
        .get("idempotency-key")
        .and_then(|v| v.to_str().ok())
        .ok_or(ApiError(StatusCode::BAD_REQUEST, "Idempotency-Key header is required"))?;
    if input.quantity <= 0 {
        return Err(ApiError(StatusCode::UNPROCESSABLE_ENTITY, "quantity must be greater than 0"));
    }
    let fingerprint = format!("{}:{}:{}", input.customer_id, input.product_id, input.quantity);

    let outcome = retry::with_retry(3, || {
        idempotent::create_order(&state.pool, key, &fingerprint, input.customer_id, input.product_id, input.quantity)
    })
    .await?;

    Ok(match outcome {
        Outcome::Created(body) => (StatusCode::CREATED, Json(body)).into_response(),
        Outcome::Replayed(body) => (StatusCode::OK, Json(body)).into_response(),
        Outcome::KeyReused => ApiError(
            StatusCode::UNPROCESSABLE_ENTITY,
            "idempotency key was used with a different request",
        )
        .into_response(),
    })
}

#[derive(Serialize, sqlx::FromRow)]
struct Product {
    id: i64,
    name: String,
    price_cents: i32,
    version: i64,
}

async fn get_product(State(state): State<AppState>, Path(id): Path<i64>) -> Result<Response, ApiError> {
    let product: Option<Product> =
        sqlx::query_as("SELECT id, name, price_cents, version FROM products WHERE id = $1")
            .bind(id)
            .fetch_optional(&state.pool)
            .await?;
    let product = product.ok_or(ApiError(StatusCode::NOT_FOUND, "product not found"))?;
    let etag = format!("\"{}\"", product.version);
    Ok(([(header::ETAG, etag)], Json(product)).into_response())
}

#[derive(Deserialize)]
struct PriceInput {
    price_cents: i32,
}

async fn put_price(
    State(state): State<AppState>,
    Path(id): Path<i64>,
    headers: HeaderMap,
    Json(input): Json<PriceInput>,
) -> Result<Response, ApiError> {
    let expected: i64 = headers
        .get(header::IF_MATCH)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.trim_matches('"').parse().ok())
        .ok_or(ApiError(StatusCode::PRECONDITION_REQUIRED, "If-Match header is required"))?;
    if input.price_cents <= 0 {
        return Err(ApiError(StatusCode::UNPROCESSABLE_ENTITY, "price_cents must be greater than 0"));
    }

    match versioned::update_price(&state.pool, id, expected, input.price_cents).await? {
        Update::Updated(version) => {
            let etag = format!("\"{version}\"");
            Ok(([(header::ETAG, etag)], Json(json!({ "version": version }))).into_response())
        }
        Update::Conflict => Err(ApiError(StatusCode::PRECONDITION_FAILED, "the product changed since it was read")),
        Update::NotFound => Err(ApiError(StatusCode::NOT_FOUND, "product not found")),
    }
}

#[derive(Deserialize)]
struct ReservationLine {
    product_id: i64,
    quantity: i32,
}

#[derive(Deserialize)]
struct Reservation {
    lines: Vec<ReservationLine>,
}

async fn reserve(State(state): State<AppState>, Json(input): Json<Reservation>) -> Result<Json<Value>, ApiError> {
    let lines: Vec<(i64, i32)> = input.lines.iter().map(|l| (l.product_id, l.quantity)).collect();
    retry::with_retry(3, || multi::reserve_all(&state.pool, &lines)).await?;
    Ok(Json(json!({ "reserved": lines.len() })))
}

async fn relay(State(state): State<AppState>) -> Result<Json<Value>, ApiError> {
    let published = outbox::relay_once(&state.pool, 100).await?;
    Ok(Json(json!({ "published": published })))
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let pool = PgPoolOptions::new()
        .max_connections(10)
        .connect(&std::env::var("DATABASE_URL")?)
        .await?;
    sqlx::migrate!("./migrations").run(&pool).await?;

    let app = Router::new()
        .route("/orders", post(create_order))
        .route("/products/{id}", get(get_product).put(put_price))
        .route("/reservations", post(reserve))
        .route("/outbox/relay", post(relay))
        .with_state(AppState { pool });

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

Start a fresh database, wait until it accepts connections, and run the service.

Terminal
docker rm -f storefront-pg
docker run --name storefront-pg -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=storefront -p 5432:5432 -d postgres:16
export DATABASE_URL=postgres://postgres:postgres@localhost:5432/storefront
cargo run

Place an order, repeat it with the same key, reuse the key with a different quantity, and order more than the remaining stock (5 minus 2 leaves 3).

Terminal
curl -s -w '\n%{http_code}\n' -X POST localhost:3000/orders -H 'content-type: application/json' -H 'Idempotency-Key: k1' -d '{"customer_id":7,"product_id":1,"quantity":2}'
curl -s -w '\n%{http_code}\n' -X POST localhost:3000/orders -H 'content-type: application/json' -H 'Idempotency-Key: k1' -d '{"customer_id":7,"product_id":1,"quantity":2}'
curl -s -w '\n%{http_code}\n' -X POST localhost:3000/orders -H 'content-type: application/json' -H 'Idempotency-Key: k1' -d '{"customer_id":7,"product_id":1,"quantity":3}'
curl -s -w '\n%{http_code}\n' -X POST localhost:3000/orders -H 'content-type: application/json' -H 'Idempotency-Key: k2' -d '{"customer_id":7,"product_id":1,"quantity":4}'
Output
{"order_id":1}
201
{"order_id":1}
200
{"error":"idempotency key was used with a different request"}
422
{"error":"out of stock"}
409

Run the relay twice. The server prints publish order.placed {"customer_id":7,"order_id":1} on the first call only.

Terminal
curl -s -X POST localhost:3000/outbox/relay
curl -s -X POST localhost:3000/outbox/relay
Output
{"published":1}
{"published":0}

Read a product, update its price with the version from the ETag, and repeat the update with the old version.

Terminal
curl -s localhost:3000/products/1
curl -s -X PUT localhost:3000/products/1 -H 'content-type: application/json' -H 'If-Match: "1"' -d '{"price_cents":500}'
curl -s -w '\n%{http_code}\n' -X PUT localhost:3000/products/1 -H 'content-type: application/json' -H 'If-Match: "1"' -d '{"price_cents":600}'
Output
{"id":1,"name":"Notebook","price_cents":450,"version":1}
{"version":2}
{"error":"the product changed since it was read"}
412

The GET response carries the header ETag: "1". Reserve two products in one transaction.

Terminal
curl -s -X POST localhost:3000/reservations -H 'content-type: application/json' -d '{"lines":[{"product_id":2,"quantity":1},{"product_id":1,"quantity":1}]}'
Output
{"reserved":2}

Summary

  • A dropped SQLx Transaction rolls back, and commit() must be called. Queries run on &mut *tx, not on the pool.
  • Transactions stay short and free of remote calls, because they hold a pooled connection and locks.
  • PostgreSQL defaults to Read Committed. Repeatable Read and Serializable can abort with 40001, and the caller retries the whole transaction.
  • Read-modify-write in application code loses updates. One atomic UPDATE ... WHERE statement or SELECT ... FOR UPDATE prevents it.
  • A fixed lock order prevents deadlocks, and a retry on 40P01 handles those that remain.
  • A version column with UPDATE ... WHERE version = $n detects concurrent edits. HTTP maps it to ETag, If-Match, and 412.
  • An idempotency key claimed in the same transaction as the work makes retried POST requests safe, and a stored fingerprint detects key reuse.
  • The outbox stores a message in the same transaction as the change and a relay publishes it with at-least-once delivery, so consumers deduplicate.