Skip to main content

Acknowledgement and delivery guarantees

This guide settles JetStream deliveries with the handler's outcome, sizes ack_wait for slow handlers, handles a message delivered twice, and shows what Core NATS does not promise.

A handler settles its delivery with the value it returns, and the runtime sends the settlement after the handler. On a JetStream consumer each outcome is an acknowledgement the server acts on:

OutcomeSent to the serverWhat the server does
ack+ACKforgets the message for this consumer
retry-NAKdelivers it again at once
retry_after(delay)-NAK with the delayholds it and delivers it again after the delay
drop+TERMnever delivers it again
no settlementnothingdelivers it again after ack_wait

A redelivery keeps the message's stream sequence, and the server counts its deliveries.

The example is a payments service. It captures authorized payments through the bank and announces each capture to a dashboard.

Settling a payment​

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

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

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "payments.captured")]
struct Captured {
    payment: u64,
    amount_cents: u64,
}

#[subscriber(
    JetStreamSubject::new("payments.authorized", "PAYMENTS")
        .durable("capture")
        .ack_wait(NonZeroDuration::from_secs(nonzero!(60))),
    reply
)]
async fn capture(
    payment: &Payment,
    Ctx(delivered): Ctx<Delivered>,
    State(bank): State<Arc<Bank>>,
) -> Result<Captured, HandlerOutcome> {
    if delivered.is_some_and(|count| count > 1) {
        tracing::info!(payment.id, ?delivered, "payment delivered again");
    }
    match bank.capture(payment).await {
        Ok(()) => Ok(Captured {
            payment: payment.id,
            amount_cents: payment.amount_cents,
        }),
        // Past the rate limit: the server holds the payment and delivers it after the delay.
        Err(Refusal::Busy) => Err(HandlerOutcome::retry_after(BACKOFF)),
        // No answer: the server delivers the payment again at once.
        Err(Refusal::Unavailable) => Err(HandlerOutcome::retry()),
        // The server never delivers a terminated payment again.
        Err(Refusal::Declined) => {
            tracing::warn!(payment.id, "card declined, payment terminated");
            Err(HandlerOutcome::drop())
        }
    }
}

The bank answers in four ways, and each needs a different outcome. A capture returns Captured, which the runtime publishes as the reply and then acknowledges the payment. Busy means the bank's rate limit is used up, so the payment waits 30 seconds on the server. Unavailable means the bank did not answer, so the payment comes back at once. Declined means the card cannot be charged, and another try gets the same answer, so the payment is terminated. A retry without a cap comes back forever; Retries and dead letters adds the cap and a dead-letter subject.

Messages delivered twice​

JetStream delivers at least once. A payment comes back after a retry, after ack_wait ran out, and after a process stopped before it settled. So a handler sees some payments twice, and its work has to accept a repeat. Here the bank takes the payment id as an idempotency key and charges a repeated capture once. Ctx<Delivered> is the server's delivery count, 1 on the first delivery, and capture logs a payment it sees again.

Slow handlers and ack_wait​

The server waits ack_wait for a settlement, 30 seconds unless the descriptor names another. After that it delivers the message again, while the first handler may still run. capture sets 60 seconds, above the longest time the bank takes to answer. Set ack_wait above the worst time of the handler, and the two deliveries never overlap. An expired ack_wait also counts as a delivery for the attempt cap.

The dashboard on Core NATS​

src/main.rs
#[subscriber("payments.captured")]
async fn refresh(captured: &Captured) -> HandlerOutcome {
    tracing::info!(captured.payment, "dashboard refreshed");
    HandlerOutcome::ack()
}

refresh reads the captures on a Core NATS subject. Core NATS delivers at most once and has no acknowledgement. Settling a Core delivery reports AckError::Unsupported, and the runtime logs it as a warning. A Core message the process loses is gone, and nothing brings it back. The dashboard can lose an update, because the next capture refreshes it again. A payment cannot, so payments travel through a stream.

The app​

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| {
            b.include(capture);
            b.include(refresh);
        })
}

The bank's client is the app's state, opened once at startup. A batch handler settles its whole batch with one outcome, or each message on its own when it returns Vec<HandlerOutcome>.

The tests in the example file run the app in process with a paused clock, so the 30 seconds of Busy pass at once.