Seek

A handler moves its group's cursor through the SeekHandle key; a delivery's Position key seeks back to that delivery.

Ctx<keys::SeekHandle> binds a RedisGroupSeeker for the group the delivery came from. A batch body reads the same key off its StreamBatchContext with ctx.context(keys::SeekHandle). seek(position) takes a RedisGroupPosition and moves the cursor of the whole group, every consumer of it included. The cursor moves as soon as seek returns. A read blocked in XREADGROUP sees the move within one block, and the entries it had selected under the old cursor are discarded, never delivered. Ctx<keys::Position> binds the position of the delivery itself; a seek there delivers that entry again and the entries after it. The group's cursor is exclusive, so that position holds the id one below the entry's, and the crate computes it. A seek leaves three things alone. Entries delivered and not yet acknowledged stay in the pending entries list, where a reclaim subscription can take them. Copies waiting in a delay queue are appended when they fall due, wherever the group reads. A replayed entry is delivered as a new one: Redis starts its delivery count in the pending entries list over at one. So a cap that reads that count, on a claiming or a reclaim subscription, counts from the replay, and the retry-count header the entry carries stays as it was. Seeking is a stream's: a list and a Pub/Sub channel hand each value out once and keep no read position. A seek after the service shut down returns an error. A position that has to hold on every start belongs in start_at.

Example

A batch that holds the void marker moves the group to the tail once it is billed, so the orders queued behind the marker are never billed.

Macros
Manual

examples/redis_positions_seek.rs
/// Bills a batch of orders. A batch holding the void marker moves the group to the tail once it
/// is billed, so the orders queued behind the marker are never billed.
#[subscriber(RedisStream::new("orders").group("billing"))]
async fn bill(orders: &[Order], ctx: &mut Context<'_, StreamBatchContext>) -> HandlerOutcome {
    tracing::info!(count = orders.len(), "orders billed");
    if orders.iter().all(|order| order.id != 0) {
        return HandlerOutcome::ack();
    }
    let group = ctx.context(keys::ConsumerGroup);
    tracing::info!(%group, "void marker, the group moves to the tail");
    match ctx
        .context(keys::SeekHandle)
        .seek(RedisGroupPosition::end())
        .await
    {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

examples/manual/redis_positions_seek.rs
/// Bills a batch of orders. A batch holding the void marker moves the group to the tail once it
/// is billed, so the orders queued behind the marker are never billed.
struct Bill;

impl Handle<[Order], (), (), StreamBatchContext> for Bill {
    async fn handle(
        &self,
        orders: &[Order],
        _outs: &(),
        ctx: &mut Context<'_, StreamBatchContext>,
    ) -> Result<(), Vec<HandlerOutcome>> {
        tracing::info!(count = orders.len(), "orders billed");
        if orders.iter().all(|order| order.id != 0) {
            return Ok(());
        }
        let group = ctx.context(keys::ConsumerGroup);
        tracing::info!(%group, "void marker, the group moves to the tail");
        ctx.context(keys::SeekHandle)
            .seek(RedisGroupPosition::end())
            .await
            .map_err(|_| orders.iter().map(|_| HandlerOutcome::retry()).collect())
    }
}

fn app() -> impl App {
    let broker = RedisBroker::standalone("redis://localhost:6379");
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(broker, |b| {
        let orders = RedisStream::new("orders").group("billing");
        b.include(subscriber(orders, Bill).batch(nonzero!(2)).build());
    })
}

Example​

Example