Fix cross-tier JetStream consumer counting for limits

Project: nats-io/nats-server | Language: Go | Concepts: consistency-models, data-structures, testing-strategy | Difficulty: mid
Link: https://github.com/nats-io/nats-server/pull/8484

The problem

When enforcing JetStream consumer limits in a clustered setup with tiered account limits (e.g., R1 and R3 tiers), the server incorrectly counted all consumers on a stream against the per-tier limit, regardless of which tier those consumers belonged to. This caused users to hit consumer limits prematurely when mixing consumers of different tiers on the same stream.

Why it's non-trivial

A naive fix would just count all consumers and compare against the lowest configured limit, but stream-level and tier-level limits are meant to be enforced independently:

• The stream limit applies to all consumers on that stream.
• The tier limit applies only to consumers belonging to that specific tier. Mixing these up leads to incorrect enforcement, especially when both limits are present and consumers of different tiers coexist.

The approach

1. Determine both stream and tier consumer limits and count separately.
Instead of using a single counter and limit, the code now tracks both the total number of consumers (for the stream-level limit) and the number of consumers in the relevant tier (for the tier-level limit):

-	maxc := sa.Config.MaxConsumers
-	if maxc <= 0 || (selectedLimits.MaxConsumers > 0 && selectedLimits.MaxConsumers < maxc) {
-		maxc = selectedLimits.MaxConsumers
-	}
-	if maxc > 0 {
+	// The stream limit caps every consumer of the stream. The account limit
+	// caps only the consumers of the selected tier, so the two need separate
+	// counts.
+	streamMaxc := sa.Config.MaxConsumers
+	tierMaxc := selectedLimits.MaxConsumers
+	if streamMaxc > 0 || tierMaxc > 0 {

Annotation: This separates the stream-level (streamMaxc) and tier-level (tierMaxc) consumer limits, allowing the server to enforce both independently.

2. Count consumers per stream and per tier.
Within the consumer assignment loop, increment the overall total for the stream and, if the consumer matches the selected tier, increment the tier-specific total:

-	total := 0
+	var total, tierTotal int
 	for ca := range js.consumerAssignmentsOrInflightSeq(acc.Name, stream) {
 		if ca.unsupported != nil {
 			continue
 		}
 		if ca.Config != nil && !ca.Config.Direct && !ca.Config.Sourcing {
-			total++
+			total++
+			if selectedTier == _EMPTY_ || isSameTier(ca.Config.replicas(&streamCfg), selectedReplicas) {
+				tierTotal++
+			}
 		}
 	}
-	if total >= maxc {
+	if (streamMaxc > 0 && total >= streamMaxc) || (tierMaxc > 0 && tierTotal >= tierMaxc) {
 		resp.Error = NewJSMaximumConsumersLimitError()
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 		return

Annotation: This ensures that the stream limit (streamMaxc) applies to all consumers, while the tier limit (tierMaxc) only applies to consumers matching the selected tier.

3. Add a test that exercises mixed-tier consumer creation and limit enforcement.
A new test confirms that the fix works as intended by creating both R1 and R3 consumers and checking that limits for each tier are enforced separately:

func TestJetStreamJWTClusteredTieredR3StreamWithMixedReplicaConsumers(t *testing.T) {
    // ... setup omitted ...
    // Add 2 R3 consumers
    for i := range 2 {
        _, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
            Name:      fmt.Sprintf("R3-%d", i+1),
            AckPolicy: nats.AckExplicitPolicy,
            Replicas:  3,
        })
        require_NoError(t, err)
    }

    // Add 1 R1 consumer
    _, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
        Name:      "R1-1",
        AckPolicy: nats.AckExplicitPolicy,
        Replicas:  1,
    })
    require_NoError(t, err)

    // Add the 3rd R3 consumer (allowed by limits)
    _, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
        Name:      "R3-3",
        AckPolicy: nats.AckExplicitPolicy,
        Replicas:  3,
    })
    require_NoError(t, err)

    // Add the 2nd R1 consumer (allowed by limits)
    _, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
        Name:      "R1-2",
        AckPolicy: nats.AckExplicitPolicy,
        Replicas:  1,
    })
    require_NoError(t, err)

    // Both tiers are full now, so one more consumer of each tier must fail.
    _, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
        Name:      "R3-4",
        AckPolicy: nats.AckExplicitPolicy,
        Replicas:  3,
    })
    require_Error(t, err, NewJSMaximumConsumersLimitError())

    _, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
        Name:      "R1-3",
        AckPolicy: nats.AckExplicitPolicy,
        Replicas:  1,
    })
    require_Error(t, err, NewJSMaximumConsumersLimitError())
    // ... final assertions ...
}

Annotation: This test confirms that the server allows up to the per-tier limit for each tier, and rejects consumers that would exceed those limits.

From the review thread

Generally LGTM, but could you squash into just one commit?

@MauriceVanVeen done

Takeaways

• When enforcing resource limits with multiple scopes (e.g., per-stream and per-tier), always count and enforce limits independently for each scope.
• Tests that mix resource types or tiers are essential to catch cross-scope accounting bugs.
• When updating limit logic, ensure that legacy/single-scope behavior is preserved for non-tiered configurations.