Перейти к основному содержимому
Нормативная версия документации - английская

Эта страница переведена с английского языковой моделью. При любом расхождении верен английский оригинал.

Повторы и потоки мёртвых писем

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

Kinesis не придерживает записи, не считает доставки, и у него нет DLQ. Все три задачи решает фреймворк: точка монтирования объявляет предел попыток и поток мёртвых писем, а повтор уходит копией записи с числом попыток в заголовке.

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

Подписчик​

Процессинг отказывает в возврате тремя способами, и каждому нужен свой исход. 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)]
#[outgoing(name = "refunds")]
struct Refund {
    id: u64,
    card: String,
    amount_cents: u64,
}

#[subscriber(KinesisStream::new("refunds").create_if_missing(1))]
async fn process_refund(
    refund: &Refund,
    State(processor): State<Arc<Processor>>,
) -> 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()
        }
    }
}

retry_after завершает запись и по истечении задержки публикует её копию обратно в refunds. Копия содержит заголовок с числом попыток (RETRY_COUNT_HEADER), увеличенным на единицу. До конца задержки копия хранится в памяти сервиса, поэтому доставляется at-most-once: процесс, который остановился до конца задержки, её теряет, а оригинал уже позади контрольной точки.

retry при объявленном пределе тоже становится копией, которая публикуется сразу. Без предела он держит контрольную точку сегмента перед записью, как показывает руководство Контрольные точки и гарантии доставки.

Предел попыток и поток мёртвых писем​

src/main.rs
fn service(broker: KinesisBroker) -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("refunds", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(broker, |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.review");
            b.include(review);
        })
}

#[ruststream::app]
fn app() -> impl App<State = Clients> {
    service(
        KinesisBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
    )
}

max_attempts считает все доставки одного возврата, включая первую. Пятая доставка не повторяется: dead_letter публикует её в refunds.review с полезной нагрузкой и заголовками в том виде, в каком они пришли. Предел без потока мёртвых писем вместо этого переносит контрольную точку за исчерпавшую попытки запись, как это делает drop. Отброшенный возврат завершается и никуда не уходит. В потоке мёртвых писем лежат только возвраты, у которых кончились попытки. Потоки принадлежат инфраструктуре сервиса. Пример создаёт их через create_if_missing для локального стенда.

Чтение потока мёртвых писем​

Поток мёртвых писем - обычный поток, и review читает его, как любой другой.

src/main.rs
#[subscriber(KinesisStream::new("refunds.review").create_if_missing(1))]
async fn review(refund: &Refund, ctx: &mut Context<'_>) -> HandlerOutcome {
    let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
    tracing::error!(refund.id, deliveries, "refund needs a person to look at it");
    HandlerOutcome::ack()
}

Заголовок с числом попыток у записи из потока мёртвых писем хранит число доставок, через которые она прошла, здесь 5. review читает его через ctx.headers() и пишет возврат в журнал, чтобы его посмотрел человек.

Копии и порядок​

Копия уходит обратно в поток, который читает подписка, поэтому регистрации на Kinesis не нужен собственный адрес для повторов. Копия сохраняет ключ сегмента записи, поэтому попадает в сегмент этого ключа. Но попадает она в конец этого сегмента, после записей, опубликованных, пока она ждала. Поэтому возврат, который ждёт 30 секунд, обрабатывается после возвратов той же карты, пришедших позже.

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

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

Тесты в файле примера запускают приложение внутри процесса с остановленными часами, поэтому tb.advance(BACKOFF) проматывает 30 секунд мгновенно. TestApp::with_state запускает приложение с подставным процессингом, который занят или молчит. Работу с TestApp описывает руководство Тестирование сервиса на Kinesis.