Replay and seeking a stream

This guide moves a consumer group's cursor on a Redis stream: back to replay entries, forward to skip them.

The example is a payments service. It rebuilds a view of balances from the ledger stream, and sends alerts to customers from another stream.

Replay from the beginning

Macros

src/main.rs

/// Rebuilds the balance view from the oldest entry the stream retains, on every start.
#[subscriber(
    RedisStream::new("ledger").group("balances"),
    start_at(RedisGroupPosition::beginning())
)]
async fn rebuild(entry: &LedgerEntry) -> HandlerOutcome {
    tracing::info!(
        entry.seq,
        entry.account,
        entry.amount_cents,
        "balance updated"
    );
    HandlerOutcome::ack()
}

Manual

src/main.rs

/// Rebuilds the balance view from the oldest entry the stream retains, on every start.
struct Rebuild;

impl Handle<LedgerEntry> for Rebuild {
    fn handle(
        &self,
        entry: &LedgerEntry,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(
            entry.seq,
            entry.account,
            entry.amount_cents,
            "balance updated"
        );
        ready(Ok(()))
    }
}

A Redis stream keeps its entries after they are read and acknowledged. A consumer group keeps one cursor over them: the id of the last entry it was given. Replaying means moving that cursor back, which by hand is XGROUP SETID.

start_at moves the cursor before the subscription's first delivery, on every start. RedisGroupPosition::beginning is the oldest entry the stream still retains, so rebuild reads the whole ledger each time the service starts. A view rebuilt this way starts from an empty state, such as one held in memory.

The cursor belongs to the group, not to the subscription. Every consumer of balances replays, in this service and in every other instance of it.

The mount

Macros

src/main.rs

fn service(broker: RedisBroker) -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        b.include(rebuild);
        b.include(notify);
    })
}

Manual

src/main.rs

fn service(broker: RedisBroker) -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        let ledger = RedisStream::new("ledger").group("balances");
        b.include(
            subscriber(ledger, Rebuild)
                .build()
                .start_at(RedisGroupPosition::beginning()),
        );
        let alerts = RedisStream::new("alerts").group("notifier");
        b.include(subscriber(alerts, Notify).build());
    })
}

With the macros, start_at(..) is a clause of #[subscriber]. In the manual form it is a step of the mount, after build().

Position	Where the group resumes
RedisGroupPosition::beginning()	the oldest entry the stream retains
RedisGroupPosition::end()	the tail: only entries added afterwards
RedisGroupPosition::after(id)	the entry after id, which itself is skipped

StreamStart is a different setting: it says where a group starts when it does not exist yet. start_at moves a group that exists, on every start.

Skip to the tail

Macros

src/main.rs

/// Sends alerts. On a halt marker the group moves to the tail, so the alerts queued behind the
/// marker are never sent, by this consumer or any other of `notifier`.
#[subscriber(RedisStream::new("alerts").group("notifier"))]
async fn notify(alert: &Alert, Ctx(seeker): Ctx<keys::SeekHandle>) -> HandlerOutcome {
    if !alert.halt {
        tracing::info!(alert.id, "alert sent");
        return HandlerOutcome::ack();
    }
    tracing::warn!(
        group = seeker.group(),
        "halt marker, the backlog is skipped"
    );
    match seeker.seek(RedisGroupPosition::end()).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(_) => HandlerOutcome::retry(),
    }
}

Manual

src/main.rs

/// Sends alerts. On a halt marker the group moves to the tail, so the alerts queued behind the
/// marker are never sent, by this consumer or any other of `notifier`.
struct Notify;

impl Handle<Alert, (), (), StreamContext> for Notify {
    async fn handle(
        &self,
        alert: &Alert,
        _outs: &(),
        ctx: &mut Context<'_, StreamContext>,
    ) -> Result<(), HandlerOutcome> {
        if !alert.halt {
            tracing::info!(alert.id, "alert sent");
            return Ok(());
        }
        let seeker = ctx.context(keys::SeekHandle);
        tracing::warn!(
            group = seeker.group(),
            "halt marker, the backlog is skipped"
        );
        seeker
            .seek(RedisGroupPosition::end())
            .await
            .map_err(|_| HandlerOutcome::retry())
    }
}

A handler moves its group while the service runs through the seek handle. Ctx<keys::SeekHandle> gives it; its group() names the group it moves.

A producer floods the alerts stream by mistake, and an operator appends a halt marker while the flood goes on. notify reads the marker and moves the group to the tail, so the alerts that follow the marker in the stream are never sent. The seek takes effect as soon as it returns. Entries the old cursor had already selected are discarded, not delivered, and a subscription blocked in XREADGROUP reads under the new cursor within one block interval.

A batch handler reads the same handle with ctx.context(keys::SeekHandle) on its batch context.

Replay one delivery

A delivery knows its own position: Ctx<keys::Position> is the cursor that redelivers it. A seek to it delivers the same entry again, then the entries after it, in stream order. A handler that must keep order on a transient failure can seek there instead of returning retry, because on a stream retry appends a copy at the tail. Ctx<keys::EntryId> is the entry's own id, and RedisGroupPosition::after(id) builds a position from an id stored elsewhere.

What a seek leaves alone

A seek moves the cursor and nothing else.

• An entry delivered and not acknowledged stays in the group's pending entries list, and a reclaim or claiming subscription still finds it.
• A copy waiting in a delay set is appended when it falls due, wherever the group reads.
• A replayed entry is read as a new delivery, and Redis counts its deliveries from one again. A claiming subscription caps its attempts from the replay on, and the framework's own retry count does not move.

Retention

Redis keeps a stream's entries until something trims them, and a replay can only reach what is kept. Trimming with XTRIM is the service's decision: a handler sends it through Ctx<keys::FredPool>, or an operator runs it. A group whose cursor points before the oldest retained entry reads from the oldest one.

Testing

The tests in the example files run in process. They check that the view reads every entry and that a halt marker skips the alerts queued behind it. A replay across a restart needs a server that keeps the stream between two starts, so that test runs live, against REDIS_TEST_URL.