Переобработка, задержки и dead-letter

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

У Redis нет задержки для отдельного сообщения и нет структуры для dead-letter, а счётчик доставок есть только внутри группы потребителей потока. Остальное даёт фреймворк: точка монтирования объявляет предел попыток и место назначения, а подписка выбирает, как хранится задержка.

Пример - сервис возвратов, который проводит возвраты через процессинг карт.

Подписчик

Процессинг отказывает в возврате тремя способами, и каждому нужен свой исход. Busy значит, что лимит запросов процессинга исчерпан. Немедленная переобработка снова получит отказ, поэтому retry_after ждёт 30 секунд. Unavailable значит, что процессинг не ответил: retry пробует снова сразу. Declined значит, что карточный счёт закрыт. Новая попытка получит тот же ответ, поэтому возврат отбрасывается.

Макросы

src/main.rs

/// How long a busy processor asks its callers to wait.
const BACKOFF: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Refund {
    id: u64,
    card: String,
    amount_cents: u64,
}

#[subscriber(
    RedisStream::new("refunds")
        .group("refunds")
        .delayed_retry(DelayedRetry::DurableZset {
            key: "refunds.delayed".to_owned(),
            ttl: None,
        })
)]
async fn process_refund(
    refund: &Refund,
    State(processor): State<Arc<Processor>>,
) -> HandlerOutcome {
    settle(&processor, refund).await
}

async fn settle(processor: &Processor, refund: &Refund) -> HandlerOutcome {
    match processor.refund(refund).await {
        Ok(()) => HandlerOutcome::ack(),
        // Past the rate limit: a delivery right now would be refused again.
        Err(Refusal::Busy) => HandlerOutcome::retry_after(BACKOFF),
        // No answer: try again at once, up to the attempt cap.
        Err(Refusal::Unavailable) => HandlerOutcome::retry(),
        Err(Refusal::Declined) => {
            tracing::warn!(refund.id, "card declined, refund dropped");
            HandlerOutcome::drop()
        }
    }
}

Вручную

src/main.rs

/// How long a busy processor asks its callers to wait.
const BACKOFF: Duration = Duration::from_secs(30);

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

impl OutgoingDestination for Refund {
    type Form = CallerName;
}

impl MessageHeaders for Refund {
    type Contract = NoHeaders;
}

struct ProcessRefund;

impl Handle<Refund, (), (), (), Clients> for ProcessRefund {
    async fn handle(
        &self,
        refund: &Refund,
        _outs: &(),
        ctx: &mut Context<'_, (), Clients>,
    ) -> Result<(), HandlerOutcome> {
        settle(&ctx.state().processor, refund).await
    }
}

async fn settle(processor: &Processor, refund: &Refund) -> Result<(), HandlerOutcome> {
    match processor.refund(refund).await {
        Ok(()) => Ok(()),
        // Past the rate limit: a delivery right now would be refused again.
        Err(Refusal::Busy) => Err(HandlerOutcome::retry_after(BACKOFF)),
        // No answer: try again at once, up to the attempt cap.
        Err(Refusal::Unavailable) => Err(HandlerOutcome::retry()),
        Err(Refusal::Declined) => {
            tracing::warn!(refund.id, "card declined, refund dropped");
            Err(HandlerOutcome::drop())
        }
    }
}

fn refunds() -> RedisStream {
    RedisStream::new("refunds")
        .group("refunds")
        .delayed_retry(DelayedRetry::DurableZset {
            key: "refunds.delayed".to_owned(),
            ttl: None,
        })
}

retry на потоке добавляет в поток копию записи и подтверждает оригинал. На других структурах он делает то, что позволяет структура:

Подписка	retry без предела попыток
RedisStream::new, reclaim	копия, добавленная в поток
RedisStream::claiming	запись остаётся неподтверждённой и возвращается через min_idle
RedisList::reliable	запись переносится обратно в список
RedisList::new, RedisPubSub	ничего: сообщение пропадает

При объявленном пределе попыток retry на любой подписке, кроме подписки с перехватом, становится копией, и счётчик передаётся вместе с сообщением. У записи подписки с перехватом свой счётчик, который ведёт Redis.

Три способа хранить задержку

retry_after завершает доставку и запрашивает её снова после задержки. Подписка хранит эту задержку одним из трёх способов.

Собственному способу фреймворка ничего не нужно от дескриптора. Рантайм держит копию и публикует её обратно в ключ потока, когда задержка истекает. У копии есть заголовок с числом попыток RETRY_COUNT_HEADER, увеличенным на единицу. Копия хранится в памяти процесса, поэтому сервис, который остановился до конца задержки, её теряет.

DelayedRetry::DurableZset хранит задержку в Redis, и так делает process_refund. Отложенная доставка попадает в упорядоченное множество refunds.delayed со сроком в качестве оценки и увеличенным числом попыток, а оригинал подтверждается. Во время чтения подписка возвращает в поток каждую запись, срок которой наступил, под новым идентификатором. Задержка переживает перезапуск сервиса. Проверка срока выполняется один раз за чтение, поэтому задержка заканчивается в пределах одного интервала block после срока, по умолчанию пяти секунд. Оценки - это миллисекунды настенных часов, поэтому часы хостов сервиса должны совпадать. ttl задаёт срок жизни брошенного множества и должен быть больше самой долгой задержки, иначе записи пропадут до наступления срока.

Третий способ - подписка с перехватом, и множество ей не нужно.

Макросы

src/main.rs

#[subscriber(RedisStream::claiming("refunds.priority", BACKOFF).group("refunds"))]
async fn process_priority_refund(
    refund: &Refund,
    State(processor): State<Arc<Processor>>,
) -> HandlerOutcome {
    settle(&processor, refund).await
}

Вручную

src/main.rs

fn priority_refunds() -> RedisStream {
    RedisStream::claiming("refunds.priority", BACKOFF).group("refunds")
}

Переобработка оставляет запись в списке неподтверждённых записей группы, и следующее чтение той же подписки забирает её обратно, когда она простоит min_idle. Ничего не копируется, и запись сохраняет свой идентификатор. Единственное ожидание, которое здесь доступно, - это min_idle, поэтому более короткая задержка округляется до него, а более длинная сокращается до него. По этой причине process_priority_refund задаёт min_idle, равный паузе перед переобработкой. Подписка с перехватом, которая указывает DurableZset, соблюдает каждую задержку в точности.

Предел попыток и поток dead-letter

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("refunds", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(RedisBroker::standalone("redis://localhost:6379"), |b| {
            // Five deliveries of one refund, the first included; then the dead-letter stream.
            b.include(process_refund)
                .max_attempts(nonzero!(5u32))
                .dead_letter("refunds.dlq");
            b.include(process_priority_refund)
                .max_attempts(nonzero!(5u32))
                .dead_letter("refunds.dlq");
            b.include(review);
        })
}

Вручную

src/main.rs

fn app() -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("refunds", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(RedisBroker::standalone("redis://localhost:6379"), |b| {
            // Five deliveries of one refund, the first included; then the dead-letter stream.
            b.include(subscriber(refunds(), ProcessRefund).build())
                .max_attempts(nonzero!(5u32))
                .dead_letter("refunds.dlq");
            b.include(subscriber(priority_refunds(), ProcessRefund).build())
                .max_attempts(nonzero!(5u32))
                .dead_letter("refunds.dlq");
            b.include(
                subscriber(
                    RedisStream::new("refunds.dlq").group("refunds-review"),
                    Review,
                )
                .build(),
            );
        })
}

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

max_attempts считает все доставки одного возврата, включая первую. Пятая доставка не переобрабатывается: dead_letter публикует её в refunds.dlq с полезной нагрузкой и заголовками в том виде, в каком они пришли. Сообщение dead-letter записывается тем же способом, каким пришёл возврат, поэтому возврат из потока попадает в поток refunds.dlq. Предел без места назначения отклоняет доставку, когда попытки исчерпаны, а место назначения без предела забирает каждую переобработку на первой же попытке. Отброшенный возврат завершается и никуда не уходит. В refunds.dlq лежат только возвраты, у которых кончились попытки.

Предел читает один счётчик, и откуда он берётся, зависит от подписки. Подписка с перехватом читает счётчик доставок, который Redis хранит для записи. Этот счётчик учитывает и доставки остановившегося воркера, поэтому возврат, во время обработки которого воркер аварийно остановился, тоже расходует попытки. Любая другая подписка читает заголовок с числом попыток на копиях, которые публикует рантайм, и множество задержек тоже увеличивает его на каждом круге.

Перемотка, которая читает записи заново, доставляет их повторно как новые. Redis снова считает доставки такой записи с единицы, поэтому подписка с перехватом отсчитывает предел попыток с момента повторного чтения, а заголовок с числом попыток не меняется.

Чтение dead-letter

Макросы

src/main.rs

#[subscriber(RedisStream::new("refunds.dlq").group("refunds-review"))]
async fn review(refund: &Refund, ctx: &mut Context<'_>) -> HandlerOutcome {
    let retry_count = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("0");
    tracing::error!(
        refund.id,
        retry_count,
        "refund needs a person to look at it"
    );
    HandlerOutcome::ack()
}

Вручную

src/main.rs

struct Review;

impl Handle<Refund, (), (), (), Clients> for Review {
    fn handle(
        &self,
        refund: &Refund,
        _outs: &(),
        ctx: &mut Context<'_, (), Clients>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let retry_count = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("0");
        tracing::error!(
            refund.id,
            retry_count,
            "refund needs a person to look at it"
        );
        ready(Ok(()))
    }
}

refunds.dlq - обычный поток, и review читает его через собственную группу. Возврат, который прошёл через копии, содержит заголовок с числом попыток, здесь 5. review пишет возврат в журнал, чтобы его посмотрел человек.

Подписка по шаблону задаёт адрес переобработки

Копия уходит по адресу, который сообщает подписка: в ключ потока, в ключ списка или в канал. Подписка Pub/Sub по шаблону читает каждый канал, подходящий под её маску, и не называет ни одного конкретного канала. Её регистрация задаёт, куда уходит копия, через out_retry, например .out_retry(Publish::default()).to("events.retry"), и без этого сервис не запускается.

Проверка задержек

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