atomic

.atomic() after .pipeline() wraps each delivery's commands and settle in MULTI and EXEC, so they apply together or not at all.

The step follows .pipeline() at the mount site: b.include(handler.pipeline().atomic()). A delivery's segment becomes MULTI, the commands its handler queued, the delivery's settle where the form has one, and EXEC. The transaction still travels inside the window's pipeline, so it costs no round trip of its own. A handler's side effects and its acknowledgement are then one unit: a reader never sees the effects of an entry that is still pending. .atomic() without .pipeline() does not compile, and the error says to write .pipeline().atomic().

The transaction is Redis's, with Redis's limits. A command that fails inside EXEC, a WRONGTYPE for example, leaves the other commands of the block executed: Redis does not roll back. On a cluster, a transaction runs only when every key in it hashes to one slot, the slot of the subscription's key. A hash tag gives the handler's keys that slot: {orders}:invoices beside a stream {orders}. A command for another slot makes Redis refuse the whole EXEC, so a segment never runs in part.

Example

An invoice is queued beside each paid order and leaves in one MULTI / EXEC with the order's XACK.

Macros
Manual

examples/redis_pipelining_atomic.rs
/// Queues an invoice beside the order; the hash tag `{orders}` keeps both keys on one cluster slot.
#[subscriber(RedisStream::new("{orders}").group("billing"))]
async fn invoice(order: &Order, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
    if pipeline
        .lpush("{orders}:invoices", order.id.to_string())
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    if !order.paid {
        tracing::warn!(order.id, "an unpaid order gets no invoice");
        return HandlerOutcome::drop();
    }
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(invoice.pipeline().atomic());
        },
    )
}

examples/manual/redis_pipelining_atomic.rs
/// Queues an invoice beside the order; the hash tag `{orders}` keeps both keys on one cluster slot.
struct Invoice;

impl Handle<Order, (), (), PipelineContext> for Invoice {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, PipelineContext>,
    ) -> Result<(), HandlerOutcome> {
        let pipeline = ctx.context(keys::Pipeline);
        if pipeline
            .lpush("{orders}:invoices", order.id.to_string())
            .await
            .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        if !order.paid {
            tracing::warn!(order.id, "an unpaid order gets no invoice");
            return Err(HandlerOutcome::drop());
        }
        Ok(())
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            let invoice = subscriber(RedisStream::new("{orders}").group("billing"), Invoice);
            b.include(invoice.pipeline().atomic().build());
        },
    )
}

Example​

Example