Retries

max_attempts(n) caps the deliveries of one message; on JetStream it counts the server's own deliveries, on Core the copies in a header.

It is declared at the mount site, and the first delivery counts as one. A JetStream consumer reports how many times it has delivered a message, so the cap reads that number. Every delivery counts: a retry, a retry_after, and a redelivery after ack_wait. A Core subject delivers each message once, so the framework counts the copies it publishes, in RETRY_COUNT_HEADER. Under a cap every retry on a Core subject travels as such a copy, an immediate one too. The delivery that reaches the cap goes to the dead-letter subject, or is terminated when the mount site names none. The cap is the framework's own: the consumer the crate creates leaves max_deliver at the server's default, unlimited, so the server delivers a retried message again until the cap ends it.

Example

Five deliveries of one payment, on a Core subject and on a stream; the fifth that still fails goes to payments.dead.

Macros
Manual

examples/nats_retry.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            // Five deliveries of one payment, the first included; the fifth that still fails goes
            // to "payments.dead" instead, payload and headers as they arrived.
            b.include(reconcile)
                .max_attempts(nonzero!(5u32))
                .dead_letter("payments.dead");
            b.include(store)
                .max_attempts(nonzero!(5u32))
                .dead_letter("payments.dead");
            b.include(on_dead_letter);
        },
    )
}

examples/manual/nats_retry.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        NatsBroker::new("nats://localhost:4222"),
        |b| {
            // Five deliveries of one payment, the first included; the fifth that still fails goes
            // to "payments.dead" instead, payload and headers as they arrived.
            b.include(subscriber("payments.settled", Reconcile).build())
                .max_attempts(nonzero!(5u32))
                .dead_letter("payments.dead");
            let stored = JetStreamSubject::new("payments.stored", "PAYMENTS").durable("reconciler");
            b.include(subscriber(stored, Reconcile).build())
                .max_attempts(nonzero!(5u32))
                .dead_letter("payments.dead");
            b.include(subscriber("payments.dead", OnDeadLetter).build());
        },
    )
}

Example​

Example