Группы потребителей и восстановление после сбоя

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

Пример - платёжный сервис. Платежи приходят в поток payments. Группа бухгалтерского журнала проводит каждый платёж, а группа антифрода его проверяет.

Группа воркеров

Макросы

src/main.rs

/// Longer than the slowest posting ever takes: an entry idle this long belongs to no live worker.
const MIN_IDLE: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "payments")]
struct Payment {
    id: u64,
    account: String,
    amount_cents: u64,
}

#[subscriber(
    RedisStream::claiming("payments", MIN_IDLE).group("ledger"),
    workers(8, by_key)
)]
async fn post_payment(payment: &Payment, ctx: &mut Context<'_>) -> HandlerOutcome {
    let earlier_deliveries = ctx.headers().get_str(DELIVERY_COUNT_HEADER).unwrap_or("0");
    if payment.account.starts_with("migrating-") {
        tracing::warn!(
            payment.id,
            earlier_deliveries,
            "account locked, payment left pending"
        );
        return HandlerOutcome::retry();
    }
    tracing::info!(payment.id, earlier_deliveries, "payment posted");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Longer than the slowest posting ever takes: an entry idle this long belongs to no live worker.
const MIN_IDLE: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Payment {
    id: u64,
    account: String,
    amount_cents: u64,
}

named!(Payment, "payments");

struct PostPayment;

impl Handle<Payment> for PostPayment {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let earlier_deliveries = ctx.headers().get_str(DELIVERY_COUNT_HEADER).unwrap_or("0");
        if payment.account.starts_with("migrating-") {
            tracing::warn!(
                payment.id,
                earlier_deliveries,
                "account locked, payment left pending"
            );
            return ready(Err(HandlerOutcome::retry()));
        }
        tracing::info!(payment.id, earlier_deliveries, "payment posted");
        ready(Ok(()))
    }
}

Поток Redis читается через группу потребителей. Группа хранит курсор в потоке, и каждая запись за курсором достаётся одному потребителю группы. Каждая подписка - это один потребитель. Поэтому каждый работающий экземпляр сервиса добавляет потребителя в ledger, и экземпляры делят платежи между собой. Потребитель получает сгенерированное имя: ruststream-0, ruststream-1 и так далее в каждом процессе. consumer задаёт имя, например по имени хоста, чтобы XINFO CONSUMERS различал экземпляры.

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

RedisStream::claiming читает оба набора за один вызов: сначала записи, которые ждут подтверждения не меньше min_idle, затем новые. Поэтому подписка, которая принимает новые платежи, перехватывает и платежи остановившегося воркера. Ей нужен Redis 8.4 или новее. На более старом сервере сервис не запускается, а ошибка называет подписку и обе версии.

retry на этой подписке не добавляет копию. Запись остаётся неподтверждённой под своим идентификатором, и следующее чтение перехватывает её, когда она простоит min_idle. post_payment пользуется этим для счёта, который переносится: платёж ждёт в списке неподтверждённых записей и возвращается каждые 30 секунд.

У перехваченной записи есть заголовок DELIVERY_COUNT_HEADER с числом доставок до этой. У новой записи он равен нулю. IDLE_MS_HEADER хранит, сколько запись простояла. Как ограничить число доставок по этому счётчику, показывает руководство Переобработка, задержки и dead-letter.

Выбор min_idle

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

Вторая группа

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "payments.screened")]
struct Screened {
    payment_id: u64,
    suspicious: bool,
}

#[subscriber(
    RedisStream::new("payments")
        .group("fraud")
        .start_id(StreamStart::Beginning),
    reply
)]
async fn screen_payment(payment: &Payment) -> Screened {
    Screened {
        payment_id: payment.id,
        suspicious: payment.amount_cents > 1_000_000,
    }
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Screened {
    payment_id: u64,
    suspicious: bool,
}

named!(Screened, "payments.screened");

struct ScreenPayment;

impl Handle<Payment, Screened> for ScreenPayment {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Screened, HandlerOutcome>> {
        ready(Ok(Screened {
            payment_id: payment.id,
            suspicious: payment.amount_cents > 1_000_000,
        }))
    }
}

У каждой группы свой курсор и свой список неподтверждённых записей, поэтому каждая группа читает каждую запись. Группа антифрода получает каждый платёж, который получает группа журнала, и ни одна не замедляет другую. На практике у каждого сервиса, который читает поток, своя группа.

Сервис создаёт группу при запуске, если её ещё нет. start_id задаёт, откуда начинает новая группа. StreamStart::New, значение по умолчанию, читает только записи, добавленные после создания группы, а StreamStart::Beginning читает все записи, которые поток ещё хранит. Группу антифрода добавили позже группы журнала, поэтому она начинает с начала и проверяет платежи, которые уже лежат в потоке. Существующая группа сохраняет свой курсор, и start_id его не сдвигает. Его сдвигает start_at.

Сервис

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(post_payment);
            b.include(screen_payment).out_reply(Publish);
        },
    )
}

Вручную

src/main.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(
                subscriber(
                    RedisStream::claiming("payments", MIN_IDLE).group("ledger"),
                    PostPayment,
                )
                .workers_by_key(nonzero!(8))
                .build(),
            );
            b.include(
                subscriber(
                    RedisStream::new("payments")
                        .group("fraud")
                        .start_id(StreamStart::Beginning),
                    ScreenPayment,
                )
                .reply()
                .build(),
            )
            .out_reply(Publish);
        },
    )
}

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

screen_payment отвечает сообщением Screened в поток payments.screened.

Порядок по счёту

workers(8, by_key) запускает post_payment в восьми партициях. Платежи с одинаковым ключом партиционирования попадают в одну партицию и обрабатываются друг за другом, поэтому платежи одного счёта проводятся в том порядке, в каком их добавили. В Redis нет разделов, поэтому ключ передаётся в заголовке PARTITION_KEY_HEADER. Производитель задаёт его шагом partition_key своей публикации. Платёж без ключа попадает в любую партицию. Порядок соблюдается внутри одного потребителя. Два экземпляра сервиса проводят каждый свою долю платежей.

Восстановление до Redis 8.4

Макросы

src/main.rs

#[subscriber(RedisStream::new("payments").group("ledger"))]
async fn post_new_payment(payment: &Payment) -> HandlerOutcome {
    tracing::info!(payment.id, "payment posted");
    HandlerOutcome::ack()
}

#[subscriber(RedisStream::reclaim("payments", MIN_IDLE).group("ledger"))]
async fn recover_payment(payment: &Payment, ctx: &mut Context<'_>) -> HandlerOutcome {
    let deliveries = ctx.headers().get_str(DELIVERY_COUNT_HEADER).unwrap_or("1");
    tracing::warn!(
        payment.id,
        deliveries,
        "payment recovered from a stopped worker"
    );
    HandlerOutcome::ack()
}

fn app_before_8_4() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(post_new_payment);
            b.include(recover_payment);
            b.include(screen_payment).out_reply(Publish);
        },
    )
}

Вручную

src/main.rs

struct PostNewPayment;

impl Handle<Payment> for PostNewPayment {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(payment.id, "payment posted");
        ready(Ok(()))
    }
}

struct RecoverPayment;

impl Handle<Payment> for RecoverPayment {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let deliveries = ctx.headers().get_str(DELIVERY_COUNT_HEADER).unwrap_or("1");
        tracing::warn!(
            payment.id,
            deliveries,
            "payment recovered from a stopped worker"
        );
        ready(Ok(()))
    }
}

fn app_before_8_4() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(
                subscriber(RedisStream::new("payments").group("ledger"), PostNewPayment).build(),
            );
            b.include(
                subscriber(
                    RedisStream::reclaim("payments", MIN_IDLE).group("ledger"),
                    RecoverPayment,
                )
                .build(),
            );
            b.include(
                subscriber(
                    RedisStream::new("payments")
                        .group("fraud")
                        .start_id(StreamStart::Beginning),
                    ScreenPayment,
                )
                .reply()
                .build(),
            )
            .out_reply(Publish);
        },
    )
}

У более старого сервера нет чтения с перехватом, поэтому восстановление - это вторая подписка на ту же группу. RedisStream::new читает новые записи. RedisStream::reclaim командой XAUTOCLAIM читает только записи, которые ждут подтверждения не меньше min_idle, и так перехватывает то, что держал остановившийся воркер. Перехваченная доставка учитывает себя в DELIVERY_COUNT_HEADER, поэтому первое восстановление сообщает как минимум 1.

На подписке new вызов retry добавляет копию записи в поток и подтверждает оригинал. Копия возвращается сразу, после записей, добавленных за это время. Для ожидания вроде переноса счёта выше там нужен retry_after, о котором рассказывает руководство Переобработка, задержки и dead-letter.

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

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