Транзакции в потоках

Это руководство публикует группы сообщений в потоки Redis по принципу «всё или ничего».

Пример - сервис заказов. Он подтверждает заказы пакетами и отправляет каждую упакованную посылку как набор отгрузок, по одной на товар.

Ответы пакета одним блоком

Макросы

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())
}

Вручную

src/main.rs

/// Confirms a batch of orders. The confirmations of one batch are published together or not at
/// all.
struct Confirm;

impl Handle<[Order], Vec<Confirmation>> for Confirm {
    fn handle(
        &self,
        orders: &[Order],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Vec<Confirmation>, Vec<HandlerOutcome>>> {
        if orders.iter().any(|order| order.items.is_empty()) {
            tracing::warn!(
                orders = orders.len(),
                "a batch with an empty order is dropped"
            );
            return ready(Err(orders.iter().map(|_| HandlerOutcome::drop()).collect()));
        }
        ready(Ok(orders
            .iter()
            .map(|order| Confirmation { order_id: order.id })
            .collect()))
    }
}

Транзакция отправляет свои публикации одним блоком MULTI и EXEC в порядке публикации. Читатель confirmations видит либо все подтверждения пакета, либо ни одного.

confirm - пакетный обработчик: он принимает срез и возвращает по одному подтверждению на заказ. Err ничего не публикует и завершает весь пакет своим исходом.

Монтирование

Макросы

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();
    })
}

Вручную

src/main.rs

fn service(broker: RedisBroker) -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        let confirm = subscriber(RedisStream::new("orders").group("confirmations"), Confirm);
        b.include(confirm.batch(nonzero!(32)).reply().build())
            .out_reply(TransactionalPublish)
            .transactional();
        let ship = subscriber(RedisStream::new("packed").group("shipping"), Ship);
        b.include(ship.build())
            .out(Shipments, TransactionalPublish)
            .build();
    })
}

.out_reply(TransactionalPublish).transactional() запрашивает заимствованную транзакцию. Рантайм открывает транзакцию на издателе ответов, публикует в неё ответы пакета и фиксирует её. TransactionalPublish - имя политики RedisTransactionalPublish в прелюдии потоков. Она отправляет командой XADD, как Publish, и работает только там, где работают транзакции.

Размер пакета равен COUNT чтения, которое его получает, поэтому здесь одна транзакция содержит не больше 32 подтверждений.

Транзакция, которой владеет обработчик

Макросы

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(),
    }
}

Вручную

src/main.rs

/// Publishes one shipment per item of a parcel, committed as one block.
struct Ship;

impl<Slot> Handle<Parcel, (), Outs<(Slot,)>> for Ship
where
    Slot: OutEntry<Shipments, Wire: OwnedTransactions>,
{
    async fn handle(
        &self,
        parcel: &Parcel,
        outs: &Outs<(Slot,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), HandlerOutcome> {
        let mut transaction = outs
            .get(Shipments)
            .transaction()
            .await
            .map_err(|_| 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.
                transaction
                    .abort()
                    .await
                    .map_err(|_| HandlerOutcome::retry())?;
                return Err(HandlerOutcome::drop());
            }
            let shipment = Shipment {
                order_id: parcel.order_id,
                item: item.clone(),
            };
            transaction
                .message(&shipment)
                .publish()
                .await
                .map_err(|_| HandlerOutcome::retry())?;
        }
        transaction
            .commit()
            .await
            .map_err(|_| HandlerOutcome::retry())
    }
}

transaction() на слоте Out возвращает значение, которое владеет своим буфером. Его публикации попадают в буфер, commit отправляет буфер одним блоком, а abort отбрасывает его. commit и abort поглощают значение, поэтому повторная фиксация или публикация после завершения не компилируется. На одном издателе можно одновременно открыть несколько таких транзакций, и завершение одной никогда не затрагивает другую. Транзакция, удалённая без commit и abort, отбрасывается как при abort, с предупреждением в журнале.

owned_transaction() делает то же над значениями: каждая публикация принимает место назначения и значение и кодирует его кодеком по умолчанию. Она подходит для хука запуска вроде after_startup, который наполняет поток начальными данными.

Что значит «нет отката»

Redis ставит команды блока в очередь и выполняет их вместе, но не отменяет их. Команда, которую сервер отказывается поставить в очередь, например с неверным числом аргументов, отбрасывает весь блок, и commit возвращает ошибку. Команда, которая возвращает ошибку во время выполнения блока, оставляет команды перед ней применёнными. Для группы XADD это случается, когда у сервера кончается память или ключ хранит другой тип.

Неудачная фиксация уже поглотила свою транзакцию, и буфера больше нет. Обработчик возвращает retry, доставка приходит снова и строит новую транзакцию.

Где работают транзакции

Транзакции работают на одиночном сервере и в развёртывании с Sentinel. Блок MULTI не может охватывать несколько хеш-слотов, поэтому в кластере сервис с политикой TransactionalPublish не запускается. У издателей списков и Pub/Sub транзакций нет.

Транзакция охватывает публикации. На стороне потребителя .pipeline().atomic() помещает собственные команды обработчика в Redis и подтверждение его доставки в один блок.

Тестирование

Тесты в файлах примера работают внутри процесса, где Redis внутри процесса выполняет MULTI и EXEC. Они проверяют, что подтверждения пакета и отгрузки посылки приходят вместе, а отброшенный пакет и прерванная посылка ничего не публикуют. Ещё один тест запускает тот же сервис с брокером кластера и проверяет, что сервис не запускается.