reclaim

reclaim reads only entries another consumer of the group fetched and never acked, idle at least min_idle, with XAUTOCLAIM.

It is the recovery path for a consumer that crashed or hung while it held entries. It runs as a second subscription on the same group, beside a RedisStream::new one that reads the fresh tail. A fresh entry is no one's unfinished work, so a reclaim subscription never reads it. 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. The subscription scans the pending entries list, and between two scans that found nothing it waits its block. A reclaimed delivery carries the server's count of its deliveries, this one included, in the DELIVERY_COUNT_HEADER header. A retry on a reclaim subscription appends a copy of the entry to the stream and acknowledges the original. On Redis 8.4 and later one claiming subscription covers both the fresh tail and the recovery.

Example

A worker reads the fresh tail of orders, and a recovery handler of the same group takes back what a worker left pending for a minute.

Macros
Manual

examples/redis_streams_reclaim.rs
/// The worker: fresh entries off the tail of `orders`.
#[subscriber(RedisStream::new("orders").group("workers"))]
async fn work(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order processed");
    HandlerOutcome::ack()
}

/// The recovery path of the same group: entries a worker took and left unacknowledged for a
/// minute, because it crashed or hung.
#[subscriber(RedisStream::reclaim("orders", Duration::from_secs(60)).group("workers"))]
async fn recover(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "order recovered");
    HandlerOutcome::ack()
}

fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(work);
        b.include(recover);
    })
}

examples/manual/redis_streams_reclaim.rs
/// The worker: fresh entries off the tail of `orders`.
struct Work;

impl Handle<Order> for Work {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order processed");
        ready(Ok(()))
    }
}

/// The recovery path: entries a worker took and left unacknowledged.
struct Recover;

impl Handle<Order> for Recover {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "order recovered");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379");
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        let fresh = RedisStream::new("orders").group("workers");
        b.include(subscriber(fresh, Work).build());
        // Idle a minute: the worker that took it crashed or hung.
        let stale = RedisStream::reclaim("orders", Duration::from_secs(60)).group("workers");
        b.include(subscriber(stale, Recover).build());
    })
}

Example​

Example