retry_after

retry_after(delay) on a plain stream makes the runtime republish a copy after the delay, which a crash before the timer loses.

Redis has no per-message delay, so the runtime serves it. It settles the delivery, holds the payload and headers in memory, and publishes a copy once the delay is up. The copy goes where the subscription's descriptor says a copy goes: the stream key, the list key or the channel, and for a pattern the channel its registration names. It carries RETRY_COUNT_HEADER one higher than the delivery it replaces. The copy waits in the service's memory, so it is at most once over the delay: a process that exits before the timer fires loses it. A simple list and Pub/Sub settle nothing, and they get the copy all the same. The copy leaves through the registration's retry publisher, which out_retry replaces. A stream that has to keep the delayed message across a crash parks it in Redis with DelayedRetry. A claiming subscription keeps the entry pending instead, so the delay rounds up to its min_idle, and a longer delay is not held that long.

Example

A card payment the gateway timed out on is charged again 30 seconds later.

Macros
Manual

examples/redis_acknowledgement_retries.rs
/// A timed-out charge comes back after `RECHECK`: the runtime holds a copy and publishes it to
/// `card-payments` once the delay is up.
#[subscriber(RedisStream::new("card-payments").group("billing"))]
async fn charge_card(payment: &Payment, State(gateway): State<PaymentGateway>) -> HandlerOutcome {
    match gateway.charge(payment).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(payment.id, %err, "charge failed, retrying later");
            HandlerOutcome::retry_after(RECHECK)
        }
    }
}

examples/manual/redis_acknowledgement_retries.rs
/// A timed-out charge comes back after `RECHECK`: the runtime holds a copy and publishes it to
/// `card-payments` once the delay is up.
struct ChargeCard;

impl Handle<Payment, (), (), (), AppState> for ChargeCard {
    async fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        ctx: &mut Context<'_, (), AppState>,
    ) -> Result<(), HandlerOutcome> {
        if let Err(err) = ctx.state().gateway.charge(payment).await {
            tracing::warn!(payment.id, %err, "charge failed, retrying later");
            return Err(HandlerOutcome::retry_after(RECHECK));
        }
        Ok(())
    }
}

fn card_payments() -> RedisStream {
    RedisStream::new("card-payments").group("billing")
}

Example​

Example