Skip to main content

Relaying events to another broker

This guide writes an event in the same transaction as the data it describes, then publishes it to another broker.

A service that saves an order and then publishes "order placed" to Kafka has two writes that can part. The order can commit while the publish fails, or the publish can go out for an order that rolled back. The transactional outbox joins them: the event is a row written in the order's transaction, and a relay publishes the rows that committed. With this crate the event table is a queue table, and the relay is an ordinary subscription.

The example is a shop. A checkout places the order and writes its event; the relay publishes each event to a bus, the in-memory broker here.

The order and its event​

src/main.rs
const SCHEMA: &str = "
CREATE TABLE IF NOT EXISTS orders (id INTEGER PRIMARY KEY, total_cents INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS checkout_jobs (
    id INTEGER PRIMARY KEY,
    locked_until TEXT,
    payload BLOB NOT NULL
);
CREATE TABLE IF NOT EXISTS order_events (
    id INTEGER PRIMARY KEY,
    aggregate TEXT NOT NULL,
    locked_until TEXT,
    payload BLOB NOT NULL
);
";
src/main.rs
use ruststream_sqlx::prelude::*;
use sqlx::Connection;
use sqlx::sqlite::SqlitePoolOptions;
use sqlx::{Sqlite, SqliteConnection};

/// One row of `order_events`: an event waiting for the relay, keyed by its order.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "order_events")]
struct OrderEvent {
    #[field(id, generated)]
    id: i64,
    #[field(partition_key)]
    aggregate: String,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(payload)]
    payload: Vec<u8>,
}

/// The order and its event, in the checkout's transaction: both commit, or neither does.
async fn place(
    conn: &mut SqliteConnection,
    order: &Order,
) -> Result<(), Box<dyn Error + Send + Sync>> {
    sqlx::query("INSERT INTO orders (id, total_cents) VALUES (?, ?)")
        .bind(order.id)
        .bind(order.total_cents)
        .execute(&mut *conn)
        .await?;
    let placed = OrderPlaced {
        order: order.id,
        total_cents: order.total_cents,
    };
    let event = OrderEvent {
        id: 0,
        aggregate: order.id.to_string(),
        locked_until: None,
        payload: serde_json::to_vec(&placed)?,
    };
    event.insert(conn).await?;
    Ok(())
}

#[subscriber(InboxQueue::<Checkout>::new("checkouts"))]
async fn checkout(order: &Order, Ctx(mut tx): Ctx<keys::Tx<Sqlite>>) -> HandlerOutcome {
    match place(&mut tx, order).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(error) => {
            tracing::warn!(subscription = "checkouts", order.id, %error, "order not placed");
            HandlerOutcome::drop()
        }
    }
}

checkout runs in transactional mode: it writes the order and the event through its delivery's transaction. Its acknowledgement commits both with the checkout's settlement. An order whose insert fails rolls its event back, and nothing is published for it. Code outside a handler, an HTTP endpoint for one, writes the same two rows in a sqlx transaction of its own, as Enqueueing jobs does. The event row wakes no subscription in that case either: the relay finds it within its poll interval.

The relay​

src/main.rs
use ruststream::memory::{MemoryBroker, MemoryPublish};

/// Each event goes out on the bus as it was written; the acknowledgement deletes its row.
#[subscriber(InboxQueue::<OrderEvent>::new("order-events"), reply)]
async fn relay(event: &OrderPlaced) -> OrderPlaced {
    event.clone()
}

/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    // The target broker is wrapped first, so its publish token exists before the scopes.
    let bus = MemoryBroker::new().bindable();
    let to_bus = bus.bind(MemoryPublish);
    RustStream::new(AppInfo::new("shop", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<Checkout>("checkouts"), |b| {
            b.include(checkout.transactional());
            b.include(relay.workers_by_key(nonzero!(4)))
                .out_reply(to_bus);
        })
        .with_broker(bus, |_b| {})
}

relay reads order_events and answers each event with itself; the reply type names its destination, orders. The reply leaves through the bus: bindable() wraps the target broker, bind(MemoryPublish) mints a publish token for it, and .out_reply(to_bus) sends the replies of relay through that token. A Kafka or NATS service puts its own broker and publish policy in the same two lines. The reply is published before the event's row settles, and the acknowledgement deletes the row.

The relay is at least once. A reply whose publish fails returns the event to the queue, and a crash after the publish and before the acknowledgement publishes it again. A consumer that must not act twice keeps the event's identity, the order here, and passes over one it has seen.

Order per aggregate​

aggregate is the event's partition key, the order's id. workers_by_key(4) relays four orders at once and keeps the events of one order in the order they were written. That order holds within one relay subscription. A relay on several instances whose events must stay in order per aggregate across all of them keeps them in a FIFO group or in the advisory lock form.

The relay needs no "sent" flag and no polling of its own: the claim is the queue's, the delay after a failed publish is a retry, and a cap with a dead-letter group parks an event the target keeps refusing. Tracking what you publish is the other shape: the outbox plugin records what a handler publishes and marks it once a consumer processed it, but its record commits on its own, outside the business transaction.

Testing​

The tests in the example file publish checkouts through the harness and read what reached the bus with tb.broker::<MemoryBroker>().published(..). One order reaches the bus once, and its event row is gone. The same order checked out twice publishes one event: the second checkout fails, and its event rolled back. The event rows wake no subscription, so the tests run on the real clock with TestApp::start_live past the poll interval.