OwnedTransactions

OwnedTransactions opens independent transactions that own their buffers, so a handler keeps several open on one slot at once.

A broker whose transactions are client buffers flushed at commit offers this kind: an AMQP confirms buffer, a Redis pipeline, the memory broker. A handler bounds its slot Out<impl OwnedTransactions, Marker> and calls transaction(). Each call returns a TypedTransaction of its own, and settling one never touches another. Publishes go through the transaction's builder and stay in its buffer until commit() flushes them together. commit() and abort() consume the value, so a publish after settling does not compile. A transaction dropped unsettled discards its buffer, the way an abort does, and logs a warning. The transaction sends from its own buffer: the slot's transforms and the app's publish layers do not run on its messages, and a test reads them on the destination record, published, rather than the slot's. Kafka-like brokers offer only the borrowed kind, TransactionalPublisher.

Example

A transfer books its entries in one transaction and its audit record in another, so an empty transfer keeps the audit record and books nothing.

Macros
Manual

examples/owned_transactions.rs
#[derive(OutSlot)]
#[publishes(Debit, Credit, Audited)]
struct Ledger;

/// The audit record and the entries settle apart: two transactions open on one slot.
#[subscriber("transfers")]
async fn book(
    transfer: &Transfer,
    Out(ledger): Out<impl OwnedTransactions, Ledger>,
) -> HandlerOutcome {
    let Ok(mut audit) = ledger.transaction().await else {
        return HandlerOutcome::retry();
    };
    let Ok(mut entries) = ledger.transaction().await else {
        return HandlerOutcome::retry();
    };
    let id = transfer.id;
    if audit.message(&Audited { id }).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    if transfer.amount == 0 {
        // Nothing to book: the entries go, the audit record stays.
        if entries.abort().await.is_err() || audit.commit().await.is_err() {
            return HandlerOutcome::retry();
        }
        return HandlerOutcome::ack();
    }
    let amount = transfer.amount;
    let debit = entries.message(&Debit { id, amount }).publish().await;
    let credit = entries.message(&Credit { id, amount }).publish().await;
    if debit.is_err()
        || credit.is_err()
        || entries.commit().await.is_err()
        || audit.commit().await.is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

examples/manual/owned_transactions.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 {}
impl PublishedThrough<Ledger> for Audited {}

/// The audit record and the entries settle apart: two transactions open on one slot.
struct Book;

impl<L> Handle<Transfer, (), Outs<(L,)>> for Book
where
    L: OutEntry<Ledger, Wire: OwnedTransactions>,
{
    async fn handle(
        &self,
        transfer: &Transfer,
        outs: &Outs<(L,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let ledger = outs.get(Ledger);
        let Ok(mut audit) = ledger.transaction().await else {
            return Err(HandlerOutcome::retry());
        };
        let Ok(mut entries) = ledger.transaction().await else {
            return Err(HandlerOutcome::retry());
        };
        let id = transfer.id;
        if audit.message(&Audited { id }).publish().await.is_err() {
            return Err(HandlerOutcome::retry());
        }
        if transfer.amount == 0 {
            // Nothing to book: the entries go, the audit record stays.
            if entries.abort().await.is_err() || audit.commit().await.is_err() {
                return Err(HandlerOutcome::retry());
            }
            return Ok(());
        }
        let amount = transfer.amount;
        let debit = entries.message(&Debit { id, amount }).publish().await;
        let credit = entries.message(&Credit { id, amount }).publish().await;
        if debit.is_err()
            || credit.is_err()
            || entries.commit().await.is_err()
            || audit.commit().await.is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

Macros
Manual

examples/owned_transactions.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(book).out(Ledger, Publish).build();
    })
}

examples/manual/owned_transactions.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(subscriber("transfers", Book).build())
            .out(Ledger, Publish)
            .build();
    })
}

Example​

Example