Повторное чтение и перемотка потока

Это руководство сдвигает курсор группы потребителей в потоке Redis: назад, чтобы прочитать записи заново, и вперёд, чтобы пропустить их.

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

Повторное чтение с начала

Макросы

src/main.rs

/// Rebuilds the balance view from the oldest entry the stream retains, on every start.
#[subscriber(
    RedisStream::new("ledger").group("balances"),
    start_at(RedisGroupPosition::beginning())
)]
async fn rebuild(entry: &LedgerEntry) -> HandlerOutcome {
    tracing::info!(
        entry.seq,
        entry.account,
        entry.amount_cents,
        "balance updated"
    );
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Rebuilds the balance view from the oldest entry the stream retains, on every start.
struct Rebuild;

impl Handle<LedgerEntry> for Rebuild {
    fn handle(
        &self,
        entry: &LedgerEntry,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(
            entry.seq,
            entry.account,
            entry.amount_cents,
            "balance updated"
        );
        ready(Ok(()))
    }
}

Поток Redis хранит записи и после того, как их прочитали и подтвердили. Группа потребителей держит над ними один курсор: идентификатор последней выданной ей записи. Повторное чтение - это сдвиг курсора назад, вручную это XGROUP SETID.

start_at сдвигает курсор перед первой доставкой подписки при каждом запуске. RedisGroupPosition::beginning - самая старая запись, которую поток ещё хранит, поэтому rebuild читает весь журнал при каждом запуске сервиса. Представление, перестроенное так, начинается с пустого состояния, например с состояния в памяти.

Курсор принадлежит группе, а не подписке. Заново читает каждый потребитель balances, в этом сервисе и в каждом другом его экземпляре.

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

Макросы

src/main.rs

fn service(broker: RedisBroker) -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        b.include(rebuild);
        b.include(notify);
    })
}

Вручную

src/main.rs

fn service(broker: RedisBroker) -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        let ledger = RedisStream::new("ledger").group("balances");
        b.include(
            subscriber(ledger, Rebuild)
                .build()
                .start_at(RedisGroupPosition::beginning()),
        );
        let alerts = RedisStream::new("alerts").group("notifier");
        b.include(subscriber(alerts, Notify).build());
    })
}

С макросами start_at(..) - аргумент макроса #[subscriber]. В ручной форме это шаг монтирования после build().

Позиция	Откуда группа продолжает чтение
RedisGroupPosition::beginning()	с самой старой записи, которую хранит поток
RedisGroupPosition::end()	с конца: только записи, добавленные позже
RedisGroupPosition::after(id)	с записи после id, сама она пропускается

StreamStart - другая настройка: она задаёт, откуда начинает группа, которой ещё нет. start_at сдвигает существующую группу при каждом запуске.

Перемотка в конец

Макросы

src/main.rs

/// Sends alerts. On a halt marker the group moves to the tail, so the alerts queued behind the
/// marker are never sent, by this consumer or any other of `notifier`.
#[subscriber(RedisStream::new("alerts").group("notifier"))]
async fn notify(alert: &Alert, Ctx(seeker): Ctx<keys::SeekHandle>) -> HandlerOutcome {
    if !alert.halt {
        tracing::info!(alert.id, "alert sent");
        return HandlerOutcome::ack();
    }
    tracing::warn!(
        group = seeker.group(),
        "halt marker, the backlog is skipped"
    );
    match seeker.seek(RedisGroupPosition::end()).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

Вручную

src/main.rs

/// Sends alerts. On a halt marker the group moves to the tail, so the alerts queued behind the
/// marker are never sent, by this consumer or any other of `notifier`.
struct Notify;

impl Handle<Alert, (), (), StreamContext> for Notify {
    async fn handle(
        &self,
        alert: &Alert,
        _outs: &(),
        ctx: &mut Context<'_, StreamContext>,
    ) -> Result<(), HandlerOutcome> {
        if !alert.halt {
            tracing::info!(alert.id, "alert sent");
            return Ok(());
        }
        let seeker = ctx.context(keys::SeekHandle);
        tracing::warn!(
            group = seeker.group(),
            "halt marker, the backlog is skipped"
        );
        seeker
            .seek(RedisGroupPosition::end())
            .await
            .map_err(|_| HandlerOutcome::retry())
    }
}

Во время работы сервиса обработчик сдвигает свою группу через дескриптор перемотки. Его даёт Ctx<keys::SeekHandle>, а его метод group() называет группу, которую он сдвигает.

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

Пакетный обработчик получает тот же дескриптор через ctx.context(keys::SeekHandle) в контексте пакета.

Повторное чтение одной доставки

Доставка знает свою позицию: Ctx<keys::Position> - это курсор для её повторной доставки. Перемотка на него доставляет ту же запись снова, а затем записи после неё в порядке потока. Обработчик, которому нужно сохранить порядок при временной ошибке, может перемотать туда вместо retry, потому что на потоке retry добавляет копию в конец. Ctx<keys::EntryId> - собственный идентификатор записи, а RedisGroupPosition::after(id) строит позицию из идентификатора, сохранённого в другом месте.

Что перемотка не меняет

Перемотка сдвигает курсор и больше ничего.

• Запись, которую доставили и не подтвердили, остаётся в списке неподтверждённых записей группы, и подписка с перехватом или reclaim по-прежнему её находит.
• Копия, которая ждёт во множестве задержек, добавляется в поток, когда наступает её срок, где бы ни читала группа.
• Повторно прочитанная запись доставляется как новая, и Redis снова считает её доставки с единицы. Подписка с перехватом отсчитывает предел попыток с момента повторного чтения, а собственный счётчик попыток фреймворка не меняется.

Хранение записей

Redis хранит записи потока, пока их что-нибудь не обрежет, и повторное чтение достаёт только то, что сохранилось. Обрезка командой XTRIM - решение сервиса: её отправляет обработчик через Ctx<keys::FredPool> или выполняет оператор. Группа, курсор которой указывает раньше самой старой сохранённой записи, читает с самой старой.

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

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