Acknowledgement

A JetStream delivery settles on the server: ack acknowledges it, retry sends a negative acknowledgement, drop terminates it.

A JetStreamSubject reads a stream through a consumer, and the consumer tracks every message it hands out until the message is settled. The handler's outcome becomes the acknowledgement the server receives. ack sends +ACK, and the consumer moves past the message. retry sends -NAK, and the consumer delivers the message again at once, with its delivery count raised by one. drop sends +TERM, and the server stops delivering the message. In the manual form Ok acknowledges and Err(outcome) settles by the outcome. A delivery nobody settles comes back once the consumer's ack_wait has passed. retry_after asks for the next delivery after a delay, and max_attempts caps how many deliveries one message gets. A Core NATS subscription has no acknowledgement at all: its deliveries are processed at most once.

Example

The first reservation after startup fails while the stock count loads, so its order is delivered again; an order of nothing is terminated.

Macros
Manual

examples/nats_settlement_ack.rs
/// Reserves the stock of each order. The outcome settles the delivery on the server: `ack` is
/// `+ACK`, `retry` is `-NAK` and the consumer delivers the order again, `drop` is `+TERM`.
#[subscriber(JetStreamSubject::new("orders.placed", "ORDERS").durable("fulfilment"))]
async fn fulfil(order: &Order, State(warehouse): State<Warehouse>) -> HandlerOutcome {
    if order.quantity == 0 {
        tracing::warn!(order.id, "an order of nothing is terminated");
        return HandlerOutcome::drop();
    }
    match warehouse.reserve(order).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(order.id, %err, "reservation failed, redelivering");
            HandlerOutcome::retry()
        }
    }
}

examples/manual/nats_settlement_ack.rs
/// Reserves the stock of each order. `Ok` is `+ACK`; `Err(retry)` is `-NAK` and the consumer
/// delivers the order again; `Err(drop)` is `+TERM`.
struct Fulfil;

impl Handle<Order, (), (), (), AppState> for Fulfil {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, (), AppState>,
    ) -> Result<(), HandlerOutcome> {
        if order.quantity == 0 {
            tracing::warn!(order.id, "an order of nothing is terminated");
            return Err(HandlerOutcome::drop());
        }
        ctx.state().warehouse.reserve(order).await.map_err(|err| {
            tracing::warn!(order.id, %err, "reservation failed, redelivering");
            HandlerOutcome::retry()
        })
    }
}

Example​

Example