Repository: nats-io/nats-server
Pull request #5823 (2024-08-23 15:11:52)

Cherry picks for v2.10.19 RC5
Includes:

  - https://github.com/nats-io/nats-server/pull/5782
  - https://github.com/nats-io/nats-server/pull/5785
  - https://github.com/nats-io/nats-server/pull/5757
  - https://github.com/nats-io/nats-server/pull/5789
  - https://github.com/nats-io/nats-server/pull/5790
  - https://github.com/nats-io/nats-server/pull/5793
  - https://github.com/nats-io/nats-server/pull/5794
  - https://github.com/nats-io/nats-server/pull/5809
  - https://github.com/nats-io/nats-server/pull/5812
  - https://github.com/nats-io/nats-server/pull/5816
  - https://github.com/nats-io/nats-server/pull/5819
  - https://github.com/nats-io/nats-server/pull/5818




Patch:
diff --git a/go.mod b/go.mod
index 0e6c6627..7c235cf2 100644
--- a/go.mod
+++ b/go.mod
@@ -10,7 +10,7 @@ require (
 	github.com/nats-io/nkeys v0.4.7
 	github.com/nats-io/nuid v1.0.1
 	go.uber.org/automaxprocs v1.5.3
-	golang.org/x/crypto v0.25.0
-	golang.org/x/sys v0.23.0
+	golang.org/x/crypto v0.26.0
+	golang.org/x/sys v0.24.0
 	golang.org/x/time v0.6.0
 )
diff --git a/go.sum b/go.sum
index c7e25f5a..e678c009 100644
--- a/go.sum
+++ b/go.sum
@@ -22,11 +22,15 @@ go.uber.org/automaxprocs v1.5.3 h1:kWazyxZUrS3Gs4qUpbwo5kEIMGe/DAvi5Z4tl2NW4j8=
 go.uber.org/automaxprocs v1.5.3/go.mod h1:eRbA25aqJrxAbsLO0xy5jVwPt7FQnRgjW+efnwa1WM0=
 golang.org/x/crypto v0.25.0 h1:ypSNr+bnYL2YhwoMt2zPxHFmbAN1KZs/njMG3hxUp30=
 golang.org/x/crypto v0.25.0/go.mod h1:T+wALwcMOSE0kXgUAnPAHqTLW+XHgcELELW8VaDgm/M=
+golang.org/x/crypto v0.26.0 h1:RrRspgV4mU+YwB4FYnuBoKsUapNIL5cohGAmSH3azsw=
+golang.org/x/crypto v0.26.0/go.mod h1:GY7jblb9wI+FOo5y8/S2oY4zWP07AkOJ4+jxCqdqn54=
 golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.22.0 h1:RI27ohtqKCnwULzJLqkv897zojh5/DwS/ENaMzUOaWI=
 golang.org/x/sys v0.22.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.23.0 h1:YfKFowiIMvtgl1UERQoTPPToxltDeZfbj4H7dVUCwmM=
 golang.org/x/sys v0.23.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.24.0 h1:Twjiwq9dn6R1fQcyiK+wQyHWfaz/BJB+YIpzU/Cv3Xg=
+golang.org/x/sys v0.24.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
 golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/time v0.6.0 h1:eTDhh4ZXt5Qf0augr54TN6suAUudPcawVZeIAPU7D4U=
diff --git a/internal/ocsp/ocsp.go b/internal/ocsp/ocsp.go
index 4fa51bd3..f317c1f1 100644
--- a/internal/ocsp/ocsp.go
+++ b/internal/ocsp/ocsp.go
@@ -178,8 +178,9 @@ func NewOCSPResponderBase(t *testing.T, issuerCertPEM, respCertPEM, respKeyPEM s
 	})
 
 	srv := &http.Server{
-		Addr:    addr,
-		Handler: mux,
+		Addr:        addr,
+		Handler:     mux,
+		ReadTimeout: time.Second * 5,
 	}
 	go srv.ListenAndServe()
 	time.Sleep(1 * time.Second)
diff --git a/server/accounts.go b/server/accounts.go
index 5e2609c1..f9897034 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -15,6 +15,7 @@ package server
 
 import (
 	"bytes"
+	"cmp"
 	"encoding/hex"
 	"errors"
 	"fmt"
@@ -25,7 +26,7 @@ import (
 	"net/http"
 	"net/textproto"
 	"reflect"
-	"sort"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -349,10 +350,8 @@ func (a *Account) updateRemoteServer(m *AccountNumConns) []*client {
 	// conservative and bit harsh here. Clients will reconnect if we over compensate.
 	var clients []*client
 	if mtce {
-		clients := a.getClientsLocked()
-		sort.Slice(clients, func(i, j int) bool {
-			return clients[i].start.After(clients[j].start)
-		})
+		clients = a.getClientsLocked()
+		slices.SortFunc(clients, func(i, j *client) int { return -i.start.Compare(j.start) }) // reserve
 		over := (len(a.clients) - int(a.sysclients) + int(a.nrclients)) - int(a.mconns)
 		if over < len(clients) {
 			clients = clients[:over]
@@ -669,7 +668,7 @@ func (a *Account) AddWeightedMappings(src string, dests ...*MapDest) error {
 			}
 			dests = append(dests, &destination{tr, aw})
 		}
-		sort.Slice(dests, func(i, j int) bool { return dests[i].weight < dests[j].weight })
+		slices.SortFunc(dests, func(i, j *destination) int { return cmp.Compare(i.weight, j.weight) })
 
 		var lw uint8
 		for _, d := range dests {
@@ -3567,9 +3566,7 @@ func (s *Server) updateAccountClaimsWithRefresh(a *Account, ac *jwt.AccountClaim
 
 	// Sort if we are over the limit.
 	if a.MaxTotalConnectionsReached() {
-		sort.Slice(clients, func(i, j int) bool {
-			return clients[i].start.After(clients[j].start)
-		})
+		slices.SortFunc(clients, func(i, j *client) int { return -i.start.Compare(j.start) }) // sort in reverse order
 	}
 
 	// If JetStream is enabled for this server we will call into configJetStream for the account
diff --git a/server/auth.go b/server/auth.go
index bdaf7335..0a9564f4 100644
--- a/server/auth.go
+++ b/server/auth.go
@@ -857,33 +857,6 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (au
 	// If we have a jwt and a userClaim, make sure we have the Account, etc associated.
 	// We need to look up the account. This will use an account resolver if one is present.
 	if juc != nil {
-		allowedConnTypes, err := convertAllowedConnectionTypes(juc.AllowedConnectionTypes)
-		if err != nil {
-			// We got an error, which means some connection types were unknown. As long as
-			// a valid one is returned, we proceed with auth. If not, we have to reject.
-			// In other words, suppose that JWT allows "WEBSOCKET" in the array. No error
-			// is returned and allowedConnTypes will contain "WEBSOCKET" only.
-			// Client will be rejected if not a websocket client, or proceed with rest of
-			// auth if it is.
-			// Now suppose JWT allows "WEBSOCKET, MQTT" and say MQTT is not known by this
-			// server. In this case, allowedConnTypes would contain "WEBSOCKET" and we
-			// would get `err` indicating that "MQTT" is an unknown connection type.
-			// If a websocket client connects, it should still be allowed, since after all
-			// the admin wanted to allow websocket and mqtt connection types.
-			// However, say that the JWT only allows "MQTT" (and again suppose this server
-			// does not know about MQTT connection type), then since the allowedConnTypes
-			// map would be empty (no valid types found), and since empty means allow-all,
-			// then we should reject because the intent was to allow connections for this
-			// user only as an MQTT client.
-			c.Debugf("%v", err)
-			if len(allowedConnTypes) == 0 {
-				return false
-			}
-		}
-		if !c.connectionTypeAllowed(allowedConnTypes) {
-			c.Debugf("Connection type not allowed")
-			return false
-		}
 		issuer := juc.Issuer
 		if juc.IssuerAccount != _EMPTY_ {
 			issuer = juc.IssuerAccount
@@ -926,6 +899,36 @@ func (s *Server) processClientOrLeafAuthentication(c *client, opts *Options) (au
 			c.Debugf("Account does not allow bearer tokens")
 			return false
 		}
+		// We check the allowed connection types, but only after processing
+		// of scoped signer (so that it updates `juc` with what is defined
+		// in the account.
+		allowedConnTypes, err := convertAllowedConnectionTypes(juc.AllowedConnectionTypes)
+		if err != nil {
+			// We got an error, which means some connection types were unknown. As long as
+			// a valid one is returned, we proceed with auth. If not, we have to reject.
+			// In other words, suppose that JWT allows "WEBSOCKET" in the array. No error
+			// is returned and allowedConnTypes will contain "WEBSOCKET" only.
+			// Client will be rejected if not a websocket client, or proceed with rest of
+			// auth if it is.
+			// Now suppose JWT allows "WEBSOCKET, MQTT" and say MQTT is not known by this
+			// server. In this case, allowedConnTypes would contain "WEBSOCKET" and we
+			// would get `err` indicating that "MQTT" is an unknown connection type.
+			// If a websocket client connects, it should still be allowed, since after all
+			// the admin wanted to allow websocket and mqtt connection types.
+			// However, say that the JWT only allows "MQTT" (and again suppose this server
+			// does not know about MQTT connection type), then since the allowedConnTypes
+			// map would be empty (no valid types found), and since empty means allow-all,
+			// then we should reject because the intent was to allow connections for this
+			// user only as an MQTT client.
+			c.Debugf("%v", err)
+			if len(allowedConnTypes) == 0 {
+				return false
+			}
+		}
+		if !c.connectionTypeAllowed(allowedConnTypes) {
+			c.Debugf("Connection type not allowed")
+			return false
+		}
 		// skip validation of nonce when presented with a bearer token
 		// FIXME: if BearerToken is only for WSS, need check for server with that port enabled
 		if !juc.BearerToken {
diff --git a/server/avl/seqset.go b/server/avl/seqset.go
index 8d404502..0b6aca67 100644
--- a/server/avl/seqset.go
+++ b/server/avl/seqset.go
@@ -14,10 +14,11 @@
 package avl
 
 import (
+	"cmp"
 	"encoding/binary"
 	"errors"
 	"math/bits"
-	"sort"
+	"slices"
 )
 
 // SequenceSet is a memory and encoding optimized set for storing unsigned ints.
@@ -209,7 +210,7 @@ func Union(ssa ...*SequenceSet) *SequenceSet {
 		return nil
 	}
 	// Sort so we can clone largest.
-	sort.Slice(ssa, func(i, j int) bool { return ssa[i].Size() > ssa[j].Size() })
+	slices.SortFunc(ssa, func(i, j *SequenceSet) int { return -cmp.Compare(i.Size(), j.Size()) }) // reverse order
 	ss := ssa[0].Clone()
 
 	// Insert the rest through range call.
diff --git a/server/client.go b/server/client.go
index b6894c09..19886e70 100644
--- a/server/client.go
+++ b/server/client.go
@@ -5857,6 +5857,14 @@ func (c *client) Warnf(format string, v ...any) {
 	c.srv.Warnf(format, v...)
 }
 
+func (c *client) rateLimitFormatWarnf(format string, v ...any) {
+	if _, loaded := c.srv.rateLimitLogging.LoadOrStore(format, time.Now()); loaded {
+		return
+	}
+	statement := fmt.Sprintf(format, v...)
+	c.Warnf("%s", statement)
+}
+
 func (c *client) RateLimitWarnf(format string, v ...any) {
 	// Do the check before adding the client info to the format...
 	statement := fmt.Sprintf(format, v...)
diff --git a/server/consumer.go b/server/consumer.go
index f8a9b2e7..38af7da9 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -21,7 +21,7 @@ import (
 	"fmt"
 	"math/rand"
 	"reflect"
-	"sort"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -712,7 +712,6 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 
 	mset.mu.RLock()
 	s, jsa, cfg, acc := mset.srv, mset.jsa, mset.cfg, mset.acc
-	retention := cfg.Retention
 	mset.mu.RUnlock()
 
 	// If we do not have the consumer currently assigned to us in cluster mode we will proceed but warn.
@@ -735,7 +734,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	// Make sure we have sane defaults. Do so with the JS lock, otherwise a
 	// badly timed meta snapshot can result in a race condition.
 	mset.js.mu.Lock()
-	setConsumerConfigDefaults(config, &mset.cfg, srvLim, selectedLimits)
+	setConsumerConfigDefaults(config, &cfg, srvLim, selectedLimits)
 	mset.js.mu.Unlock()
 
 	if err := checkConsumerCfg(config, srvLim, &cfg, acc, selectedLimits, isRecovering); err != nil {
@@ -781,7 +780,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 				return nil, NewJSConsumerAlreadyExistsError()
 			}
 			// Check for overlapping subjects if we are a workqueue
-			if mset.cfg.Retention == WorkQueuePolicy {
+			if cfg.Retention == WorkQueuePolicy {
 				subjects := gatherSubjectFilters(config.FilterSubject, config.FilterSubjects)
 				if !mset.partitionUnique(cName, subjects) {
 					return nil, NewJSConsumerWQConsumerNotUniqueError()
@@ -803,7 +802,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	// but if not we use the value from account limits, if account limits is more restrictive
 	// than stream config we prefer the account limits to handle cases where account limits are
 	// updated during the lifecycle of the stream
-	maxc := mset.cfg.MaxConsumers
+	maxc := cfg.MaxConsumers
 	if maxc <= 0 || (selectedLimits.MaxConsumers > 0 && selectedLimits.MaxConsumers < maxc) {
 		maxc = selectedLimits.MaxConsumers
 	}
@@ -813,7 +812,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	}
 
 	// Check on stream type conflicts with WorkQueues.
-	if mset.cfg.Retention == WorkQueuePolicy && !config.Direct {
+	if cfg.Retention == WorkQueuePolicy && !config.Direct {
 		// Force explicit acks here.
 		if config.AckPolicy != AckExplicit {
 			mset.mu.Unlock()
@@ -871,7 +870,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 		sfreq:     int32(sampleFreq),
 		maxdc:     uint64(config.MaxDeliver),
 		maxp:      config.MaxAckPending,
-		retention: retention,
+		retention: cfg.Retention,
 		created:   time.Now().UTC(),
 	}
 
@@ -904,17 +903,17 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 		}
 	}
 	// Create ackMsgs queue now that we have a consumer name
-	o.ackMsgs = newIPQueue[*jsAckMsg](s, fmt.Sprintf("[ACC:%s] consumer '%s' on stream '%s' ackMsgs", accName, o.name, mset.cfg.Name))
+	o.ackMsgs = newIPQueue[*jsAckMsg](s, fmt.Sprintf("[ACC:%s] consumer '%s' on stream '%s' ackMsgs", accName, o.name, cfg.Name))
 
 	// Create our request waiting queue.
 	if o.isPullMode() {
 		o.waiting = newWaitQueue(config.MaxWaiting)
 		// Create our internal queue for next msg requests.
-		o.nextMsgReqs = newIPQueue[*nextMsgReq](s, fmt.Sprintf("[ACC:%s] consumer '%s' on stream '%s' pull requests", accName, o.name, mset.cfg.Name))
+		o.nextMsgReqs = newIPQueue[*nextMsgReq](s, fmt.Sprintf("[ACC:%s] consumer '%s' on stream '%s' pull requests", accName, o.name, cfg.Name))
 	}
 
 	// already under lock, mset.Name() would deadlock
-	o.stream = mset.cfg.Name
+	o.stream = cfg.Name
 	o.ackEventT = JSMetricConsumerAckPre + "." + o.stream + "." + o.name
 	o.nakEventT = JSAdvisoryConsumerMsgNakPre + "." + o.stream + "." + o.name
 	o.deliveryExcEventT = JSAdvisoryConsumerMaxDeliveryExceedPre + "." + o.stream + "." + o.name
@@ -999,7 +998,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	// that to scanf them back in.
 	// Escape '%' in consumer and stream names, as `pre` is used as a template later
 	// in consumer.ackReply(), resulting in erroneous formatting of the ack subject.
-	mn := strings.ReplaceAll(mset.cfg.Name, "%", "%%")
+	mn := strings.ReplaceAll(cfg.Name, "%", "%%")
 	pre := fmt.Sprintf(jsAckT, mn, strings.ReplaceAll(o.name, "%", "%%"))
 	o.ackReplyT = fmt.Sprintf("%s.%%d.%%d.%%d.%%d.%%d", pre)
 	o.ackSubj = fmt.Sprintf("%s.*.*.*.*.*", pre)
@@ -1011,7 +1010,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	if o.isPushMode() {
 		// Check if we are running only 1 replica and that the delivery subject has interest.
 		// Check in place here for interest. Will setup properly in setLeader.
-		if config.replicas(&mset.cfg) == 1 {
+		if config.replicas(&cfg) == 1 {
 			interest := o.acc.sl.HasInterest(o.cfg.DeliverSubject)
 			if !o.hasDeliveryInterest(interest) {
 				// Let the interest come to us eventually, but setup delete timer.
@@ -1193,7 +1192,7 @@ func (o *consumer) setLeader(isLeader bool) {
 		}
 
 		mset.mu.RLock()
-		s, jsa, stream, lseq := mset.srv, mset.jsa, mset.cfg.Name, mset.lseq
+		s, jsa, stream, lseq := mset.srv, mset.jsa, mset.getCfgName(), mset.lseq
 		mset.mu.RUnlock()
 
 		o.mu.Lock()
@@ -1697,7 +1696,7 @@ func (o *consumer) forceExpirePending() {
 		}
 	}
 	if len(expired) > 0 {
-		sort.Slice(expired, func(i, j int) bool { return expired[i] < expired[j] })
+		slices.Sort(expired)
 		o.addToRedeliverQueue(expired...)
 		// Now we should update the timestamp here since we are redelivering.
 		// We will use an incrementing time to preserve order for any other redelivery.
@@ -1745,6 +1744,8 @@ func (o *consumer) setRateLimit(bps uint64) {
 
 	// Burst should be set to maximum msg size for this account, etc.
 	var burst int
+	// We don't need to get cfgMu's rlock here since this function
+	// is already invoked under mset.mu.RLock(), which superseeds cfgMu.
 	if mset.cfg.MaxMsgSize > 0 {
 		burst = int(mset.cfg.MaxMsgSize)
 	} else if mset.jsa.account.limits.mpay > 0 {
@@ -2465,7 +2466,7 @@ func (o *consumer) checkRedelivered(slseq uint64) {
 	if shouldUpdateState {
 		if err := o.writeStoreStateUnlocked(); err != nil && o.srv != nil && o.mset != nil && !o.closed {
 			s, acc, mset, name := o.srv, o.acc, o.mset, o.name
-			s.Warnf("Consumer '%s > %s > %s' error on write store state from check redelivered: %v", acc, mset.cfg.Name, name, err)
+			s.Warnf("Consumer '%s > %s > %s' error on write store state from check redelivered: %v", acc, mset.getCfgName(), name, err)
 		}
 	}
 }
@@ -2883,16 +2884,21 @@ func (o *consumer) isFiltered() bool {
 		return true
 	}
 
+	// Protect access to mset.cfg with the cfgMu mutex.
+	mset.cfgMu.RLock()
+	msetSubjects := mset.cfg.Subjects
+	mset.cfgMu.RUnlock()
+
 	// `isFiltered` need to be performant, so we do
 	// as any checks as possible to avoid unnecessary work.
 	// Here we avoid iteration over slices if there is only one subject in stream
 	// and one filter for the consumer.
-	if len(mset.cfg.Subjects) == 1 && len(o.subjf) == 1 {
-		return mset.cfg.Subjects[0] != o.subjf[0].subject
+	if len(msetSubjects) == 1 && len(o.subjf) == 1 {
+		return msetSubjects[0] != o.subjf[0].subject
 	}
 
 	// if the list is not equal length, we can return early, as this is filtered.
-	if len(mset.cfg.Subjects) != len(o.subjf) {
+	if len(msetSubjects) != len(o.subjf) {
 		return true
 	}
 
@@ -2904,7 +2910,7 @@ func (o *consumer) isFiltered() bool {
 	for _, val := range o.subjf {
 		cfilters[val.subject] = struct{}{}
 	}
-	for _, val := range mset.cfg.Subjects {
+	for _, val := range msetSubjects {
 		if _, ok := cfilters[val]; !ok {
 			return true
 		}
@@ -3941,10 +3947,10 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 	inch := o.inch
 	o.mu.Unlock()
 
-	// Grab the stream's retention policy
-	mset.mu.RLock()
-	rp := mset.cfg.Retention
-	mset.mu.RUnlock()
+	// Grab the stream's retention policy and name
+	mset.cfgMu.RLock()
+	stream, rp := mset.cfg.Name, mset.cfg.Retention
+	mset.cfgMu.RUnlock()
 
 	var err error
 
@@ -4000,12 +4006,12 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 				goto waitForMsgs
 			} else if err == errPartialCache {
 				s.Warnf("Unexpected partial cache error looking up message for consumer '%s > %s > %s'",
-					o.mset.acc, o.mset.cfg.Name, o.cfg.Name)
+					o.mset.acc, stream, o.cfg.Name)
 				goto waitForMsgs
 
 			} else {
 				s.Errorf("Received an error looking up message for consumer '%s > %s > %s': %v",
-					o.mset.acc, o.mset.cfg.Name, o.cfg.Name, err)
+					o.mset.acc, stream, o.cfg.Name, err)
 				goto waitForMsgs
 			}
 		}
@@ -4656,7 +4662,7 @@ func (o *consumer) checkPending() {
 
 	if len(expired) > 0 {
 		// We need to sort.
-		sort.Slice(expired, func(i, j int) bool { return expired[i] < expired[j] })
+		slices.Sort(expired)
 		o.addToRedeliverQueue(expired...)
 		// Now we should update the timestamp here since we are redelivering.
 		// We will use an incrementing time to preserve order for any other redelivery.
@@ -4690,7 +4696,7 @@ func (o *consumer) checkPending() {
 	if shouldUpdateState {
 		if err := o.writeStoreStateUnlocked(); err != nil && o.srv != nil && o.mset != nil && !o.closed {
 			s, acc, mset, name := o.srv, o.acc, o.mset, o.name
-			s.Warnf("Consumer '%s > %s > %s' error on write store state from check pending: %v", acc, mset.cfg.Name, name, err)
+			s.Warnf("Consumer '%s > %s > %s' error on write store state from check pending: %v", acc, mset.getCfgName(), name, err)
 		}
 	}
 }
@@ -4811,7 +4817,10 @@ func (o *consumer) selectStartingSeqNo() {
 			} else if o.cfg.DeliverPolicy == DeliverLastPerSubject {
 				// If our parent stream is set to max msgs per subject of 1 this is just
 				// a normal consumer at this point. We can avoid any heavy lifting.
-				if o.mset.cfg.MaxMsgsPer == 1 {
+				o.mset.cfgMu.RLock()
+				mmp := o.mset.cfg.MaxMsgsPer
+				o.mset.cfgMu.RUnlock()
+				if mmp == 1 {
 					o.sseq = state.FirstSeq
 				} else {
 					// A threshold for when we switch from get last msg to subjects state.
@@ -4841,9 +4850,7 @@ func (o *consumer) selectStartingSeqNo() {
 					}
 					// Sort the skip list if needed.
 					if len(lss.seqs) > 1 {
-						sort.Slice(lss.seqs, func(i, j int) bool {
-							return lss.seqs[j] > lss.seqs[i]
-						})
+						slices.Sort(lss.seqs)
 					}
 					if len(lss.seqs) == 0 {
 						o.sseq = state.LastSeq
@@ -4882,12 +4889,16 @@ func (o *consumer) selectStartingSeqNo() {
 			o.sseq = o.cfg.OptStartSeq
 		}
 
-		if state.FirstSeq == 0 {
-			o.sseq = 1
-		} else if o.sseq < state.FirstSeq {
-			o.sseq = state.FirstSeq
-		} else if o.sseq > state.LastSeq {
-			o.sseq = state.LastSeq + 1
+		// Only clip the sseq if the OptStartSeq is not provided, otherwise
+		// it's possible that the stream just doesn't contain OptStartSeq yet.
+		if o.cfg.OptStartSeq == 0 {
+			if state.FirstSeq == 0 {
+				o.sseq = 1
+			} else if o.sseq < state.FirstSeq {
+				o.sseq = state.FirstSeq
+			} else if o.sseq > state.LastSeq {
+				o.sseq = state.LastSeq + 1
+			}
 		}
 	}
 
@@ -5202,6 +5213,7 @@ func (o *consumer) stopWithFlags(dflag, sdflag, doSignal, advisory bool) error {
 	if mset != nil {
 		mset.mu.Lock()
 		mset.removeConsumer(o)
+		// No need for cfgMu's lock since mset.mu.Lock superseeds it.
 		rp = mset.cfg.Retention
 		mset.mu.Unlock()
 	}
diff --git a/server/events.go b/server/events.go
index 2ebfc5eb..39de871f 100644
--- a/server/events.go
+++ b/server/events.go
@@ -130,6 +130,7 @@ type internal struct {
 	replies        map[string]msgHandler
 	sendq          *ipQueue[*pubMsg]
 	recvq          *ipQueue[*inSysMsg]
+	recvqp         *ipQueue[*inSysMsg] // For STATSZ/Pings
 	resetCh        chan struct{}
 	wg             sync.WaitGroup
 	sq             *sendq
@@ -412,15 +413,7 @@ type TypedEvent struct {
 
 // internalReceiveLoop will be responsible for dispatching all messages that
 // a server receives and needs to internally process, e.g. internal subs.
-func (s *Server) internalReceiveLoop() {
-	s.mu.RLock()
-	if s.sys == nil || s.sys.recvq == nil {
-		s.mu.RUnlock()
-		return
-	}
-	recvq := s.sys.recvq
-	s.mu.RUnlock()
-
+func (s *Server) internalReceiveLoop(recvq *ipQueue[*inSysMsg]) {
 	for s.eventsRunning() {
 		select {
 		case <-recvq.ch:
@@ -1139,7 +1132,7 @@ func (s *Server) initEventTracking() {
 	}
 	// Listen for ping messages that will be sent to all servers for statsz.
 	// This subscription is kept for backwards compatibility. Got replaced by ...PING.STATZ from below
-	if _, err := s.sysSubscribe(serverStatsPingReqSubj, s.noInlineCallback(s.statszReq)); err != nil {
+	if _, err := s.sysSubscribe(serverStatsPingReqSubj, s.noInlineCallbackStatsz(s.statszReq)); err != nil {
 		s.Errorf("Error setting up internal tracking: %v", err)
 		return
 	}
@@ -1182,23 +1175,42 @@ func (s *Server) initEventTracking() {
 			optz := &HealthzEventOptions{}
 			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (any, error) { return s.healthz(&optz.HealthzOptions), nil })
 		},
-		"PROFILEZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
-			optz := &ProfilezEventOptions{}
-			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (any, error) { return s.profilez(&optz.ProfilezOptions), nil })
-		},
+		"PROFILEZ": nil, // Special case, see below
 		"EXPVARZ": func(sub *subscription, c *client, _ *Account, subject, reply string, hdr, msg []byte) {
 			optz := &ExpvarzEventOptions{}
 			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (any, error) { return s.expvarz(optz), nil })
 		},
 	}
+	profilez := func(_ *subscription, c *client, _ *Account, _, rply string, rmsg []byte) {
+		hdr, msg := c.msgParts(rmsg)
+		// Need to copy since we are passing those to the go routine below.
+		hdr, msg = copyBytes(hdr), copyBytes(msg)
+		// Execute in its own go routine because CPU profiling, for instance,
+		// could take several seconds to complete.
+		go func() {
+			optz := &ProfilezEventOptions{}
+			s.zReq(c, rply, hdr, msg, &optz.EventFilterOptions, optz, func() (any, error) {
+				return s.profilez(&optz.ProfilezOptions), nil
+			})
+		}()
+	}
 	for name, req := range monSrvc {
+		var h msgHandler
+		switch name {
+		case "PROFILEZ":
+			h = profilez
+		case "STATSZ":
+			h = s.noInlineCallbackStatsz(req)
+		default:
+			h = s.noInlineCallback(req)
+		}
 		subject = fmt.Sprintf(serverDirectReqSubj, s.info.ID, name)
-		if _, err := s.sysSubscribe(subject, s.noInlineCallback(req)); err != nil {
+		if _, err := s.sysSubscribe(subject, h); err != nil {
 			s.Errorf("Error setting up internal tracking: %v", err)
 			return
 		}
 		subject = fmt.Sprintf(serverPingReqSubj, name)
-		if _, err := s.sysSubscribe(subject, s.noInlineCallback(req)); err != nil {
+		if _, err := s.sysSubscribe(subject, h); err != nil {
 			s.Errorf("Error setting up internal tracking: %v", err)
 			return
 		}
@@ -2451,16 +2463,39 @@ func (s *Server) sendAccountAuthErrorEvent(c *client, acc *Account, reason strin
 // rmsg contains header and the message. use client.msgParts(rmsg) to split them apart
 type msgHandler func(sub *subscription, client *client, acc *Account, subject, reply string, rmsg []byte)
 
+const (
+	recvQMuxed  = 1
+	recvQStatsz = 2
+)
+
 // Create a wrapped callback handler for the subscription that will move it to an
 // internal recvQ for processing not inline with routes etc.
 func (s *Server) noInlineCallback(cb sysMsgHandler) msgHandler {
+	return s.noInlineCallbackRecvQSelect(cb, recvQMuxed)
+}
+
+// Create a wrapped callback handler for the subscription that will move it to an
+// internal recvQ for Statsz/Pings for processing not inline with routes etc.
+func (s *Server) noInlineCallbackStatsz(cb sysMsgHandler) msgHandler {
+	return s.noInlineCallbackRecvQSelect(cb, recvQStatsz)
+}
+
+// Create a wrapped callback handler for the subscription that will move it to an
+// internal IPQueue for processing not inline with routes etc.
+func (s *Server) noInlineCallbackRecvQSelect(cb sysMsgHandler, recvQSelect int) msgHandler {
 	s.mu.RLock()
 	if !s.eventsEnabled() {
 		s.mu.RUnlock()
 		return nil
 	}
 	// Capture here for direct reference to avoid any unnecessary blocking inline with routes, gateways etc.
-	recvq := s.sys.recvq
+	var recvq *ipQueue[*inSysMsg]
+	switch recvQSelect {
+	case recvQStatsz:
+		recvq = s.sys.recvqp
+	default:
+		recvq = s.sys.recvq
+	}
 	s.mu.RUnlock()
 
 	return func(sub *subscription, c *client, acc *Account, subj, rply string, rmsg []byte) {
diff --git a/server/filestore.go b/server/filestore.go
index a1a54fb3..6b959a34 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -32,6 +32,7 @@ import (
 	"net"
 	"os"
 	"path/filepath"
+	"slices"
 	"sort"
 	"strings"
 	"sync"
@@ -1069,7 +1070,7 @@ func (fs *fileStore) addLostData(ld *LostStreamData) {
 		}
 		if added {
 			msgs := fs.ld.Msgs
-			sort.Slice(msgs, func(i, j int) bool { return msgs[i] < msgs[j] })
+			slices.Sort(msgs)
 			fs.ld.Bytes += ld.Bytes
 		}
 	} else {
@@ -1079,17 +1080,10 @@ func (fs *fileStore) addLostData(ld *LostStreamData) {
 
 // Helper to see if we already have this sequence reported in our lost data.
 func (ld *LostStreamData) exists(seq uint64) (int, bool) {
-	i, found := sort.Find(len(ld.Msgs), func(i int) int {
-		tseq := ld.Msgs[i]
-		if tseq < seq {
-			return -1
-		}
-		if tseq > seq {
-			return +1
-		}
-		return 0
+	i := slices.IndexFunc(ld.Msgs, func(i uint64) bool {
+		return i == seq
 	})
-	return i, found
+	return i, i > -1
 }
 
 func (fs *fileStore) removeFromLostData(seq uint64) {
@@ -1242,6 +1236,13 @@ func (mb *msgBlock) convertToEncrypted() error {
 	return nil
 }
 
+// Return the mb's index.
+func (mb *msgBlock) getIndex() uint32 {
+	mb.mu.RLock()
+	defer mb.mu.RUnlock()
+	return mb.index
+}
+
 // Rebuild the state of the blk based on what we have on disk in the N.blk file.
 // We will return any lost data, and we will return any delete tombstones we encountered.
 func (mb *msgBlock) rebuildState() (*LostStreamData, []uint64, error) {
@@ -2600,14 +2601,14 @@ func (fs *fileStore) FilteredState(sseq uint64, subj string) SimpleState {
 	return ss
 }
 
-// This is used to see if we can selectively jump start blocks based on filter subject and a floor block index.
-// Will return -1 if no matches at all.
-func (fs *fileStore) checkSkipFirstBlock(filter string, wc bool) (int, int) {
-	if filter == _EMPTY_ {
-		filter = fwcs
-		wc = true
+// This is used to see if we can selectively jump start blocks based on filter subject and a starting block index.
+// Will return -1 and ErrStoreEOF if no matches at all or no more from where we are.
+func (fs *fileStore) checkSkipFirstBlock(filter string, wc bool, bi int) (int, error) {
+	// If we match everything, just move to next blk.
+	if filter == _EMPTY_ || filter == fwcs {
+		return bi + 1, nil
 	}
-
+	// Move through psim to gather start and stop bounds.
 	start, stop := uint32(math.MaxUint32), uint32(0)
 	if wc {
 		fs.psim.Match(stringToBytes(filter), func(_ []byte, psi *psi) {
@@ -2621,51 +2622,26 @@ func (fs *fileStore) checkSkipFirstBlock(filter string, wc bool) (int, int) {
 	} else if psi, ok := fs.psim.Find(stringToBytes(filter)); ok {
 		start, stop = psi.fblk, psi.lblk
 	}
-	// Nothing found.
+	// Nothing was found.
 	if start == uint32(math.MaxUint32) {
-		return -1, -1
+		return -1, ErrStoreEOF
 	}
-	// Here we need to translate this to index into fs.blks properly.
-	mb := fs.bim[start]
-	if mb == nil {
-		// psim fblk can be lazy.
-		i := start + 1
-		for ; i <= stop; i++ {
-			mb = fs.bim[i]
-			if mb == nil {
-				continue
-			}
-			if _, f, _ := mb.filteredPending(filter, wc, 0); f > 0 {
-				break
-			}
-		}
-		// Update fblk since fblk was outdated.
-		if !wc {
-			if psi, ok := fs.psim.Find(stringToBytes(filter)); ok {
-				psi.fblk = i
-			}
-		} else {
-			fs.psim.Match(stringToBytes(filter), func(subj []byte, psi *psi) {
-				if i > psi.fblk {
-					psi.fblk = i
-				}
-			})
-		}
+	// Can not be nil so ok to inline dereference.
+	mbi := fs.blks[bi].getIndex()
+	// All matching msgs are behind us.
+	// Less than AND equal is important because we were called because we missed searching bi.
+	if stop <= mbi {
+		return -1, ErrStoreEOF
 	}
-	// Still nothing.
-	if mb == nil {
-		return -1, -1
-	}
-	// Grab first index.
-	fi, _ := fs.selectMsgBlockWithIndex(atomic.LoadUint64(&mb.last.seq))
-
-	// Grab last if applicable.
-	var li int
-	if mb = fs.bim[stop]; mb != nil {
-		li, _ = fs.selectMsgBlockWithIndex(atomic.LoadUint64(&mb.last.seq))
+	// If start is > index return dereference of fs.blks index.
+	if start > mbi {
+		if mb := fs.bim[start]; mb != nil {
+			ni, _ := fs.selectMsgBlockWithIndex(atomic.LoadUint64(&mb.last.seq))
+			return ni, nil
+		}
 	}
-
-	return fi, li
+	// Otherwise just bump to the next one.
+	return bi + 1, nil
 }
 
 // Optimized way for getting all num pending matching a filter subject.
@@ -6441,10 +6417,13 @@ func (fs *fileStore) LoadNextMsg(filter string, wc bool, start uint64, sm *Store
 				// We should not do this at all if we are already on the last block.
 				// Also if we are a wildcard do not check if large subject space.
 				const wcMaxSizeToCheck = 64 * 1024
+
+				// if len(blks) - 1 > N
+
 				if i == bi && i < len(fs.blks)-1 && (!wc || fs.psim.Size() < wcMaxSizeToCheck) {
-					nbi, lbi := fs.checkSkipFirstBlock(filter, wc)
+					nbi, err := fs.checkSkipFirstBlock(filter, wc, bi)
 					// Nothing available.
-					if nbi < 0 || lbi <= bi {
+					if err == ErrStoreEOF {
 						return nil, fs.state.LastSeq, ErrStoreEOF
 					}
 					// See if we can jump ahead here.
@@ -6546,9 +6525,7 @@ func (fs *fileStore) State() StreamState {
 
 	// Can not be guaranteed to be sorted.
 	if len(state.Deleted) > 0 {
-		sort.Slice(state.Deleted, func(i, j int) bool {
-			return state.Deleted[i] < state.Deleted[j]
-		})
+		slices.Sort(state.Deleted)
 		state.NumDeleted = len(state.Deleted)
 	}
 	return state
diff --git a/server/gateway.go b/server/gateway.go
index 1f462175..82df196e 100644
--- a/server/gateway.go
+++ b/server/gateway.go
@@ -15,6 +15,7 @@ package server
 
 import (
 	"bytes"
+	"cmp"
 	"crypto/sha256"
 	"crypto/tls"
 	"encoding/json"
@@ -22,7 +23,7 @@ import (
 	"math/rand"
 	"net"
 	"net/url"
-	"sort"
+	"slices"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -1730,9 +1731,7 @@ func (s *Server) getOutboundGatewayConnections(a *[]*client) {
 // Gateway write lock is held on entry
 func (g *srvGateway) orderOutboundConnectionsLocked() {
 	// Order the gateways by lowest RTT
-	sort.Slice(g.outo, func(i, j int) bool {
-		return g.outo[i].getRTTValue() < g.outo[j].getRTTValue()
-	})
+	slices.SortFunc(g.outo, func(i, j *client) int { return cmp.Compare(i.getRTTValue(), j.getRTTValue()) })
 }
 
 // Orders the array of outbound connections.
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index c675bd1d..479babf8 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -15,6 +15,7 @@ package server
 
 import (
 	"bytes"
+	"cmp"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -22,7 +23,7 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
-	"sort"
+	"slices"
 	"strconv"
 	"strings"
 	"sync/atomic"
@@ -824,10 +825,10 @@ func (js *jetStream) apiDispatch(sub *subscription, c *client, acc *Account, sub
 	// 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.
 	// Check pending and warn if getting backed up.
-	const warnThresh = 32
+	const warnThresh = 128
 	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)
+	if pending >= warnThresh {
+		s.rateLimitFormatWarnf("JetStream request queue has high pending count: %d", pending)
 	}
 }
 
@@ -1203,8 +1204,8 @@ func (s *Server) jsTemplateNamesRequest(sub *subscription, c *client, _ *Account
 	}
 
 	ts := acc.templates()
-	sort.Slice(ts, func(i, j int) bool {
-		return strings.Compare(ts[i].StreamTemplateConfig.Name, ts[j].StreamTemplateConfig.Name) < 0
+	slices.SortFunc(ts, func(i, j *streamTemplate) int {
+		return cmp.Compare(i.StreamTemplateConfig.Name, j.StreamTemplateConfig.Name)
 	})
 
 	tcnt := len(ts)
@@ -1624,7 +1625,7 @@ func (s *Server) jsStreamNamesRequest(sub *subscription, c *client, _ *Account,
 		}
 		js.mu.RUnlock()
 		if len(resp.Streams) > 1 {
-			sort.Slice(resp.Streams, func(i, j int) bool { return strings.Compare(resp.Streams[i], resp.Streams[j]) < 0 })
+			slices.Sort(resp.Streams)
 		}
 		numStreams = len(resp.Streams)
 		if offset > numStreams {
@@ -1640,9 +1641,7 @@ func (s *Server) jsStreamNamesRequest(sub *subscription, c *client, _ *Account,
 		msets := acc.filteredStreams(filter)
 		// Since we page results order matters.
 		if len(msets) > 1 {
-			sort.Slice(msets, func(i, j int) bool {
-				return strings.Compare(msets[i].cfg.Name, msets[j].cfg.Name) < 0
-			})
+			slices.SortFunc(msets, func(i, j *stream) int { return cmp.Compare(i.cfg.Name, j.cfg.Name) })
 		}
 
 		numStreams = len(msets)
@@ -1739,9 +1738,7 @@ func (s *Server) jsStreamListRequest(sub *subscription, c *client, _ *Account, s
 		msets = acc.filteredStreams(filter)
 	}
 
-	sort.Slice(msets, func(i, j int) bool {
-		return strings.Compare(msets[i].cfg.Name, msets[j].cfg.Name) < 0
-	})
+	slices.SortFunc(msets, func(i, j *stream) int { return cmp.Compare(i.cfg.Name, j.cfg.Name) })
 
 	scnt := len(msets)
 	if offset > scnt {
@@ -1944,7 +1941,7 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, a *Account, s
 					subjs = append(subjs, subj)
 				}
 				// Sort it
-				sort.Strings(subjs)
+				slices.Sort(subjs)
 
 				if offset > len(subjs) {
 					offset = len(subjs)
@@ -4074,7 +4071,7 @@ func (s *Server) jsConsumerNamesRequest(sub *subscription, c *client, _ *Account
 			resp.Consumers = append(resp.Consumers, consumer)
 		}
 		if len(resp.Consumers) > 1 {
-			sort.Slice(resp.Consumers, func(i, j int) bool { return strings.Compare(resp.Consumers[i], resp.Consumers[j]) < 0 })
+			slices.Sort(resp.Consumers)
 		}
 		numConsumers = len(resp.Consumers)
 		if offset > numConsumers {
@@ -4095,9 +4092,7 @@ func (s *Server) jsConsumerNamesRequest(sub *subscription, c *client, _ *Account
 		}
 
 		obs := mset.getPublicConsumers()
-		sort.Slice(obs, func(i, j int) bool {
-			return strings.Compare(obs[i].name, obs[j].name) < 0
-		})
+		slices.SortFunc(obs, func(i, j *consumer) int { return cmp.Compare(i.name, j.name) })
 
 		numConsumers = len(obs)
 		if offset > numConsumers {
@@ -4190,9 +4185,7 @@ func (s *Server) jsConsumerListRequest(sub *subscription, c *client, _ *Account,
 	}
 
 	obs := mset.getPublicConsumers()
-	sort.Slice(obs, func(i, j int) bool {
-		return strings.Compare(obs[i].name, obs[j].name) < 0
-	})
+	slices.SortFunc(obs, func(i, j *consumer) int { return cmp.Compare(i.name, j.name) })
 
 	ocnt := len(obs)
 	if offset > ocnt {
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index ce18b77c..18a67c32 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -15,6 +15,7 @@ package server
 
 import (
 	"bytes"
+	"cmp"
 	crand "crypto/rand"
 	"encoding/binary"
 	"encoding/json"
@@ -25,7 +26,7 @@ import (
 	"os"
 	"path/filepath"
 	"reflect"
-	"sort"
+	"slices"
 	"strconv"
 	"strings"
 	"sync/atomic"
@@ -2260,7 +2261,12 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 	// Make sure to stop the raft group on exit to prevent accidental memory bloat.
 	// This should be below the checkInMonitor call though to avoid stopping it out
 	// from underneath the one that is running since it will be the same raft node.
-	defer n.Stop()
+	defer func() {
+		// We might be closing during shutdown, don't pre-emptively stop here since we'll still want to install snapshots.
+		if !mset.closed.Load() {
+			n.Stop()
+		}
+	}()
 
 	qch, mqch, lch, aq, uch, ourPeerId := n.QuitC(), mset.monitorQuitC(), n.LeadChangeC(), n.ApplyQ(), mset.updateC(), meta.ID()
 
@@ -2419,6 +2425,9 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 			return
 		case <-aq.ch:
 			var ne, nb uint64
+			// If we bump clfs we will want to write out snapshot if within our time window.
+			pclfs := mset.getCLFS()
+
 			ces := aq.pop()
 			for _, ce := range ces {
 				// No special processing needed for when we are caught up on restart.
@@ -2435,6 +2444,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 					}
 					continue
 				}
+
 				// Apply our entries.
 				if err := js.applyStreamEntries(mset, ce, isRecovering); err == nil {
 					// Update our applied.
@@ -2466,7 +2476,13 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 
 			// Check about snapshotting
 			// If we have at least min entries to compact, go ahead and try to snapshot/compact.
-			if ne >= compactNumMin || nb > compactSizeMin {
+			if ne >= compactNumMin || nb > compactSizeMin || mset.getCLFS() > pclfs {
+				// We want to make sure we do not short circuit if transistioning from no clfs.
+				if pclfs == 0 {
+					// This is always false by default.
+					lastState.firstNeedsUpdate = true
+					lastSnapTime = time.Time{}
+				}
 				doSnapshot()
 			}
 
@@ -3125,13 +3141,10 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 				}
 			}
 
-			if !isRecovering && !mset.IsLeader() {
+			if isRecovering || !mset.IsLeader() {
 				if err := mset.processSnapshot(ss); err != nil {
 					return err
 				}
-			} else if isRecovering {
-				// On recovery, reset CLFS/FAILED.
-				mset.setCLFS(ss.Failed)
 			}
 		} else if e.Type == EntryRemovePeer {
 			js.mu.RLock()
@@ -5905,15 +5918,15 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 		return nil, &err
 	}
 	// Sort based on available from most to least, breaking ties by number of total streams assigned to the peer.
-	sort.Slice(nodes, func(i, j int) bool {
-		if nodes[i].avail == nodes[j].avail {
-			return nodes[i].ns < nodes[j].ns
+	slices.SortFunc(nodes, func(i, j wn) int {
+		if i.avail == j.avail {
+			return cmp.Compare(i.ns, j.ns)
 		}
-		return nodes[i].avail > nodes[j].avail
+		return -cmp.Compare(i.avail, j.avail) // reverse
 	})
 	// If we are placing a replicated stream, let's sort based on HAAssets, as that is more important to balance.
 	if cfg.Replicas > 1 {
-		sort.SliceStable(nodes, func(i, j int) bool { return nodes[i].ha < nodes[j].ha })
+		slices.SortStableFunc(nodes, func(i, j wn) int { return cmp.Compare(i.ha, j.ha) })
 	}
 
 	var results []string
@@ -6673,9 +6686,7 @@ func (s *Server) jsClusteredStreamListRequest(acc *Account, ci *ClientInfo, filt
 
 	// Needs to be sorted for offsets etc.
 	if len(streams) > 1 {
-		sort.Slice(streams, func(i, j int) bool {
-			return strings.Compare(streams[i].Config.Name, streams[j].Config.Name) < 0
-		})
+		slices.SortFunc(streams, func(i, j *streamAssignment) int { return cmp.Compare(i.Config.Name, j.Config.Name) })
 	}
 
 	scnt := len(streams)
@@ -6788,9 +6799,7 @@ LOOP:
 
 	// Needs to be sorted as well.
 	if len(resp.Streams) > 1 {
-		sort.Slice(resp.Streams, func(i, j int) bool {
-			return strings.Compare(resp.Streams[i].Config.Name, resp.Streams[j].Config.Name) < 0
-		})
+		slices.SortFunc(resp.Streams, func(i, j *StreamInfo) int { return cmp.Compare(i.Config.Name, j.Config.Name) })
 	}
 
 	resp.Total = scnt
@@ -6823,9 +6832,7 @@ func (s *Server) jsClusteredConsumerListRequest(acc *Account, ci *ClientInfo, of
 	}
 	// Needs to be sorted.
 	if len(consumers) > 1 {
-		sort.Slice(consumers, func(i, j int) bool {
-			return strings.Compare(consumers[i].Name, consumers[j].Name) < 0
-		})
+		slices.SortFunc(consumers, func(i, j *consumerAssignment) int { return cmp.Compare(i.Config.Name, j.Config.Name) })
 	}
 
 	ocnt := len(consumers)
@@ -6935,9 +6942,7 @@ LOOP:
 
 	// Needs to be sorted as well.
 	if len(resp.Consumers) > 1 {
-		sort.Slice(resp.Consumers, func(i, j int) bool {
-			return strings.Compare(resp.Consumers[i].Name, resp.Consumers[j].Name) < 0
-		})
+		slices.SortFunc(resp.Consumers, func(i, j *ConsumerInfo) int { return cmp.Compare(i.Name, j.Name) })
 	}
 
 	resp.Total = ocnt
@@ -7810,13 +7815,16 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 		}
 		// Expected last sequence per subject.
 		// We can check for last sequence per subject but only if the expected seq <= lseq.
-		if seq, exists := getExpectedLastSeqPerSubject(hdr); exists && store != nil && seq > 0 && seq <= lseq {
+		if seq, exists := getExpectedLastSeqPerSubject(hdr); exists && store != nil && seq <= lseq {
 			var smv StoreMsg
 			var fseq uint64
 			sm, err := store.LoadLastMsg(subject, &smv)
 			if sm != nil {
 				fseq = sm.seq
 			}
+			if err == ErrStoreMsgNotFound && seq == 0 {
+				fseq, err = 0, nil
+			}
 			if err != nil || fseq != seq {
 				if canRespond {
 					var resp = &JSPubAckResponse{PubAck: &PubAck{Stream: name}}
@@ -8548,9 +8556,7 @@ func (js *jetStream) clusterInfo(rg *raftGroup) *ClusterInfo {
 	}
 	// Order the result based on the name so that we get something consistent
 	// when doing repeated stream info in the CLI, etc...
-	sort.Slice(ci.Replicas, func(i, j int) bool {
-		return ci.Replicas[i].Name < ci.Replicas[j].Name
-	})
+	slices.SortFunc(ci.Replicas, func(i, j *PeerInfo) int { return cmp.Compare(i.Name, j.Name) })
 	return ci
 }
 
@@ -8610,9 +8616,8 @@ func (js *jetStream) streamAlternates(ci *ClientInfo, stream string) []StreamAlt
 	}
 
 	// Sort based on our weights that originate from the request itself.
-	sort.Slice(alts, func(i, j int) bool {
-		return weights[alts[i].Cluster] > weights[alts[j].Cluster]
-	})
+	// reverse sort
+	slices.SortFunc(alts, func(i, j StreamAlternate) int { return -cmp.Compare(weights[i.Cluster], weights[j.Cluster]) })
 
 	return alts
 }
diff --git a/server/log.go b/server/log.go
index e1b9078a..68222658 100644
--- a/server/log.go
+++ b/server/log.go
@@ -219,6 +219,14 @@ func (s *Server) Warnf(format string, v ...any) {
 	}, format, v...)
 }
 
+func (s *Server) rateLimitFormatWarnf(format string, v ...any) {
+	if _, loaded := s.rateLimitLogging.LoadOrStore(format, time.Now()); loaded {
+		return
+	}
+	statement := fmt.Sprintf(format, v...)
+	s.Warnf("%s", statement)
+}
+
 func (s *Server) RateLimitWarnf(format string, v ...any) {
 	statement := fmt.Sprintf(format, v...)
 	if _, loaded := s.rateLimitLogging.LoadOrStore(statement, time.Now()); loaded {
diff --git a/server/monitor.go b/server/monitor.go
index 8396555d..46e262e5 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -15,6 +15,7 @@ package server
 
 import (
 	"bytes"
+	"cmp"
 	"crypto/sha256"
 	"crypto/tls"
 	"crypto/x509"
@@ -29,6 +30,7 @@ import (
 	"path/filepath"
 	"runtime"
 	"runtime/pprof"
+	"slices"
 	"sort"
 	"strconv"
 	"strings"
@@ -3018,9 +3020,7 @@ func (s *Server) Jsz(opts *JSzOptions) (*JSInfo, error) {
 		accounts = []*jsAccount{accounts[filterIdx]}
 	} else if opts.Accounts {
 		if opts.Offset != 0 {
-			sort.Slice(accounts, func(i, j int) bool {
-				return strings.Compare(accounts[i].acc().Name, accounts[j].acc().Name) < 0
-			})
+			slices.SortFunc(accounts, func(i, j *jsAccount) int { return cmp.Compare(i.acc().Name, j.acc().Name) })
 			if opts.Offset > len(accounts) {
 				accounts = []*jsAccount{}
 			} else {
diff --git a/server/mqtt.go b/server/mqtt.go
index 33a00109..1c6a98a2 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -15,6 +15,7 @@ package server
 
 import (
 	"bytes"
+	"cmp"
 	"crypto/tls"
 	"encoding/binary"
 	"encoding/json"
@@ -23,7 +24,7 @@ import (
 	"io"
 	"net"
 	"net/http"
-	"sort"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -4276,9 +4277,7 @@ func (s *Server) mqttCheckPubRetainedPerms() {
 			})
 		}
 		asm.mu.RUnlock()
-		sort.Slice(rms, func(i, j int) bool {
-			return rms[i].rmsg.sseq < rms[j].rmsg.sseq
-		})
+		slices.SortFunc(rms, func(i, j retainedMsg) int { return cmp.Compare(i.rmsg.sseq, j.rmsg.sseq) })
 
 		perms := map[string]*perm{}
 		deletes := map[string]uint64{}
diff --git a/server/reload.go b/server/reload.go
index d15525d5..15bbae1e 100644
--- a/server/reload.go
+++ b/server/reload.go
@@ -14,12 +14,13 @@
 package server
 
 import (
+	"cmp"
 	"crypto/tls"
 	"errors"
 	"fmt"
 	"net/url"
 	"reflect"
-	"sort"
+	"slices"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -1131,38 +1132,24 @@ func (s *Server) reloadOptions(curOpts, newOpts *Options) error {
 func imposeOrder(value any) error {
 	switch value := value.(type) {
 	case []*Account:
-		sort.Slice(value, func(i, j int) bool {
-			return value[i].Name < value[j].Name
-		})
+		slices.SortFunc(value, func(i, j *Account) int { return cmp.Compare(i.Name, j.Name) })
 		for _, a := range value {
-			sort.Slice(a.imports.streams, func(i, j int) bool {
-				return a.imports.streams[i].acc.Name < a.imports.streams[j].acc.Name
-			})
+			slices.SortFunc(a.imports.streams, func(i, j *streamImport) int { return cmp.Compare(i.acc.Name, j.acc.Name) })
 		}
 	case []*User:
-		sort.Slice(value, func(i, j int) bool {
-			return value[i].Username < value[j].Username
-		})
+		slices.SortFunc(value, func(i, j *User) int { return cmp.Compare(i.Username, j.Username) })
 	case []*NkeyUser:
-		sort.Slice(value, func(i, j int) bool {
-			return value[i].Nkey < value[j].Nkey
-		})
+		slices.SortFunc(value, func(i, j *NkeyUser) int { return cmp.Compare(i.Nkey, j.Nkey) })
 	case []*url.URL:
-		sort.Slice(value, func(i, j int) bool {
-			return value[i].String() < value[j].String()
-		})
+		slices.SortFunc(value, func(i, j *url.URL) int { return cmp.Compare(i.String(), j.String()) })
 	case []string:
-		sort.Strings(value)
+		slices.Sort(value)
 	case []*jwt.OperatorClaims:
-		sort.Slice(value, func(i, j int) bool {
-			return value[i].Issuer < value[j].Issuer
-		})
+		slices.SortFunc(value, func(i, j *jwt.OperatorClaims) int { return cmp.Compare(i.Issuer, j.Issuer) })
 	case GatewayOpts:
-		sort.Slice(value.Gateways, func(i, j int) bool {
-			return value.Gateways[i].Name < value.Gateways[j].Name
-		})
+		slices.SortFunc(value.Gateways, func(i, j *RemoteGatewayOpts) int { return cmp.Compare(i.Name, j.Name) })
 	case WebsocketOpts:
-		sort.Strings(value.AllowedOrigins)
+		slices.Sort(value.AllowedOrigins)
 	case string, bool, uint8, int, int32, int64, time.Duration, float64, nil, LeafNodeOpts, ClusterOpts, *tls.Config, PinnedCertSet,
 		*URLAccResolver, *MemAccResolver, *DirAccResolver, *CacheDirAccResolver, Authentication, MQTTOpts, jwt.TagList,
 		*OCSPConfig, map[string]string, JSLimitOpts, StoreCipher, *OCSPResponseCacheConfig:
diff --git a/server/server.go b/server/server.go
index 58be465a..4907bd5b 100644
--- a/server/server.go
+++ b/server/server.go
@@ -1702,12 +1702,14 @@ func (s *Server) setSystemAccount(acc *Account) error {
 		replies: make(map[string]msgHandler),
 		sendq:   newIPQueue[*pubMsg](s, "System sendQ"),
 		recvq:   newIPQueue[*inSysMsg](s, "System recvQ"),
+		recvqp:  newIPQueue[*inSysMsg](s, "System recvQ Pings"),
 		resetCh: make(chan struct{}),
 		sq:      s.newSendQ(),
 		statsz:  eventsHBInterval,
 		orphMax: 5 * eventsHBInterval,
 		chkOrph: 3 * eventsHBInterval,
 	}
+	recvq, recvqp := s.sys.recvq, s.sys.recvqp
 	s.sys.wg.Add(1)
 	s.mu.Unlock()
 
@@ -1721,7 +1723,9 @@ func (s *Server) setSystemAccount(acc *Account) error {
 	go s.internalSendLoop(&s.sys.wg)
 
 	// Start the internal loop for inbound messages.
-	go s.internalReceiveLoop()
+	go s.internalReceiveLoop(recvq)
+	// Start the internal loop for inbound STATSZ/Ping messages.
+	go s.internalReceiveLoop(recvqp)
 
 	// Start up our general subscriptions
 	s.initEventTracking()
@@ -2779,6 +2783,7 @@ func (s *Server) StartProfiler() {
 		Addr:           hp,
 		Handler:        http.DefaultServeMux,
 		MaxHeaderBytes: 1 << 20,
+		ReadTimeout:    time.Second * 5,
 	}
 	s.profiler = l
 	s.profilingServer = srv
@@ -2978,10 +2983,11 @@ func (s *Server) startMonitoring(secure bool) error {
 	// to return empty response or unable to display page if the
 	// server needs more time to build the response.
 	srv := &http.Server{
-		Addr:           hp,
-		Handler:        mux,
-		MaxHeaderBytes: 1 << 20,
-		ErrorLog:       log.New(&captureHTTPServerLog{s, "monitoring: "}, _EMPTY_, 0),
+		Addr:              hp,
+		Handler:           mux,
+		MaxHeaderBytes:    1 << 20,
+		ErrorLog:          log.New(&captureHTTPServerLog{s, "monitoring: "}, _EMPTY_, 0),
+		ReadHeaderTimeout: time.Second * 5,
 	}
 	s.mu.Lock()
 	s.http = httpListener
diff --git a/server/stream.go b/server/stream.go
index 1279efae..95b18e15 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -236,6 +236,7 @@ type stream struct {
 	consumers map[string]*consumer    // The consumers for this stream.
 	numFilter int                     // The number of filtered consumers.
 	cfg       StreamConfig            // The stream's config.
+	cfgMu     sync.RWMutex            // Config mutex used to solve some races with consumer code
 	created   time.Time               // Time the stream was created.
 	stype     StorageType             // The storage type.
 	tier      string                  // The tier is the number of replicas for the stream (e.g. "R1" or "R3").
@@ -1018,6 +1019,9 @@ func (mset *stream) lastSeqAndCLFS() (uint64, uint64) {
 }
 
 func (mset *stream) getCLFS() uint64 {
+	if mset == nil {
+		return 0
+	}
 	mset.clMu.Lock()
 	defer mset.clMu.Unlock()
 	return mset.clfs
@@ -1950,7 +1954,14 @@ func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool)
 	}
 
 	// Now update config and store's version of our config.
+	// Although we are under the stream write lock, we will also assign the new
+	// configuration under mset.cfgMu lock. This is so that in places where
+	// mset.mu cannot be acquired (like many cases in consumer.go where code
+	// is under the consumer's lock), and the stream's configuration needs to
+	// be inspected, one can use mset.cfgMu's read lock to do that safely.
+	mset.cfgMu.Lock()
 	mset.cfg = *cfg
+	mset.cfgMu.Unlock()
 
 	// If we're changing retention and haven't errored because of consumer
 	// replicas by now, whip through and update the consumer retention.
@@ -2001,6 +2012,15 @@ func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool)
 	return nil
 }
 
+// Small helper to return the Name field from mset.cfg, protected by
+// the mset.cfgMu mutex. This is simply because we have several places
+// in consumer.go where we need it.
+func (mset *stream) getCfgName() string {
+	mset.cfgMu.RLock()
+	defer mset.cfgMu.RUnlock()
+	return mset.cfg.Name
+}
+
 // Purge will remove all messages from the stream and underlying store based on the request.
 func (mset *stream) purge(preq *JSApiStreamPurgeRequest) (purged uint64, err error) {
 	mset.mu.RLock()
@@ -4369,7 +4389,6 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 		}
 
 		// Expected last sequence per subject.
-		// If we are clustered we have prechecked seq > 0.
 		if seq, exists := getExpectedLastSeqPerSubject(hdr); exists {
 			// TODO(dlc) - We could make a new store func that does this all in one.
 			var smv StoreMsg
@@ -4521,7 +4540,7 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 		mset.lmsgId = msgId
 		// If we have a msgId make sure to save.
 		if msgId != _EMPTY_ {
-			mset.storeMsgIdLocked(&ddentry{msgId, seq, ts})
+			mset.storeMsgIdLocked(&ddentry{msgId, mset.lseq, ts})
 		}
 		if canRespond {
 			response = append(pubAck, strconv.FormatUint(mset.lseq, 10)...)
@@ -5071,6 +5090,10 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 
 	// Kick monitor and collect consumers first.
 	mset.mu.Lock()
+
+	// Mark closed.
+	mset.closed.Store(true)
+
 	// Signal to the monitor loop.
 	// Can't use qch here.
 	if mset.mqch != nil {
@@ -5131,9 +5154,6 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 		mset.sendDeleteAdvisoryLocked()
 	}
 
-	// Mark closed.
-	mset.closed.Store(true)
-
 	// Quit channel, do this after sending the delete advisory
 	if mset.qch != nil {
 		close(mset.qch)


Test patch:
diff --git a/server/auth_callout_test.go b/server/auth_callout_test.go
index 1458dca2..b9282b4d 100644
--- a/server/auth_callout_test.go
+++ b/server/auth_callout_test.go
@@ -22,7 +22,7 @@ import (
 	"fmt"
 	"os"
 	"reflect"
-	"sort"
+	"slices"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -709,9 +709,9 @@ func TestAuthCalloutOperatorModeBasics(t *testing.T) {
 		t.Fatalf("Expected to be switched to %q, but got %q", tpub, userInfo.Account)
 	}
 	require_True(t, len(userInfo.Permissions.Publish.Allow) == 2)
-	sort.Strings(userInfo.Permissions.Publish.Allow)
+	slices.Sort(userInfo.Permissions.Publish.Allow)
 	require_Equal(t, "foo.>", userInfo.Permissions.Publish.Allow[1])
-	sort.Strings(userInfo.Permissions.Subscribe.Allow)
+	slices.Sort(userInfo.Permissions.Subscribe.Allow)
 	require_True(t, len(userInfo.Permissions.Subscribe.Allow) == 2)
 	require_Equal(t, "foo.>", userInfo.Permissions.Subscribe.Allow[1])
 }
@@ -834,9 +834,9 @@ func testAuthCalloutScopedUser(t *testing.T, allowAnyAccount bool) {
 		t.Fatalf("Expected to be switched to %q, but got %q", tpub, userInfo.Account)
 	}
 	require_True(t, len(userInfo.Permissions.Publish.Allow) == 2)
-	sort.Strings(userInfo.Permissions.Publish.Allow)
+	slices.Sort(userInfo.Permissions.Publish.Allow)
 	require_Equal(t, "foo.>", userInfo.Permissions.Publish.Allow[1])
-	sort.Strings(userInfo.Permissions.Subscribe.Allow)
+	slices.Sort(userInfo.Permissions.Subscribe.Allow)
 	require_True(t, len(userInfo.Permissions.Subscribe.Allow) == 2)
 	require_Equal(t, "foo.>", userInfo.Permissions.Subscribe.Allow[1])
 
diff --git a/server/events_test.go b/server/events_test.go
index 0d52ca20..6545db66 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -2846,6 +2846,62 @@ func TestServerEventsPingStatsZ(t *testing.T) {
 	}
 }
 
+func TestServerEventsPingStatsZDedicatedRecvQ(t *testing.T) {
+	sa, _, sb, optsB, akp := runTrustedCluster(t)
+	defer sa.Shutdown()
+	defer sb.Shutdown()
+	url := fmt.Sprintf("nats://%s:%d", optsB.Host, optsB.Port)
+	nc, err := nats.Connect(url, createUserCreds(t, sb, akp))
+	require_NoError(t, err)
+	defer nc.Close()
+	// We need to wait a little bit for the $SYS.SERVER.ACCOUNT.%s.CONNS
+	// event to be pushed in the mux'ed queue.
+	time.Sleep(300 * time.Millisecond)
+
+	testReq := func(t *testing.T, subj string, expectTwo bool) {
+		for _, s := range []*Server{sa, sb} {
+			s.mu.RLock()
+			recvq := s.sys.recvq
+			s.mu.RUnlock()
+			recvq.Lock()
+			defer recvq.Unlock()
+		}
+		reply := nc.NewRespInbox()
+		sub := natsSubSync(t, nc, reply)
+		nc.PublishRequest(subj, reply, nil)
+		msg := natsNexMsg(t, sub, time.Second)
+		if len(msg.Data) == 0 {
+			t.Fatal("Unexpected empty response")
+		}
+		// Make sure its a statsz
+		m := ServerStatsMsg{}
+		err := json.Unmarshal(msg.Data, &m)
+		require_NoError(t, err)
+		require_False(t, m.Stats.Start.IsZero())
+		if expectTwo {
+			msg = natsNexMsg(t, sub, time.Second)
+			err = json.Unmarshal(msg.Data, &m)
+			require_NoError(t, err)
+			require_False(t, m.Stats.Start.IsZero())
+		}
+	}
+	const statsz = "STATSZ"
+	for _, test := range []struct {
+		name      string
+		f         func() string
+		expectTwo bool
+	}{
+		{"server stats ping request subject", func() string { return serverStatsPingReqSubj }, true},
+		{"server ping request subject", func() string { return fmt.Sprintf(serverPingReqSubj, statsz) }, true},
+		{"server a direct request subject", func() string { return fmt.Sprintf(serverDirectReqSubj, sa.ID(), statsz) }, false},
+		{"server b direct request subject", func() string { return fmt.Sprintf(serverDirectReqSubj, sb.ID(), statsz) }, false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			testReq(t, test.f(), test.expectTwo)
+		})
+	}
+}
+
 func TestServerEventsPingStatsZFilter(t *testing.T) {
 	sa, _, sb, optsB, akp := runTrustedCluster(t)
 	defer sa.Shutdown()
@@ -3551,10 +3607,71 @@ func TestClusterSetupMsgs(t *testing.T) {
 
 	var totalOut int
 	for _, server := range c.servers {
-		totalOut += int(server.outMsgs)
+		totalOut += int(atomic.LoadInt64(&server.outMsgs))
 	}
 	totalExpected := numServers * numServers
 	if totalOut >= totalExpected {
 		t.Fatalf("Total outMsgs is %d, expected < %d\n", totalOut, totalExpected)
 	}
 }
+
+func TestServerEventsProfileZNotBlockingRecvQ(t *testing.T) {
+	sa, _, sb, optsB, akp := runTrustedCluster(t)
+	defer sa.Shutdown()
+	defer sb.Shutdown()
+	// For this test, we will run a single server because the profiling
+	// would fail for the second server since it would detect that
+	// one profiling is already running (2 servers, but same process).
+	sa.Shutdown()
+	url := fmt.Sprintf("nats://%s:%d", optsB.Host, optsB.Port)
+	nc, err := nats.Connect(url, createUserCreds(t, sb, akp))
+	require_NoError(t, err)
+	defer nc.Close()
+	// We need to wait a little bit for the $SYS.SERVER.ACCOUNT.%s.CONNS
+	// event to be pushed in the mux'ed queue.
+	time.Sleep(300 * time.Millisecond)
+
+	po := ProfilezOptions{Name: "cpu", Duration: 1 * time.Second}
+	req, err := json.Marshal(po)
+	require_NoError(t, err)
+
+	testReq := func(t *testing.T, subj string) {
+		// Block the recvQ by locking it for the duration of this test.
+		sb.mu.RLock()
+		recvq := sb.sys.recvq
+		sb.mu.RUnlock()
+		recvq.Lock()
+		defer recvq.Unlock()
+
+		// Send the profilez request on the given subject.
+		reply := nc.NewRespInbox()
+		sub := natsSubSync(t, nc, reply)
+		nc.PublishRequest(subj, reply, req)
+		msg := natsNexMsg(t, sub, 10*time.Second)
+		if len(msg.Data) == 0 {
+			t.Fatal("Unexpected empty response")
+		}
+		// Make sure its a ServerAPIResponse
+		resp := ServerAPIResponse{Data: &ProfilezStatus{}}
+		err := json.Unmarshal(msg.Data, &resp)
+		require_NoError(t, err)
+		// Check profile status to make sure that we got something.
+		ps := resp.Data.(*ProfilezStatus)
+		if ps.Error != _EMPTY_ {
+			t.Fatalf("%s", ps.Error)
+		}
+		require_True(t, len(ps.Profile) > 0)
+	}
+	const profilez = "PROFILEZ"
+	for _, test := range []struct {
+		name string
+		f    func() string
+	}{
+		{"server profilez request subject", func() string { return fmt.Sprintf(serverPingReqSubj, profilez) }},
+		{"server direct request subject", func() string { return fmt.Sprintf(serverDirectReqSubj, sb.ID(), profilez) }},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			testReq(t, test.f())
+		})
+	}
+}
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 7e74368a..1ab23b09 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -7248,28 +7248,6 @@ func TestFileStoreCheckSkipFirstBlockBug(t *testing.T) {
 	require_NoError(t, err)
 }
 
-// https://github.com/nats-io/nats-server/issues/5705
-func TestFileStoreCheckSkipFirstBlockEmptyFilter(t *testing.T) {
-	sd := t.TempDir()
-	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: sd, BlockSize: 128},
-		StreamConfig{Name: "zzz", Subjects: []string{"foo.*.*"}, Storage: FileStorage})
-	require_NoError(t, err)
-	defer fs.Stop()
-
-	msg := []byte("hello")
-	// Create 4 blocks, each block holds 2 msgs
-	for i := 0; i < 4; i++ {
-		fs.StoreMsg("foo.22.bar", nil, msg)
-		fs.StoreMsg("foo.22.baz", nil, msg)
-	}
-	require_Equal(t, fs.numMsgBlocks(), 4)
-
-	nbi, lbi := fs.checkSkipFirstBlock(_EMPTY_, false)
-	require_Equal(t, nbi, 0)
-	require_Equal(t, lbi, 3)
-}
-
 // https://github.com/nats-io/nats-server/issues/5702
 func TestFileStoreTombstoneRbytes(t *testing.T) {
 	fs, err := newFileStore(
@@ -7336,6 +7314,71 @@ func TestFileStoreMsgBlockShouldCompact(t *testing.T) {
 	require_False(t, shouldCompact)
 }
 
+func TestFileStoreCheckSkipFirstBlockNotLoadOldBlocks(t *testing.T) {
+	sd := t.TempDir()
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: sd, BlockSize: 128},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	msg := []byte("hello")
+
+	fs.StoreMsg("foo.BB.bar", nil, msg)
+	fs.StoreMsg("foo.AA.bar", nil, msg)
+	for i := 0; i < 6; i++ {
+		fs.StoreMsg("foo.BB.bar", nil, msg)
+	}
+	fs.StoreMsg("foo.AA.bar", nil, msg) // Sequence 9
+	fs.StoreMsg("foo.AA.bar", nil, msg) // Sequence 10
+
+	for i := 0; i < 4; i++ {
+		fs.StoreMsg("foo.BB.bar", nil, msg)
+	}
+
+	// Should have created 7 blocks.
+	// BB AA | BB BB | BB BB | BB BB | AA AA | BB BB | BB BB
+	require_Equal(t, fs.numMsgBlocks(), 7)
+
+	fs.RemoveMsg(1)
+	fs.RemoveMsg(2)
+
+	// First block should be gone now.
+	// -- -- | BB BB | BB BB | BB BB | AA AA | BB BB | BB BB
+	require_Equal(t, fs.numMsgBlocks(), 6)
+
+	// Remove all blk cache and fss.
+	fs.mu.RLock()
+	for _, mb := range fs.blks {
+		mb.mu.Lock()
+		mb.fss, mb.cache = nil, nil
+		mb.mu.Unlock()
+	}
+	fs.mu.RUnlock()
+
+	// But this means that the psim still points fblk to block 1.
+	// So when we try to load AA from near the end (last AA sequence), it will not find anything and will then
+	// check if we can skip ahead, but in the process reload blocks 2, 3, 4 amd 5..
+	// This can trigger for an up to date consumer near the end of the stream that gets a new pull request that will pop it out of msgWait
+	// and it will call LoadNextMsg() like we do here with starting sequence of 11.
+	_, _, err = fs.LoadNextMsg("foo.AA.bar", false, 11, nil)
+	require_Error(t, err, ErrStoreEOF)
+
+	// Now make sure we did not load fss and cache.
+	var loaded int
+	fs.mu.RLock()
+	for _, mb := range fs.blks {
+		mb.mu.RLock()
+		if mb.cache != nil || mb.fss != nil {
+			loaded++
+		}
+		mb.mu.RUnlock()
+	}
+	fs.mu.RUnlock()
+	// We will load last block for starting seq 9, but no others should have loaded.
+	require_Equal(t, loaded, 1)
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Benchmarks
 ///////////////////////////////////////////////////////////////////////////
diff --git a/server/jetstream_cluster_4_test.go b/server/jetstream_cluster_4_test.go
index 3d619e93..44a6a815 100644
--- a/server/jetstream_cluster_4_test.go
+++ b/server/jetstream_cluster_4_test.go
@@ -2416,3 +2416,130 @@ func TestJetStreamClusterMetaSyncOrphanCleanup(t *testing.T) {
 		require_Equal(t, state.Msgs, 10)
 	}
 }
+
+func TestJetStreamClusterKeyValueDesyncAfterHardKill(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3F", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.serverByName("S-1"))
+	defer nc.Close()
+
+	kv, err := js.CreateKeyValue(&nats.KeyValueConfig{
+		Bucket:   "inconsistency",
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	// First create should succeed.
+	revision, err := kv.Create("key.exists", []byte("1"))
+	require_NoError(t, err)
+	require_Equal(t, revision, 1)
+
+	// Second create will be rejected but bump CLFS.
+	_, err = kv.Create("key.exists", []byte("2"))
+	require_Error(t, err)
+
+	// Insert a new message, should only be applied once, even if we hard kill and replay afterward.
+	revision, err = kv.Put("key.put", []byte("3"))
+	require_NoError(t, err)
+	require_Equal(t, revision, 2)
+
+	// Restart a server
+	s3 := c.serverByName("S-3")
+	// We will remove the index.db file after we shutdown.
+	mset, err := s3.GlobalAccount().lookupStream("KV_inconsistency")
+	require_NoError(t, err)
+	fs := mset.store.(*fileStore)
+	ifile := filepath.Join(fs.fcfg.StoreDir, msgDir, "index.db")
+
+	s3.Shutdown()
+	s3.WaitForShutdown()
+	// Remove the index.db file to simulate a hard kill where server can not write out the index.db file.
+	require_NoError(t, os.Remove(ifile))
+
+	c.restartServer(s3)
+	c.waitOnClusterReady()
+	c.waitOnAllCurrent()
+
+	getStreamDetails := func(t *testing.T, c *cluster, accountName, streamName string) *StreamDetail {
+		t.Helper()
+		srv := c.streamLeader(accountName, streamName)
+		if srv == nil {
+			return nil
+		}
+		jsz, err := srv.Jsz(&JSzOptions{Accounts: true, Streams: true, Consumer: true})
+		require_NoError(t, err)
+		for _, acc := range jsz.AccountDetails {
+			if acc.Name == accountName {
+				for _, stream := range acc.Streams {
+					if stream.Name == streamName {
+						return &stream
+					}
+				}
+			}
+		}
+		t.Error("Could not find account details")
+		return nil
+	}
+
+	checkState := func(t *testing.T, c *cluster, accountName, streamName string) error {
+		t.Helper()
+
+		leaderSrv := c.streamLeader(accountName, streamName)
+		if leaderSrv == nil {
+			return fmt.Errorf("no leader server found for stream %q", streamName)
+		}
+		streamLeader := getStreamDetails(t, c, accountName, streamName)
+		if streamLeader == nil {
+			return fmt.Errorf("no leader found for stream %q", streamName)
+		}
+		var errs []error
+		for _, srv := range c.servers {
+			if srv == leaderSrv {
+				// Skip self
+				continue
+			}
+			acc, err := srv.LookupAccount(accountName)
+			require_NoError(t, err)
+			stream, err := acc.lookupStream(streamName)
+			require_NoError(t, err)
+			state := stream.state()
+
+			if state.Msgs != streamLeader.State.Msgs {
+				err := fmt.Errorf("[%s] Leader %v has %d messages, Follower %v has %d messages",
+					streamName, leaderSrv, streamLeader.State.Msgs,
+					srv, state.Msgs,
+				)
+				errs = append(errs, err)
+			}
+			if state.FirstSeq != streamLeader.State.FirstSeq {
+				err := fmt.Errorf("[%s] Leader %v FirstSeq is %d, Follower %v is at %d",
+					streamName, leaderSrv, streamLeader.State.FirstSeq,
+					srv, state.FirstSeq,
+				)
+				errs = append(errs, err)
+			}
+			if state.LastSeq != streamLeader.State.LastSeq {
+				err := fmt.Errorf("[%s] Leader %v LastSeq is %d, Follower %v is at %d",
+					streamName, leaderSrv, streamLeader.State.LastSeq,
+					srv, state.LastSeq,
+				)
+				errs = append(errs, err)
+			}
+			if state.NumDeleted != streamLeader.State.NumDeleted {
+				err := fmt.Errorf("[%s] Leader %v NumDeleted is %d, Follower %v is at %d",
+					streamName, leaderSrv, streamLeader.State.NumDeleted,
+					srv, state.NumDeleted,
+				)
+				errs = append(errs, err)
+			}
+		}
+		if len(errs) > 0 {
+			return errors.Join(errs...)
+		}
+		return nil
+	}
+
+	err = checkState(t, c, "$G", "KV_inconsistency")
+	require_NoError(t, err)
+}
diff --git a/server/jetstream_consumer_test.go b/server/jetstream_consumer_test.go
index e408621d..46ce0851 100644
--- a/server/jetstream_consumer_test.go
+++ b/server/jetstream_consumer_test.go
@@ -20,7 +20,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"math/rand"
-	"sort"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -162,9 +162,7 @@ func TestJetStreamConsumerMultipleFiltersRace(t *testing.T) {
 		if len(seqs) != 3*total {
 			return fmt.Errorf("found %d messages instead of %d", len(seqs), 3*total)
 		}
-		sort.Slice(seqs, func(i, j int) bool {
-			return seqs[i] < seqs[j]
-		})
+		slices.Sort(seqs)
 		for i := 1; i < len(seqs); i++ {
 			if seqs[i] != seqs[i-1]+1 {
 				fmt.Printf("seqs: %+v\n", seqs)
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 6ff7ca09..4c1fba80 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -34,7 +34,7 @@ import (
 	"reflect"
 	"runtime"
 	"runtime/debug"
-	"sort"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -7872,7 +7872,7 @@ func TestJetStreamRequestAPI(t *testing.T) {
 	if len(tListResp.Templates) != 2 {
 		t.Fatalf("Expected 2 templates but got %d", len(tListResp.Templates))
 	}
-	sort.Strings(tListResp.Templates)
+	slices.Sort(tListResp.Templates)
 	if tListResp.Templates[0] != "kv" {
 		t.Fatalf("Expected to get %q, but got %q", "kv", tListResp.Templates[0])
 	}
@@ -22463,7 +22463,7 @@ func TestInterestConsumerFilterEdit(t *testing.T) {
 }
 
 // https://github.com/nats-io/nats-server/issues/5383
-func TestInterestStreamWithFilterSubjectsConsumer(t *testing.T) {
+func TestJetStreamInterestStreamWithFilterSubjectsConsumer(t *testing.T) {
 	s := RunBasicJetStreamServer(t)
 	defer s.Shutdown()
 
@@ -22803,3 +22803,88 @@ func TestJetStreamConsumerInfoNumPending(t *testing.T) {
 	require_NoError(t, err)
 	require_Equal(t, ci.NumPending, 100)
 }
+
+func TestJetStreamConsumerStartSequenceNotInStream(t *testing.T) {
+	// This test is checking that we still correctly set the start
+	// sequence of a consumer if that start sequence doesn't appear
+	// in the stream yet. Previously this would have been clipped
+	// back to between the first and last seq from the stream state.
+
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"test"},
+	})
+	require_NoError(t, err)
+
+	sub, err := js.PullSubscribe("test", "test_consumer", nats.StartSequence(10))
+	require_NoError(t, err)
+
+	stream, err := s.gacc.lookupStream("TEST")
+	require_NoError(t, err)
+	consumer := stream.lookupConsumer("test_consumer")
+
+	func() {
+		consumer.mu.RLock()
+		defer consumer.mu.RUnlock()
+
+		require_Equal(t, consumer.dseq, 1)
+		require_Equal(t, consumer.sseq, 10)
+	}()
+
+	for i := 1; i <= 10; i++ {
+		_, err = js.Publish("test", []byte{byte(i)})
+		require_NoError(t, err)
+	}
+
+	msgs, err := sub.Fetch(1)
+	require_NoError(t, err)
+	require_Len(t, len(msgs), 1)
+	require_Equal(t, msgs[0].Data[0], 10)
+
+	require_NoError(t, msgs[0].AckSync())
+
+	func() {
+		consumer.mu.RLock()
+		defer consumer.mu.RUnlock()
+
+		require_Equal(t, consumer.dseq, 2)
+		require_Equal(t, consumer.adflr, 1)
+		require_Equal(t, consumer.sseq, 11)
+		require_Equal(t, consumer.asflr, 10)
+	}()
+}
+
+func TestJetStreamInterestStreamWithDuplicateMessages(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	cfg := &nats.StreamConfig{
+		Name:      "INTEREST",
+		Subjects:  []string{"interest"},
+		Replicas:  1,
+		Retention: nats.InterestPolicy,
+	}
+	_, err := js.AddStream(cfg)
+	require_NoError(t, err)
+
+	// Publishing the first time should give a sequence, even when there's no interest.
+	pa, err := js.Publish("interest", nil, nats.MsgId("dedupe"))
+	require_NoError(t, err)
+	require_Equal(t, pa.Sequence, 1)
+	require_Equal(t, pa.Duplicate, false)
+
+	// Publishing a duplicate with no interest should return the same sequence as above.
+	pa, err = js.Publish("interest", nil, nats.MsgId("dedupe"))
+	require_NoError(t, err)
+	require_Equal(t, pa.Sequence, 1)
+	require_Equal(t, pa.Duplicate, true)
+}
diff --git a/server/jwt_test.go b/server/jwt_test.go
index cb3a3391..d45b1dd5 100644
--- a/server/jwt_test.go
+++ b/server/jwt_test.go
@@ -5779,6 +5779,7 @@ func TestJWScopedSigningKeys(t *testing.T) {
 	signer.Key = aSignScopedPub
 	signer.Template.Pub.Deny.Add("denied")
 	signer.Template.Payload = 5
+	signer.Template.AllowedConnectionTypes.Add(jwt.ConnectionTypeStandard)
 	accClaim.SigningKeys.AddScopedSigner(signer)
 	accJwt := encodeClaim(t, accClaim, aExpPub)
 
@@ -5795,11 +5796,16 @@ func TestJWScopedSigningKeys(t *testing.T) {
 			type: full
 			dir: '%s'
 		}
+		websocket {
+			listen: 127.0.0.1:-1
+			no_tls: true
+		}
     `, ojwt, syspub, dirSrv)))
 	s, opts := RunServerWithConfig(cf)
 	defer s.Shutdown()
 
 	url := fmt.Sprintf("nats://%s:%d", opts.Host, opts.Port)
+	wsUrl := fmt.Sprintf("ws://%s:%d", opts.Websocket.Host, opts.Websocket.Port)
 	errChan := make(chan error, 1)
 	defer close(errChan)
 	awaitError := func(expected bool) {
@@ -5853,6 +5859,10 @@ func TestJWScopedSigningKeys(t *testing.T) {
 		nc.Flush()
 		awaitError(true)
 	})
+	t.Run("scoped-signing-key-allowed-conn-types", func(t *testing.T) {
+		_, err := nats.Connect(wsUrl, nats.UserCredentials(aScopedCreds))
+		require_Error(t, err)
+	})
 	t.Run("scoped-signing-key-reload", func(t *testing.T) {
 		reconChan := make(chan struct{}, 1)
 		defer close(reconChan)
@@ -5867,6 +5877,7 @@ func TestJWScopedSigningKeys(t *testing.T) {
 			nats.ReconnectHandler(func(conn *nats.Conn) {
 				reconChan <- struct{}{}
 			}),
+			nats.ReconnectWait(100*time.Millisecond),
 		)
 		defer nc.Close()
 		_, err := nc.ChanSubscribe("denied", msgChan)
@@ -5879,6 +5890,7 @@ func TestJWScopedSigningKeys(t *testing.T) {
 		// Alter scoped permissions and update
 		signer.Template.Payload = -1
 		signer.Template.Pub.Deny.Remove("denied")
+		signer.Template.AllowedConnectionTypes.Add(jwt.ConnectionTypeWebsocket)
 		accClaim.SigningKeys.AddScopedSigner(signer)
 		accUpdatedJwt := encodeClaim(t, accClaim, aExpPub)
 		if updateJwt(t, url, sysCreds, accUpdatedJwt, 1) != 1 {
@@ -5894,6 +5906,11 @@ func TestJWScopedSigningKeys(t *testing.T) {
 		msg := <-msgChan
 		require_Equal(t, string(msg.Data), "way.too.long.for.old.payload.limit")
 		require_Len(t, len(msgChan), 0)
+		nc.Close()
+		nc, err = nats.Connect(wsUrl, nats.UserCredentials(aScopedCreds))
+		require_NoError(t, err)
+		nc.Flush()
+		defer nc.Close()
 	})
 	require_Len(t, len(errChan), 0)
 }
diff --git a/server/norace_test.go b/server/norace_test.go
index 738abda4..3f8b7c87 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -36,7 +36,7 @@ import (
 	"reflect"
 	"runtime"
 	"runtime/debug"
-	"sort"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -10124,7 +10124,7 @@ func TestNoRaceConnectionObjectReleased(t *testing.T) {
 			// Don't fail the test since there is no guarantee that
 			// finalizers are invoked.
 			t.Logf("Got %v out of %v finalizers", len(clients), total)
-			sort.Strings(clients)
+			slices.Sort(clients)
 			for _, cs := range clients {
 				t.Logf("  => %s", cs)
 			}
@@ -10273,7 +10273,7 @@ func TestNoRaceWQAndMultiSubjectFilters(t *testing.T) {
 		}
 	}
 
-	sort.Slice(received, func(i, j int) bool { return received[i] < received[j] })
+	slices.Sort(received)
 
 	var pseq, gaps uint64
 	for _, seq := range received {
diff --git a/server/server_test.go b/server/server_test.go
index 3eb91eea..915d2a98 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -28,7 +28,7 @@ import (
 	"os"
 	"reflect"
 	"runtime"
-	"sort"
+	"slices"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -747,7 +747,7 @@ func TestProfilingNoTimeout(t *testing.T) {
 	if srv == nil {
 		t.Fatalf("Profiling server not set")
 	}
-	if srv.ReadTimeout != 0 {
+	if srv.ReadTimeout != time.Second*5 {
 		t.Fatalf("ReadTimeout should not be set, was set to %v", srv.ReadTimeout)
 	}
 	if srv.WriteTimeout != 0 {
@@ -1046,9 +1046,9 @@ func TestLameDuckModeInfo(t *testing.T) {
 		t.Helper()
 		var si *serverInfo
 		for i, ws := range []bool{false, true} {
-			sort.Strings(expected[i])
+			slices.Sort(expected[i])
 			si = getInfo(ws)
-			sort.Strings(si.ConnectURLs)
+			slices.Sort(si.ConnectURLs)
 			if !reflect.DeepEqual(expected[i], si.ConnectURLs) {
 				t.Fatalf("Expected %q, got %q", expected, si.ConnectURLs)
 			}
@@ -1997,16 +1997,20 @@ func TestServerRateLimitLogging(t *testing.T) {
 
 	s.RateLimitWarnf("Warning number 1")
 	s.RateLimitWarnf("Warning number 2")
+	s.rateLimitFormatWarnf("warning value %d", 1)
 	s.RateLimitWarnf("Warning number 1")
 	s.RateLimitWarnf("Warning number 2")
+	s.rateLimitFormatWarnf("warning value %d", 2)
 
 	checkLog := func(c1, c2 *client) {
 		t.Helper()
 
 		nb1 := "Warning number 1"
 		nb2 := "Warning number 2"
+		nbv := "warning value"
 		gotOne := 0
 		gotTwo := 0
+		gotFormat := 0
 		for done := false; !done; {
 			select {
 			case w := <-l.warn:
@@ -2014,6 +2018,8 @@ func TestServerRateLimitLogging(t *testing.T) {
 					gotOne++
 				} else if strings.Contains(w, nb2) {
 					gotTwo++
+				} else if strings.Contains(w, nbv) {
+					gotFormat++
 				}
 			case <-time.After(150 * time.Millisecond):
 				done = true
@@ -2025,27 +2031,39 @@ func TestServerRateLimitLogging(t *testing.T) {
 		if gotTwo != 1 {
 			t.Fatalf("Should have had only 1 warning for nb2, got %v", gotTwo)
 		}
+		if gotFormat != 1 {
+			t.Fatalf("Should have had only 1 warning for format, got %v", gotFormat)
+		}
 
 		// Wait for more than the expiration interval
 		time.Sleep(200 * time.Millisecond)
 		if c1 == nil {
 			s.RateLimitWarnf(nb1)
+			s.rateLimitFormatWarnf("warning value %d", 1)
 		} else {
 			c1.RateLimitWarnf(nb1)
 			c2.RateLimitWarnf(nb1)
+			c1.rateLimitFormatWarnf("warning value %d", 1)
 		}
 		gotOne = 0
+		gotFormat = 0
 		for {
 			select {
 			case w := <-l.warn:
 				if strings.Contains(w, nb1) {
 					gotOne++
+				} else if strings.Contains(w, nbv) {
+					gotFormat++
 				}
 			case <-time.After(200 * time.Millisecond):
 				if gotOne == 0 {
 					t.Fatalf("Warning was still suppressed")
 				} else if gotOne > 1 {
 					t.Fatalf("Should have had only 1 warning for nb1, got %v", gotOne)
+				} else if gotFormat == 0 {
+					t.Fatalf("Warning was still suppressed")
+				} else if gotFormat > 1 {
+					t.Fatalf("Should have had only 1 warning for format, got %v", gotFormat)
 				} else {
 					// OK! we are done
 					return
@@ -2087,8 +2105,10 @@ func TestServerRateLimitLogging(t *testing.T) {
 
 	c1.RateLimitWarnf("Warning number 1")
 	c1.RateLimitWarnf("Warning number 2")
+	c1.rateLimitFormatWarnf("warning value %d", 1)
 	c2.RateLimitWarnf("Warning number 1")
 	c2.RateLimitWarnf("Warning number 2")
+	c2.rateLimitFormatWarnf("warning value %d", 2)
 
 	checkLog(c1, c2)
 }