Repository: nats-io/nats-server
Pull request #4884 (2023-12-14 00:47:58)

[IMPROVED] JSAPI internal routing and reporting and Source and Mirror setup
A customer had an issue with JS api requests backing up in the internal queue we use to avoid blocking non-clients. The reason for the buildup was known due to time spent in NumPending calculations for R1 consumers with wildcard filtered subjects and large subject/key space. These consumers were being created from a large fan in stream with many sources.

1. We now properly track the queue depth of this internal queue, warn if we are getting backed up and make sure to properly report under JS API usage stats - Inflight.
2. Use multiple Go routines in a pool to process this API queue vs 1.
3. When we fail to create a consumer for a mIrror or a source, back off retries.
4. When we fail do immediate cleanup, do not wait.
5. Only create internal interest for the consumer once we get a response from the source stream.

Signed-off-by: Derek Collison <derek@nats.io>



Patch:
diff --git a/server/consumer.go b/server/consumer.go
index 68f55418..ef0eb7d2 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -3350,7 +3350,7 @@ func (o *consumer) notifyDeliveryExceeded(sseq, dc uint64) {
 	o.sendAdvisory(o.deliveryExcEventT, j)
 }
 
-// Check if the candidate subject matches a filter if its present.
+// Check to see if the candidate subject matches a filter if its present.
 // Lock should be held.
 func (o *consumer) isFilteredMatch(subj string) bool {
 	// No filter is automatic match.
@@ -3374,25 +3374,6 @@ func (o *consumer) isFilteredMatch(subj string) bool {
 	return false
 }
 
-// Check if the candidate filter subject is equal to or a subset match
-// of one of the filter subjects.
-// Lock should be held.
-func (o *consumer) isEqualOrSubsetMatch(subj string) bool {
-	for _, filter := range o.subjf {
-		if !filter.hasWildcard && subj == filter.subject {
-			return true
-		}
-	}
-	tsa := [32]string{}
-	tts := tokenizeSubjectIntoSlice(tsa[:0], subj)
-	for _, filter := range o.subjf {
-		if isSubsetMatchTokenized(filter.tokenizedSubject, tts) {
-			return true
-		}
-	}
-	return false
-}
-
 var (
 	errMaxAckPending = errors.New("max ack pending reached")
 	errBadConsumer   = errors.New("consumer not valid")
diff --git a/server/filestore.go b/server/filestore.go
index 5eac125b..573edb29 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -2591,6 +2591,8 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 		}
 	}
 
+	var tsa, fsa [32]string
+	fts := tokenizeSubjectIntoSlice(fsa[:0], filter)
 	isAll := filter == _EMPTY_ || filter == fwcs
 	wc := subjectHasWildcard(filter)
 
@@ -2607,9 +2609,6 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 		return fs.state.LastSeq - sseq + 1, validThrough
 	}
 
-	var tsa, fsa [32]string
-	fts := tokenizeSubjectIntoSlice(fsa[:0], filter)
-
 	isMatch := func(subj string) bool {
 		if isAll {
 			return true
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index c4163c85..96ec1f62 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -21,6 +21,7 @@ import (
 	"math/rand"
 	"os"
 	"path/filepath"
+	"runtime"
 	"sort"
 	"strconv"
 	"strings"
@@ -749,7 +750,7 @@ func (js *jetStream) apiDispatch(sub *subscription, c *client, acc *Account, sub
 		}
 	}
 
-	// Shortcircuit.
+	// Short circuit for no interest.
 	if len(rr.psubs)+len(rr.qsubs) == 0 {
 		return
 	}
@@ -774,20 +775,30 @@ func (js *jetStream) apiDispatch(sub *subscription, c *client, acc *Account, sub
 
 	// If we are here we have received this request over a non-client connection.
 	// We need to make sure not to block. We will send the request to a long-lived
-	// go routine.
+	// pool of go routines.
+
+	// Increment inflight. Do this before queueing.
+	atomic.AddInt64(&js.apiInflight, 1)
 
 	// Copy the state. Note the JSAPI only uses the hdr index to piece apart the
 	// header from the msg body. No other references are needed.
-	s.jsAPIRoutedReqs.push(&jsAPIRoutedReq{jsub, sub, acc, subject, reply, copyBytes(rmsg), c.pa})
+	// Check pending and warn if getting backed up.
+	const warnThresh = 32
+	pending := s.jsAPIRoutedReqs.push(&jsAPIRoutedReq{jsub, sub, acc, subject, reply, copyBytes(rmsg), c.pa})
+	if pending > warnThresh {
+		s.RateLimitWarnf("JetStream request queue has high pending count: %d", pending)
+	}
 }
 
 func (s *Server) processJSAPIRoutedRequests() {
 	defer s.grWG.Done()
 
-	s.mu.Lock()
+	s.mu.RLock()
 	queue := s.jsAPIRoutedReqs
 	client := &client{srv: s, kind: JETSTREAM}
-	s.mu.Unlock()
+	s.mu.RUnlock()
+
+	js := s.getJetStream()
 
 	for {
 		select {
@@ -800,6 +811,7 @@ func (s *Server) processJSAPIRoutedRequests() {
 				if dur := time.Since(start); dur >= readLoopReportThreshold {
 					s.Warnf("Internal subscription on %q took too long: %v", r.subject, dur)
 				}
+				atomic.AddInt64(&js.apiInflight, -1)
 			}
 			queue.recycle(&reqs)
 		case <-s.quitCh:
@@ -816,8 +828,16 @@ func (s *Server) setJetStreamExportSubs() error {
 
 	// Start the go routine that will process API requests received by the
 	// subscription below when they are coming from routes, etc..
+	const maxProcs = 16
+	mp := runtime.GOMAXPROCS(0)
+	// Cap at 16 max for now on larger core setups.
+	if mp > maxProcs {
+		mp = maxProcs
+	}
 	s.jsAPIRoutedReqs = newIPQueue[*jsAPIRoutedReq](s, "Routed JS API Requests")
-	s.startGoRoutine(s.processJSAPIRoutedRequests)
+	for i := 0; i < mp; i++ {
+		s.startGoRoutine(s.processJSAPIRoutedRequests)
+	}
 
 	// This is the catch all now for all JetStream API calls.
 	if _, err := s.sysSubscribe(jsAllAPI, js.apiDispatch); err != nil {
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 9ca7fd9c..0995e7f5 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -8080,6 +8080,7 @@ func (mset *stream) processCatchupMsg(msg []byte) (uint64, error) {
 	st := mset.cfg.Storage
 	ddloaded := mset.ddloaded
 	tierName := mset.tier
+	replicas := mset.cfg.Replicas
 
 	if mset.hasAllPreAcks(seq, subj) {
 		mset.clearAllPreAcks(seq)
@@ -8090,7 +8091,7 @@ func (mset *stream) processCatchupMsg(msg []byte) (uint64, error) {
 
 	if mset.js.limitsExceeded(st) {
 		return 0, NewJSInsufficientResourcesError()
-	} else if exceeded, apiErr := mset.jsa.limitsExceeded(st, tierName, mset.cfg.Replicas); apiErr != nil {
+	} else if exceeded, apiErr := mset.jsa.limitsExceeded(st, tierName, replicas); apiErr != nil {
 		return 0, apiErr
 	} else if exceeded {
 		return 0, NewJSInsufficientResourcesError()
diff --git a/server/stream.go b/server/stream.go
index 2241efdb..9de84cbc 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -309,6 +309,7 @@ type sourceInfo struct {
 	start time.Time
 	lag   uint64
 	err   *ApiError
+	fails int
 	last  time.Time
 	lreq  time.Time
 	qch   chan struct{}
@@ -2001,10 +2002,11 @@ func (mset *stream) purge(preq *JSApiStreamPurgeRequest) (purged uint64, err err
 		// no subject was specified, we can purge all consumers sequences
 		doPurge := preq == nil ||
 			preq.Subject == _EMPTY_ ||
-			// consumer filter subject is equal to purged subject
-			// or consumer filter subject is subset of purged subject,
+			// or consumer filter subject is equal to purged subject
+			preq.Subject == o.cfg.FilterSubject ||
+			// or consumer subject is subset of purged subject,
 			// but not the other way around.
-			o.isEqualOrSubsetMatch(preq.Subject)
+			subjectIsSubsetMatch(o.cfg.FilterSubject, preq.Subject)
 		o.mu.RUnlock()
 		if doPurge {
 			o.purge(fseq, lseq)
@@ -2390,13 +2392,30 @@ func (mset *stream) skipMsgs(start, end uint64) {
 	}
 }
 
+const (
+	// Base retry backoff duration.
+	retryBackOff = 5 * time.Second
+	// Maximum amount we will wait.
+	retryMaximum = 2 * time.Minute
+)
+
+// Calculate our backoff based on number of failures.
+func calculateRetryBackoff(fails int) time.Duration {
+	backoff := time.Duration(retryBackOff) * time.Duration(fails*2)
+	if backoff > retryMaximum {
+		backoff = retryMaximum
+	}
+	return backoff
+}
+
 // This will schedule a call to setupMirrorConsumer, taking into account the last
-// time it was retried and determine the soonest setSourceConsumer can be called
-// without tripping the sourceConsumerRetryThreshold.
+// time it was retried and determine the soonest setupMirrorConsumer can be called
+// without tripping the sourceConsumerRetryThreshold. We will also take into account
+// number of failures and will back off our retries.
 // The mset.mirror pointer has been verified to be not nil by the caller.
 //
 // Lock held on entry
-func (mset *stream) scheduleSetupMirrorConsumerRetryAsap() {
+func (mset *stream) scheduleSetupMirrorConsumerRetry() {
 	// We are trying to figure out how soon we can retry. setupMirrorConsumer will reject
 	// a retry if last was done less than "sourceConsumerRetryThreshold" ago.
 	next := sourceConsumerRetryThreshold - time.Since(mset.mirror.lreq)
@@ -2404,9 +2423,12 @@ func (mset *stream) scheduleSetupMirrorConsumerRetryAsap() {
 		// It means that we have passed the threshold and so we are ready to go.
 		next = 0
 	}
-	// To make *sure* that the next request will not fail, add a bit of buffer
-	// and some randomness.
-	next += time.Duration(rand.Intn(int(10*time.Millisecond))) + 10*time.Millisecond
+	// Take into account failures here.
+	next += calculateRetryBackoff(mset.mirror.fails)
+
+	// Add some jitter.
+	next += time.Duration(rand.Intn(int(100*time.Millisecond))) + 100*time.Millisecond
+
 	time.AfterFunc(next, func() {
 		mset.mu.Lock()
 		mset.setupMirrorConsumer()
@@ -2417,6 +2439,9 @@ func (mset *stream) scheduleSetupMirrorConsumerRetryAsap() {
 // Setup our mirror consumer.
 // Lock should be held.
 func (mset *stream) setupMirrorConsumer() error {
+	if mset.closed {
+		return errStreamClosed
+	}
 	if mset.outq == nil {
 		return errors.New("outq required")
 	}
@@ -2448,7 +2473,7 @@ func (mset *stream) setupMirrorConsumer() error {
 	// We want to throttle here in terms of how fast we request new consumers,
 	// or if the previous is still in progress.
 	if last := time.Since(mirror.lreq); last < sourceConsumerRetryThreshold || mirror.sip {
-		mset.scheduleSetupMirrorConsumerRetryAsap()
+		mset.scheduleSetupMirrorConsumerRetry()
 		return nil
 	}
 	mirror.lreq = time.Now()
@@ -2505,22 +2530,23 @@ func (mset *stream) setupMirrorConsumer() error {
 		mirror.sf = mset.cfg.Mirror.FilterSubject
 	}
 
-	sfs := make([]string, len(mset.cfg.Mirror.SubjectTransforms))
-	trs := make([]*subjectTransform, len(mset.cfg.Mirror.SubjectTransforms))
+	if lst := len(mset.cfg.Mirror.SubjectTransforms); lst > 0 {
+		sfs := make([]string, lst)
+		trs := make([]*subjectTransform, lst)
 
-	for i, tr := range mset.cfg.Mirror.SubjectTransforms {
-		// will not fail as already checked before that the transform will work
-		subjectTransform, err := NewSubjectTransform(tr.Source, tr.Destination)
-		if err != nil {
-			mset.srv.Errorf("Unable to get transform for mirror consumer: %v", err)
+		for i, tr := range mset.cfg.Mirror.SubjectTransforms {
+			// will not fail as already checked before that the transform will work
+			subjectTransform, err := NewSubjectTransform(tr.Source, tr.Destination)
+			if err != nil {
+				mset.srv.Errorf("Unable to get transform for mirror consumer: %v", err)
+			}
+			sfs[i] = tr.Source
+			trs[i] = subjectTransform
 		}
-
-		sfs[i] = tr.Source
-		trs[i] = subjectTransform
+		mirror.sfs = sfs
+		mirror.trs = trs
+		req.Config.FilterSubjects = sfs
 	}
-	mirror.sfs = sfs
-	mirror.trs = trs
-	req.Config.FilterSubjects = sfs
 
 	respCh := make(chan *JSApiConsumerCreateResponse, 1)
 	reply := infoReplySubject()
@@ -2538,7 +2564,7 @@ func (mset *stream) setupMirrorConsumer() error {
 	})
 	if err != nil {
 		mirror.err = NewJSMirrorConsumerSetupFailedError(err, Unless(err))
-		mset.scheduleSetupMirrorConsumerRetryAsap()
+		mset.scheduleSetupMirrorConsumerRetry()
 		return nil
 	}
 
@@ -2556,26 +2582,9 @@ func (mset *stream) setupMirrorConsumer() error {
 		subject = strings.ReplaceAll(subject, "..", ".")
 	}
 
-	// We need to create the subscription that will receive the messages prior
-	// to sending the consumer create request, because in some complex topologies
-	// with gateways and optimistic mode, it is possible that the consumer starts
-	// delivering messages as soon as the consumer request is received.
-	qname := fmt.Sprintf("[ACC:%s] stream mirror '%s' of '%s' msgs", mset.acc.Name, mset.cfg.Name, mset.cfg.Mirror.Name)
-	// Create a new queue each time
-	mirror.msgs = newIPQueue[*inMsg](mset.srv, qname)
-	msgs := mirror.msgs
-	sub, err := mset.subscribeInternal(deliverSubject, func(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
-		hdr, msg := c.msgParts(copyBytes(rmsg)) // Need to copy.
-		mset.queueInbound(msgs, subject, reply, hdr, msg)
-	})
-	if err != nil {
-		mirror.err = NewJSMirrorConsumerSetupFailedError(err, Unless(err))
-		mset.unsubscribeUnlocked(crSub)
-		mset.scheduleSetupMirrorConsumerRetryAsap()
-		return nil
-	}
+	// Reset
+	mirror.msgs = nil
 	mirror.err = nil
-	mirror.sub = sub
 	mirror.sip = true
 
 	// Send the consumer create request
@@ -2591,7 +2600,13 @@ func (mset *stream) setupMirrorConsumer() error {
 				mset.mirror.sip = false
 				// If we need to retry, schedule now
 				if retry {
-					mset.scheduleSetupMirrorConsumerRetryAsap()
+					mset.mirror.fails++
+					// Cancel here since we can not do anything with this consumer at this point.
+					mset.cancelSourceInfo(mset.mirror)
+					mset.scheduleSetupMirrorConsumerRetry()
+				} else {
+					// Clear on success.
+					mset.mirror.fails = 0
 				}
 			}
 			mset.mu.Unlock()
@@ -2617,7 +2632,26 @@ func (mset *stream) setupMirrorConsumer() error {
 				mirror.err = ccr.Error
 				// Let's retry as soon as possible, but we are gated by sourceConsumerRetryThreshold
 				retry = true
+				mset.mu.Unlock()
+				return
 			} else {
+				// Setup actual subscription to process messages from our source.
+				qname := fmt.Sprintf("[ACC:%s] stream mirror '%s' of '%s' msgs", mset.acc.Name, mset.cfg.Name, mset.cfg.Mirror.Name)
+				// Create a new queue each time
+				mirror.msgs = newIPQueue[*inMsg](mset.srv, qname)
+				msgs := mirror.msgs
+				sub, err := mset.subscribeInternal(deliverSubject, func(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
+					hdr, msg := c.msgParts(copyBytes(rmsg)) // Need to copy.
+					mset.queueInbound(msgs, subject, reply, hdr, msg)
+				})
+				if err != nil {
+					mirror.err = NewJSMirrorConsumerSetupFailedError(err, Unless(err))
+					retry = true
+					mset.mu.Unlock()
+					return
+				}
+				// Save our sub.
+				mirror.sub = sub
 
 				// When an upstream stream expires messages or in general has messages that we want
 				// that are no longer available we need to adjust here.
@@ -2733,8 +2767,10 @@ func (mset *stream) cancelSourceInfo(si *sourceInfo) {
 		close(si.qch)
 		si.qch = nil
 	}
-	si.msgs.drain()
-	si.msgs.unregister()
+	if si.msgs != nil {
+		si.msgs.drain()
+		si.msgs.unregister()
+	}
 }
 
 const sourceConsumerRetryThreshold = 2 * time.Second
@@ -2744,7 +2780,7 @@ const sourceConsumerRetryThreshold = 2 * time.Second
 // without tripping the sourceConsumerRetryThreshold.
 //
 // Lock held on entry
-func (mset *stream) scheduleSetSourceConsumerRetryAsap(si *sourceInfo, seq uint64, startTime time.Time) {
+func (mset *stream) scheduleSetSourceConsumerRetry(si *sourceInfo, seq uint64, startTime time.Time) {
 	// We are trying to figure out how soon we can retry. setSourceConsumer will reject
 	// a retry if last was done less than "sourceConsumerRetryThreshold" ago.
 	next := sourceConsumerRetryThreshold - time.Since(si.lreq)
@@ -2752,16 +2788,19 @@ func (mset *stream) scheduleSetSourceConsumerRetryAsap(si *sourceInfo, seq uint6
 		// It means that we have passed the threshold and so we are ready to go.
 		next = 0
 	}
+	// Take into account failures here.
+	next += calculateRetryBackoff(si.fails)
+
 	// To make *sure* that the next request will not fail, add a bit of buffer
 	// and some randomness.
 	next += time.Duration(rand.Intn(int(10*time.Millisecond))) + 10*time.Millisecond
-	mset.scheduleSetSourceConsumerRetry(si.iname, seq, next, startTime)
+	mset.scheduleSetSourceConsumer(si.iname, seq, next, startTime)
 }
 
 // Simply schedules setSourceConsumer at the given delay.
 //
 // Lock held on entry
-func (mset *stream) scheduleSetSourceConsumerRetry(iname string, seq uint64, delay time.Duration, startTime time.Time) {
+func (mset *stream) scheduleSetSourceConsumer(iname string, seq uint64, delay time.Duration, startTime time.Time) {
 	if mset.sourceRetries == nil {
 		mset.sourceRetries = map[string]*time.Timer{}
 	}
@@ -2783,6 +2822,11 @@ func (mset *stream) scheduleSetSourceConsumerRetry(iname string, seq uint64, del
 
 // Lock should be held.
 func (mset *stream) setSourceConsumer(iname string, seq uint64, startTime time.Time) {
+	// Ignore if closed.
+	if mset.closed {
+		return
+	}
+
 	si := mset.sources[iname]
 	if si == nil {
 		return
@@ -2798,7 +2842,7 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64, startTime time.T
 	// We want to throttle here in terms of how fast we request new consumers,
 	// or if the previous is still in progress.
 	if last := time.Since(si.lreq); last < sourceConsumerRetryThreshold || si.sip {
-		mset.scheduleSetSourceConsumerRetryAsap(si, seq, startTime)
+		mset.scheduleSetSourceConsumerRetry(si, seq, startTime)
 		return
 	}
 	si.lreq = time.Now()
@@ -2877,7 +2921,7 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64, startTime time.T
 	})
 	if err != nil {
 		si.err = NewJSSourceConsumerSetupFailedError(err, Unless(err))
-		mset.scheduleSetSourceConsumerRetryAsap(si, seq, startTime)
+		mset.scheduleSetSourceConsumerRetry(si, seq, startTime)
 		return
 	}
 
@@ -2903,26 +2947,9 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64, startTime time.T
 	// Marshal request.
 	b, _ := json.Marshal(req)
 
-	// We need to create the subscription that will receive the messages prior
-	// to sending the consumer create request, because in some complex topologies
-	// with gateways and optimistic mode, it is possible that the consumer starts
-	// delivering messages as soon as the consumer request is received.
-	qname := fmt.Sprintf("[ACC:%s] stream source '%s' from '%s' msgs", mset.acc.Name, mset.cfg.Name, si.name)
-	// Create a new queue each time
-	si.msgs = newIPQueue[*inMsg](mset.srv, qname)
-	msgs := si.msgs
-	sub, err := mset.subscribeInternal(deliverSubject, func(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
-		hdr, msg := c.msgParts(copyBytes(rmsg)) // Need to copy.
-		mset.queueInbound(msgs, subject, reply, hdr, msg)
-	})
-	if err != nil {
-		si.err = NewJSSourceConsumerSetupFailedError(err, Unless(err))
-		mset.unsubscribeUnlocked(crSub)
-		mset.scheduleSetSourceConsumerRetryAsap(si, seq, startTime)
-		return
-	}
+	// Reset
+	si.msgs = nil
 	si.err = nil
-	si.sub = sub
 	si.sip = true
 
 	// Send the consumer create request
@@ -2938,7 +2965,13 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64, startTime time.T
 				si.sip = false
 				// If we need to retry, schedule now
 				if retry {
-					mset.scheduleSetSourceConsumerRetryAsap(si, seq, startTime)
+					si.fails++
+					// Cancel here since we can not do anything with this consumer at this point.
+					mset.cancelSourceInfo(si)
+					mset.scheduleSetSourceConsumerRetry(si, seq, startTime)
+				} else {
+					// Clear on success.
+					si.fails = 0
 				}
 			}
 			mset.mu.Unlock()
@@ -2953,7 +2986,7 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64, startTime time.T
 			ready := sync.WaitGroup{}
 			mset.mu.Lock()
 			// Check that it has not been removed or canceled (si.sub would be nil)
-			if si := mset.sources[iname]; si != nil && si.sub != nil {
+			if si := mset.sources[iname]; si != nil {
 				si.err = nil
 				if ccr.Error != nil || ccr.ConsumerInfo == nil {
 					// Note: this warning can happen a few times when starting up the server when sourcing streams are
@@ -2963,7 +2996,27 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64, startTime time.T
 					si.err = ccr.Error
 					// Let's retry as soon as possible, but we are gated by sourceConsumerRetryThreshold
 					retry = true
+					mset.mu.Unlock()
+					return
 				} else {
+					// Setup actual subscription to process messages from our source.
+					qname := fmt.Sprintf("[ACC:%s] stream source '%s' from '%s' msgs", mset.acc.Name, mset.cfg.Name, si.name)
+					// Create a new queue each time
+					si.msgs = newIPQueue[*inMsg](mset.srv, qname)
+					msgs := si.msgs
+					sub, err := mset.subscribeInternal(deliverSubject, func(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
+						hdr, msg := c.msgParts(copyBytes(rmsg)) // Need to copy.
+						mset.queueInbound(msgs, subject, reply, hdr, msg)
+					})
+					if err != nil {
+						si.err = NewJSSourceConsumerSetupFailedError(err, Unless(err))
+						retry = true
+						mset.mu.Unlock()
+						return
+					}
+					// Save our sub.
+					si.sub = sub
+
 					if si.sseq != ccr.ConsumerInfo.Delivered.Stream {
 						si.sseq = ccr.ConsumerInfo.Delivered.Stream + 1
 					}
@@ -3488,7 +3541,7 @@ func (mset *stream) subscribeToStream() error {
 		mset.mirror.sfs = sfs
 		mset.mirror.trs = trs
 		// delay the actual mirror consumer creation for after a delay
-		mset.scheduleSetupMirrorConsumerRetryAsap()
+		mset.scheduleSetupMirrorConsumerRetry()
 	} else if len(mset.cfg.Sources) > 0 {
 		// Setup the initial source infos for the sources
 		mset.resetSourceInfo()
diff --git a/server/subject_transform.go b/server/subject_transform.go
index e7d72576..63876f43 100644
--- a/server/subject_transform.go
+++ b/server/subject_transform.go
@@ -561,29 +561,3 @@ func (tr *subjectTransform) reverse() *subjectTransform {
 	rtr, _ := NewSubjectTransformStrict(nsrc, ndest)
 	return rtr
 }
-
-// Will share relevant info regarding the subject.
-// Returns valid, tokens, num pwcs, has fwc.
-func subjectInfo(subject string) (bool, []string, int, bool) {
-	if subject == "" {
-		return false, nil, 0, false
-	}
-	npwcs := 0
-	sfwc := false
-	tokens := strings.Split(subject, tsep)
-	for _, t := range tokens {
-		if len(t) == 0 || sfwc {
-			return false, nil, 0, false
-		}
-		if len(t) > 1 {
-			continue
-		}
-		switch t[0] {
-		case fwc:
-			sfwc = true
-		case pwc:
-			npwcs++
-		}
-	}
-	return true, tokens, npwcs, sfwc
-}
diff --git a/server/sublist.go b/server/sublist.go
index 781458ed..a78ff86c 100644
--- a/server/sublist.go
+++ b/server/sublist.go
@@ -1101,6 +1101,32 @@ func IsValidSubject(subject string) bool {
 	return true
 }
 
+// Will share relevant info regarding the subject.
+// Returns valid, tokens, num pwcs, has fwc.
+func subjectInfo(subject string) (bool, []string, int, bool) {
+	if subject == "" {
+		return false, nil, 0, false
+	}
+	npwcs := 0
+	sfwc := false
+	tokens := strings.Split(subject, tsep)
+	for _, t := range tokens {
+		if len(t) == 0 || sfwc {
+			return false, nil, 0, false
+		}
+		if len(t) > 1 {
+			continue
+		}
+		switch t[0] {
+		case fwc:
+			sfwc = true
+		case pwc:
+			npwcs++
+		}
+	}
+	return true, tokens, npwcs, sfwc
+}
+
 // IsValidLiteralSubject returns true if a subject is valid and literal (no wildcards), false otherwise
 func IsValidLiteralSubject(subject string) bool {
 	return isValidLiteralSubject(strings.Split(subject, tsep))


Test patch:
diff --git a/server/jetstream_cluster_1_test.go b/server/jetstream_cluster_1_test.go
index ea8d70ab..644543f2 100644
--- a/server/jetstream_cluster_1_test.go
+++ b/server/jetstream_cluster_1_test.go
@@ -5353,7 +5353,7 @@ func TestJetStreamClusterSourcesFilteringAndUpdating(t *testing.T) {
 
 	checkSync := func(msgsTest, msgsM uint64) {
 		t.Helper()
-		checkFor(t, 10*time.Second, 500*time.Millisecond, func() error {
+		checkFor(t, 20*time.Second, 500*time.Millisecond, func() error {
 			if tsi, err := js.StreamInfo("TEST"); err != nil {
 				return err
 			} else if msi, err := js.StreamInfo("M"); err != nil {
@@ -5604,7 +5604,7 @@ func TestJetStreamClusterMirrorAndSourcesClusterRestart(t *testing.T) {
 
 		checkSync := func(msgsTest, msgsM uint64) {
 			t.Helper()
-			checkFor(t, 10*time.Second, 500*time.Millisecond, func() error {
+			checkFor(t, 20*time.Second, 500*time.Millisecond, func() error {
 				if tsi, err := js.StreamInfo("TEST"); err != nil {
 					return err
 				} else if msi, err := js.StreamInfo("M"); err != nil {
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index 1e2cb80a..819a3a32 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -1010,7 +1010,7 @@ func TestJetStreamClusterSourceWithOptStartTime(t *testing.T) {
 
 		checkCount := func(sname string, expected int) {
 			t.Helper()
-			checkFor(t, 2*time.Second, 50*time.Millisecond, func() error {
+			checkFor(t, 10*time.Second, 50*time.Millisecond, func() error {
 				si, err := js.StreamInfo(sname)
 				if err != nil {
 					return err
diff --git a/server/jetstream_consumer_test.go b/server/jetstream_consumer_test.go
index ec8fdf12..ed3de3ad 100644
--- a/server/jetstream_consumer_test.go
+++ b/server/jetstream_consumer_test.go
@@ -871,36 +871,6 @@ func TestJetStreamConsumerIsFilteredMatch(t *testing.T) {
 	}
 }
 
-func TestJetStreamConsumerIsEqualOrSubsetMatch(t *testing.T) {
-	for _, test := range []struct {
-		name           string
-		filterSubjects []string
-		subject        string
-		result         bool
-	}{
-		{"no filter", []string{}, "foo.bar", false},
-		{"literal match", []string{"foo.baz", "foo.bar"}, "foo.bar", true},
-		{"literal mismatch", []string{"foo.baz", "foo.bar"}, "foo.ban", false},
-		{"literal match", []string{"bar.>", "foo.>"}, "foo.>", true},
-		{"subset match", []string{"bar.foo.>", "foo.bar.>"}, "bar.>", true},
-		{"subset mismatch", []string{"bar.>", "foo.>"}, "baz.foo.>", false},
-		{"literal match", filterSubjects(100), "foo.bar.*.xyz.abcdef", true},
-		{"subset match", filterSubjects(100), "foo.bar.>", true},
-	} {
-		test := test
-
-		t.Run(test.name, func(t *testing.T) {
-			t.Parallel()
-
-			c := consumerWithFilterSubjects(test.filterSubjects)
-			if res := c.isEqualOrSubsetMatch(test.subject); res != test.result {
-				t.Fatalf("Subject %q subset match of %v, should be %v, got %v",
-					test.subject, test.filterSubjects, test.result, res)
-			}
-		})
-	}
-}
-
 func Benchmark____JetStreamConsumerIsFilteredMatch(b *testing.B) {
 	subject := "foo.bar.do.not.match.any.filter.subject"
 	for n := 1; n <= 1024; n *= 2 {
diff --git a/server/jetstream_super_cluster_test.go b/server/jetstream_super_cluster_test.go
index f209f089..b7b7f69f 100644
--- a/server/jetstream_super_cluster_test.go
+++ b/server/jetstream_super_cluster_test.go
@@ -659,7 +659,7 @@ func TestJetStreamSuperClusterConsumersBrokenGateways(t *testing.T) {
 	}
 
 	// Make sure we can deal with data loss at the end.
-	checkFor(t, 10*time.Second, 250*time.Millisecond, func() error {
+	checkFor(t, 20*time.Second, 250*time.Millisecond, func() error {
 		si, err := js.StreamInfo("S")
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
diff --git a/server/norace_test.go b/server/norace_test.go
index b4504893..022d52ab 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -1751,7 +1751,6 @@ func TestNoRaceJetStreamSuperClusterMixedModeMirrors(t *testing.T) {
 }
 
 func TestNoRaceJetStreamSuperClusterSources(t *testing.T) {
-
 	sc := createJetStreamSuperCluster(t, 3, 3)
 	defer sc.shutdown()
 
@@ -1890,7 +1889,7 @@ func TestNoRaceJetStreamSuperClusterSources(t *testing.T) {
 	sc.clusterForName("C3").waitOnStreamLeader("$G", "MS2")
 	<-doneCh
 
-	checkFor(t, 15*time.Second, time.Second, func() error {
+	checkFor(t, 20*time.Second, time.Second, func() error {
 		si, err := js2.StreamInfo("MS2")
 		if err != nil {
 			return err
@@ -2116,7 +2115,7 @@ func TestNoRaceJetStreamClusterMirrorExpirationAndMissingSequences(t *testing.T)
 
 	checkStream := func(stream string, num uint64) {
 		t.Helper()
-		checkFor(t, 10*time.Second, 50*time.Millisecond, func() error {
+		checkFor(t, 20*time.Second, 20*time.Millisecond, func() error {
 			si, err := js.StreamInfo(stream)
 			if err != nil {
 				return err
@@ -2134,7 +2133,7 @@ func TestNoRaceJetStreamClusterMirrorExpirationAndMissingSequences(t *testing.T)
 	// Origin
 	_, err := js.AddStream(&nats.StreamConfig{
 		Name:   "TEST",
-		MaxAge: 100 * time.Millisecond,
+		MaxAge: 500 * time.Millisecond,
 	})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
@@ -9145,3 +9144,172 @@ func TestNoRaceFileStoreLargeMsgsAndFirstMatching(t *testing.T) {
 func TestNoRaceWSNoCorruptionWithFrameSizeLimit(t *testing.T) {
 	testWSNoCorruptionWithFrameSizeLimit(t, 50000)
 }
+
+func TestNoRaceJetStreamAPIDispatchQueuePending(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	// Setup the KV bucket and use for making assertions.
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo.*.*"},
+	})
+	require_NoError(t, err)
+
+	// Queue up 500k messages all with different subjects.
+	// We want to make num pending for a consumer expensive, so a large subject
+	// space and wildcards for now does the trick.
+	toks := []string{"foo", "bar", "baz"} // for second token.
+	for i := 1; i <= 500_000; i++ {
+		subj := fmt.Sprintf("foo.%s.%d", toks[rand.Intn(len(toks))], i)
+		_, err := js.PublishAsync(subj, nil, nats.StallWait(time.Second))
+		require_NoError(t, err)
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(20 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	// To back up our pending queue we will create lots of filtered, with wildcards, R1 consumers
+	// from a different server then the one hosting the stream.
+	// ok to share this connection here.
+	sldr := c.streamLeader(globalAccountName, "TEST")
+	for _, s := range c.servers {
+		if s != sldr {
+			nc, js = jsClientConnect(t, s)
+			defer nc.Close()
+			break
+		}
+	}
+
+	ngr, ncons := 100, 10
+	startCh, errCh := make(chan bool), make(chan error, ngr)
+	var wg, swg sync.WaitGroup
+	wg.Add(ngr)
+	swg.Add(ngr)
+
+	// The wildcard in the filter subject is the key.
+	cfg := &nats.ConsumerConfig{FilterSubject: "foo.*.22"}
+	var tt atomic.Int64
+
+	for i := 0; i < ngr; i++ {
+		go func() {
+			defer wg.Done()
+			swg.Done()
+			// Make them all fire at once.
+			<-startCh
+
+			for i := 0; i < ncons; i++ {
+				start := time.Now()
+				if _, err := js.AddConsumer("TEST", cfg); err != nil {
+					errCh <- err
+					t.Logf("Got err creating consumer: %v", err)
+				}
+				elapsed := time.Since(start)
+				tt.Add(int64(elapsed))
+			}
+		}()
+	}
+	swg.Wait()
+	close(startCh)
+	time.Sleep(time.Millisecond)
+	jsz, _ := sldr.Jsz(nil)
+	// This could be 0 legit, so just log, don't fail.
+	if jsz.JetStreamStats.API.Inflight == 0 {
+		t.Log("Expected a non-zero inflight")
+	}
+	wg.Wait()
+
+	if len(errCh) > 0 {
+		t.Fatalf("Expected no errors, got %d", len(errCh))
+	}
+}
+
+func TestNoRaceJetStreamMirrorAndSourceConsumerFailBackoff(t *testing.T) {
+	// Check calculations first.
+	for i := 1; i <= 20; i++ {
+		backoff := calculateRetryBackoff(i)
+		if i < 12 {
+			require_Equal(t, backoff, time.Duration(i)*10*time.Second)
+		} else {
+			require_Equal(t, backoff, retryMaximum)
+		}
+	}
+
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	// Setup the KV bucket and use for making assertions.
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo.*.*"},
+	})
+	require_NoError(t, err)
+	sl := c.streamLeader(globalAccountName, "TEST")
+
+	// Create a mirror.
+	ml := sl
+	// Make sure not on the same server. Should not happened in general but possible.
+	for ml == sl {
+		js.DeleteStream("MIRROR")
+		_, err = js.AddStream(&nats.StreamConfig{
+			Name:   "MIRROR",
+			Mirror: &nats.StreamSource{Name: "TEST"},
+		})
+		require_NoError(t, err)
+		ml = c.streamLeader(globalAccountName, "MIRROR")
+	}
+
+	// Create sub to watch for the consumer create requests.
+	nc, _ = jsClientConnect(t, ml)
+	defer nc.Close()
+	sub, err := nc.SubscribeSync("$JS.API.CONSUMER.CREATE.>")
+	require_NoError(t, err)
+
+	// Kill the server where the source is..
+	sldr := c.streamLeader(globalAccountName, "TEST")
+	sldr.Shutdown()
+
+	// Wait for just greater than 10s. We should only see 1 request during this time.
+	time.Sleep(11 * time.Second)
+	n, _, _ := sub.Pending()
+	require_Equal(t, n, 1)
+
+	// Now make sure that the fails is set properly.
+	mset, err := ml.GlobalAccount().lookupStream("MIRROR")
+	require_NoError(t, err)
+	mset.mu.RLock()
+	fails := mset.mirror.fails
+	mset.mu.RUnlock()
+	require_Equal(t, fails, 1)
+
+	js.DeleteStream("MIRROR")
+	// Clear sub
+	sub.NextMsg(time.Second)
+	// Make sure sources behave similarly.
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:    "SOURCE",
+		Sources: []*nats.StreamSource{{Name: "TEST"}},
+	})
+	require_NoError(t, err)
+
+	// Wait for just greater than 10s. We should only see 1 request during this time.
+	time.Sleep(11 * time.Second)
+	n, _, _ = sub.Pending()
+	require_Equal(t, n, 1)
+
+	mset, err = c.streamLeader(globalAccountName, "SOURCE").GlobalAccount().lookupStream("SOURCE")
+	require_NoError(t, err)
+	mset.mu.RLock()
+	si := mset.sources["TEST > >"]
+	mset.mu.RUnlock()
+	require_True(t, si != nil)
+	require_Equal(t, si.fails, 1)
+}