Repository: nats-io/nats-server
Pull request #4887 (2023-12-15 16:34:29)

Release v2.10.8 cherry picks
Includes the following:

- #4873
- #4880
- #4881
- #4884
- #4885
- #4890
- #4889



Patch:
diff --git a/server/configs/reload/tls_multi_cert_1.conf b/server/configs/reload/tls_multi_cert_1.conf
new file mode 100644
index 00000000..0269e4af
--- /dev/null
+++ b/server/configs/reload/tls_multi_cert_1.conf
@@ -0,0 +1,13 @@
+# Simple TLS config file
+
+listen:   127.0.0.1:-1
+
+tls {
+  certs = [
+    {
+      cert_file: "../test/configs/certs/srva-cert.pem"
+      key_file:  "../test/configs/certs/srva-key.pem"
+    }
+  ]
+  timeout:    2
+}
diff --git a/server/configs/reload/tls_multi_cert_2.conf b/server/configs/reload/tls_multi_cert_2.conf
new file mode 100644
index 00000000..63288aa8
--- /dev/null
+++ b/server/configs/reload/tls_multi_cert_2.conf
@@ -0,0 +1,19 @@
+# Simple TLS config file
+
+listen:   127.0.0.1:-1
+
+tls {
+  certs = [
+    {
+      cert_file: "../test/configs/certs/srva-cert.pem"
+      key_file:  "../test/configs/certs/srva-key.pem"
+    },
+    {
+      cert_file: "../test/configs/certs/srvb-cert.pem"
+      key_file:  "../test/configs/certs/srvb-key.pem"
+    }
+  ]
+  ca_file: "../test/configs/certs/ca.pem"
+  verify:  true
+  timeout: 2
+}
diff --git a/server/configs/reload/tls_multi_cert_3.conf b/server/configs/reload/tls_multi_cert_3.conf
new file mode 100644
index 00000000..281cfc8d
--- /dev/null
+++ b/server/configs/reload/tls_multi_cert_3.conf
@@ -0,0 +1,13 @@
+# Simple TLS config file
+
+listen:   127.0.0.1:-1
+
+tls {
+  certs = [
+    {
+      cert_file: "../test/configs/certs/srvb-cert.pem"
+      key_file:  "../test/configs/certs/srvb-key.pem"
+    }
+  ]
+  timeout: 2
+}
diff --git a/server/consumer.go b/server/consumer.go
index ef0eb7d2..68f55418 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 to see if the candidate subject matches a filter if its present.
+// Check 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,6 +3374,25 @@ 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 573edb29..50361a5e 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -1590,6 +1590,13 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 				// We could reference the underlying buffer, but we could guess wrong if
 				// number of blocks is large and subjects is low, since we would reference buf.
 				subj := string(buf[bi : bi+lsubj])
+				// We had a bug that could cause memory corruption in the PSIM that could have gotten stored to disk.
+				// Only would affect subjects, so do quick check.
+				if !isValidSubject(subj, true) {
+					os.Remove(fn)
+					fs.warn("Stream state corrupt subject detected")
+					return errCorruptState
+				}
 				bi += lsubj
 				psi := &psi{total: readU64(), fblk: uint32(readU64())}
 				if psi.total > 1 {
@@ -2581,8 +2588,7 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 	}
 
 	// Track starting for both block for the sseq and staring block that matches any subject.
-	var seqStart, subjStart int
-
+	var seqStart int
 	// See if we need to figure out starting block per sseq.
 	if sseq > fs.state.FirstSeq {
 		// This should not, but can return -1, so make sure we check to avoid panic below.
@@ -2591,8 +2597,6 @@ 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)
 
@@ -2602,13 +2606,16 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 	}
 
 	// If we are isAll and have no deleted we can do a simpler calculation.
-	if isAll && (fs.state.LastSeq-fs.state.FirstSeq+1) == fs.state.Msgs {
+	if !lastPerSubject && isAll && (fs.state.LastSeq-fs.state.FirstSeq+1) == fs.state.Msgs {
 		if sseq == 0 {
 			return fs.state.Msgs, validThrough
 		}
 		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
@@ -2620,58 +2627,112 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 		return isSubsetMatchTokenized(tts, fts)
 	}
 
-	// If we would need to scan more from the beginning, revert back to calculating directly here.
-	// TODO(dlc) - Redo properly with sublists etc for subject-based filtering.
-	if lastPerSubject || seqStart >= (len(fs.blks)/2) {
-		// If we need to track seen for last per subject.
-		var seen map[string]bool
-		if lastPerSubject {
-			seen = make(map[string]bool)
+	// Handle last by subject a bit differently.
+	// We will scan PSIM since we accurately track the last block we have seen the subject in. This
+	// allows us to only need to load at most one block now.
+	// For the last block, we need to track the subjects that we know are in that block, and track seen
+	// while in the block itself, but complexity there worth it.
+	if lastPerSubject {
+		// If we want all and our start sequence is equal or less than first return number of subjects.
+		if isAll && sseq <= fs.state.FirstSeq {
+			return uint64(len(fs.psim)), validThrough
+		}
+		// If we are here we need to scan. We are going to scan the PSIM looking for lblks that are >= seqStart.
+		// This will build up a list of all subjects from the selected block onward.
+		lbm := make(map[string]bool)
+		mb := fs.blks[seqStart]
+		bi := mb.index
+
+		for subj, psi := range fs.psim {
+			// If the select blk start is greater than entry's last blk skip.
+			if bi > psi.lblk {
+				continue
+			}
+			if isMatch(subj) {
+				total++
+				// We will track the subjects that are an exact match to the last block.
+				// This is needed for last block processing.
+				if psi.lblk == bi {
+					lbm[subj] = true
+				}
+			}
+		}
+		// Now check if we need to inspect the seqStart block.
+		// Grab write lock in case we need to load in msgs.
+		mb.mu.Lock()
+		var shouldExpire bool
+		// We need to walk this block to correct accounting from above.
+		if sseq > mb.first.seq {
+			// Track the ones we add back in case more than one.
+			seen := make(map[string]bool)
+			// We need to discount the total by subjects seen before sseq, but also add them right back in if they are >= sseq for this blk.
+			// This only should be subjects we know have the last blk in this block.
+			if mb.cacheNotLoaded() {
+				mb.loadMsgsWithLock()
+				shouldExpire = true
+			}
+			var smv StoreMsg
+			for seq, lseq := atomic.LoadUint64(&mb.first.seq), atomic.LoadUint64(&mb.last.seq); seq <= lseq; seq++ {
+				sm, _ := mb.cacheLookup(seq, &smv)
+				if sm == nil || sm.subj == _EMPTY_ || !lbm[sm.subj] {
+					continue
+				}
+				if isMatch(sm.subj) {
+					// If less than sseq adjust off of total as long as this subject matched the last block.
+					if seq < sseq {
+						if !seen[sm.subj] {
+							total--
+							seen[sm.subj] = true
+						}
+					} else if seen[sm.subj] {
+						// This is equal or more than sseq, so add back in.
+						total++
+						// Make sure to not process anymore.
+						delete(seen, sm.subj)
+					}
+				}
+			}
 		}
+		// If we loaded the block try to force expire.
+		if shouldExpire {
+			mb.tryForceExpireCacheLocked()
+		}
+		mb.mu.Unlock()
+		return total, validThrough
+	}
 
+	// If we would need to scan more from the beginning, revert back to calculating directly here.
+	// TODO(dlc) - Redo properly with sublists etc for subject-based filtering.
+	if seqStart >= (len(fs.blks) / 2) {
 		for i := seqStart; i < len(fs.blks); i++ {
+			var shouldExpire bool
 			mb := fs.blks[i]
+			// Hold write lock in case we need to load cache.
 			mb.mu.Lock()
 			var t uint64
 			if isAll && sseq <= atomic.LoadUint64(&mb.first.seq) {
-				if lastPerSubject {
-					mb.ensurePerSubjectInfoLoaded()
-					for subj := range mb.fss {
-						if !seen[subj] {
-							total++
-							seen[subj] = true
-						}
-					}
-				} else {
-					total += mb.msgs
-				}
+				total += mb.msgs
 				mb.mu.Unlock()
 				continue
 			}
-
 			// If we are here we need to at least scan the subject fss.
 			// Make sure we have fss loaded.
-			mb.ensurePerSubjectInfoLoaded()
+			if mb.cacheNotLoaded() {
+				mb.loadMsgsWithLock()
+				shouldExpire = true
+			}
 			var havePartial bool
 			for subj, ss := range mb.fss {
-				if !seen[subj] && isMatch(subj) {
-					if lastPerSubject {
-						// Can't have a partials with last by subject.
-						if sseq <= ss.Last {
-							t++
-							seen[subj] = true
-						}
-					} else {
-						if ss.firstNeedsUpdate {
-							mb.recalculateFirstForSubj(subj, ss.First, ss)
-						}
-						if sseq <= ss.First {
-							t += ss.Msgs
-						} else if sseq <= ss.Last {
-							// We matched but its a partial.
-							havePartial = true
-							break
-						}
+				if isMatch(subj) {
+					if ss.firstNeedsUpdate {
+						mb.recalculateFirstForSubj(subj, ss.First, ss)
+					}
+					if sseq <= ss.First {
+						t += ss.Msgs
+					} else if sseq <= ss.Last {
+						// We matched but its a partial.
+						havePartial = true
+						break
 					}
 				}
 			}
@@ -2679,22 +2740,16 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 			if havePartial {
 				// Clear on partial.
 				t = 0
-				// If we load the cache for a linear scan we want to expire that cache upon exit.
-				var shouldExpire bool
-				if mb.cacheNotLoaded() {
-					mb.loadMsgsWithLock()
-					shouldExpire = true
-				}
 				var smv StoreMsg
 				for seq, lseq := sseq, atomic.LoadUint64(&mb.last.seq); seq <= lseq; seq++ {
-					if sm, _ := mb.cacheLookup(seq, &smv); sm != nil && (isAll || isMatch(sm.subj)) {
+					if sm, _ := mb.cacheLookup(seq, &smv); sm != nil && isMatch(sm.subj) {
 						t++
 					}
 				}
-				// If we loaded this block for this operation go ahead and expire it here.
-				if shouldExpire {
-					mb.tryForceExpireCacheLocked()
-				}
+			}
+			// If we loaded this block for this operation go ahead and expire it here.
+			if shouldExpire {
+				mb.tryForceExpireCacheLocked()
 			}
 			mb.mu.Unlock()
 			total += t
@@ -2702,24 +2757,15 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 		return total, validThrough
 	}
 
-	// If we are here its better to calculate totals from psim and adjust downward by scanning less blocks.
+	// If we are here it's better to calculate totals from psim and adjust downward by scanning less blocks.
 	// TODO(dlc) - Eventually when sublist uses generics, make this sublist driven instead.
 	start := uint32(math.MaxUint32)
 	for subj, psi := range fs.psim {
 		if isMatch(subj) {
-			if lastPerSubject {
-				total++
-				// Keep track of start index for this subject.
-				// Use last block in this case.
-				if psi.lblk < start {
-					start = psi.lblk
-				}
-			} else {
-				total += psi.total
-				// Keep track of start index for this subject.
-				if psi.fblk < start {
-					start = psi.fblk
-				}
+			total += psi.total
+			// Keep track of start index for this subject.
+			if psi.fblk < start {
+				start = psi.fblk
 			}
 		}
 	}
@@ -2729,11 +2775,8 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 	}
 
 	// If we are here we need to calculate partials for the first blocks.
-	subjStart = int(start)
-	firstSubjBlk := fs.bim[uint32(subjStart)]
+	firstSubjBlk := fs.bim[start]
 	var firstSubjBlkFound bool
-	var smv StoreMsg
-
 	// Adjust in case not found.
 	if firstSubjBlk == nil {
 		firstSubjBlkFound = true
@@ -2741,62 +2784,53 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 
 	// Track how many we need to adjust against the total.
 	var adjust uint64
-
 	for i := 0; i <= seqStart; i++ {
 		mb := fs.blks[i]
-
 		// We can skip blks if we know they are below the first one that has any subject matches.
 		if !firstSubjBlkFound {
-			if mb == firstSubjBlk {
-				firstSubjBlkFound = true
-			} else {
+			if firstSubjBlkFound = (mb == firstSubjBlk); !firstSubjBlkFound {
 				continue
 			}
 		}
-
 		// We need to scan this block.
 		var shouldExpire bool
 		mb.mu.Lock()
 		// Check if we should include all of this block in adjusting. If so work with metadata.
 		if sseq > atomic.LoadUint64(&mb.last.seq) {
-			if isAll && !lastPerSubject {
+			if isAll {
 				adjust += mb.msgs
 			} else {
 				// We need to adjust for all matches in this block.
-				// We will scan fss state vs messages themselves.
-				// Make sure we have fss loaded.
-				mb.ensurePerSubjectInfoLoaded()
+				// Make sure we have fss loaded. This loads whole block now.
+				if mb.cacheNotLoaded() {
+					mb.loadMsgsWithLock()
+					shouldExpire = true
+				}
 				for subj, ss := range mb.fss {
 					if isMatch(subj) {
-						if lastPerSubject {
-							adjust++
-						} else {
-							adjust += ss.Msgs
-						}
+						adjust += ss.Msgs
 					}
 				}
 			}
 		} else {
 			// This is the last block. We need to scan per message here.
 			if mb.cacheNotLoaded() {
-				if err := mb.loadMsgsWithLock(); err != nil {
-					mb.mu.Unlock()
-					return 0, 0
-				}
+				mb.loadMsgsWithLock()
 				shouldExpire = true
 			}
-
 			var last = atomic.LoadUint64(&mb.last.seq)
 			if sseq < last {
 				last = sseq
 			}
+			// We need to walk all messages in this block
+			var smv StoreMsg
 			for seq := atomic.LoadUint64(&mb.first.seq); seq < last; seq++ {
 				sm, _ := mb.cacheLookup(seq, &smv)
-				if sm == nil {
+				if sm == nil || sm.subj == _EMPTY_ {
 					continue
 				}
 				// Check if it matches our filter.
-				if isMatch(sm.subj) && sm.seq < sseq {
+				if sm.seq < sseq && isMatch(sm.subj) {
 					adjust++
 				}
 			}
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/opts.go b/server/opts.go
index ff46afb0..1b2621a2 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -669,6 +669,13 @@ type TLSConfigOpts struct {
 	CertMatchBy       certstore.MatchByType
 	CertMatch         string
 	OCSPPeerConfig    *certidp.OCSPPeerConfig
+	Certificates      []*TLSCertPairOpt
+}
+
+// TLSCertPairOpt are the paths to a certificate and private key.
+type TLSCertPairOpt struct {
+	CertFile string
+	KeyFile  string
 }
 
 // OCSPConfig represents the options of OCSP stapling options.
@@ -4180,7 +4187,7 @@ func parseTLS(v interface{}, isClientCtx bool) (t *TLSConfigOpts, retErr error)
 	)
 	defer convertPanicToError(&lt, &retErr)
 
-	_, v = unwrapValue(v, &lt)
+	tk, v := unwrapValue(v, &lt)
 	tlsm = v.(map[string]interface{})
 	for mk, mv := range tlsm {
 		tk, mv := unwrapValue(mv, &lt)
@@ -4381,10 +4388,46 @@ func parseTLS(v interface{}, isClientCtx bool) (t *TLSConfigOpts, retErr error)
 			default:
 				return nil, &configErr{tk, fmt.Sprintf("error parsing ocsp peer config: unsupported type %T", v)}
 			}
+		case "certs", "certificates":
+			certs, ok := mv.([]interface{})
+			if !ok {
+				return nil, &configErr{tk, fmt.Sprintf("error parsing certificates config: unsupported type %T", v)}
+			}
+			tc.Certificates = make([]*TLSCertPairOpt, len(certs))
+			for i, v := range certs {
+				tk, vv := unwrapValue(v, &lt)
+				pair, ok := vv.(map[string]interface{})
+				if !ok {
+					return nil, &configErr{tk, fmt.Sprintf("error parsing certificates config: unsupported type %T", vv)}
+				}
+				certPair := &TLSCertPairOpt{}
+				for k, v := range pair {
+					tk, vv = unwrapValue(v, &lt)
+					file, ok := vv.(string)
+					if !ok {
+						return nil, &configErr{tk, fmt.Sprintf("error parsing certificates config: unsupported type %T", vv)}
+					}
+					switch k {
+					case "cert_file":
+						certPair.CertFile = file
+					case "key_file":
+						certPair.KeyFile = file
+					default:
+						return nil, &configErr{tk, fmt.Sprintf("error parsing tls certs config, unknown field %q", k)}
+					}
+				}
+				if certPair.CertFile == _EMPTY_ || certPair.KeyFile == _EMPTY_ {
+					return nil, &configErr{tk, "error parsing certificates config: both 'cert_file' and 'cert_key' options are required"}
+				}
+				tc.Certificates[i] = certPair
+			}
 		default:
-			return nil, &configErr{tk, fmt.Sprintf("error parsing tls config, unknown field [%q]", mk)}
+			return nil, &configErr{tk, fmt.Sprintf("error parsing tls config, unknown field %q", mk)}
 		}
 	}
+	if len(tc.Certificates) > 0 && tc.CertFile != _EMPTY_ {
+		return nil, &configErr{tk, "error parsing tls config, cannot combine 'cert_file' option with 'certs' option"}
+	}
 
 	// If cipher suites were not specified then use the defaults
 	if tc.Ciphers == nil {
@@ -4696,6 +4739,20 @@ func GenTLSConfig(tc *TLSConfigOpts) (*tls.Config, error) {
 		if err != nil {
 			return nil, err
 		}
+	case tc.Certificates != nil:
+		// Multiple certificate support.
+		config.Certificates = make([]tls.Certificate, len(tc.Certificates))
+		for i, certPair := range tc.Certificates {
+			cert, err := tls.LoadX509KeyPair(certPair.CertFile, certPair.KeyFile)
+			if err != nil {
+				return nil, fmt.Errorf("error parsing X509 certificate/key pair %d/%d: %v", i+1, len(tc.Certificates), err)
+			}
+			cert.Leaf, err = x509.ParseCertificate(cert.Certificate[0])
+			if err != nil {
+				return nil, fmt.Errorf("error parsing certificate %d/%d: %v", i+1, len(tc.Certificates), err)
+			}
+			config.Certificates[i] = cert
+		}
 	}
 
 	// Require client certificates as needed
diff --git a/server/stream.go b/server/stream.go
index f4c630d0..7317837b 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,11 +2002,10 @@ 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_ ||
-			// or consumer filter subject is equal to purged subject
-			preq.Subject == o.cfg.FilterSubject ||
-			// or consumer subject is subset of purged subject,
+			// consumer filter subject is equal to purged subject
+			// or consumer filter subject is subset of purged subject,
 			// but not the other way around.
-			subjectIsSubsetMatch(o.cfg.FilterSubject, preq.Subject)
+			o.isEqualOrSubsetMatch(preq.Subject)
 		o.mu.RUnlock()
 		if doPurge {
 			o.purge(fseq, lseq)
@@ -2391,13 +2391,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)
@@ -2405,9 +2422,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()
@@ -2418,6 +2438,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")
 	}
@@ -2449,7 +2472,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()
@@ -2506,22 +2529,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()
@@ -2539,7 +2563,7 @@ func (mset *stream) setupMirrorConsumer() error {
 	})
 	if err != nil {
 		mirror.err = NewJSMirrorConsumerSetupFailedError(err, Unless(err))
-		mset.scheduleSetupMirrorConsumerRetryAsap()
+		mset.scheduleSetupMirrorConsumerRetry()
 		return nil
 	}
 
@@ -2557,26 +2581,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
@@ -2592,7 +2599,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()
@@ -2618,7 +2631,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.
@@ -2734,8 +2766,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
@@ -2745,7 +2779,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)
@@ -2753,16 +2787,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{}
 	}
@@ -2784,6 +2821,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
@@ -2799,7 +2841,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()
@@ -2878,7 +2920,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
 	}
 
@@ -2904,26 +2946,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
@@ -2939,7 +2964,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()
@@ -2954,7 +2985,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
@@ -2964,7 +2995,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
 					}
@@ -3489,7 +3540,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 63876f43..e7d72576 100644
--- a/server/subject_transform.go
+++ b/server/subject_transform.go
@@ -561,3 +561,29 @@ 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 a78ff86c..fdd7cc5f 100644
--- a/server/sublist.go
+++ b/server/sublist.go
@@ -19,6 +19,7 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"unicode/utf8"
 )
 
 // Sublist is a routing mechanism to handle subject distribution and
@@ -1075,9 +1076,22 @@ func IsValidPublishSubject(subject string) bool {
 
 // IsValidSubject returns true if a subject is valid, false otherwise
 func IsValidSubject(subject string) bool {
+	return isValidSubject(subject, false)
+}
+
+func isValidSubject(subject string, checkRunes bool) bool {
 	if subject == _EMPTY_ {
 		return false
 	}
+	if checkRunes {
+		// Since casting to a string will always produce valid UTF-8, we need to look for replacement runes.
+		// This signals something is off or corrupt.
+		for _, r := range subject {
+			if r == utf8.RuneError {
+				return false
+			}
+		}
+	}
 	sfwc := false
 	tokens := strings.Split(subject, tsep)
 	for _, t := range tokens {
@@ -1101,32 +1115,6 @@ 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/config_check_test.go b/server/config_check_test.go
index a07962b7..53299492 100644
--- a/server/config_check_test.go
+++ b/server/config_check_test.go
@@ -298,7 +298,7 @@ func TestConfigCheck(t *testing.T) {
 		  hello = "world"
 		}
 		`,
-			err:       errors.New(`error parsing tls config, unknown field ["hello"]`),
+			err:       errors.New(`error parsing tls config, unknown field "hello"`),
 			errorLine: 3,
 			errorPos:  5,
 		},
@@ -311,7 +311,7 @@ func TestConfigCheck(t *testing.T) {
 		  }
 		}
 		`,
-			err:       errors.New(`error parsing tls config, unknown field ["foo"]`),
+			err:       errors.New(`error parsing tls config, unknown field "foo"`),
 			errorLine: 4,
 			errorPos:  7,
 		},
@@ -326,7 +326,7 @@ func TestConfigCheck(t *testing.T) {
 		    preferences = []
 		}
 		`,
-			err:       errors.New(`error parsing tls config, unknown field ["preferences"]`),
+			err:       errors.New(`error parsing tls config, unknown field "preferences"`),
 			errorLine: 7,
 			errorPos:  7,
 		},
@@ -342,7 +342,7 @@ func TestConfigCheck(t *testing.T) {
 		    suites = []
 		}
 		`,
-			err:       errors.New(`error parsing tls config, unknown field ["suites"]`),
+			err:       errors.New(`error parsing tls config, unknown field "suites"`),
 			errorLine: 8,
 			errorPos:  7,
 		},
@@ -1832,6 +1832,93 @@ func TestConfigCheck(t *testing.T) {
 			errorLine: 4,
 			errorPos:  6,
 		},
+		{
+			name: "TLS multiple certs",
+			config: `
+				port: -1
+				tls {
+					certs: [
+					  { cert_file: "configs/certs/server.pem", key_file: "configs/certs/key.pem"},
+					  { cert_file: "configs/certs/cert.new.pem", key_file: "configs/certs/key.new.pem"},
+					]
+				}
+			`,
+			err: nil,
+		},
+		{
+			name: "TLS multiple certs, bad type",
+			config: `
+				port: -1
+				tls {
+					certs: [
+					  { cert_file: "configs/certs/server.pem", key_file: 123 },
+					  { cert_file: "configs/certs/cert.new.pem", key_file: "configs/certs/key.new.pem"},
+					]
+				}
+			`,
+			err:       fmt.Errorf("error parsing certificates config: unsupported type int64"),
+			errorLine: 5,
+			errorPos:  49,
+		},
+		{
+			name: "TLS multiple certs, missing key_file",
+			config: `
+				port: -1
+				tls {
+					certs: [
+					  { cert_file: "configs/certs/server.pem" }
+					  { cert_file: "configs/certs/cert.new.pem", key_file: "configs/certs/key.new.pem"}
+					]
+				}
+			`,
+			err:       fmt.Errorf("error parsing certificates config: both 'cert_file' and 'cert_key' options are required"),
+			errorLine: 5,
+			errorPos:  10,
+		},
+		{
+			name: "TLS multiple certs and single cert options at the same time",
+			config: `
+				port: -1
+				tls {
+					cert_file: "configs/certs/server.pem"
+					key_file: "configs/certs/key.pem"
+					certs: [
+					  { cert_file: "configs/certs/server.pem", key_file: "configs/certs/key.pem"},
+					  { cert_file: "configs/certs/cert.new.pem", key_file: "configs/certs/key.new.pem"},
+					]
+				}
+			`,
+			err:       fmt.Errorf("error parsing tls config, cannot combine 'cert_file' option with 'certs' option"),
+			errorLine: 3,
+			errorPos:  5,
+		},
+		{
+			name: "TLS multiple certs used but not configured, but cert_file configured",
+			config: `
+				port: -1
+				tls {
+					cert_file: "configs/certs/server.pem"
+					key_file: "configs/certs/key.pem"
+					certs: []
+				}
+			`,
+			err: nil,
+		},
+		{
+			name: "TLS multiple certs, missing bad path",
+			config: `
+				port: -1
+				tls {
+					certs: [
+					  { cert_file: "configs/certs/cert.new.pem", key_file: "configs/certs/key.new.pem"}
+					  { cert_file: "configs/certs/server.pem", key_file: "configs/certs/key.new.pom" }
+					]
+				}
+			`,
+			err:       fmt.Errorf("error parsing X509 certificate/key pair 2/2: open configs/certs/key.new.pom: no such file or directory"),
+			errorLine: 3,
+			errorPos:  5,
+		},
 	}
 
 	checkConfig := func(config string) error {
diff --git a/server/filestore_test.go b/server/filestore_test.go
index ceea7e9d..872eda3f 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -6184,6 +6184,104 @@ func TestFileStoreSubjectCorruption(t *testing.T) {
 	}
 }
 
+// Since 2.10 we no longer have fss, and the approach for calculating NumPending would branch
+// based on the old fss metadata being present. This meant that calculating NumPending in >= 2.10.x
+// would load all blocks to complete. This test makes sure we do not do that anymore.
+func TestFileStoreNumPendingLastBySubject(t *testing.T) {
+	sd, blkSize := t.TempDir(), uint64(1024)
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: sd, BlockSize: blkSize},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	numSubjects := 20
+	msg := bytes.Repeat([]byte("ABC"), 25)
+	for i := 1; i <= 1000; i++ {
+		subj := fmt.Sprintf("foo.%d.%d", rand.Intn(numSubjects)+1, i)
+		fs.StoreMsg(subj, nil, msg)
+	}
+	// Each block has ~8 msgs.
+	require_True(t, fs.numMsgBlocks() > 100)
+
+	calcCacheLoads := func() (cloads uint64) {
+		fs.mu.RLock()
+		defer fs.mu.RUnlock()
+		for _, mb := range fs.blks {
+			mb.mu.RLock()
+			cloads += mb.cloads
+			mb.mu.RUnlock()
+		}
+		return cloads
+	}
+
+	total, _ := fs.NumPending(0, "foo.*.*", true)
+	require_Equal(t, total, 1000)
+	// Make sure no blocks were loaded to calculate this as a new consumer.
+	require_Equal(t, calcCacheLoads(), 0)
+
+	checkResult := func(sseq, np uint64, filter string) {
+		t.Helper()
+		var checkTotal uint64
+		var smv StoreMsg
+		for seq := sseq; seq <= 1000; seq++ {
+			sm, err := fs.LoadMsg(seq, &smv)
+			require_NoError(t, err)
+			if subjectIsSubsetMatch(sm.subj, filter) {
+				checkTotal++
+			}
+		}
+		require_Equal(t, np, checkTotal)
+	}
+
+	// Make sure partials work properly.
+	for _, filter := range []string{"foo.10.*", "*.22.*", "*.*.222", "foo.5.999", "*.2.*"} {
+		sseq := uint64(rand.Intn(250) + 200) // Between 200-450
+		total, _ = fs.NumPending(sseq, filter, true)
+		checkResult(sseq, total, filter)
+	}
+}
+
+// We had a bug that could cause internal memory corruption of the psim keys in memory
+// which could have been written to disk via index.db.
+func TestFileStoreCorruptPSIMOnDisk(t *testing.T) {
+	sd := t.TempDir()
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: sd},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	fs.StoreMsg("foo.bar", nil, []byte("ABC"))
+	fs.StoreMsg("foo.baz", nil, []byte("XYZ"))
+
+	// Force bad subject.
+	fs.mu.Lock()
+	psi := fs.psim["foo.bar"]
+	bad := make([]byte, 7)
+	crand.Read(bad)
+	fs.psim[string(bad)] = psi
+	delete(fs.psim, "foo.bar")
+	fs.dirty++
+	fs.mu.Unlock()
+
+	// Restart
+	fs.Stop()
+	fs, err = newFileStore(
+		FileStoreConfig{StoreDir: sd},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	sm, err := fs.LoadLastMsg("foo.bar", nil)
+	require_NoError(t, err)
+	require_True(t, bytes.Equal(sm.msg, []byte("ABC")))
+
+	sm, err = fs.LoadLastMsg("foo.baz", nil)
+	require_NoError(t, err)
+	require_True(t, bytes.Equal(sm.msg, []byte("XYZ")))
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Benchmarks
 ///////////////////////////////////////////////////////////////////////////
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 ed3de3ad..ec8fdf12 100644
--- a/server/jetstream_consumer_test.go
+++ b/server/jetstream_consumer_test.go
@@ -871,6 +871,36 @@ 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/jetstream_test.go b/server/jetstream_test.go
index 8cca50e1..565a0386 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -21964,6 +21964,7 @@ func TestJetStreamKVReductionInHistory(t *testing.T) {
 	require_NoError(t, err)
 
 	checkAllKeys := func() {
+		t.Helper()
 		// Make sure we can retrieve all of the keys.
 		keys, err := kv.Keys()
 		require_NoError(t, 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)
+}
diff --git a/server/reload_test.go b/server/reload_test.go
index 4fe67a55..5be29aa7 100644
--- a/server/reload_test.go
+++ b/server/reload_test.go
@@ -492,6 +492,102 @@ func TestConfigReloadDisableTLS(t *testing.T) {
 	nc.Close()
 }
 
+func TestConfigReloadRotateTLSMultiCert(t *testing.T) {
+	server, opts, config := runReloadServerWithConfig(t, "./configs/reload/tls_multi_cert_1.conf")
+	defer server.Shutdown()
+
+	// Ensure we can connect as a sanity check.
+	addr := fmt.Sprintf("nats://%s:%d", opts.Host, server.Addr().(*net.TCPAddr).Port)
+
+	rawCerts := make(chan []byte, 3)
+	nc, err := nats.Connect(addr, nats.Secure(&tls.Config{
+		VerifyConnection: func(s tls.ConnectionState) error {
+			rawCerts <- s.PeerCertificates[0].Raw
+			return nil
+		},
+		InsecureSkipVerify: true,
+	}))
+	if err != nil {
+		t.Fatalf("Error creating client: %v", err)
+	}
+	defer nc.Close()
+	sub, err := nc.SubscribeSync("foo")
+	if err != nil {
+		t.Fatalf("Error subscribing: %v", err)
+	}
+	defer sub.Unsubscribe()
+
+	// Rotate cert and enable client verification.
+	changeCurrentConfigContent(t, config, "./configs/reload/tls_multi_cert_2.conf")
+	if err := server.Reload(); err != nil {
+		t.Fatalf("Error reloading config: %v", err)
+	}
+
+	// Ensure connecting fails.
+	if _, err := nats.Connect(addr, nats.Secure(&tls.Config{InsecureSkipVerify: true})); err == nil {
+		t.Fatal("Expected connect to fail")
+	}
+
+	// Ensure connecting succeeds when client presents cert.
+	cert := nats.ClientCert("../test/configs/certs/client-cert.pem", "../test/configs/certs/client-key.pem")
+	conn, err := nats.Connect(addr, cert, nats.RootCAs("../test/configs/certs/ca.pem"), nats.Secure(&tls.Config{
+		VerifyConnection: func(s tls.ConnectionState) error {
+			rawCerts <- s.PeerCertificates[0].Raw
+			return nil
+		},
+	}))
+	if err != nil {
+		t.Fatalf("Error creating client: %v", err)
+	}
+	conn.Close()
+
+	// Ensure the original connection can still publish/receive.
+	if err := nc.Publish("foo", []byte("hello")); err != nil {
+		t.Fatalf("Error publishing: %v", err)
+	}
+	nc.Flush()
+	msg, err := sub.NextMsg(2 * time.Second)
+	if err != nil {
+		t.Fatalf("Error receiving msg: %v", err)
+	}
+	if string(msg.Data) != "hello" {
+		t.Fatalf("Msg is incorrect.\nexpected: %+v\ngot: %+v", []byte("hello"), msg.Data)
+	}
+
+	// Rotate cert and disable client verification.
+	changeCurrentConfigContent(t, config, "./configs/reload/tls_multi_cert_3.conf")
+	if err := server.Reload(); err != nil {
+		t.Fatalf("Error reloading config: %v", err)
+	}
+
+	nc, err = nats.Connect(addr, nats.Secure(&tls.Config{
+		VerifyConnection: func(s tls.ConnectionState) error {
+			rawCerts <- s.PeerCertificates[0].Raw
+			return nil
+		},
+		InsecureSkipVerify: true,
+	}))
+	if err != nil {
+		t.Fatalf("Error creating client: %v", err)
+	}
+	defer nc.Close()
+	sub, err = nc.SubscribeSync("foo")
+	if err != nil {
+		t.Fatalf("Error subscribing: %v", err)
+	}
+	defer sub.Unsubscribe()
+
+	certA := <-rawCerts
+	certB := <-rawCerts
+	certC := <-rawCerts
+	if !bytes.Equal(certA, certB) {
+		t.Error("Expected the same cert")
+	}
+	if bytes.Equal(certB, certC) {
+		t.Error("Expected a different cert")
+	}
+}
+
 // Ensure Reload supports single user authentication config changes. Test this
 // by starting a server with authentication enabled, connect to it to verify,
 // reload config using a different username/password, ensure reconnect fails,