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

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

Переобработка и недоставленные сообщения

Это руководство переобрабатывает доставку сразу или позже, ограничивает число попыток и сохраняет то, что раз за разом не обрабатывается, в стриме для недоставленных сообщений: на консьюмере стрима, на субъекте Core и на шаблоне.

В NATS нет механизма недоставленных сообщений. JetStream умеет ограничивать число доставок сообщения, но последнюю доставку никуда не отправляет. Поэтому предел и субъект для недоставленных сообщений объявляет точка монтирования, а фреймворк применяет их к обеим моделям.

Пример - платёжный сервис, который проводит платежи и отправляет возвраты через банк. Недоставленные сообщения он хранит в отдельном стриме:

nats stream add PAYMENTS --subjects 'payments.settle' --defaults
nats stream add DEAD --subjects 'dead.*' --defaults

Переобработка на стриме​

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

/// The outcome a refusal of the bank calls for.
fn outcome(refusal: Refusal) -> HandlerOutcome {
    match refusal {
        // Past the rate limit: a call right now would be refused again.
        Refusal::Busy => HandlerOutcome::retry_after(BACKOFF),
        // No answer: try again at once, up to the attempt cap.
        Refusal::Unavailable => HandlerOutcome::retry(),
        Refusal::Declined => HandlerOutcome::drop(),
    }
}

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

#[subscriber(JetStreamSubject::new("payments.settle", "PAYMENTS").durable("settler"))]
async fn settle(payment: &Payment, State(bank): State<Arc<Bank>>) -> HandlerOutcome {
    match bank.charge(&payment.card).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(refusal) => {
            tracing::warn!(payment.id, ?refusal, "payment refused by the bank");
            outcome(refusal)
        }
    }
}

outcome превращает отказ банка в исход обработчика. Если банк занят, retry_after повторяет обращение через 30 секунд. На консьюмере JetStream задержка передаётся в отрицательном подтверждении: сервер придерживает платёж и повторно доставляет его тому же консьюмеру с тем же порядковым номером в стриме. Ничего не публикуется, и процесс, остановленный в промежутке, ничего не теряет. Если банк молчит, retry повторяет обращение сразу, а платёж по отклонённой карте отбрасывается.

Предел попыток и недоставленные сообщения​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("payments", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(NatsBroker::new("nats://localhost:4222"), |b| {
            // Five deliveries of one payment, the first included; then the dead-letter subject.
            b.include(settle)
                .max_attempts(nonzero!(5u32))
                .dead_letter("dead.payments");
            b.include(review);
            // A pattern is no subject to publish a copy to, so the mount site names one.
            b.include(send_refund)
                .max_attempts(nonzero!(3u32))
                .dead_letter("dead.refunds")
                .out_retry(Publish)
                .to("refunds.retry");
        })
}

max_attempts считает все доставки одного платежа, включая первую. На консьюмере JetStream счёт ведёт сам сервер, поэтому истёкший ack_wait тоже учитывается. Пятая доставка не переобрабатывается: dead_letter публикует её в dead.payments с полезной нагрузкой и заголовками в том виде, в каком они пришли. Отброшенный платёж завершается и никуда не уходит, поэтому среди недоставленных сообщений лежат только платежи, у которых кончились попытки. Консьюмер с пределом, но без субъекта для недоставленных сообщений прекращает свою последнюю доставку.

dead.payments - обычная публикация, а Core NATS не хранит то, что никто не читает. Её сохраняет стрим DEAD, поэтому недоставленное сообщение ждёт, пока его кто-нибудь посмотрит. Ни один субъект, который читает приложение, не подходит к субъекту для недоставленных сообщений. Иначе исчерпанная доставка вернулась бы к обработчику, который от неё отказался.

Чтение недоставленных сообщений​

src/main.rs
#[subscriber(JetStreamSubject::new("dead.payments", "DEAD").durable("review"))]
async fn review(payment: &Payment) -> HandlerOutcome {
    tracing::error!(payment.id, "payment needs a person to look at it");
    HandlerOutcome::ack()
}

review читает недоставленные платежи из стрима DEAD, как любой другой стрим, и пишет каждый в журнал, чтобы его посмотрел человек.

Переобработка на субъектах Core​

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

#[subscriber(CoreWildcard::new("refunds.*"))]
async fn send_refund(refund: &Refund, State(bank): State<Arc<Bank>>) -> HandlerOutcome {
    match bank.charge(&refund.card).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(refusal) => {
            tracing::warn!(refund.id, ?refusal, "refund refused by the bank");
            outcome(refusal)
        }
    }
}

Core NATS не умеет придерживать сообщение. На подписке Core переобработка - это копия доставки, которую публикует фреймворк: retry публикует её сразу, retry_after - по истечении задержки. Число доставок копия хранит в заголовке x-ruststream-retry-count, и предел читает его оттуда. Отложенная копия доставляется at-most-once: процесс, который остановился до конца задержки, её теряет.

CoreSubject публикует свои копии в собственный субъект. send_refund читает шаблон refunds.*, у которого нет единственного субъекта для публикации. Поэтому приложение называет его через out_retry(Publish).to("refunds.retry"). Это субъект, к которому подходит шаблон, и копии возвращаются в send_refund. Регистрация на шаблон без него не запускается, просит её обработчик переобработку или нет. Третья доставка возврата уходит в dead.refunds с числом доставок в заголовке.

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