[IMPROVED] Defer consumer starting seq scan off meta goroutine

Signed-off-by: Maurice van Veen <github@mauricevanveen.com>

Backport of nats-io#8051 (6b884f4) adapted for release/v2.12.7.
The v2.12.7 signature of selectStartingSeqNo does not return an error, so the
leader-elect path logs/stepdown handling was simplified accordingly.

Author: MauriceVanVeen

server/consumer.go

@@ -1228,13 +1228,15 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 		}
 	}
 
+	standalone := !s.JetStreamIsClustered() && s.standAloneMode()
 	if o.store != nil && o.store.HasState() {
 		// Restore our saved state.
 		o.mu.Lock()
 		o.readStoredState()
 		o.mu.Unlock()
-	} else {
-		// Select starting sequence number
+	} else if config.Direct || standalone {
+		// Clustered non-direct consumers defer this to setLeader so the
+		// expensive store scans don't block the meta apply goroutine.
 		o.selectStartingSeqNo()
 	}
 
@@ -1304,7 +1306,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	mset.setConsumer(o)
 	mset.mu.Unlock()
 
-	if config.Direct || (!s.JetStreamIsClustered() && s.standAloneMode()) {
+	if config.Direct || standalone {
 		o.setLeader(true)
 	}
 
@@ -1472,18 +1474,29 @@ func (o *consumer) isLeader() bool {
 	return o.leader.Load()
 }
 
-func (o *consumer) setLeader(isLeader bool) {
+func (o *consumer) setLeader(isLeader bool) error {
 	o.mu.RLock()
 	mset, closed := o.mset, o.closed
 	movingToClustered := o.node != nil && o.pch == nil
 	movingToNonClustered := o.node == nil && o.pch != nil
 	wasLeader := o.leader.Swap(isLeader)
+
+	// For clustered new consumers, starting seq selection was deferred from
+	// addConsumerWithAssignment so the scan wouldn't block the meta apply
+	// goroutine, run it here on leader-elect instead.
+	needsSelect := isLeader && !wasLeader && o.dseq == 0 && (o.store == nil || !o.store.HasState())
 	o.mu.RUnlock()
 
 	// If we are here we have a change in leader status.
 	if isLeader {
 		if closed || mset == nil {
-			return
+			return nil
+		}
+
+		if needsSelect {
+			o.mu.Lock()
+			o.selectStartingSeqNo()
+			o.mu.Unlock()
 		}
 
 		if wasLeader {
@@ -1512,7 +1525,7 @@ func (o *consumer) setLeader(isLeader bool) {
 				}
 				o.mu.Unlock()
 			}
-			return
+			return nil
 		}
 
 		mset.mu.RLock()
@@ -1551,15 +1564,15 @@ func (o *consumer) setLeader(isLeader bool) {
 		if o.cfg.AckPolicy != AckNone {
 			if o.ackSub, err = o.subscribeInternal(o.ackSubj, o.pushAck); err != nil {
 				o.mu.Unlock()
-				return
+				return nil
 			}
 		}
 
 		// Setup the internal sub for next message requests regardless.
 		// Will error if wrong mode to provide feedback to users.
 		if o.reqSub, err = o.subscribeInternal(o.nextMsgSubj, o.processNextMsgReq); err != nil {
 			o.mu.Unlock()
-			return
+			return nil
 		}
 
 		// Check on flow control settings.
@@ -1568,7 +1581,7 @@ func (o *consumer) setLeader(isLeader bool) {
 			fcsubj := fmt.Sprintf(jsFlowControl, stream, o.name)
 			if o.fcSub, err = o.subscribeInternal(fcsubj, o.processFlowControl); err != nil {
 				o.mu.Unlock()
-				return
+				return nil
 			}
 		}
 
@@ -1704,6 +1717,7 @@ func (o *consumer) setLeader(isLeader bool) {
 		}
 		o.mu.Unlock()
 	}
+	return nil
 }
 
 // This is coming on the wire so do not block here.
@@ -3167,15 +3181,23 @@ func (o *consumer) infoWithSnapAndReply(snap bool, reply string) *ConsumerInfo {
 		return nil
 	}
 
+	dseq, sseq := o.dseq, o.sseq
+	if dseq <= 0 {
+		dseq = 1
+	}
+	if sseq <= 0 {
+		sseq = 1
+	}
+
 	cfg := o.cfg
 	info := &ConsumerInfo{
 		Stream:  o.stream,
 		Name:    o.name,
 		Created: o.created,
 		Config:  &cfg,
 		Delivered: SequenceInfo{
-			Consumer: o.dseq - 1,
-			Stream:   o.sseq - 1,
+			Consumer: dseq - 1,
+			Stream:   sseq - 1,
 		},
 		AckFloor: SequenceInfo{
 			Consumer: o.adflr,

server/jetstream_cluster.go

@@ -5806,10 +5806,15 @@ func (js *jetStream) processClusterCreateConsumer(oca, ca *consumerAssignment, s
 						func() {
 							defer s.grWG.Done()
 							defer o.clearMonitorRunning()
-							o.setLeader(true)
+							err := o.setLeader(true)
 							var resp = JSApiConsumerCreateResponse{ApiResponse: ApiResponse{Type: JSApiConsumerCreateResponseType}}
-							resp.ConsumerInfo = setDynamicConsumerInfoMetadata(o.info())
-							s.sendAPIResponse(client, acc, subject, reply, _EMPTY_, s.jsonResponse(&resp))
+							if err != nil {
+								resp.Error = NewJSConsumerCreateError(err, Unless(err))
+								s.sendAPIErrResponse(client, acc, subject, reply, _EMPTY_, s.jsonResponse(&resp))
+							} else {
+								resp.ConsumerInfo = setDynamicConsumerInfoMetadata(o.info())
+								s.sendAPIResponse(client, acc, subject, reply, _EMPTY_, s.jsonResponse(&resp))
+							}
 						},
 						pprofLabels{
 							"type":     "consumer",
@@ -6506,6 +6511,9 @@ func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry, isLea
 				if !o.isLeader() && sseq > o.sseq {
 					o.sseq = sseq
 				}
+				if o.dseq == 0 {
+					o.dseq = 1
+				}
 				if o.store != nil {
 					o.store.UpdateStarting(sseq - 1)
 				}
@@ -6677,7 +6685,9 @@ func (js *jetStream) processConsumerLeaderChangeWithAssignment(o *consumer, ca *
 	}
 
 	// Tell consumer to switch leader status.
-	o.setLeader(isLeader)
+	if lerr := o.setLeader(isLeader); lerr != nil && err == nil {
+		err = lerr
+	}
 
 	if !isLeader || hasResponded {
 		if isLeader {

server/jetstream_cluster_3_test.go

@@ -8233,3 +8233,64 @@ func TestJetStreamClusterStreamLeaderStepsDownIfSnapshotCatchupRequired(t *testi
 		return checkState(t, c, globalAccountName, "TEST")
 	})
 }
+
+func TestJetStreamClusterConsumerSelectStartingSeqDeferred(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	_, err = js.Publish("foo", nil)
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:   "C",
+		AckPolicy: nats.AckExplicitPolicy,
+		Replicas:  3,
+	})
+	require_NoError(t, err)
+
+	leader := c.consumerLeader(globalAccountName, "TEST", "C")
+	require_NotNil(t, leader)
+	follower := c.randomNonConsumerLeader(globalAccountName, "TEST", "C")
+	require_NotNil(t, follower)
+
+	getConsumer := func(s *Server) *consumer {
+		t.Helper()
+		mset, err := s.globalAccount().lookupStream("TEST")
+		require_NoError(t, err)
+		o := mset.lookupConsumer("C")
+		require_NotNil(t, o)
+		return o
+	}
+
+	// On the leader, selectStartingSeqNo ran inside setLeader(true).
+	l := getConsumer(leader)
+	l.mu.RLock()
+	ldseq, lsseq := l.dseq, l.sseq
+	l.mu.RUnlock()
+	require_Equal(t, ldseq, 1)
+	require_Equal(t, lsseq, 1)
+
+	// On the follower, meta apply must not have run selectStartingSeqNo.
+	checkFor(t, 2*time.Second, 200*time.Millisecond, func() error {
+		f := getConsumer(follower)
+		f.mu.RLock()
+		defer f.mu.RUnlock()
+		if f.dseq != 1 {
+			return fmt.Errorf("expected follower dseq 1, got %d", f.dseq)
+		}
+		if f.sseq != 1 {
+			return fmt.Errorf("expected follower sseq 1, got %d", f.sseq)
+		}
+		return nil
+	})
+}