retry_after

retry_after(delay) on JetStream puts the delay in the negative acknowledgement; on Core the framework publishes a copy after the delay.

A JetStream consumer takes the delay as part of -NAK. The server holds the message for that long and delivers it again on the same consumer. Nothing is published, so the message keeps its stream sequence, and its delivery count rises by one. On a durable consumer the delayed message survives a restart of the service, because the server holds it.

A Core subject has no acknowledgement to carry a delay. The framework waits out the delay in the service's memory and publishes a copy of the message, payload and headers, back to the subject the subscription reads. The copy carries RETRY_COUNT_HEADER, raised by one on every copy. A process that exits before the delay is over loses the copy. A CoreWildcard reads many subjects, so its mount site names where the copy goes with out_retry. A cap declared at the mount site ends the retries of one message.

Example

A payment the ledger has not caught up with is checked again after 30 seconds, on a Core subject and on a stream.

Macros
Manual

examples/nats_retry.rs
/// The ledger has not caught up yet, so the delivery asks to come back later. On a Core subject
/// the framework publishes a copy of the payment to the subject once the delay is over.
#[subscriber("payments.settled")]
async fn reconcile(payment: &Payment) -> HandlerOutcome {
    if payment.settled {
        return HandlerOutcome::ack();
    }
    tracing::info!(payment.id, "not settled yet, checking again later");
    HandlerOutcome::retry_after(RECHECK)
}

/// The same work through a stream: the delay rides the negative acknowledgement, and the server
/// holds the message and delivers it again on this consumer.
#[subscriber(JetStreamSubject::new("payments.stored", "PAYMENTS").durable("reconciler"))]
async fn store(payment: &Payment) -> HandlerOutcome {
    if payment.settled {
        return HandlerOutcome::ack();
    }
    tracing::info!(payment.id, "not settled yet, checking again later");
    HandlerOutcome::retry_after(RECHECK)
}

examples/manual/nats_retry.rs
/// The ledger has not caught up yet, so the delivery asks to come back later. Mounted on a Core
/// subject, the framework publishes a copy of the payment to the subject once the delay is over;
/// mounted on a stream consumer, the delay rides the negative acknowledgement and the server
/// holds the message.
struct Reconcile;

impl Handle<Payment> for Reconcile {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if payment.settled {
            return ready(Ok(()));
        }
        tracing::info!(payment.id, "not settled yet, checking again later");
        ready(Err(HandlerOutcome::retry_after(RECHECK)))
    }
}

Example​

Example