build(deps): bump github.com/nats-io/nats-server/v2

Bumps [github.com/nats-io/nats-server/v2](https://github.com/nats-io/nats-server) from 2.10.14 to 2.10.15. - [Release notes](https://github.com/nats-io/nats-server/releases) - [Changelog](https://github.com/nats-io/nats-server/blob/main/.goreleaser.yml) - [Commits](https://github.com/nats-io/nats-server/compare/v2.10.14...v2.10.15) --- updated-dependencies: - dependency-name: github.com/nats-io/nats-server/v2 dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com>
2026-07-19 20:14:17 -04:00 · 2024-05-17 06:21:22 +00:00
parent a090aab0f7
commit b85d2c39a9
31 changed files with 1321 additions and 870 deletions
--- a/go.mod
+++ b/go.mod
@@ -63,7 +63,7 @@ require (
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/mna/pigeon v1.2.1
 	github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826
-	github.com/nats-io/nats-server/v2 v2.10.14
+	github.com/nats-io/nats-server/v2 v2.10.15
 	github.com/nats-io/nats.go v1.34.1
 	github.com/oklog/run v1.1.0
 	github.com/olekukonko/tablewriter v0.0.5
@@ -282,7 +282,7 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/mschoch/smat v0.2.0 // indirect
-	github.com/nats-io/jwt/v2 v2.5.5 // indirect
+	github.com/nats-io/jwt/v2 v2.5.6 // indirect
 	github.com/nats-io/nkeys v0.4.7 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/nxadm/tail v1.4.8 // indirect
--- a/go.sum
+++ b/go.sum
@@ -1755,10 +1755,10 @@ github.com/mschoch/smat v0.2.0/go.mod h1:kc9mz7DoBKqDyiRL7VZN8KvXQMWeTaVnttLRXOl
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
 github.com/namedotcom/go v0.0.0-20180403034216-08470befbe04/go.mod h1:5sN+Lt1CaY4wsPvgQH/jsuJi4XO2ssZbdsIizr4CVC8=
-github.com/nats-io/jwt/v2 v2.5.5 h1:ROfXb50elFq5c9+1ztaUbdlrArNFl2+fQWP6B8HGEq4=
-github.com/nats-io/jwt/v2 v2.5.5/go.mod h1:ZdWS1nZa6WMZfFwwgpEaqBV8EPGVgOTDHN/wTbz0Y5A=
-github.com/nats-io/nats-server/v2 v2.10.14 h1:98gPJFOAO2vLdM0gogh8GAiHghwErrSLhugIqzRC+tk=
-github.com/nats-io/nats-server/v2 v2.10.14/go.mod h1:a0TwOVBJZz6Hwv7JH2E4ONdpyFk9do0C18TEwxnHdRk=
+github.com/nats-io/jwt/v2 v2.5.6 h1:Cp618+z4q042sWqHiSoIHFT08OZtAskui0hTmRfmGGQ=
+github.com/nats-io/jwt/v2 v2.5.6/go.mod h1:ZdWS1nZa6WMZfFwwgpEaqBV8EPGVgOTDHN/wTbz0Y5A=
+github.com/nats-io/nats-server/v2 v2.10.15 h1:O/l+ZT91ltMiiRJKjWLQJcGg7ypzjlb/bC5bFIRVw3M=
+github.com/nats-io/nats-server/v2 v2.10.15/go.mod h1:ul+pGt5I7e4U+nI09ZFDG4vqM+6Ce2Tou7UbVSnLiIw=
 github.com/nats-io/nats.go v1.34.1 h1:syWey5xaNHZgicYBemv0nohUPPmaLteiBEUT6Q5+F/4=
 github.com/nats-io/nats.go v1.34.1/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
 github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
--- a/vendor/github.com/nats-io/jwt/v2/signingkeys.go
+++ b/vendor/github.com/nats-io/jwt/v2/signingkeys.go
@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 The NATS Authors
+ * Copyright 2020-2024 The NATS Authors
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
@@ -67,10 +67,11 @@ func (t *ScopeType) UnmarshalJSON(b []byte) error {
 }

 type UserScope struct {
-	Kind     ScopeType            `json:"kind"`
-	Key      string               `json:"key"`
-	Role     string               `json:"role"`
-	Template UserPermissionLimits `json:"template"`
+	Kind        ScopeType            `json:"kind"`
+	Key         string               `json:"key"`
+	Role        string               `json:"role"`
+	Template    UserPermissionLimits `json:"template"`
+	Description string               `json:"description"`
 }

 func NewUserScope() *UserScope {
--- a/vendor/github.com/nats-io/nats-server/v2/conf/lex.go
+++ b/vendor/github.com/nats-io/nats-server/v2/conf/lex.go
@@ -1,4 +1,4 @@
-// Copyright 2013-2018 The NATS Authors
+// Copyright 2013-2024 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -156,7 +156,6 @@ func (lx *lexer) pop() stateFn {

 func (lx *lexer) emit(typ itemType) {
 	val := strings.Join(lx.stringParts, "") + lx.input[lx.start:lx.pos]
-
 	// Position of item in line where it started.
 	pos := lx.pos - lx.ilstart - len(val)
 	lx.items <- item{typ, val, lx.line, pos}
@@ -263,7 +262,8 @@ func lexTop(lx *lexer) stateFn {

 	switch r {
 	case topOptStart:
-		return lexSkip(lx, lexTop)
+		lx.push(lexTop)
+		return lexSkip(lx, lexBlockStart)
 	case commentHashStart:
 		lx.push(lexTop)
 		return lexCommentStart
@@ -318,6 +318,105 @@ func lexTopValueEnd(lx *lexer) stateFn {
 		"comment or EOF, but got '%v' instead.", r)
 }

+func lexBlockStart(lx *lexer) stateFn {
+	r := lx.next()
+	if unicode.IsSpace(r) {
+		return lexSkip(lx, lexBlockStart)
+	}
+
+	switch r {
+	case topOptStart:
+		lx.push(lexBlockEnd)
+		return lexSkip(lx, lexBlockStart)
+	case topOptTerm:
+		lx.ignore()
+		return lx.pop()
+	case commentHashStart:
+		lx.push(lexBlockEnd)
+		return lexCommentStart
+	case commentSlashStart:
+		rn := lx.next()
+		if rn == commentSlashStart {
+			lx.push(lexBlockEnd)
+			return lexCommentStart
+		}
+		lx.backup()
+		fallthrough
+	case eof:
+		if lx.pos > lx.start {
+			return lx.errorf("Unexpected EOF.")
+		}
+		lx.emit(itemEOF)
+		return nil
+	}
+
+	// At this point, the only valid item can be a key, so we back up
+	// and let the key lexer do the rest.
+	lx.backup()
+	lx.push(lexBlockValueEnd)
+	return lexKeyStart
+}
+
+// lexBlockValueEnd is entered whenever a block-level value has been consumed.
+// It must see only whitespace, and will turn back to lexBlockStart upon a new line.
+// If it sees EOF, it will quit the lexer successfully.
+func lexBlockValueEnd(lx *lexer) stateFn {
+	r := lx.next()
+	switch {
+	case r == commentHashStart:
+		// a comment will read to a new line for us.
+		lx.push(lexBlockValueEnd)
+		return lexCommentStart
+	case r == commentSlashStart:
+		rn := lx.next()
+		if rn == commentSlashStart {
+			lx.push(lexBlockValueEnd)
+			return lexCommentStart
+		}
+		lx.backup()
+		fallthrough
+	case isWhitespace(r):
+		return lexBlockValueEnd
+	case isNL(r) || r == optValTerm || r == topOptValTerm:
+		lx.ignore()
+		return lexBlockStart
+	case r == topOptTerm:
+		lx.backup()
+		return lexBlockEnd
+	}
+	return lx.errorf("Expected a block-level value to end with a new line, "+
+		"comment or EOF, but got '%v' instead.", r)
+}
+
+// lexBlockEnd is entered whenever a block-level value has been consumed.
+// It must see only whitespace, and will turn back to lexTop upon a "}".
+func lexBlockEnd(lx *lexer) stateFn {
+	r := lx.next()
+	switch {
+	case r == commentHashStart:
+		// a comment will read to a new line for us.
+		lx.push(lexBlockStart)
+		return lexCommentStart
+	case r == commentSlashStart:
+		rn := lx.next()
+		if rn == commentSlashStart {
+			lx.push(lexBlockStart)
+			return lexCommentStart
+		}
+		lx.backup()
+		fallthrough
+	case isNL(r) || isWhitespace(r):
+		return lexBlockEnd
+	case r == optValTerm || r == topOptValTerm:
+		lx.ignore()
+		return lexBlockStart
+	case r == topOptTerm:
+		lx.ignore()
+		return lx.pop()
+	}
+	return lx.errorf("Expected a block-level to end with a '}', but got '%v' instead.", r)
+}
+
 // lexKeyStart consumes a key name up until the first non-whitespace character.
 // lexKeyStart will ignore whitespace. It will also eat enclosing quotes.
 func lexKeyStart(lx *lexer) stateFn {
--- a/vendor/github.com/nats-io/nats-server/v2/server/accounts.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/accounts.go
@@ -82,7 +82,7 @@ type Account struct {
 	js           *jsAccount
 	jsLimits     map[string]JetStreamAccountLimits
 	limits
-	expired      bool
+	expired      atomic.Bool
 	incomplete   bool
 	signingKeys  map[string]jwt.Scope
 	extAuth      *jwt.ExternalAuthorization
@@ -759,7 +759,7 @@ func (a *Account) selectMappedSubject(dest string) (string, bool) {
 		return dest, false
 	}

-	a.mu.Lock()
+	a.mu.RLock()
 	// In case we have to tokenize for subset matching.
 	tsa := [32]string{}
 	tts := tsa[:0]
@@ -790,7 +790,7 @@ func (a *Account) selectMappedSubject(dest string) (string, bool) {
 	}

 	if m == nil {
-		a.mu.Unlock()
+		a.mu.RUnlock()
 		return dest, false
 	}

@@ -829,7 +829,7 @@ func (a *Account) selectMappedSubject(dest string) (string, bool) {
 		}
 	}

-	a.mu.Unlock()
+	a.mu.RUnlock()
 	return ndest, true
 }

@@ -2189,7 +2189,7 @@ const (
 // This is where all service export responses are handled.
 func (a *Account) processServiceImportResponse(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
 	a.mu.RLock()
-	if a.expired || len(a.exports.responses) == 0 {
+	if a.expired.Load() || len(a.exports.responses) == 0 {
 		a.mu.RUnlock()
 		return
 	}
@@ -2664,7 +2664,7 @@ func (a *Account) getWildcardServiceExport(from string) *serviceExport {
 // These are import stream specific versions for when an activation expires.
 func (a *Account) streamActivationExpired(exportAcc *Account, subject string) {
 	a.mu.RLock()
-	if a.expired || a.imports.streams == nil {
+	if a.expired.Load() || a.imports.streams == nil {
 		a.mu.RUnlock()
 		return
 	}
@@ -2699,7 +2699,7 @@ func (a *Account) streamActivationExpired(exportAcc *Account, subject string) {
 // These are import service specific versions for when an activation expires.
 func (a *Account) serviceActivationExpired(subject string) {
 	a.mu.RLock()
-	if a.expired || a.imports.services == nil {
+	if a.expired.Load() || a.imports.services == nil {
 		a.mu.RUnlock()
 		return
 	}
@@ -2960,18 +2960,13 @@ func (a *Account) checkServiceImportAuthorizedNoLock(account *Account, subject s

 // IsExpired returns expiration status.
 func (a *Account) IsExpired() bool {
-	a.mu.RLock()
-	exp := a.expired
-	a.mu.RUnlock()
-	return exp
+	return a.expired.Load()
 }

 // Called when an account has expired.
 func (a *Account) expiredTimeout() {
 	// Mark expired first.
-	a.mu.Lock()
-	a.expired = true
-	a.mu.Unlock()
+	a.expired.Store(true)

 	// Collect the clients and expire them.
 	cs := a.getClients()
@@ -3016,17 +3011,17 @@ func (a *Account) checkExpiration(claims *jwt.ClaimsData) {

 	a.clearExpirationTimer()
 	if claims.Expires == 0 {
-		a.expired = false
+		a.expired.Store(false)
 		return
 	}
 	tn := time.Now().Unix()
 	if claims.Expires <= tn {
-		a.expired = true
+		a.expired.Store(true)
 		return
 	}
 	expiresAt := time.Duration(claims.Expires - tn)
 	a.setExpirationTimer(expiresAt * time.Second)
-	a.expired = false
+	a.expired.Store(false)
 }

 // hasIssuer returns true if the issuer matches the account
@@ -3072,9 +3067,9 @@ func (s *Server) SetAccountResolver(ar AccountResolver) {

 // AccountResolver returns the registered account resolver.
 func (s *Server) AccountResolver() AccountResolver {
-	s.mu.Lock()
+	s.mu.RLock()
 	ar := s.accResolver
-	s.mu.Unlock()
+	s.mu.RUnlock()
 	return ar
 }

--- a/vendor/github.com/nats-io/nats-server/v2/server/client.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/client.go
@@ -474,8 +474,8 @@ func (rcf readCacheFlag) isSet(c readCacheFlag) bool {
 }

 const (
-	defaultMaxPerAccountCacheSize   = 4096
-	defaultPrunePerAccountCacheSize = 256
+	defaultMaxPerAccountCacheSize   = 8192
+	defaultPrunePerAccountCacheSize = 1024
 	defaultClosedSubsCheckInterval  = 5 * time.Minute
 )

@@ -2349,24 +2349,11 @@ func (c *client) generateClientInfoJSON(info Info) []byte {
 	info.MaxPayload = c.mpay
 	if c.isWebsocket() {
 		info.ClientConnectURLs = info.WSConnectURLs
-		if c.srv != nil { // Otherwise lame duck info can panic
-			c.srv.websocket.mu.RLock()
-			info.TLSAvailable = c.srv.websocket.tls
-			if c.srv.websocket.tls && c.srv.websocket.server != nil {
-				if tc := c.srv.websocket.server.TLSConfig; tc != nil {
-					info.TLSRequired = !tc.InsecureSkipVerify
-				}
-			}
-			if c.srv.websocket.listener != nil {
-				laddr := c.srv.websocket.listener.Addr().String()
-				if h, p, err := net.SplitHostPort(laddr); err == nil {
-					if p, err := strconv.Atoi(p); err == nil {
-						info.Host = h
-						info.Port = p
-					}
-				}
-			}
-			c.srv.websocket.mu.RUnlock()
+		// Otherwise lame duck info can panic
+		if c.srv != nil {
+			ws := &c.srv.websocket
+			info.TLSAvailable, info.TLSRequired = ws.tls, ws.tls
+			info.Host, info.Port = ws.host, ws.port
 		}
 	}
 	info.WSConnectURLs = nil
@@ -4909,6 +4896,23 @@ func adjustPingInterval(kind int, d time.Duration) time.Duration {
 	return d
 }

+// This is used when a connection cannot yet start to send PINGs because
+// the remote would not be able to handle them (case of compression,
+// or outbound gateway, etc...), but we still want to close the connection
+// if the timer has not been reset by the time we reach the time equivalent
+// to have sent the max number of pings.
+//
+// Lock should be held
+func (c *client) watchForStaleConnection(pingInterval time.Duration, pingMax int) {
+	c.ping.tmr = time.AfterFunc(pingInterval*time.Duration(pingMax+1), func() {
+		c.mu.Lock()
+		c.Debugf("Stale Client Connection - Closing")
+		c.enqueueProto([]byte(fmt.Sprintf(errProto, "Stale Connection")))
+		c.mu.Unlock()
+		c.closeConnection(StaleConnection)
+	})
+}
+
 // Lock should be held
 func (c *client) setPingTimer() {
 	if c.srv == nil {
@@ -5436,13 +5440,13 @@ func (c *client) getAccAndResultFromCache() (*Account, *SublistResult) {
 		// Match against the account sublist.
 		r = sl.Match(string(c.pa.subject))

-		// Store in our cache
-		c.in.pacache[string(c.pa.pacache)] = &perAccountCache{acc, r, atomic.LoadUint64(&sl.genid)}
-
 		// Check if we need to prune.
-		if len(c.in.pacache) > maxPerAccountCacheSize {
+		if len(c.in.pacache) >= maxPerAccountCacheSize {
 			c.prunePerAccountCache()
 		}
+
+		// Store in our cache,make sure to do so after we prune.
+		c.in.pacache[string(c.pa.pacache)] = &perAccountCache{acc, r, atomic.LoadUint64(&sl.genid)}
 	}
 	return acc, r
 }
--- a/vendor/github.com/nats-io/nats-server/v2/server/const.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/const.go
@@ -41,7 +41,7 @@ var (

 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.10.14"
+	VERSION = "2.10.15"

 	// PROTO is the currently supported protocol.
 	// 0 was the original
--- a/vendor/github.com/nats-io/nats-server/v2/server/consumer.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/consumer.go
@@ -345,6 +345,7 @@ type consumer struct {
 	rdq               []uint64
 	rdqi              avl.SequenceSet
 	rdc               map[uint64]uint64
+	replies           map[uint64]string
 	maxdc             uint64
 	waiting           *waitQueue
 	cfg               ConsumerConfig
@@ -423,8 +424,7 @@ const (
 	// JsDeleteWaitTimeDefault is the default amount of time we will wait for non-durable
 	// consumers to be in an inactive state before deleting them.
 	JsDeleteWaitTimeDefault = 5 * time.Second
-	// JsFlowControlMaxPending specifies default pending bytes during flow control that can be
-	// outstanding.
+	// JsFlowControlMaxPending specifies default pending bytes during flow control that can be outstanding.
 	JsFlowControlMaxPending = 32 * 1024 * 1024
 	// JsDefaultMaxAckPending is set for consumers with explicit ack that do not set the max ack pending.
 	JsDefaultMaxAckPending = 1000
@@ -1143,23 +1143,19 @@ func (o *consumer) clearNode() {

 // IsLeader will return if we are the current leader.
 func (o *consumer) IsLeader() bool {
-	o.mu.RLock()
-	defer o.mu.RUnlock()
 	return o.isLeader()
 }

 // Lock should be held.
 func (o *consumer) isLeader() bool {
-	if o.node != nil {
-		return o.node.Leader()
-	}
-	return true
+	return o.leader.Load()
 }

 func (o *consumer) setLeader(isLeader bool) {
 	o.mu.RLock()
 	mset, closed := o.mset, o.closed
 	movingToClustered := o.node != nil && o.pch == nil
+	movingToNonClustered := o.node == nil && o.pch != nil
 	wasLeader := o.leader.Swap(isLeader)
 	o.mu.RUnlock()

@@ -1183,6 +1179,17 @@ func (o *consumer) setLeader(isLeader bool) {
 					}
 				}
 				o.mu.Unlock()
+			} else if movingToNonClustered {
+				// We are moving from clustered to non-clustered now.
+				// Set pch to nil so if we scale back up we will recreate the loopAndForward from above.
+				o.mu.Lock()
+				pch := o.pch
+				o.pch = nil
+				select {
+				case pch <- struct{}{}:
+				default:
+				}
+				o.mu.Unlock()
 			}
 			return
 		}
@@ -1191,9 +1198,6 @@ func (o *consumer) setLeader(isLeader bool) {
 		s, jsa, stream, lseq := mset.srv, mset.jsa, mset.cfg.Name, mset.lseq
 		mset.mu.RUnlock()

-		// Register as a leader with our parent stream.
-		mset.setConsumerAsLeader(o)
-
 		o.mu.Lock()
 		o.rdq = nil
 		o.rdqi.Empty()
@@ -1365,13 +1369,10 @@ func (o *consumer) setLeader(isLeader bool) {
 		// If we were the leader make sure to drain queued up acks.
 		if wasLeader {
 			o.ackMsgs.drain()
+			// Also remove any pending replies since we should not be the one to respond at this point.
+			o.replies = nil
 		}
 		o.mu.Unlock()
-
-		// Unregister as a leader with our parent stream.
-		if mset != nil {
-			mset.removeConsumerAsLeader(o)
-		}
 	}
 }

@@ -1480,15 +1481,15 @@ func (o *consumer) setCreatedTime(created time.Time) {
 // that, but in the absence of local interest and presence of gateways or service imports we need
 // to check those as well.
 func (o *consumer) hasDeliveryInterest(localInterest bool) bool {
-	o.mu.Lock()
+	o.mu.RLock()
 	mset := o.mset
 	if mset == nil {
-		o.mu.Unlock()
+		o.mu.RUnlock()
 		return false
 	}
 	acc := o.acc
 	deliver := o.cfg.DeliverSubject
-	o.mu.Unlock()
+	o.mu.RUnlock()

 	if localInterest {
 		return true
@@ -1920,6 +1921,13 @@ func (o *consumer) updateConfig(cfg *ConsumerConfig) error {
 	// Allowed but considered no-op, [Description, SampleFrequency, MaxWaiting, HeadersOnly]
 	o.cfg = *cfg

+	// Cleanup messages that lost interest.
+	if o.retention == InterestPolicy {
+		o.mu.Unlock()
+		o.cleanupNoInterestMessages(o.mset, false)
+		o.mu.Lock()
+	}
+
 	// Re-calculate num pending on update.
 	o.streamNumPending()

@@ -1962,9 +1970,9 @@ func configsEqualSansDelivery(a, b ConsumerConfig) bool {

 // Helper to send a reply to an ack.
 func (o *consumer) sendAckReply(subj string) {
-	o.mu.Lock()
-	defer o.mu.Unlock()
-	o.sendAdvisory(subj, nil)
+	o.mu.RLock()
+	defer o.mu.RUnlock()
+	o.outq.sendMsg(subj, nil)
 }

 type jsAckMsg struct {
@@ -2022,9 +2030,11 @@ func (o *consumer) processAck(subject, reply string, hdr int, rmsg []byte) {

 	switch {
 	case len(msg) == 0, bytes.Equal(msg, AckAck), bytes.Equal(msg, AckOK):
-		o.processAckMsg(sseq, dseq, dc, true)
+		o.processAckMsg(sseq, dseq, dc, reply, true)
+		// We handle replies for acks in updateAcks
+		skipAckReply = true
 	case bytes.HasPrefix(msg, AckNext):
-		o.processAckMsg(sseq, dseq, dc, true)
+		o.processAckMsg(sseq, dseq, dc, _EMPTY_, true)
 		o.processNextMsgRequest(reply, msg[len(AckNext):])
 		skipAckReply = true
 	case bytes.HasPrefix(msg, AckNak):
@@ -2036,7 +2046,9 @@ func (o *consumer) processAck(subject, reply string, hdr int, rmsg []byte) {
 		if buf := msg[len(AckTerm):]; len(buf) > 0 {
 			reason = string(bytes.TrimSpace(buf))
 		}
-		o.processTerm(sseq, dseq, dc, reason)
+		o.processTerm(sseq, dseq, dc, reason, reply)
+		// We handle replies for acks in updateAcks
+		skipAckReply = true
 	}

 	// Ack the ack if requested.
@@ -2071,6 +2083,13 @@ func (o *consumer) updateSkipped(seq uint64) {
 }

 func (o *consumer) loopAndForwardProposals(qch chan struct{}) {
+	// On exit make sure we nil out pch.
+	defer func() {
+		o.mu.Lock()
+		o.pch = nil
+		o.mu.Unlock()
+	}()
+
 	o.mu.RLock()
 	node, pch := o.node, o.pch
 	o.mu.RUnlock()
@@ -2081,7 +2100,7 @@ func (o *consumer) loopAndForwardProposals(qch chan struct{}) {

 	forwardProposals := func() error {
 		o.mu.Lock()
-		if o.node != node || node.State() != Leader {
+		if o.node == nil || o.node.State() != Leader {
 			o.mu.Unlock()
 			return errors.New("no longer leader")
 		}
@@ -2168,8 +2187,17 @@ func (o *consumer) updateDelivered(dseq, sseq, dc uint64, ts int64) {
 	o.ldt = time.Now()
 }

+// Used to remember a pending ack reply in a replicated consumer.
 // Lock should be held.
-func (o *consumer) updateAcks(dseq, sseq uint64) {
+func (o *consumer) addAckReply(sseq uint64, reply string) {
+	if o.replies == nil {
+		o.replies = make(map[uint64]string)
+	}
+	o.replies[sseq] = reply
+}
+
+// Lock should be held.
+func (o *consumer) updateAcks(dseq, sseq uint64, reply string) {
 	if o.node != nil {
 		// Inline for now, use variable compression.
 		var b [2*binary.MaxVarintLen64 + 1]byte
@@ -2178,8 +2206,15 @@ func (o *consumer) updateAcks(dseq, sseq uint64) {
 		n += binary.PutUvarint(b[n:], dseq)
 		n += binary.PutUvarint(b[n:], sseq)
 		o.propose(b[:n])
+		if reply != _EMPTY_ {
+			o.addAckReply(sseq, reply)
+		}
 	} else if o.store != nil {
 		o.store.UpdateAcks(dseq, sseq)
+		if reply != _EMPTY_ {
+			// Already locked so send direct.
+			o.outq.sendMsg(reply, nil)
+		}
 	}
 	// Update activity.
 	o.lat = time.Now()
@@ -2369,9 +2404,9 @@ func (o *consumer) processNak(sseq, dseq, dc uint64, nak []byte) {
 }

 // Process a TERM
-func (o *consumer) processTerm(sseq, dseq, dc uint64, reason string) {
+func (o *consumer) processTerm(sseq, dseq, dc uint64, reason, reply string) {
 	// Treat like an ack to suppress redelivery.
-	o.processAckMsg(sseq, dseq, dc, false)
+	o.processAckMsg(sseq, dseq, dc, reply, false)

 	o.mu.Lock()
 	defer o.mu.Unlock()
@@ -2474,6 +2509,7 @@ func (o *consumer) applyState(state *ConsumerState) {
 		// This is on startup or leader change. We want to check pending
 		// sooner in case there are inconsistencies etc. Pick between 500ms - 1.5s
 		delay := 500*time.Millisecond + time.Duration(rand.Int63n(1000))*time.Millisecond
+
 		// If normal is lower than this just use that.
 		if o.cfg.AckWait < delay {
 			delay = o.ackWait(0)
@@ -2699,7 +2735,7 @@ func (o *consumer) sampleAck(sseq, dseq, dc uint64) {
 	o.sendAdvisory(o.ackEventT, j)
 }

-func (o *consumer) processAckMsg(sseq, dseq, dc uint64, doSample bool) {
+func (o *consumer) processAckMsg(sseq, dseq, dc uint64, reply string, doSample bool) {
 	o.mu.Lock()
 	if o.closed {
 		o.mu.Unlock()
@@ -2739,12 +2775,12 @@ func (o *consumer) processAckMsg(sseq, dseq, dc uint64, doSample bool) {
 					}
 				}
 			}
-			// If nothing left set to current delivered.
+			// If nothing left set consumer to current delivered.
+			// Do not update stream.
 			if len(o.pending) == 0 {
-				o.adflr, o.asflr = o.dseq-1, o.sseq-1
+				o.adflr = o.dseq - 1
 			}
 		}
-		// We do these regardless.
 		delete(o.rdc, sseq)
 		o.removeFromRedeliverQueue(sseq)
 	case AckAll:
@@ -2770,7 +2806,7 @@ func (o *consumer) processAckMsg(sseq, dseq, dc uint64, doSample bool) {
 	}

 	// Update underlying store.
-	o.updateAcks(dseq, sseq)
+	o.updateAcks(dseq, sseq, reply)

 	clustered := o.node != nil

@@ -3356,15 +3392,9 @@ func (o *consumer) incDeliveryCount(sseq uint64) uint64 {
 // Lock should be held.
 func (o *consumer) decDeliveryCount(sseq uint64) {
 	if o.rdc == nil {
-		return
-	}
-	if dc, ok := o.rdc[sseq]; ok {
-		if dc == 1 {
-			delete(o.rdc, sseq)
-		} else {
-			o.rdc[sseq] -= 1
-		}
+		o.rdc = make(map[uint64]uint64)
 	}
+	o.rdc[sseq] -= 1
 }

 // send a delivery exceeded advisory.
@@ -3512,21 +3542,23 @@ func (o *consumer) getNextMsg() (*jsPubMsg, uint64, error) {

 	// Grab next message applicable to us.
 	// We will unlock here in case lots of contention, e.g. WQ.
+	filters, subjf, fseq := o.filters, o.subjf, o.sseq
 	o.mu.Unlock()
 	// Check if we are multi-filtered or not.
-	if o.filters != nil {
-		sm, sseq, err = store.LoadNextMsgMulti(o.filters, o.sseq, &pmsg.StoreMsg)
-	} else if o.subjf != nil { // Means single filtered subject since o.filters means > 1.
-		filter, wc := o.subjf[0].subject, o.subjf[0].hasWildcard
-		sm, sseq, err = store.LoadNextMsg(filter, wc, o.sseq, &pmsg.StoreMsg)
+	if filters != nil {
+		sm, sseq, err = store.LoadNextMsgMulti(filters, fseq, &pmsg.StoreMsg)
+	} else if subjf != nil { // Means single filtered subject since o.filters means > 1.
+		filter, wc := subjf[0].subject, subjf[0].hasWildcard
+		sm, sseq, err = store.LoadNextMsg(filter, wc, fseq, &pmsg.StoreMsg)
 	} else {
 		// No filter here.
-		sm, sseq, err = store.LoadNextMsg(_EMPTY_, false, o.sseq, &pmsg.StoreMsg)
+		sm, sseq, err = store.LoadNextMsg(_EMPTY_, false, fseq, &pmsg.StoreMsg)
 	}
 	if sm == nil {
 		pmsg.returnToPool()
 		pmsg = nil
 	}
+	// Re-acquire lock.
 	o.mu.Lock()
 	// Check if we should move our o.sseq.
 	if sseq >= o.sseq {
@@ -3666,7 +3698,7 @@ func (o *consumer) checkAckFloor() {
 			o.mu.RUnlock()
 			// If it was pending for us, get rid of it.
 			if isPending {
-				o.processTerm(seq, p.Sequence, rdc, ackTermLimitsReason)
+				o.processTerm(seq, p.Sequence, rdc, ackTermLimitsReason, _EMPTY_)
 			}
 		}
 	} else if numPending > 0 {
@@ -3691,7 +3723,7 @@ func (o *consumer) checkAckFloor() {

 		for i := 0; i < len(toTerm); i += 3 {
 			seq, dseq, rdc := toTerm[i], toTerm[i+1], toTerm[i+2]
-			o.processTerm(seq, dseq, rdc, ackTermLimitsReason)
+			o.processTerm(seq, dseq, rdc, ackTermLimitsReason, _EMPTY_)
 		}
 	}

@@ -3705,9 +3737,12 @@ func (o *consumer) checkAckFloor() {
 	}

 	// If we are here, and this should be rare, we still are off with our ack floor.
+	// We will make sure we are not doing un-necessary work here if only off by a bit
+	// since this could be normal for a high activity wq or stream.
 	// We will set it explicitly to 1 behind our current lowest in pending, or if
 	// pending is empty, to our current delivered -1.
-	if o.asflr < ss.FirstSeq-1 {
+	const minOffThreshold = 50
+	if o.asflr < ss.FirstSeq-minOffThreshold {
 		var psseq, pdseq uint64
 		for seq, p := range o.pending {
 			if psseq == 0 || seq < psseq {
@@ -3719,7 +3754,7 @@ func (o *consumer) checkAckFloor() {
 			psseq, pdseq = o.sseq-1, o.dseq-1
 			// If still not adjusted.
 			if psseq < ss.FirstSeq-1 {
-				psseq, pdseq = ss.FirstSeq-1, ss.FirstSeq-1
+				psseq = ss.FirstSeq - 1
 			}
 		} else {
 			// Since this was set via the pending, we should not include
@@ -3735,6 +3770,7 @@ func (o *consumer) processInboundAcks(qch chan struct{}) {
 	o.mu.RLock()
 	s, mset := o.srv, o.mset
 	hasInactiveThresh := o.cfg.InactiveThreshold > 0
+
 	o.mu.RUnlock()

 	if s == nil || mset == nil {
@@ -3873,7 +3909,7 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 		o.mu.Lock()

 		// consumer is closed when mset is set to nil.
-		if o.mset == nil {
+		if o.closed || o.mset == nil {
 			o.mu.Unlock()
 			return
 		}
@@ -3948,9 +3984,17 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 			}
 		} else {
 			// We will redo this one as long as this is not a redelivery.
-			if dc == 1 {
+			// Need to also test that this is not going backwards since if
+			// we fail to deliver we can end up here from rdq but we do not
+			// want to decrement o.sseq if that is the case.
+			if dc == 1 && pmsg.seq == o.sseq-1 {
 				o.sseq--
 				o.npc++
+			} else if !o.onRedeliverQueue(pmsg.seq) {
+				// We are not on the rdq so decrement the delivery count
+				// and add it back.
+				o.decDeliveryCount(pmsg.seq)
+				o.addToRedeliverQueue(pmsg.seq)
 			}
 			pmsg.returnToPool()
 			goto waitForMsgs
@@ -3995,7 +4039,8 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {

 		// If given request fulfilled batch size, but there are still pending bytes, send information about it.
 		if wrn <= 0 && wrb > 0 {
-			o.outq.send(newJSPubMsg(dsubj, _EMPTY_, _EMPTY_, fmt.Appendf(nil, JsPullRequestRemainingBytesT, JSPullRequestPendingMsgs, wrn, JSPullRequestPendingBytes, wrb), nil, nil, 0))
+			msg := fmt.Appendf(nil, JsPullRequestRemainingBytesT, JSPullRequestPendingMsgs, wrn, JSPullRequestPendingBytes, wrb)
+			o.outq.send(newJSPubMsg(dsubj, _EMPTY_, _EMPTY_, msg, nil, nil, 0))
 		}
 		// Reset our idle heartbeat timer if set.
 		if hb != nil {
@@ -4260,7 +4305,7 @@ func (o *consumer) deliverMsg(dsubj, ackReply string, pmsg *jsPubMsg, dc uint64,
 		if o.node == nil || o.cfg.Direct {
 			mset.ackq.push(seq)
 		} else {
-			o.updateAcks(dseq, seq)
+			o.updateAcks(dseq, seq, _EMPTY_)
 		}
 	}
 }
@@ -4386,6 +4431,9 @@ func (o *consumer) didNotDeliver(seq uint64, subj string) {
 		o.active = false
 		checkDeliveryInterest = true
 	} else if o.pending != nil {
+		// Good chance we did not deliver because no interest so force a check.
+		o.processWaiting(false)
+		// If it is still there credit it back.
 		o.creditWaitingRequest(subj)
 		// pull mode and we have pending.
 		if _, ok := o.pending[seq]; ok {
@@ -4394,7 +4442,9 @@ func (o *consumer) didNotDeliver(seq uint64, subj string) {
 			// we know it was not delivered
 			if !o.onRedeliverQueue(seq) {
 				o.addToRedeliverQueue(seq)
-				o.signalNewMessages()
+				if !o.waiting.isEmpty() {
+					o.signalNewMessages()
+				}
 			}
 		}
 	}
@@ -4523,6 +4573,10 @@ func (o *consumer) checkPending() {
 		if len(o.cfg.BackOff) > 0 {
 			// This is ok even if o.rdc is nil, we would get dc == 0, which is what we want.
 			dc := int(o.rdc[seq])
+			if dc < 0 {
+				// Prevent consumer backoff from going backwards.
+				dc = 0
+			}
 			// This will be the index for the next backoff, will set to last element if needed.
 			nbi := dc + 1
 			if dc+1 >= len(o.cfg.BackOff) {
@@ -5075,7 +5129,6 @@ func (o *consumer) stopWithFlags(dflag, sdflag, doSignal, advisory bool) error {
 	if dflag {
 		ca = o.ca
 	}
-	sigSubs := o.sigSubs
 	js := o.js
 	o.mu.Unlock()

@@ -5092,45 +5145,15 @@ func (o *consumer) stopWithFlags(dflag, sdflag, doSignal, advisory bool) error {

 	var rp RetentionPolicy
 	if mset != nil {
-		if len(sigSubs) > 0 {
-			mset.removeConsumerAsLeader(o)
-		}
 		mset.mu.Lock()
 		mset.removeConsumer(o)
 		rp = mset.cfg.Retention
 		mset.mu.Unlock()
 	}

-	// We need to optionally remove all messages since we are interest based retention.
-	// We will do this consistently on all replicas. Note that if in clustered mode the
-	// non-leader consumers will need to restore state first.
+	// Cleanup messages that lost interest.
 	if dflag && rp == InterestPolicy {
-		state := mset.state()
-		stop := state.LastSeq
-		o.mu.Lock()
-		if !o.isLeader() {
-			o.readStoredState(stop)
-		}
-		start := o.asflr
-		o.mu.Unlock()
-		// Make sure we start at worst with first sequence in the stream.
-		if start < state.FirstSeq {
-			start = state.FirstSeq
-		}
-
-		var rmseqs []uint64
-		mset.mu.Lock()
-		for seq := start; seq <= stop; seq++ {
-			if mset.noInterest(seq, o) {
-				rmseqs = append(rmseqs, seq)
-			}
-		}
-		mset.mu.Unlock()
-
-		// These can be removed.
-		for _, seq := range rmseqs {
-			mset.store.RemoveMsg(seq)
-		}
+		o.cleanupNoInterestMessages(mset, true)
 	}

 	// Cluster cleanup.
@@ -5173,6 +5196,46 @@ func (o *consumer) stopWithFlags(dflag, sdflag, doSignal, advisory bool) error {
 	return err
 }

+// We need to optionally remove all messages since we are interest based retention.
+// We will do this consistently on all replicas. Note that if in clustered mode the non-leader
+// consumers will need to restore state first.
+// ignoreInterest marks whether the consumer should be ignored when determining interest.
+// No lock held on entry.
+func (o *consumer) cleanupNoInterestMessages(mset *stream, ignoreInterest bool) {
+	state := mset.state()
+	stop := state.LastSeq
+	o.mu.Lock()
+	if !o.isLeader() {
+		o.readStoredState(stop)
+	}
+	start := o.asflr
+	o.mu.Unlock()
+	// Make sure we start at worst with first sequence in the stream.
+	if start < state.FirstSeq {
+		start = state.FirstSeq
+	}
+
+	// Consumer's interests are ignored by default. If we should not ignore interest, unset.
+	co := o
+	if !ignoreInterest {
+		co = nil
+	}
+
+	var rmseqs []uint64
+	mset.mu.Lock()
+	for seq := start; seq <= stop; seq++ {
+		if mset.noInterest(seq, co) {
+			rmseqs = append(rmseqs, seq)
+		}
+	}
+	mset.mu.Unlock()
+
+	// These can be removed.
+	for _, seq := range rmseqs {
+		mset.store.RemoveMsg(seq)
+	}
+}
+
 // Check that we do not form a cycle by delivering to a delivery subject
 // that is part of the interest group.
 func deliveryFormsCycle(cfg *StreamConfig, deliverySubject string) bool {
@@ -5227,7 +5290,7 @@ func (o *consumer) decStreamPending(sseq uint64, subj string) {
 	if wasPending {
 		// We could have the lock for the stream so do this in a go routine.
 		// TODO(dlc) - We should do this with ipq vs naked go routines.
-		go o.processTerm(sseq, p.Sequence, rdc, ackTermUnackedLimitsReason)
+		go o.processTerm(sseq, p.Sequence, rdc, ackTermUnackedLimitsReason, _EMPTY_)
 	}
 }

@@ -5263,18 +5326,24 @@ func (o *consumer) signalSubs() []*subscription {
 }

 // This is what will be called when our parent stream wants to kick us regarding a new message.
-// We know that we are the leader and that this subject matches us by how the parent handles registering
-// us with the signaling sublist.
+// We know that this subject matches us by how the parent handles registering us with the signaling sublist,
+// but we must check if we are leader.
 // We do need the sequence of the message however and we use the msg as the encoded seq.
 func (o *consumer) processStreamSignal(_ *subscription, _ *client, _ *Account, subject, _ string, seqb []byte) {
-	var le = binary.LittleEndian
-	seq := le.Uint64(seqb)
-
+	// We can get called here now when not leader, so bail fast
+	// and without acquiring any locks.
+	if !o.leader.Load() {
+		return
+	}
 	o.mu.Lock()
 	defer o.mu.Unlock()
 	if o.mset == nil {
 		return
 	}
+
+	var le = binary.LittleEndian
+	seq := le.Uint64(seqb)
+
 	if seq > o.npf {
 		o.npc++
 	}
--- a/vendor/github.com/nats-io/nats-server/v2/server/events.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/events.go
@@ -1120,6 +1120,10 @@ func (s *Server) initEventTracking() {
 			optz := &ProfilezEventOptions{}
 			s.zReq(c, reply, hdr, msg, &optz.EventFilterOptions, optz, func() (any, error) { return s.profilez(&optz.ProfilezOptions), nil })
 		},
+		"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 })
+		},
 	}
 	for name, req := range monSrvc {
 		subject = fmt.Sprintf(serverDirectReqSubj, s.info.ID, name)
@@ -1788,6 +1792,11 @@ type ProfilezEventOptions struct {
 	EventFilterOptions
 }

+// In the context of system events, ExpvarzEventOptions are options passed to Expvarz
+type ExpvarzEventOptions struct {
+	EventFilterOptions
+}
+
 // returns true if the request does NOT apply to this server and can be ignored.
 // DO NOT hold the server lock when
 func (s *Server) filterRequest(fOpts *EventFilterOptions) bool {
--- a/vendor/github.com/nats-io/nats-server/v2/server/filestore.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/filestore.go
@@ -52,6 +52,8 @@ type FileStoreConfig struct {
 	BlockSize uint64
 	// CacheExpire is how long with no activity until we expire the cache.
 	CacheExpire time.Duration
+	// SubjectStateExpire is how long with no activity until we expire a msg block's subject state.
+	SubjectStateExpire time.Duration
 	// SyncInterval is how often we sync to disk in the background.
 	SyncInterval time.Duration
 	// SyncAlways is when the stream should sync all data writes.
@@ -182,7 +184,6 @@ type fileStore struct {
 	adml        int
 	hh          hash.Hash64
 	qch         chan struct{}
-	fch         chan struct{}
 	fsld        chan struct{}
 	cmu         sync.RWMutex
 	cfs         []ConsumerStore
@@ -218,11 +219,13 @@ type msgBlock struct {
 	lwts       int64
 	llts       int64
 	lrts       int64
+	lsts       int64
 	llseq      uint64
 	hh         hash.Hash64
 	cache      *cache
 	cloads     uint64
 	cexp       time.Duration
+	fexp       time.Duration
 	ctmr       *time.Timer
 	werr       error
 	dmap       avl.SequenceSet
@@ -297,6 +300,8 @@ const (
 	defaultSyncInterval = 2 * time.Minute
 	// default idle timeout to close FDs.
 	closeFDsIdle = 30 * time.Second
+	// default expiration time for mb.fss when idle.
+	defaultFssExpiration = 10 * time.Second
 	// coalesceMinimum
 	coalesceMinimum = 16 * 1024
 	// maxFlushWait is maximum we will wait to gather messages to flush.
@@ -360,6 +365,9 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 	if fcfg.CacheExpire == 0 {
 		fcfg.CacheExpire = defaultCacheBufferExpiration
 	}
+	if fcfg.SubjectStateExpire == 0 {
+		fcfg.SubjectStateExpire = defaultFssExpiration
+	}
 	if fcfg.SyncInterval == 0 {
 		fcfg.SyncInterval = defaultSyncInterval
 	}
@@ -390,7 +398,6 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 		prf:    prf,
 		oldprf: oldprf,
 		qch:    make(chan struct{}),
-		fch:    make(chan struct{}, 1),
 		fsld:   make(chan struct{}),
 		srv:    fcfg.srv,
 	}
@@ -451,7 +458,6 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 		}
 		// Since we recovered here, make sure to kick ourselves to write out our stream state.
 		fs.dirty++
-		defer fs.kickFlushStateLoop()
 	}

 	// Also make sure we get rid of old idx and fss files on return.
@@ -525,8 +531,8 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 	// Setup our sync timer.
 	fs.setSyncTimer()

-	// Spin up the go routine that will write out or full state stream index.
-	go fs.flushStreamStateLoop(fs.fch, fs.qch, fs.fsld)
+	// Spin up the go routine that will write out our full state stream index.
+	go fs.flushStreamStateLoop(fs.qch, fs.fsld)

 	return fs, nil
 }
@@ -548,6 +554,13 @@ func (fs *fileStore) unlockAllMsgBlocks() {
 }

 func (fs *fileStore) UpdateConfig(cfg *StreamConfig) error {
+	start := time.Now()
+	defer func() {
+		if took := time.Since(start); took > time.Minute {
+			fs.warn("UpdateConfig took %v", took.Round(time.Millisecond))
+		}
+	}()
+
 	if fs.isClosed() {
 		return ErrStoreClosed
 	}
@@ -873,7 +886,14 @@ func (fs *fileStore) noTrackSubjects() bool {

 // Will init the basics for a message block.
 func (fs *fileStore) initMsgBlock(index uint32) *msgBlock {
-	mb := &msgBlock{fs: fs, index: index, cexp: fs.fcfg.CacheExpire, noTrack: fs.noTrackSubjects(), syncAlways: fs.fcfg.SyncAlways}
+	mb := &msgBlock{
+		fs:         fs,
+		index:      index,
+		cexp:       fs.fcfg.CacheExpire,
+		fexp:       fs.fcfg.SubjectStateExpire,
+		noTrack:    fs.noTrackSubjects(),
+		syncAlways: fs.fcfg.SyncAlways,
+	}

 	mdir := filepath.Join(fs.fcfg.StoreDir, msgDir)
 	mb.mfn = filepath.Join(mdir, fmt.Sprintf(blkScan, index))
@@ -955,16 +975,6 @@ func (mb *msgBlock) ensureLastChecksumLoaded() {
 	copy(mb.lchk[0:], mb.lastChecksum())
 }

-// Perform a recover but do not update PSIM.
-// Lock should be held.
-func (fs *fileStore) recoverMsgBlockNoSubjectUpdates(index uint32) (*msgBlock, error) {
-	psim, tsl := fs.psim, fs.tsl
-	fs.psim = nil
-	mb, err := fs.recoverMsgBlock(index)
-	fs.psim, fs.tsl = psim, tsl
-	return mb, err
-}
-
 // Lock held on entry
 func (fs *fileStore) recoverMsgBlock(index uint32) (*msgBlock, error) {
 	mb := fs.initMsgBlock(index)
@@ -1002,6 +1012,7 @@ func (fs *fileStore) recoverMsgBlock(index uint32) (*msgBlock, error) {
 	file.Close()

 	// Read our index file. Use this as source of truth if possible.
+	// This not applicable in >= 2.10 servers. Here for upgrade paths from < 2.10.
 	if err := mb.readIndexInfo(); err == nil {
 		// Quick sanity check here.
 		// Note this only checks that the message blk file is not newer then this file, or is empty and we expect empty.
@@ -1733,13 +1744,13 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 		// Reverse update of tracking state for this mb, will add new state in below.
 		mstate.Msgs -= mb.msgs
 		mstate.Bytes -= mb.bytes
-		if nmb, err := fs.recoverMsgBlockNoSubjectUpdates(mb.index); err != nil && !os.IsNotExist(err) {
+		if nmb, err := fs.recoverMsgBlock(mb.index); err != nil && !os.IsNotExist(err) {
 			fs.warn("Stream state could not recover last msg block")
 			os.Remove(fn)
 			return errCorruptState
 		} else if nmb != nil {
 			fs.adjustAccounting(mb, nmb)
-			updateTrackingState(&mstate, mb)
+			updateTrackingState(&mstate, nmb)
 		}
 	}

@@ -1796,10 +1807,26 @@ func (fs *fileStore) adjustAccounting(mb, nmb *msgBlock) {
 	}
 	nmb.ensurePerSubjectInfoLoaded()

+	var smv StoreMsg
+
+	// Need to walk previous messages and undo psim stats.
+	// We already undid msgs and bytes accounting.
+	for seq, lseq := atomic.LoadUint64(&mb.first.seq), atomic.LoadUint64(&mb.last.seq); seq <= lseq; seq++ {
+		// Lookup the message. If an error will be deleted, so can skip.
+		sm, err := nmb.cacheLookup(seq, &smv)
+		if err != nil {
+			continue
+		}
+		if len(sm.subj) > 0 && fs.psim != nil {
+			if info, ok := fs.psim.Find(stringToBytes(sm.subj)); ok {
+				info.total--
+			}
+		}
+	}
+
 	// Walk only new messages and update accounting at fs level. Any messages that should have
 	// triggered limits exceeded will be handled after the recovery and prior to the stream
 	// being available to the system.
-	var smv StoreMsg
 	for seq, lseq := atomic.LoadUint64(&mb.last.seq)+1, atomic.LoadUint64(&nmb.last.seq); seq <= lseq; seq++ {
 		// Lookup the message. If an error will be deleted, so can skip.
 		sm, err := nmb.cacheLookup(seq, &smv)
@@ -1809,17 +1836,6 @@ func (fs *fileStore) adjustAccounting(mb, nmb *msgBlock) {
 		// Since we found it we just need to adjust fs totals and psim.
 		fs.state.Msgs++
 		fs.state.Bytes += fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
-		if len(sm.subj) > 0 && fs.psim != nil {
-			if info, ok := fs.psim.Find(stringToBytes(sm.subj)); ok {
-				info.total++
-				if nmb.index > info.lblk {
-					info.lblk = nmb.index
-				}
-			} else {
-				fs.psim.Insert(stringToBytes(sm.subj), psi{total: 1, fblk: nmb.index, lblk: nmb.index})
-				fs.tsl += len(sm.subj)
-			}
-		}
 	}

 	// Now check to see if we had a higher first for the recovered state mb vs nmb.
@@ -2191,7 +2207,6 @@ func (fs *fileStore) expireMsgsOnRecover() {
 	// If we purged anything, make sure we kick flush state loop.
 	if purged > 0 {
 		fs.dirty++
-		fs.kickFlushStateLoop()
 	}
 }

@@ -2273,7 +2288,7 @@ func (mb *msgBlock) firstMatchingMulti(sl *Sublist, start uint64, sm *StoreMsg)
 		}
 		expireOk := seq == lseq && mb.llseq == seq

-		if r := sl.Match(fsm.subj); len(r.psubs) > 0 {
+		if sl.HasInterest(fsm.subj) {
 			return fsm, expireOk, nil
 		}
 		// If we are here we did not match, so put the llseq back.
@@ -2283,6 +2298,7 @@ func (mb *msgBlock) firstMatchingMulti(sl *Sublist, start uint64, sm *StoreMsg)
 }

 // Find the first matching message.
+// fs lock should be held.
 func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *StoreMsg) (*StoreMsg, bool, error) {
 	mb.mu.Lock()
 	defer mb.mu.Unlock()
@@ -2295,6 +2311,8 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 		mb.loadMsgsWithLock()
 		didLoad = true
 	}
+	// Mark fss activity.
+	mb.lsts = time.Now().UnixNano()

 	// If we only have 1 subject currently and it matches our filter we can also set isAll.
 	if !isAll && len(mb.fss) == 1 {
@@ -2319,8 +2337,16 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 			return isSubsetMatchTokenized(tts, fts)
 		}
 	}
-	// Only do linear scan if isAll or we are wildcarded and have to traverse more fss than actual messages.
-	doLinearScan := isAll || (wc && len(mb.fss) > int(lseq-fseq))
+
+	subjs := mb.fs.cfg.Subjects
+	// If isAll or our single filter matches the filter arg do linear scan.
+	doLinearScan := isAll || (wc && len(subjs) == 1 && subjs[0] == filter)
+	// If we do not think we should do a linear scan check how many fss we
+	// would need to scan vs the full range of the linear walk. Optimize for
+	// 25th quantile of a match in a linear walk. Filter should be a wildcard.
+	if !doLinearScan && wc {
+		doLinearScan = len(mb.fss)*4 > int(lseq-fseq)
+	}

 	if !doLinearScan {
 		// If we have a wildcard match against all tracked subjects we know about.
@@ -2360,7 +2386,7 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 	// If we guess to not do a linear scan, but the above resulted in alot of subs that will
 	// need to be checked for every scanned message, revert.
 	// TODO(dlc) - we could memoize the subs across calls.
-	if len(subs) > int(lseq-fseq) {
+	if !doLinearScan && len(subs) > int(lseq-fseq) {
 		doLinearScan = true
 	}

@@ -2653,6 +2679,8 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState {
 			mb.loadMsgsWithLock()
 			shouldExpire = true
 		}
+		// Mark fss activity.
+		mb.lsts = time.Now().UnixNano()
 		for subj, ss := range mb.fss {
 			if subject == _EMPTY_ || subject == fwcs || subjectIsSubsetMatch(subj, subject) {
 				if ss.firstNeedsUpdate {
@@ -2832,6 +2860,9 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 				mb.loadMsgsWithLock()
 				shouldExpire = true
 			}
+			// Mark fss activity.
+			mb.lsts = time.Now().UnixNano()
+
 			var havePartial bool
 			for subj, ss := range mb.fss {
 				if isMatch(subj) {
@@ -2847,6 +2878,7 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 					}
 				}
 			}
+
 			// See if we need to scan msgs here.
 			if havePartial {
 				// Make sure we have the cache loaded.
@@ -2920,6 +2952,9 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 					mb.loadMsgsWithLock()
 					shouldExpire = true
 				}
+				// Mark fss activity.
+				mb.lsts = time.Now().UnixNano()
+
 				for subj, ss := range mb.fss {
 					if isMatch(subj) {
 						adjust += ss.Msgs
@@ -3088,10 +3123,6 @@ func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 	// Add to our list of blocks and mark as last.
 	fs.addMsgBlock(mb)

-	if fs.dirty > 0 {
-		fs.kickFlushStateLoop()
-	}
-
 	return mb, nil
 }

@@ -3458,7 +3489,7 @@ func (fs *fileStore) rebuildFirst() {
 // Optimized helper function to return first sequence.
 // subj will always be publish subject here, meaning non-wildcard.
 // We assume a fast check that this subj even exists already happened.
-// Lock should be held.
+// Write lock should be held.
 func (fs *fileStore) firstSeqForSubj(subj string) (uint64, error) {
 	if len(fs.blks) == 0 {
 		return 0, nil
@@ -3475,16 +3506,25 @@ func (fs *fileStore) firstSeqForSubj(subj string) (uint64, error) {
 		if mb == nil {
 			continue
 		}
+		// If we need to load msgs here and we need to walk multiple blocks this
+		// could tie up the upper fs lock, so release while dealing with the block.
+		fs.mu.Unlock()
+
 		mb.mu.Lock()
 		var shouldExpire bool
 		if mb.fssNotLoaded() {
 			// Make sure we have fss loaded.
 			if err := mb.loadMsgsWithLock(); err != nil {
 				mb.mu.Unlock()
+				// Re-acquire fs lock
+				fs.mu.Lock()
 				return 0, err
 			}
 			shouldExpire = true
 		}
+		// Mark fss activity.
+		mb.lsts = time.Now().UnixNano()
+
 		if ss := mb.fss[subj]; ss != nil {
 			// Adjust first if it was not where we thought it should be.
 			if i != start {
@@ -3496,6 +3536,8 @@ func (fs *fileStore) firstSeqForSubj(subj string) (uint64, error) {
 				mb.recalculateFirstForSubj(subj, ss.First, ss)
 			}
 			mb.mu.Unlock()
+			// Re-acquire fs lock
+			fs.mu.Lock()
 			return ss.First, nil
 		}
 		// If we did not find it and we loaded this msgBlock try to expire as long as not the last.
@@ -3504,6 +3546,8 @@ func (fs *fileStore) firstSeqForSubj(subj string) (uint64, error) {
 			mb.tryForceExpireCacheLocked()
 		}
 		mb.mu.Unlock()
+		// Re-acquire fs lock
+		fs.mu.Lock()
 	}
 	return 0, nil
 }
@@ -3547,6 +3591,13 @@ func (fs *fileStore) enforceBytesLimit() {
 // will most likely only be the last one, so can take a more conservative approach.
 // Lock should be held.
 func (fs *fileStore) enforceMsgPerSubjectLimit(fireCallback bool) {
+	start := time.Now()
+	defer func() {
+		if took := time.Since(start); took > time.Minute {
+			fs.warn("enforceMsgPerSubjectLimit took %v", took.Round(time.Millisecond))
+		}
+	}()
+
 	maxMsgsPer := uint64(fs.cfg.MaxMsgsPer)

 	// We may want to suppress callbacks from remove during this process
@@ -3828,7 +3879,6 @@ func (fs *fileStore) removeMsg(seq uint64, secure, viaLimits, needFSLock bool) (
 		// TODO(dlc) - This should not be inline, should kick the sync routine.
 		if mb.rbytes > compactMinimum && mb.bytes*2 < mb.rbytes && !isLastBlock {
 			mb.compact()
-			fs.kickFlushStateLoop()
 		}
 	}

@@ -3933,8 +3983,9 @@ func (mb *msgBlock) compact() {
 		if !isDeleted(seq) {
 			// Check for tombstones.
 			if seq&tbit != 0 {
-				// If we are last mb we should consider to keep these unless the tombstone reflects a seq in this mb.
-				if mb == mb.fs.lmb && seq < fseq {
+				seq = seq &^ tbit
+				// If this entry is for a lower seq than ours then keep around.
+				if seq < fseq {
 					nbuf = append(nbuf, buf[index:index+rl]...)
 				}
 			} else {
@@ -3990,6 +4041,9 @@ func (mb *msgBlock) compact() {
 		return
 	}

+	// Make sure to sync
+	mb.needSync = true
+
 	// Capture the updated rbytes.
 	mb.rbytes = uint64(len(nbuf))

@@ -4419,9 +4473,16 @@ func (mb *msgBlock) clearCacheAndOffset() {

 // Lock should be held.
 func (mb *msgBlock) clearCache() {
-	if mb.ctmr != nil && mb.fss == nil {
-		mb.ctmr.Stop()
-		mb.ctmr = nil
+	if mb.ctmr != nil {
+		tsla := mb.sinceLastActivity()
+		if mb.fss == nil || tsla > mb.fexp {
+			// Force
+			mb.fss = nil
+			mb.ctmr.Stop()
+			mb.ctmr = nil
+		} else {
+			mb.resetCacheExpireTimer(mb.fexp - tsla)
+		}
 	}

 	if mb.cache == nil {
@@ -4486,7 +4547,7 @@ func (mb *msgBlock) tryExpireWriteCache() []byte {

 // Lock should be held.
 func (mb *msgBlock) expireCacheLocked() {
-	if mb.cache == nil {
+	if mb.cache == nil && mb.fss == nil {
 		if mb.ctmr != nil {
 			mb.ctmr.Stop()
 			mb.ctmr = nil
@@ -4694,6 +4755,8 @@ func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte
 		if err := mb.ensurePerSubjectInfoLoaded(); err != nil {
 			return err
 		}
+		// Mark fss activity.
+		mb.lsts = time.Now().UnixNano()
 		if ss := mb.fss[subj]; ss != nil {
 			ss.Msgs++
 			ss.Last = seq
@@ -5094,7 +5157,6 @@ func (fs *fileStore) syncBlocks() {
 	}
 	blks := append([]*msgBlock(nil), fs.blks...)
 	lmb := fs.lmb
-	syncInterval := fs.fcfg.SyncInterval
 	fs.mu.RUnlock()

 	var markDirty bool
@@ -5109,11 +5171,6 @@ func (fs *fileStore) syncBlocks() {
 		if mb.mfd != nil && mb.sinceLastWriteActivity() > closeFDsIdle {
 			mb.dirtyCloseWithRemove(false)
 		}
-		// Check our fss subject metadata.
-		// If we have no activity within sync interval remove.
-		if mb.fssLoaded() && mb.sinceLastActivity() > syncInterval {
-			mb.fss = nil
-		}

 		// Check if we should compact here as well.
 		// Do not compact last mb.
@@ -5308,6 +5365,8 @@ func (mb *msgBlock) indexCacheBuf(buf []byte) error {
 		mb.fss = make(map[string]*SimpleState)
 		popFss = true
 	}
+	// Mark fss activity.
+	mb.lsts = time.Now().UnixNano()

 	lbuf := uint32(len(buf))
 	var seq uint64
@@ -5580,12 +5639,6 @@ func (mb *msgBlock) fssNotLoaded() bool {
 	return mb.fss == nil && !mb.noTrack
 }

-// Report if we have our fss loaded.
-// Lock should be held.
-func (mb *msgBlock) fssLoaded() bool {
-	return mb.fss != nil
-}
-
 // Wrap openBlock for the gated semaphore processing.
 // Lock should be held
 func (mb *msgBlock) openBlock() (*os.File, error) {
@@ -5706,7 +5759,7 @@ checkCache:
 	// We want to hold the mb lock here to avoid any changes to state.
 	buf, err := mb.loadBlock(nil)
 	if err != nil {
-		mb.fs.warn("loadBlock error: ", err)
+		mb.fs.warn("loadBlock error: %v", err)
 		if err == errNoBlkData {
 			if ld, _, err := mb.rebuildStateLocked(); err != nil && ld != nil {
 				// Rebuild fs state too.
@@ -5812,6 +5865,7 @@ var (
 	errUnknownCipher = errors.New("unknown cipher")
 	errNoMainKey     = errors.New("encrypted store encountered with no main key")
 	errNoBlkData     = errors.New("message block data missing")
+	errStateTooBig   = errors.New("store state too big for optional write")
 )

 const (
@@ -6077,6 +6131,9 @@ func (fs *fileStore) loadLast(subj string, sm *StoreMsg) (lsm *StoreMsg, err err
 			mb.mu.Unlock()
 			return nil, err
 		}
+		// Mark fss activity.
+		mb.lsts = time.Now().UnixNano()
+
 		var l uint64
 		// Optimize if subject is not a wildcard.
 		if !wc {
@@ -6129,7 +6186,7 @@ func (fs *fileStore) LoadNextMsgMulti(sl *Sublist, start uint64, smp *StoreMsg)
 	if fs.closed {
 		return nil, 0, ErrStoreClosed
 	}
-	if fs.state.Msgs == 0 {
+	if fs.state.Msgs == 0 || start > fs.state.LastSeq {
 		return nil, fs.state.LastSeq, ErrStoreEOF
 	}
 	if start < fs.state.FirstSeq {
@@ -6163,7 +6220,7 @@ func (fs *fileStore) LoadNextMsg(filter string, wc bool, start uint64, sm *Store
 	if fs.closed {
 		return nil, 0, ErrStoreClosed
 	}
-	if fs.state.Msgs == 0 {
+	if fs.state.Msgs == 0 || start > fs.state.LastSeq {
 		return nil, fs.state.LastSeq, ErrStoreEOF
 	}
 	if start < fs.state.FirstSeq {
@@ -6325,6 +6382,9 @@ func (mb *msgBlock) sinceLastActivity() time.Duration {
 	if mb.llts > last {
 		last = mb.llts
 	}
+	if mb.lsts > last {
+		last = mb.lsts
+	}
 	return time.Since(time.Unix(0, last).UTC())
 }

@@ -6633,8 +6693,6 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 	cb := fs.scb
 	fs.mu.Unlock()

-	fs.kickFlushStateLoop()
-
 	if cb != nil {
 		cb(-int64(purged), -int64(bytes), 0, _EMPTY_)
 	}
@@ -6827,6 +6885,9 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 		if smb != fs.lmb {
 			smb.dirtyCloseWithRemove(true)
 			deleted++
+		} else {
+			// Make sure to sync changes.
+			smb.needSync = true
 		}
 		// Update fs first here as well.
 		fs.state.FirstSeq = atomic.LoadUint64(&smb.last.seq) + 1
@@ -6908,6 +6969,10 @@ SKIP:
 		purged = fs.state.Msgs
 	}
 	fs.state.Msgs -= purged
+	if fs.state.Msgs == 0 {
+		fs.state.FirstSeq = fs.state.LastSeq + 1
+		fs.state.FirstTime = time.Time{}
+	}

 	if bytes > fs.state.Bytes {
 		bytes = fs.state.Bytes
@@ -6915,7 +6980,6 @@ SKIP:
 	fs.state.Bytes -= bytes

 	fs.dirty++
-	fs.kickFlushStateLoop()

 	cb := fs.scb
 	fs.mu.Unlock()
@@ -6969,7 +7033,6 @@ func (fs *fileStore) reset() error {
 	// If we purged anything, make sure we kick flush state loop.
 	if purged > 0 {
 		fs.dirty++
-		fs.kickFlushStateLoop()
 	}

 	fs.mu.Unlock()
@@ -7056,7 +7119,6 @@ func (fs *fileStore) Truncate(seq uint64) error {
 	fs.resetGlobalPerSubjectInfo()

 	fs.dirty++
-	fs.kickFlushStateLoop()

 	cb := fs.scb
 	fs.mu.Unlock()
@@ -7143,8 +7205,6 @@ func (mb *msgBlock) dirtyCloseWithRemove(remove bool) {
 		mb.ctmr.Stop()
 		mb.ctmr = nil
 	}
-	// Clear any tracking by subject.
-	mb.fss = nil
 	// Close cache
 	mb.clearCacheAndOffset()
 	// Quit our loops.
@@ -7157,6 +7217,8 @@ func (mb *msgBlock) dirtyCloseWithRemove(remove bool) {
 		mb.mfd = nil
 	}
 	if remove {
+		// Clear any tracking by subject if we are removing.
+		mb.fss = nil
 		if mb.mfn != _EMPTY_ {
 			os.Remove(mb.mfn)
 			mb.mfn = _EMPTY_
@@ -7164,8 +7226,6 @@ func (mb *msgBlock) dirtyCloseWithRemove(remove bool) {
 		if mb.kfn != _EMPTY_ {
 			os.Remove(mb.kfn)
 		}
-		// Since we are removing a block kick the state flusher.
-		mb.fs.kickFlushStateLoop()
 	}
 }

@@ -7310,6 +7370,8 @@ func (mb *msgBlock) generatePerSubjectInfo() error {
 	if len(mb.fss) > 0 {
 		// Make sure we run the cache expire timer.
 		mb.llts = time.Now().UnixNano()
+		// Mark fss activity.
+		mb.lsts = time.Now().UnixNano()
 		mb.startCacheExpireTimer()
 	}
 	return nil
@@ -7319,6 +7381,10 @@ func (mb *msgBlock) generatePerSubjectInfo() error {
 // Lock should be held
 func (mb *msgBlock) ensurePerSubjectInfoLoaded() error {
 	if mb.fss != nil || mb.noTrack {
+		if mb.fss != nil {
+			// Mark fss activity.
+			mb.lsts = time.Now().UnixNano()
+		}
 		return nil
 	}
 	if mb.msgs == 0 {
@@ -7481,53 +7547,26 @@ const (
 	fullStateVersion = uint8(1)
 )

-// This go routine runs and receives kicks to write out our full stream state index.
-// This will get kicked when we create a new block or when we delete a block in general.
-// This is also called during Stop().
-func (fs *fileStore) flushStreamStateLoop(fch, qch, done chan struct{}) {
-	// Make sure we do not try to write these out too fast.
-	const writeThreshold = time.Minute
-	lastWrite := time.Time{}
-
-	// We will use these to complete the full state write while not doing them too fast.
-	var dt *time.Timer
-	var dtc <-chan time.Time
-
+// This go routine periodically writes out our full stream state index.
+func (fs *fileStore) flushStreamStateLoop(qch, done chan struct{}) {
+	// Signal we are done on exit.
 	defer close(done)

+	// Make sure we do not try to write these out too fast.
+	const writeThreshold = 2 * time.Minute
+	t := time.NewTicker(writeThreshold)
+	defer t.Stop()
+
 	for {
 		select {
-		case <-fch:
-			if elapsed := time.Since(lastWrite); elapsed > writeThreshold {
-				fs.writeFullState()
-				lastWrite = time.Now()
-				if dt != nil {
-					dt.Stop()
-					dt, dtc = nil, nil
-				}
-			} else if dtc == nil {
-				fireIn := time.Until(lastWrite.Add(writeThreshold))
-				if fireIn < 0 {
-					fireIn = 100 * time.Millisecond
-				}
-				dt = time.NewTimer(fireIn)
-				dtc = dt.C
-			}
-		case <-dtc:
+		case <-t.C:
 			fs.writeFullState()
-			lastWrite = time.Now()
-			dt, dtc = nil, nil
 		case <-qch:
 			return
 		}
 	}
 }

-// Kick the flusher.
-func (fs *fileStore) kickFlushStateLoop() {
-	kickFlusher(fs.fch)
-}
-
 // Helper since unixnano of zero time undefined.
 func timestampNormalized(t time.Time) int64 {
 	if t.IsZero() {
@@ -7536,6 +7575,17 @@ func timestampNormalized(t time.Time) int64 {
 	return t.UnixNano()
 }

+// writeFullState will proceed to write the full meta state iff not complex and time consuming.
+// Since this is for quick recovery it is optional and should not block/stall normal operations.
+func (fs *fileStore) writeFullState() error {
+	return fs._writeFullState(false)
+}
+
+// forceWriteFullState will proceed to write the full meta state. This should only be called by stop()
+func (fs *fileStore) forceWriteFullState() error {
+	return fs._writeFullState(true)
+}
+
 // This will write the full binary state for the stream.
 // This plus everything new since last hash will be the total recovered state.
 // This state dump will have the following.
@@ -7543,13 +7593,34 @@ func timestampNormalized(t time.Time) int64 {
 // 2. PSIM - Per Subject Index Map - Tracks first and last blocks with subjects present.
 // 3. MBs - Index, Bytes, First and Last Sequence and Timestamps, and the deleted map (avl.seqset).
 // 4. Last block index and hash of record inclusive to this stream state.
-func (fs *fileStore) writeFullState() error {
+func (fs *fileStore) _writeFullState(force bool) error {
+	start := time.Now()
 	fs.mu.Lock()
 	if fs.closed || fs.dirty == 0 {
 		fs.mu.Unlock()
 		return nil
 	}

+	// For calculating size and checking time costs for non forced calls.
+	numSubjects := fs.numSubjects()
+
+	// If we are not being forced to write out our state, check the complexity for time costs as to not
+	// block or stall normal operations.
+	// We will base off of number of subjects and interior deletes. A very large number of msg blocks could also
+	// be used, but for next server version will redo all meta handling to be disk based. So this is temporary.
+	if !force {
+		const numThreshold = 1_000_000
+		// Calculate interior deletes.
+		var numDeleted int
+		if fs.state.LastSeq > fs.state.FirstSeq {
+			numDeleted = int((fs.state.LastSeq - fs.state.FirstSeq + 1) - fs.state.Msgs)
+		}
+		if numSubjects > numThreshold || numDeleted > numThreshold {
+			fs.mu.Unlock()
+			return errStateTooBig
+		}
+	}
+
 	// We track this through subsequent runs to get an avg per blk used for subsequent runs.
 	avgDmapLen := fs.adml
 	// If first time through could be 0
@@ -7557,11 +7628,8 @@ func (fs *fileStore) writeFullState() error {
 		avgDmapLen = 1024
 	}

-	// For calculating size.
-	numSubjects := fs.psim.Size()
-
 	// Calculate and estimate of the uper bound on the  size to avoid multiple allocations.
-	sz := 2 + // Magic and Version
+	sz := hdrLen + // Magic and Version
 		(binary.MaxVarintLen64 * 6) + // FS data
 		binary.MaxVarintLen64 + fs.tsl + // NumSubjects + total subject length
 		numSubjects*(binary.MaxVarintLen64*4) + // psi record
@@ -7575,7 +7643,7 @@ func (fs *fileStore) writeFullState() error {

 	if sz <= ssz {
 		var _buf [ssz]byte
-		buf, sz = _buf[0:2:ssz], ssz
+		buf, sz = _buf[0:hdrLen:ssz], ssz
 	} else {
 		buf = make([]byte, hdrLen, sz)
 	}
@@ -7684,8 +7752,6 @@ func (fs *fileStore) writeFullState() error {
 		fs.warn("Stream state encountered internal inconsistency on write")
 		// Rebuild our fs state from the mb state.
 		fs.rebuildState(nil)
-		// Make sure to reprocess.
-		fs.kickFlushStateLoop()
 		return errCorruptState
 	}

@@ -7693,32 +7759,23 @@ func (fs *fileStore) writeFullState() error {
 		fs.debug("WriteFullState reallocated from %d to %d", sz, cap(buf))
 	}

-	// Write to a tmp file and rename.
-	const tmpPre = streamStreamStateFile + tsep
-	f, err := os.CreateTemp(filepath.Join(fs.fcfg.StoreDir, msgDir), tmpPre)
-	if err != nil {
-		return err
-	}
-	tmpName := f.Name()
-	defer os.Remove(tmpName)
-	if _, err = f.Write(buf); err == nil && fs.fcfg.SyncAlways {
-		f.Sync()
-	}
-	f.Close()
-	if err != nil {
-		return err
+	// Only warn about construction time since file write not holding any locks.
+	if took := time.Since(start); took > time.Minute {
+		fs.warn("WriteFullState took %v (%d bytes)", took.Round(time.Millisecond), len(buf))
 	}

-	// Rename into position under our lock, clear prior dirty pending on success.
-	fs.mu.Lock()
-	if !fs.closed {
-		if err := os.Rename(tmpName, fn); err != nil {
-			fs.mu.Unlock()
-			return err
-		}
+	// Write our update index.db
+	// Protect with dios.
+	<-dios
+	err := os.WriteFile(fn, buf, defaultFilePerms)
+	dios <- struct{}{}
+
+	// Update dirty if successful.
+	if err == nil {
+		fs.mu.Lock()
 		fs.dirty -= priorDirty
+		fs.mu.Unlock()
 	}
-	fs.mu.Unlock()

 	return nil
 }
@@ -7760,7 +7817,7 @@ func (fs *fileStore) stop(writeState bool) error {
 		fs.mu.Unlock()
 		<-fsld
 		// Write full state if needed. If not dirty this is a no-op.
-		fs.writeFullState()
+		fs.forceWriteFullState()
 		fs.mu.Lock()
 	}

--- a/vendor/github.com/nats-io/nats-server/v2/server/gateway.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/gateway.go
@@ -907,6 +907,15 @@ func (s *Server) createGateway(cfg *gatewayCfg, url *url.URL, conn net.Conn) {
 		c.Debugf("TLS version %s, cipher suite %s", tlsVersion(cs.Version), tlsCipher(cs.CipherSuite))
 	}

+	// For outbound, we can't set the normal ping timer yet since the other
+	// side would fail with a parse error should it receive anything but the
+	// CONNECT protocol as the first protocol. We still want to make sure
+	// that the connection is not stale until the first INFO from the remote
+	// is received.
+	if solicit {
+		c.watchForStaleConnection(adjustPingInterval(GATEWAY, opts.PingInterval), opts.MaxPingsOut)
+	}
+
 	c.mu.Unlock()

 	// Announce ourselves again to new connections.
--- a/vendor/github.com/nats-io/nats-server/v2/server/jetstream_cluster.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/jetstream_cluster.go
@@ -47,8 +47,8 @@ type jetStreamCluster struct {
 	// concurrent requests that would otherwise be accepted.
 	// We also record the group for the stream. This is needed since if we have
 	// concurrent requests for same account and stream we need to let it process to get
-	// a response but they need to be same group, peers etc.
-	inflight map[string]map[string]*raftGroup
+	// a response but they need to be same group, peers etc. and sync subjects.
+	inflight map[string]map[string]*inflightInfo
 	// Signals meta-leader should check the stream assignments.
 	streamsCheck bool
 	// Server.
@@ -70,6 +70,12 @@ type jetStreamCluster struct {
 	qch chan struct{}
 }

+// Used to track inflight stream add requests to properly re-use same group and sync subject.
+type inflightInfo struct {
+	rg   *raftGroup
+	sync string
+}
+
 // Used to guide placement of streams and meta controllers in clustered JetStream.
 type Placement struct {
 	Cluster string   `json:"cluster,omitempty"`
@@ -1576,6 +1582,7 @@ func (js *jetStream) applyMetaSnapshot(buf []byte, ru *recoveryUpdates, isRecove
 			}
 		}
 	}
+
 	// Now walk the ones to check and process consumers.
 	var caAdd, caDel []*consumerAssignment
 	for _, sa := range saChk {
@@ -2356,11 +2363,18 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 	}
 	defer stopDirectMonitoring()

+	// For checking interest state if applicable.
+	var cist *time.Ticker
+	var cistc <-chan time.Time
+
 	if mset != nil && mset.isInterestRetention() {
 		// Wait on our consumers to be assigned and running before proceeding.
 		// This can become important when a server has lots of assets
 		// since we process streams first then consumers as an asset class.
 		mset.waitOnConsumerAssignments()
+		// Setup a periodic check here.
+		cist = time.NewTicker(30 * time.Second)
+		cistc = cist.C
 	}

 	// This is triggered during a scale up from R1 to clustered mode. We need the new followers to catchup,
@@ -2484,6 +2498,9 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 				}
 			}

+		case <-cistc:
+			mset.checkInterestState()
+
 		case <-datc:
 			if mset == nil || isRecovering {
 				continue
@@ -2645,6 +2662,11 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 					uch = mset.updateC()
 					// Also update our mqch
 					mqch = mset.monitorQuitC()
+					// Setup a periodic check here if we are interest based as well.
+					if mset.isInterestRetention() {
+						cist = time.NewTicker(30 * time.Second)
+						cistc = cist.C
+					}
 				}
 			}
 			if err != nil {
@@ -3549,7 +3571,6 @@ func (js *jetStream) processClusterUpdateStream(acc *Account, osa, sa *streamAss
 		var needsSetLeader bool
 		if !alreadyRunning && numReplicas > 1 {
 			if needsNode {
-				mset.setLeader(false)
 				js.createRaftGroup(acc.GetName(), rg, storage, pprofLabels{
 					"type":    "stream",
 					"account": mset.accName(),
@@ -3569,10 +3590,14 @@ func (js *jetStream) processClusterUpdateStream(acc *Account, osa, sa *streamAss
 		} else if numReplicas == 1 && alreadyRunning {
 			// We downgraded to R1. Make sure we cleanup the raft node and the stream monitor.
 			mset.removeNode()
-			// Make sure we are leader now that we are R1.
-			needsSetLeader = true
 			// In case we need to shutdown the cluster specific subs, etc.
-			mset.setLeader(false)
+			mset.mu.Lock()
+			// Stop responding to sync requests.
+			mset.stopClusterSubs()
+			// Clear catchup state
+			mset.clearAllCatchupPeers()
+			mset.mu.Unlock()
+			// Remove from meta layer.
 			js.mu.Lock()
 			rg.node = nil
 			js.mu.Unlock()
@@ -4761,9 +4786,9 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 				o.checkStateForInterestStream()
 				// Do a snapshot.
 				doSnapshot(true)
-				// Synchronize followers to our state. Only send out if we have state.
+				// Synchronize followers to our state. Only send out if we have state and nothing pending.
 				if n != nil {
-					if _, _, applied := n.Progress(); applied > 0 {
+					if _, _, applied := n.Progress(); applied > 0 && aq.len() == 0 {
 						if snap, err := o.store.EncodedState(); err == nil {
 							n.SendSnapshot(snap)
 						}
@@ -4986,6 +5011,13 @@ var errConsumerClosed = errors.New("consumer closed")

 func (o *consumer) processReplicatedAck(dseq, sseq uint64) error {
 	o.mu.Lock()
+	// Update activity.
+	o.lat = time.Now()
+
+	// Do actual ack update to store.
+	// Always do this to have it recorded.
+	o.store.UpdateAcks(dseq, sseq)
+
 	mset := o.mset
 	if o.closed || mset == nil {
 		o.mu.Unlock()
@@ -4996,11 +5028,11 @@ func (o *consumer) processReplicatedAck(dseq, sseq uint64) error {
 		return errStreamClosed
 	}

-	// Update activity.
-	o.lat = time.Now()
-
-	// Do actual ack update to store.
-	o.store.UpdateAcks(dseq, sseq)
+	// Check if we have a reply that was requested.
+	if reply := o.replies[sseq]; reply != _EMPTY_ {
+		o.outq.sendMsg(reply, nil)
+		delete(o.replies, sseq)
+	}

 	if o.retention == LimitsPolicy {
 		o.mu.Unlock()
@@ -5481,7 +5513,7 @@ func (js *jetStream) processLeaderChange(isLeader bool) {
 		for acc, asa := range cc.streams {
 			for _, sa := range asa {
 				if sa.Sync == _EMPTY_ {
-					s.Warnf("Stream assigment corrupt for stream '%s > %s'", acc, sa.Config.Name)
+					s.Warnf("Stream assignment corrupt for stream '%s > %s'", acc, sa.Config.Name)
 					nsa := &streamAssignment{Group: sa.Group, Config: sa.Config, Subject: sa.Subject, Reply: sa.Reply, Client: sa.Client}
 					nsa.Sync = syncSubjForStream()
 					cc.meta.Propose(encodeUpdateStreamAssignment(nsa))
@@ -5985,6 +6017,7 @@ func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject,

 	var self *streamAssignment
 	var rg *raftGroup
+	var syncSubject string

 	// Capture if we have existing assignment first.
 	if osa := js.streamAssignment(acc.Name, cfg.Name); osa != nil {
@@ -5994,7 +6027,7 @@ func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject,
 			return
 		}
 		// This is an equal assignment.
-		self, rg = osa, osa.Group
+		self, rg, syncSubject = osa, osa.Group, osa.Sync
 	}

 	if cfg.Sealed {
@@ -6018,19 +6051,22 @@ func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject,
 		return
 	}

+	// Make sure inflight is setup properly.
+	if cc.inflight == nil {
+		cc.inflight = make(map[string]map[string]*inflightInfo)
+	}
+	streams, ok := cc.inflight[acc.Name]
+	if !ok {
+		streams = make(map[string]*inflightInfo)
+		cc.inflight[acc.Name] = streams
+	}
+
 	// Raft group selection and placement.
 	if rg == nil {
 		// Check inflight before proposing in case we have an existing inflight proposal.
-		if cc.inflight == nil {
-			cc.inflight = make(map[string]map[string]*raftGroup)
-		}
-		streams, ok := cc.inflight[acc.Name]
-		if !ok {
-			streams = make(map[string]*raftGroup)
-			cc.inflight[acc.Name] = streams
-		} else if existing, ok := streams[cfg.Name]; ok {
-			// We have existing for same stream. Re-use same group.
-			rg = existing
+		if existing, ok := streams[cfg.Name]; ok {
+			// We have existing for same stream. Re-use same group and syncSubject.
+			rg, syncSubject = existing.rg, existing.sync
 		}
 	}
 	// Create a new one here if needed.
@@ -6046,14 +6082,17 @@ func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject,
 		rg.setPreferred()
 	}

+	if syncSubject == _EMPTY_ {
+		syncSubject = syncSubjForStream()
+	}
 	// Sync subject for post snapshot sync.
-	sa := &streamAssignment{Group: rg, Sync: syncSubjForStream(), Config: cfg, Subject: subject, Reply: reply, Client: ci, Created: time.Now().UTC()}
+	sa := &streamAssignment{Group: rg, Sync: syncSubject, Config: cfg, Subject: subject, Reply: reply, Client: ci, Created: time.Now().UTC()}
 	if err := cc.meta.Propose(encodeAddStreamAssignment(sa)); err == nil {
 		// On success, add this as an inflight proposal so we can apply limits
 		// on concurrent create requests while this stream assignment has
 		// possibly not been processed yet.
 		if streams, ok := cc.inflight[acc.Name]; ok {
-			streams[cfg.Name] = rg
+			streams[cfg.Name] = &inflightInfo{rg, syncSubject}
 		}
 	}
 }
@@ -7468,7 +7507,7 @@ func encodeStreamMsg(subject, reply string, hdr, msg []byte, lseq uint64, ts int

 // Threshold for compression.
 // TODO(dlc) - Eventually make configurable.
-const compressThreshold = 256
+const compressThreshold = 8192 // 8k

 // If allowed and contents over the threshold we will compress.
 func encodeStreamMsgAllowCompress(subject, reply string, hdr, msg []byte, lseq uint64, ts int64, compressOK bool) []byte {
@@ -7625,7 +7664,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 	s, js, jsa, st, r, tierName, outq, node := mset.srv, mset.js, mset.jsa, mset.cfg.Storage, mset.cfg.Replicas, mset.tier, mset.outq, mset.node
 	maxMsgSize, lseq := int(mset.cfg.MaxMsgSize), mset.lseq
 	interestPolicy, discard, maxMsgs, maxBytes := mset.cfg.Retention != LimitsPolicy, mset.cfg.Discard, mset.cfg.MaxMsgs, mset.cfg.MaxBytes
-	isLeader, isSealed := mset.isLeader(), mset.cfg.Sealed
+	isLeader, isSealed, compressOK := mset.isLeader(), mset.cfg.Sealed, mset.compressOK
 	mset.mu.RUnlock()

 	// This should not happen but possible now that we allow scale up, and scale down where this could trigger.
@@ -7813,7 +7852,7 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 		}
 	}

-	esm := encodeStreamMsgAllowCompress(subject, reply, hdr, msg, mset.clseq, time.Now().UnixNano(), mset.compressOK)
+	esm := encodeStreamMsgAllowCompress(subject, reply, hdr, msg, mset.clseq, time.Now().UnixNano(), compressOK)
 	// Do proposal.
 	err := node.Propose(esm)
 	if err == nil {
@@ -7854,7 +7893,13 @@ type streamSyncRequest struct {
 }

 // Given a stream state that represents a snapshot, calculate the sync request based on our current state.
+// Stream lock must be held.
 func (mset *stream) calculateSyncRequest(state *StreamState, snap *StreamReplicatedState) *streamSyncRequest {
+	// Shouldn't happen, but consequences are pretty bad if we have the lock held and
+	// our caller tries to take the lock again on panic defer, as in processSnapshot.
+	if state == nil || snap == nil || mset.node == nil {
+		return nil
+	}
 	// Quick check if we are already caught up.
 	if state.LastSeq >= snap.LastSeq {
 		return nil
@@ -7982,7 +8027,7 @@ func (mset *stream) isCurrent() bool {
 }

 // Maximum requests for the whole server that can be in flight at the same time.
-const maxConcurrentSyncRequests = 16
+const maxConcurrentSyncRequests = 32

 var (
 	errCatchupCorruptSnapshot = errors.New("corrupt stream snapshot detected")
@@ -7990,6 +8035,7 @@ var (
 	errCatchupStreamStopped   = errors.New("stream has been stopped") // when a catchup is terminated due to the stream going away.
 	errCatchupBadMsg          = errors.New("bad catchup msg")
 	errCatchupWrongSeqForSkip = errors.New("wrong sequence for skipped msg")
+	errCatchupTooManyRetries  = errors.New("catchup failed, too many retries")
 )

 // Process a stream snapshot.
@@ -8054,6 +8100,13 @@ func (mset *stream) processSnapshot(snap *StreamReplicatedState) (e error) {
 			o.mu.Unlock()
 		}
 		mset.mu.Unlock()
+
+		// If we are interest based make sure to check our ack floor state.
+		// We will delay a bit to allow consumer states to also catchup.
+		if mset.isInterestRetention() {
+			fire := time.Duration(rand.Intn(10)+5) * time.Second
+			time.AfterFunc(fire, mset.checkInterestState)
+		}
 	}()

 	var releaseSem bool
@@ -8098,7 +8151,7 @@ RETRY:
 	numRetries++
 	if numRetries >= maxRetries {
 		// Force a hard reset here.
-		return errFirstSequenceMismatch
+		return errCatchupTooManyRetries
 	}

 	// Block here if we have too many requests in flight.
@@ -8175,7 +8228,6 @@ RETRY:
 				// Check for eof signaling.
 				if len(msg) == 0 {
 					msgsQ.recycle(&mrecs)
-					mset.checkInterestState()
 					return nil
 				}
 				if _, err := mset.processCatchupMsg(msg); err == nil {
--- a/vendor/github.com/nats-io/nats-server/v2/server/memstore.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/memstore.go
@@ -1012,7 +1012,7 @@ func (ms *memStore) LoadNextMsgMulti(sl *Sublist, start uint64, smp *StoreMsg) (
 		if !ok {
 			continue
 		}
-		if r := sl.Match(sm.subj); len(r.psubs) > 0 {
+		if sl.HasInterest(sm.subj) {
 			if smp == nil {
 				smp = new(StoreMsg)
 			}
--- a/vendor/github.com/nats-io/nats-server/v2/server/monitor.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/monitor.go
@@ -1,4 +1,4 @@
-// Copyright 2013-2023 The NATS Authors
+// Copyright 2013-2024 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -20,6 +20,7 @@ import (
 	"crypto/x509"
 	"encoding/hex"
 	"encoding/json"
+	"expvar"
 	"fmt"
 	"net"
 	"net/http"
@@ -2202,9 +2203,9 @@ func (s *Server) Leafz(opts *LeafzOptions) (*Leafz, error) {
 	}
 	s.mu.Unlock()

-	var leafnodes []*LeafInfo
+	leafnodes := make([]*LeafInfo, 0, len(lconns))
+
 	if len(lconns) > 0 {
-		leafnodes = make([]*LeafInfo, 0, len(lconns))
 		for _, ln := range lconns {
 			ln.mu.Lock()
 			lni := &LeafInfo{
@@ -2231,6 +2232,7 @@ func (s *Server) Leafz(opts *LeafzOptions) (*Leafz, error) {
 			leafnodes = append(leafnodes, lni)
 		}
 	}
+
 	return &Leafz{
 		ID:       s.ID(),
 		Now:      time.Now().UTC(),
@@ -2695,7 +2697,7 @@ func (s *Server) accountInfo(accName string) (*AccountInfo, error) {
 		accName,
 		a.updated.UTC(),
 		isSys,
-		a.expired,
+		a.expired.Load(),
 		!a.incomplete,
 		a.js != nil,
 		a.numLocalLeafNodes(),
@@ -3671,6 +3673,30 @@ func (s *Server) healthz(opts *HealthzOptions) *HealthStatus {
 	return health
 }

+type ExpvarzStatus struct {
+	Memstats json.RawMessage `json:"memstats"`
+	Cmdline  json.RawMessage `json:"cmdline"`
+}
+
+func (s *Server) expvarz(_ *ExpvarzEventOptions) *ExpvarzStatus {
+	var stat ExpvarzStatus
+
+	const memStatsKey = "memstats"
+	const cmdLineKey = "cmdline"
+
+	expvar.Do(func(v expvar.KeyValue) {
+		switch v.Key {
+		case memStatsKey:
+			stat.Memstats = json.RawMessage(v.Value.String())
+
+		case cmdLineKey:
+			stat.Cmdline = json.RawMessage(v.Value.String())
+		}
+	})
+
+	return &stat
+}
+
 type ProfilezStatus struct {
 	Profile []byte `json:"profile"`
 	Error   string `json:"error"`
--- a/vendor/github.com/nats-io/nats-server/v2/server/raft.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/raft.go
@@ -1,4 +1,4 @@
-// Copyright 2020-2023 The NATS Authors
+// Copyright 2020-2024 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -150,6 +150,7 @@ type raft struct {
 	pae     map[uint64]*appendEntry        // Pending append entries

 	elect  *time.Timer // Election timer, normally accessed via electTimer
+	etlr   time.Time   // Election timer last reset time, for unit tests only
 	active time.Time   // Last activity time, i.e. for heartbeats
 	llqrt  time.Time   // Last quorum lost time
 	lsut   time.Time   // Last scale-up time
@@ -469,6 +470,11 @@ func (s *Server) startRaftNode(accName string, cfg *RaftConfig, labels pprofLabe
 				}
 			}
 		}
+	} else if n.pterm == 0 && n.pindex == 0 {
+		// We have recovered no state, either through our WAL or snapshots,
+		// so inherit from term from our tav.idx file and pindex from our last sequence.
+		n.pterm = n.term
+		n.pindex = state.LastSeq
 	}

 	// Make sure to track ourselves.
@@ -1766,6 +1772,7 @@ func (n *raft) resetElectionTimeoutWithLock() {

 // Lock should be held.
 func (n *raft) resetElect(et time.Duration) {
+	n.etlr = time.Now()
 	if n.elect == nil {
 		n.elect = time.NewTimer(et)
 	} else {
@@ -2342,31 +2349,38 @@ func (n *raft) runAsLeader() {
 			n.resp.recycle(&ars)
 		case <-n.prop.ch:
 			const maxBatch = 256 * 1024
+			const maxEntries = 512
 			var entries []*Entry

-			es := n.prop.pop()
-			sz := 0
-			for i, b := range es {
+			es, sz := n.prop.pop(), 0
+			for _, b := range es {
 				if b.Type == EntryRemovePeer {
 					n.doRemovePeerAsLeader(string(b.Data))
 				}
 				entries = append(entries, b)
-				sz += len(b.Data) + 1
-				if i != len(es)-1 && sz < maxBatch && len(entries) < math.MaxUint16 {
-					continue
-				}
-				n.sendAppendEntry(entries)
-
 				// If this is us sending out a leadership transfer stepdown inline here.
 				if b.Type == EntryLeaderTransfer {
+					// Send out what we have and switch to follower.
+					n.sendAppendEntry(entries)
 					n.prop.recycle(&es)
 					n.debug("Stepping down due to leadership transfer")
 					n.switchToFollower(noLeader)
 					return
 				}
+				// Increment size.
+				sz += len(b.Data) + 1
+				// If below thresholds go ahead and send.
+				if sz < maxBatch && len(entries) < maxEntries {
+					continue
+				}
+				n.sendAppendEntry(entries)
+				// Reset our sz and entries.
 				// We need to re-create `entries` because there is a reference
 				// to it in the node's pae map.
-				entries = nil
+				sz, entries = 0, nil
+			}
+			if len(entries) > 0 {
+				n.sendAppendEntry(entries)
 			}
 			n.prop.recycle(&es)

@@ -2411,9 +2425,9 @@ func (n *raft) Quorum() bool {
 	n.RLock()
 	defer n.RUnlock()

-	now, nc := time.Now().UnixNano(), 1
-	for _, peer := range n.peers {
-		if now-peer.ts < int64(lostQuorumInterval) {
+	now, nc := time.Now().UnixNano(), 0
+	for id, peer := range n.peers {
+		if id == n.id || time.Duration(now-peer.ts) < lostQuorumInterval {
 			nc++
 			if nc >= n.qn {
 				return true
@@ -2435,9 +2449,9 @@ func (n *raft) lostQuorumLocked() bool {
 		return false
 	}

-	now, nc := time.Now().UnixNano(), 1
-	for _, peer := range n.peers {
-		if now-peer.ts < int64(lostQuorumInterval) {
+	now, nc := time.Now().UnixNano(), 0
+	for id, peer := range n.peers {
+		if id == n.id || time.Duration(now-peer.ts) < lostQuorumInterval {
 			nc++
 			if nc >= n.qn {
 				return false
@@ -2682,15 +2696,11 @@ func (n *raft) applyCommit(index uint64) error {
 		n.debug("Ignoring apply commit for %d, already processed", index)
 		return nil
 	}
-	original := n.commit
-	n.commit = index

 	if n.State() == Leader {
 		delete(n.acks, index)
 	}

-	var fpae bool
-
 	ae := n.pae[index]
 	if ae == nil {
 		var state StreamState
@@ -2708,15 +2718,14 @@ func (n *raft) applyCommit(index uint64) error {
 				// Reset and cancel any catchup.
 				n.resetWAL()
 				n.cancelCatchup()
-			} else {
-				n.commit = original
 			}
 			return errEntryLoadFailed
 		}
 	} else {
-		fpae = true
+		defer delete(n.pae, index)
 	}

+	n.commit = index
 	ae.buf = nil

 	var committed []*Entry
@@ -2791,9 +2800,6 @@ func (n *raft) applyCommit(index uint64) error {
 			committed = append(committed, e)
 		}
 	}
-	if fpae {
-		delete(n.pae, index)
-	}
 	// Pass to the upper layers if we have normal entries. It is
 	// entirely possible that 'committed' might be an empty slice here,
 	// which will happen if we've processed updates inline (like peer
@@ -3121,9 +3127,13 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	if n.State() == Leader {
 		// If we are the same we should step down to break the tie.
 		if ae.term >= n.term {
-			n.term = ae.term
-			n.vote = noVote
-			n.writeTermVote()
+			// If the append entry term is newer than the current term, erase our
+			// vote.
+			if ae.term > n.term {
+				n.term = ae.term
+				n.vote = noVote
+				n.writeTermVote()
+			}
 			n.debug("Received append entry from another leader, stepping down to %q", ae.leader)
 			n.stepdown.push(ae.leader)
 		} else {
@@ -3142,13 +3152,18 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	// another node has taken on the leader role already, so we should convert
 	// to a follower of that node instead.
 	if n.State() == Candidate {
-		n.debug("Received append entry in candidate state from %q, converting to follower", ae.leader)
-		if n.term < ae.term {
-			n.term = ae.term
-			n.vote = noVote
-			n.writeTermVote()
+		// Ignore old terms, otherwise we might end up stepping down incorrectly.
+		if ae.term >= n.term {
+			// If the append entry term is newer than the current term, erase our
+			// vote.
+			if ae.term > n.term {
+				n.term = ae.term
+				n.vote = noVote
+				n.writeTermVote()
+			}
+			n.debug("Received append entry in candidate state from %q, converting to follower", ae.leader)
+			n.stepdown.push(ae.leader)
 		}
-		n.stepdown.push(ae.leader)
 	}

 	// Catching up state.
@@ -3258,6 +3273,15 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 		// so make sure this is a snapshot entry. If it is not start the catchup process again since it
 		// means we may have missed additional messages.
 		if catchingUp {
+			// This means we already entered into a catchup state but what the leader sent us did not match what we expected.
+			// Snapshots and peerstate will always be together when a leader is catching us up in this fashion.
+			if len(ae.entries) != 2 || ae.entries[0].Type != EntrySnapshot || ae.entries[1].Type != EntryPeerState {
+				n.warn("Expected first catchup entry to be a snapshot and peerstate, will retry")
+				n.cancelCatchup()
+				n.Unlock()
+				return
+			}
+
 			// Check if only our terms do not match here.
 			if ae.pindex == n.pindex {
 				// Make sure pterms match and we take on the leader's.
@@ -3267,14 +3291,6 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 				n.Unlock()
 				return
 			}
-			// This means we already entered into a catchup state but what the leader sent us did not match what we expected.
-			// Snapshots and peerstate will always be together when a leader is catching us up in this fashion.
-			if len(ae.entries) != 2 || ae.entries[0].Type != EntrySnapshot || ae.entries[1].Type != EntryPeerState {
-				n.warn("Expected first catchup entry to be a snapshot and peerstate, will retry")
-				n.cancelCatchup()
-				n.Unlock()
-				return
-			}

 			if ps, err := decodePeerState(ae.entries[1].Data); err == nil {
 				n.processPeerState(ps)
@@ -3739,11 +3755,11 @@ func (n *raft) readTermVote() (term uint64, voted string, err error) {
 	if err != nil {
 		return 0, noVote, err
 	}
-	if len(buf) < termVoteLen {
-		return 0, noVote, nil
-	}
 	var le = binary.LittleEndian
 	term = le.Uint64(buf[0:])
+	if len(buf) < termVoteLen {
+		return term, noVote, nil
+	}
 	voted = string(buf[8:])
 	return term, voted, nil
 }
@@ -3809,6 +3825,8 @@ func (n *raft) writeTermVote() {
 	// Stamp latest and write the term & vote file.
 	n.wtv = b
 	if err := writeTermVote(n.sd, n.wtv); err != nil && !n.isClosed() {
+		// Clear wtv since we failed.
+		n.wtv = nil
 		n.setWriteErrLocked(err)
 		n.warn("Error writing term and vote file for %q: %v", n.group, err)
 	}
@@ -3875,7 +3893,6 @@ func (n *raft) processVoteRequest(vr *voteRequest) error {
 	}

 	n.Lock()
-	n.resetElectionTimeout()

 	vresp := &voteResponse{n.term, n.id, false}
 	defer n.debug("Sending a voteResponse %+v -> %q", vresp, vr.reply)
@@ -3907,6 +3924,7 @@ func (n *raft) processVoteRequest(vr *voteRequest) error {
 		n.term = vr.term
 		n.vote = vr.candidate
 		n.writeTermVote()
+		n.resetElectionTimeout()
 	} else {
 		if vr.term >= n.term && n.vote == noVote {
 			n.term = vr.term
@@ -3996,25 +4014,27 @@ func (n *raft) updateLeadChange(isLeader bool) {

 // Lock should be held.
 func (n *raft) switchState(state RaftState) {
-	if n.State() == Closed {
+	pstate := n.State()
+	if pstate == Closed {
 		return
 	}

 	// Reset the election timer.
 	n.resetElectionTimeout()
+	// Set our state.
+	n.state.Store(int32(state))

-	if n.State() == Leader && state != Leader {
+	if pstate == Leader && state != Leader {
 		n.updateLeadChange(false)
 		// Drain the response queue.
 		n.resp.drain()
-	} else if state == Leader && n.State() != Leader {
+	} else if state == Leader && pstate != Leader {
 		if len(n.pae) > 0 {
 			n.pae = make(map[uint64]*appendEntry)
 		}
 		n.updateLeadChange(true)
 	}

-	n.state.Store(int32(state))
 	n.writeTermVote()
 }

--- a/vendor/github.com/nats-io/nats-server/v2/server/route.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/route.go
@@ -1398,7 +1398,7 @@ func (c *client) processRemoteSub(argo []byte, hasOrigin bool) (err error) {
 		var isNew bool
 		if acc, isNew = srv.LookupOrRegisterAccount(accountName); isNew {
 			acc.mu.Lock()
-			acc.expired = true
+			acc.expired.Store(true)
 			acc.incomplete = true
 			acc.mu.Unlock()
 		}
@@ -1788,13 +1788,7 @@ func (s *Server) createRoute(conn net.Conn, rURL *url.URL, accName string) *clie
 		if opts.Cluster.MaxPingsOut > 0 {
 			pingMax = opts.MaxPingsOut
 		}
-		c.ping.tmr = time.AfterFunc(pingInterval*time.Duration(pingMax+1), func() {
-			c.mu.Lock()
-			c.Debugf("Stale Client Connection - Closing")
-			c.enqueueProto([]byte(fmt.Sprintf(errProto, "Stale Connection")))
-			c.mu.Unlock()
-			c.closeConnection(StaleConnection)
-		})
+		c.watchForStaleConnection(adjustPingInterval(ROUTER, pingInterval), pingMax)
 	} else {
 		// Set the Ping timer
 		c.setFirstPingTimer()
--- a/vendor/github.com/nats-io/nats-server/v2/server/server.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/server.go
@@ -2478,12 +2478,7 @@ func (s *Server) Shutdown() {
 	}

 	// Kick websocket server
-	if s.websocket.server != nil {
-		doneExpected++
-		s.websocket.server.Close()
-		s.websocket.server = nil
-		s.websocket.listener = nil
-	}
+	doneExpected += s.closeWebsocketServer()

 	// Kick MQTT accept loop
 	if s.mqtt.listener != nil {
@@ -2571,6 +2566,24 @@ func (s *Server) Shutdown() {
 	close(s.shutdownComplete)
 }

+// Close the websocket server if running. If so, returns 1, else 0.
+// Server lock held on entry.
+func (s *Server) closeWebsocketServer() int {
+	ws := &s.websocket
+	ws.mu.Lock()
+	hs := ws.server
+	if hs != nil {
+		ws.server = nil
+		ws.listener = nil
+	}
+	ws.mu.Unlock()
+	if hs != nil {
+		hs.Close()
+		return 1
+	}
+	return 0
+}
+
 // WaitForShutdown will block until the server has been fully shutdown.
 func (s *Server) WaitForShutdown() {
 	<-s.shutdownComplete
@@ -4090,12 +4103,7 @@ func (s *Server) lameDuckMode() {
 	expected := 1
 	s.listener.Close()
 	s.listener = nil
-	if s.websocket.server != nil {
-		expected++
-		s.websocket.server.Close()
-		s.websocket.server = nil
-		s.websocket.listener = nil
-	}
+	expected += s.closeWebsocketServer()
 	s.ldmCh = make(chan bool, expected)
 	opts := s.getOpts()
 	gp := opts.LameDuckGracePeriod
@@ -4396,8 +4404,10 @@ func (s *Server) changeRateLimitLogInterval(d time.Duration) {

 // DisconnectClientByID disconnects a client by connection ID
 func (s *Server) DisconnectClientByID(id uint64) error {
-	client := s.clients[id]
-	if client != nil {
+	if s == nil {
+		return ErrServerNotRunning
+	}
+	if client := s.getClient(id); client != nil {
 		client.closeConnection(Kicked)
 		return nil
 	}
@@ -4406,23 +4416,27 @@ func (s *Server) DisconnectClientByID(id uint64) error {

 // LDMClientByID sends a Lame Duck Mode info message to a client by connection ID
 func (s *Server) LDMClientByID(id uint64) error {
+	if s == nil {
+		return ErrServerNotRunning
+	}
+	s.mu.RLock()
+	c := s.clients[id]
+	if c == nil {
+		s.mu.RUnlock()
+		return errors.New("no such client id")
+	}
 	info := s.copyInfo()
 	info.LameDuckMode = true
-
-	c := s.clients[id]
-	if c != nil {
-		c.mu.Lock()
-		defer c.mu.Unlock()
-		if c.opts.Protocol >= ClientProtoInfo &&
-			c.flags.isSet(firstPongSent) {
-			// sendInfo takes care of checking if the connection is still
-			// valid or not, so don't duplicate tests here.
-			c.Debugf("sending Lame Duck Mode info to client")
-			c.enqueueProto(c.generateClientInfoJSON(info))
-			return nil
-		} else {
-			return errors.New("ClientProtoInfo < ClientOps.Protocol or first pong not sent")
-		}
+	s.mu.RUnlock()
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.opts.Protocol >= ClientProtoInfo && c.flags.isSet(firstPongSent) {
+		// sendInfo takes care of checking if the connection is still
+		// valid or not, so don't duplicate tests here.
+		c.Debugf("Sending Lame Duck Mode info to client")
+		c.enqueueProto(c.generateClientInfoJSON(info))
+		return nil
+	} else {
+		return errors.New("client does not support Lame Duck Mode or is not ready to receive the notification")
 	}
-	return errors.New("no such client id")
 }
--- a/vendor/github.com/nats-io/nats-server/v2/server/store.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/store.go
@@ -721,7 +721,7 @@ func isOutOfSpaceErr(err error) bool {
 var errFirstSequenceMismatch = errors.New("first sequence mismatch")

 func isClusterResetErr(err error) bool {
-	return err == errLastSeqMismatch || err == ErrStoreEOF || err == errFirstSequenceMismatch
+	return err == errLastSeqMismatch || err == ErrStoreEOF || err == errFirstSequenceMismatch || err == errCatchupTooManyRetries
 }

 // Copy all fields.
--- a/vendor/github.com/nats-io/nats-server/v2/server/stream.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/stream.go
--- a/vendor/github.com/nats-io/nats-server/v2/server/stree/dump.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/stree/dump.go
@@ -38,7 +38,7 @@ func (t *SubjectTree[T]) dump(w io.Writer, n node, depth int) {
 	} else {
 		// We are a node type here, grab meta portion.
 		bn := n.base()
-		fmt.Fprintf(w, "%s %s Prefix: %q\n", dumpPre(depth), n.kind(), bn.prefix[:bn.prefixLen])
+		fmt.Fprintf(w, "%s %s Prefix: %q\n", dumpPre(depth), n.kind(), bn.prefix)
 		depth++
 		n.iter(func(n node) bool {
 			t.dump(w, n, depth)
--- a/vendor/github.com/nats-io/nats-server/v2/server/stree/node.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/stree/node.go
@@ -32,13 +32,22 @@ type node interface {
 	path() []byte
 }

-// Maximum prefix len
-// We expect the most savings to come from long shared prefixes.
-const maxPrefixLen = 24
-
-// 64 bytes total - an L1 cache line.
 type meta struct {
-	prefix    [maxPrefixLen]byte
-	prefixLen uint16
-	size      uint16
+	prefix []byte
+	size   uint16
+}
+
+func (n *meta) isLeaf() bool { return false }
+func (n *meta) base() *meta  { return n }
+
+func (n *meta) setPrefix(pre []byte) {
+	n.prefix = append([]byte(nil), pre...)
+}
+
+func (n *meta) numChildren() uint16 { return n.size }
+func (n *meta) path() []byte        { return n.prefix }
+
+// Will match parts against our prefix.
+func (n *meta) matchParts(parts [][]byte) ([][]byte, bool) {
+	return matchParts(parts, n.prefix)
 }
--- a/vendor/github.com/nats-io/nats-server/v2/server/stree/node16.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/stree/node16.go
@@ -26,16 +26,6 @@ func newNode16(prefix []byte) *node16 {
 	return nn
 }

-func (n *node16) isLeaf() bool { return false }
-func (n *node16) base() *meta  { return &n.meta }
-
-func (n *node16) setPrefix(pre []byte) {
-	n.prefixLen = uint16(min(len(pre), maxPrefixLen))
-	for i := uint16(0); i < n.prefixLen; i++ {
-		n.prefix[i] = pre[i]
-	}
-}
-
 // Currently we do not keep node16 sorted or use bitfields for traversal so just add to the end.
 // TODO(dlc) - We should revisit here with more detailed benchmarks.
 func (n *node16) addChild(c byte, nn node) {
@@ -47,9 +37,6 @@ func (n *node16) addChild(c byte, nn node) {
 	n.size++
 }

-func (n *node16) numChildren() uint16 { return n.size }
-func (n *node16) path() []byte        { return n.prefix[:n.prefixLen] }
-
 func (n *node16) findChild(c byte) *node {
 	for i := uint16(0); i < n.size; i++ {
 		if n.key[i] == c {
@@ -62,7 +49,7 @@ func (n *node16) findChild(c byte) *node {
 func (n *node16) isFull() bool { return n.size >= 16 }

 func (n *node16) grow() node {
-	nn := newNode256(n.prefix[:n.prefixLen])
+	nn := newNode256(n.prefix)
 	for i := 0; i < 16; i++ {
 		nn.addChild(n.key[i], n.child[i])
 	}
@@ -101,11 +88,6 @@ func (n *node16) shrink() node {
 	return nn
 }

-// Will match parts against our prefix.no
-func (n *node16) matchParts(parts [][]byte) ([][]byte, bool) {
-	return matchParts(parts, n.prefix[:n.prefixLen])
-}
-
 // Iterate over all children calling func f.
 func (n *node16) iter(f func(node) bool) {
 	for i := uint16(0); i < n.size; i++ {
--- a/vendor/github.com/nats-io/nats-server/v2/server/stree/node256.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/stree/node256.go
@@ -25,24 +25,11 @@ func newNode256(prefix []byte) *node256 {
 	return nn
 }

-func (n *node256) isLeaf() bool { return false }
-func (n *node256) base() *meta  { return &n.meta }
-
-func (n *node256) setPrefix(pre []byte) {
-	n.prefixLen = uint16(min(len(pre), maxPrefixLen))
-	for i := uint16(0); i < n.prefixLen; i++ {
-		n.prefix[i] = pre[i]
-	}
-}
-
 func (n *node256) addChild(c byte, nn node) {
 	n.child[c] = nn
 	n.size++
 }

-func (n *node256) numChildren() uint16 { return n.size }
-func (n *node256) path() []byte        { return n.prefix[:n.prefixLen] }
-
 func (n *node256) findChild(c byte) *node {
 	if n.child[c] != nil {
 		return &n.child[c]
@@ -75,11 +62,6 @@ func (n *node256) shrink() node {
 	return nn
 }

-// Will match parts against our prefix.
-func (n *node256) matchParts(parts [][]byte) ([][]byte, bool) {
-	return matchParts(parts, n.prefix[:n.prefixLen])
-}
-
 // Iterate over all children calling func f.
 func (n *node256) iter(f func(node) bool) {
 	for i := 0; i < 256; i++ {
--- a/vendor/github.com/nats-io/nats-server/v2/server/stree/node4.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/stree/node4.go
@@ -26,16 +26,6 @@ func newNode4(prefix []byte) *node4 {
 	return nn
 }

-func (n *node4) isLeaf() bool { return false }
-func (n *node4) base() *meta  { return &n.meta }
-
-func (n *node4) setPrefix(pre []byte) {
-	n.prefixLen = uint16(min(len(pre), maxPrefixLen))
-	for i := uint16(0); i < n.prefixLen; i++ {
-		n.prefix[i] = pre[i]
-	}
-}
-
 // Currently we do not need to keep sorted for traversal so just add to the end.
 func (n *node4) addChild(c byte, nn node) {
 	if n.size >= 4 {
@@ -46,9 +36,6 @@ func (n *node4) addChild(c byte, nn node) {
 	n.size++
 }

-func (n *node4) numChildren() uint16 { return n.size }
-func (n *node4) path() []byte        { return n.prefix[:n.prefixLen] }
-
 func (n *node4) findChild(c byte) *node {
 	for i := uint16(0); i < n.size; i++ {
 		if n.key[i] == c {
@@ -61,7 +48,7 @@ func (n *node4) findChild(c byte) *node {
 func (n *node4) isFull() bool { return n.size >= 4 }

 func (n *node4) grow() node {
-	nn := newNode16(n.prefix[:n.prefixLen])
+	nn := newNode16(n.prefix)
 	for i := 0; i < 4; i++ {
 		nn.addChild(n.key[i], n.child[i])
 	}
@@ -96,11 +83,6 @@ func (n *node4) shrink() node {
 	return nil
 }

-// Will match parts against our prefix.
-func (n *node4) matchParts(parts [][]byte) ([][]byte, bool) {
-	return matchParts(parts, n.prefix[:n.prefixLen])
-}
-
 // Iterate over all children calling func f.
 func (n *node4) iter(f func(node) bool) {
 	for i := uint16(0); i < n.size; i++ {
--- a/vendor/github.com/nats-io/nats-server/v2/server/stree/parts.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/stree/parts.go
@@ -45,6 +45,14 @@ func genParts(filter []byte, parts [][]byte) [][]byte {
 				start = i + 1
 			}
 		} else if filter[i] == pwc || filter[i] == fwc {
+			// Wildcard must be at the start or preceded by tsep.
+			if prev := i - 1; prev >= 0 && filter[prev] != tsep {
+				continue
+			}
+			// Wildcard must be at the end or followed by tsep.
+			if next := i + 1; next == e || next < e && filter[next] != tsep {
+				continue
+			}
 			// We start with a pwc or fwc.
 			parts = append(parts, filter[i:i+1])
 			if i+1 <= e {
--- a/vendor/github.com/nats-io/nats-server/v2/server/stree/stree.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/stree/stree.go
@@ -60,7 +60,7 @@ func (t *SubjectTree[T]) Insert(subject []byte, value T) (*T, bool) {

 // Find will find the value and return it or false if it was not found.
 func (t *SubjectTree[T]) Find(subject []byte) (*T, bool) {
-	var si uint16
+	var si int
 	for n := t.root; n != nil; {
 		if n.isLeaf() {
 			if ln := n.(*leaf[T]); ln.match(subject[si:]) {
@@ -69,13 +69,13 @@ func (t *SubjectTree[T]) Find(subject []byte) (*T, bool) {
 			return nil, false
 		}
 		// We are a node type here, grab meta portion.
-		if bn := n.base(); bn.prefixLen > 0 {
-			end := min(int(si+bn.prefixLen), len(subject))
-			if !bytes.Equal(subject[si:end], bn.prefix[:bn.prefixLen]) {
+		if bn := n.base(); len(bn.prefix) > 0 {
+			end := min(si+len(bn.prefix), len(subject))
+			if !bytes.Equal(subject[si:end], bn.prefix) {
 				return nil, false
 			}
 			// Increment our subject index.
-			si += bn.prefixLen
+			si += len(bn.prefix)
 		}
 		if an := n.findChild(pivot(subject, si)); an != nil {
 			n = *an
@@ -157,9 +157,9 @@ func (t *SubjectTree[T]) insert(np *node, subject []byte, value T, si int) (*T,

 	// Non-leaf nodes.
 	bn := n.base()
-	if bn.prefixLen > 0 {
-		cpi := commonPrefixLen(bn.prefix[:bn.prefixLen], subject[si:])
-		if pli := int(bn.prefixLen); cpi >= pli {
+	if len(bn.prefix) > 0 {
+		cpi := commonPrefixLen(bn.prefix, subject[si:])
+		if pli := len(bn.prefix); cpi >= pli {
 			// Move past this node. We look for an existing child node to recurse into.
 			// If one does not exist we can create a new leaf node.
 			si += pli
@@ -180,7 +180,7 @@ func (t *SubjectTree[T]) insert(np *node, subject []byte, value T, si int) (*T,
 			// We will insert a new node4 and attach our current node below after adjusting prefix.
 			nn := newNode4(prefix)
 			// Shift the prefix for our original node.
-			n.setPrefix(bn.prefix[cpi:bn.prefixLen])
+			n.setPrefix(bn.prefix[cpi:])
 			nn.addChild(pivot(bn.prefix[:], 0), n)
 			// Add in our new leaf.
 			nn.addChild(pivot(subject[si:], 0), newLeaf(subject[si:], value))
@@ -203,7 +203,7 @@ func (t *SubjectTree[T]) insert(np *node, subject []byte, value T, si int) (*T,
 }

 // internal function to recursively find the leaf to delete. Will do compaction if the item is found and removed.
-func (t *SubjectTree[T]) delete(np *node, subject []byte, si uint16) (*T, bool) {
+func (t *SubjectTree[T]) delete(np *node, subject []byte, si int) (*T, bool) {
 	if t == nil || np == nil || *np == nil || len(subject) == 0 {
 		return nil, false
 	}
@@ -217,12 +217,12 @@ func (t *SubjectTree[T]) delete(np *node, subject []byte, si uint16) (*T, bool)
 		return nil, false
 	}
 	// Not a leaf node.
-	if bn := n.base(); bn.prefixLen > 0 {
-		if !bytes.Equal(subject[si:si+bn.prefixLen], bn.prefix[:bn.prefixLen]) {
+	if bn := n.base(); len(bn.prefix) > 0 {
+		if !bytes.Equal(subject[si:si+len(bn.prefix)], bn.prefix) {
 			return nil, false
 		}
 		// Increment our subject index.
-		si += bn.prefixLen
+		si += len(bn.prefix)
 	}
 	p := pivot(subject, si)
 	nna := n.findChild(p)
@@ -238,7 +238,7 @@ func (t *SubjectTree[T]) delete(np *node, subject []byte, si uint16) (*T, bool)
 			if sn := n.shrink(); sn != nil {
 				bn := n.base()
 				// Make sure to set cap so we force an append to copy below.
-				pre := bn.prefix[:bn.prefixLen:bn.prefixLen]
+				pre := bn.prefix[:len(bn.prefix):len(bn.prefix)]
 				// Need to fix up prefixes/suffixes.
 				if sn.isLeaf() {
 					ln := sn.(*leaf[T])
@@ -248,7 +248,7 @@ func (t *SubjectTree[T]) delete(np *node, subject []byte, si uint16) (*T, bool)
 					// We are a node here, we need to add in the old prefix.
 					if len(pre) > 0 {
 						bsn := sn.base()
-						sn.setPrefix(append(pre, bsn.prefix[:bsn.prefixLen]...))
+						sn.setPrefix(append(pre, bsn.prefix...))
 					}
 				}
 				*np = sn
@@ -287,9 +287,9 @@ func (t *SubjectTree[T]) match(n node, parts [][]byte, pre []byte, cb func(subje
 		// We have normal nodes here.
 		// We need to append our prefix
 		bn := n.base()
-		if bn.prefixLen > 0 {
+		if len(bn.prefix) > 0 {
 			// Note that this append may reallocate, but it doesn't modify "pre" at the "match" callsite.
-			pre = append(pre, bn.prefix[:bn.prefixLen]...)
+			pre = append(pre, bn.prefix...)
 		}

 		// Check our remaining parts.
@@ -359,7 +359,7 @@ func (t *SubjectTree[T]) iter(n node, pre []byte, cb func(subject []byte, val *T
 	// We are normal node here.
 	bn := n.base()
 	// Note that this append may reallocate, but it doesn't modify "pre" at the "iter" callsite.
-	pre = append(pre, bn.prefix[:bn.prefixLen]...)
+	pre = append(pre, bn.prefix...)
 	// Collect nodes since unsorted.
 	var _nodes [256]node
 	nodes := _nodes[:0]
--- a/vendor/github.com/nats-io/nats-server/v2/server/stree/util.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/stree/util.go
@@ -29,7 +29,7 @@ func commonPrefixLen(s1, s2 []byte) int {
 			break
 		}
 	}
-	return min(i, maxPrefixLen)
+	return i
 }

 // Helper to copy bytes.
--- a/vendor/github.com/nats-io/nats-server/v2/server/subject_transform.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/subject_transform.go
@@ -31,6 +31,8 @@ var (
 	sliceFromLeftMappingFunctionRegEx  = regexp.MustCompile(`{{\s*[sS]lice[fF]rom[lL]eft\s*\((.*)\)\s*}}`)
 	sliceFromRightMappingFunctionRegEx = regexp.MustCompile(`{{\s*[sS]lice[fF]rom[rR]ight\s*\((.*)\)\s*}}`)
 	splitMappingFunctionRegEx          = regexp.MustCompile(`{{\s*[sS]plit\s*\((.*)\)\s*}}`)
+	leftMappingFunctionRegEx           = regexp.MustCompile(`{{\s*[lL]eft\s*\((.*)\)\s*}}`)
+	rightMappingFunctionRegEx          = regexp.MustCompile(`{{\s*[rR]ight\s*\((.*)\)\s*}}`)
 )

 // Enum for the subject mapping subjectTransform function types
@@ -44,6 +46,8 @@ const (
 	SliceFromLeft
 	SliceFromRight
 	Split
+	Left
+	Right
 )

 // Transforms for arbitrarily mapping subjects from one to another for maps, tees and filters.
@@ -297,6 +301,18 @@ func indexPlaceHolders(token string) (int16, []int, int32, string, error) {
 				return transformIndexIntArgsHelper(token, args, SliceFromRight)
 			}

+			// Right(token, length)
+			args = getMappingFunctionArgs(rightMappingFunctionRegEx, token)
+			if args != nil {
+				return transformIndexIntArgsHelper(token, args, Right)
+			}
+
+			// Left(token, length)
+			args = getMappingFunctionArgs(leftMappingFunctionRegEx, token)
+			if args != nil {
+				return transformIndexIntArgsHelper(token, args, Left)
+			}
+
 			// split(token, deliminator)
 			args = getMappingFunctionArgs(splitMappingFunctionRegEx, token)
 			if args != nil {
@@ -515,6 +531,24 @@ func (tr *subjectTransform) TransformTokenizedSubject(tokens []string) string {
 						b.WriteString(tsep)
 					}
 				}
+			case Left:
+				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
+				sourceTokenLen := len(sourceToken)
+				sliceSize := int(tr.dtokmfintargs[i])
+				if sliceSize > 0 && sliceSize < sourceTokenLen {
+					b.WriteString(sourceToken[0:sliceSize])
+				} else { // too small to slice at the requested size: don't slice
+					b.WriteString(sourceToken)
+				}
+			case Right:
+				sourceToken := tokens[tr.dtokmftokindexesargs[i][0]]
+				sourceTokenLen := len(sourceToken)
+				sliceSize := int(tr.dtokmfintargs[i])
+				if sliceSize > 0 && sliceSize < sourceTokenLen {
+					b.WriteString(sourceToken[sourceTokenLen-sliceSize : sourceTokenLen])
+				} else { // too small to slice at the requested size: don't slice
+					b.WriteString(sourceToken)
+				}
 			}
 		}

--- a/vendor/github.com/nats-io/nats-server/v2/server/sublist.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/sublist.go
@@ -530,6 +530,12 @@ func (s *Sublist) Match(subject string) *SublistResult {
 	return s.match(subject, true)
 }

+// HasInterest will return whether or not there is any interest in the subject.
+// In cases where more detail is not required, this may be faster than Match.
+func (s *Sublist) HasInterest(subject string) bool {
+	return s.hasInterest(subject, true)
+}
+
 func (s *Sublist) matchNoLock(subject string) *SublistResult {
 	return s.match(subject, false)
 }
@@ -608,6 +614,49 @@ func (s *Sublist) match(subject string, doLock bool) *SublistResult {
 	return result
 }

+func (s *Sublist) hasInterest(subject string, doLock bool) bool {
+	// Check cache first.
+	if doLock {
+		s.RLock()
+	}
+	var matched bool
+	if s.cache != nil {
+		if r, ok := s.cache[subject]; ok {
+			matched = len(r.psubs)+len(r.qsubs) > 0
+		}
+	}
+	if doLock {
+		s.RUnlock()
+	}
+	if matched {
+		atomic.AddUint64(&s.cacheHits, 1)
+		return true
+	}
+
+	tsa := [32]string{}
+	tokens := tsa[:0]
+	start := 0
+	for i := 0; i < len(subject); i++ {
+		if subject[i] == btsep {
+			if i-start == 0 {
+				return false
+			}
+			tokens = append(tokens, subject[start:i])
+			start = i + 1
+		}
+	}
+	if start >= len(subject) {
+		return false
+	}
+	tokens = append(tokens, subject[start:])
+
+	if doLock {
+		s.RLock()
+		defer s.RUnlock()
+	}
+	return matchLevelForAny(s.root, tokens)
+}
+
 // Remove entries in the cache until we are under the maximum.
 // TODO(dlc) this could be smarter now that its not inline.
 func (s *Sublist) reduceCacheCount() {
@@ -720,6 +769,36 @@ func matchLevel(l *level, toks []string, results *SublistResult) {
 	}
 }

+func matchLevelForAny(l *level, toks []string) bool {
+	var pwc, n *node
+	for i, t := range toks {
+		if l == nil {
+			return false
+		}
+		if l.fwc != nil {
+			return true
+		}
+		if pwc = l.pwc; pwc != nil {
+			if match := matchLevelForAny(pwc.next, toks[i+1:]); match {
+				return true
+			}
+		}
+		n = l.nodes[t]
+		if n != nil {
+			l = n.next
+		} else {
+			l = nil
+		}
+	}
+	if n != nil {
+		return len(n.plist) > 0 || len(n.psubs) > 0 || len(n.qsubs) > 0
+	}
+	if pwc != nil {
+		return len(pwc.plist) > 0 || len(pwc.psubs) > 0 || len(pwc.qsubs) > 0
+	}
+	return false
+}
+
 // lnt is used to track descent into levels for a removal for pruning.
 type lnt struct {
 	l *level
--- a/vendor/github.com/nats-io/nats-server/v2/server/websocket.go
+++ b/vendor/github.com/nats-io/nats-server/v2/server/websocket.go
@@ -125,12 +125,17 @@ type srvWebsocket struct {
 	server         *http.Server
 	listener       net.Listener
 	listenerErr    error
-	tls            bool
 	allowedOrigins map[string]*allowedOrigin // host will be the key
 	sameOrigin     bool
 	connectURLs    []string
 	connectURLsMap refCountedUrlSet
 	authOverride   bool // indicate if there is auth override in websocket config
+
+	// These are immutable and can be accessed without lock.
+	// This is the case when generating the client INFO.
+	tls  bool   // True if TLS is required (TLSConfig is specified).
+	host string // Host/IP the webserver is listening on (shortcut to opts.Websocket.Host).
+	port int    // Port the webserver is listening on. This is after an ephemeral port may have been selected (shortcut to opts.Websocket.Port).
 }

 type allowedOrigin struct {
@@ -1102,7 +1107,12 @@ func (s *Server) startWebsocketServer() {
 		s.Warnf("Websocket not configured with TLS. DO NOT USE IN PRODUCTION!")
 	}

-	s.websocket.tls = proto == "wss"
+	// These 3 are immutable and will be accessed without lock by the client
+	// when generating/sending the INFO protocols.
+	s.websocket.tls = proto == wsSchemePrefixTLS
+	s.websocket.host, s.websocket.port = o.Host, o.Port
+
+	// This will be updated when/if the cluster changes.
 	s.websocket.connectURLs, err = s.getConnectURLs(o.Advertise, o.Host, o.Port)
 	if err != nil {
 		s.Fatalf("Unable to get websocket connect URLs: %v", err)
@@ -1141,8 +1151,10 @@ func (s *Server) startWebsocketServer() {
 		ReadTimeout: o.HandshakeTimeout,
 		ErrorLog:    log.New(&captureHTTPServerLog{s, "websocket: "}, _EMPTY_, 0),
 	}
+	s.websocket.mu.Lock()
 	s.websocket.server = hs
 	s.websocket.listener = hl
+	s.websocket.mu.Unlock()
 	go func() {
 		if err := hs.Serve(hl); err != http.ErrServerClosed {
 			s.Fatalf("websocket listener error: %v", err)
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1406,10 +1406,10 @@ github.com/mohae/deepcopy
 # github.com/mschoch/smat v0.2.0
 ## explicit; go 1.13
 github.com/mschoch/smat
-# github.com/nats-io/jwt/v2 v2.5.5
+# github.com/nats-io/jwt/v2 v2.5.6
 ## explicit; go 1.18
 github.com/nats-io/jwt/v2
-# github.com/nats-io/nats-server/v2 v2.10.14
+# github.com/nats-io/nats-server/v2 v2.10.15
 ## explicit; go 1.20
 github.com/nats-io/nats-server/v2/conf
 github.com/nats-io/nats-server/v2/internal/fastrand