Подтверждение и гарантии доставки

Это руководство завершает доставки JetStream исходом обработчика, подбирает ack_wait для медленных обработчиков, обрабатывает сообщение, доставленное дважды, и показывает, чего не обещает Core NATS.

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

Исход	Что получает сервер	Что делает сервер
ack	+ACK	забывает сообщение для этого консьюмера
retry	-NAK	сразу доставляет его повторно
retry_after(delay)	-NAK с задержкой	придерживает его и доставляет повторно после задержки
drop	+TERM	больше никогда его не доставляет
нет исхода	ничего	доставляет его повторно после ack_wait

Повторная доставка сохраняет порядковый номер сообщения в стриме, а сервер считает его доставки.

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

Завершение платежа

Макросы

src/main.rs

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

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

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "payments.captured")]
struct Captured {
    payment: u64,
    amount_cents: u64,
}

#[subscriber(
    JetStreamSubject::new("payments.authorized", "PAYMENTS")
        .durable("capture")
        .ack_wait(NonZeroDuration::from_secs(nonzero!(60))),
    reply
)]
async fn capture(
    payment: &Payment,
    Ctx(delivered): Ctx<Delivered>,
    State(bank): State<Arc<Bank>>,
) -> Result<Captured, HandlerOutcome> {
    if delivered.is_some_and(|count| count > 1) {
        tracing::info!(payment.id, ?delivered, "payment delivered again");
    }
    match bank.capture(payment).await {
        Ok(()) => Ok(Captured {
            payment: payment.id,
            amount_cents: payment.amount_cents,
        }),
        // Past the rate limit: the server holds the payment and delivers it after the delay.
        Err(Refusal::Busy) => Err(HandlerOutcome::retry_after(BACKOFF)),
        // No answer: the server delivers the payment again at once.
        Err(Refusal::Unavailable) => Err(HandlerOutcome::retry()),
        // The server never delivers a terminated payment again.
        Err(Refusal::Declined) => {
            tracing::warn!(payment.id, "card declined, payment terminated");
            Err(HandlerOutcome::drop())
        }
    }
}

Вручную

src/main.rs

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

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

impl OutgoingDestination for Payment {
    type Form = FixedName;

    const DESTINATION: &'static str = "payments.authorized";
}

impl MessageHeaders for Payment {
    type Contract = NoHeaders;
}

impl MessageInfo for Payment {
    const NAME: &'static str = "Payment";
}

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Captured {
    payment: u64,
    amount_cents: u64,
}

impl OutgoingDestination for Captured {
    type Form = FixedName;

    const DESTINATION: &'static str = "payments.captured";
}

impl MessageHeaders for Captured {
    type Contract = NoHeaders;
}

impl MessageInfo for Captured {
    const NAME: &'static str = "Captured";
}

struct Capture;

impl Handle<Payment, Captured, (), JetStreamContext, Clients> for Capture {
    async fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        ctx: &mut Context<'_, JetStreamContext, Clients>,
    ) -> Result<Captured, HandlerOutcome> {
        let delivered = ctx.context(DELIVERED);
        if delivered.is_some_and(|count| count > 1) {
            tracing::info!(payment.id, ?delivered, "payment delivered again");
        }
        match ctx.state().bank.capture(payment).await {
            Ok(()) => Ok(Captured {
                payment: payment.id,
                amount_cents: payment.amount_cents,
            }),
            // Past the rate limit: the server holds the payment and delivers it after the delay.
            Err(Refusal::Busy) => Err(HandlerOutcome::retry_after(BACKOFF)),
            // No answer: the server delivers the payment again at once.
            Err(Refusal::Unavailable) => Err(HandlerOutcome::retry()),
            // The server never delivers a terminated payment again.
            Err(Refusal::Declined) => {
                tracing::warn!(payment.id, "card declined, payment terminated");
                Err(HandlerOutcome::drop())
            }
        }
    }
}

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

Сообщения, доставленные дважды

JetStream доставляет сообщения at-least-once. Платёж возвращается после retry, после истечения ack_wait и после остановки процесса, который не успел завершить его доставку. Поэтому обработчик видит некоторые платежи дважды, и его работа должна допускать повтор. Здесь банк принимает идентификатор платежа как ключ идемпотентности и списывает повторный платёж один раз. Ctx<Delivered> - счётчик доставок сервера, на первой доставке он равен 1. capture пишет в журнал платёж, который видит снова.

Медленные обработчики и ack_wait

Сервер ждёт исхода ack_wait: 30 секунд, если дескриптор не задаёт другое значение. Затем он доставляет сообщение повторно, хотя первый обработчик может ещё работать. capture задаёт 60 секунд, больше самого долгого ответа банка. Задайте ack_wait больше худшего времени обработчика, и две доставки никогда не пересекутся. Истёкший ack_wait тоже считается доставкой для предела попыток.

Панель мониторинга на Core NATS

Макросы

src/main.rs

#[subscriber("payments.captured")]
async fn refresh(captured: &Captured) -> HandlerOutcome {
    tracing::info!(captured.payment, "dashboard refreshed");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

struct Refresh;

impl Handle<Captured, (), (), (), Clients> for Refresh {
    fn handle(
        &self,
        captured: &Captured,
        _outs: &(),
        _ctx: &mut Context<'_, (), Clients>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(captured.payment, "dashboard refreshed");
        ready(Ok(()))
    }
}

refresh читает списания из субъекта Core NATS. Core NATS доставляет at-most-once, и подтверждения в нём нет. Завершение доставки Core возвращает AckError::Unsupported, и рантайм пишет это в журнал как предупреждение. Сообщение Core, которое потерял процесс, пропадает, и ничто его не вернёт. Панель может пропустить обновление, потому что следующее списание снова её обновит. Платёж пропасть не может, поэтому платежи проходят через стрим.

Приложение

Макросы

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| {
            b.include(capture);
            b.include(refresh);
        })
}

Вручную

src/main.rs

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| {
            b.include(
                subscriber(
                    JetStreamSubject::new("payments.authorized", "PAYMENTS")
                        .durable("capture")
                        .ack_wait(NonZeroDuration::from_secs(nonzero!(60))),
                    Capture,
                )
                .reply()
                .build(),
            );
            b.include(subscriber("payments.captured", Refresh).build());
        })
}

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

Клиент банка - состояние приложения, его открывают один раз при запуске. Обработчик пакетов завершает весь пакет одним исходом или каждое сообщение отдельно, если возвращает Vec<HandlerOutcome>.

Тесты в файле примера запускают приложение внутри процесса с остановленными часами, поэтому 30 секунд Busy проходят мгновенно.