Recovery

recovery_zset starts a watchdog that returns entries a dead consumer left on its processing list, once idle min_idle.

A reliable list keeps no idle time of its own, so the subscription records every claim in the named sorted set, scored by the time of the claim. Each read of the subscription sweeps the set and pushes an entry claimed more than min_idle ago back onto the queue, where a live consumer takes it. An ack or a drop removes the entry's record from the set. min_idle has no default, and a recovery set without it does not start. It has to exceed the longest time one entry takes to handle: a slower handler has its entry returned and handled twice. The sweep runs once per read, so the block interval sets how soon a due entry returns, and one pass moves at most 128 entries. recovery_ttl expires the set once no claim refreshes it, and it has to exceed min_idle. recovery_zset turns the reliable form on by itself.

Example

An invoice claimed by a worker that died is sent by another one a minute later.

Macros

examples/redis_lists.rs

/// Every claim is recorded in `invoices.claims`. An entry still unsettled after `MIN_IDLE` goes
/// back on `invoices`, so set it above the longest time one invoice takes.
#[subscriber(
    RedisList::new("invoices")
        .reliable()
        .min_idle(MIN_IDLE)
        .recovery_zset("invoices.claims")
)]
async fn send_invoice(job: &Job) -> HandlerOutcome {
    tracing::info!(job.id, "invoice sent");
    HandlerOutcome::ack()
}

Manual

examples/manual/redis_lists.rs

/// Every claim is recorded in `invoices.claims`. An entry still unsettled after `MIN_IDLE` goes
/// back on `invoices`, so set it above the longest time one invoice takes.
struct SendInvoice;

impl Handle<Job, (), (), (), AppState> for SendInvoice {
    fn handle(
        &self,
        job: &Job,
        _outs: &(),
        _ctx: &mut Context<'_, (), AppState>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(job.id, "invoice sent");
        ready(Ok(()))
    }
}

fn invoices() -> RedisList {
    RedisList::new("invoices")
        .reliable()
        .min_idle(MIN_IDLE)
        .recovery_zset("invoices.claims")
}