Fan-out, Retention & Replay

A stream retains messages independently of consumption. Each consumer group has its own cursor and exception rows, so several groups can process the same message with different handlers and retry policies. They share database capacity and the stream’s retention policy.

Independent groups

On orders.created, email-receipts sends receipts while fraud-screening scores the same orders. Each group registers against the same stream handle:

orders := client.Stream[OrderCreated]("orders.created")
receipts := orders.Consumer("email-receipts")
receiptsConsumer, err := receipts.Register(ctx, nil)
if err != nil {
	return err
}
screening := orders.Consumer("fraud-screening")
screeningConsumer, err := screening.Register(ctx, nil)
if err != nil {
	return err
}

If message 101 succeeds for email-receipts and becomes dead for fraud-screening, neither outcome changes the other group’s delivery. A slow group does not hold another group’s cursor, though it can prevent shared history from expiring. Each group can also declare bindings to read only matching routing keys.

Reading history

A new group reads retained history by default. Add search-indexer after orders have been flowing for months and it can build its projection from the messages still retained, then continue into live traffic.

A group that only needs new traffic declares Start: sqlstreams.Head(). This chooses its initial cursor once; redeclaring an existing group does not move it. Where a New Group Starts shows the declaration and how to check its position.

The registration boundary

Head() uses the visible MAX(id) inside the registration transaction. Suppose orders.created (stream id 1) has a visible head of 1204318. The new search-indexer group (id 7) gets this cursor:

SELECT
    consumer_group_id,
    claimed,
    committed,
    settled_head
FROM sqlstreams.consumer_group_cursor_1
WHERE consumer_group_id = 7;
-- consumer_group_id | claimed | committed | settled_head
-- 7                 | 1204318 | 1204318   | 1204318

All three positions start at the head. settled_head tells fresh claims to treat ids at or below that position as settled. If a produce already allocated id 1204317 but commits after registration, this group skips it too: its cursor is already past that id. A new message with id 1204319 is ahead of the cursor and can be read.

For a group that must receive traffic around its startup, register it before producers start. Head() is a visible id boundary, not a commit timestamp filter.

Progress and failures

Read cursor progress alongside pending and dead deliveries. In the example above, fraud-screening can reach the head while message 101 remains dead. Zero cursor lag does not mean every handler succeeded.

For stream id 1, inspect the cursors:

SELECT
    g.name AS consumer_group,
    c.claimed,
    c.committed,
    (SELECT COALESCE(max(id), 0) FROM sqlstreams.message_log_1) - c.committed AS cursor_distance
FROM sqlstreams.consumer_group_cursor_1 c
JOIN sqlstreams.consumer_group_config g ON g.id = c.consumer_group_id;

The distance is measured in ids, not a count of messages: ids can have gaps and groups can skip rows through version or routing filters. Inspect exception state separately:

SELECT
    g.name AS consumer_group,
    d.status,
    count(*) AS messages,
    min(d.created_at) AS oldest_created_at
FROM sqlstreams.exception_queue_1 d
JOIN sqlstreams.consumer_group_config g ON g.id = d.consumer_group_id
GROUP BY g.name, d.status
ORDER BY g.name, d.status;

Signal	What it tells you
Cursor distance	progress through the message log
ready, inflight, deferred rows	deliveries still unresolved, including requested delays and key waits
Oldest unresolved row	how long pending delivery state has existed
dead rows	deliveries requiring investigation even if the cursor is caught up

The metrics API exposes cursor, exception, and lease state. Use Consumer Tuning for growing backlog and Dead-lettered Messages for dead rows.

Retention

StreamConfig.RetentionTTL defaults to 0, which retains messages indefinitely. A nonzero TTL lets the janitor drop whole id-range partitions whose newest message has expired and sweep the remaining expired rows. Consuming a message does not delete it.

By default, AllowDropPastCommitted: false prevents retention from passing the slowest group’s committed cursor. A lagging search-indexer can therefore keep old partitions on disk even after email-receipts has caught up. Observe storage as well as delivery progress.

Set AllowDropPastCommitted: true only when dropping unread history is acceptable. Exception rows do not provide separate retention: when retention deletes the source message, its retry, delayed, or dead state is deleted too. See Stream for the fields.