transaction

transaction hands back a value that owns its buffer, so several can be open at once and a double commit does not compile.

This is the owned kind, the core's OwnedTransactions. A handler takes its slot as Out<impl OwnedTransactions, Slot> and calls transaction() on it. The value it gets back publishes like the slot, txn.message(&value).publish(), and admits the types the slot admits. Each call opens a buffer of its own, so the workers of a workers(n) subscription never share one, and the slot keeps publishing directly meanwhile. commit sends the buffer as one MULTI / EXEC block; abort discards it. Both consume the value, so a second commit or a publish after settling does not compile.

Dropping a transaction that was neither committed nor aborted discards its buffer, as an abort does, and writes a warning. A commit that fails has consumed the transaction, and its buffer is gone. The recovery is to redeliver the input, which a retry outcome does, not to send the buffer again. The live transaction is a RedisTransaction, with the same limits as the borrowed kind: standalone and sentinel only, no rollback. The slot's policy is TransactionalPublish, so a cluster refuses it at startup.

Example

An order ships one entry per item in one block, on four workers at once; an item without a name aborts the transaction, which discards the entries already buffered.

Macros
Manual

examples/redis_transactions_transaction.rs
/// One shipment per item, committed as one block; four orders ship at once.
#[subscriber(RedisStream::new("orders").group("shipping"), workers(4))]
async fn ship(
    order: &Order,
    Out(shipments): Out<impl OwnedTransactions, Shipments>,
) -> HandlerOutcome {
    let Ok(mut txn) = shipments.transaction().await else {
        return HandlerOutcome::retry();
    };
    for item in &order.items {
        if item.is_empty() {
            tracing::warn!(order.id, "an item has no name, the order ships nothing");
            // The abort discards the shipments the transaction buffered so far.
            return match txn.abort().await {
                Ok(()) => HandlerOutcome::drop(),
                Err(_) => HandlerOutcome::retry(),
            };
        }
        let shipment = Shipment {
            order_id: order.id,
            item: item.clone(),
        };
        if txn.message(&shipment).publish().await.is_err() {
            return HandlerOutcome::retry();
        }
    }
    match txn.commit().await {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(ship).out(Shipments, TransactionalPublish).build();
        },
    )
}

examples/manual/redis_transactions_transaction.rs
/// One shipment per item, committed as one block; four orders ship at once.
struct Ship;

impl<S> Handle<Order, (), Outs<(S,)>> for Ship
where
    S: OutEntry<Shipments, Wire: OwnedTransactions>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(S,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let mut txn = outs
            .get(Shipments)
            .transaction()
            .await
            .map_err(|_| HandlerOutcome::retry())?;
        for item in &order.items {
            if item.is_empty() {
                tracing::warn!(order.id, "an item has no name, the order ships nothing");
                // The abort discards the shipments the transaction buffered so far.
                txn.abort().await.map_err(|_| HandlerOutcome::retry())?;
                return Err(HandlerOutcome::drop());
            }
            let shipment = Shipment {
                order_id: order.id,
                item: item.clone(),
            };
            txn.message(&shipment)
                .publish()
                .await
                .map_err(|_| HandlerOutcome::retry())?;
        }
        txn.commit().await.map_err(|_| HandlerOutcome::retry())
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            let ship = subscriber(RedisStream::new("orders").group("shipping"), Ship);
            b.include(ship.workers(nonzero!(4)).build())
                .out(Shipments, TransactionalPublish)
                .build();
        },
    )
}

Example​

Example