结算：retry、retry_after、死信

支付服务商每次拒绝，charge 都给出所需的结果：立即重试、稍后重试、丢弃，或者转入死信主题。

每次投递都以一个结果结束。 ack 完成它，drop 丢弃它，retry 请求 Broker 再投递一次。 retry_after 在一段延迟之后再投递一次，不会阻塞排在它后面的投递。一直不成功的重试会无休止地回来，因此挂载点限定尝试次数。超过上限后，投递进入死信主题，运维人员或其他服务可以在那里查看它。

订阅者

支付服务商有三种拒绝方式，每种需要不同的结果。 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()
}

内存 Broker 自己完成重新投递，延迟正是 retry_after 要求的时长。如果 Broker 无法延迟一次投递，就把一份副本发布回该订阅。重试副本的去向讲解了这一点。第 8 章测试全部三条路径，包括延迟，而且不必真的等待。

订阅者​

上限与死信主题​

订阅者

上限与死信主题