Skip to main content

Retries and dead letters

This guide retries a delivery now or later, caps the attempts and keeps what keeps failing in a dead-letter stream, on a stream consumer, a Core subject and a pattern.

NATS has no dead-letter mechanism. JetStream can cap the deliveries of a message, but it sends the last one nowhere. So the mount site declares the cap and the dead-letter subject, and the framework applies them to both models.

The example is a payments service that settles payments and sends refunds through a bank. It keeps its dead letters in a stream of their own:

nats stream add PAYMENTS --subjects 'payments.settle' --defaults
nats stream add DEAD --subjects 'dead.*' --defaults

Retrying on a stream​

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

/// The outcome a refusal of the bank calls for.
fn outcome(refusal: Refusal) -> HandlerOutcome {
    match refusal {
        // Past the rate limit: a call right now would be refused again.
        Refusal::Busy => HandlerOutcome::retry_after(BACKOFF),
        // No answer: try again at once, up to the attempt cap.
        Refusal::Unavailable => HandlerOutcome::retry(),
        Refusal::Declined => HandlerOutcome::drop(),
    }
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "payments.settle")]
struct Payment {
    id: u64,
    card: String,
    amount_cents: u64,
}

#[subscriber(JetStreamSubject::new("payments.settle", "PAYMENTS").durable("settler"))]
async fn settle(payment: &Payment, State(bank): State<Arc<Bank>>) -> HandlerOutcome {
    match bank.charge(&payment.card).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(refusal) => {
            tracing::warn!(payment.id, ?refusal, "payment refused by the bank");
            outcome(refusal)
        }
    }
}

outcome turns the bank's refusal into the handler's outcome. A busy bank is asked again in 30 seconds with retry_after. On a JetStream consumer the delay travels in the negative acknowledgement: the server holds the payment and delivers it again on the same consumer, with the same stream sequence. Nothing is published, and a process that stops in between loses nothing. A silent bank is asked again at once with retry, and a declined card is dropped.

The cap and the dead letters​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("payments", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(NatsBroker::new("nats://localhost:4222"), |b| {
            // Five deliveries of one payment, the first included; then the dead-letter subject.
            b.include(settle)
                .max_attempts(nonzero!(5u32))
                .dead_letter("dead.payments");
            b.include(review);
            // A pattern is no subject to publish a copy to, so the mount site names one.
            b.include(send_refund)
                .max_attempts(nonzero!(3u32))
                .dead_letter("dead.refunds")
                .out_retry(Publish)
                .to("refunds.retry");
        })
}

max_attempts counts every delivery of one payment, the first included. On a JetStream consumer the count is the server's own, so an ack_wait that ran out counts too. The fifth delivery is not retried: dead_letter publishes it to dead.payments, payload and headers as they arrived. A dropped payment settles and goes nowhere, so the dead letters hold only the payments that ran out of attempts. A capped consumer without a dead-letter subject terminates its last delivery.

dead.payments is a plain publish, and Core NATS keeps nothing nobody reads. The DEAD stream stores it, so a dead letter waits until someone looks at it. No subject the app reads matches a dead-letter subject, or a spent delivery would come back to the handler that gave up on it.

Reading the dead letters​

src/main.rs
#[subscriber(JetStreamSubject::new("dead.payments", "DEAD").durable("review"))]
async fn review(payment: &Payment) -> HandlerOutcome {
    tracing::error!(payment.id, "payment needs a person to look at it");
    HandlerOutcome::ack()
}

review reads the dead payments from the DEAD stream like any other stream and logs each for a person to look at.

Retrying on Core subjects​

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

#[subscriber(CoreWildcard::new("refunds.*"))]
async fn send_refund(refund: &Refund, State(bank): State<Arc<Bank>>) -> HandlerOutcome {
    match bank.charge(&refund.card).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(refusal) => {
            tracing::warn!(refund.id, ?refusal, "refund refused by the bank");
            outcome(refusal)
        }
    }
}

Core NATS cannot hold a message back. On a Core subscription a retry is a copy of the delivery that the framework publishes: retry publishes it at once, retry_after when the delay ends. The copy carries the delivery count in its x-ruststream-retry-count header, which the cap reads. A delayed copy is at most once: a process that stops before the delay ends loses it.

A CoreSubject publishes its copies to its own subject. send_refund reads the pattern refunds.*, which has no single subject to publish to. So the app names one with out_retry(Publish).to("refunds.retry"), a subject the pattern matches, and the copies come back to send_refund. A registration over a pattern without it refuses to start, whether its handler retries or not. The third delivery of a refund goes to dead.refunds, with its count in the header.

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. Testing a NATS service covers the harness.