Attempts

max_attempts(n) ends a message that keeps coming back, and dead_letter(name) is where it goes then.

max_attempts(n) counts the deliveries one message gets, the first one included. The delivery that reaches the cap is not retried: it is published to the dead_letter destination with its payload and headers. A cap without a destination rejects the delivery instead, and a warning names the subscription. The count is the broker's own where the transport keeps one, read through redelivery_count, and the framework's RETRY_COUNT_HEADER where it keeps none. The in-memory broker counts its deliveries, so the cap there counts its redeliveries. A retry() and a retry_after(delay) are both capped. Where the broker moves a spent delivery itself (a queue with a delivery limit and a dead-letter exchange), the declaration reaches the subscription descriptor at startup, and the broker applies it. The two steps come right after include, before .out_retry(..), the slot the dead-letter copy leaves through, and each is written once.

Example

Macros

examples/retry.rs

/// The gateway rejects this payment every time, so on its own it would never stop coming back.
#[subscriber("payments")]
async fn charge(payment: &Payment) -> HandlerOutcome {
    tracing::warn!(payment.id, "gateway refused the charge");
    HandlerOutcome::retry()
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // Three deliveries, then the payment goes to the dead-letter subject.
        b.include(charge)
            .max_attempts(nonzero!(3u32))
            .dead_letter("payments.dead");
    })
}

Manual

examples/manual/retry.rs

/// The gateway rejects this payment every time, so on its own it would never stop coming back.
struct Charge;

impl Handle<Payment> for Charge {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::warn!(payment.id, "gateway refused the charge");
        ready(Err(HandlerOutcome::retry()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        // Three deliveries, then the payment goes to the dead-letter subject.
        b.include(subscriber("payments", Charge).build())
            .max_attempts(nonzero!(3u32))
            .dead_letter("payments.dead");
    })
}