Counts

The claiming modes report Redis's delivery count and idle time as headers, and a cap reads that count.

A consumer group counts how often it has handed out each pending entry. The two stream modes that claim entries, claiming and reclaim, put that count on every delivery in DELIVERY_COUNT_HEADER. IDLE_MS_HEADER carries how long the entry had been pending, in milliseconds. A claiming delivery reports the attempts made before it, so a fresh entry carries zero for both headers. A reclaim delivery reports the count including itself. The runtime reads the same count through redelivery_count, which counts the current delivery in both modes. That is the count a max_attempts cap reads. An entry a crashed worker took and never settled counts towards the cap, although no process saw it fail. A reclaiming read that found entries issues one XPENDING to learn their counts. Every other subscription has no count from Redis, and its cap counts through the runtime's retry-count header instead.

Example

The handler logs both values with every attempt.

Macros
Manual

examples/redis_acknowledgement_counts.rs
/// A claimed entry carries how often Redis has handed it out before and how long it sat pending;
/// the handler logs both with every attempt.
#[subscriber(RedisStream::claiming("payouts", MIN_IDLE).group("payouts"))]
async fn pay(
    payout: &Payout,
    ctx: &mut Context<'_>,
    State(gateway): State<PaymentGateway>,
) -> HandlerOutcome {
    let earlier = ctx.headers().get_str(DELIVERY_COUNT_HEADER).unwrap_or("0");
    let idle_ms = ctx.headers().get_str(IDLE_MS_HEADER).unwrap_or("0");
    tracing::info!(payout.id, earlier, idle_ms, "paying out");
    match gateway.pay(payout).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(payout.id, %err, "payout failed, left pending");
            HandlerOutcome::retry()
        }
    }
}

examples/manual/redis_acknowledgement_counts.rs
/// A claimed entry carries how often Redis has handed it out before and how long it sat pending;
/// the handler logs both with every attempt.
struct Pay;

impl Handle<Payout, (), (), (), AppState> for Pay {
    async fn handle(
        &self,
        payout: &Payout,
        _outs: &(),
        ctx: &mut Context<'_, (), AppState>,
    ) -> Result<(), HandlerOutcome> {
        let earlier = ctx.headers().get_str(DELIVERY_COUNT_HEADER).unwrap_or("0");
        let idle_ms = ctx.headers().get_str(IDLE_MS_HEADER).unwrap_or("0");
        tracing::info!(payout.id, earlier, idle_ms, "paying out");
        if let Err(err) = ctx.state().gateway.pay(payout).await {
            tracing::warn!(payout.id, %err, "payout failed, left pending");
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

fn payouts() -> RedisStream {
    RedisStream::claiming("payouts", MIN_IDLE).group("payouts")
}

The cap at the mount site reads the same count: the third failed delivery goes to payouts.dlq.

Macros
Manual

examples/redis_acknowledgement_counts.rs
#[ruststream::app]
fn app() -> impl App<State = AppState> {
    RustStream::new(AppInfo::new("payouts", "0.1.0"))
        .on_startup(async move |()| {
            let gateway = PaymentGateway {
                endpoint: "https://gateway.internal".into(),
            };
            Ok::<_, Infallible>(AppState { gateway })
        })
        .with_broker(RedisBroker::standalone("redis://localhost:6379"), |b| {
            // The cap reads Redis's own count, so an entry a crashed worker never settled counts
            // too. The third failed delivery goes to `payouts.dlq`.
            b.include(pay)
                .max_attempts(nonzero!(3u32))
                .dead_letter("payouts.dlq");
        })
}

examples/manual/redis_acknowledgement_counts.rs
fn app() -> impl App<State = AppState> {
    RustStream::new(AppInfo::new("payouts", "0.1.0"))
        .on_startup(async move |()| {
            let gateway = PaymentGateway {
                endpoint: "https://gateway.internal".into(),
            };
            Ok::<_, Infallible>(AppState { gateway })
        })
        .with_broker(RedisBroker::standalone("redis://localhost:6379"), |b| {
            // The cap reads Redis's own count, so an entry a crashed worker never settled counts
            // too. The third failed delivery goes to `payouts.dlq`.
            b.include(subscriber(payouts(), Pay).build())
                .max_attempts(nonzero!(3u32))
                .dead_letter("payouts.dlq");
        })
}

Example​

Example