Replaying History
Edit this pageRegister a fresh consumer group to process retained history with a new or corrected handler. An existing group’s cursor cannot be rewound with the shipped API.
Shipped today: bootstrap a new service from history
Suppose orders.placed already contains months of orders and a new
search-indexer needs to build a projection. Register it with default
Start so it reads retained history, then continues with live traffic:
orders := client.Stream[OrderPlaced]("orders.placed")
search := orders.Consumer("search-indexer")
indexer, err := search.Register(ctx, nil)
if err != nil {
return err
}
return indexer.Consume(ctx, indexOrder, nil)
If that name already has a cursor, it resumes the existing position.
Use a new name for a separate rebuild. A group that needs only live
traffic uses Start: sqlstreams.Head().
For a damaged projection, run the corrected handler under a fresh group
and write into a separate projection. For example, rebuild orders 101–104
with search-indexer-v2, check the resulting records and unresolved
deliveries, then switch readers to the rebuilt projection and retire the
old group. Cursor progress alone does not prove the rebuild succeeded;
inspect failures too.
Other groups keep their cursors, but the rebuild adds database reads, handler load, and possible retention pressure.
Proposed: rewind an existing group
Tuesday 09:14: a deploy ships a bug in warehouse-sync that silently
mis-handles EU orders. Wednesday: you notice. The proposed loop:
-
Deploy the corrected handler.
-
Rewind the group to before the damage — the proposed verb takes a message id or a timestamp and moves the group’s cursor back:
// PROPOSED — not shipped API. err := client.Stream[OrderPlaced](registered.Name).Consumer("warehouse-sync"). Rewind(ctx, sqlstreams.AtTime(tuesdayMorning))The position is the same
sqlstreams.CursorPositiona new group’sStarttakes — one type for every verb that places a cursor;AtTimeandAtMessageIdare the two constructors this verb adds. -
Watch it re-run the window. Handlers are idempotent (they already had to be — at-least-once delivery), so re-processing messages it had handled correctly is harmless.
Spec questions the proposal has to answer before it ships — recorded here so the eventual design is reviewable against them:
- Delivery rows in the window. A rewind re-runs messages that may
have open or
deaddelivery rows; the verb must define whether those reset, keep their attempt history, or block the rewind. - Safety rail. Rewinding a group that fires emails is exciting in the wrong way — the verb should report what it would redeliver (count, id range, time span) before doing it.
- Retention interplay. A rewind target older than retention is an error naming the oldest retained message, not a silent partial replay.
What replay costs
Rebuilding reads retained messages and runs the handler again. Plan for that database and downstream load; idempotency must cover the entire history window, including any external provider’s deduplication limits. Retention determines which messages remain available:
registered, err := client.Stream[OrderPlaced]("orders").Register(ctx,
&sqlstreams.StreamConfig{RetentionTTL: 30 * 24 * time.Hour})
if err != nil {
return err
}
The default retains messages indefinitely; a TTL trades history against
storage. By default,
retention never drops past the slowest group’s cursor
unless the stream opts into AllowDropPastCommitted.