retry_after

retry_after(delay) holds the delivery unacknowledged in the service for the delay, then negatively acknowledges it; nothing is republished.

A Pulsar negative acknowledgement carries no delay of its own, so the wait happens in the service. When it is over, the subscription delivers the same message again, with its own payload, properties and position. The redelivery is an ordinary one, so a dead-letter policy counts it like any other. The wait runs on the runtime the broker connected on. A process that exits during the wait loses the wait, not the message: the broker delivers it again once the ack_timeout elapses, or at once when the consumer disconnects. A delay that is not shorter than the subscription's ack_timeout is refused when the handler returns it; a subscription with no ack_timeout accepts any delay. A test waits the delay out with tb.advance, in process on a paused clock and against a live broker in real time.

Retries and dead letters holds a refund back while the card processor's rate limit is used up.

Example

An invoice whose customer has not reached the directory yet comes back after the replication lag.

Macros
Manual

examples/pulsar_settlement_retry_after.rs
/// How long a customer the directory has not replicated yet gets before the next try.
const REPLICATION_LAG: Duration = Duration::from_secs(2);

/// Issues each invoice. An invoice whose customer is missing waits in the service for the lag,
/// then the broker delivers it again.
#[subscriber(PulsarSubscription::new("invoices", "billing"))]
async fn issue(invoice: &Invoice, State(directory): State<Directory>) -> HandlerOutcome {
    match directory.find(invoice.customer).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(invoice.id, %err, "customer not found, trying again later");
            HandlerOutcome::retry_after(REPLICATION_LAG)
        }
    }
}

examples/manual/pulsar_settlement_retry_after.rs
/// How long a customer the directory has not replicated yet gets before the next try.
const REPLICATION_LAG: Duration = Duration::from_secs(2);

/// Issues each invoice. An invoice whose customer is missing waits in the service for the lag,
/// then the broker delivers it again.
struct Issue;

impl Handle<Invoice, (), (), (), AppState> for Issue {
    async fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        ctx: &mut Context<'_, (), AppState>,
    ) -> Result<(), HandlerOutcome> {
        ctx.state()
            .directory
            .find(invoice.customer)
            .await
            .map_err(|err| {
                tracing::warn!(invoice.id, %err, "customer not found, trying again later");
                HandlerOutcome::retry_after(REPLICATION_LAG)
            })
    }
}

Example​

Example