Transactional outbox

The outbox records each message a service publishes under a registered name until a consumer processed it, and publishes the rest again at startup.

The outbox is a plugin over any RustStream broker and needs no broker of its own. The outbox feature turns it on, beside a driver feature: postgres, mysql, sqlite or any. It is independent of the inbox, and a service may enable both.

A tracked message goes through four steps. The publish middleware writes a record of the message, then sends it with the record's id in the x-ruststream-outbox-id header. The subscription middleware takes a delivery that carries an id into work by that id, and runs the handler. The handler's outcome then settles the record: an acknowledgement marks it processed. At startup the republish sends every unprocessed record again. A message under a name the registry does not hold passes both middlewares after a comparison of names.

A message is delivered at least once. The record is written before the message is sent, so a send that fails, a consumer that stops before its handler finishes, or a handler that asks for the message again leaves the record unprocessed for the next startup. A copy whose record is processed already is acknowledged without running the handler. Two instances of a service that start together may both send the same record again, so a consumer that must not process a message twice checks that itself.

A message whose name is not registered costs one comparison per registered name: no allocation and no database call. A tracked publish takes a connection for the record's insert and allocates the id header's value. A tracked delivery takes a connection for the fetch, returns it before the handler runs, and takes one again for the settlement. Under sqlx's default test_before_acquire, each acquire pings the connection first; a pool built with test_before_acquire(false) skips that round trip. The default retry takes no connection, and the republish allocates its body once per startup.

The repository's examples/outbox runs the same flow over Redis Pub/Sub, with migrations and a fetch of its own.

Example

The service answers a request on checkout with an OrderPlaced on orders, a registered name: the reply is recorded before it is sent, the acknowledgement of fulfil marks the record, and the startup sends what no consumer processed. RUSTSTREAM_SQLX_OUTBOX=on cargo run -p ruststream-sqlx --example sqlx_outbox --features outbox,sqlite,testing starts it; the switch is there because testing is among the example's features.

Macros

examples/sqlx_outbox.rs

use ruststream::memory::prelude::*;
use ruststream_sqlx::{Outbox, outbox};
use sqlx::SqlitePool;

/// One record of the `outbox` table.
#[derive(Outbox, sqlx::FromRow)]
#[outbox(table = "outbox")]
struct OrderOutbox {
    #[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; the outbox sends its id with the message.
impl outbox::Publish<Sqlite> for OrderOutbox {
    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
    }
}

/// The reply to `orders` is recorded before it is sent.
#[subscriber("checkout", reply)]
async fn place(cmd: &PlaceOrder) -> OrderPlaced {
    OrderPlaced { id: cmd.id }
}

/// After the acknowledgement, the record is marked processed.
#[subscriber("orders")]
async fn fulfil(order: &OrderPlaced) -> HandlerOutcome {
    tracing::info!(order = order.id, "fulfilling the order");
    HandlerOutcome::ack()
}

fn service(pool: SqlitePool) -> impl App {
    let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .layer(tracking.layer())
        .publish_layer(tracking.publish_layer())
        .with_broker(MemoryBroker::new(), |b| {
            b.include(place).out_reply(Publish);
            b.include(fulfil);
            // Once the subscriptions are open, the unprocessed records are sent again.
            b.after_startup(Publish, tracking.republish());
        })
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let pool = SqlitePool::connect("sqlite:orders.db?mode=rwc").await?;
    sqlx::raw_sql(SCHEMA).execute(&pool).await?;
    service(pool.clone()).run().await?;
    pool.close().await;
    Ok(())
}

Manual

examples/manual/sqlx_outbox.rs

use ruststream::memory::prelude::*;
use ruststream_sqlx::outbox::{self, Outbox, OutboxSpec, OutboxTable, TrackedName};
use sqlx::SqlitePool;

/// One record of the `outbox` table: the struct holds what the outbox reads.
#[derive(sqlx::FromRow)]
struct OrderOutbox {
    id: i64,
    name: String,
    payload: Vec<u8>,
}

impl OutboxTable for OrderOutbox {
    type Id = i64;
    type Table = OutboxSpec<(ProcessedAt,)>;
    const TABLE: Self::Table = OutboxSpec::new(
        "outbox",
        Column::new("id"),
        Column::new("name"),
        Column::new("payload"),
    )
    .processed_at(Column::new("processed_at"));

    fn id(&self) -> &i64 {
        &self.id
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn payload(&self) -> &[u8] {
        &self.payload
    }
}

/// The service writes the record; the outbox sends its id with the message.
impl outbox::Publish<Sqlite> for OrderOutbox {
    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
    }
}

/// The name `OrderPlaced` is published under.
struct Orders;

impl TrackedName for Orders {
    const NAME: &'static str = "orders";
}

/// The reply to `orders` is recorded before it is sent.
struct Place;

impl Handle<PlaceOrder, OrderPlaced> for Place {
    fn handle(
        &self,
        cmd: &PlaceOrder,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<OrderPlaced, HandlerOutcome>> {
        ready(Ok(OrderPlaced { id: cmd.id }))
    }
}

/// After the acknowledgement, the record is marked processed.
struct Fulfil;

impl Handle<OrderPlaced> for Fulfil {
    fn handle(
        &self,
        order: &OrderPlaced,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order = order.id, "fulfilling the order");
        ready(Ok(()))
    }
}

fn service(pool: SqlitePool) -> impl App {
    let tracking = Outbox::new(pool).track::<OrderOutbox, Orders>();
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .layer(tracking.layer())
        .publish_layer(tracking.publish_layer())
        .with_broker(MemoryBroker::new(), |b| {
            b.include(subscriber("checkout", Place).reply().build())
                .out_reply(Publish);
            b.include(subscriber("orders", Fulfil).build());
            // Once the subscriptions are open, the unprocessed records are sent again.
            b.after_startup(Publish, tracking.republish());
        })
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let pool = SqlitePool::connect("sqlite:orders.db?mode=rwc").await?;
    sqlx::raw_sql(SCHEMA).execute(&pool).await?;
    service(pool.clone()).run().await?;
    pool.close().await;
    Ok(())
}