Побочные эффекты вместе с подтверждением

Это руководство отправляет собственные записи обработчика в Redis вместе с подтверждением его доставки.

Пример - сервис заказов. Он ставит позиции каждого заказа в очередь комплектации склада, отвечает пронумерованной квитанцией и проводит платежи по бухгалтерскому журналу.

Постановка команды в очередь

Макросы

src/main.rs

/// Puts every line of an order on the warehouse's pick queue and answers with a numbered receipt.
#[subscriber(RedisStream::new("orders").group("fulfilment"), reply)]
async fn accept(
    order: &Order,
    Ctx(pipeline): Ctx<keys::Pipeline>,
    Ctx(pool): Ctx<keys::FredPool>,
) -> Result<Receipt, HandlerOutcome> {
    for line in &order.lines {
        if line.quantity == 0 {
            tracing::warn!(order.id, line.sku, "an empty line drops the order");
            return Err(HandlerOutcome::drop());
        }
        if pipeline.lpush("picks", line.sku.as_str()).await.is_err() {
            return Err(HandlerOutcome::retry());
        }
    }
    // The receipt needs its number now, so the `INCR` goes over the pool, not into the round.
    let number = pool
        .incr::<i64, _>("receipts:sequence")
        .await
        .map_err(|err| {
            tracing::warn!(order.id, %err, "the receipt sequence is out of reach");
            HandlerOutcome::retry()
        })?;
    Ok(Receipt {
        order_id: order.id,
        number,
    })
}

Вручную

src/main.rs

/// Puts every line of an order on the warehouse's pick queue and answers with a numbered receipt.
struct Accept;

impl Handle<Order, Receipt, (), PipelineContext> for Accept {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, PipelineContext>,
    ) -> Result<Receipt, HandlerOutcome> {
        let pipeline = ctx.context(keys::Pipeline);
        for line in &order.lines {
            if line.quantity == 0 {
                tracing::warn!(order.id, line.sku, "an empty line drops the order");
                return Err(HandlerOutcome::drop());
            }
            if pipeline.lpush("picks", line.sku.as_str()).await.is_err() {
                return Err(HandlerOutcome::retry());
            }
        }
        // The receipt needs its number now, so the `INCR` goes over the pool, not into the round.
        let number = ctx
            .context(keys::FredPool)
            .incr::<i64, _>("receipts:sequence")
            .await
            .map_err(|err| {
                tracing::warn!(order.id, %err, "the receipt sequence is out of reach");
                HandlerOutcome::retry()
            })?;
        Ok(Receipt {
            order_id: order.id,
            number,
        })
    }
}

Обработчик, который пишет в Redis, а затем подтверждает доставку, тратит два обмена с сервером. Сбой между ними применяет запись без подтверждения, и запись потока доставляется снова.

Ctx<keys::Pipeline> даёт обработчику раунд доставки, RedisPipeline. Его методы команд совпадают с методами fred: lpush, hincrby, xadd, publish, eval и остальные. Вызов ставит команду в очередь и сразу возвращает управление. Ok означает, что команда поставлена в очередь, а не что Redis её выполнил.

Команды из очереди уходят после возврата из обработчика и только если он подтверждает доставку. Сначала идут они, затем XACK, в одном конвейере.

Исход	Команды из очереди	Завершение доставки
ack	отправляются до завершения	XACK
drop	отбрасываются	XACK
retry, retry_after	отбрасываются	копия для переобработки, затем XACK
не завершена	отбрасываются	нет: запись остаётся неподтверждённой

accept ставит в очередь по одной команде добавления на позицию. Позиция без количества отбрасывает заказ, и вместе с ним отбрасываются команды, поставленные для предыдущих позиций.

Чтение значения сразу

У команды из очереди нет ответа внутри обработчика. Значение, которое обработчику нужно сразу, он получает через пул соединений брокера, который даёт Ctx<keys::FredPool>. Так accept нумерует каждую квитанцию командой INCR. Рядом с keys::Pipeline параметр keys::FredPool стоит в сигнатуре вторым.

Ответы в раунде

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            // The receipt joins the delivery's round and leaves after the picks.
            b.include(accept.pipeline())
                .out_reply(Publish)
                .transform(InRound);
            b.include(capture.pipeline().atomic());
        },
    )
}

Вручную

src/main.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            let accept = subscriber(RedisStream::new("orders").group("fulfilment"), Accept);
            // The receipt joins the delivery's round and leaves after the picks.
            b.include(accept.pipeline().reply().build())
                .out_reply(Publish)
                .transform(InRound);
            let capture = subscriber(RedisStream::new("{payments}").group("ledger"), Capture);
            b.include(capture.pipeline().atomic().build());
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

.pipeline() в точке монтирования открывает окно, и команды обработчика можно поставить в очередь только на подписке, смонтированной с этим шагом. Обработчик, который принимает Ctx<keys::Pipeline> и смонтирован без этого шага, не компилируется.

Без преобразования InRound ответ публикуется сразу после возврата из обработчика, раньше раунда. Это преобразование на ответе помещает его в раунд после команд, поэтому квитанция уходит только вместе с комплектацией. Публикация через слот Out входит в раунд, когда обработчик привязывает слот вызовом pipeline.bind(out).

Окно собирает раунды всех доставок одного чтения. Окно отправляется, когда завершены все доставки чтения, поэтому под нагрузкой весь полученный пакет уходит за один обмен с сервером. При редком потоке сообщений окно отправляется сразу после возврата из последнего обработчика. Об ошибке отправки сообщается один раз, с подпиской и числом команд и завершений в окне. Записи, XACK которых не прошёл, остаются неподтверждёнными и доставляются снова.

Пакетный обработчик берёт раунд из своего контекста, ctx.context(keys::Pipeline), и его команды уходят, только если подтверждена каждая запись пакета (Отправка).

Атомарные сегменты

Макросы

src/main.rs

/// Books a payment: the ledger entry, the balance and the acknowledgement commit as one block.
#[subscriber(RedisStream::new("{payments}").group("ledger"))]
async fn capture(payment: &Payment, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
    let entry = format!("{}:{}", payment.id, payment.amount_cents);
    if pipeline.lpush("{payments}:ledger", entry).await.is_err()
        || pipeline
            .hincrby(
                "{payments}:balances",
                payment.account.as_str(),
                payment.amount_cents,
            )
            .await
            .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Books a payment: the ledger entry, the balance and the acknowledgement commit as one block.
struct Capture;

impl Handle<Payment, (), (), PipelineContext> for Capture {
    async fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        ctx: &mut Context<'_, PipelineContext>,
    ) -> Result<(), HandlerOutcome> {
        let pipeline = ctx.context(keys::Pipeline);
        let entry = format!("{}:{}", payment.id, payment.amount_cents);
        if pipeline.lpush("{payments}:ledger", entry).await.is_err()
            || pipeline
                .hincrby(
                    "{payments}:balances",
                    payment.account.as_str(),
                    payment.amount_cents,
                )
                .await
                .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

Конвейер - не транзакция: команда может выполниться, а XACK после неё вернуть ошибку. .atomic() после .pipeline() заключает команды каждой доставки и её завершение в MULTI и EXEC. Тогда запись в журнале, баланс и подтверждение платежа выполняются вместе или не выполняются вовсе.

В Redis нет отката. Команда, которая возвращает ошибку внутри EXEC, например HINCRBY на ключе другого типа, оставляет остальные применёнными.

В кластере блок MULTI выполняется в одном хеш-слоте. Поток называется {payments}, а ключи обработчика {payments}:ledger и {payments}:balances: хеш-тег в фигурных скобках помещает их все в слот потока. В имени потока есть фигурные скобки, поэтому тип платежа не объявляет место назначения, а производитель задаёт поток при публикации.

Скрипты Lua

eval и evalsha ставят скрипт Lua в очередь, как любую другую команду, и скрипт выполняется в раунде вместе с завершением доставки. Скрипт выполняется на сервере атомарно, поэтому подходит для проверки и записи, которые нельзя разделять. Redis внутри процесса не выполняет скрипты, поэтому тест обработчика, который ставит скрипт в очередь, работает с сервером.

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

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