TransactionalPublisher

TransactionalPublisher puts several publishes in one broker transaction: every reply of a batch, or a scope a handler opens on its slot.

Messages published inside a transaction become visible together on commit, or not at all. A batch handler whose reply chain ends with .transactional() publishes every reply of the batch in one transaction, and acks the batch after the commit. A policy whose live publisher has no transactions does not compile there. A broker's prelude names its transactional policy TransactionalPublish.

By hand, a handler bounds its slot Out<impl TransactionalPublisher, Marker> and calls begin(). It returns a TransactionScope that owns the transaction, and publishes go through the scope's builder. commit() and abort() consume the scope, so a second commit or a publish after settling does not compile. A scope dropped unsettled logs a warning and leaves the broker transaction open, so a handler settles every scope it opens. The scope sends straight into the transaction: the slot's transforms and the app's publish layers do not run on its messages. This is the borrowed kind of transaction: the publisher carries one broker transaction, and one scope is open on it at a time. Kafka-like brokers, whose client holds one transaction per producer, offer this kind only. Brokers whose transactions are client buffers also offer OwnedTransactions.

Example

Every confirmation of a batch in one transaction.

Macros

examples/transactional_publisher.rs

/// Every confirmation of a batch becomes visible at once, on commit.
#[subscriber("orders", reply)]
async fn confirm(orders: &[Order]) -> Vec<Confirmed> {
    orders
        .iter()
        .map(|order| Confirmed { id: order.id })
        .collect()
}

Manual

examples/manual/transactional_publisher.rs

/// Every confirmation of a batch becomes visible at once, on commit.
struct Confirm;

impl Handle<[Order], Vec<Confirmed>> for Confirm {
    fn handle(
        &self,
        orders: &[Order],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Vec<Confirmed>, Vec<HandlerOutcome>>> {
        ready(Ok(orders
            .iter()
            .map(|order| Confirmed { id: order.id })
            .collect()))
    }
}

A scope opened on a slot: both entries of a payment, or neither.

Macros

examples/transactional_publisher.rs

#[derive(OutSlot)]
#[publishes(Debit, Credit)]
struct Ledger;

/// Both entries of a payment, or neither.
#[subscriber("payments")]
async fn book(
    payment: &Payment,
    Out(ledger): Out<impl TransactionalPublisher, Ledger>,
) -> HandlerOutcome {
    let Ok(scope) = ledger.begin().await else {
        return HandlerOutcome::retry();
    };
    let (id, amount) = (payment.id, payment.amount);
    let debit = scope.message(&Debit { id, amount }).publish().await;
    let credit = scope.message(&Credit { id, amount }).publish().await;
    if debit.is_err() || credit.is_err() {
        if let Err(err) = scope.abort().await {
            tracing::warn!(payment.id, %err, "payments: aborting the ledger transaction failed");
        }
        return HandlerOutcome::retry();
    }
    match scope.commit().await {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

Manual

examples/manual/transactional_publisher.rs

struct Ledger;

impl OutSlot for Ledger {
    const NAME: &'static str = "Ledger";
    type Destination = Reads;
}

impl PublishedThrough<Ledger> for Debit {}
impl PublishedThrough<Ledger> for Credit {}

/// Both entries of a payment, or neither.
struct Book;

impl<L> Handle<Payment, (), Outs<(L,)>> for Book
where
    L: OutEntry<Ledger, Wire: TransactionalPublisher>,
{
    async fn handle(
        &self,
        payment: &Payment,
        outs: &Outs<(L,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let Ok(scope) = outs.get(Ledger).begin().await else {
            return Err(HandlerOutcome::retry());
        };
        let (id, amount) = (payment.id, payment.amount);
        let debit = scope.message(&Debit { id, amount }).publish().await;
        let credit = scope.message(&Credit { id, amount }).publish().await;
        if debit.is_err() || credit.is_err() {
            if let Err(err) = scope.abort().await {
                tracing::warn!(payment.id, %err, "payments: aborting the ledger transaction failed");
            }
            return Err(HandlerOutcome::retry());
        }
        scope.commit().await.map_err(|_| HandlerOutcome::retry())
    }
}

Macros

examples/transactional_publisher.rs

fn app(broker: MemoryBroker) -> impl App {
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(broker, |b| {
        // One transaction per batch, committed before the batch is acked.
        b.include(confirm.batch(nonzero!(64)))
            .out_reply(TransactionalPublish)
            .transactional();
        b.include(book).out(Ledger, TransactionalPublish).build();
    })
}

Manual

examples/manual/transactional_publisher.rs

fn app(broker: MemoryBroker) -> impl App {
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(broker, |b| {
        // One transaction per batch, committed before the batch is acked.
        b.include(
            subscriber("orders", Confirm)
                .reply()
                .batch(nonzero!(64))
                .build(),
        )
        .out_reply(TransactionalPublish)
        .transactional();
        b.include(subscriber("payments", Book).build())
            .out(Ledger, TransactionalPublish)
            .build();
    })
}