Reliable Dispatch — Transactional Outbox, LISTEN/NOTIFY and Sharded Redis Streams
On this page · 10 sections
- Step 1: Name queues and shards
- Step 2: Write the dispatch into the outbox
- Step 3: The outbox table and its trigger
- Step 4: The enqueuer's main loop
- Step 5: Listen on a dedicated connection
- Step 6: Claim and publish a batch
- Step 7: Retry, dead-letter and clean up
- Step 8: The stream interfaces
- Step 9: Implement the consumer on Redis Streams
- Step 10: Publish the shard count
The coordinator saved "dispatch task t8" as an outbox row. Now get that message to a worker. You build three things in this page:
EnqueueDispatch— the outboxINSERTinside the decision's transaction.- The outbox enqueuer — a loop that publishes committed rows to Redis, woken instantly by
NOTIFY. - The stream layer — queue and shard naming, plus a Redis Streams consumer for workers.
Step 1: Name queues and shards
A queue like http can carry a huge volume. One Redis stream is one key, and one key lives on one Redis Cluster node. So split each queue into shards — several streams — and spread tasks across them.
Create pkg/stream/naming.go:
// pkg/stream/naming.go
package stream
const (
// TaskStreamPrefix prefixes every task stream: workflow.tasks.<queue>.<shard>.
TaskStreamPrefix = "workflow.tasks."
// DefaultShards is the number of streams per queue when not configured.
DefaultShards = 4
// ShardsKey is the Redis key where the coordinator publishes its shard
// count, so workers read exactly the streams the coordinator writes.
ShardsKey = "composer:stream_shards"
)
// TaskStream returns the stream name for one shard of a task queue. Each shard
// is its own key, so in Redis Cluster the shards of a queue spread across nodes.
func TaskStream(queue string, shard int) string {
return fmt.Sprintf("%s%s.%d", TaskStreamPrefix, queue, shard)
}
// ShardFor picks the shard for a task. Any stable spread works: shards only
// distribute load, they carry no ordering guarantee.
func ShardFor(taskID string, shards int) int {
if shards <= 1 {
return 0
}
h := fnv.New32a()
_, _ = h.Write([]byte(taskID))
return int(h.Sum32() % uint32(shards))
}
// QueueStreams returns every shard stream of the given queues.
func QueueStreams(queues []string, shards int) []string {
if shards < 1 {
shards = 1
}
out := make([]string, 0, len(queues)*shards)
for _, q := range queues {
for s := range shards {
out = append(out, TaskStream(q, s))
}
}
return out
}
Workflow steps have no ordering relationship between different tasks (the state machine already orders them), so a simple hash spread is enough.
Step 2: Write the dispatch into the outbox
Create pkg/coord/repository_dispatch.go. The message a worker receives:
// pkg/coord/repository_dispatch.go
// taskDispatch is the message a worker receives for one task attempt.
type taskDispatch struct {
TaskID string
RunID string
WorkflowID string
WorkflowVersion int
NodeID string
NodeType string
Queue string
Attempt int
Input json.RawMessage
}
// EnqueueDispatch writes the task's dispatch message to the outbox, addressed
// to one shard of the task's queue. The outbox enqueuer publishes it no
// earlier than notBefore.
func (r repository) EnqueueDispatch(ctx context.Context, q db.Querier, d *taskDispatch, notBefore time.Time) error {
payload, err := json.Marshal(map[string]any{
"task_id": d.TaskID,
"run_id": d.RunID,
"workflow_id": d.WorkflowID,
"workflow_version": d.WorkflowVersion,
"node_id": d.NodeID,
"node_type": d.NodeType,
"attempt": d.Attempt,
"input": string(d.Input),
})
if err != nil {
return fmt.Errorf("marshal dispatch payload: %w", err)
}
streamName := stream.TaskStream(d.Queue, stream.ShardFor(d.TaskID, r.streamShards))
if _, err := q.Exec(ctx, `
INSERT INTO outbox (id, task_id, stream, payload, type, status, next_attempt_at)
VALUES ($1, $2, $3, $4::jsonb, 'TASK_EXECUTION', 'PENDING', $5)`,
uuid.NewString(), d.TaskID, streamName, payload, notBefore); err != nil {
return fmt.Errorf("insert outbox entry for task %s: %w", d.TaskID, err)
}
return nil
}
The message is self-contained: it carries the resolved input, the attempt and the run ID. A worker never needs to read the database to execute a task.
Step 3: The outbox table and its trigger
-- Outbox table for transactional message publishing
CREATE TABLE IF NOT EXISTS outbox(
id TEXT PRIMARY KEY,
task_id TEXT, -- task this row dispatches, so recovery can find tasks without a pending dispatch
stream TEXT NOT NULL,
payload JSONB,
type TEXT,
status TEXT NOT NULL DEFAULT 'PENDING',
attempt_count INT NOT NULL DEFAULT 0,
max_retries INT NOT NULL DEFAULT 5,
next_attempt_at TIMESTAMPTZ,
last_error TEXT,
backoff_seconds INT NOT NULL DEFAULT 5,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
sent_at TIMESTAMPTZ
);
CREATE INDEX IF NOT EXISTS idx_outbox_status ON outbox(status, created_at);
-- Partial index matching the outbox poll query's filter and ORDER BY created_at.
CREATE INDEX IF NOT EXISTS idx_outbox_pending_next
ON outbox (created_at)
WHERE status IN ('PENDING', 'RETRY');
CREATE INDEX IF NOT EXISTS idx_outbox_task_id ON outbox (task_id) WHERE task_id IS NOT NULL;
An outbox row moves through these states:
If the enqueuer only polled every second, every step would wait up to a second for dispatch. A ten-step workflow would spend ten seconds idle. Instead, add a trigger that fires NOTIFY when outbox rows are inserted:
-- Wake outbox enqueuers (LISTEN outbox_new) as soon as a transaction that
-- inserted outbox rows commits. Statement-level, and Postgres folds duplicate
-- notifications within a transaction, so a batch insert sends one.
CREATE OR REPLACE FUNCTION composer_notify_outbox() RETURNS TRIGGER
LANGUAGE plpgsql AS $$
BEGIN
PERFORM pg_notify('outbox_new', '');
RETURN NULL;
END;
$$;
DROP TRIGGER IF EXISTS outbox_notify ON outbox;
CREATE TRIGGER outbox_notify AFTER INSERT ON outbox
FOR EACH STATEMENT EXECUTE FUNCTION composer_notify_outbox();
PostgreSQL delivers notifications only when the transaction commits. That is exactly the timing you want: the enqueuer wakes the moment the outbox row becomes visible, never before.
Step 4: The enqueuer's main loop
Create pkg/outbox/enqueuer.go:
// pkg/outbox/enqueuer.go
package outbox
// notifyChannel is the Postgres channel the outbox insert trigger notifies.
const notifyChannel = "outbox_new"
// OutboxEnqueuer processes outbox rows and pushes them to Redis streams.
//
// It is woken by Postgres NOTIFY as soon as a transaction that inserted outbox
// rows commits, so dispatch latency does not depend on the poll interval. The
// poll still runs as a fallback (lost notifications, listener reconnects) and
// to pick up rows whose next_attempt_at has come due.
type OutboxEnqueuer struct {
cfg EnqueuerConfig
wg sync.WaitGroup
quit chan struct{}
wake chan struct{} // signalled by the LISTEN loop
}
Run processes batches with bounded concurrency, and between batches waits for whichever comes first: a full batch (more work waiting), a NOTIFY, or the poll interval:
// Run starts the enqueuer loop and returns when ctx is done (or Stop called).
func (o *OutboxEnqueuer) Run(ctx context.Context) {
o.cfg.Logger.Info().Int("batch", o.cfg.BatchSize).Msg("starting outbox enqueuer")
// Start worker goroutines
workerChan := make(chan struct{}, o.cfg.Concurrency)
for i := 0; i < o.cfg.Concurrency; i++ {
workerChan <- struct{}{}
}
// batchFull receives a signal when a batch was at full capacity, meaning
// more rows are likely waiting. The main loop skips its sleep in that case
// so we drain the queue without introducing artificial delay.
batchFull := make(chan struct{}, 1)
// Start periodic cleanup and recovery in background
o.wg.Add(2)
go func() {
defer o.wg.Done()
o.maintenanceLoop(ctx)
}()
go func() {
defer o.wg.Done()
o.listenLoop(ctx)
}()
for {
select {
case <-ctx.Done():
o.cfg.Logger.Info().Msg("context canceled, stopping enqueuer")
o.shutdown()
return
case <-o.quit:
o.cfg.Logger.Info().Msg("quit signal received, stopping enqueuer")
o.shutdown()
return
default:
}
// Acquire a worker slot
select {
case <-workerChan:
// process one batch in goroutine
o.wg.Add(1)
go func() {
defer func() { workerChan <- struct{}{} }()
defer o.wg.Done()
count, err := o.processBatch(ctx)
if err != nil {
o.cfg.Logger.Error().Err(err).Msg("processBatch failed")
}
if count >= o.cfg.BatchSize {
select {
case batchFull <- struct{}{}:
default:
}
}
}()
case <-ctx.Done():
o.shutdown()
return
}
// Throttle: only sleep when the queue appears empty. If the last batch
// was at full capacity there are likely more rows ready, so poll again
// immediately instead of waiting for the full PollInterval.
select {
case <-batchFull:
// more work pending — re-poll immediately
case <-o.wake:
// new rows committed
case <-time.After(o.cfg.PollInterval):
case <-ctx.Done():
o.shutdown()
return
case <-o.quit:
o.shutdown()
return
}
}
}
In practice the loop behaves like this:
- Idle, then a commit: the
NOTIFYwakes the loop at once, and it publishes the new rows. - Idle, nothing new: after the 1 s poll it claims nothing (retries not due yet) and waits again.
- A burst: a full batch of 50 means more rows are waiting, so the loop goes straight to the next batch without sleeping until the queue is drained.
Step 5: Listen on a dedicated connection
LISTEN is session state. If you LISTEN on a pooled connection and return it to the pool, some unrelated query later inherits your subscription. So take a connection out of the pool for good:
// listenLoop holds a dedicated connection on LISTEN outbox_new and signals
// wake for each notification, reconnecting with backoff on failure.
func (o *OutboxEnqueuer) listenLoop(ctx context.Context) {
// WaitForNotification only returns on ctx, so Stop must cancel it too.
ctx, cancel := context.WithCancel(ctx)
defer cancel()
go func() {
select {
case <-o.quit:
cancel()
case <-ctx.Done():
}
}()
backoff := time.Second
for ctx.Err() == nil {
err := o.listen(ctx)
if ctx.Err() != nil {
return
}
o.cfg.Logger.Warn().Err(err).Dur("retry_in", backoff).Msg("outbox listener disconnected; polling until it reconnects")
select {
case <-time.After(backoff):
case <-ctx.Done():
return
}
backoff = min(backoff*2, 30*time.Second)
}
}
func (o *OutboxEnqueuer) listen(ctx context.Context) error {
pooled, err := o.cfg.DB.Acquire(ctx)
if err != nil {
return err
}
// Take the connection out of the pool: its LISTEN session state must not
// leak to other users, and it is closed when listening stops.
conn := pooled.Hijack()
defer conn.Close(context.Background()) //nolint:errcheck
if _, err := conn.Exec(ctx, "LISTEN "+notifyChannel); err != nil {
return err
}
o.signalWake() // catch rows committed while we were not listening
for {
if _, err := conn.WaitForNotification(ctx); err != nil {
return err
}
o.signalWake()
}
}
func (o *OutboxEnqueuer) signalWake() {
select {
case o.wake <- struct{}{}:
default:
}
}
Two small details matter:
wakehas a buffer of one andsignalWakenever blocks. A thousand notifications in a burst collapse into one wake-up, and the batch loop drains everything anyway.- After every (re)connect it signals a wake immediately, to catch rows committed while it was not listening. Combined with the poll fallback, a lost notification can delay dispatch but never lose it.
Step 6: Claim and publish a batch
Several coordinators run an enqueuer each. They must not publish the same row. FOR UPDATE SKIP LOCKED lets each one grab a different set of rows without waiting on the others:
// processBatch claims a batch of due outbox rows (marking them IN_FLIGHT so
// other coordinators skip them), publishes them and records each outcome.
// Returns the number of rows claimed and any error.
func (o *OutboxEnqueuer) processBatch(ctx context.Context) (int, error) {
rows, err := o.cfg.DB.Query(ctx, `
UPDATE outbox SET status = 'IN_FLIGHT', updated_at = NOW()
WHERE id IN (
SELECT id FROM outbox
WHERE status IN ('PENDING', 'RETRY')
AND (next_attempt_at IS NULL OR next_attempt_at <= NOW())
ORDER BY created_at
LIMIT $1
FOR UPDATE SKIP LOCKED)
RETURNING id, stream, payload, attempt_count, max_retries, next_attempt_at, backoff_seconds`,
o.cfg.BatchSize)
if err != nil {
return 0, err
}
rowsToProcess, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (outboxRow, error) {
var r outboxRow
err := row.Scan(&r.ID, &r.Stream, &r.Payload, &r.Attempt, &r.MaxRetries, &r.NextAttempt, &r.BackoffSeconds)
return r, err
})
if err != nil {
return 0, err
}
// Publish each row; failures are rescheduled individually, successes are
// deleted together in one statement.
sent := make([]string, 0, len(rowsToProcess))
for _, r := range rowsToProcess {
util.RecordOutboxInFlight(ctx, r.Stream, 1)
startTime := time.Now()
if err := o.publishRow(ctx, r); err != nil {
o.cfg.Logger.Warn().Err(err).Str("outbox_id", r.ID).Str("stream", r.Stream).Msg("publish failed; scheduling retry/dlq")
util.RecordOutboxFailed(ctx, r.Stream, err.Error())
if err := o.scheduleRetry(ctx, r, err); err != nil {
o.cfg.Logger.Error().Err(err).Str("outbox_id", r.ID).Msg("failed to schedule outbox retry")
}
} else {
sent = append(sent, r.ID)
util.RecordOutboxProcessed(ctx, r.Stream)
o.cfg.Logger.Debug().Str("outbox_id", r.ID).Dur("duration", time.Since(startTime)).Msg("outbox row published")
}
util.RecordOutboxInFlight(ctx, r.Stream, -1)
}
if len(sent) > 0 {
if _, err := o.cfg.DB.Exec(ctx, `DELETE FROM outbox WHERE id = ANY($1)`, sent); err != nil {
// Fallback: mark SENT so the maintenance loop cleans them up and
// the stale IN_FLIGHT recovery does not publish them again.
o.cfg.Logger.Error().Err(err).Int("count", len(sent)).Msg("failed to delete sent outbox rows, marking SENT")
_, _ = o.cfg.DB.Exec(ctx,
`UPDATE outbox SET status='SENT', sent_at=NOW(), updated_at=NOW() WHERE id = ANY($1)`, sent)
}
}
return len(rowsToProcess), nil
}
The next_attempt_at <= NOW() filter is how retry backoff works for free: a retry's dispatch row simply is not claimable yet.
Publishing itself is one call through the stream.Publisher interface:
// publishRow pushes a single outbox row to its stream.
func (o *OutboxEnqueuer) publishRow(ctx context.Context, r outboxRow) error {
var payload map[string]interface{}
if len(r.Payload) == 0 {
payload = map[string]interface{}{"id": r.ID}
} else if err := json.Unmarshal(r.Payload, &payload); err != nil {
payload = map[string]interface{}{"data": string(r.Payload)}
}
stream := r.Stream
if stream == "" {
stream = "workflow.tasks"
}
return o.cfg.Publisher.Publish(ctx, stream, payload)
}
Step 7: Retry, dead-letter and clean up
A failed publish (Redis down) is rescheduled with exponential backoff and jitter. After max_retries it goes to the dead-letter table:
// scheduleRetry reschedules a row whose publish failed with exponential
// backoff, or moves it to the DLQ once its retries are exhausted.
func (o *OutboxEnqueuer) scheduleRetry(ctx context.Context, r outboxRow, publishErr error) error {
attempt := r.Attempt + 1
maxRetries := r.MaxRetries
if maxRetries <= 0 {
maxRetries = 5
}
if attempt > maxRetries {
if _, err := o.cfg.DB.Exec(ctx, `INSERT INTO outbox_dlq (id, stream, payload, last_error, attempts, created_at, moved_at)
VALUES ($1,$2,$3::jsonb,$4,$5,NOW(),NOW()) ON CONFLICT (id) DO NOTHING`,
r.ID, r.Stream, r.Payload, publishErr.Error(), attempt); err != nil {
o.cfg.Logger.Error().Err(err).Str("outbox_id", r.ID).Msg("failed move to dlq")
}
if _, err := o.cfg.DB.Exec(ctx, `UPDATE outbox SET status='DLQ', last_error=$1, attempt_count=$2, updated_at=NOW() WHERE id=$3`,
publishErr.Error(), attempt, r.ID); err != nil {
return err
}
util.RecordOutboxDLQ(ctx, r.Stream)
return nil
}
nextAt := time.Now().Add(computeBackoffDuration(r.BackoffSeconds, attempt))
_, err := o.cfg.DB.Exec(ctx, `UPDATE outbox SET status='RETRY', attempt_count=$1, next_attempt_at=$2, last_error=$3, updated_at=NOW() WHERE id=$4`,
attempt, nextAt, publishErr.Error(), r.ID)
return err
}
func computeBackoffDuration(baseSeconds, attempt int) time.Duration {
if baseSeconds <= 0 {
baseSeconds = 5
}
// exponential factor
exp := float64(baseSeconds) * math.Pow(2, float64(attempt-1))
// full jitter with guaranteed minimum of baseSeconds
jitter := float64(baseSeconds) + rand.Float64()*exp
return time.Duration(int64(jitter)) * time.Second
}
Jitter matters when Redis comes back after an outage: without it every enqueuer would retry at the same instant and hammer it again.
A maintenance loop handles the crash cases. If an enqueuer dies after marking rows IN_FLIGHT, those rows would be stuck forever — so after two minutes they go back to RETRY:
// maintenanceLoop periodically cleans up SENT rows and recovers stale IN_FLIGHT rows.
func (o *OutboxEnqueuer) maintenanceLoop(ctx context.Context) {
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-o.quit:
return
case <-ticker.C:
o.cleanupSentRows(ctx)
o.recoverStaleInFlight(ctx)
}
}
}
// recoverStaleInFlight resets IN_FLIGHT rows that have been stuck for too long
// (e.g., the enqueuer crashed after marking IN_FLIGHT but before completing processing).
func (o *OutboxEnqueuer) recoverStaleInFlight(ctx context.Context) {
result, err := o.cfg.DB.Exec(ctx,
`UPDATE outbox SET status = 'RETRY', updated_at = NOW(), next_attempt_at = NOW()
WHERE status = 'IN_FLIGHT' AND updated_at < NOW() - INTERVAL '2 minutes'`)
if err != nil {
o.cfg.Logger.Error().Err(err).Msg("failed to recover stale IN_FLIGHT outbox rows")
return
}
if result.RowsAffected() > 0 {
o.cfg.Logger.Warn().Int64("recovered", result.RowsAffected()).Msg("recovered stale IN_FLIGHT outbox rows")
}
}
That recovery can publish a row a second time (the crash happened after XADD). That is fine: the duplicate message reaches a worker, the worker fails to acquire the lease, and drops it.
Step 8: The stream interfaces
The coordinator and worker must not depend on Redis directly. Define the contract in pkg/stream/stream.go:
// pkg/stream/stream.go
// Message is a transport-independent representation of a single stream entry.
type Message struct {
// ID is the stream-assigned message identifier (e.g. Redis stream ID or
// Kafka offset string). It must be passed back to Ack unchanged.
ID string
// Values contains the key/value pairs published with the message.
Values map[string]any
// DeliveryCount is how many times this message has been delivered.
// 1 = first delivery; >1 = redelivery after a previous consumer failed.
// Populated only for messages returned by ReclaimOrphaned.
DeliveryCount int64
}
// Publisher writes messages to a named stream.
// Implementations must be safe for concurrent use.
type Publisher interface {
// Publish appends values to the named stream. The call is synchronous;
// a nil error means the message is durably stored in the stream.
Publish(ctx context.Context, stream string, values map[string]any) error
}
// Consumer reads messages from a stream with consumer-group semantics,
// providing at-least-once delivery, orphan reclaim, and delivery-count
// tracking needed for DLQ enforcement.
// Implementations must be safe for concurrent use.
type Consumer interface {
Setup(ctx context.Context) error
ReadNew(ctx context.Context, max int64) ([]Message, error)
ReclaimOrphaned(ctx context.Context, minIdle time.Duration) ([]Message, error)
Ack(ctx context.Context, msgID string) error
Close() error
}
Step 9: Implement the consumer on Redis Streams
Redis Streams with consumer groups give you exactly the semantics a task queue needs:
XREADGROUP … >hands each new message to exactly one consumer in the group.- A delivered message stays in the pending entries list (PEL) until it is acked.
XAUTOCLAIMtransfers messages idle in someone else's PEL — the crash-recovery path.
Create pkg/stream/redis/consumer.go:
// pkg/stream/redis/consumer.go
// Setup creates the consumer group (and the stream if absent). Safe to call
// on every startup — ignores BUSYGROUP errors.
func (c *Consumer) Setup(ctx context.Context) error {
err := c.rdb.XGroupCreateMkStream(ctx, c.cfg.StreamName, c.cfg.ConsumerGroup, "0")
if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
return fmt.Errorf("xgroup create: %w", err)
}
return nil
}
// ReadNew blocks up to BlockDuration and returns at most max newly dispatched
// messages (those not yet assigned to any consumer). Returns nil, nil on timeout.
func (c *Consumer) ReadNew(ctx context.Context, max int64) ([]stream.Message, error) {
if max < 1 {
max = 1
}
args := &redis.XReadGroupArgs{
Group: c.cfg.ConsumerGroup,
Consumer: c.cfg.ConsumerName,
Streams: []string{c.cfg.StreamName, ">"},
Count: max,
Block: c.cfg.BlockDuration,
}
res, err := c.rdb.XReadGroup(ctx, args)
if err != nil {
if err == redis.Nil {
return nil, nil
}
return nil, fmt.Errorf("xreadgroup: %w", err)
}
var msgs []stream.Message
for _, s := range res {
for _, m := range s.Messages {
msgs = append(msgs, stream.Message{ID: m.ID, Values: m.Values})
}
}
return msgs, nil
}
Reclaiming orphans also fetches delivery counts, so the worker can dead-letter a message that keeps crashing workers ("poison message"):
// ReclaimOrphaned uses XAUTOCLAIM to take ownership of messages idle longer
// than minIdle, then fetches their delivery counts via XPENDING so callers
// can enforce DLQ limits without an extra round-trip per message.
func (c *Consumer) ReclaimOrphaned(ctx context.Context, minIdle time.Duration) ([]stream.Message, error) {
raw, _, err := c.rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{
Stream: c.cfg.StreamName,
Group: c.cfg.ConsumerGroup,
Consumer: c.cfg.ConsumerName,
MinIdle: minIdle,
Start: "0-0",
Count: c.cfg.ReclaimCount,
})
if err != nil {
return nil, fmt.Errorf("xautoclaim: %w", err)
}
if len(raw) == 0 {
return nil, nil
}
// Fetch delivery counts for all claimed messages in one XPENDING range.
// XAUTOCLAIM already incremented each counter before returning.
counts := c.fetchDeliveryCounts(ctx, raw)
msgs := make([]stream.Message, len(raw))
for i, m := range raw {
msgs[i] = stream.Message{
ID: m.ID,
Values: m.Values,
DeliveryCount: counts[m.ID],
}
}
return msgs, nil
}
// Ack removes the message from the consumer group's PEL (XACK) and deletes
// it from the stream (XDEL) to reclaim memory.
func (c *Consumer) Ack(ctx context.Context, msgID string) error {
_, _ = c.rdb.XAck(ctx, c.cfg.StreamName, c.cfg.ConsumerGroup, msgID)
_, _ = c.rdb.XDel(ctx, c.cfg.StreamName, msgID)
return nil
}
Ack also deletes the entry (XDEL). The task row in PostgreSQL is the record of truth, so there is no reason to keep processed messages in Redis memory.
Step 10: Publish the shard count
Workers must read exactly the streams the coordinator writes to. The coordinator stores its shard count in Redis at startup (you will see this in Coordinator.Run on the recovery page), and the worker reads it:
// cmd/worker/main.go
// streamShards returns the shard count the coordinator published in Redis,
// falling back to STREAM_SHARDS or the default when it is not set yet.
func streamShards(ctx context.Context, rs *redisq.RedisStream, logger zerolog.Logger) int {
if n, err := rs.Client().Get(ctx, stream.ShardsKey).Int(); err == nil && n > 0 {
return n
}
n := int(envInt("STREAM_SHARDS", stream.DefaultShards))
logger.Warn().Int("shards", n).Msg("coordinator has not published a shard count yet; using STREAM_SHARDS")
return n
}
Messages are now flowing into sharded streams within milliseconds of each commit. Next, build the workers that consume them.