Consumer

stream.Consumer(name) names a consumer group on its stream: no I/O, no failure. Register writes the group’s declaration and returns the ConsumerInstance[T] whose Consume runs a session. What the group means is declared at Register and stored on the group’s rows; how one session runs is passed to Consume (consumer group config explains the split).

payments, err := client.Stream[PaymentRequestedV1]("payments.requested").Consumer("charge-cards").Register(ctx,
	&sqlstreams.ConsumerConfig{
		Message: &sqlstreams.MessageOptions{
			Timeout: 10 * time.Second,
			Retry:   &sqlstreams.RetryPolicy{MaxRetries: 3, BaseDelay: 2 * time.Second},
		},
	})
if err != nil {
	return err
}

return payments.Consume(ctx, func(ctx context.Context, payment *PaymentRequestedV1) error {
	fmt.Printf("charging %s\n", payment.OrderId)
	return nil
}, &sqlstreams.ConsumeOptions{BatchLimit: 20})

Verbs

On the handle:

verb	returns	notes
Register(ctx, cfg)	*ConsumerInstance[T]	resolves the stream, writes the declaration, creates the cursor row on first registration; nil cfg is the defaults; newest declaration wins, and a differing one logs SQL0059
Get(ctx)	*Consumer	the comma-ok read: (nil, nil) when the stream or the group is not registered
Workers(ctx)	[]*Worker	the group’s own worker rows, each with its stored config document in Worker.Metadata
Destroy(ctx, options)	error	deletes the group’s cursor, bindings, leases, delivery rows, workers, and schedules; the stream and its messages stay; ErrDestroyDisabled unless ClientConfig.AllowDestroy; ErrConsumerGroupLive while an instance is live and ErrConsumerGroupDeliveriesPending while delivery rows remain, both skipped by DestroyOptions.Force
Binding()	*BindingHandle	no I/O; its Get(ctx) reads the group’s effective binding set, (nil, nil) when the group never declared one and reads the whole stream
Metrics()	*ConsumerMetricsHandle	no I/O; Metrics
Alerts()	*ConsumerAlertsHandle	no I/O; Alerts

On the instance:

verb	returns	notes
Consume(ctx, consumerFunc, options)	error	blocks for the session; cancel ctx to start graceful shutdown and get nil back; nil options is the defaults; a second call on the same instance returns ErrAlreadyConsuming

consumerFunc is func(ctx context.Context, message *T) error, and its return value is the delivery’s outcome: nil succeeds, any error retries, sqlstreams.Terminal(err) dead-letters now, sqlstreams.Delay(d) runs later without counting a failure (handler outcomes). sqlstreams.MetaFromContext(ctx) inside the handler returns the delivery’s MessageMeta: message id, routing and message keys, attempt count, and the resolved options it runs under.

Consume runs the system manager beside the session unless ClientConfig.DisableManager is set, so one live consumer keeps the whole deployment’s upkeep running (Manager). A context that can never be cancelled returns ErrLifecycleContextNotCancellable unless DisableGracefulShutdown is set, and that case runs no manager.

CLI

command	client operation
consumer list orders.created	Stream(name).Consumers(ctx)
consumer get orders.created billing	Consumer(name).Get(ctx)
consumer worker list orders.created billing [key]	Consumer(name).Workers(ctx), displaying the stored config keys per worker

List accepts --quiet for names only. Get accepts --quiet for an existence check: no output, exit 0 when registered, exit 1 when absent. Both accept --output json; list returns the consumer array, and get returns the consumer row or null with exit 1 when absent. --quiet and --output json cannot be combined.

Worker list accepts an optional stored config key, such as exception_initial_backoff or message.timeout. Its JSON document contains stream, consumer, and keys, each with its worker name and stored value. Session settings such as ConsumeOptions.ClaimPollRate are not stored worker config. Workers are declared at Register; running instances refresh their stored settings at ConfigRefreshInterval.

Config

ConsumerConfig

Declared at Register, stored on the group’s worker_config rows, the same for every instance.

field	default	what it decides
Message	timeout 30s, retry MaxRetries: 3 on the default curve	the MessageOptions filling whatever the produced message left unset
MessageMin	nil	per-option floors on what a message may request
MessageMax	Message’s values	per-option ceilings; a message cannot request above the group’s defaults unless raised here
ConcurrencyOverride	"" (honor the message’s own)	run every message under this policy: ConcurrencyParallel, ConcurrencyExclusive, ConcurrencyOrdered
Start	sqlstreams.Beginning()	where a new group’s cursor is placed; read once, at creation (where a new group starts)
Bindings	nil (the whole stream)	the group’s whole pattern set (routing)
ExceptionInitialBackoff	5s	the first can_run_after delay on a fresh exception row
MaxRangeReclaims	3	reclaims after which a range is quarantined

ConsumeOptions

Passed to Consume, one session’s own.

field	default	what it decides
BatchLimit	4	messages claimed per poll
QueueSize	BatchLimit	claimed messages buffered ahead of processing; at least BatchLimit
MessageConcurrency	1	messages this instance processes at once
ClaimPollRate	500ms	how often an idle instance polls
QueueMargin	15s	lease padding for time a claim sits queued
RecordMargin	2s	lease padding for recording the outcome
TimeoutGrace	100ms	slack for a handler that respected ctx.Done() to unwind before the hard cutoff abandons it
SlowDispatchThreshold	0 (off)	a dispatch running longer logs SQL0039
InstanceTTL	30s	how long this instance’s worker_instance rows stay live without a heartbeat
BindingRetryInterval	10s	how often a waiting binding declaration is retried
ConfigRefreshInterval	30s	how often the stored group config is re-read: the staleness window for a redeclaration
ShutdownTimeout	MessageMax.Timeout + TimeoutGrace + RecordMargin	how long the drain waits for in-flight handlers before the rest is released
DisableGracefulShutdown	false	accept a context that can never be cancelled, leaving process exit as the only stop

Queue and lease budgets

See consumer tuning for workload-specific starting settings and the measurements to compare before raising them.

The default consumer claims four messages at a time and buffers up to four ordinary messages ahead of processing. Handlers remain serial unless MessageConcurrency is raised. Idle claim and exception polling share ClaimPollRate; a shorter interval also increases database query traffic.

The default range lease is 30s + 100ms + 15s + 2s = 47.1s. A crash can leave claimed messages waiting that long before a later poll reclaims them. ShutdownTimeout remains 32.1s: it covers in-flight processing, while QueueMargin covers waiting before processing.

For example, charge-cards processing ids 1–8 with a three-second handler can leave the next four messages waiting roughly twelve seconds. The fifteen-second margin covers that wait with some room for database work. Longer handlers need a smaller batch and queue or a larger QueueMargin. QueueSize must stay at least BatchLimit. An explicit queue below 4 requires an explicit compatible batch limit; set both to 1 for shallow prefetch. Choose the handler timeout from legitimate runtime; raising it and MessageMax.Timeout together does not increase queue allowance.

Ordered messages on one key remain serial and can wait for the committed cursor between ranges. ClaimPollRate is not a delivery-latency guarantee.

The Info-level starting log includes resolved session settings and the registration-time timeout budgets. Stored group settings are read when consumer workers start and on config refresh. SQL0105 warns when a queued message has insufficient lease time to start. Existing warning suppression collapses repeated warnings within the instance’s one-minute window.

Gotchas

• MessageConcurrency is how many messages this instance runs at once; ConcurrencyOverride is what a message key means. They sound alike and sit on opposite sides of the split.
• A group’s config wants exactly one declaring service. Two services declaring charge-cards differently overwrite each other on every restart, and the same SQL0059 line on every restart is the tell.
• Handlers should be idempotent: redelivery after a crash or timeout is normal at-least-once behavior (side effects and retries).
• Worker.Metadata is the stored document, sparse: an absent timeout means it was omitted, not that deliveries have no timeout. Defaults are resolved when an instance reads it, and worker-specific fields are outside the stable contract.