Skip to main content

Interplay

The outbox wraps inbox handlers as any other; a record written from a transactional handler commits on its own, whatever the delivery's outcome.

The two middlewares wrap the handlers of an inbox subscription as they wrap the handlers of any broker, and a tracked name may lead into an inbox table. A tracked publish writes its record on a connection of its own, never in the delivery's transaction. So a handler in transactional mode commits its own writes with the delivery, while the record of what it published commits at once: a delivery that is dropped or retried discards the handler's writes and leaves the record. The guarantee stays the outbox's own: the message is delivered at least once, and a consumer that must not process it twice checks that itself.

A message tracked into an inbox table carries its record id only where the table keeps headers: a headers column that the table's Publish writes from the message's headers. The consumer's subscription then reads the id back, and its acknowledgement marks the record. A table without a headers column refuses a publish that carries a header, the id among them.

SQLite takes one writer at a time, and a transaction holds the write lock from its first write until it ends, or from its start in immediate or exclusive mode. A handler there publishes its tracked messages before its first write through the delivery's transaction: a record's insert on its own connection needs that transaction's write lock, and the transaction waits for the handler. A tracked publish that comes later fails with OutboxError::WriteLockHeld and is not sent. In immediate or exclusive mode it fails at once. After a statement through Tx the record waits for the lock at most the pool's acquire timeout, since SQLite tells whether that statement wrote only by the wait. A record written or marked from a subscription on dedicated threads opens its connection on the app's runtime, so the pool outlives the threads.

Example​

take reads an order from order_jobs in transactional mode: it sends a tracked Receipt into receipt_jobs, then writes the order through the delivery's transaction. A dropped order leaves no row in orders, and its receipt is still recorded, delivered and marked.

examples/sqlx_outbox_interplay.rs
use ruststream_sqlx::prelude::*;
use ruststream_sqlx::{Outbox, outbox};
use sqlx::SqlitePool;

/// The `headers` column: a JSON object of strings, `NULL` for none.
type Headers = Option<Json<BTreeMap<String, String>>>;

/// A queue of receipts whose headers column carries the outbox's record id.
#[derive(Inbox, sqlx::FromRow)]
#[inbox(table = "receipt_jobs")]
struct ReceiptJob {
    #[field(id, generated)]
    id: i64,
    #[field(locked_until)]
    locked_until: Option<DateTime<Utc>>,
    #[field(headers)]
    headers: Headers,
    #[field(payload)]
    payload: Vec<u8>,
}

/// The row keeps the message's headers, the record id among them.
impl Publish<Sqlite> for ReceiptJob {
    async fn publish(
        conn: &mut SqliteConnection,
        message: &OutgoingMessage<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query("INSERT INTO receipt_jobs (headers, payload) VALUES (?, ?)")
            .bind(Headers::from_headers(message.headers()))
            .bind(message.payload())
            .execute(conn)
            .await?;
        Ok(())
    }
}

/// The receipt's record commits on its own; the order's row commits with the delivery.
#[subscriber(InboxQueue::<OrderJob>::new("orders"))]
async fn take(
    order: &PlaceOrder,
    Ctx(mut tx): Ctx<keys::Tx<Sqlite>>,
    Out(out): Out<impl Publisher>,
) -> HandlerOutcome {
    if out
        .message(&Receipt { order: order.id })
        .publish()
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    let written = sqlx::query("INSERT INTO orders (id) VALUES (?)")
        .bind(order.id)
        .execute(&mut *tx)
        .await;
    match (written, order.valid) {
        (Ok(_), true) => HandlerOutcome::ack(),
        // Dropped: the transaction discards the order's row, and the receipt stays sent.
        (Ok(_), false) => HandlerOutcome::drop(),
        (Err(_), _) => HandlerOutcome::retry(),
    }
}

/// The acknowledgement marks the receipt's record processed.
#[subscriber(InboxQueue::<ReceiptJob>::new("receipts"))]
async fn file(receipt: &Receipt) -> HandlerOutcome {
    tracing::info!(order = receipt.order, "filing the receipt");
    HandlerOutcome::ack()
}

fn service(pool: SqlitePool) -> impl App {
    let tracking = outbox! { pool: pool.clone(), "receipts" => ReceiptOutbox };
    let broker = SqlxBroker::new(pool).route::<OrderJob>("orders");
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .layer(tracking.layer())
        .publish_layer(tracking.publish_layer())
        .with_broker(broker, |b| {
            b.include(take.transactional())
                .out(DefaultSlot, Repository::<ReceiptJob>::default())
                .build();
            b.include(file);
            b.after_startup(Repository::<ReceiptJob>::default(), tracking.republish());
        })
}