Retries

max_attempts(n) and dead_letter(name) cap the deliveries of one message; on a classic queue the runtime keeps the count.

max_attempts(n) is how many deliveries one message gets, the first included. dead_letter(name) is where it goes once they run out. A classic queue counts nothing, so under a cap every retry, an immediate one too, travels as a copy. The runtime settles the delivery and publishes a copy of the message back to the queue. The copy carries RETRY_COUNT_HEADER (x-ruststream-retry-count) raised by one, and the handler reads it like any other header. The copy goes back under the queue's own name: on the default exchange a routing key addresses the queue of that name. The delivery that reaches the cap is published under name instead, so name is the queue the message lands in, with its count. A bare queue name, #[subscriber("payments")], takes the same cap.

The copies leave through the broker's default publish policy. out_retry(policy) names another one, and the position takes a slot's steps: .codec(..), .transform(..), .to(name). The example names the confirms policy, whose publish resolves once the broker has confirmed the copy. A retry policy aimed at a topic or direct exchange reaches the queue only through a binding under the queue's name. On a quorum queue the same two steps become the queue's own delivery limit.

Example

A declined card is retried at once. The third delivery is the last: the payment goes to payments.dead, where a second subscription reads the count.

Macros
Manual

examples/lapin_settlement_retries.rs
/// A card the gateway declines is asked for again, at once.
#[subscriber(RabbitQueue::new("payments"))]
async fn charge(payment: &Payment) -> HandlerOutcome {
    if payment.card.starts_with("0000") {
        tracing::warn!(payment.id, "card declined, retrying");
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// The payments that ran out of deliveries, with the count the runtime kept.
#[subscriber("payments.dead")]
async fn bury(payment: &Payment, ctx: &mut Context<'_>) -> HandlerOutcome {
    let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
    tracing::error!(payment.id, deliveries, "payment dead-lettered");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672");
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        // Three deliveries, the first included, then the payment goes to `payments.dead`. The
        // copies leave through the confirms policy, which waits for the broker to confirm each.
        b.include(charge)
            .max_attempts(nonzero!(3u32))
            .dead_letter("payments.dead")
            .out_retry(TransactionalPublish::default());
        b.include(bury);
    })
}

examples/manual/lapin_settlement_retries.rs
/// A card the gateway declines is asked for again, at once.
struct Charge;

impl Handle<Payment> for Charge {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if payment.card.starts_with("0000") {
            tracing::warn!(payment.id, "card declined, retrying");
            return ready(Err(HandlerOutcome::retry()));
        }
        ready(Ok(()))
    }
}

/// The payments that ran out of deliveries, with the count the runtime kept.
struct Bury;

impl Handle<Payment> for Bury {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
        tracing::error!(payment.id, deliveries, "payment dead-lettered");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672");
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        // Three deliveries, the first included, then the payment goes to `payments.dead`. The
        // copies leave through the confirms policy, which waits for the broker to confirm each.
        b.include(subscriber(RabbitQueue::new("payments"), Charge).build())
            .max_attempts(nonzero!(3u32))
            .dead_letter("payments.dead")
            .out_retry(TransactionalPublish::default());
        b.include(subscriber("payments.dead", Bury).build());
    })
}

Example​

Example