Acknowledgement

What ack and nack do on each form: XACK on a stream, LREM on a reliable list, nothing on a simple list or a channel.

A handler ends a delivery with an outcome: ack, retry or drop. The runtime settles the delivery with that outcome, and each form turns it into its own Redis commands.

Form	ack	retry	drop
Stream	`XACK`	`XADD` of a copy, then `XACK`	`XACK`
Stream, `claiming`	`XACK`	the entry stays pending and comes back after `min_idle`	`XACK`
List, reliable	`LREM` off the processing list	`LPUSH` back onto the queue, then `LREM`	`LREM`
List, simple	nothing	nothing	nothing
Channel or pattern	nothing	nothing	nothing

On a stream a retry appends a copy and then acknowledges the original, so a crash between the two leaves a duplicate rather than a loss. The copy is a new entry with a new id. A claiming subscription retries through the pending entries list instead: nothing is appended, and the entry keeps its id and its delivery count. A simple list and Pub/Sub keep nothing to settle: the entry left Redis when it was popped, and a message is gone once it is pushed. An ack, a drop or a retry there reports AckError::Unsupported, and the runtime logs an ack / nack failed warning for each delivery. A retry there delivers nothing again, unless it is a retry_after or a cap is declared: both travel as a copy the runtime publishes.

Example

A payment is charged, dropped when it is malformed, or retried when the gateway is busy.

Macros
Manual

examples/redis_acknowledgement.rs
/// One handler, three outcomes: the ack settles the entry, the drop settles it without a copy,
/// the retry appends a copy to `payments` and settles the original.
#[subscriber(RedisStream::new("payments").group("billing"))]
async fn charge(payment: &Payment, State(gateway): State<PaymentGateway>) -> HandlerOutcome {
    if payment.amount_cents <= 0 {
        tracing::warn!(payment.id, "malformed payment dropped");
        return HandlerOutcome::drop();
    }
    match gateway.charge(payment).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(payment.id, %err, "charge failed, retrying");
            HandlerOutcome::retry()
        }
    }
}

examples/manual/redis_acknowledgement.rs
/// One handler, three outcomes: the ack settles the entry, the drop settles it without a copy,
/// the retry appends a copy to `payments` and settles the original.
struct Charge;

impl Handle<Payment, (), (), (), AppState> for Charge {
    async fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        ctx: &mut Context<'_, (), AppState>,
    ) -> Result<(), HandlerOutcome> {
        if payment.amount_cents <= 0 {
            tracing::warn!(payment.id, "malformed payment dropped");
            return Err(HandlerOutcome::drop());
        }
        if let Err(err) = ctx.state().gateway.charge(payment).await {
            tracing::warn!(payment.id, %err, "charge failed, retrying");
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

fn payments() -> RedisStream {
    RedisStream::new("payments").group("billing")
}

Example​

Example