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.

Macros

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());
        })
}

Manual

examples/manual/sqlx_outbox_interplay.rs

use ruststream_sqlx::keys::TxContext;
use ruststream_sqlx::outbox::Outbox;
use ruststream_sqlx::prelude::*;
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(sqlx::FromRow)]
struct ReceiptJob {
    id: i64,
    headers: Headers,
    payload: Vec<u8>,
}

impl InboxTable for ReceiptJob {
    type Id = i64;
    type Table = InboxSpec<(Lease<DateTime<Utc>>, HeadersColumn, Payload)>;
    const TABLE: Self::Table = InboxSpec::new("receipt_jobs", Column::new("id").generated())
        .lease(Column::new("locked_until"))
        .headers(Column::new("headers"))
        .payload(Column::new("payload"));

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

impl HeaderRow for ReceiptJob {
    type Column = Headers;

    fn headers_mut(&mut self) -> &mut Headers {
        &mut self.headers
    }
}

impl PayloadRow for ReceiptJob {
    type Column = Vec<u8>;

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

/// 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.
struct Take;

// Transactional mode lends the handler the delivery's transaction through its context.
impl<Receipts> Handle<PlaceOrder, (), Outs<(Receipts,)>, TxContext<Sqlite>> for Take
where
    Receipts: OutEntry<DefaultSlot, Wire: Publisher>,
{
    async fn handle(
        &self,
        order: &PlaceOrder,
        outs: &Outs<(Receipts,)>,
        ctx: &mut Context<'_, TxContext<Sqlite>>,
    ) -> Result<(), HandlerOutcome> {
        let receipt = Receipt { order: order.id };
        if outs
            .get(DefaultSlot)
            .message(&receipt)
            .publish()
            .await
            .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        let Ok(Ctx(mut tx)) = Ctx::<keys::Tx<Sqlite>>::from_context(ctx).await;
        let written = sqlx::query("INSERT INTO orders (id) VALUES (?)")
            .bind(order.id)
            .execute(&mut *tx)
            .await;
        match (written, order.valid) {
            (Ok(_), true) => Ok(()),
            // Dropped: the transaction discards the order's row, and the receipt stays sent.
            (Ok(_), false) => Err(HandlerOutcome::drop()),
            (Err(_), _) => Err(HandlerOutcome::retry()),
        }
    }
}

/// The acknowledgement marks the receipt's record processed.
struct File;

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

/// The name `Receipt` is published under.
struct Receipts;

impl TrackedName for Receipts {
    const NAME: &'static str = "receipts";
}

fn service(pool: SqlitePool) -> impl App {
    let tracking = Outbox::new(pool.clone()).track::<ReceiptOutbox, Receipts>();
    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| {
            let take = subscriber("orders", Take)
                .map_source(|_| InboxQueue::<OrderJob>::new("orders"))
                .transactional();
            b.include(take.build())
                .out(DefaultSlot, Repository::<ReceiptJob>::default())
                .build();
            let file = subscriber("receipts", File)
                .map_source(|_| InboxQueue::<ReceiptJob>::new("receipts"));
            b.include(file.build());
            b.after_startup(Repository::<ReceiptJob>::default(), tracking.republish());
        })
}