Bump github.com/nats-io/nats-server/v2 from 2.10.2 to 2.10.4

Bumps [github.com/nats-io/nats-server/v2](https://github.com/nats-io/nats-server) from 2.10.2 to 2.10.4. - [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.2...v2.10.4) --- updated-dependencies: - dependency-name: github.com/nats-io/nats-server/v2 dependency-type: direct:production ... Signed-off-by: dependabot[bot] <support@github.com>
2023-11-02 11:41:06 +01:00
parent 378bcd55ab
commit 5c3b2991b1
35 changed files with 938 additions and 387 deletions
@@ -57,7 +57,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.2
+	github.com/nats-io/nats-server/v2 v2.10.4
 	github.com/oklog/run v1.1.0
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/onsi/ginkgo v1.16.5
@@ -243,7 +243,7 @@ require (
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/juliangruber/go-intersect v1.1.0 // indirect
 	github.com/kevinburke/ssh_config v1.2.0 // indirect
-	github.com/klauspost/compress v1.17.0 // indirect
+	github.com/klauspost/compress v1.17.2 // indirect
 	github.com/klauspost/cpuid/v2 v2.1.0 // indirect
 	github.com/leodido/go-urn v1.2.0 // indirect
 	github.com/libregraph/oidc-go v1.0.0 // indirect
@@ -272,8 +272,8 @@ require (
 	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.2 // indirect
-	github.com/nats-io/nats.go v1.30.2 // indirect
-	github.com/nats-io/nkeys v0.4.5 // indirect
+	github.com/nats-io/nats.go v1.31.0 // indirect
+	github.com/nats-io/nkeys v0.4.6 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/nxadm/tail v1.4.8 // indirect
 	github.com/opencontainers/runtime-spec v1.0.3-0.20210326190908-1c3f411f0417 // indirect
@@ -1568,8 +1568,8 @@ github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/asmfmt v1.3.2/go.mod h1:AG8TuvYojzulgDAMCnYn50l/5QV3Bs/tp6j0HLHbNSE=
 github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
-github.com/klauspost/compress v1.17.0 h1:Rnbp4K9EjcDuVuHtd0dgA4qNuv9yKDYKK1ulpJwgrqM=
-github.com/klauspost/compress v1.17.0/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4=
+github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
 github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.0.4/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
@@ -1723,12 +1723,12 @@ github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRW
 github.com/namedotcom/go v0.0.0-20180403034216-08470befbe04/go.mod h1:5sN+Lt1CaY4wsPvgQH/jsuJi4XO2ssZbdsIizr4CVC8=
 github.com/nats-io/jwt/v2 v2.5.2 h1:DhGH+nKt+wIkDxM6qnVSKjokq5t59AZV5HRcFW0zJwU=
 github.com/nats-io/jwt/v2 v2.5.2/go.mod h1:24BeQtRwxRV8ruvC4CojXlx/WQ/VjuwlYiH+vu/+ibI=
-github.com/nats-io/nats-server/v2 v2.10.2 h1:2o/OOyc/dxeMCQtrF1V/9er0SU0A3LKhDlv/+rqreBM=
-github.com/nats-io/nats-server/v2 v2.10.2/go.mod h1:lzrskZ/4gyMAh+/66cCd+q74c6v7muBypzfWhP/MAaM=
-github.com/nats-io/nats.go v1.30.2 h1:aloM0TGpPorZKQhbAkdCzYDj+ZmsJDyeo3Gkbr72NuY=
-github.com/nats-io/nats.go v1.30.2/go.mod h1:dcfhUgmQNN4GJEfIb2f9R7Fow+gzBF4emzDHrVBd5qM=
-github.com/nats-io/nkeys v0.4.5 h1:Zdz2BUlFm4fJlierwvGK+yl20IAKUm7eV6AAZXEhkPk=
-github.com/nats-io/nkeys v0.4.5/go.mod h1:XUkxdLPTufzlihbamfzQ7mw/VGx6ObUs+0bN5sNvt64=
+github.com/nats-io/nats-server/v2 v2.10.4 h1:uB9xcwon3tPXWAdmTJqqqC6cie3yuPWHJjjTBgaPNus=
+github.com/nats-io/nats-server/v2 v2.10.4/go.mod h1:eWm2JmHP9Lqm2oemB6/XGi0/GwsZwtWf8HIPUsh+9ns=
+github.com/nats-io/nats.go v1.31.0 h1:/WFBHEc/dOKBF6qf1TZhrdEfTmOZ5JzdJ+Y3m6Y/p7E=
+github.com/nats-io/nats.go v1.31.0/go.mod h1:di3Bm5MLsoB4Bx61CBTsxuarI36WbhAwOm8QrW39+i8=
+github.com/nats-io/nkeys v0.4.6 h1:IzVe95ru2CT6ta874rt9saQRkWfe2nFj1NtvYSLqMzY=
+github.com/nats-io/nkeys v0.4.6/go.mod h1:4DxZNzenSVd1cYQoAa8948QY3QDjrHfcfVADymtkpts=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/nbio/st v0.0.0-20140626010706-e9e8d9816f32/go.mod h1:9wM+0iRr9ahx58uYLpLIr5fm8diHn0JbqRycJi6w0Ms=
@@ -120,8 +120,9 @@ func (h *huffmanDecoder) init(lengths []int) bool {
 	const sanity = false

 	if h.chunks == nil {
-		h.chunks = &[huffmanNumChunks]uint16{}
+		h.chunks = new([huffmanNumChunks]uint16)
 	}
+
 	if h.maxRead != 0 {
 		*h = huffmanDecoder{chunks: h.chunks, links: h.links}
 	}
@@ -175,6 +176,7 @@ func (h *huffmanDecoder) init(lengths []int) bool {
 	}

 	h.maxRead = min
+
 	chunks := h.chunks[:]
 	for i := range chunks {
 		chunks[i] = 0
@@ -202,8 +204,7 @@ func (h *huffmanDecoder) init(lengths []int) bool {
 			if cap(h.links[off]) < numLinks {
 				h.links[off] = make([]uint16, numLinks)
 			} else {
-				links := h.links[off][:0]
-				h.links[off] = links[:numLinks]
+				h.links[off] = h.links[off][:numLinks]
 			}
 		}
 	} else {
@@ -277,7 +278,7 @@ func (h *huffmanDecoder) init(lengths []int) bool {
 	return true
 }

-// The actual read interface needed by NewReader.
+// Reader is the actual read interface needed by NewReader.
 // If the passed in io.Reader does not also have ReadByte,
 // the NewReader will introduce its own buffering.
 type Reader interface {
@@ -285,6 +286,18 @@ type Reader interface {
 	io.ByteReader
 }

+type step uint8
+
+const (
+	copyData step = iota + 1
+	nextBlock
+	huffmanBytesBuffer
+	huffmanBytesReader
+	huffmanBufioReader
+	huffmanStringsReader
+	huffmanGenericReader
+)
+
 // Decompress state.
 type decompressor struct {
 	// Input source.
@@ -303,7 +316,7 @@ type decompressor struct {

 	// Next step in the decompression,
 	// and decompression state.
-	step      func(*decompressor)
+	step      step
 	stepState int
 	err       error
 	toRead    []byte
@@ -342,7 +355,7 @@ func (f *decompressor) nextBlock() {
 		// compressed, fixed Huffman tables
 		f.hl = &fixedHuffmanDecoder
 		f.hd = nil
-		f.huffmanBlockDecoder()()
+		f.huffmanBlockDecoder()
 		if debugDecode {
 			fmt.Println("predefinied huffman block")
 		}
@@ -353,7 +366,7 @@ func (f *decompressor) nextBlock() {
 		}
 		f.hl = &f.h1
 		f.hd = &f.h2
-		f.huffmanBlockDecoder()()
+		f.huffmanBlockDecoder()
 		if debugDecode {
 			fmt.Println("dynamic huffman block")
 		}
@@ -379,14 +392,16 @@ func (f *decompressor) Read(b []byte) (int, error) {
 		if f.err != nil {
 			return 0, f.err
 		}
-		f.step(f)
+
+		f.doStep()
+
 		if f.err != nil && len(f.toRead) == 0 {
 			f.toRead = f.dict.readFlush() // Flush what's left in case of error
 		}
 	}
 }

-// Support the io.WriteTo interface for io.Copy and friends.
+// WriteTo implements the io.WriteTo interface for io.Copy and friends.
 func (f *decompressor) WriteTo(w io.Writer) (int64, error) {
 	total := int64(0)
 	flushed := false
@@ -410,7 +425,7 @@ func (f *decompressor) WriteTo(w io.Writer) (int64, error) {
 			return total, f.err
 		}
 		if f.err == nil {
-			f.step(f)
+			f.doStep()
 		}
 		if len(f.toRead) == 0 && f.err != nil && !flushed {
 			f.toRead = f.dict.readFlush() // Flush what's left in case of error
@@ -631,7 +646,7 @@ func (f *decompressor) copyData() {

 	if f.dict.availWrite() == 0 || f.copyLen > 0 {
 		f.toRead = f.dict.readFlush()
-		f.step = (*decompressor).copyData
+		f.step = copyData
 		return
 	}
 	f.finishBlock()
@@ -644,7 +659,28 @@ func (f *decompressor) finishBlock() {
 		}
 		f.err = io.EOF
 	}
-	f.step = (*decompressor).nextBlock
+	f.step = nextBlock
+}
+
+func (f *decompressor) doStep() {
+	switch f.step {
+	case copyData:
+		f.copyData()
+	case nextBlock:
+		f.nextBlock()
+	case huffmanBytesBuffer:
+		f.huffmanBytesBuffer()
+	case huffmanBytesReader:
+		f.huffmanBytesReader()
+	case huffmanBufioReader:
+		f.huffmanBufioReader()
+	case huffmanStringsReader:
+		f.huffmanStringsReader()
+	case huffmanGenericReader:
+		f.huffmanGenericReader()
+	default:
+		panic("BUG: unexpected step state")
+	}
 }

 // noEOF returns err, unless err == io.EOF, in which case it returns io.ErrUnexpectedEOF.
@@ -747,7 +783,7 @@ func (f *decompressor) Reset(r io.Reader, dict []byte) error {
 		h1:       f.h1,
 		h2:       f.h2,
 		dict:     f.dict,
-		step:     (*decompressor).nextBlock,
+		step:     nextBlock,
 	}
 	f.dict.init(maxMatchOffset, dict)
 	return nil
@@ -768,7 +804,7 @@ func NewReader(r io.Reader) io.ReadCloser {
 	f.r = makeReader(r)
 	f.bits = new([maxNumLit + maxNumDist]int)
 	f.codebits = new([numCodes]int)
-	f.step = (*decompressor).nextBlock
+	f.step = nextBlock
 	f.dict.init(maxMatchOffset, nil)
 	return &f
 }
@@ -787,7 +823,7 @@ func NewReaderDict(r io.Reader, dict []byte) io.ReadCloser {
 	f.r = makeReader(r)
 	f.bits = new([maxNumLit + maxNumDist]int)
 	f.codebits = new([numCodes]int)
-	f.step = (*decompressor).nextBlock
+	f.step = nextBlock
 	f.dict.init(maxMatchOffset, dict)
 	return &f
 }
@@ -85,7 +85,7 @@ readLiteral:
 			dict.writeByte(byte(v))
 			if dict.availWrite() == 0 {
 				f.toRead = dict.readFlush()
-				f.step = (*decompressor).huffmanBytesBuffer
+				f.step = huffmanBytesBuffer
 				f.stepState = stateInit
 				f.b, f.nb = fb, fnb
 				return
@@ -251,7 +251,7 @@ copyHistory:

 		if dict.availWrite() == 0 || f.copyLen > 0 {
 			f.toRead = dict.readFlush()
-			f.step = (*decompressor).huffmanBytesBuffer // We need to continue this work
+			f.step = huffmanBytesBuffer // We need to continue this work
 			f.stepState = stateDict
 			f.b, f.nb = fb, fnb
 			return
@@ -336,7 +336,7 @@ readLiteral:
 			dict.writeByte(byte(v))
 			if dict.availWrite() == 0 {
 				f.toRead = dict.readFlush()
-				f.step = (*decompressor).huffmanBytesReader
+				f.step = huffmanBytesReader
 				f.stepState = stateInit
 				f.b, f.nb = fb, fnb
 				return
@@ -502,7 +502,7 @@ copyHistory:

 		if dict.availWrite() == 0 || f.copyLen > 0 {
 			f.toRead = dict.readFlush()
-			f.step = (*decompressor).huffmanBytesReader // We need to continue this work
+			f.step = huffmanBytesReader // We need to continue this work
 			f.stepState = stateDict
 			f.b, f.nb = fb, fnb
 			return
@@ -587,7 +587,7 @@ readLiteral:
 			dict.writeByte(byte(v))
 			if dict.availWrite() == 0 {
 				f.toRead = dict.readFlush()
-				f.step = (*decompressor).huffmanBufioReader
+				f.step = huffmanBufioReader
 				f.stepState = stateInit
 				f.b, f.nb = fb, fnb
 				return
@@ -753,7 +753,7 @@ copyHistory:

 		if dict.availWrite() == 0 || f.copyLen > 0 {
 			f.toRead = dict.readFlush()
-			f.step = (*decompressor).huffmanBufioReader // We need to continue this work
+			f.step = huffmanBufioReader // We need to continue this work
 			f.stepState = stateDict
 			f.b, f.nb = fb, fnb
 			return
@@ -838,7 +838,7 @@ readLiteral:
 			dict.writeByte(byte(v))
 			if dict.availWrite() == 0 {
 				f.toRead = dict.readFlush()
-				f.step = (*decompressor).huffmanStringsReader
+				f.step = huffmanStringsReader
 				f.stepState = stateInit
 				f.b, f.nb = fb, fnb
 				return
@@ -1004,7 +1004,7 @@ copyHistory:

 		if dict.availWrite() == 0 || f.copyLen > 0 {
 			f.toRead = dict.readFlush()
-			f.step = (*decompressor).huffmanStringsReader // We need to continue this work
+			f.step = huffmanStringsReader // We need to continue this work
 			f.stepState = stateDict
 			f.b, f.nb = fb, fnb
 			return
@@ -1089,7 +1089,7 @@ readLiteral:
 			dict.writeByte(byte(v))
 			if dict.availWrite() == 0 {
 				f.toRead = dict.readFlush()
-				f.step = (*decompressor).huffmanGenericReader
+				f.step = huffmanGenericReader
 				f.stepState = stateInit
 				f.b, f.nb = fb, fnb
 				return
@@ -1255,7 +1255,7 @@ copyHistory:

 		if dict.availWrite() == 0 || f.copyLen > 0 {
 			f.toRead = dict.readFlush()
-			f.step = (*decompressor).huffmanGenericReader // We need to continue this work
+			f.step = huffmanGenericReader // We need to continue this work
 			f.stepState = stateDict
 			f.b, f.nb = fb, fnb
 			return
@@ -1265,19 +1265,19 @@ copyHistory:
 	// Not reached
 }

-func (f *decompressor) huffmanBlockDecoder() func() {
+func (f *decompressor) huffmanBlockDecoder() {
 	switch f.r.(type) {
 	case *bytes.Buffer:
-		return f.huffmanBytesBuffer
+		f.huffmanBytesBuffer()
 	case *bytes.Reader:
-		return f.huffmanBytesReader
+		f.huffmanBytesReader()
 	case *bufio.Reader:
-		return f.huffmanBufioReader
+		f.huffmanBufioReader()
 	case *strings.Reader:
-		return f.huffmanStringsReader
+		f.huffmanStringsReader()
 	case Reader:
-		return f.huffmanGenericReader
+		f.huffmanGenericReader()
 	default:
-		return f.huffmanGenericReader
+		f.huffmanGenericReader()
 	}
 }
@@ -57,7 +57,7 @@ func Encode(dst, src []byte) []byte {
 // The function returns -1 if no improvement could be achieved.
 // Using actual compression will most often produce better compression than the estimate.
 func EstimateBlockSize(src []byte) (d int) {
-	if len(src) < 6 || int64(len(src)) > 0xffffffff {
+	if len(src) <= inputMargin || int64(len(src)) > 0xffffffff {
 		return -1
 	}
 	if len(src) <= 1024 {
@@ -157,6 +157,9 @@ func encodeBlockBest(dst, src []byte, dict *Dict) (d int) {
 				return m
 			}
 			matchDict := func(candidate, s int, first uint32, rep bool) match {
+				if s >= MaxDictSrcOffset {
+					return match{offset: candidate, s: s}
+				}
 				// Calculate offset as if in continuous array with s
 				offset := -len(dict.dict) + candidate
 				if best.length != 0 && best.s-best.offset == s-offset && !rep {
@@ -316,6 +316,7 @@ func matchLen(a []byte, b []byte) int {
 	return len(a) + checked
 }

+// input must be > inputMargin
 func calcBlockSize(src []byte) (d int) {
 	// Initialize the hash table.
 	const (
@@ -501,6 +502,7 @@ emitRemainder:
 	return d
 }

+// length must be > inputMargin.
 func calcBlockSizeSmall(src []byte) (d int) {
 	// Initialize the hash table.
 	const (
@@ -141,6 +141,7 @@ const (
 	expectConnect                                 // Marks if this connection is expected to send a CONNECT
 	connectProcessFinished                        // Marks if this connection has finished the connect process.
 	compressionNegotiated                         // Marks if this connection has negotiated compression level with remote.
+	didTLSFirst                                   // Marks if this connection requested and was accepted doing the TLS handshake first (prior to INFO).
 )

 // set the flag (would be equivalent to set the boolean to true)
@@ -2800,6 +2801,11 @@ func (c *client) addShadowSubscriptions(acc *Account, sub *subscription) error {
 	)

 	acc.mu.RLock()
+	// If this is from a service import, ignore.
+	if sub.si {
+		acc.mu.RUnlock()
+		return nil
+	}
 	subj := string(sub.subject)
 	if len(acc.imports.streams) > 0 {
 		tokens = tokenizeSubjectIntoSlice(tsa[:0], subj)
@@ -5226,7 +5232,7 @@ func (c *client) reconnect() {

 	// Check for a solicited route. If it was, start up a reconnect unless
 	// we are already connected to the other end.
-	if c.isSolicitedRoute() || retryImplicit {
+	if didSolicit := c.isSolicitedRoute(); didSolicit || retryImplicit {
 		srv.mu.Lock()
 		defer srv.mu.Unlock()

@@ -5236,7 +5242,7 @@ func (c *client) reconnect() {
 		rtype := c.route.routeType
 		rurl := c.route.url
 		accName := string(c.route.accName)
-		checkRID := accName == _EMPTY_ && srv.routesPoolSize <= 1 && rid != _EMPTY_
+		checkRID := accName == _EMPTY_ && srv.getOpts().Cluster.PoolSize < 1 && rid != _EMPTY_
 		c.mu.Unlock()

 		// It is possible that the server is being shutdown.
@@ -5246,6 +5252,14 @@ func (c *client) reconnect() {
 		}

 		if checkRID && srv.routes[rid] != nil {
+			// This is the case of "no pool". Make sure that the registered one
+			// is upgraded to solicited if the connection trying to reconnect
+			// was a solicited one.
+			if didSolicit {
+				if remote := srv.routes[rid][0]; remote != nil {
+					upgradeRouteToSolicited(remote, rurl, rtype)
+				}
+			}
 			srv.Debugf("Not attempting reconnect for solicited route, already connected to %q", rid)
 			return
 		} else if rid == srv.info.ID {
@@ -5305,10 +5319,10 @@ func (c *client) getAccAndResultFromCache() (*Account, *SublistResult) {
 	// Check our cache.
 	if pac, ok = c.in.pacache[string(c.pa.pacache)]; ok {
 		// Check the genid to see if it's still valid.
-		// sl could be swapped out on reload so need to lock.
-		pac.acc.mu.RLock()
+		// Since v2.10.0, the config reload of accounts has been fixed
+		// and an account's sublist pointer should not change, so no need to
+		// lock to access it.
 		sl := pac.acc.sl
-		pac.acc.mu.RUnlock()

 		if genid := atomic.LoadUint64(&sl.genid); genid != pac.genid {
 			ok = false
@@ -5320,15 +5334,15 @@ func (c *client) getAccAndResultFromCache() (*Account, *SublistResult) {
 	}

 	if !ok {
-		// Match correct account and sublist.
-		if acc, _ = c.srv.LookupAccount(string(c.pa.account)); acc == nil {
-			return nil, nil
+		if c.kind == ROUTER && len(c.route.accName) > 0 {
+			acc = c.acc
+		} else {
+			// Match correct account and sublist.
+			if acc, _ = c.srv.LookupAccount(string(c.pa.account)); acc == nil {
+				return nil, nil
+			}
 		}
-
-		// sl could be swapped out on reload so need to lock.
-		acc.mu.RLock()
 		sl := acc.sl
-		acc.mu.RUnlock()

 		// Match against the account sublist.
 		r = sl.Match(string(c.pa.subject))
@@ -41,7 +41,7 @@ var (

 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.10.2"
+	VERSION = "2.10.4"

 	// PROTO is the currently supported protocol.
 	// 0 was the original
@@ -82,6 +82,12 @@ const (
 	// TLS_TIMEOUT is the TLS wait time.
 	TLS_TIMEOUT = 2 * time.Second

+	// DEFAULT_TLS_HANDSHAKE_FIRST_FALLBACK_DELAY is the default amount of
+	// time for the server to wait for the TLS handshake with a client to
+	// be initiated before falling back to sending the INFO protocol first.
+	// See TLSHandshakeFirst and TLSHandshakeFirstFallback options.
+	DEFAULT_TLS_HANDSHAKE_FIRST_FALLBACK_DELAY = 50 * time.Millisecond
+
 	// AUTH_TIMEOUT is the authorization wait time.
 	AUTH_TIMEOUT = 2 * time.Second

@@ -272,6 +272,10 @@ var (
 	AckNext = []byte("+NXT")
 	// Terminate delivery of the message.
 	AckTerm = []byte("+TERM")
+
+	// reasons to supply when terminating messages using limits
+	ackTermLimitsReason        = "Message deleted by stream limits"
+	ackTermUnackedLimitsReason = "Unacknowledged message was deleted"
 )

 // Calculate accurate replicas for the consumer config with the parent stream config.
@@ -292,6 +296,7 @@ type consumer struct {
 	// This will be checked in checkPending to abort processing
 	// and let ack be processed in priority.
 	awl               int64
+	leader            atomic.Bool
 	mu                sync.RWMutex
 	js                *jetStream
 	mset              *stream
@@ -767,6 +772,13 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 			if action == ActionCreate && !reflect.DeepEqual(*config, eo.config()) {
 				return nil, NewJSConsumerAlreadyExistsError()
 			}
+			// Check for overlapping subjects.
+			if mset.cfg.Retention == WorkQueuePolicy {
+				subjects := gatherSubjectFilters(config.FilterSubject, config.FilterSubjects)
+				if !mset.partitionUnique(cName, subjects) {
+					return nil, NewJSConsumerWQConsumerNotUniqueError()
+				}
+			}
 			err := eo.updateConfig(config)
 			if err == nil {
 				return eo, nil
@@ -805,7 +817,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 			if len(subjects) == 0 {
 				mset.mu.Unlock()
 				return nil, NewJSConsumerWQMultipleUnfilteredError()
-			} else if !mset.partitionUnique(subjects) {
+			} else if !mset.partitionUnique(cName, subjects) {
 				// Prior to v2.9.7, on a stream with WorkQueue policy, the servers
 				// were not catching the error of having multiple consumers with
 				// overlapping filter subjects depending on the scope, for instance
@@ -1124,19 +1136,21 @@ func (o *consumer) isLeader() bool {

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

 	// If we are here we have a change in leader status.
 	if isLeader {
-		if mset == nil {
+		if closed || mset == nil {
 			return
 		}
-		if isRunning {
+
+		if wasLeader {
 			// If we detect we are scaling up, make sure to create clustered routines and channels.
-			o.mu.Lock()
-			if o.node != nil && o.pch == nil {
+			if movingToClustered {
+				o.mu.Lock()
 				// We are moving from R1 to clustered.
 				o.pch = make(chan struct{}, 1)
 				go o.loopAndForwardProposals(o.qch)
@@ -1146,8 +1160,8 @@ func (o *consumer) setLeader(isLeader bool) {
 					default:
 					}
 				}
+				o.mu.Unlock()
 			}
-			o.mu.Unlock()
 			return
 		}

@@ -1981,8 +1995,12 @@ func (o *consumer) processAck(subject, reply string, hdr int, rmsg []byte) {
 		o.processNak(sseq, dseq, dc, msg)
 	case bytes.Equal(msg, AckProgress):
 		o.progressUpdate(sseq)
-	case bytes.Equal(msg, AckTerm):
-		o.processTerm(sseq, dseq, dc)
+	case bytes.HasPrefix(msg, AckTerm):
+		var reason string
+		if buf := msg[len(AckTerm):]; len(buf) > 0 {
+			reason = string(bytes.TrimSpace(buf))
+		}
+		o.processTerm(sseq, dseq, dc, reason)
 	}

 	// Ack the ack if requested.
@@ -2309,7 +2327,7 @@ func (o *consumer) processNak(sseq, dseq, dc uint64, nak []byte) {
 }

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

@@ -2328,6 +2346,7 @@ func (o *consumer) processTerm(sseq, dseq, dc uint64) {
 		ConsumerSeq: dseq,
 		StreamSeq:   sseq,
 		Deliveries:  dc,
+		Reason:      reason,
 		Domain:      o.srv.getOpts().JetStreamDomain,
 	}

@@ -2536,7 +2555,11 @@ func (o *consumer) infoWithSnapAndReply(snap bool, reply string) *ConsumerInfo {

 	// If we are replicated and we are not the leader we need to pull certain data from our store.
 	if rg != nil && rg.node != nil && !o.isLeader() && o.store != nil {
-		state, _ := o.store.BorrowState()
+		state, err := o.store.BorrowState()
+		if err != nil {
+			o.mu.Unlock()
+			return nil
+		}
 		info.Delivered.Consumer, info.Delivered.Stream = state.Delivered.Consumer, state.Delivered.Stream
 		info.AckFloor.Consumer, info.AckFloor.Stream = state.AckFloor.Consumer, state.AckFloor.Stream
 		info.NumAckPending = len(state.Pending)
@@ -3636,7 +3659,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)
+				o.processTerm(seq, p.Sequence, rdc, ackTermLimitsReason)
 			}
 		}
 	} else if numPending > 0 {
@@ -3661,7 +3684,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)
+			o.processTerm(seq, dseq, rdc, ackTermLimitsReason)
 		}
 	}

@@ -4634,16 +4657,6 @@ type lastSeqSkipList struct {
 	seqs   []uint64
 }

-// Will create a skip list for us from a store's subjects state.
-func createLastSeqSkipList(mss map[string]SimpleState) []uint64 {
-	seqs := make([]uint64, 0, len(mss))
-	for _, ss := range mss {
-		seqs = append(seqs, ss.Last)
-	}
-	sort.Slice(seqs, func(i, j int) bool { return seqs[i] < seqs[j] })
-	return seqs
-}
-
 // Let's us know we have a skip list, which is for deliver last per subject and we are just starting.
 // Lock should be held.
 func (o *consumer) hasSkipListPending() bool {
@@ -4674,37 +4687,49 @@ func (o *consumer) selectStartingSeqNo() {
 					}
 				}
 			} else if o.cfg.DeliverPolicy == DeliverLastPerSubject {
-				if o.subjf == nil {
-					if mss := o.mset.store.SubjectsState(o.cfg.FilterSubject); len(mss) > 0 {
-						o.lss = &lastSeqSkipList{
-							resume: state.LastSeq,
-							seqs:   createLastSeqSkipList(mss),
-						}
-						o.sseq = o.lss.seqs[0]
+				// If our parent stream is set to max msgs per subject of 1 this is just
+				// a normal consumer at this point. We can avoid any heavy lifting.
+				if o.mset.cfg.MaxMsgsPer == 1 {
+					o.sseq = state.FirstSeq
+				} else {
+					// A threshold for when we switch from get last msg to subjects state.
+					const numSubjectsThresh = 256
+					lss := &lastSeqSkipList{resume: state.LastSeq}
+					var filters []string
+					if o.subjf == nil {
+						filters = append(filters, o.cfg.FilterSubject)
 					} else {
-						// If no mapping info just set to last.
+						for _, filter := range o.subjf {
+							filters = append(filters, filter.subject)
+						}
+					}
+					for _, filter := range filters {
+						if st := o.mset.store.SubjectsTotals(filter); len(st) < numSubjectsThresh {
+							var smv StoreMsg
+							for subj := range st {
+								if sm, err := o.mset.store.LoadLastMsg(subj, &smv); err == nil {
+									lss.seqs = append(lss.seqs, sm.seq)
+								}
+							}
+						} else if mss := o.mset.store.SubjectsState(filter); len(mss) > 0 {
+							for _, ss := range mss {
+								lss.seqs = append(lss.seqs, ss.Last)
+							}
+						}
+					}
+					// Sort the skip list if needed.
+					if len(lss.seqs) > 1 {
+						sort.Slice(lss.seqs, func(i, j int) bool {
+							return lss.seqs[j] > lss.seqs[i]
+						})
+					}
+					if len(lss.seqs) == 0 {
 						o.sseq = state.LastSeq
+					} else {
+						o.sseq = lss.seqs[0]
 					}
-					return
-				}
-				lss := &lastSeqSkipList{
-					resume: state.LastSeq,
-				}
-				for _, filter := range o.subjf {
-					if mss := o.mset.store.SubjectsState(filter.subject); len(mss) > 0 {
-						lss.seqs = append(lss.seqs, createLastSeqSkipList(mss)...)
-					}
-				}
-				if len(lss.seqs) == 0 {
-					o.sseq = state.LastSeq
-				}
-				// Sort the skip list
-				sort.Slice(lss.seqs, func(i, j int) bool {
-					return lss.seqs[j] > lss.seqs[i]
-				})
-				o.lss = lss
-				if len(o.lss.seqs) != 0 {
-					o.sseq = o.lss.seqs[0]
+					// Assign skip list.
+					o.lss = lss
 				}
 			} else if o.cfg.OptStartTime != nil {
 				// If we are here we are time based.
@@ -5137,11 +5162,10 @@ func (o *consumer) decStreamPending(sseq uint64, subj string) {
 	o.mu.Unlock()

 	// If it was pending process it like an ack.
-	// TODO(dlc) - we could do a term here instead with a reason to generate the advisory.
 	if wasPending {
 		// We could have lock for 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)
+		go o.processTerm(sseq, p.Sequence, rdc, ackTermUnackedLimitsReason)
 	}
 }

@@ -1598,14 +1598,16 @@ func (s *Server) shutdownEventing() {
 	s.mu.Lock()
 	clearTimer(&s.sys.sweeper)
 	clearTimer(&s.sys.stmr)
-	sys := s.sys
+	rc := s.sys.resetCh
+	s.sys.resetCh = nil
+	wg := &s.sys.wg
 	s.mu.Unlock()

 	// We will queue up a shutdown event and wait for the
 	// internal send loop to exit.
 	s.sendShutdownEvent()
-	sys.wg.Wait()
-	close(sys.resetCh)
+	wg.Wait()
+	close(rc)

 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -176,6 +176,7 @@ type fileStore struct {
 	blks        []*msgBlock
 	bim         map[uint32]*msgBlock
 	psim        map[string]*psi
+	tsl         int
 	hh          hash.Hash64
 	qch         chan struct{}
 	fch         chan struct{}
@@ -428,7 +429,7 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 		prior := fs.state
 		// Reset anything that could have been set from above.
 		fs.state = StreamState{}
-		fs.psim = make(map[string]*psi)
+		fs.psim, fs.tsl = make(map[string]*psi), 0
 		fs.bim = make(map[uint32]*msgBlock)
 		fs.blks = nil
 		fs.tombs = nil
@@ -467,7 +468,7 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 	// Check if we have any left over tombstones to process.
 	if len(fs.tombs) > 0 {
 		for _, seq := range fs.tombs {
-			fs.removeMsg(seq, false, false, false)
+			fs.removeMsg(seq, false, true, false)
 			fs.removeFromLostData(seq)
 		}
 		// Not needed after this phase.
@@ -486,7 +487,7 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim

 	// If we have max msgs per subject make sure the is also enforced.
 	if fs.cfg.MaxMsgsPer > 0 {
-		fs.enforceMsgPerSubjectLimit()
+		fs.enforceMsgPerSubjectLimit(false)
 	}

 	// Grab first sequence for check below while we have lock.
@@ -589,7 +590,7 @@ func (fs *fileStore) UpdateConfig(cfg *StreamConfig) error {
 	}

 	if fs.cfg.MaxMsgsPer > 0 && fs.cfg.MaxMsgsPer < old_cfg.MaxMsgsPer {
-		fs.enforceMsgPerSubjectLimit()
+		fs.enforceMsgPerSubjectLimit(true)
 	}
 	fs.mu.Unlock()

@@ -937,6 +938,16 @@ 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 := fs.psim
+	fs.psim = nil
+	mb, err := fs.recoverMsgBlock(index)
+	fs.psim = psim
+	return mb, err
+}
+
 // Lock held on entry
 func (fs *fileStore) recoverMsgBlock(index uint32) (*msgBlock, error) {
 	mb := fs.initMsgBlock(index)
@@ -1218,6 +1229,8 @@ func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, []uint64, error) {
 	mb.clearCacheAndOffset()

 	buf, err := mb.loadBlock(nil)
+	defer recycleMsgBlockBuf(buf)
+
 	if err != nil || len(buf) == 0 {
 		var ld *LostStreamData
 		// No data to rebuild from here.
@@ -1553,7 +1566,7 @@ func (fs *fileStore) recoverFullState() (rerr error) {

 	// Check for per subject info.
 	if numSubjects := int(readU64()); numSubjects > 0 {
-		fs.psim = make(map[string]*psi, numSubjects)
+		fs.psim, fs.tsl = make(map[string]*psi, numSubjects), 0
 		for i := 0; i < numSubjects; i++ {
 			if lsubj := int(readU64()); lsubj > 0 {
 				if bi+lsubj > len(buf) {
@@ -1570,6 +1583,7 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 					psi.lblk = psi.fblk
 				}
 				fs.psim[subj] = psi
+				fs.tsl += len(subj)
 			}
 		}
 	}
@@ -1646,18 +1660,18 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 		return errPriorState
 	}
 	if matched = bytes.Equal(mb.lastChecksum(), lchk[:]); !matched {
-		// Remove the last message block since we will re-process below.
+		// Remove the last message block since recover will add in the new one.
 		fs.removeMsgBlockFromList(mb)
+		if nmb, err := fs.recoverMsgBlockNoSubjectUpdates(mb.index); err != nil && !os.IsNotExist(err) {
+			os.Remove(fn)
+			return errCorruptState
+		} else if nmb != nil {
+			fs.adjustAccounting(mb, nmb)
+		}
 	}

 	// We may need to check other blocks. Even if we matched last checksum we will see if there is another block.
-	// If we did not match we re-process the last block.
-	start := blkIndex
-	if matched {
-		start++
-	}
-
-	for bi := start; ; bi++ {
+	for bi := blkIndex + 1; ; bi++ {
 		nmb, err := fs.recoverMsgBlock(bi)
 		if err != nil {
 			if os.IsNotExist(err) {
@@ -1668,13 +1682,6 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 			return err
 		}
 		if nmb != nil {
-			// Check if we have to account for a partial message block.
-			if !matched && mb != nil && mb.index == nmb.index {
-				if err := fs.adjustAccounting(mb, nmb); err != nil {
-					fs.warn("Stream state could not adjust accounting")
-					return err
-				}
-			}
 			// Update top level accounting.
 			if fs.state.FirstSeq == 0 || nmb.first.seq < fs.state.FirstSeq {
 				fs.state.FirstSeq = nmb.first.seq
@@ -1691,9 +1698,9 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 }

 // adjustAccounting will be called when a stream state was only partially accounted for
-// with a message block, e.g. additional records were added after the stream state.
+// within a message block, e.g. additional records were added after the stream state.
 // Lock should be held.
-func (fs *fileStore) adjustAccounting(mb, nmb *msgBlock) error {
+func (fs *fileStore) adjustAccounting(mb, nmb *msgBlock) {
 	nmb.mu.Lock()
 	defer nmb.mu.Unlock()

@@ -1703,23 +1710,29 @@ func (fs *fileStore) adjustAccounting(mb, nmb *msgBlock) error {
 	}
 	nmb.ensurePerSubjectInfoLoaded()

-	// Walk all the original mb's sequences that were included in the stream state.
+	// 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 := mb.first.seq; seq <= mb.last.seq; seq++ {
-		// If we had already declared it deleted we can move on since you can not undelete.
-		if mb.dmap.Exists(seq) {
-			continue
-		}
-		// Lookup the message.
+	for seq := mb.last.seq + 1; seq <= nmb.last.seq; seq++ {
+		// Lookup the message. If an error will be deleted, so can skip.
 		sm, err := nmb.cacheLookup(seq, &smv)
 		if err != nil {
-			return err
+			continue
 		}
 		// 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)
+		fs.state.Msgs++
+		fs.state.Bytes += fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
 		if len(sm.subj) > 0 && fs.psim != nil {
-			fs.removePerSubject(sm.subj)
+			if info, ok := fs.psim[sm.subj]; ok {
+				info.total++
+				if nmb.index > info.lblk {
+					info.lblk = nmb.index
+				}
+			} else {
+				fs.psim[sm.subj] = &psi{total: 1, fblk: nmb.index, lblk: nmb.index}
+				fs.tsl += len(sm.subj)
+			}
 		}
 	}

@@ -1729,7 +1742,15 @@ func (fs *fileStore) adjustAccounting(mb, nmb *msgBlock) error {
 		nmb.first = mb.first
 	}

-	return nil
+	// Update top level accounting.
+	if fs.state.FirstSeq == 0 || nmb.first.seq < fs.state.FirstSeq {
+		fs.state.FirstSeq = nmb.first.seq
+		fs.state.FirstTime = time.Unix(0, nmb.first.ts).UTC()
+	}
+	if nmb.last.seq > fs.state.LastSeq {
+		fs.state.LastSeq = nmb.last.seq
+		fs.state.LastTime = time.Unix(0, nmb.last.ts).UTC()
+	}
 }

 // Grabs last checksum for the named block file.
@@ -1907,13 +1928,10 @@ func (fs *fileStore) expireMsgsOnRecover() {
 		}
 		// Make sure we do subject cleanup as well.
 		mb.ensurePerSubjectInfoLoaded()
-		for subj := range mb.fss {
-			fs.removePerSubject(subj)
-		}
-		// Make sure we do subject cleanup as well.
-		mb.ensurePerSubjectInfoLoaded()
-		for subj := range mb.fss {
-			fs.removePerSubject(subj)
+		for subj, ss := range mb.fss {
+			for i := uint64(0); i < ss.Msgs; i++ {
+				fs.removePerSubject(subj)
+			}
 		}
 		mb.dirtyCloseWithRemove(true)
 		deleted++
@@ -2044,7 +2062,7 @@ func (fs *fileStore) expireMsgsOnRecover() {
 			lmb.writeTombstone(last.seq, last.ts)
 		}
 		// Clear any global subject state.
-		fs.psim = make(map[string]*psi)
+		fs.psim, fs.tsl = make(map[string]*psi), 0
 	}

 	// If we purged anything, make sure we kick flush state loop.
@@ -2408,7 +2426,7 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState {
 	fs.mu.RLock()
 	defer fs.mu.RUnlock()

-	if fs.state.Msgs == 0 {
+	if fs.state.Msgs == 0 || fs.noTrackSubjects() {
 		return nil
 	}

@@ -2435,8 +2453,12 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState {
 		}

 		mb.mu.Lock()
-		// Make sure we have fss loaded.
-		mb.ensurePerSubjectInfoLoaded()
+		var shouldExpire bool
+		if mb.fss == nil {
+			// Make sure we have fss loaded.
+			mb.loadMsgsWithLock()
+			shouldExpire = true
+		}
 		for subj, ss := range mb.fss {
 			if subject == _EMPTY_ || subject == fwcs || subjectIsSubsetMatch(subj, subject) {
 				if ss.firstNeedsUpdate {
@@ -2452,6 +2474,10 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState {
 				}
 			}
 		}
+		if shouldExpire {
+			// Expire this cache before moving on.
+			mb.tryForceExpireCacheLocked()
+		}
 		mb.mu.Unlock()

 		if mb == stop {
@@ -2943,6 +2969,7 @@ func (fs *fileStore) storeRawMsg(subj string, hdr, msg []byte, seq uint64, ts in
 			}
 		} else {
 			fs.psim[subj] = &psi{total: 1, fblk: index, lblk: index}
+			fs.tsl += len(subj)
 		}
 	}

@@ -3190,14 +3217,16 @@ func (fs *fileStore) enforceBytesLimit() {
 // We will make sure to go through all msg blocks etc. but in practice this
 // will most likely only be the last one, so can take a more conservative approach.
 // Lock should be held.
-func (fs *fileStore) enforceMsgPerSubjectLimit() {
+func (fs *fileStore) enforceMsgPerSubjectLimit(fireCallback bool) {
 	maxMsgsPer := uint64(fs.cfg.MaxMsgsPer)

-	// We want to suppress callbacks from remove during this process
+	// We may want to suppress callbacks from remove during this process
 	// since these should have already been deleted and accounted for.
-	cb := fs.scb
-	fs.scb = nil
-	defer func() { fs.scb = cb }()
+	if !fireCallback {
+		cb := fs.scb
+		fs.scb = nil
+		defer func() { fs.scb = cb }()
+	}

 	var numMsgs uint64

@@ -3215,7 +3244,7 @@ func (fs *fileStore) enforceMsgPerSubjectLimit() {
 	if numMsgs != fs.state.Msgs {
 		fs.warn("Detected skew in subject-based total (%d) vs raw total (%d), rebuilding", numMsgs, fs.state.Msgs)
 		// Clear any global subject state.
-		fs.psim = make(map[string]*psi)
+		fs.psim, fs.tsl = make(map[string]*psi), 0
 		for _, mb := range fs.blks {
 			ld, _, err := mb.rebuildState()
 			if err != nil && ld != nil {
@@ -3251,6 +3280,9 @@ func (fs *fileStore) enforceMsgPerSubjectLimit() {
 			}
 			// Grab the ss entry for this subject in case sparse.
 			mb.mu.Lock()
+			if mb.cacheNotLoaded() {
+				mb.loadMsgsWithLock()
+			}
 			mb.ensurePerSubjectInfoLoaded()
 			ss := mb.fss[subj]
 			if ss != nil && ss.firstNeedsUpdate {
@@ -3321,6 +3353,7 @@ func (fs *fileStore) removePerSubject(subj string) {
 			info.fblk = info.lblk
 		} else if info.total == 0 {
 			delete(fs.psim, subj)
+			fs.tsl -= len(subj)
 		}
 	}
 }
@@ -3491,14 +3524,13 @@ func (fs *fileStore) removeMsg(seq uint64, secure, viaLimits, needFSLock bool) (
 	} else if !isEmpty {
 		// Out of order delete.
 		mb.dmap.Insert(seq)
-		// Check if <25% utilization and minimum size met.
-		if mb.rbytes > compactMinimum && !isLastBlock {
-			// Remove the interior delete records
-			rbytes := mb.rbytes - uint64(mb.dmap.Size()*emptyRecordLen)
-			if rbytes>>2 > mb.bytes {
-				mb.compact()
-				fs.kickFlushStateLoop()
-			}
+		// Make simple check here similar to Compact(). If we can save 50% and over a certain threshold do inline.
+		// All other more thorough cleanup will happen in syncBlocks logic.
+		// Note that we do not have to store empty records for the deleted, so don't use to calculate.
+		// 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()
 		}
 	}

@@ -3572,7 +3604,9 @@ func (mb *msgBlock) compact() {
 	}

 	buf := mb.cache.buf
-	nbuf := make([]byte, 0, len(buf))
+	nbuf := getMsgBlockBuf(len(buf))
+	// Recycle our nbuf when we are done.
+	defer recycleMsgBlockBuf(nbuf)

 	var le = binary.LittleEndian
 	var firstSet bool
@@ -3622,9 +3656,16 @@ func (mb *msgBlock) compact() {
 	}

 	// Handle compression
-	var err error
-	if nbuf, err = mb.cmp.Compress(nbuf); err != nil {
-		return
+	if mb.cmp != NoCompression {
+		cbuf, err := mb.cmp.Compress(nbuf)
+		if err != nil {
+			return
+		}
+		meta := &CompressionInfo{
+			Algorithm:    mb.cmp,
+			OriginalSize: uint64(len(nbuf)),
+		}
+		nbuf = append(meta.MarshalMetadata(), cbuf...)
 	}

 	// Check for encryption.
@@ -4258,7 +4299,7 @@ func (fs *fileStore) checkMsgs() *LostStreamData {
 	fs.checkAndFlushAllBlocks()

 	// Clear any global subject state.
-	fs.psim = make(map[string]*psi)
+	fs.psim, fs.tsl = make(map[string]*psi), 0

 	for _, mb := range fs.blks {
 		// Make sure encryption loaded if needed for the block.
@@ -4701,6 +4742,24 @@ func (mb *msgBlock) decompressIfNeeded(buf []byte) ([]byte, error) {
 	}
 }

+// Lock should be held.
+func (mb *msgBlock) ensureRawBytesLoaded() error {
+	if mb.rbytes > 0 {
+		return nil
+	}
+	f, err := os.Open(mb.mfn)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	if fi, err := f.Stat(); fi != nil && err == nil {
+		mb.rbytes = uint64(fi.Size())
+	} else {
+		return err
+	}
+	return nil
+}
+
 // Sync msg and index files as needed. This is called from a timer.
 func (fs *fileStore) syncBlocks() {
 	fs.mu.RLock()
@@ -4709,8 +4768,10 @@ func (fs *fileStore) syncBlocks() {
 		return
 	}
 	blks := append([]*msgBlock(nil), fs.blks...)
+	lmb := fs.lmb
 	fs.mu.RUnlock()

+	var markDirty bool
 	for _, mb := range blks {
 		// Do actual sync. Hold lock for consistency.
 		mb.mu.Lock()
@@ -4722,24 +4783,33 @@ func (fs *fileStore) syncBlocks() {
 		if mb.mfd != nil && mb.sinceLastWriteActivity() > closeFDsIdle {
 			mb.dirtyCloseWithRemove(false)
 		}
+		// Check if we should compact here as well.
+		// Do not compact last mb.
+		if mb != lmb && mb.ensureRawBytesLoaded() == nil && mb.rbytes > mb.bytes {
+			mb.compact()
+			markDirty = true
+		}
+
 		// Check if we need to sync. We will not hold lock during actual sync.
-		var fn string
-		if mb.needSync {
+		needSync, fn := mb.needSync, mb.mfn
+		if needSync {
 			// Flush anything that may be pending.
-			if mb.pendingWriteSizeLocked() > 0 {
-				mb.flushPendingMsgsLocked()
-			}
-			fn = mb.mfn
-			mb.needSync = false
+			mb.flushPendingMsgsLocked()
 		}
 		mb.mu.Unlock()

 		// Check if we need to sync.
 		// This is done not holding any locks.
-		if fn != _EMPTY_ {
+		if needSync {
 			if fd, _ := os.OpenFile(fn, os.O_RDWR, defaultFilePerms); fd != nil {
-				fd.Sync()
+				canClear := fd.Sync() == nil
 				fd.Close()
+				// Only clear sync flag on success.
+				if canClear {
+					mb.mu.Lock()
+					mb.needSync = false
+					mb.mu.Unlock()
+				}
 			}
 		}
 	}
@@ -4748,8 +4818,12 @@ func (fs *fileStore) syncBlocks() {
 	fs.syncTmr = time.AfterFunc(fs.fcfg.SyncInterval, fs.syncBlocks)
 	fn := filepath.Join(fs.fcfg.StoreDir, msgDir, streamStreamStateFile)
 	syncAlways := fs.fcfg.SyncAlways
+	if markDirty {
+		fs.dirty++
+	}
 	fs.mu.Unlock()

+	// Sync state file if we are not running with sync always.
 	if !syncAlways {
 		if fd, _ := os.OpenFile(fn, os.O_RDWR, defaultFilePerms); fd != nil {
 			fd.Sync()
@@ -5942,7 +6016,7 @@ func compareFn(subject string) func(string, string) bool {
 // Will return the number of purged messages.
 func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint64, err error) {
 	if subject == _EMPTY_ || subject == fwcs {
-		if keep == 0 && (sequence == 0 || sequence == 1) {
+		if keep == 0 && sequence == 0 {
 			return fs.Purge()
 		}
 		if sequence > 1 {
@@ -6098,7 +6172,7 @@ func (fs *fileStore) purge(fseq uint64) (uint64, error) {
 	fs.lmb = nil
 	fs.bim = make(map[uint32]*msgBlock)
 	// Clear any per subject tracking.
-	fs.psim = make(map[string]*psi)
+	fs.psim, fs.tsl = make(map[string]*psi), 0
 	// Mark dirty
 	fs.dirty++

@@ -6185,8 +6259,10 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 		bytes += mb.bytes
 		// Make sure we do subject cleanup as well.
 		mb.ensurePerSubjectInfoLoaded()
-		for subj := range mb.fss {
-			fs.removePerSubject(subj)
+		for subj, ss := range mb.fss {
+			for i := uint64(0); i < ss.Msgs; i++ {
+				fs.removePerSubject(subj)
+			}
 		}
 		// Now close.
 		mb.dirtyCloseWithRemove(true)
@@ -6369,7 +6445,7 @@ func (fs *fileStore) reset() error {
 	fs.blks, fs.lmb = nil, nil

 	// Reset subject mappings.
-	fs.psim = make(map[string]*psi)
+	fs.psim, fs.tsl = make(map[string]*psi), 0
 	fs.bim = make(map[uint32]*msgBlock)

 	// If we purged anything, make sure we kick flush state loop.
@@ -6629,7 +6705,12 @@ func (mb *msgBlock) recalculateFirstForSubj(subj string, startSeq uint64, ss *Si

 	var le = binary.LittleEndian
 	for slot := startSlot; slot < len(mb.cache.idx); slot++ {
-		li := int(mb.cache.idx[slot]&^hbit) - mb.cache.off
+		bi := mb.cache.idx[slot] &^ hbit
+		if bi == dbit {
+			// delete marker so skip.
+			continue
+		}
+		li := int(bi) - mb.cache.off
 		if li >= len(mb.cache.buf) {
 			ss.First = ss.Last
 			return
@@ -6639,10 +6720,7 @@ func (mb *msgBlock) recalculateFirstForSubj(subj string, startSeq uint64, ss *Si
 		slen := int(le.Uint16(hdr[20:]))
 		if subj == string(buf[msgHdrSize:msgHdrSize+slen]) {
 			seq := le.Uint64(hdr[4:])
-			if seq < mb.first.seq || seq&ebit != 0 {
-				continue
-			}
-			if mb.dmap.Exists(seq) {
+			if seq < mb.first.seq || seq&ebit != 0 || mb.dmap.Exists(seq) {
 				continue
 			}
 			ss.First = seq
@@ -6654,7 +6732,7 @@ func (mb *msgBlock) recalculateFirstForSubj(subj string, startSeq uint64, ss *Si
 // Lock should be held.
 func (fs *fileStore) resetGlobalPerSubjectInfo() {
 	// Clear any global subject state.
-	fs.psim = make(map[string]*psi)
+	fs.psim, fs.tsl = make(map[string]*psi), 0
 	for _, mb := range fs.blks {
 		fs.populateGlobalPerSubjectInfo(mb)
 	}
@@ -6678,7 +6756,7 @@ func (mb *msgBlock) generatePerSubjectInfo() error {
 		if err := mb.loadMsgsWithLock(); err != nil {
 			return err
 		}
-		// indexCaceheBuf can produce fss now, so if non-nil we are good.
+		// indexCacheBuf can produce fss now, so if non-nil we are good.
 		if mb.fss != nil {
 			return nil
 		}
@@ -6752,6 +6830,7 @@ func (fs *fileStore) populateGlobalPerSubjectInfo(mb *msgBlock) {
 				}
 			} else {
 				fs.psim[subj] = &psi{total: ss.Msgs, fblk: mb.index, lblk: mb.index}
+				fs.tsl += len(subj)
 			}
 		}
 	}
@@ -6902,10 +6981,19 @@ func (fs *fileStore) writeFullState() error {
 		return nil
 	}

-	var _buf [32 * 1024]byte
-	_buf[0], _buf[1] = fullStateMagic, fullStateVersion
-	buf := _buf[:hdrLen]
+	// For calculating size.
+	numSubjects := len(fs.psim)

+	// Calculate and estimate of the uper bound on the  size to avoid multiple allocations.
+	sz := 2 + // Magic and Version
+		(binary.MaxVarintLen64 * 6) + // FS data
+		binary.MaxVarintLen64 + fs.tsl + // NumSubjects + total subject length
+		numSubjects*(binary.MaxVarintLen64*4) + // psi record
+		len(fs.blks)*((binary.MaxVarintLen64*6)+512) + // msg blocks, 512 is est for dmap
+		binary.MaxVarintLen64 + 8 // last index + checksum
+
+	buf := make([]byte, hdrLen, sz)
+	buf[0], buf[1] = fullStateMagic, fullStateVersion
 	buf = binary.AppendUvarint(buf, fs.state.Msgs)
 	buf = binary.AppendUvarint(buf, fs.state.Bytes)
 	buf = binary.AppendUvarint(buf, fs.state.FirstSeq)
@@ -6914,9 +7002,7 @@ func (fs *fileStore) writeFullState() error {
 	buf = binary.AppendVarint(buf, timestampNormalized(fs.state.LastTime))

 	// Do per subject information map if applicable.
-	numSubjects := len(fs.psim)
 	buf = binary.AppendUvarint(buf, uint64(numSubjects))
-
 	if numSubjects > 0 {
 		for subj, psi := range fs.psim {
 			buf = binary.AppendUvarint(buf, uint64(len(subj)))
@@ -6938,6 +7024,7 @@ func (fs *fileStore) writeFullState() error {

 	// Use basetime to save some space.
 	baseTime := timestampNormalized(fs.state.FirstTime)
+	var scratch [8 * 1024]byte

 	for _, mb := range fs.blks {
 		mb.mu.RLock()
@@ -6951,7 +7038,6 @@ func (fs *fileStore) writeFullState() error {
 		numDeleted := mb.dmap.Size()
 		buf = binary.AppendUvarint(buf, uint64(numDeleted))
 		if numDeleted > 0 {
-			var scratch [8 * 1024]byte
 			dmap, _ := mb.dmap.Encode(scratch[:0])
 			buf = append(buf, dmap...)
 		}
@@ -6991,6 +7077,10 @@ func (fs *fileStore) writeFullState() error {
 	// Release lock.
 	fs.mu.Unlock()

+	if cap(buf) > sz {
+		fs.warn("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)
@@ -7270,6 +7360,9 @@ func (fs *fileStore) Snapshot(deadline time.Duration, checkMsgs, includeConsumer
 		}
 	}

+	// Write out full state as well before proceeding.
+	fs.writeFullState()
+
 	pr, pw := net.Pipe()

 	// Set a write deadline here to protect ourselves.
@@ -1038,6 +1038,10 @@ func (a *Account) EnableJetStream(limits map[string]JetStreamAccountLimits) erro
 	}

 	s.mu.RLock()
+	if s.sys == nil {
+		s.mu.RUnlock()
+		return ErrServerNotRunning
+	}
 	sendq := s.sys.sendq
 	s.mu.RUnlock()

@@ -4120,7 +4120,9 @@ func (s *Server) jsConsumerListRequest(sub *subscription, c *client, _ *Account,
 	}

 	for _, o := range obs[offset:] {
-		resp.Consumers = append(resp.Consumers, o.info())
+		if cinfo := o.info(); cinfo != nil {
+			resp.Consumers = append(resp.Consumers, cinfo)
+		}
 		if len(resp.Consumers) >= JSApiListLimit {
 			break
 		}
@@ -2016,7 +2016,7 @@ func (js *jetStream) createRaftGroup(accName string, rg *raftGroup, storage Stor
 	if storage == FileStorage {
 		fs, err := newFileStoreWithCreated(
 			FileStoreConfig{StoreDir: storeDir, BlockSize: defaultMediumBlockSize, AsyncFlush: false, SyncInterval: 5 * time.Minute, srv: s},
-			StreamConfig{Name: rg.Name, Storage: FileStorage},
+			StreamConfig{Name: rg.Name, Storage: FileStorage, Metadata: labels},
 			time.Now().UTC(),
 			s.jsKeyGen(s.getOpts().JetStreamKey, rg.Name),
 			s.jsKeyGen(s.getOpts().JetStreamOldKey, rg.Name),
@@ -6472,7 +6472,7 @@ func (s *Server) jsClusteredStreamListRequest(acc *Account, ci *ClientInfo, filt
 	s.sys.replies[inbox] = func(sub *subscription, _ *client, _ *Account, subject, _ string, msg []byte) {
 		var si StreamInfo
 		if err := json.Unmarshal(msg, &si); err != nil {
-			s.Warnf("Error unmarshaling clustered stream info response:%v", err)
+			s.Warnf("Error unmarshalling clustered stream info response:%v", err)
 			return
 		}
 		select {
@@ -8121,7 +8121,7 @@ func (mset *stream) handleClusterSyncRequest(sub *subscription, c *client, _ *Ac
 func (js *jetStream) offlineClusterInfo(rg *raftGroup) *ClusterInfo {
 	s := js.srv

-	ci := &ClusterInfo{Name: s.ClusterName()}
+	ci := &ClusterInfo{Name: s.ClusterName(), RaftGroup: rg.Name}
 	for _, peer := range rg.Peers {
 		if sir, ok := s.nodeToInfo.Load(peer); ok && sir != nil {
 			si := sir.(nodeInfo)
@@ -8150,8 +8150,9 @@ func (js *jetStream) clusterInfo(rg *raftGroup) *ClusterInfo {

 	n := rg.node
 	ci := &ClusterInfo{
-		Name:   s.cachedClusterName(),
-		Leader: s.serverNameForNode(n.GroupLeader()),
+		Name:      s.cachedClusterName(),
+		Leader:    s.serverNameForNode(n.GroupLeader()),
+		RaftGroup: rg.Name,
 	}

 	now := time.Now()
@@ -134,6 +134,7 @@ type JSConsumerDeliveryTerminatedAdvisory struct {
 	ConsumerSeq uint64 `json:"consumer_seq"`
 	StreamSeq   uint64 `json:"stream_seq"`
 	Deliveries  uint64 `json:"deliveries"`
+	Reason      string `json:"reason,omitempty"`
 	Domain      string `json:"domain,omitempty"`
 }

@@ -427,9 +427,17 @@ func (ms *memStore) filteredStateLocked(sseq uint64, filter string, lastPerSubje
 		// We will adjust from the totals above by scanning what we need to exclude.
 		ss.First = first
 		var adjust uint64
+		var tss *SimpleState
+
 		for seq := ms.state.FirstSeq; seq < first; seq++ {
 			if sm, ok := ms.msgs[seq]; ok && !seen[sm.subj] && isMatch(sm.subj) {
-				adjust++
+				if lastPerSubject {
+					tss = ms.fss[sm.subj]
+				}
+				// If we are last per subject, make sure to only adjust if all messages are before our first.
+				if tss == nil || tss.Last < first {
+					adjust++
+				}
 				if seen != nil {
 					seen[sm.subj] = true
 				}
@@ -515,8 +523,7 @@ func (ms *memStore) SubjectsTotals(filterSubject string) map[string]uint64 {

 // NumPending will return the number of pending messages matching the filter subject starting at sequence.
 func (ms *memStore) NumPending(sseq uint64, filter string, lastPerSubject bool) (total, validThrough uint64) {
-	// This needs to be a write lock, as filteredStateLocked can
-	// mutate the per-subject state.
+	// This needs to be a write lock, as filteredStateLocked can mutate the per-subject state.
 	ms.mu.Lock()
 	defer ms.mu.Unlock()

@@ -629,7 +636,7 @@ func (ms *memStore) expireMsgs() {
 // Will return the number of purged messages.
 func (ms *memStore) PurgeEx(subject string, sequence, keep uint64) (purged uint64, err error) {
 	if subject == _EMPTY_ || subject == fwcs {
-		if keep == 0 && (sequence == 0 || sequence == 1) {
+		if keep == 0 && sequence == 0 {
 			return ms.Purge()
 		}
 		if sequence > 1 {
@@ -130,6 +130,7 @@ type ConnInfo struct {
 	TLSVersion     string         `json:"tls_version,omitempty"`
 	TLSCipher      string         `json:"tls_cipher_suite,omitempty"`
 	TLSPeerCerts   []*TLSPeerCert `json:"tls_peer_certs,omitempty"`
+	TLSFirst       bool           `json:"tls_first,omitempty"`
 	AuthorizedUser string         `json:"authorized_user,omitempty"`
 	Account        string         `json:"account,omitempty"`
 	Subs           []string       `json:"subscriptions_list,omitempty"`
@@ -568,6 +569,7 @@ func (ci *ConnInfo) fill(client *client, nc net.Conn, now time.Time, auth bool)
 			if auth && len(cs.PeerCertificates) > 0 {
 				ci.TLSPeerCerts = makePeerCerts(cs.PeerCertificates)
 			}
+			ci.TLSFirst = client.flags.isSet(didTLSFirst)
 		}
 	}

@@ -85,6 +85,7 @@ const (
 	mqttConnAckRCServerUnavailable           = byte(0x3)
 	mqttConnAckRCBadUserOrPassword           = byte(0x4)
 	mqttConnAckRCNotAuthorized               = byte(0x5)
+	mqttConnAckRCQoS2WillRejected            = byte(0x10)

 	// Maximum payload size of a control packet
 	mqttMaxPayloadSize = 0xFFFFFFF
@@ -345,6 +346,14 @@ type mqtt struct {
 	asm  *mqttAccountSessionManager // quick reference to account session manager, immutable after processConnect()
 	sess *mqttSession               // quick reference to session, immutable after processConnect()
 	cid  string                     // client ID
+
+	// rejectQoS2Pub tells the MQTT client to not accept QoS2 PUBLISH, instead
+	// error and terminate the connection.
+	rejectQoS2Pub bool
+
+	// downgradeQOS2Sub tells the MQTT client to downgrade QoS2 SUBSCRIBE
+	// requests to QoS1.
+	downgradeQoS2Sub bool
 }

 type mqttPending struct {
@@ -475,7 +484,11 @@ func (s *Server) createMQTTClient(conn net.Conn, ws *websocket) *client {
 	}
 	now := time.Now()

-	c := &client{srv: s, nc: conn, mpay: maxPay, msubs: maxSubs, start: now, last: now, mqtt: &mqtt{}, ws: ws}
+	mqtt := &mqtt{
+		rejectQoS2Pub:    opts.MQTT.rejectQoS2Pub,
+		downgradeQoS2Sub: opts.MQTT.downgradeQoS2Sub,
+	}
+	c := &client{srv: s, nc: conn, mpay: maxPay, msubs: maxSubs, start: now, last: now, mqtt: mqtt, ws: ws}
 	c.headers = true
 	c.mqtt.pp = &mqttPublish{}
 	// MQTT clients don't send NATS CONNECT protocols. So make it an "echo"
@@ -598,9 +611,9 @@ func validateMQTTOptions(o *Options) error {
 	if mo.Port == 0 {
 		return nil
 	}
-	// We have to force the server name to be explicitly set. There are conditions
-	// where we need a unique, repeatable name.
-	if o.ServerName == _EMPTY_ {
+	// We have to force the server name to be explicitly set and be unique when
+	// in cluster mode.
+	if o.ServerName == _EMPTY_ && (o.Cluster.Port != 0 || o.Gateway.Port != 0) {
 		return errMQTTServerNameMustBeSet
 	}
 	// If there is a NoAuthUser, we need to have Users defined and
@@ -2029,9 +2042,6 @@ func (as *mqttAccountSessionManager) sendJSAPIrequests(s *Server, c *client, acc
 // Add/Replace this message from the retained messages map.
 // If a message for this topic already existed, the existing record is updated
 // with the provided information.
-// This function will return the stream sequence of the record before its update,
-// or 0 if the record was added instead of updated.
-//
 // Lock not held on entry.
 func (as *mqttAccountSessionManager) handleRetainedMsg(key string, rm *mqttRetainedMsgRef) {
 	as.mu.Lock()
@@ -2040,7 +2050,7 @@ func (as *mqttAccountSessionManager) handleRetainedMsg(key string, rm *mqttRetai
 		as.retmsgs = make(map[string]*mqttRetainedMsgRef)
 		as.sl = NewSublistWithCache()
 	} else {
-		// Check if we already had one. If so, update the existing one.
+		// Check if we already had one retained message. If so, update the existing one.
 		if erm, exists := as.retmsgs[key]; exists {
 			// If the new sequence is below the floor or the existing one,
 			// then ignore the new one.
@@ -2057,6 +2067,7 @@ func (as *mqttAccountSessionManager) handleRetainedMsg(key string, rm *mqttRetai
 				erm.sub = &subscription{subject: []byte(key)}
 				as.sl.Insert(erm.sub)
 			}
+			return
 		}
 	}
 	rm.sub = &subscription{subject: []byte(key)}
@@ -2228,6 +2239,10 @@ func (as *mqttAccountSessionManager) processSubs(sess *mqttSession, c *client,
 		if f.qos > 2 {
 			f.qos = 2
 		}
+		if c.mqtt.downgradeQoS2Sub && f.qos == 2 {
+			c.Warnf("Downgrading subscription QoS2 to QoS1 for %q, as configured", f.filter)
+			f.qos = 1
+		}
 		subject := f.filter
 		sid := subject

@@ -2331,6 +2346,10 @@ func (as *mqttAccountSessionManager) serializeRetainedMsgsForSub(sess *mqttSessi
 		if qos > sub.mqtt.qos {
 			qos = sub.mqtt.qos
 		}
+		if c.mqtt.rejectQoS2Pub && qos == 2 {
+			c.Warnf("Rejecting retained message with QoS2 for subscription %q, as configured", sub.subject)
+			continue
+		}
 		if qos > 0 {
 			pi = sess.trackPublishRetained()

@@ -2891,10 +2910,11 @@ func (sess *mqttSession) trackAsPubRel(pi uint16, jsAckSubject string) {

 	sseq, _, _ := ackReplyInfo(jsAckSubject)

-	var sseqToPi map[uint64]uint16
 	if sess.cpending == nil {
 		sess.cpending = make(map[string]map[uint64]uint16)
-	} else if sseqToPi = sess.cpending[jsDur]; sseqToPi == nil {
+	}
+	sseqToPi := sess.cpending[jsDur]
+	if sseqToPi == nil {
 		sseqToPi = make(map[uint64]uint16)
 		sess.cpending[jsDur] = sseqToPi
 	}
@@ -3002,6 +3022,10 @@ func (c *client) mqttParseConnect(r *mqttReader, pl int, hasMappings bool) (byte
 		hasWill = true
 	}

+	if c.mqtt.rejectQoS2Pub && hasWill && wqos == 2 {
+		return mqttConnAckRCQoS2WillRejected, nil, fmt.Errorf("server does not accept QoS2 for Will messages")
+	}
+
 	// Spec [MQTT-3.1.2-19]
 	hasUser := cp.flags&mqttConnFlagUsernameFlag != 0
 	// Spec [MQTT-3.1.2-21]
@@ -3386,6 +3410,11 @@ func (c *client) mqttParsePub(r *mqttReader, pl int, pp *mqttPublish, hasMapping
 	if qos > 2 {
 		return fmt.Errorf("QoS=%v is invalid in MQTT", qos)
 	}
+
+	if c.mqtt.rejectQoS2Pub && qos == 2 {
+		return fmt.Errorf("QoS=2 is disabled for PUBLISH messages")
+	}
+
 	// Keep track of where we are when starting to read the variable header
 	start := r.pos

@@ -327,11 +327,23 @@ type Options struct {
 	TLSConfig             *tls.Config       `json:"-"`
 	TLSPinnedCerts        PinnedCertSet     `json:"-"`
 	TLSRateLimit          int64             `json:"-"`
-	AllowNonTLS           bool              `json:"-"`
-	WriteDeadline         time.Duration     `json:"-"`
-	MaxClosedClients      int               `json:"-"`
-	LameDuckDuration      time.Duration     `json:"-"`
-	LameDuckGracePeriod   time.Duration     `json:"-"`
+	// When set to true, the server will perform the TLS handshake before
+	// sending the INFO protocol. For clients that are not configured
+	// with a similar option, their connection will fail with some sort
+	// of timeout or EOF error since they are expecting to receive an
+	// INFO protocol first.
+	TLSHandshakeFirst bool `json:"-"`
+	// If TLSHandshakeFirst is true and this value is strictly positive,
+	// the server will wait for that amount of time for the TLS handshake
+	// to start before falling back to previous behavior of sending the
+	// INFO protocol first. It allows for a mix of newer clients that can
+	// require a TLS handshake first, and older clients that can't.
+	TLSHandshakeFirstFallback time.Duration `json:"-"`
+	AllowNonTLS               bool          `json:"-"`
+	WriteDeadline             time.Duration `json:"-"`
+	MaxClosedClients          int           `json:"-"`
+	LameDuckDuration          time.Duration `json:"-"`
+	LameDuckGracePeriod       time.Duration `json:"-"`

 	// MaxTracedMsgLen is the maximum printable length for traced messages.
 	MaxTracedMsgLen int `json:"-"`
@@ -550,6 +562,14 @@ type MQTTOpts struct {

 	// Snapshot of configured TLS options.
 	tlsConfigOpts *TLSConfigOpts
+
+	// rejectQoS2Pub tells the MQTT client to not accept QoS2 PUBLISH, instead
+	// error and terminate the connection.
+	rejectQoS2Pub bool
+
+	// downgradeQOS2Sub tells the MQTT client to downgrade QoS2 SUBSCRIBE
+	// requests to QoS1.
+	downgradeQoS2Sub bool
 }

 type netResolver interface {
@@ -638,7 +658,8 @@ type TLSConfigOpts struct {
 	Insecure          bool
 	Map               bool
 	TLSCheckKnownURLs bool
-	HandshakeFirst    bool // Indicate that the TLS handshake should occur first, before sending the INFO protocol
+	HandshakeFirst    bool          // Indicate that the TLS handshake should occur first, before sending the INFO protocol.
+	FallbackDelay     time.Duration // Where supported, indicates how long to wait for the handshake before falling back to sending the INFO protocol first.
 	Timeout           float64
 	RateLimit         int64
 	Ciphers           []uint16
@@ -1072,6 +1093,8 @@ func (o *Options) processConfigFileLine(k string, v interface{}, errors *[]error
 		o.TLSMap = tc.Map
 		o.TLSPinnedCerts = tc.PinnedCerts
 		o.TLSRateLimit = tc.RateLimit
+		o.TLSHandshakeFirst = tc.HandshakeFirst
+		o.TLSHandshakeFirstFallback = tc.FallbackDelay

 		// Need to keep track of path of the original TLS config
 		// and certs path for OCSP Stapling monitoring.
@@ -4312,7 +4335,30 @@ func parseTLS(v interface{}, isClientCtx bool) (t *TLSConfigOpts, retErr error)
 			}
 			tc.CertMatch = certMatch
 		case "handshake_first", "first", "immediate":
-			tc.HandshakeFirst = mv.(bool)
+			switch mv := mv.(type) {
+			case bool:
+				tc.HandshakeFirst = mv
+			case string:
+				switch strings.ToLower(mv) {
+				case "true", "on":
+					tc.HandshakeFirst = true
+				case "false", "off":
+					tc.HandshakeFirst = false
+				case "auto", "auto_fallback":
+					tc.HandshakeFirst = true
+					tc.FallbackDelay = DEFAULT_TLS_HANDSHAKE_FIRST_FALLBACK_DELAY
+				default:
+					// Check to see if this is a duration.
+					if dur, err := time.ParseDuration(mv); err == nil {
+						tc.HandshakeFirst = true
+						tc.FallbackDelay = dur
+						break
+					}
+					return nil, &configErr{tk, fmt.Sprintf("field %q's value %q is invalid", mk, mv)}
+				}
+			default:
+				return nil, &configErr{tk, fmt.Sprintf("field %q should be a boolean or a string, got %T", mk, mv)}
+			}
 		case "ocsp_peer":
 			switch vv := mv.(type) {
 			case bool:
@@ -4593,6 +4639,11 @@ func parseMQTT(v interface{}, o *Options, errors *[]error, warnings *[]error) er
 		case "consumer_inactive_threshold", "consumer_auto_cleanup":
 			o.MQTT.ConsumerInactiveThreshold = parseDuration("consumer_inactive_threshold", tk, mv, errors, warnings)

+		case "reject_qos2_publish":
+			o.MQTT.rejectQoS2Pub = mv.(bool)
+		case "downgrade_qos2_subscribe":
+			o.MQTT.downgradeQoS2Sub = mv.(bool)
+
 		default:
 			if !tk.IsUsedVariable() {
 				err := &unknownConfigFieldErr{
@@ -133,8 +133,7 @@ type raft struct {
 	wtype    StorageType
 	track    bool
 	werr     error
-	state    RaftState
-	isLeader atomic.Bool
+	state    atomic.Int32 // RaftState
 	hh       hash.Hash64
 	snapfile string
 	csz      int
@@ -382,7 +381,6 @@ func (s *Server) startRaftNode(accName string, cfg *RaftConfig, labels pprofLabe
 		wal:      cfg.Log,
 		wtype:    cfg.Log.Type(),
 		track:    cfg.Track,
-		state:    Follower,
 		csz:      ps.clusterSize,
 		qn:       ps.clusterSize/2 + 1,
 		hash:     hash,
@@ -653,12 +651,11 @@ func (s *Server) transferRaftLeaders() bool {
 // Propose will propose a new entry to the group.
 // This should only be called on the leader.
 func (n *raft) Propose(data []byte) error {
-	n.RLock()
-	if state := n.state; state != Leader {
-		n.RUnlock()
+	if state := n.State(); state != Leader {
 		n.debug("Proposal ignored, not leader (state: %v)", state)
 		return errNotLeader
 	}
+	n.RLock()
 	// Error if we had a previous write error.
 	if werr := n.werr; werr != nil {
 		n.RUnlock()
@@ -674,12 +671,11 @@ func (n *raft) Propose(data []byte) error {
 // ProposeDirect will propose entries directly.
 // This should only be called on the leader.
 func (n *raft) ProposeDirect(entries []*Entry) error {
-	n.RLock()
-	if n.state != Leader {
-		n.RUnlock()
-		n.debug("Direct proposal ignored, not leader (state: %v)", n.state)
+	if state := n.State(); state != Leader {
+		n.debug("Direct proposal ignored, not leader (state: %v)", state)
 		return errNotLeader
 	}
+	n.RLock()
 	// Error if we had a previous write error.
 	if werr := n.werr; werr != nil {
 		n.RUnlock()
@@ -706,11 +702,10 @@ func (n *raft) ForwardProposal(entry []byte) error {

 // ProposeAddPeer is called to add a peer to the group.
 func (n *raft) ProposeAddPeer(peer string) error {
-	n.RLock()
-	if n.state != Leader {
-		n.RUnlock()
+	if n.State() != Leader {
 		return errNotLeader
 	}
+	n.RLock()
 	// Error if we had a previous write error.
 	if werr := n.werr; werr != nil {
 		n.RUnlock()
@@ -742,7 +737,7 @@ func (n *raft) doRemovePeerAsLeader(peer string) {
 func (n *raft) ProposeRemovePeer(peer string) error {
 	n.RLock()
 	prop, subj := n.prop, n.rpsubj
-	isLeader := n.state == Leader
+	isLeader := n.State() == Leader
 	werr := n.werr
 	n.RUnlock()

@@ -794,11 +789,10 @@ func (n *raft) AdjustBootClusterSize(csz int) error {
 // AdjustClusterSize will change the cluster set size.
 // Must be the leader.
 func (n *raft) AdjustClusterSize(csz int) error {
-	n.Lock()
-	if n.state != Leader {
-		n.Unlock()
+	if n.State() != Leader {
 		return errNotLeader
 	}
+	n.Lock()
 	// Same floor as bootstrap.
 	if csz < 2 {
 		csz = 2
@@ -816,14 +810,15 @@ func (n *raft) AdjustClusterSize(csz int) error {
 // PauseApply will allow us to pause processing of append entries onto our
 // external apply chan.
 func (n *raft) PauseApply() error {
+	if n.State() == Leader {
+		return errAlreadyLeader
+	}
+
 	n.Lock()
 	defer n.Unlock()

-	if n.state == Leader {
-		return errAlreadyLeader
-	}
 	// If we are currently a candidate make sure we step down.
-	if n.state == Candidate {
+	if n.State() == Candidate {
 		n.stepdown.push(noLeader)
 	}

@@ -944,12 +939,12 @@ func (n *raft) SendSnapshot(data []byte) error {
 // all of the log entries up to and including index. This should not be called with
 // entries that have been applied to the FSM but have not been applied to the raft state.
 func (n *raft) InstallSnapshot(data []byte) error {
-	n.Lock()
-	if n.state == Closed {
-		n.Unlock()
+	if n.State() == Closed {
 		return errNodeClosed
 	}

+	n.Lock()
+
 	if werr := n.werr; werr != nil {
 		n.Unlock()
 		return werr
@@ -1164,7 +1159,7 @@ func (n *raft) Leader() bool {
 	if n == nil {
 		return false
 	}
-	return n.isLeader.Load()
+	return n.State() == Leader
 }

 func (n *raft) isCatchingUp() bool {
@@ -1178,7 +1173,7 @@ func (n *raft) isCatchingUp() bool {
 // Lock should be held.
 func (n *raft) isCurrent(includeForwardProgress bool) bool {
 	// Check if we are closed.
-	if n.state == Closed {
+	if n.State() == Closed {
 		n.debug("Not current, node is closed")
 		return false
 	}
@@ -1190,7 +1185,7 @@ func (n *raft) isCurrent(includeForwardProgress bool) bool {
 	}

 	// Make sure we are the leader or we know we have heard from the leader recently.
-	if n.state == Leader {
+	if n.State() == Leader {
 		return true
 	}

@@ -1300,7 +1295,7 @@ func (n *raft) StepDown(preferred ...string) error {
 		return errTooManyPrefs
 	}

-	if n.state != Leader {
+	if n.State() != Leader {
 		n.Unlock()
 		return errNotLeader
 	}
@@ -1394,7 +1389,7 @@ func randCampaignTimeout() time.Duration {
 // Lock should be held.
 func (n *raft) campaign() error {
 	n.debug("Starting campaign")
-	if n.state == Leader {
+	if n.State() == Leader {
 		return errAlreadyLeader
 	}
 	n.resetElect(randCampaignTimeout())
@@ -1405,7 +1400,7 @@ func (n *raft) campaign() error {
 // Lock should be held.
 func (n *raft) xferCampaign() error {
 	n.debug("Starting transfer campaign")
-	if n.state == Leader {
+	if n.State() == Leader {
 		n.lxfer = false
 		return errAlreadyLeader
 	}
@@ -1415,9 +1410,7 @@ func (n *raft) xferCampaign() error {

 // State returns the current state for this node.
 func (n *raft) State() RaftState {
-	n.RLock()
-	defer n.RUnlock()
-	return n.state
+	return RaftState(n.state.Load())
 }

 // Progress returns the current index, commit and applied values.
@@ -1478,7 +1471,7 @@ func (n *raft) UpdateKnownPeers(knownPeers []string) {
 	// Process like peer state update.
 	ps := &peerState{knownPeers, len(knownPeers), n.extSt}
 	n.processPeerState(ps)
-	isLeader := n.state == Leader
+	isLeader := n.State() == Leader
 	n.Unlock()

 	// If we are the leader send this update out as well.
@@ -1507,7 +1500,14 @@ func (n *raft) Delete() {

 func (n *raft) shutdown(shouldDelete bool) {
 	n.Lock()
-	if n.state == Closed {
+
+	// Returned swap value is the previous state. It looks counter-intuitive
+	// to do this atomic operation with the lock held, but we have to do so in
+	// order to make sure that switchState() is not already running. If it is
+	// then it can potentially update the n.state back to a non-closed state,
+	// allowing shutdown() to be called again. If that happens then the below
+	// close(n.quit) will panic from trying to close an already-closed channel.
+	if n.state.Swap(int32(Closed)) == int32(Closed) {
 		n.Unlock()
 		return
 	}
@@ -1525,7 +1525,6 @@ func (n *raft) shutdown(shouldDelete bool) {
 		}
 		c.closeConnection(InternalClient)
 	}
-	n.state = Closed
 	s, g, wal := n.s, n.group, n.wal

 	// Delete our peer state and vote state and any snapshots.
@@ -1788,7 +1787,7 @@ func (n *raft) processAppendEntries() {
 }

 func (n *raft) runAsFollower() {
-	for {
+	for n.State() == Follower {
 		elect := n.electTimer()

 		select {
@@ -2156,11 +2155,11 @@ func (n *raft) handleForwardedProposal(sub *subscription, c *client, _ *Account,
 }

 func (n *raft) runAsLeader() {
-	n.RLock()
-	if n.state == Closed {
-		n.RUnlock()
+	if n.State() == Closed {
 		return
 	}
+
+	n.RLock()
 	psubj, rpsubj := n.psubj, n.rpsubj
 	n.RUnlock()

@@ -2193,7 +2192,7 @@ func (n *raft) runAsLeader() {
 	lq := time.NewTicker(lostQuorumCheck)
 	defer lq.Stop()

-	for {
+	for n.State() == Leader {
 		select {
 		case <-n.s.quitCh:
 			n.shutdown(false)
@@ -2541,7 +2540,7 @@ func (n *raft) loadEntry(index uint64) (*appendEntry, error) {
 // applyCommit will update our commit index and apply the entry to the apply chan.
 // lock should be held.
 func (n *raft) applyCommit(index uint64) error {
-	if n.state == Closed {
+	if n.State() == Closed {
 		return errNodeClosed
 	}
 	if index <= n.commit {
@@ -2551,7 +2550,7 @@ func (n *raft) applyCommit(index uint64) error {
 	original := n.commit
 	n.commit = index

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

@@ -2568,7 +2567,7 @@ func (n *raft) applyCommit(index uint64) error {
 		if ae, err = n.loadEntry(index); err != nil {
 			if err != ErrStoreClosed && err != ErrStoreEOF {
 				n.warn("Got an error loading %d index: %v - will reset", index, err)
-				if n.state == Leader {
+				if n.State() == Leader {
 					n.stepdown.push(n.selectNextLeader())
 				}
 				// Reset and cancel any catchup.
@@ -2596,7 +2595,7 @@ func (n *raft) applyCommit(index uint64) error {
 		case EntrySnapshot:
 			committed = append(committed, e)
 		case EntryPeerState:
-			if n.state != Leader {
+			if n.State() != Leader {
 				if ps, err := decodePeerState(e.Data); err == nil {
 					n.processPeerState(ps)
 				}
@@ -2646,7 +2645,7 @@ func (n *raft) applyCommit(index uint64) error {
 			}

 			// If this is us and we are the leader we should attempt to stepdown.
-			if peer == n.id && n.state == Leader {
+			if peer == n.id && n.State() == Leader {
 				n.stepdown.push(n.selectNextLeader())
 			}

@@ -2674,12 +2673,12 @@ func (n *raft) applyCommit(index uint64) error {

 // Used to track a success response and apply entries.
 func (n *raft) trackResponse(ar *appendEntryResponse) {
-	n.Lock()
-	if n.state == Closed {
-		n.Unlock()
+	if n.State() == Closed {
 		return
 	}

+	n.Lock()
+
 	// Update peer's last index.
 	if ps := n.peers[ar.peer]; ps != nil && ar.index > ps.li {
 		ps.li = ar.index
@@ -2736,7 +2735,7 @@ func (n *raft) adjustClusterSizeAndQuorum() {
 		n.lsut = time.Now()
 	} else if ncsz < pcsz {
 		n.debug("Decreasing our clustersize: %d -> %d", pcsz, ncsz)
-		if n.state == Leader {
+		if n.State() == Leader {
 			go n.sendHeartbeat()
 		}
 	}
@@ -2749,7 +2748,7 @@ func (n *raft) trackPeer(peer string) error {
 	if n.removed != nil {
 		_, isRemoved = n.removed[peer]
 	}
-	if n.state == Leader {
+	if n.State() == Leader {
 		if lp, ok := n.peers[peer]; !ok || !lp.kp {
 			// Check if this peer had been removed previously.
 			needPeerAdd = !isRemoved
@@ -2780,7 +2779,7 @@ func (n *raft) runAsCandidate() {
 	// We vote for ourselves.
 	votes := 1

-	for {
+	for n.State() == Candidate {
 		elect := n.electTimer()
 		select {
 		case <-n.entry.ch:
@@ -2962,7 +2961,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	}

 	// Just return if closed or we had previous write error.
-	if n.state == Closed || n.werr != nil {
+	if n.State() == Closed || n.werr != nil {
 		n.Unlock()
 		return
 	}
@@ -2972,7 +2971,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	arbuf := scratch[:]

 	// Are we receiving from another leader.
-	if n.state == Leader {
+	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
@@ -2993,7 +2992,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	}

 	// If we received an append entry as a candidate we should convert to a follower.
-	if n.state == Candidate {
+	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
@@ -3059,13 +3058,13 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 		if isNew {
 			n.writeTermVote()
 		}
-		if n.state != Follower {
-			n.debug("Term higher than ours and we are not a follower: %v, stepping down to %q", n.state, ae.leader)
+		if n.State() != Follower {
+			n.debug("Term higher than ours and we are not a follower: %v, stepping down to %q", n.State(), ae.leader)
 			n.stepdown.push(ae.leader)
 		}
 	}

-	if isNew && n.leader != ae.leader && n.state == Follower {
+	if isNew && n.leader != ae.leader && n.State() == Follower {
 		n.debug("AppendEntry updating leader to %q", ae.leader)
 		n.updateLeader(ae.leader)
 		n.writeTermVote()
@@ -3333,7 +3332,7 @@ func (n *raft) storeToWAL(ae *appendEntry) error {
 	// Sanity checking for now.
 	if index := ae.pindex + 1; index != seq {
 		n.warn("Wrong index, ae is %+v, index stored was %d, n.pindex is %d, will reset", ae, seq, n.pindex)
-		if n.state == Leader {
+		if n.State() == Leader {
 			n.stepdown.push(n.selectNextLeader())
 		}
 		// Reset and cancel any catchup.
@@ -3564,7 +3563,7 @@ func (n *raft) readTermVote() (term uint64, voted string, err error) {
 // Lock should be held.
 func (n *raft) setWriteErrLocked(err error) {
 	// Check if we are closed already.
-	if n.state == Closed {
+	if n.State() == Closed {
 		return
 	}
 	// Ignore if already set.
@@ -3598,9 +3597,7 @@ func (n *raft) setWriteErrLocked(err error) {

 // Helper to check if we are closed when we do not hold a lock already.
 func (n *raft) isClosed() bool {
-	n.RLock()
-	defer n.RUnlock()
-	return n.state == Closed
+	return n.State() == Closed
 }

 // Capture our write error if any and hold.
@@ -3746,9 +3743,9 @@ func (n *raft) processVoteRequest(vr *voteRequest) error {

 	// If this is a higher term go ahead and stepdown.
 	if vr.term > n.term {
-		if n.state != Follower {
+		if n.State() != Follower {
 			n.debug("Stepping down from %s, detected higher term: %d vs %d",
-				strings.ToLower(n.state.String()), vr.term, n.term)
+				strings.ToLower(n.State().String()), vr.term, n.term)
 			n.stepdown.push(noLeader)
 			n.term = vr.term
 		}
@@ -3787,7 +3784,7 @@ func (n *raft) handleVoteRequest(sub *subscription, c *client, _ *Account, subje

 func (n *raft) requestVote() {
 	n.Lock()
-	if n.state != Candidate {
+	if n.State() != Candidate {
 		n.Unlock()
 		return
 	}
@@ -3829,9 +3826,6 @@ func (n *raft) quorumNeeded() int {

 // Lock should be held.
 func (n *raft) updateLeadChange(isLeader bool) {
-	// Update our atomic about being the leader.
-	n.isLeader.Store(isLeader)
-
 	// We don't care about values that have not been consumed (transitory states),
 	// so we dequeue any state that is pending and push the new one.
 	for {
@@ -3851,25 +3845,25 @@ func (n *raft) updateLeadChange(isLeader bool) {

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

 	// Reset the election timer.
 	n.resetElectionTimeout()

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

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

@@ -3879,11 +3873,13 @@ const (
 )

 func (n *raft) switchToFollower(leader string) {
-	n.Lock()
-	defer n.Unlock()
-	if n.state == Closed {
+	if n.State() == Closed {
 		return
 	}
+
+	n.Lock()
+	defer n.Unlock()
+
 	n.debug("Switching to follower")

 	n.lxfer = false
@@ -3892,17 +3888,19 @@ func (n *raft) switchToFollower(leader string) {
 }

 func (n *raft) switchToCandidate() {
-	n.Lock()
-	defer n.Unlock()
-	if n.state == Closed {
+	if n.State() == Closed {
 		return
 	}
+
+	n.Lock()
+	defer n.Unlock()
+
 	// If we are catching up or are in observer mode we can not switch.
 	if n.observer || n.paused {
 		return
 	}

-	if n.state != Candidate {
+	if n.State() != Candidate {
 		n.debug("Switching to candidate")
 	} else {
 		if n.lostQuorumLocked() && time.Since(n.llqrt) > 20*time.Second {
@@ -3919,11 +3917,12 @@ func (n *raft) switchToCandidate() {
 }

 func (n *raft) switchToLeader() {
-	n.Lock()
-	if n.state == Closed {
-		n.Unlock()
+	if n.State() == Closed {
 		return
 	}
+
+	n.Lock()
+
 	n.debug("Switching to leader")

 	var state StreamState
@@ -266,6 +266,28 @@ func (t *tlsPinnedCertOption) Apply(server *Server) {
 	server.Noticef("Reloaded: %d pinned_certs", len(t.newValue))
 }

+// tlsHandshakeFirst implements the option interface for the tls `handshake first` setting.
+type tlsHandshakeFirst struct {
+	noopOption
+	newValue bool
+}
+
+// Apply is a no-op because the timeout will be reloaded after options are applied.
+func (t *tlsHandshakeFirst) Apply(server *Server) {
+	server.Noticef("Reloaded: Client TLS handshake first: %v", t.newValue)
+}
+
+// tlsHandshakeFirstFallback implements the option interface for the tls `handshake first fallback delay` setting.
+type tlsHandshakeFirstFallback struct {
+	noopOption
+	newValue time.Duration
+}
+
+// Apply is a no-op because the timeout will be reloaded after options are applied.
+func (t *tlsHandshakeFirstFallback) Apply(server *Server) {
+	server.Noticef("Reloaded: Client TLS handshake first fallback delay: %v", t.newValue)
+}
+
 // authOption is a base struct that provides default option behaviors.
 type authOption struct {
 	noopOption
@@ -1222,6 +1244,10 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 			diffOpts = append(diffOpts, &tlsTimeoutOption{newValue: newValue.(float64)})
 		case "tlspinnedcerts":
 			diffOpts = append(diffOpts, &tlsPinnedCertOption{newValue: newValue.(PinnedCertSet)})
+		case "tlshandshakefirst":
+			diffOpts = append(diffOpts, &tlsHandshakeFirst{newValue: newValue.(bool)})
+		case "tlshandshakefirstfallback":
+			diffOpts = append(diffOpts, &tlsHandshakeFirstFallback{newValue: newValue.(time.Duration)})
 		case "username":
 			diffOpts = append(diffOpts, &usernameOption{})
 		case "password":
@@ -1071,9 +1071,30 @@ func (s *Server) processImplicitRoute(info *Info, routeNoPool bool) {
 // in the server's opts.Routes, false otherwise.
 // Server lock is assumed to be held by caller.
 func (s *Server) hasThisRouteConfigured(info *Info) bool {
-	urlToCheckExplicit := strings.ToLower(net.JoinHostPort(info.Host, strconv.Itoa(info.Port)))
-	for _, ri := range s.getOpts().Routes {
-		if strings.ToLower(ri.Host) == urlToCheckExplicit {
+	routes := s.getOpts().Routes
+	if len(routes) == 0 {
+		return false
+	}
+	// This could possibly be a 0.0.0.0 host so we will also construct a second
+	// url with the host section of the `info.IP` (if present).
+	sPort := strconv.Itoa(info.Port)
+	urlOne := strings.ToLower(net.JoinHostPort(info.Host, sPort))
+	var urlTwo string
+	if info.IP != _EMPTY_ {
+		if u, _ := url.Parse(info.IP); u != nil {
+			urlTwo = strings.ToLower(net.JoinHostPort(u.Hostname(), sPort))
+			// Ignore if same than the first
+			if urlTwo == urlOne {
+				urlTwo = _EMPTY_
+			}
+		}
+	}
+	for _, ri := range routes {
+		rHost := strings.ToLower(ri.Host)
+		if rHost == urlOne {
+			return true
+		}
+		if urlTwo != _EMPTY_ && rHost == urlTwo {
 			return true
 		}
 	}
@@ -1860,6 +1881,17 @@ const (
 func (s *Server) addRoute(c *client, didSolicit bool, info *Info, accName string) bool {
 	id := info.ID

+	var acc *Account
+	if accName != _EMPTY_ {
+		var err error
+		acc, err = s.LookupAccount(accName)
+		if err != nil {
+			c.sendErrAndErr(fmt.Sprintf("Unable to lookup account %q: %v", accName, err))
+			c.closeConnection(MissingAccount)
+			return false
+		}
+	}
+
 	s.mu.Lock()
 	if !s.isRunning() || s.routesReject {
 		s.mu.Unlock()
@@ -1935,6 +1967,9 @@ func (s *Server) addRoute(c *client, didSolicit bool, info *Info, accName string
 			if c.last.IsZero() {
 				c.last = time.Now()
 			}
+			if acc != nil {
+				c.acc = acc
+			}
 			c.mu.Unlock()

 			// Store this route with key being the route id hash + account name
@@ -1984,6 +2019,21 @@ func (s *Server) addRoute(c *client, didSolicit bool, info *Info, accName string
 				s.mu.Unlock()
 				return false
 			}
+			// Look if there is a solicited route in the pool. If there is one,
+			// they should all be, so stop at the first.
+			if url, rtype, hasSolicited := hasSolicitedRoute(conns); hasSolicited {
+				upgradeRouteToSolicited(c, url, rtype)
+			}
+		} else {
+			// If we solicit, upgrade to solicited all non-solicited routes that
+			// we may have registered.
+			c.mu.Lock()
+			url := c.route.url
+			rtype := c.route.routeType
+			c.mu.Unlock()
+			for _, r := range conns {
+				upgradeRouteToSolicited(r, url, rtype)
+			}
 		}
 		// For all cases (solicited and not) we need to count how many connections
 		// we already have, and for solicited route, we will find a free spot in
@@ -2109,6 +2159,41 @@ func (s *Server) addRoute(c *client, didSolicit bool, info *Info, accName string
 	return !exists
 }

+func hasSolicitedRoute(conns []*client) (*url.URL, RouteType, bool) {
+	var url *url.URL
+	var rtype RouteType
+	for _, r := range conns {
+		if r == nil {
+			continue
+		}
+		r.mu.Lock()
+		if r.route.didSolicit {
+			url = r.route.url
+			rtype = r.route.routeType
+		}
+		r.mu.Unlock()
+		if url != nil {
+			return url, rtype, true
+		}
+	}
+	return nil, 0, false
+}
+
+func upgradeRouteToSolicited(r *client, url *url.URL, rtype RouteType) {
+	if r == nil {
+		return
+	}
+	r.mu.Lock()
+	if !r.route.didSolicit {
+		r.route.didSolicit = true
+		r.route.url = url
+	}
+	if rtype == Explicit {
+		r.route.routeType = Explicit
+	}
+	r.mu.Unlock()
+}
+
 func handleDuplicateRoute(remote, c *client, setNoReconnect bool) {
 	// We used to clear some fields when closing a duplicate connection
 	// to prevent sending INFO protocols for the remotes to update
@@ -2573,6 +2573,9 @@ func (s *Server) AcceptLoop(clr chan struct{}) {
 	// Alert of TLS enabled.
 	if opts.TLSConfig != nil {
 		s.Noticef("TLS required for client connections")
+		if opts.TLSHandshakeFirst && opts.TLSHandshakeFirstFallback == 0 {
+			s.Warnf("Clients that are not using \"TLS Handshake First\" option will fail to connect")
+		}
 	}

 	// If server was started with RANDOM_PORT (-1), opts.Port would be equal
@@ -3041,10 +3044,37 @@ func (s *Server) createClientEx(conn net.Conn, inProcess bool) *client {

 	c.Debugf("Client connection created")

-	// Send our information.
-	// Need to be sent in place since writeLoop cannot be started until
-	// TLS handshake is done (if applicable).
-	c.sendProtoNow(c.generateClientInfoJSON(info))
+	// Save info.TLSRequired value since we may neeed to change it back and forth.
+	orgInfoTLSReq := info.TLSRequired
+
+	var tlsFirstFallback time.Duration
+	// Check if we should do TLS first.
+	tlsFirst := opts.TLSConfig != nil && opts.TLSHandshakeFirst
+	if tlsFirst {
+		// Make sure info.TLSRequired is set to true (it could be false
+		// if AllowNonTLS is enabled).
+		info.TLSRequired = true
+		// Get the fallback delay value if applicable.
+		if f := opts.TLSHandshakeFirstFallback; f > 0 {
+			tlsFirstFallback = f
+		} else if inProcess {
+			// For in-process connection, we will always have a fallback
+			// delay. It allows support for non-TLS, TLS and "TLS First"
+			// in-process clients to successfully connect.
+			tlsFirstFallback = DEFAULT_TLS_HANDSHAKE_FIRST_FALLBACK_DELAY
+		}
+	}
+
+	// Decide if we are going to require TLS or not and generate INFO json.
+	tlsRequired := info.TLSRequired
+	infoBytes := c.generateClientInfoJSON(info)
+
+	// Send our information, except if TLS and TLSHandshakeFirst is requested.
+	if !tlsFirst {
+		// Need to be sent in place since writeLoop cannot be started until
+		// TLS handshake is done (if applicable).
+		c.sendProtoNow(infoBytes)
+	}

 	// Unlock to register
 	c.mu.Unlock()
@@ -3077,20 +3107,50 @@ func (s *Server) createClientEx(conn net.Conn, inProcess bool) *client {
 	}
 	s.clients[c.cid] = c

-	tlsRequired := info.TLSRequired
 	s.mu.Unlock()

 	// Re-Grab lock
 	c.mu.Lock()

-	// Connection could have been closed while sending the INFO proto.
 	isClosed := c.isClosed()
-
 	var pre []byte
+	// We need first to check for "TLS First" fallback delay.
+	if !isClosed && tlsFirstFallback > 0 {
+		// We wait and see if we are getting any data. Since we did not send
+		// the INFO protocol yet, only clients that use TLS first should be
+		// sending data (the TLS handshake). We don't really check the content:
+		// if it is a rogue agent and not an actual client performing the
+		// TLS handshake, the error will be detected when performing the
+		// handshake on our side.
+		pre = make([]byte, 4)
+		c.nc.SetReadDeadline(time.Now().Add(tlsFirstFallback))
+		n, _ := io.ReadFull(c.nc, pre[:])
+		c.nc.SetReadDeadline(time.Time{})
+		// If we get any data (regardless of possible timeout), we will proceed
+		// with the TLS handshake.
+		if n > 0 {
+			pre = pre[:n]
+		} else {
+			// We did not get anything so we will send the INFO protocol.
+			pre = nil
+
+			// Restore the original info.TLSRequired value if it is
+			// different that the current value and regenerate infoBytes.
+			if orgInfoTLSReq != info.TLSRequired {
+				info.TLSRequired = orgInfoTLSReq
+				infoBytes = c.generateClientInfoJSON(info)
+			}
+			c.sendProtoNow(infoBytes)
+			// Set the boolean to false for the rest of the function.
+			tlsFirst = false
+			// Check closed status again
+			isClosed = c.isClosed()
+		}
+	}
 	// If we have both TLS and non-TLS allowed we need to see which
 	// one the client wants. We'll always allow this for in-process
 	// connections.
-	if !isClosed && opts.TLSConfig != nil && (inProcess || opts.AllowNonTLS) {
+	if !isClosed && !tlsFirst && opts.TLSConfig != nil && (inProcess || opts.AllowNonTLS) {
 		pre = make([]byte, 4)
 		c.nc.SetReadDeadline(time.Now().Add(secondsToDuration(opts.TLSTimeout)))
 		n, _ := io.ReadFull(c.nc, pre[:])
@@ -3125,12 +3185,18 @@ func (s *Server) createClientEx(conn net.Conn, inProcess bool) *client {
 		}
 	}

-	// If connection is marked as closed, bail out.
+	// Now, send the INFO if it was delayed
+	if !isClosed && tlsFirst {
+		c.flags.set(didTLSFirst)
+		c.sendProtoNow(infoBytes)
+		// Check closed status
+		isClosed = c.isClosed()
+	}
+
+	// Connection could have been closed while sending the INFO proto.
 	if isClosed {
 		c.mu.Unlock()
-		// Connection could have been closed due to TLS timeout or while trying
-		// to send the INFO protocol. We need to call closeConnection() to make
-		// sure that proper cleanup is done.
+		// We need to call closeConnection() to make sure that proper cleanup is done.
 		c.closeConnection(WriteError)
 		return nil
 	}
@@ -34,10 +34,15 @@ func (s *Server) handleSignals() {
 	signal.Notify(c, os.Interrupt, syscall.SIGTERM)

 	go func() {
-		for sig := range c {
-			s.Debugf("Trapped %q signal", sig)
-			s.Shutdown()
-			os.Exit(0)
+		for {
+			select {
+			case sig := <-c:
+				s.Debugf("Trapped %q signal", sig)
+				s.Shutdown()
+				os.Exit(0)
+			case <-s.quitCh:
+				return
+			}
 		}
 	}()
 }
@@ -160,9 +160,10 @@ type StreamAlternate struct {
 // ClusterInfo shows information about the underlying set of servers
 // that make up the stream or consumer.
 type ClusterInfo struct {
-	Name     string      `json:"name,omitempty"`
-	Leader   string      `json:"leader,omitempty"`
-	Replicas []*PeerInfo `json:"replicas,omitempty"`
+	Name      string      `json:"name,omitempty"`
+	RaftGroup string      `json:"raft_group,omitempty"`
+	Leader    string      `json:"leader,omitempty"`
+	Replicas  []*PeerInfo `json:"replicas,omitempty"`
 }

 // PeerInfo shows information about all the peers in the cluster that
@@ -2126,6 +2127,14 @@ func (mset *stream) processMirrorMsgs(mirror *sourceInfo, ready *sync.WaitGroup)
 	// Signal the caller that we have captured the above fields.
 	ready.Done()

+	// Make sure we have valid ipq for msgs.
+	if msgs == nil {
+		mset.mu.Lock()
+		mset.cancelMirrorConsumer()
+		mset.mu.Unlock()
+		return
+	}
+
 	t := time.NewTicker(sourceHealthCheckInterval)
 	defer t.Stop()

@@ -5266,9 +5275,13 @@ func (mset *stream) Store() StreamStore {

 // Determines if the new proposed partition is unique amongst all consumers.
 // Lock should be held.
-func (mset *stream) partitionUnique(partitions []string) bool {
+func (mset *stream) partitionUnique(name string, partitions []string) bool {
 	for _, partition := range partitions {
-		for _, o := range mset.consumers {
+		for n, o := range mset.consumers {
+			// Skip the consumer being checked.
+			if n == name {
+				continue
+			}
 			if o.subjf == nil {
 				return false
 			}
@@ -22,7 +22,7 @@ before_script:
 - golangci-lint run ./jetstream/...
 script:
 - go test -modfile=go_test.mod -v -run=TestNoRace -p=1 ./... --failfast -vet=off
- if [[ "$TRAVIS_GO_VERSION" =~ 1.21 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off; fi
+- if [[ "$TRAVIS_GO_VERSION" =~ 1.21 ]]; then ./scripts/cov.sh TRAVIS; else go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off -tags=internal_testing; fi
 after_success:
 - if [[ "$TRAVIS_GO_VERSION" =~ 1.21 ]]; then $HOME/gopath/bin/goveralls -coverprofile=acc.out -service travis-ci; fi

@@ -29,7 +29,7 @@ When using or transitioning to Go modules support:
 ```bash
 # Go client latest or explicit version
 go get github.com/nats-io/nats.go/@latest
-go get github.com/nats-io/nats.go/@v1.30.2
+go get github.com/nats-io/nats.go/@v1.31.0

 # For latest NATS Server, add /v2 at the end
 go get github.com/nats-io/nats-server/v2
@@ -2,12 +2,14 @@

 This file lists the dependencies used in this repository.

-| Dependency | License |
-|-|-|
-| Go | BSD 3-Clause "New" or "Revised" License |
-| github.com/nats-io/nats.go | Apache License 2.0 |
-| github.com/golang/protobuf v1.4.2 | BSD 3-Clause "New" or "Revised" License |
-| github.com/nats-io/nats-server/v2 v2.1.8-0.20201115145023-f61fa8529a0f | Apache License 2.0 |
-| github.com/nats-io/nkeys v0.2.0 | Apache License 2.0 |
-| github.com/nats-io/nuid v1.0.1 | Apache License 2.0 |
-| google.golang.org/protobuf v1.23.0 | BSD 3-Clause License |
+| Dependency                        | License      |
+|-----------------------------------|--------------|
+| Go                                | BSD 3-Clause |
+| github.com/golang/protobuf/proto  | BSD-3-Clause |
+| github.com/klauspost/compress     | BSD-3-Clause |
+| github.com/nats-io/nats-server/v2 | Apache-2.0   |
+| github.com/nats-io/nkeys          | Apache-2.0   |
+| github.com/nats-io/nuid           | Apache-2.0   |
+| go.uber.org/goleak                | MIT          |
+| golang.org/x/text                 | BSD-3-Clause |
+| google.golang.org/protobuf        | BSD-3-Clause |
@@ -3223,6 +3223,9 @@ func (js *js) getConsumerInfoContext(ctx context.Context, stream, consumer strin
 		}
 		return nil, info.Error
 	}
+	if info.Error == nil && info.ConsumerInfo == nil {
+		return nil, ErrConsumerNotFound
+	}
 	return info.ConsumerInfo, nil
 }

@@ -47,7 +47,7 @@ import (

 // Default Constants
 const (
-	Version                   = "1.30.2"
+	Version                   = "1.31.0"
 	DefaultURL                = "nats://127.0.0.1:4222"
 	DefaultPort               = 4222
 	DefaultMaxReconnect       = 60
@@ -311,6 +311,13 @@ type Options struct {
 	// TLSCertCB is used to fetch and return custom tls certificate.
 	TLSCertCB TLSCertHandler

+	// TLSHandshakeFirst is used to instruct the library perform
+	// the TLS handshake right after the connect and before receiving
+	// the INFO protocol from the server. If this option is enabled
+	// but the server is not configured to perform the TLS handshake
+	// first, the connection will fail.
+	TLSHandshakeFirst bool
+
 	// RootCAsCB is used to fetch and return a set of root certificate
 	// authorities that clients use when verifying server certificates.
 	RootCAsCB RootCAsHandler
@@ -1315,6 +1322,17 @@ func SkipHostLookup() Option {
 	}
 }

+// TLSHandshakeFirst is an Option to perform the TLS handshake first, that is
+// before receiving the INFO protocol. This requires the server to also be
+// configured with such option, otherwise the connection will fail.
+func TLSHandshakeFirst() Option {
+	return func(o *Options) error {
+		o.TLSHandshakeFirst = true
+		o.Secure = true
+		return nil
+	}
+}
+
 // Handler processing

 // SetDisconnectHandler will set the disconnect event handler.
@@ -1481,6 +1499,12 @@ func (o Options) Connect() (*Conn, error) {
 		}
 	}

+	// If the TLSHandshakeFirst option is specified, make sure that
+	// the Secure boolean is true.
+	if nc.Opts.TLSHandshakeFirst {
+		nc.Opts.Secure = true
+	}
+
 	if err := nc.setupServerPool(); err != nil {
 		return nil, err
 	}
@@ -2235,6 +2259,14 @@ func (nc *Conn) processConnectInit() error {
 	// Set our status to connecting.
 	nc.changeConnStatus(CONNECTING)

+	// If we need to have a TLS connection and want the TLS handshake to occur
+	// first, do it now.
+	if nc.Opts.Secure && nc.Opts.TLSHandshakeFirst {
+		if err := nc.makeTLSConn(); err != nil {
+			return err
+		}
+	}
+
 	// Process the INFO protocol received from the server
 	err := nc.processExpectedInfo()
 	if err != nil {
@@ -2351,8 +2383,13 @@ func (nc *Conn) checkForSecure() error {
 		o.Secure = true
 	}

-	// Need to rewrap with bufio
 	if o.Secure {
+		// If TLS handshake first is true, we have already done
+		// the handshake, so we are done here.
+		if o.TLSHandshakeFirst {
+			return nil
+		}
+		// Need to rewrap with bufio
 		if err := nc.makeTLSConn(); err != nil {
 			return err
 		}
@@ -3040,28 +3077,6 @@ func (nc *Conn) waitForMsgs(s *Subscription) {
 // Return what is to be used. If we return nil the message will be dropped.
 type msgFilter func(m *Msg) *Msg

-func (nc *Conn) addMsgFilter(subject string, filter msgFilter) {
-	nc.subsMu.Lock()
-	defer nc.subsMu.Unlock()
-
-	if nc.filters == nil {
-		nc.filters = make(map[string]msgFilter)
-	}
-	nc.filters[subject] = filter
-}
-
-func (nc *Conn) removeMsgFilter(subject string) {
-	nc.subsMu.Lock()
-	defer nc.subsMu.Unlock()
-
-	if nc.filters != nil {
-		delete(nc.filters, subject)
-		if len(nc.filters) == 0 {
-			nc.filters = nil
-		}
-	}
-}
-
 // processMsg is called by parse and will place the msg on the
 // appropriate channel/pending queue for processing. If the channel is full,
 // or the pending queue is over the pending limits, the connection is
@@ -0,0 +1,59 @@
+// Copyright 2023 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
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build internal_testing
+// +build internal_testing
+
+// Functions in this file are only available when building nats.go with the
+// internal_testing build tag. They are used by the nats.go test suite.
+package nats
+
+// AddMsgFilter adds a message filter for the given subject
+// to the connection. The filter will be called for each
+// message received on the subject. If the filter returns
+// nil, the message will be dropped.
+func (nc *Conn) AddMsgFilter(subject string, filter msgFilter) {
+	nc.subsMu.Lock()
+	defer nc.subsMu.Unlock()
+
+	if nc.filters == nil {
+		nc.filters = make(map[string]msgFilter)
+	}
+	nc.filters[subject] = filter
+}
+
+// RemoveMsgFilter removes a message filter for the given subject.
+func (nc *Conn) RemoveMsgFilter(subject string) {
+	nc.subsMu.Lock()
+	defer nc.subsMu.Unlock()
+
+	if nc.filters != nil {
+		delete(nc.filters, subject)
+		if len(nc.filters) == 0 {
+			nc.filters = nil
+		}
+	}
+}
+
+// IsJSControlMessage returns true if the message is a JetStream control message.
+func IsJSControlMessage(msg *Msg) (bool, int) {
+	return isJSControlMessage(msg)
+}
+
+// CloseTCPConn closes the underlying TCP connection.
+// It can be used to simulate a disconnect.
+func (nc *Conn) CloseTCPConn() {
+	nc.mu.Lock()
+	defer nc.mu.Unlock()
+	nc.conn.Close()
+}
@@ -19,7 +19,7 @@ package nkeys
 import "io"

 // Version is our current version
-const Version = "0.4.5"
+const Version = "0.4.6"

 // KeyPair provides the central interface to nkeys.
 type KeyPair interface {
@@ -1,4 +1,4 @@
-// Copyright 2022 The NATS Authors
+// Copyright 2022-2023 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
@@ -38,12 +38,13 @@ type ckp struct {
 	seed [curveKeyLen]byte // Private raw key.
 }

-// CreateUser will create a User typed KeyPair.
+// CreateCurveKeys will create a Curve typed KeyPair.
 func CreateCurveKeys() (KeyPair, error) {
 	return CreateCurveKeysWithRand(rand.Reader)
 }

-// CreateUser will create a User typed KeyPair with specified rand source.
+// CreateCurveKeysWithRand will create a Curve typed KeyPair
+// with specified rand source.
 func CreateCurveKeysWithRand(rr io.Reader) (KeyPair, error) {
 	var kp ckp
 	_, err := io.ReadFull(rr, kp.seed[:])
@@ -85,7 +86,7 @@ func (pair *ckp) PrivateKey() ([]byte, error) {
 	return Encode(PrefixBytePrivate, pair.seed[:])
 }

-func decodePubCurveKey(src string, dest [curveKeyLen]byte) error {
+func decodePubCurveKey(src string, dest []byte) error {
 	var raw [curveDecodeLen]byte // should always be 35
 	n, err := b32Enc.Decode(raw[:], []byte(src))
 	if err != nil {
@@ -112,7 +113,7 @@ func decodePubCurveKey(src string, dest [curveKeyLen]byte) error {
 	}

 	// Copy over, ignore prefix byte.
-	copy(dest[:], raw[1:end])
+	copy(dest, raw[1:end])
 	return nil
 }

@@ -134,7 +135,7 @@ func (pair *ckp) SealWithRand(input []byte, recipient string, rr io.Reader) ([]b
 		err   error
 	)

-	if err = decodePubCurveKey(recipient, rpub); err != nil {
+	if err = decodePubCurveKey(recipient, rpub[:]); err != nil {
 		return nil, ErrInvalidRecipient
 	}
 	if _, err := io.ReadFull(rr, nonce[:]); err != nil {
@@ -159,7 +160,7 @@ func (pair *ckp) Open(input []byte, sender string) ([]byte, error) {
 	}
 	copy(nonce[:], input[vlen:vlen+curveNonceLen])

-	if err = decodePubCurveKey(sender, spub); err != nil {
+	if err = decodePubCurveKey(sender, spub[:]); err != nil {
 		return nil, ErrInvalidSender
 	}

@@ -1210,7 +1210,7 @@ github.com/justinas/alice
 # github.com/kevinburke/ssh_config v1.2.0
 ## explicit
 github.com/kevinburke/ssh_config
-# github.com/klauspost/compress v1.17.0
+# github.com/klauspost/compress v1.17.2
 ## explicit; go 1.18
 github.com/klauspost/compress/flate
 github.com/klauspost/compress/s2
@@ -1367,7 +1367,7 @@ github.com/mschoch/smat
 # github.com/nats-io/jwt/v2 v2.5.2
 ## explicit; go 1.18
 github.com/nats-io/jwt/v2
-# github.com/nats-io/nats-server/v2 v2.10.2
+# github.com/nats-io/nats-server/v2 v2.10.4
 ## explicit; go 1.20
 github.com/nats-io/nats-server/v2/conf
 github.com/nats-io/nats-server/v2/internal/ldap
@@ -1378,13 +1378,13 @@ github.com/nats-io/nats-server/v2/server/certidp
 github.com/nats-io/nats-server/v2/server/certstore
 github.com/nats-io/nats-server/v2/server/pse
 github.com/nats-io/nats-server/v2/server/sysmem
-# github.com/nats-io/nats.go v1.30.2
+# github.com/nats-io/nats.go v1.31.0
 ## explicit; go 1.20
 github.com/nats-io/nats.go
 github.com/nats-io/nats.go/encoders/builtin
 github.com/nats-io/nats.go/internal/parser
 github.com/nats-io/nats.go/util
-# github.com/nats-io/nkeys v0.4.5
+# github.com/nats-io/nkeys v0.4.6
 ## explicit; go 1.19
 github.com/nats-io/nkeys
 # github.com/nats-io/nuid v1.0.1