Skip to main content

Retries and dead letters

This guide retries a Pulsar message now or later, caps the deliveries, and moves what keeps failing to a dead-letter topic that a handler reads.

Pulsar redelivers a message itself, so a retry publishes nothing. The mount site declares the cap and the dead-letter topic, and the Pulsar client applies them.

The example is a refunds service that sends refunds through a card processor. Processor stands in for the processor's client, and the app opens it at startup.

The subscriber​

The 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.

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)]
#[outgoing(name = "refunds")]
struct Refund {
    id: u64,
    card: String,
    amount_cents: u64,
}

/// A refund left unsettled for two minutes comes back; a wait must be shorter than that.
#[subscriber(PulsarSubscription::new("refunds", "refunds").ack_timeout(Duration::from_secs(120)))]
async fn process_refund(
    refund: &Refund,
    State(processor): State<Arc<Processor>>,
) -> HandlerOutcome {
    match processor.refund(refund).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()
        }
    }
}

retry negatively acknowledges the refund, and the broker delivers it again at once. retry_after holds the refund unacknowledged in the service for the delay, then negatively acknowledges it. Pulsar's negative acknowledgement carries no delay of its own, so the wait belongs to the process. A process that stops during the wait loses the wait, not the refund: the broker delivers it again after the acknowledgement timeout, or at once when the consumer disconnects.

The delay has to be shorter than the subscription's acknowledgement timeout, or the timeout would bring the refund back first. A longer delay is refused when the handler returns it, with an error that names both values. A subscription without an acknowledgement timeout accepts any delay.

The cap and the dead-letter topic​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("refunds", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(PulsarBroker::new("pulsar://localhost:6650"), |b| {
            // Five deliveries of one refund, the first included; then the dead-letter topic.
            b.include(process_refund)
                .max_attempts(nonzero!(5u32))
                .dead_letter("refunds-dlq");
            b.include(review);
        })
}

max_attempts counts every delivery of one refund, the first included. dead_letter names where a refund goes once the deliveries are spent. The pair becomes the consumer's own dead-letter policy. The Pulsar client counts the redeliveries, publishes the fifth failed refund to refunds-dlq and acknowledges the original. A retry, a retry_after and an expired acknowledgement timeout each count as a redelivery. A dropped refund settles and goes nowhere; the dead-letter topic holds only the refunds that ran out of attempts.

Write both steps or neither: a cap without a destination, or a destination without a cap, stops the service at startup, and the error names the missing step. The broker counts redeliveries only on a Shared or a KeyShared subscription, so the cap belongs there. A cap on an Exclusive or a Failover subscription would never be reached, and such a registration stops the service at startup with an error that names the type. A handler mounted by a bare topic name takes the same declaration, on the broker's default subscription.

Reading the dead-letter topic​

The dead-letter topic is an ordinary topic, and review reads it like any other.

src/main.rs
/// Reads the refunds that ran out of attempts.
#[subscriber(PulsarSubscription::new("refunds-dlq", "refunds-review"))]
async fn review(refund: &Refund) -> HandlerOutcome {
    tracing::error!(refund.id, "refund needs a person to look at it");
    HandlerOutcome::ack()
}

A dead-lettered refund keeps the payload and the headers it arrived with. The client adds two properties: REAL_TOPIC, the topic the refund came from, and ORIGIN_MESSAGE_ID, its position there. review logs the refund for a person to look at.

Testing the delay​

The tests in the example file run the app in process with TestApp::with_state, which hands the handlers a processor that fails on purpose. The delay test starts on a paused clock, so tb.advance(BACKOFF) lets the 30 seconds pass at once. The in-process Pulsar counts the deliveries and moves the spent refund to refunds-dlq the way the client does. Testing a Pulsar service covers the harness.