Skip to main content

Tracking what you publish: the outbox plugin

This guide makes every message a service publishes reach a handler that processes it, on a broker that keeps nothing.

Some brokers deliver a message only to whoever listens at that moment, Redis Pub/Sub and NATS core among them, and a consumer may stop halfway through any message. The outbox records each message the service publishes under a registered name, marks the record once a consumer processed the message, and publishes the unprocessed records again at startup. It is a plugin over any RustStream broker and needs no relay process. Its outbox feature is independent of the inbox.

The example takes payments at checkout and ships the paid orders, over the in-memory broker.

The table​

src/main.rs
const SCHEMA: &str = "CREATE TABLE IF NOT EXISTS outbox (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    payload BLOB NOT NULL,
    processed_at TEXT
)";

The table is the service's: it creates it, with its own migrations in production. processed_at stays empty until a consumer processed the message.

The record​

src/main.rs
use ruststream_sqlx::{Outbox, outbox};

/// One record of the `outbox` table.
#[derive(Outbox, sqlx::FromRow)]
#[outbox(table = "outbox")]
struct PaymentOutbox {
    #[field(id)]
    id: i64,
    #[field(name)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
}

/// The service writes the record and returns its id; the outbox sends the id with the message.
impl outbox::Publish<Sqlite> for PaymentOutbox {
    async fn publish(conn: &mut SqliteConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
        sqlx::query_scalar("INSERT INTO outbox (name, payload) VALUES (?, ?) RETURNING id")
            .bind(msg.name())
            .bind(msg.payload())
            .fetch_one(conn)
            .await
    }
}

#[derive(Outbox)] describes the record: id, name and payload are required, headers and processed_at optional. The service writes the record with its own outbox::Publish and returns its id; the crate builds the other statements. In the manual form, OutboxSpec says the same, and a TrackedName type names what is tracked.

The handlers​

src/main.rs
use ruststream::memory::prelude::*;

/// The reply to `payments` is recorded before it is sent.
#[subscriber("checkout", reply)]
async fn take_payment(checkout: &Checkout) -> PaymentTaken {
    tracing::info!(order = checkout.order, "payment taken");
    PaymentTaken {
        order: checkout.order,
    }
}

/// The acknowledgement marks the payment's record processed.
#[subscriber("payments")]
async fn ship(payment: &PaymentTaken) -> HandlerOutcome {
    tracing::info!(order = payment.order, "shipping");
    HandlerOutcome::ack()
}

The handlers know nothing of the outbox: the middlewares of the app do the tracking.

The app​

src/main.rs
use std::convert::Infallible;
use std::io;

use sqlx::SqlitePool;

fn service(url: String) -> impl App {
    // The registry starts without a pool: sqlx builds one only inside the Tokio runtime.
    let tracking = outbox! { "payments" => PaymentOutbox };
    let registry = tracking.clone();

    RustStream::new(AppInfo::new("payments", "0.1.0"))
        .on_startup(async move |()| {
            let pool = SqlitePool::connect(&url).await.map_err(io::Error::other)?;
            sqlx::raw_sql(SCHEMA)
                .execute(&pool)
                .await
                .map_err(io::Error::other)?;
            registry.set_pool(pool.clone()).map_err(io::Error::other)?;
            Ok::<_, io::Error>(pool)
        })
        .layer(tracking.layer())
        .publish_layer(tracking.publish_layer())
        .after_shutdown(async move |pool| {
            pool.close().await;
            Ok::<_, Infallible>(())
        })
        .with_broker(MemoryBroker::new(), |b| {
            b.include(take_payment).out_reply(Publish);
            b.include(ship);
            // Once the subscriptions are open, the unprocessed records are sent again.
            b.after_startup(Publish, tracking.republish());
        })
}

#[ruststream::app]
fn app() -> impl App {
    service("sqlite:payments.db?mode=rwc".to_owned())
}

outbox! registers payments with its record; a name registered twice does not compile. #[ruststream::app] builds the app before the Tokio runtime exists, and sqlx builds a pool only inside one. So the registry starts without a pool, and on_startup builds the pool and hands it over with set_pool; after_shutdown closes it.

A tracked message goes through four steps:

  1. The publish layer writes the record, then sends the message with the record's id in the x-ruststream-outbox-id header. A record that cannot be written fails the publish.
  2. The subscription layer takes a delivery that carries an id into work by that id and runs the handler; a copy whose record is processed already is acknowledged without the handler.
  3. The handler's acknowledgement or drop marks the record processed; a retry leaves it as it was.
  4. At startup, republish sends every unprocessed record again, once the subscriptions are open.

The consumer marks the record in the producer's table, so producer and consumer share the registry and the database, as both handlers in one app do here.

What the outbox guarantees​

A message is delivered at least once. The record is written before the message is sent, so a failed send, a consumer that stops mid-handler, or a handler that asks for the message again leaves the record for the next startup. Two instances that start together may both send the same record again, so a consumer that must not process a message twice checks that itself.

The record is written on a connection of its own and commits at once. It does not join a transaction of the service: a transactional inbox handler that is retried discards its own writes and keeps the record of what it published. On SQLite such a handler publishes before its first write through its transaction, because the record's insert needs that transaction's write lock; a publish after that write fails with OutboxError::WriteLockHeld (Interplay). An event that must commit with the business rows is a row of the business transaction, relayed afterwards: Relaying events to another broker.

A publish outside a handler, an HTTP endpoint for one, takes the running app's publisher and wraps it in the registry, so it is tracked the same way. The tracking costs a connection for each tracked publish, and two for each tracked delivery: one for the fetch, one for the settlement. A message under a name the registry does not hold costs a comparison of names. The repository's examples/outbox runs the same flow over Redis Pub/Sub, with migrations and a fetch of its own.

Testing​

A test build leaves the outbox off unless RUSTSTREAM_SQLX_OUTBOX=on turns it on (Switch), so a service's own tests run its handlers untracked. The tests in the example file that check the tracking skip without the switch; one more test checks that the service runs untracked with it off:

cargo test
RUSTSTREAM_SQLX_OUTBOX=on cargo test

With the switch on, one test checks that a payment is recorded, sent with its record's id, and marked once shipped. Another writes a record an earlier instance left unprocessed and finds it shipped after the startup. Both run with TestApp::start on a SQLite database in memory, which the service's own pool opens by its URL.