Skip to main content

Acknowledgements, retries and dead letters

This guide settles RabbitMQ deliveries, retries them now or later, caps the attempts and moves what keeps failing to a dead-letter queue.

The example is a refunds service. Refunds arrive on a classic queue, payouts on a quorum queue, and both go to a dead-letter queue after five deliveries.

What an outcome sends​

A handler's outcome becomes an AMQP frame on the delivery's channel:

OutcomeFrameWhat RabbitMQ does
HandlerOutcome::ack()basic.ackremoves the message from the queue
HandlerOutcome::retry()basic.reject with requeueputs the message back in the queue
HandlerOutcome::drop()basic.reject without requeueremoves the message, or dead-letters it where the queue has a dead-letter exchange
HandlerOutcome::retry_after(delay)an acknowledged delivery and a delayed copydelivers the copy again once the delay has passed

RabbitMQ counts every rejection as a delivery the message has spent. So a handler that keeps asking for its message back runs a quorum queue's own limit down. A delivery the service took and never settled goes back to the queue when the channel or the connection closes.

The subscriber​

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

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Refund {
    id: u64,
    card: String,
    amount_cents: u64,
}

/// Refunds wait out a delay in the broker's waiting queue, `refunds.retry`.
#[subscriber(RabbitQueue::new("refunds").delay(Delay::dlx_ttl()))]
async fn process_refund(
    refund: &Refund,
    State(processor): State<Arc<Processor>>,
) -> HandlerOutcome {
    match processor.refund(&refund.card).await {
        Ok(()) => HandlerOutcome::ack(),
        // Past the rate limit: a delivery right now would be refused again.
        Err(Refusal::Busy) => HandlerOutcome::retry_after(BACKOFF),
        // No answer: try again at once, up to the attempt cap.
        Err(Refusal::Unavailable) => HandlerOutcome::retry(),
        Err(Refusal::Declined) => {
            tracing::warn!(refund.id, "card declined, refund dropped");
            HandlerOutcome::drop()
        }
    }
}

The card processor refuses a refund in three ways, and each needs a different outcome. Busy means its rate limit is used up: a retry right now would be refused again, so retry_after waits 30 seconds. Unavailable means it did not answer: retry tries again at once. Declined means the card account is closed, and another try gets the same answer, so the refund is dropped.

The cap on a classic queue​

src/main.rs
fn service(broker: LapinBroker) -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("refunds", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(broker, |b| {
            // The runtime counts these five deliveries, on the copies it publishes back.
            b.include(process_refund)
                .max_attempts(nonzero!(5u32))
                .dead_letter("refunds.dead");
            b.include(review_refund);
            // The server counts these: they become the queue's delivery limit and route.
            b.include(send_payout)
                .max_attempts(nonzero!(5u32))
                .dead_letter("payouts.dead");
            b.include(review_payout);
        })
}

#[ruststream::app]
fn app() -> impl App<State = Clients> {
    service(LapinBroker::new("amqp://localhost:5672").declare_topology(true))
}

max_attempts counts every delivery of one refund, the first included. The fifth delivery is not retried: dead_letter publishes it to refunds.dead. A destination is a routing key on the default exchange, so the refund lands in the queue of that name.

A classic queue counts nothing, so the cap is the service's. Under a cap, a retry acknowledges the delivery and publishes a copy back to the queue under its own name. The copy carries the attempt count in the x-ruststream-retry-count header (RETRY_COUNT_HEADER), raised by one. The copies leave through the broker's default publish policy; .out_retry(policy) names another, for example TransactionalPublish to wait for RabbitMQ's confirm of each copy.

A dropped refund does not reach refunds.dead: the dead-letter queue holds only the refunds that ran out of attempts. To keep dropped messages too, give the queue its own dead-letter exchange (Topology).

The cap on a quorum queue​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Payout {
    id: u64,
    card: String,
    amount_cents: u64,
}

/// Payouts live in a quorum queue, which counts their deliveries itself.
#[subscriber(RabbitQuorumQueue::new("payouts"))]
async fn send_payout(payout: &Payout, State(processor): State<Arc<Processor>>) -> HandlerOutcome {
    match processor.refund(&payout.card).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(Refusal::Busy | Refusal::Unavailable) => HandlerOutcome::retry(),
        Err(Refusal::Declined) => HandlerOutcome::drop(),
    }
}

A quorum queue counts the deliveries of each message itself. On payouts, the same two steps at the mount site become the queue's x-delivery-limit and its dead-letter route. The server then moves a payout to payouts.dead after its fifth delivery, whether or not a service is running, and the service publishes nothing. A retry stays a plain basic.reject with requeue.

The arguments reach the queue only where the service declares it, so this needs declare_topology(true). On a broker that declares nothing, the service refuses to start: a queue declared elsewhere carries its owner's arguments, and the cap would apply nowhere. There, set the limit and the route on the queue itself, by its declaration or a policy, and mount the handler without the two steps. Half a declaration is refused too, because a limit with nowhere to send the spent message drops it. .out_retry(..) does not compile on a quorum queue: there is no copy to publish.

A quorum queue declared without a limit has RabbitMQ 4's default of 20 deliveries; delivery_limit sets another on the descriptor. A quorum delivery reports how many deliveries it has spent through redelivery_count(). A classic delivery reports only Redelivered, whether RabbitMQ handed it out before.

Delayed retries​

AMQP has no per-message delay. Without a delay on the queue, retry_after holds the copy in the service and publishes it when the delay has passed. That copy is at-most-once: a service that stops during the delay loses it.

delay(Delay::dlx_ttl()) moves the wait to RabbitMQ. The copy waits in the queue refunds.retry under a per-message TTL, and the waiting queue dead-letters it back to refunds when the TTL fires. The original is acknowledged after the copy is published on the same channel, so a failure between the two duplicates the refund rather than losing it. The copy is a new message, so it carries the retry count raised by one, and the cap ends a delayed loop as it ends an immediate one.

A waiting queue releases messages only from its head. A message with a long delay holds back every message behind it, so mixed delays want one waiting queue per delay, named with Delay::dlx_ttl_named. The plugin-dme feature adds Delay::plugin_dme, which waits in RabbitMQ's delayed-message-exchange plugin, where a short delay never waits behind a long one.

The waiting queue and the delayed exchange are infrastructure. Under declare_topology(true) the service declares them. With declaration off, a subscription with a delay checks at startup that its waiting queue or its delayed exchange exists, and the service does not start without it.

Reading the dead-letter queue​

src/main.rs
#[subscriber("refunds.dead")]
async fn review_refund(refund: &Refund, ctx: &mut Context<'_>) -> HandlerOutcome {
    let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
    tracing::error!(refund.id, deliveries, "refund needs a person to look at it");
    HandlerOutcome::ack()
}

#[subscriber("payouts.dead")]
async fn review_payout(payout: &Payout) -> HandlerOutcome {
    tracing::error!(payout.id, "payout needs a person to look at it");
    HandlerOutcome::ack()
}

A dead-letter queue is an ordinary queue, and a handler reads it like any other. A refund the service dead-lettered carries the retry-count header with the number of deliveries it went through, 5 here. A payout the quorum queue dead-lettered carries RabbitMQ's own x-death header instead. Both handlers log the message for a person to look at.

Testing​

The tests in the example file run the app in process with a paused clock, so tb.advance(BACKOFF) lets the 30 seconds pass at once. The in-process RabbitMQ releases the waiting queue's copy, counts a quorum queue's deliveries and dead-letters a spent one through its route, as the server does. The delayed-message-exchange plugin works only on a server, so its tests run against a live RabbitMQ. Testing a RabbitMQ service covers the harness.