Завершение доставки: retry, retry_after, недоставленные сообщения

charge отвечает на каждый отказ платёжного провайдера нужным исходом: переобработкой сразу или позже, отбрасыванием или темой недоставленных сообщений.

Каждая доставка завершается одним исходом. ack завершает её, drop отбрасывает, а retry просит брокер доставить сообщение повторно. retry_after доставляет его повторно после задержки и не задерживает доставки, идущие следом. Переобработка, которая никогда не завершается успехом, повторялась бы бесконечно, поэтому точка монтирования ограничивает число попыток. Когда попытки исчерпаны, доставка уходит в тему для недоставленных сообщений (dead letter), где её может изучить оператор или другой сервис.

Подписчик

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

Макросы

payments.rs

/// How long a busy provider asks its callers to wait.
pub(crate) const BACKOFF: Duration = Duration::from_secs(1);

/// Charges an order through the payment provider.
#[subscriber("payments.requested", reply)]
async fn charge(
    request: &PaymentRequest,
    State(gateway): State<Arc<Gateway>>,
) -> Result<Paid, HandlerOutcome> {
    match gateway.charge(request).await {
        Ok(()) => Ok(Paid {
            order: request.order,
            customer: request.customer.clone(),
            lines: request.lines.clone(),
        }),
        // Past the rate limit: a redelivery 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()),
        // A refused card stays refused.
        Err(Refusal::Declined) => {
            tracing::warn!(order = request.order, "card declined, payment dropped");
            Err(HandlerOutcome::drop())
        }
    }
}

Вручную

payments.rs

/// How long a busy provider asks its callers to wait.
pub(crate) const BACKOFF: Duration = Duration::from_secs(1);

/// Charges an order through the payment provider.
struct Charge;

// The state type is the last parameter of `Handle`; the mount checks it against the app's.
impl Handle<PaymentRequest, Paid, (), (), Shop> for Charge {
    async fn handle(
        &self,
        request: &PaymentRequest,
        _outs: &(),
        ctx: &mut Context<'_, (), Shop>,
    ) -> Result<Paid, HandlerOutcome> {
        let State(gateway) = State::<Arc<Gateway>>::from_context(ctx).await?;
        match gateway.charge(request).await {
            Ok(()) => Ok(Paid {
                order: request.order,
                customer: request.customer.clone(),
                lines: request.lines.clone(),
            }),
            // Past the rate limit: a redelivery 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()),
            // A refused card stays refused.
            Err(Refusal::Declined) => {
                tracing::warn!(order = request.order, "card declined, payment dropped");
                Err(HandlerOutcome::drop())
            }
        }
    }
}

Ограничение попыток и тема недоставленных сообщений

max_attempts считает все доставки одного сообщения, включая первую. Пятая доставка не переобрабатывается: dead_letter публикует её в payments.dead с полезной нагрузкой и заголовками в том виде, в каком они пришли. Роутер декодирует запросы из CBOR и так же кодирует ответ.

Макросы

payments.rs

/// The payments side's subscriber: five deliveries at most, then the dead-letter topic.
pub(crate) fn routes() -> impl RouterDef<MemoryBroker, Shop> {
    Router::new()
        .with_codec(INTERNAL)
        .include(charge)
        .max_attempts(nonzero!(5u32))
        .dead_letter("payments.dead")
        .out_reply(Publish)
        .codec(INTERNAL)
        .build()
}

Вручную

payments.rs

/// The payments side's subscriber: five deliveries at most, then the dead-letter topic.
pub(crate) fn routes() -> impl RouterDef<MemoryBroker, Shop> {
    Router::new()
        .with_codec(INTERNAL)
        .include(
            subscriber(PaymentRequest::DESTINATION, Charge)
                .describe("Charges an order through the payment provider.")
                .reply()
                .build(),
        )
        .max_attempts(nonzero!(5u32))
        .dead_letter("payments.dead")
        .out_reply(Publish)
        .codec(INTERNAL)
        .build()
}

In-memory брокер сам доставляет сообщение повторно после задержки, которую запросил retry_after. Если брокер не умеет откладывать доставку, рантайм публикует копию сообщения обратно в подписку. Подробнее об этом - в разделе куда уходит копия для переобработки. Глава 8 проверяет все три пути, включая задержку, и не ждёт её истечения.