DelayedRetry

DelayedRetry::DurableZset parks a delayed delivery in a sorted set, so the retry survives a crash and fires at its due time.

A stream subscription takes it with RedisStream::delayed_retry. A retry_after then adds the entry to the sorted set with ZADD, scored by its due time, with its retry count raised by one. The original is acknowledged with XACK once the entry is parked. Each read of the subscription sweeps the set and appends the entries that are due back onto the stream, each under a new id. Nothing about the delay lives in the service, so a restart or a crash loses no retry. The sweep runs once per read, so the block interval sets how late a due entry may come back, and one pass moves at most 128 entries. Scores are wall-clock epoch milliseconds, so the clocks of the services that share a set have to agree. ttl expires a set no retry has refreshed for that long, and it has to exceed the longest delay, or entries are dropped before they are due. The set costs keys, memory and a sweep, so a subscription has it only when its descriptor names it. On a claiming subscription the set honours the delay as asked, instead of rounding it to min_idle.

Example

A bank payment the gateway timed out on waits in bank-payments.delayed for 30 seconds.

Macros
Manual

examples/redis_acknowledgement_retries.rs
/// The delay waits in the `bank-payments.delayed` sorted set in Redis, and the subscription puts
/// the entry back on `bank-payments` once it is due.
#[subscriber(
    RedisStream::new("bank-payments")
        .group("billing")
        .delayed_retry(DelayedRetry::DurableZset {
            key: "bank-payments.delayed".to_owned(),
            ttl: None,
        })
)]
async fn charge_bank(payment: &Payment, State(gateway): State<PaymentGateway>) -> HandlerOutcome {
    match gateway.charge(payment).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(payment.id, %err, "charge failed, parked for later");
            HandlerOutcome::retry_after(RECHECK)
        }
    }
}

examples/manual/redis_acknowledgement_retries.rs
/// The delay waits in the `bank-payments.delayed` sorted set in Redis, and the subscription puts
/// the entry back on `bank-payments` once it is due.
struct ChargeBank;

impl Handle<Payment, (), (), (), AppState> for ChargeBank {
    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, parked for later");
            return Err(HandlerOutcome::retry_after(RECHECK));
        }
        Ok(())
    }
}

fn bank_payments() -> RedisStream {
    RedisStream::new("bank-payments")
        .group("billing")
        .delayed_retry(DelayedRetry::DurableZset {
            key: "bank-payments.delayed".to_owned(),
            ttl: None,
        })
}

Example​

Example