Skip to main content

Transactions on streams

This guide publishes groups of messages to Redis streams all or nothing.

The example is an order service. It confirms orders in batches and ships each packed parcel as one shipment per item.

A batch's replies in one block​

src/main.rs
/// Confirms a batch of orders. The confirmations of one batch are published together or not at
/// all.
#[subscriber(RedisStream::new("orders").group("confirmations"), reply)]
async fn confirm(orders: &[Order]) -> Result<Vec<Confirmation>, HandlerOutcome> {
    if orders.iter().any(|order| order.items.is_empty()) {
        tracing::warn!(
            orders = orders.len(),
            "a batch with an empty order is dropped"
        );
        return Err(HandlerOutcome::drop());
    }
    Ok(orders
        .iter()
        .map(|order| Confirmation { order_id: order.id })
        .collect())
}

A transaction sends its publishes as one MULTI and EXEC block, in publish order. A reader of confirmations sees every confirmation of a batch, or none of them.

confirm is a batch handler: it takes a slice and returns one confirmation per order. An Err publishes nothing and settles the whole batch with its outcome.

The mount​

src/main.rs
fn service(broker: RedisBroker) -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(confirm.batch(nonzero!(32)))
            .out_reply(TransactionalPublish)
            .transactional();
        b.include(ship).out(Shipments, TransactionalPublish).build();
    })
}

.out_reply(TransactionalPublish).transactional() asks for the borrowed kind. The runtime opens a transaction on the reply publisher, publishes the batch's replies into it and commits. TransactionalPublish is RedisTransactionalPublish under the stream prelude's name. It sends with XADD, like Publish, and runs only where transactions run.

The size of the batch is the COUNT of the read that fetches it, so one transaction holds at most 32 confirmations here.

A transaction the handler owns​

src/main.rs
/// Publishes one shipment per item of a parcel, committed as one block.
#[subscriber(RedisStream::new("packed").group("shipping"))]
async fn ship(
    parcel: &Parcel,
    Out(shipments): Out<impl OwnedTransactions, Shipments>,
) -> HandlerOutcome {
    let Ok(mut transaction) = shipments.transaction().await else {
        return HandlerOutcome::retry();
    };
    for item in &parcel.items {
        if item.is_empty() {
            tracing::warn!(parcel.order_id, "a nameless item, the parcel ships nothing");
            // The abort discards the shipments buffered so far.
            return match transaction.abort().await {
                Ok(()) => HandlerOutcome::drop(),
                Err(_) => HandlerOutcome::retry(),
            };
        }
        let shipment = Shipment {
            order_id: parcel.order_id,
            item: item.clone(),
        };
        if transaction.message(&shipment).publish().await.is_err() {
            return HandlerOutcome::retry();
        }
    }
    match transaction.commit().await {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

transaction() on an Out slot returns a value that owns its buffer. Its publishes go into the buffer, commit sends the buffer as one block, and abort discards it. commit and abort consume the value, so a second commit or a publish after the end does not compile. Several owned transactions can be open on one publisher at once, and settling one never touches another. A transaction dropped without either is discarded like an abort, with a warning in the log.

owned_transaction() is the same over values: each publish takes a destination and a value and encodes it with the default codec. It suits a startup hook, such as after_startup, that seeds a stream.

What "no rollback" means​

Redis queues the commands of a block and runs them together; it does not undo them. A command the server refuses to queue, such as one with a wrong number of arguments, discards the whole block, and commit returns an error. A command that fails while the block runs leaves the commands before it applied. For a group of XADDs that happens when the server runs out of memory or a key holds another type.

A failed commit has consumed its transaction, and the buffer is gone. The handler returns retry, and the delivery comes again and builds a new transaction.

Where transactions run​

Transactions run on a standalone server and on a sentinel deployment. A MULTI block cannot span hash slots, so on a cluster a service that mounts TransactionalPublish does not start. The list and Pub/Sub publishers have no transactions.

A transaction covers publishes. On the consuming side, .pipeline().atomic() puts a handler's own Redis commands and the acknowledgement of its delivery in one block.

Testing​

The tests in the example files run in process, where the in-process Redis runs MULTI and EXEC. They check that a batch's confirmations and a parcel's shipments arrive together, and that a dropped batch and an aborted parcel publish nothing. One more test starts the same service with a cluster broker and checks that it does not start.