claiming

claiming reads pending entries idle at least min_idle and the fresh tail in one XREADGROUP ... CLAIM call; it needs Redis 8.4.

The read takes the stale entries first, the longest idle first, and fills the rest with fresh ones. One subscription and one handler cover new work and recovery alike. min_idle has no default. Set it above the longest time a handler takes, or an entry a healthy consumer is still working on is taken over and handled twice. A retry leaves the entry in the pending entries list: nothing is appended, and the entry keeps its id. The same subscription reads it again once it has been idle min_idle. So a retry waits out min_idle, and a retry_after shorter than that waits min_idle too. Each delivery reports the attempts before it in the DELIVERY_COUNT_HEADER header, zero for an entry read off the tail, and its idle time in IDLE_MS_HEADER. A server older than 8.4 refuses the subscription, and the service does not start. The error, RedisError::ServerTooOld, names the subscription, the mode and both versions. On such a server the recovery is a reclaim subscription beside a new one.

Example

An order out of stock stays pending and comes back after thirty seconds.

Macros
Manual

examples/redis_streams_claiming.rs
/// Fulfils new orders and the ones a worker took and never finished. An order out of stock stays
/// pending, and the subscription takes it back once it has been idle 30 seconds.
#[subscriber(RedisStream::claiming("orders", Duration::from_secs(30)).group("fulfilment"))]
async fn fulfil(order: &Order) -> HandlerOutcome {
    if !order.in_stock {
        tracing::warn!(order.id, "out of stock, the order comes back");
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, "order fulfilled");
    HandlerOutcome::ack()
}

examples/manual/redis_streams_claiming.rs
/// Fulfils new orders and the ones a worker took and never finished. An order out of stock stays
/// pending, and the subscription takes it back once it has been idle 30 seconds.
struct Fulfil;

impl Handle<Order> for Fulfil {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if !order.in_stock {
            tracing::warn!(order.id, "out of stock, the order comes back");
            return ready(Err(HandlerOutcome::retry()));
        }
        tracing::info!(order.id, "order fulfilled");
        ready(Ok(()))
    }
}

fn app(url: &str) -> impl App {
    RustStream::new(AppInfo::new("fulfilment", "0.1.0")).with_broker(
        RedisBroker::standalone(url),
        |b| {
            let orders =
                RedisStream::claiming("orders", Duration::from_secs(30)).group("fulfilment");
            b.include(subscriber(orders, Fulfil).build());
        },
    )
}

Example​

Example