Consumer Group Config

A consumer group’s config splits in two. Group config is what the group means — the message timeout, the retry budget, the concurrency override, the bindings — and it must be the same for every instance of the group. Instance config is how one process runs — batch size, poll rate, shutdown budget — and it may differ per box. Both used to live in one struct on NewConsumer, and nothing made the group half actually shared: two replicas of charge-cards with different Message.Retry gave the same message a different retry budget depending on which replica claimed it. The verbs and every field’s default are on the Consumer reference thread.

One question sorts every field

If two instances of the same group set a field differently, is that a bug or a deployment choice? A bug -> group config. A choice -> instance config.

Group config — ConsumerConfig, handed to Register and written to the group’s rows by it:

field	what it decides
Message	default MessageOptions — timeout and retry curve — filling whatever the produced message left unset
MessageMin / MessageMax	floors and ceilings on what a message may request
ConcurrencyOverride	run every message under this policy, over the message’s own
Start	where a brand-new group’s cursor is placed — read once, at creation (where a new group starts)
Bindings	the group’s whole pattern set (routing); nil is the whole stream, and a group with any binding stops receiving messages produced without a routing key
ExceptionInitialBackoff	the first can_run_after delay written on a fresh exception row
MaxRangeReclaims	how many reclaims before a range is quarantined

The last two read like tuning, which is why they’re worth listing: both decide durable outcomes on shared rows, so instances disagreeing on them is exactly the divergence this split closes.

Instance config becomes ConsumeOptions, handed to Consume. A Consume call is already one session — fresh session id, counters at zero — and these are session facts: the throughput knobs (BatchLimit, QueueSize, MessageConcurrency), this process’s own latencies (ClaimPollRate, QueueMargin, RecordMargin, TimeoutGrace), and its lifecycle (InstanceTTL, ShutdownTimeout, DisableGracefulShutdown). A bigger box runs a bigger batch, and that’s a choice, so none of it is declared anywhere; every session gets its own. The ambient Logger and Retry are the client’s (ClientConfig), so neither struct carries them.

One near-collision to keep straight: MessageConcurrency is how many messages this instance processes at once (ConsumeOptions); ConcurrencyOverride is what a message key means (group config). They sound alike and sit on opposite sides of the split.

One rule places everything: the declaration at the handle’s Register verb, session options at the verb that runs. client.Stream[T](stream).Consumer(group).Register(ctx, cfg) carries the declaration it writes, the way client.Stream[T](name).Register already does. ConsumeOptions mirrors ProduceOptions: act verbs take options, nil for the defaults.

Declared at Register, stored on the row

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})

Consumer(name).Register resolves the stream, writes the declaration it was handed, and returns the typed instance. Newest wins, and an overwrite warns SQL0059 (what a declaration does).

Register writes the config into the group’s worker_config rows — the fleet table where every group-owned worker (message_consumer, exception_consumer, delivery_consumer) already declares its config as a metadata document. Those documents already carry Message, ConcurrencyOverride, ExceptionInitialBackoff, and MaxRangeReclaims; what changed is the direction — instances read the declared document back instead of each process writing in its own copy. The newest declaration wins, which is the metadata-replace path the fleet already has, and every replace appends a full snapshot to worker_config_log with declared_by and declared_at. No new table, no new column; an overwrite that changes values logs a warn naming the difference.

Two carve-outs from newest-wins:

• Bindings moves into ConsumerConfig, and only there — a binding set switch resequences deliveries, so it keeps its own install/wait outcomes (how a set change lands).
• Start is read once, when Register creates the cursor row. A later declaration’s Start changes nothing.

A declaration is the whole document. Fields left unset mean the library default; there is no field-level patch, and nil declares the empty document — the same total-set rule bindings already have. The cost: two services both declaring charge-cards with different configs will overwrite each other on every restart, visibly in the log but endlessly. There is no strict form. A group’s config wants exactly one declaring service, and that is a convention: the library enforces nothing, every differing overwrite logs SQL0059 naming the change, and the same line on every restart is the tell.

Sparse stored, resolved at read

The metadata document holds only the fields the declaration set — a worker’s fields land on that worker’s row. Everything unset resolves to the library default at read time, so a library upgrade moves the defaults of every group that never pinned them, and a pinned value stays pinned until redeclared. Storing resolved values instead would freeze whatever defaults the declaring library version had — rejected. The inspect verb shows both layers: the declared document and the effective values.

How a change reaches running instances

Never by a read on the claim path — claiming uses an in-memory copy only. Each instance keeps the group config in a mutex-guarded copy that a background loop refreshes from the group’s worker rows on a poll cadence. Redeclare -> every instance’s next refresh picks it up -> the group converges without a deploy. The staleness window is the refresh interval (ConsumeOptions.ConfigRefreshInterval, 30 seconds unless set); that’s the price of keeping config reads off the hot path.

Where each field takes effect:

• Message, MessageMin/MessageMax, ExceptionInitialBackoff, MaxRangeReclaims -> the next claim or exception write uses the current copy. A message mid-retry follows the new curve from its next attempt.
• ConcurrencyOverride -> pinned when a message is claimed; claims already handed out finish under the value they were claimed with. Without the pin, two instances straddling a refresh could resolve the same key exclusive and parallel at once and run it twice.
• ShutdownTimeout defaults from MessageMax.Timeout, so a refresh that moves the ceiling re-derives the shutdown budget with it.

Worked case: charge-cards runs MaxRetries: 3 and a flaky card processor needs more room. Redeclare with MaxRetries: 5 from the owning service (or the future CLI verb) and check the row:

SELECT w.name, w.metadata
FROM sqlstreams.worker_config w
JOIN sqlstreams.consumer_group_config g ON g.id = w.consumer_group_id
WHERE g.name = 'charge-cards';
--  name             | metadata
--  message_consumer | {"message": {"timeout": "10s", "retry": {"max_retries": 5}}, ...}

worker_config_log holds every earlier document, with who declared it and when.

Every replica follows within one refresh interval. A message sitting at attempt 3, about to dead-letter, gets attempts 4 and 5 under the new budget instead.

The same ladder everywhere

The client holds the ambient config once (ClientConfig: Schema, Logger, Retry, AllowDestroy). Every durable resource is declared by a Register verb that carries its config and stores it on the resource’s row: the stream with StreamConfig, the system with SystemConfig, the scheduler with SchedulerConfig beside the expression and payload. Session and per-call options ride the verb that acts: Consume, Produce.

The producer is the same ladder minus the stored half. Its Message defaults have no stored home on purpose: they are merged under each produce and written into the message’s own options document, so they are already durable, per message, where the consumer reads them. There is no durable producer-side resource to hang a config on, and inventing one would add a second place a message’s options come from.

The manager owns no config at all. Every knob its workers run under is a declared metadata document on a worker_config row, read from the fleet at provision, the same ecosystem the group config joins.