Transactions

A stream publisher sends a group of publishes as one MULTI and EXEC block; a batch's replies commit together with .transactional().

The stream publisher, RedisPublisher, buffers the publishes of a transaction and sends them on commit, in publish order. Subscribers then see the whole group or none of it. It offers both kinds of transaction the framework defines. A mount that publishes in transactions names RedisTransactionalPublish, TransactionalPublish in the stream prelude, which pairs into the same publisher. This page is the borrowed kind; transaction and owned_transaction are the owned kind.

The borrowed kind is the core's TransactionalPublisher: one transaction on the publisher handle. begin_transaction starts buffering, commit flushes the buffer, abort discards it, and a clone of the handle works with the same open transaction. A second begin_transaction while one is open returns RedisError::TransactionBusy and leaves the open one alone. A commit or an abort with nothing open returns RedisError::NoTransaction. Its usual shape is a batch handler whose replies commit together: .transactional() after .out_reply(TransactionalPublish) asks for it, and a policy without transactions does not compile there.

Transactions run on standalone and sentinel deployments. A MULTI block cannot span hash slots, so on a cluster TransactionalPublish refuses to pair and the service does not start. The error is RedisError::InvalidOptions, raised before the service reads its first batch. Redis does not roll back: a command that fails inside EXEC leaves the commands before it committed. For a group of XADDs that happens when Redis runs out of memory or a key holds another type. A command the server refuses to queue discards the whole block instead. The list and Pub/Sub publishers send each message on its own.

Example

Each batch's replies reach processed in one block.

Macros
Manual

examples/redis_transactions.rs
/// Processes a batch; subscribers of `processed` see all of its replies or none.
#[subscriber(RedisStream::new("orders").group("workers"), reply)]
async fn process(orders: &[Order]) -> Result<Vec<Processed>, HandlerOutcome> {
    tracing::info!(orders = orders.len(), "batch processed");
    Ok(orders
        .iter()
        .map(|order| Processed { order_id: order.id })
        .collect())
}

#[ruststream::app]
fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(process.batch(nonzero!(32)))
            .out_reply(TransactionalPublish)
            .transactional();
    })
}

examples/manual/redis_transactions.rs
/// Processes a batch; subscribers of `processed` see all of its replies or none.
struct Process;

impl Handle<[Order], Vec<Processed>> for Process {
    fn handle(
        &self,
        orders: &[Order],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Vec<Processed>, Vec<HandlerOutcome>>> {
        tracing::info!(orders = orders.len(), "batch processed");
        ready(Ok(orders
            .iter()
            .map(|order| Processed { order_id: order.id })
            .collect()))
    }
}

fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        let process = subscriber(RedisStream::new("orders").group("workers"), Process);
        b.include(process.batch(nonzero!(32)).reply().build())
            .out_reply(TransactionalPublish)
            .transactional();
    })
}

Example​

Example