Repository: nats-io/nats-server
Pull request #2178 (2021-05-04 17:45:52)

[FIXED] MQTT fixes and improvements
Some issues that have been fixed would manifest by timeouts on
connect, unexpected memory usage on high publish message rate.

Some details:
- Replies were not always GW routed properly because we were looking
at the wrong connection's rsubs
- GW routed replies would not be found because they were tracked
in the subscription's client object, which may not be the same used
to send the reply
- Increased the mqtt timeout to wait for JS replies since in some
tests it was sometimes taking more than the original 2 seconds
- Incoming gateway messages destined for an MQTT internal subscription
may have been rejected as a no interest if the account had service imports
- Don't use time.After(), instead create explicit timer so it can
be stopped when not timing out.
- Unnecessary copy of a slice since we were converting to a string anyway.

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>



Patch:
diff --git a/go.mod b/go.mod
index d9a490b0..ff9c0d72 100644
--- a/go.mod
+++ b/go.mod
@@ -6,7 +6,7 @@ require (
 	github.com/golang/protobuf v1.4.2 // indirect
 	github.com/klauspost/compress v1.11.12
 	github.com/minio/highwayhash v1.0.1
-	github.com/nats-io/jwt/v2 v2.0.2
+	github.com/nats-io/jwt/v2 v2.0.1
 	github.com/nats-io/nats.go v1.10.1-0.20210419223411-20527524c393
 	github.com/nats-io/nkeys v0.3.0
 	github.com/nats-io/nuid v1.0.1
diff --git a/go.sum b/go.sum
index 7df6a1ba..9c617a6c 100644
--- a/go.sum
+++ b/go.sum
@@ -14,8 +14,8 @@ github.com/minio/highwayhash v1.0.1 h1:dZ6IIu8Z14VlC0VpfKofAhCy74wu/Qb5gcn52yWoz
 github.com/minio/highwayhash v1.0.1/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
 github.com/nats-io/jwt v1.2.2 h1:w3GMTO969dFg+UOKTmmyuu7IGdusK+7Ytlt//OYH/uU=
 github.com/nats-io/jwt v1.2.2/go.mod h1:/xX356yQA6LuXI9xWW7mZNpxgF2mBmGecH+Fj34sP5Q=
-github.com/nats-io/jwt/v2 v2.0.2 h1:ejVCLO8gu6/4bOKIHQpmB5UhhUJfAQw55yvLWpfmKjI=
-github.com/nats-io/jwt/v2 v2.0.2/go.mod h1:VRP+deawSXyhNjXmxPCHskrR6Mq50BqpEI5SEcNiGlY=
+github.com/nats-io/jwt/v2 v2.0.1 h1:SycklijeduR742i/1Y3nRhURYM7imDzZZ3+tuAQqhQA=
+github.com/nats-io/jwt/v2 v2.0.1/go.mod h1:VRP+deawSXyhNjXmxPCHskrR6Mq50BqpEI5SEcNiGlY=
 github.com/nats-io/nats.go v1.10.1-0.20210419223411-20527524c393 h1:GQxfDz4otI9mde5QqJlpyRNpa2tfURHiPy0YLf7hy4c=
 github.com/nats-io/nats.go v1.10.1-0.20210419223411-20527524c393/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
 github.com/nats-io/nkeys v0.2.0/go.mod h1:XdZpAbhgyyODYqjTawOnIOI7VlbKSarI9Gfy1tqEu/s=
diff --git a/server/accounts.go b/server/accounts.go
index b2e5e6bc..f3e045fd 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -42,6 +42,7 @@ const globalAccountName = DEFAULT_GLOBAL_ACCOUNT
 // Account are subject namespace definitions. By default no messages are shared between accounts.
 // You can share via Exports and Imports of Streams and Services.
 type Account struct {
+	gwReplyMapping
 	Name         string
 	Nkey         string
 	Issuer       string
@@ -1839,7 +1840,7 @@ func (a *Account) addServiceImportSub(si *serviceImport) error {
 	cb := func(sub *subscription, c *client, subject, reply string, msg []byte) {
 		c.processServiceImport(si, a, msg)
 	}
-	_, err := c.processSubEx([]byte(subject), nil, []byte(sid), cb, true, true)
+	_, err := c.processSubEx([]byte(subject), nil, []byte(sid), cb, true, true, false)
 
 	return err
 }
@@ -2039,9 +2040,7 @@ func (a *Account) createRespWildcard() []byte {
 	a.mu.Unlock()
 
 	// Create subscription and internal callback for all the wildcard response subjects.
-	if sub, err := c.processSub(wcsub, nil, []byte(sid), a.processServiceImportResponse, false); err == nil {
-		sub.rsi = true
-	}
+	c.processSubEx(wcsub, nil, []byte(sid), a.processServiceImportResponse, false, false, true)
 
 	return pre
 }
diff --git a/server/client.go b/server/client.go
index 69a833a5..cf6ff14d 100644
--- a/server/client.go
+++ b/server/client.go
@@ -215,11 +215,7 @@ const (
 type client struct {
 	// Here first because of use of atomics, and memory alignment.
 	stats
-	// Indicate if we should check gwrm or not. Since checking gwrm is done
-	// when processing inbound messages and requires the lock we want to
-	// check only when needed. This is set/get using atomic, so needs to
-	// be memory aligned.
-	cgwrt int32
+	gwReplyMapping
 	kind  int
 	srv   *Server
 	acc   *Account
@@ -262,9 +258,6 @@ type client struct {
 	ws    *websocket
 	mqtt  *mqtt
 
-	// To keep track of gateway replies mapping
-	gwrm map[string]*gwReplyMap
-
 	flags clientFlag // Compact booleans into a single field. Size will be increased when needed.
 
 	rref byte
@@ -2354,12 +2347,12 @@ func (c *client) parseSub(argo []byte, noForward bool) error {
 }
 
 func (c *client) processSub(subject, queue, bsid []byte, cb msgHandler, noForward bool) (*subscription, error) {
-	return c.processSubEx(subject, queue, bsid, cb, noForward, false)
+	return c.processSubEx(subject, queue, bsid, cb, noForward, false, false)
 }
 
-func (c *client) processSubEx(subject, queue, bsid []byte, cb msgHandler, noForward, si bool) (*subscription, error) {
+func (c *client) processSubEx(subject, queue, bsid []byte, cb msgHandler, noForward, si, rsi bool) (*subscription, error) {
 	// Create the subscription
-	sub := &subscription{client: c, subject: subject, queue: queue, sid: bsid, icb: cb, si: si}
+	sub := &subscription{client: c, subject: subject, queue: queue, sid: bsid, icb: cb, si: si, rsi: rsi}
 
 	c.mu.Lock()
 
@@ -3065,22 +3058,18 @@ func (c *client) deliverMsg(sub *subscription, subject, reply, mh, msg []byte, g
 
 	// Check for internal subscriptions.
 	if sub.icb != nil && !c.noIcb {
-		ireply := string(reply)
-		// For internal callbacks we don't want to rely on the GW routed reply tracking
-		// since clients can change for who receives vs sends response. GW routed
-		// currently pins to a destination client connection. Also for internal its
-		// ok if we do not hide the raw reply like we do for external clients. We
-		// will not remap service import replies since they are handled different.
-		if gwrply && !isServiceReply(reply) {
-			ireply = string(c.pa.reply)
+		if gwrply {
+			// We will store in the account, not the client since it will likely
+			// be a different client that will send the reply.
+			srv.trackGWReply(nil, client.acc, reply, c.pa.reply)
 		}
 		client.mu.Unlock()
 
 		// Internal account clients are for service imports and need the '\r\n'.
 		if client.kind == ACCOUNT {
-			sub.icb(sub, c, string(subject), ireply, msg)
+			sub.icb(sub, c, string(subject), string(reply), msg)
 		} else {
-			sub.icb(sub, c, string(subject), ireply, msg[:msgSize])
+			sub.icb(sub, c, string(subject), string(reply), msg[:msgSize])
 		}
 		return true
 	}
@@ -3116,7 +3105,7 @@ func (c *client) deliverMsg(sub *subscription, subject, reply, mh, msg []byte, g
 			// connection (`client`). The routed reply subject is in `c.pa.reply`,
 			// should that change, we would have to pass the GW routed reply as
 			// a parameter of deliverMsg().
-			srv.trackGWReply(client, c.pa.reply)
+			srv.trackGWReply(client, nil, reply, c.pa.reply)
 		}
 
 		// If we do not have a registered RTT queue that up now.
@@ -3454,9 +3443,19 @@ func (c *client) processInboundClientMsg(msg []byte) (bool, bool) {
 		c.mqttHandlePubRetain()
 	}
 
-	// Check if this client's gateway replies map is not empty
-	if atomic.LoadInt32(&c.cgwrt) > 0 && c.handleGWReplyMap(msg) {
-		return true, false
+	// Doing this inline as opposed to create a function (which otherwise has a measured
+	// performance impact reported in our bench)
+	var isGWRouted bool
+	if c.kind != CLIENT {
+		if atomic.LoadInt32(&c.acc.gwReplyMapping.check) > 0 {
+			c.acc.mu.RLock()
+			c.pa.subject, isGWRouted = c.acc.gwReplyMapping.get(c.pa.subject)
+			c.acc.mu.RUnlock()
+		}
+	} else if atomic.LoadInt32(&c.gwReplyMapping.check) > 0 {
+		c.mu.Lock()
+		c.pa.subject, isGWRouted = c.gwReplyMapping.get(c.pa.subject)
+		c.mu.Unlock()
 	}
 
 	// If we have an exported service and we are doing remote tracking, check this subject
@@ -3482,6 +3481,13 @@ func (c *client) processInboundClientMsg(msg []byte) (bool, bool) {
 		}
 	}
 
+	// If the subject was converted to the gateway routed subject, then handle it now
+	// and be done with the rest of this function.
+	if isGWRouted {
+		c.handleGWReplyMap(msg)
+		return true, false
+	}
+
 	// Match the subscriptions. We will use our own L1 map if
 	// it's still valid, avoiding contention on the shared sublist.
 	var r *SublistResult
@@ -3571,43 +3577,10 @@ func (c *client) subForReply(reply []byte) *subscription {
 	return nil
 }
 
-// This is invoked knowing that this client has some GW replies
-// in its map. It will check if one is find for the c.pa.subject
-// and if so will process it directly (send to GWs and LEAFs) and
-// return true to notify the caller that the message was handled.
-// If there is no mapping for the subject, false is returned.
+// This is invoked knowing that c.pa.subject has been set to the gateway routed subject.
+// This function will send the message to possibly LEAFs and directly back to the origin
+// gateway.
 func (c *client) handleGWReplyMap(msg []byte) bool {
-	c.mu.Lock()
-	rm, ok := c.gwrm[string(c.pa.subject)]
-	if !ok {
-		c.mu.Unlock()
-		return false
-	}
-	// Set subject to the mapped reply subject
-	c.pa.subject = []byte(rm.ms)
-
-	var rl *remoteLatency
-
-	if c.rrTracking != nil {
-		rl = c.rrTracking.rmap[string(c.pa.subject)]
-		if rl != nil {
-			delete(c.rrTracking.rmap, string(c.pa.subject))
-		}
-	}
-	c.mu.Unlock()
-
-	if rl != nil {
-		sl := &rl.M2
-		// Fill this in and send it off to the other side.
-		sl.Status = 200
-		sl.Responder = c.getClientInfo(true)
-		sl.ServiceLatency = time.Since(sl.RequestStart) - sl.Responder.RTT
-		sl.TotalLatency = sl.ServiceLatency + sl.Responder.RTT
-		sanitizeLatencyMetric(sl)
-		lsub := remoteLatencySubjectForResponse(c.pa.subject)
-		c.srv.sendInternalAccountMsg(nil, lsub, rl) // Send to SYS account
-	}
-
 	// Check for leaf nodes
 	if c.srv.gwLeafSubs.Count() > 0 {
 		if r := c.srv.gwLeafSubs.Match(string(c.pa.subject)); len(r.psubs) > 0 {
@@ -3617,7 +3590,6 @@ func (c *client) handleGWReplyMap(msg []byte) bool {
 	if c.srv.gateway.enabled {
 		c.sendMsgToGateways(c.acc, msg, c.pa.subject, c.pa.reply, nil)
 	}
-
 	return true
 }
 
diff --git a/server/const.go b/server/const.go
index 2b1b74b3..c43ef2e6 100644
--- a/server/const.go
+++ b/server/const.go
@@ -41,7 +41,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.2.3-beta.5"
+	VERSION = "2.2.3-beta.4"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/errors.go b/server/errors.go
index 684cce51..15eff79c 100644
--- a/server/errors.go
+++ b/server/errors.go
@@ -1,4 +1,4 @@
-// Copyright 2012-2021 The NATS Authors
+// Copyright 2012-2020 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
@@ -202,12 +202,6 @@ var (
 
 	// ErrJetStreamResourcesExceeded is returned when a call would exceed internal resource limits.
 	ErrJetStreamResourcesExceeded = errors.New("jetstream resources exceeded for server")
-
-	// ErrStorageResourcesExceeded is returned when storage resources would be exceeded.
-	ErrStorageResourcesExceeded = errors.New("insufficient storage resources available")
-
-	// ErrMemoryResourcesExceeded is returned when memory resources would be exceeded.
-	ErrMemoryResourcesExceeded = errors.New("insufficient memory resources available")
 )
 
 // configErr is a configuration error.
diff --git a/server/filestore.go b/server/filestore.go
index 3bada57a..c1f71a01 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -65,24 +65,25 @@ type FileConsumerInfo struct {
 }
 
 type fileStore struct {
-	mu      sync.RWMutex
-	state   StreamState
-	ld      *LostStreamData
-	scb     StorageUpdateHandler
-	ageChk  *time.Timer
-	syncTmr *time.Timer
-	cfg     FileStreamInfo
-	fcfg    FileStoreConfig
-	lmb     *msgBlock
-	blks    []*msgBlock
-	hh      hash.Hash64
-	qch     chan struct{}
-	cfs     []*consumerFileStore
-	fsi     map[string]seqSlice
-	fsis    *simpleState
-	closed  bool
-	fip     bool
-	sips    int
+	mu       sync.RWMutex
+	state    StreamState
+	ld       *LostStreamData
+	scb      StorageUpdateHandler
+	ageChk   *time.Timer
+	syncTmr  *time.Timer
+	cfg      FileStreamInfo
+	fcfg     FileStoreConfig
+	lmb      *msgBlock
+	blks     []*msgBlock
+	hh       hash.Hash64
+	qch      chan struct{}
+	cfs      []*consumerFileStore
+	fsi      map[string]seqSlice
+	fsis     *simpleState
+	closed   bool
+	expiring bool
+	fip      bool
+	sips     int
 }
 
 // Represents a message store block and its data.
@@ -97,9 +98,8 @@ type msgBlock struct {
 	ifn     string
 	ifd     *os.File
 	liwsz   int64
-	index   uint64 // User visible message count.
-	bytes   uint64 // User visible bytes count.
-	rbytes  uint64 // Total bytes (raw) including deleted. Used for rolling to new blk.
+	index   uint64
+	bytes   uint64
 	msgs    uint64
 	lwits   int64
 	lwts    int64
@@ -203,7 +203,7 @@ func newFileStore(fcfg FileStoreConfig, cfg StreamConfig) (*fileStore, error) {
 }
 
 func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created time.Time) (*fileStore, error) {
-	if cfg.Name == _EMPTY_ {
+	if cfg.Name == "" {
 		return nil, fmt.Errorf("name required")
 	}
 	if cfg.Storage != FileStorage {
@@ -299,7 +299,7 @@ func (fs *fileStore) UpdateConfig(cfg *StreamConfig) error {
 		return ErrStoreClosed
 	}
 
-	if cfg.Name == _EMPTY_ {
+	if cfg.Name == "" {
 		return fmt.Errorf("name required")
 	}
 	if cfg.Storage != FileStorage {
@@ -315,11 +315,9 @@ func (fs *fileStore) UpdateConfig(cfg *StreamConfig) error {
 		fs.mu.Unlock()
 		return err
 	}
-
 	// Limits checks and enforcement.
 	fs.enforceMsgLimit()
 	fs.enforceBytesLimit()
-
 	// Do age timers.
 	if fs.ageChk == nil && fs.cfg.MaxAge != 0 {
 		fs.startAgeChk()
@@ -411,12 +409,6 @@ func (fs *fileStore) recoverMsgBlock(fi os.FileInfo, index uint64) *msgBlock {
 	}
 	defer file.Close()
 
-	if fi, _ := file.Stat(); fi != nil {
-		mb.rbytes = uint64(fi.Size())
-	} else {
-		return nil
-	}
-
 	// Read our index file. Use this as source of truth if possible.
 	if err := mb.readIndexInfo(); err == nil {
 		// Quick sanity check here.
@@ -486,7 +478,7 @@ func (mb *msgBlock) rebuildState() (*LostStreamData, error) {
 	startLastSeq := mb.last.seq
 
 	// Clear state we need to rebuild.
-	mb.msgs, mb.bytes, mb.rbytes = 0, 0, 0
+	mb.msgs, mb.bytes = 0, 0
 	mb.last.seq, mb.last.ts = 0, 0
 	firstNeedsSet := true
 
@@ -494,7 +486,6 @@ func (mb *msgBlock) rebuildState() (*LostStreamData, error) {
 	if err != nil {
 		return nil, err
 	}
-	mb.rbytes = uint64(len(buf))
 
 	addToDmap := func(seq uint64) {
 		if seq == 0 {
@@ -622,7 +613,6 @@ func (mb *msgBlock) rebuildState() (*LostStreamData, error) {
 
 			mb.msgs++
 			mb.bytes += uint64(rl)
-			mb.rbytes += uint64(rl)
 		}
 
 		index += rl
@@ -858,15 +848,12 @@ func (mb *msgBlock) setupWriteCache(buf []byte) {
 	if mb.cache != nil {
 		return
 	}
-	if buf != nil {
-		buf = buf[:0]
-	}
 	mb.cache = &cache{buf: buf}
 	// Make sure we set the proper cache offset if we have existing data.
 	var fi os.FileInfo
 	if mb.mfd != nil {
 		fi, _ = mb.mfd.Stat()
-	} else if mb.mfn != _EMPTY_ {
+	} else {
 		fi, _ = os.Stat(mb.mfn)
 	}
 	if fi != nil {
@@ -1145,7 +1132,7 @@ func (fs *fileStore) enforceMsgLimit() {
 		return
 	}
 	for nmsgs := fs.state.Msgs; nmsgs > uint64(fs.cfg.MaxMsgs); nmsgs = fs.state.Msgs {
-		if removed, err := fs.deleteFirstMsg(); err != nil || !removed {
+		if removed, err := fs.deleteFirstMsgLocked(); err != nil || !removed {
 			fs.rebuildFirst()
 			return
 		}
@@ -1159,20 +1146,28 @@ func (fs *fileStore) enforceBytesLimit() {
 		return
 	}
 	for bs := fs.state.Bytes; bs > uint64(fs.cfg.MaxBytes); bs = fs.state.Bytes {
-		if removed, err := fs.deleteFirstMsg(); err != nil || !removed {
+		if removed, err := fs.deleteFirstMsgLocked(); err != nil || !removed {
 			fs.rebuildFirst()
 			return
 		}
 	}
 }
 
-// Lock should be held on entry but will be released during actual remove.
-func (fs *fileStore) deleteFirstMsg() (bool, error) {
+// Lock should be held but will be released during actual remove.
+func (fs *fileStore) deleteFirstMsgLocked() (bool, error) {
 	fs.mu.Unlock()
 	defer fs.mu.Lock()
 	return fs.removeMsg(fs.state.FirstSeq, false)
 }
 
+// Lock should NOT be held.
+func (fs *fileStore) deleteFirstMsg() (bool, error) {
+	fs.mu.RLock()
+	seq := fs.state.FirstSeq
+	fs.mu.RUnlock()
+	return fs.removeMsg(seq, false)
+}
+
 // RemoveMsg will remove the message from this store.
 // Will return the number of bytes removed.
 func (fs *fileStore) RemoveMsg(seq uint64) (bool, error) {
@@ -1460,7 +1455,7 @@ func (mb *msgBlock) flushLoop(fch, qch chan struct{}) {
 					waiting = newWaiting
 					ts *= 2
 				}
-				mb.flushPendingMsgsAndWait()
+				mb.flushPendingMsgs()
 				// Check if we are no longer the last message block. If we are
 				// not we can close FDs and exit.
 				mb.fs.mu.RLock()
@@ -1588,8 +1583,6 @@ func (mb *msgBlock) truncate(sm *fileStoredMsg) (nmsgs, nbytes uint64, err error
 	// Do local mb stat updates.
 	mb.msgs -= purged
 	mb.bytes -= bytes
-	mb.rbytes -= bytes
-
 	// Update our last msg.
 	mb.last.seq = sm.seq
 	mb.last.ts = sm.ts
@@ -1777,44 +1770,48 @@ func (fs *fileStore) expireMsgsLocked() {
 	fs.mu.Lock()
 }
 
-// Lock should be held.
-func (fs *fileStore) resetAgeChk(delta int64) {
-	fireIn := fs.cfg.MaxAge
-	if delta > 0 {
-		fireIn = time.Duration(delta)
-	}
-	if fs.ageChk != nil {
-		fs.ageChk.Reset(fireIn)
-	} else {
-		fs.ageChk = time.AfterFunc(fireIn, fs.expireMsgs)
-	}
-}
-
-// Lock should be held.
-func (fs *fileStore) cancelAgeChk() {
-	if fs.ageChk != nil {
-		fs.ageChk.Stop()
-		fs.ageChk = nil
-	}
-}
-
 // Will expire msgs that are too old.
 func (fs *fileStore) expireMsgs() {
-	// We need to delete one by one here and can not optimize for the time being.
-	// Reason is that we need more information to adjust ack pending in consumers.
-	var sm *fileStoredMsg
-	minAge := time.Now().UnixNano() - int64(fs.cfg.MaxAge)
-	for sm, _ = fs.msgForSeq(0); sm != nil && sm.ts <= minAge; sm, _ = fs.msgForSeq(0) {
-		fs.removeMsg(sm.seq, false)
+	// Make sure this is only running one at a time.
+	fs.mu.Lock()
+	if fs.expiring {
+		fs.mu.Unlock()
+		return
 	}
+	fs.expiring = true
+	fs.mu.Unlock()
 
-	fs.mu.Lock()
-	defer fs.mu.Unlock()
+	defer func() {
+		fs.mu.Lock()
+		fs.expiring = false
+		fs.mu.Unlock()
+	}()
 
-	if sm == nil {
-		fs.cancelAgeChk()
-	} else {
-		fs.resetAgeChk(sm.ts - minAge)
+	now := time.Now().UnixNano()
+	minAge := now - int64(fs.cfg.MaxAge)
+
+	for {
+		sm, _ := fs.msgForSeq(0)
+		if sm != nil && sm.ts <= minAge {
+			fs.deleteFirstMsg()
+		} else {
+			fs.mu.Lock()
+			if sm == nil {
+				if fs.ageChk != nil {
+					fs.ageChk.Stop()
+					fs.ageChk = nil
+				}
+			} else {
+				fireIn := time.Duration(sm.ts-now) + fs.cfg.MaxAge
+				if fs.ageChk != nil {
+					fs.ageChk.Reset(fireIn)
+				} else {
+					fs.ageChk = time.AfterFunc(fireIn, fs.expireMsgs)
+				}
+			}
+			fs.mu.Unlock()
+			return
+		}
 	}
 }
 
@@ -1925,7 +1922,7 @@ func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte
 	// If we should be flushing in place do so here. We will also flip to flushing in place if we
 	// had a write error.
 	if flush || werr != nil {
-		if err := mb.flushPendingMsgsAndWait(); err != nil {
+		if err := mb.flushPendingMsgs(); err != nil && err != errFlushRunning && err != errNoPending {
 			return err
 		}
 		if writeIndex {
@@ -1955,10 +1952,8 @@ func (mb *msgBlock) pendingWriteSize() int {
 	return pending
 }
 
-// Lock should NOT be held.
+// Lock should be held.
 func (mb *msgBlock) clearFlushing() {
-	mb.mu.Lock()
-	defer mb.mu.Unlock()
 	if mb.cache != nil {
 		mb.cache.flush = false
 	}
@@ -2013,10 +2008,10 @@ func (mb *msgBlock) bytesPending() ([]byte, error) {
 	return buf, nil
 }
 
-// Returns the current blkSize including deleted msgs etc.
-func (mb *msgBlock) blkSize() uint64 {
+// Return the number of bytes in this message block.
+func (mb *msgBlock) numBytes() uint64 {
 	mb.mu.RLock()
-	nb := mb.rbytes
+	nb := mb.bytes
 	mb.mu.RUnlock()
 	return nb
 }
@@ -2032,7 +2027,6 @@ func (mb *msgBlock) updateAccounting(seq uint64, ts int64, rl uint64) {
 	atomic.StoreUint64(&mb.last.seq, seq)
 	mb.last.ts = ts
 	mb.bytes += rl
-	mb.rbytes += rl
 	mb.msgs++
 }
 
@@ -2047,7 +2041,7 @@ func (fs *fileStore) writeMsgRecord(seq uint64, ts int64, subj string, hdr, msg
 	}
 	// Grab our current last message block.
 	mb := fs.lmb
-	if mb == nil || mb.blkSize()+rl > fs.fcfg.BlockSize {
+	if mb == nil || mb.numBytes()+rl > fs.fcfg.BlockSize {
 		if mb, err = fs.newMsgBlockForWrite(); err != nil {
 			return 0, err
 		}
@@ -2212,22 +2206,14 @@ func (mb *msgBlock) quitChan() chan struct{} {
 // This function is called for in place flushing so we need to wait.
 func (mb *msgBlock) flushPendingMsgsAndWait() error {
 	var err error
-	var t *time.Timer
-	const delay = time.Millisecond
 
 	// If we are in flush wait for that to clear.
 	for err = mb.flushPendingMsgs(); err == errFlushRunning; err = mb.flushPendingMsgs() {
 		qch := mb.quitChan()
-		if t == nil {
-			t = time.NewTimer(delay)
-			defer t.Stop()
-		} else {
-			t.Reset(delay)
-		}
 		select {
 		case <-qch:
 			return nil
-		case <-t.C:
+		case <-time.After(time.Millisecond):
 		}
 	}
 	return err
@@ -2250,7 +2236,7 @@ func (mb *msgBlock) flushPendingMsgs() error {
 	if err != nil {
 		mb.mu.Unlock()
 		// No pending data to be written is not an error.
-		if err == errNoPending || err == errNoCache {
+		if err == errNoPending {
 			err = nil
 		}
 		return err
@@ -2261,9 +2247,6 @@ func (mb *msgBlock) flushPendingMsgs() error {
 
 	// Only one can be flushing at a time.
 	mb.setFlushing()
-	// Clear on exit.
-	defer mb.clearFlushing()
-
 	mfd := mb.mfd
 	mb.mu.Unlock()
 
@@ -2289,20 +2272,20 @@ func (mb *msgBlock) flushPendingMsgs() error {
 		woff += int64(n)
 		tn += n
 
-		// Partial write.
-		if n != lbb {
-			buf = buf[n:]
-		} else {
-			// Done.
+		// Success
+		if n == lbb {
 			break
 		}
+		// Partial write..
+		buf = buf[n:]
 	}
 
 	// We did a successful write.
 	// Re-acquire lock to update.
 	mb.mu.Lock()
 	defer mb.mu.Unlock()
-
+	// Clear on exit.
+	defer mb.clearFlushing()
 	// set write err to any error.
 	mb.werr = err
 
@@ -2646,7 +2629,7 @@ func (fs *fileStore) LoadMsg(seq uint64) (string, []byte, []byte, int64, error)
 	if sm != nil {
 		return sm.subj, sm.hdr, sm.msg, sm.ts, nil
 	}
-	return _EMPTY_, nil, nil, 0, err
+	return "", nil, nil, 0, err
 }
 
 // Type returns the type of the underlying store.
@@ -2984,11 +2967,6 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 		fs.mu.Unlock()
 		return 0, nil
 	}
-	if err := smb.loadMsgs(); err != nil {
-		fs.mu.Unlock()
-		return 0, err
-	}
-
 	// All msgblocks up to this one can be thrown away.
 	for i, mb := range fs.blks {
 		if mb == smb {
@@ -3001,13 +2979,16 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 		mb.dirtyCloseWithRemove(true)
 		mb.mu.Unlock()
 	}
+	fs.mu.Unlock()
+
+	if err := smb.loadMsgs(); err != nil {
+		return purged, err
+	}
 
 	smb.mu.Lock()
 	for mseq := smb.first.seq; mseq < seq; mseq++ {
 		if sm, _ := smb.cacheLookupWithLock(mseq); sm != nil && smb.msgs > 0 {
-			sz := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
-			smb.bytes -= sz
-			bytes += sz
+			smb.bytes -= fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
 			smb.msgs--
 			purged++
 		}
@@ -3023,17 +3004,12 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 
 	if sm != nil {
 		// Reset our version of first.
+		fs.mu.Lock()
 		fs.state.FirstSeq = sm.seq
 		fs.state.FirstTime = time.Unix(0, sm.ts).UTC()
 		fs.state.Msgs -= purged
 		fs.state.Bytes -= bytes
-	}
-
-	cb := fs.scb
-	fs.mu.Unlock()
-
-	if cb != nil {
-		cb(-int64(purged), -int64(bytes), 0, _EMPTY_)
+		fs.mu.Unlock()
 	}
 
 	return purged, nil
@@ -3253,14 +3229,6 @@ func (fs *fileStore) Delete() error {
 	return os.RemoveAll(fs.fcfg.StoreDir)
 }
 
-// Lock should be held.
-func (fs *fileStore) cancelSyncTimer() {
-	if fs.syncTmr != nil {
-		fs.syncTmr.Stop()
-		fs.syncTmr = nil
-	}
-}
-
 func (fs *fileStore) Stop() error {
 	fs.mu.Lock()
 	if fs.closed {
@@ -3273,8 +3241,14 @@ func (fs *fileStore) Stop() error {
 	fs.checkAndFlushAllBlocks()
 	fs.closeAllMsgBlocks(false)
 
-	fs.cancelSyncTimer()
-	fs.cancelAgeChk()
+	if fs.syncTmr != nil {
+		fs.syncTmr.Stop()
+		fs.syncTmr = nil
+	}
+	if fs.ageChk != nil {
+		fs.ageChk.Stop()
+		fs.ageChk = nil
+	}
 
 	var _cfs [256]*consumerFileStore
 	cfs := append(_cfs[:0], fs.cfs...)
@@ -3517,7 +3491,7 @@ func (fs *fileStore) ConsumerStore(name string, cfg *ConsumerConfig) (ConsumerSt
 	if fs.isClosed() {
 		return nil, ErrStoreClosed
 	}
-	if cfg == nil || name == _EMPTY_ {
+	if cfg == nil || name == "" {
 		return nil, fmt.Errorf("bad consumer config")
 	}
 	odir := path.Join(fs.fcfg.StoreDir, consumerDir, name)
diff --git a/server/gateway.go b/server/gateway.go
index f0b94296..d9ed2889 100644
--- a/server/gateway.go
+++ b/server/gateway.go
@@ -238,6 +238,29 @@ type gwReplyMap struct {
 	exp int64
 }
 
+type gwReplyMapping struct {
+	// Indicate if we should check the map or not. Since checking the map is done
+	// when processing inbound messages and requires the lock we want to
+	// check only when needed. This is set/get using atomic, so needs to
+	// be memory aligned.
+	check int32
+	// To keep track of gateway replies mapping
+	mapping map[string]*gwReplyMap
+}
+
+// Returns the corresponding gw routed subject, and `true` to indicate that a
+// mapping was found. If no entry is found, the passed subject is returned
+// as-is and `false` is returned to indicate that no mapping was found.
+// Caller is responsible to ensure the locking.
+func (g *gwReplyMapping) get(subject []byte) ([]byte, bool) {
+	rm, ok := g.mapping[string(subject)]
+	if !ok {
+		return subject, false
+	}
+	subj := []byte(rm.ms)
+	return subj, true
+}
+
 // clone returns a deep copy of the RemoteGatewayOpts object
 func (r *RemoteGatewayOpts) clone() *RemoteGatewayOpts {
 	if r == nil {
@@ -2332,8 +2355,9 @@ func hasGWRoutedReplyPrefix(subj []byte) bool {
 func (g *srvGateway) shouldMapReplyForGatewaySend(c *client, acc *Account, reply []byte) bool {
 	// If the reply is a service reply (_R_), we will use the account's internal
 	// client instead of the client handed to us. This client holds the wildcard
-	// for all service replies.
-	if isServiceReply(reply) {
+	// for all service replies. For other kind of connections, we still use the
+	// given `client` object.
+	if isServiceReply(reply) && c.kind == CLIENT {
 		acc.mu.Lock()
 		c = acc.internalClient()
 		acc.mu.Unlock()
@@ -2495,7 +2519,7 @@ func (c *client) sendMsgToGateways(acc *Account, msg, subject, reply []byte, qgr
 				mh = append(mh, "| "...) // Only queues
 			}
 			mh = append(mh, queues...)
-		} else if reply != nil {
+		} else if len(reply) > 0 {
 			mh = append(mh, mreply...)
 			mh = append(mh, ' ')
 		}
@@ -2863,7 +2887,10 @@ func (c *client) processInboundGatewayMsg(msg []byte) {
 			// make sure we send the proper no interest if the service import is the only interest.
 			noInterest = true
 			for _, sub := range r.psubs {
-				if sub.client.kind != ACCOUNT {
+				// sub.si indicates that this is a subscription for service import, and is immutable.
+				// So sub.si is false, then this is a subscription for something else, so there is
+				// actually proper interest.
+				if !sub.si {
 					noInterest = false
 					break
 				}
@@ -3021,33 +3048,50 @@ func (c *client) gatewaySwitchAccountToSendAllSubs(e *insie, accName string) {
 	})
 }
 
-// Keeps track of the routed reply to be used when/if application
-// sends back a message on the reply without the prefix.
-// Client lock held on entry. This is a server receiver because
-// we use a timer interval that is avail in Server.gateway object.
-func (s *Server) trackGWReply(c *client, reply []byte) {
-	ttl := s.gateway.recSubExp
-	rm := c.gwrm
-	var we bool // will be true if map was empty on entry
-	if rm == nil {
-		rm = make(map[string]*gwReplyMap)
-		c.gwrm = rm
-		we = true
+// Keeps track of the routed reply to be used when/if application sends back a
+// message on the reply without the prefix.
+// If `client` is not nil, it will be stored in the client gwReplyMapping structure,
+// and client lock is held on entry.
+// If `client` is nil, the mapping is stored in the client's account's gwReplyMapping
+// structure. Account lock will be explicitly acquired.
+// This is a server receiver because we use a timer interval that is avail in
+/// Server.gateway object.
+func (s *Server) trackGWReply(c *client, acc *Account, reply, routedReply []byte) {
+	var l sync.Locker
+	var g *gwReplyMapping
+	if acc != nil {
+		acc.mu.Lock()
+		defer acc.mu.Unlock()
+		g = &acc.gwReplyMapping
+		l = &acc.mu
 	} else {
-		we = len(rm) == 0
+		g = &c.gwReplyMapping
+		l = &c.mu
 	}
+	ttl := s.gateway.recSubExp
+	wasEmpty := len(g.mapping) == 0
+	if g.mapping == nil {
+		g.mapping = make(map[string]*gwReplyMap)
+	}
+	// The reason we pass both `reply` and `routedReply`, is that in some cases,
+	// `routedReply` may have a deliver subject appended, something look like:
+	// "_GR_.xxx.yyy.$JS.ACK.$MQTT_msgs.someid.1.1.1.1620086713306484000.0@$MQTT.msgs.foo"
+	// but `reply` has already been cleaned up (delivery subject removed from tail):
+	// "$JS.ACK.$MQTT_msgs.someid.1.1.1.1620086713306484000.0"
+	// So we will use that knowledge so we don't have to make any cleaning here.
+	routedReply = routedReply[:gwSubjectOffset+len(reply)]
 	// We need to make a copy so that we don't reference the underlying
 	// read buffer.
-	ms := string(reply)
+	ms := string(routedReply)
 	grm := &gwReplyMap{ms: ms, exp: time.Now().Add(ttl).UnixNano()}
 	// If we are here with the same key but different mapped replies
 	// (say $GNR._.A.srv1.bar and then $GNR._.B.srv2.bar), we need to
 	// store it otherwise we would take the risk of the reply not
 	// making it back.
-	rm[ms[gwSubjectOffset:]] = grm
-	if we {
-		atomic.StoreInt32(&c.cgwrt, 1)
-		s.gwrm.m.Store(c, nil)
+	g.mapping[ms[gwSubjectOffset:]] = grm
+	if wasEmpty {
+		atomic.StoreInt32(&g.check, 1)
+		s.gwrm.m.Store(g, l)
 		if atomic.CompareAndSwapInt32(&s.gwrm.w, 0, 1) {
 			select {
 			case s.gwrm.ch <- ttl:
@@ -3077,19 +3121,20 @@ func (s *Server) startGWReplyMapExpiration() {
 				}
 				now := time.Now().UnixNano()
 				mapEmpty := true
-				s.gwrm.m.Range(func(k, _ interface{}) bool {
-					c := k.(*client)
-					c.mu.Lock()
-					for k, grm := range c.gwrm {
+				s.gwrm.m.Range(func(k, v interface{}) bool {
+					g := k.(*gwReplyMapping)
+					l := v.(sync.Locker)
+					l.Lock()
+					for k, grm := range g.mapping {
 						if grm.exp <= now {
-							delete(c.gwrm, k)
-							if len(c.gwrm) == 0 {
-								atomic.StoreInt32(&c.cgwrt, 0)
-								s.gwrm.m.Delete(c)
+							delete(g.mapping, k)
+							if len(g.mapping) == 0 {
+								atomic.StoreInt32(&g.check, 0)
+								s.gwrm.m.Delete(g)
 							}
 						}
 					}
-					c.mu.Unlock()
+					l.Unlock()
 					mapEmpty = false
 					return true
 				})
diff --git a/server/jetstream.go b/server/jetstream.go
index db25b1b3..b0dcdff2 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -1317,33 +1317,14 @@ func (jsa *jsAccount) checkLimits(config *StreamConfig) error {
 func (jsa *jsAccount) checkBytesLimits(addBytes int64, storage StorageType) error {
 	switch storage {
 	case MemoryStorage:
-		// Account limits defined.
-		if jsa.limits.MaxMemory > 0 {
-			if jsa.memReserved+addBytes > jsa.limits.MaxMemory {
-				return ErrMemoryResourcesExceeded
-			}
-		} else {
-			// Account is unlimited, check if this server can handle request.
-			js := jsa.js
-			if js.memReserved+addBytes > js.config.MaxMemory {
-				return ErrMemoryResourcesExceeded
-			}
+		if jsa.memReserved+addBytes > jsa.limits.MaxMemory {
+			return fmt.Errorf("insufficient memory resources available")
 		}
 	case FileStorage:
-		// Account limits defined.
-		if jsa.limits.MaxStore > 0 {
-			if jsa.storeReserved+addBytes > jsa.limits.MaxStore {
-				return ErrStorageResourcesExceeded
-			}
-		} else {
-			// Account is unlimited, check if this server can handle request.
-			js := jsa.js
-			if js.storeReserved+addBytes > js.config.MaxStore {
-				return ErrStorageResourcesExceeded
-			}
+		if jsa.storeReserved+addBytes > jsa.limits.MaxStore {
+			return fmt.Errorf("insufficient storage resources available")
 		}
 	}
-
 	return nil
 }
 
@@ -1440,12 +1421,11 @@ func (js *jetStream) sufficientResources(limits *JetStreamAccountLimits) error {
 	if limits == nil {
 		return nil
 	}
-
 	if js.memReserved+limits.MaxMemory > js.config.MaxMemory {
-		return ErrMemoryResourcesExceeded
+		return fmt.Errorf("insufficient memory resources available")
 	}
 	if js.storeReserved+limits.MaxStore > js.config.MaxStore {
-		return ErrStorageResourcesExceeded
+		return fmt.Errorf("insufficient storage resources available")
 	}
 	return nil
 }
diff --git a/server/jwt.go b/server/jwt.go
index 538af9b9..a41f79d7 100644
--- a/server/jwt.go
+++ b/server/jwt.go
@@ -17,6 +17,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"net"
+	"regexp"
 	"strings"
 	"time"
 
@@ -24,6 +25,8 @@ import (
 	"github.com/nats-io/nkeys"
 )
 
+var nscDecoratedRe = regexp.MustCompile(`\s*(?:(?:[-]{3,}[^\n]*[-]{3,}\n)(.+)(?:\n\s*[-]{3,}[^\n]*[-]{3,}[\n]*))`)
+
 // All JWTs once encoded start with this
 const jwtPrefix = "eyJ"
 
@@ -45,9 +48,17 @@ func readOperatorJWT(jwtfile string) (string, *jwt.OperatorClaims, error) {
 	}
 	defer wipeSlice(contents)
 
-	theJWT, err := jwt.ParseDecoratedJWT(contents)
-	if err != nil {
-		return "", nil, err
+	var theJWT string
+	items := nscDecoratedRe.FindAllSubmatch(contents, -1)
+	if len(items) == 0 {
+		theJWT = string(contents)
+	} else {
+		// First result should be the JWT.
+		// We copy here so that if the file contained a seed file too we wipe appropriately.
+		raw := items[0][1]
+		tmp := make([]byte, len(raw))
+		copy(tmp, raw)
+		theJWT = string(tmp)
 	}
 	opc, err := jwt.DecodeOperatorClaims(theJWT)
 	if err != nil {
diff --git a/server/mqtt.go b/server/mqtt.go
index 5bf71835..4324ff96 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -168,6 +168,7 @@ var (
 	mqttNatsHeaderB   = []byte(mqttNatsHeader)
 	mqttSessJailDur   = mqttSessFlappingJailDur
 	mqttFlapCleanItvl = mqttSessFlappingCleanupInterval
+	mqttJSAPITimeout  = 4 * time.Second
 )
 
 var (
@@ -175,6 +176,20 @@ var (
 	errMQTTTopicFilterCannotBeEmpty = errors.New("topic filter cannot be empty")
 	errMQTTMalformedVarInt          = errors.New("malformed variable int")
 	errMQTTSecondConnectPacket      = errors.New("received a second CONNECT packet")
+	errMQTTServerNameMustBeSet      = errors.New("mqtt requires server name to be explicitly set")
+	errMQTTUserMixWithUsersNKeys    = errors.New("mqtt authentication username not compatible with presence of users/nkeys")
+	errMQTTTokenMixWIthUsersNKeys   = errors.New("mqtt authentication token not compatible with presence of users/nkeys")
+	errMQTTAckWaitMustBePositive    = errors.New("ack wait must be a positive value")
+	errMQTTStandaloneNeedsJetStream = errors.New("mqtt requires JetStream to be enabled if running in standalone mode")
+	errMQTTConnFlagReserved         = errors.New("connect flags reserved bit not set to 0")
+	errMQTTWillAndRetainFlag        = errors.New("if Will flag is set to 0, Will Retain flag must be 0 too")
+	errMQTTPasswordFlagAndNoUser    = errors.New("password flag set but username flag is not")
+	errMQTTCIDEmptyNeedsCleanFlag   = errors.New("when client ID is empty, clean session flag must be set to 1")
+	errMQTTEmptyWillTopic           = errors.New("empty Will topic not allowed")
+	errMQTTEmptyUsername            = errors.New("empty user name not allowed")
+	errMQTTTopicIsEmpty             = errors.New("topic cannot be empty")
+	errMQTTPacketIdentifierIsZero   = errors.New("packet identifier cannot be 0")
+	errMQTTUnsupportedCharacters    = errors.New("characters ' ' and '.' not supported for MQTT topics")
 )
 
 type srvMQTT struct {
@@ -208,10 +223,10 @@ type mqttJSA struct {
 	mu      sync.Mutex
 	id      string
 	c       *client
-	rid     int64
 	sendq   chan *mqttJSPubMsg
 	rplyr   string
 	replies sync.Map
+	nuid    *nuid.NUID
 	quitCh  chan struct{}
 }
 
@@ -507,6 +522,11 @@ 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_ {
+		return errMQTTServerNameMustBeSet
+	}
 	// If there is a NoAuthUser, we need to have Users defined and
 	// the user to be present.
 	if mo.NoAuthUser != _EMPTY_ {
@@ -517,18 +537,22 @@ func validateMQTTOptions(o *Options) error {
 	// Token/Username not possible if there are users/nkeys
 	if len(o.Users) > 0 || len(o.Nkeys) > 0 {
 		if mo.Username != _EMPTY_ {
-			return fmt.Errorf("mqtt authentication username not compatible with presence of users/nkeys")
+			return errMQTTUserMixWithUsersNKeys
 		}
 		if mo.Token != _EMPTY_ {
-			return fmt.Errorf("mqtt authentication token not compatible with presence of users/nkeys")
+			return errMQTTTokenMixWIthUsersNKeys
 		}
 	}
 	if mo.AckWait < 0 {
-		return fmt.Errorf("ack wait must be a positive value")
+		return errMQTTAckWaitMustBePositive
 	}
-	// If standalone and there is no JS enabled, then it won't work...
-	if o.Cluster.Port == 0 && o.Gateway.Port == 0 && !o.JetStream {
-		return fmt.Errorf("mqtt requires JetStream to be enabled if running in standalone mode")
+	// If strictly standalone and there is no JS enabled, then it won't work...
+	// For leafnodes, we could either have remote(s) and it would be ok, or no
+	// remote but accept from a remote side that has "hub" property set, which
+	// then would ok too. So we fail only if we have no leafnode config at all.
+	if !o.JetStream && o.Cluster.Port == 0 && o.Gateway.Port == 0 &&
+		o.LeafNode.Port == 0 && len(o.LeafNode.Remotes) == 0 {
+		return errMQTTStandaloneNeedsJetStream
 	}
 	return nil
 }
@@ -849,9 +873,6 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 	c := s.createInternalAccountClient()
 	c.acc = acc
 
-	sc := s.createInternalAccountClient()
-	sc.acc = acc
-
 	id := string(getHash(s.Name()))
 	replicas := s.mqttDetermineReplicas()
 	s.Noticef("Creating MQTT streams/consumers with replicas %v for account %q", replicas, accName)
@@ -866,6 +887,7 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 			c:      c,
 			rplyr:  mqttJSARepliesPrefix + id + ".",
 			sendq:  make(chan *mqttJSPubMsg, 8192),
+			nuid:   nuid.New(),
 			quitCh: quitCh,
 		},
 	}
@@ -884,6 +906,13 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 		close(closeCh)
 	}()
 
+	// We create all subscriptions before starting the go routine that will do
+	// sends otherwise we could get races.
+	// Note that using two different clients (one for the subs, one for the
+	// sends) would cause other issues such as registration of recent subs in
+	// the "sub" client would be invisible to the check for GW routed replies
+	// (shouldMapReplyForGatewaySend) since the client there would be the "sender".
+
 	jsa := &as.jsa
 	sid := int64(1)
 	// This is a subscription that will process all JS API replies. We could split to
@@ -901,10 +930,25 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 		return nil, err
 	}
 
+	// We create the subscription on "$MQTT.sub.<nuid>" to limit the subjects
+	// that a user would allow permissions on.
+	rmsubj := mqttSubPrefix + nuid.Next()
+	if err := as.createSubscription(rmsubj, as.processRetainedMsg, &sid, &subs); err != nil {
+		return nil, err
+	}
+
+	// Create a subscription to be notified of retained messages delete requests.
+	rmdelsubj := mqttJSARepliesPrefix + "*." + mqttJSARetainedMsgDel
+	if err := as.createSubscription(rmdelsubj, as.processRetainedMsgDel, &sid, &subs); err != nil {
+		return nil, err
+	}
+
+	// No more creation of subscriptions past this point otherwise RACEs may happen.
+
 	// Start the go routine that will send JS API requests.
 	s.startGoRoutine(func() {
 		defer s.grWG.Done()
-		as.sendJSAPIrequests(s, sc, accName, closeCh)
+		as.sendJSAPIrequests(s, c, accName, closeCh)
 	})
 
 	// Create the stream for the messages.
@@ -949,32 +993,14 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 			as.rrmDoneCh = rmDoneCh
 		}
 	}
-	// We create the subscription on "$MQTT.sub.<nuid>" to limit the subjects
-	// that a user would allow permissions on.
-	rmsubj := mqttSubPrefix + nuid.Next()
-	if err := as.createSubscription(rmsubj, as.processRetainedMsg, &sid, &subs); err != nil {
-		return nil, err
-	}
-
-	// Create a subscription to be notified of retained messages delete requests.
-	rmdelsubj := mqttJSARepliesPrefix + "*." + mqttJSARetainedMsgDel
-	if err := as.createSubscription(rmdelsubj, as.processRetainedMsgDel, &sid, &subs); err != nil {
-		return nil, err
-	}
 
 	// Using ephemeral consumer is too risky because if this server were to be
 	// disconnected from the rest for few seconds, then the leader would remove
 	// the consumer, so even after a reconnect, we would not longer receive
-	// retained messages. So delete any existing durable that we have for that
-	// and recreate here. In non cluster mode, we will use the retained messages
-	// stream name as the name, since it will be unique for this account and
-	// we will know what is the name on restart.
-	rmDurName := mqttRetainedMsgsStreamName
-	// In cluster mode, we will add our id (server name hash) since there may
-	// be many of those durables (1 per node).
-	if s.JetStreamIsClustered() {
-		rmDurName += "_" + jsa.id
-	}
+	// retained messages. Delete any existing durable that we have for that
+	// and recreate here.
+	// The name for the durable is $MQTT_rmsgs_<server name hash> (which is jsa.id)
+	rmDurName := mqttRetainedMsgsStreamName + "_" + jsa.id
 	resp, err := jsa.deleteConsumer(mqttRetainedMsgsStreamName, rmDurName)
 	// If error other than "not found" then fail, otherwise proceed with creating
 	// the durable consumer.
@@ -998,7 +1024,7 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 	if lastSeq > 0 {
 		select {
 		case <-rmDoneCh:
-		case <-time.After(2 * time.Second):
+		case <-time.After(mqttJSAPITimeout):
 			s.Warnf("Timing out waiting to load %v retained messages", st.Msgs)
 		case <-quitCh:
 			return nil, ErrServerNotRunning
@@ -1043,13 +1069,15 @@ func (s *Server) mqttDetermineReplicas() int {
 //////////////////////////////////////////////////////////////////////////////
 
 func (jsa *mqttJSA) newRequest(kind, subject string, hdr int, msg []byte) (interface{}, error) {
-	return jsa.newRequestEx(kind, subject, hdr, msg, 2*time.Second)
+	return jsa.newRequestEx(kind, subject, hdr, msg, mqttJSAPITimeout)
 }
 
 func (jsa *mqttJSA) newRequestEx(kind, subject string, hdr int, msg []byte, timeout time.Duration) (interface{}, error) {
 	jsa.mu.Lock()
-	jsa.rid++
-	reply := jsa.rplyr + kind + "." + strconv.FormatInt(jsa.rid, 10)
+	// Either we use nuid.Next() which uses a global lock, or our own nuid object, but
+	// then it needs to be "write" protected. This approach will reduce across account
+	// contention since we won't use the global nuid's lock.
+	reply := jsa.rplyr + kind + "." + jsa.nuid.Next()
 	jsa.mu.Unlock()
 
 	ch := make(chan interface{}, 1)
@@ -1063,13 +1091,18 @@ func (jsa *mqttJSA) newRequestEx(kind, subject string, hdr int, msg []byte, time
 	}
 
 	var i interface{}
+	// We don't want to use time.After() which causes memory growth because the timer
+	// can't be stopped and will need to expire to then be garbage collected.
+	t := time.NewTimer(timeout)
 	select {
 	case i = <-ch:
+		// Ensure we stop the timer so it can be quickly garbage collected.
+		t.Stop()
 	case <-jsa.quitCh:
 		return nil, ErrServerNotRunning
-	case <-time.After(timeout):
+	case <-t.C:
 		jsa.replies.Delete(reply)
-		return nil, fmt.Errorf("timeout for request type %q on %q", kind, subject)
+		return nil, fmt.Errorf("timeout for request type %q on %q (reply=%q)", kind, subject, reply)
 	}
 	return i, nil
 }
@@ -1409,7 +1442,10 @@ func (as *mqttAccountSessionManager) createSubscription(subject string, cb msgHa
 //
 // No lock held on entry.
 func (as *mqttAccountSessionManager) sendJSAPIrequests(s *Server, c *client, accName string, closeCh chan struct{}) {
-	cluster := s.cachedClusterName()
+	var cluster string
+	if s.JetStreamEnabled() {
+		cluster = s.cachedClusterName()
+	}
 	as.mu.RLock()
 	sendq := as.jsa.sendq
 	quitCh := as.jsa.quitCh
@@ -2170,7 +2206,7 @@ func (c *client) mqttParseConnect(r *mqttReader, pl int) (byte, *mqttConnectProt
 
 	// Spec [MQTT-3.1.2-3]
 	if cp.flags&mqttConnFlagReserved != 0 {
-		return 0, nil, fmt.Errorf("connect flags reserved bit not set to 0")
+		return 0, nil, errMQTTConnFlagReserved
 	}
 
 	var hasWill bool
@@ -2184,7 +2220,7 @@ func (c *client) mqttParseConnect(r *mqttReader, pl int) (byte, *mqttConnectProt
 		}
 		// Spec [MQTT-3.1.2-15]
 		if wretain {
-			return 0, nil, fmt.Errorf("if Will flag is set to 0, Will Retain flag must be 0 too")
+			return 0, nil, errMQTTWillAndRetainFlag
 		}
 	} else {
 		// Spec [MQTT-3.1.2-14]
@@ -2200,7 +2236,7 @@ func (c *client) mqttParseConnect(r *mqttReader, pl int) (byte, *mqttConnectProt
 	hasPassword := cp.flags&mqttConnFlagPasswordFlag != 0
 	// Spec [MQTT-3.1.2-22]
 	if !hasUser && hasPassword {
-		return 0, nil, fmt.Errorf("password flag set but username flag is not")
+		return 0, nil, errMQTTPasswordFlagAndNoUser
 	}
 
 	// Keep alive
@@ -2225,7 +2261,7 @@ func (c *client) mqttParseConnect(r *mqttReader, pl int) (byte, *mqttConnectProt
 	// Spec [MQTT-3.1.3-7]
 	if cp.clientID == _EMPTY_ {
 		if cp.flags&mqttConnFlagCleanSession == 0 {
-			return mqttConnAckRCIdentifierRejected, nil, fmt.Errorf("when client ID is empty, clean session flag must be set to 1")
+			return mqttConnAckRCIdentifierRejected, nil, errMQTTCIDEmptyNeedsCleanFlag
 		}
 		// Spec [MQTT-3.1.3-6]
 		cp.clientID = nuid.Next()
@@ -2248,10 +2284,10 @@ func (c *client) mqttParseConnect(r *mqttReader, pl int) (byte, *mqttConnectProt
 			return 0, nil, err
 		}
 		if len(topic) == 0 {
-			return 0, nil, fmt.Errorf("empty Will topic not allowed")
+			return 0, nil, errMQTTEmptyWillTopic
 		}
 		if !utf8.Valid(topic) {
-			return 0, nil, fmt.Errorf("invalide utf8 for Will topic %q", topic)
+			return 0, nil, fmt.Errorf("invalid utf8 for Will topic %q", topic)
 		}
 		cp.will.topic = topic
 		// Convert MQTT topic to NATS subject
@@ -2272,7 +2308,7 @@ func (c *client) mqttParseConnect(r *mqttReader, pl int) (byte, *mqttConnectProt
 			return 0, nil, err
 		}
 		if c.opts.Username == _EMPTY_ {
-			return mqttConnAckRCBadUserOrPassword, nil, fmt.Errorf("empty user name not allowed")
+			return mqttConnAckRCBadUserOrPassword, nil, errMQTTEmptyUsername
 		}
 		// Spec [MQTT-3.1.3-11]
 		if !utf8.ValidString(c.opts.Username) {
@@ -2561,7 +2597,7 @@ func (c *client) mqttParsePub(r *mqttReader, pl int, pp *mqttPublish) error {
 		return err
 	}
 	if len(pp.topic) == 0 {
-		return fmt.Errorf("topic cannot be empty")
+		return errMQTTTopicIsEmpty
 	}
 	// Convert the topic to a NATS subject. This call will also check that
 	// there is no MQTT wildcards (Spec [MQTT-3.3.2-2] and [MQTT-4.7.1-1])
@@ -2667,7 +2703,7 @@ func (c *client) mqttHandlePubRetain() {
 		rm := &mqttRetainedMsg{
 			Origin:  asm.jsa.id,
 			Subject: key,
-			Topic:   string(copyBytes(pp.topic)),
+			Topic:   string(pp.topic),
 			Msg:     copyBytes(pp.msg),
 			Flags:   pp.flags,
 			Source:  c.opts.Username,
@@ -2839,7 +2875,7 @@ func mqttParsePubAck(r *mqttReader, pl int) (uint16, error) {
 		return 0, err
 	}
 	if pi == 0 {
-		return 0, fmt.Errorf("packet identifier cannot be 0")
+		return 0, errMQTTPacketIdentifierIsZero
 	}
 	return pi, nil
 }
@@ -3002,7 +3038,7 @@ func mqttSubscribeTrace(pi uint16, filters []*mqttFilter) string {
 // message and this is the callback for a QoS1 subscription because in
 // that case, it will be handled by the other callback. This avoid getting
 // duplicate deliveries.
-func mqttDeliverMsgCbQos0(sub *subscription, pc *client, subject, reply string, rmsg []byte) {
+func mqttDeliverMsgCbQos0(sub *subscription, pc *client, subject, _ string, rmsg []byte) {
 	if pc.kind == JETSTREAM {
 		return
 	}
@@ -3539,7 +3575,7 @@ func mqttToNATSSubjectConversion(mt []byte, wcOk bool) ([]byte, error) {
 			}
 		case btsep, ' ':
 			// As of now, we cannot support '.' or ' ' in the MQTT topic/filter.
-			return nil, fmt.Errorf("characters ' ' and '.' not supported for MQTT topics")
+			return nil, errMQTTUnsupportedCharacters
 		case mqttSingleLevelWC, mqttMultiLevelWC:
 			if !wcOk {
 				// Spec [MQTT-3.3.2-2] and [MQTT-4.7.1-1]
diff --git a/vendor/github.com/nats-io/jwt/v2/creds_utils.go b/vendor/github.com/nats-io/jwt/v2/creds_utils.go
index c532c887..93ba6d18 100644
--- a/vendor/github.com/nats-io/jwt/v2/creds_utils.go
+++ b/vendor/github.com/nats-io/jwt/v2/creds_utils.go
@@ -97,7 +97,7 @@ NKEYs are sensitive and should be treated as secrets.
 	return w.Bytes(), nil
 }
 
-var userConfigRE = regexp.MustCompile(`\s*(?:(?:[-]{3,}.*[-]{3,}\r?\n)([\w\-.=]+)(?:\r?\n[-]{3,}.*[-]{3,}(\r?\n|\z)))`)
+var userConfigRE = regexp.MustCompile(`\s*(?:(?:[-]{3,}.*[-]{3,}\r?\n)([\w\-.=]+)(?:\r?\n[-]{3,}.*[-]{3,}\r?\n))`)
 
 // An user config file looks like this:
 //  -----BEGIN NATS USER JWT-----
diff --git a/vendor/github.com/nats-io/jwt/v2/header.go b/vendor/github.com/nats-io/jwt/v2/header.go
index 2fd587ca..a1063467 100644
--- a/vendor/github.com/nats-io/jwt/v2/header.go
+++ b/vendor/github.com/nats-io/jwt/v2/header.go
@@ -23,7 +23,7 @@ import (
 
 const (
 	// Version is semantic version.
-	Version = "2.0.2"
+	Version = "2.0.1"
 
 	// TokenTypeJwt is the JWT token type supported JWT tokens
 	// encoded and decoded by this library
diff --git a/vendor/modules.txt b/vendor/modules.txt
index fa549e09..f4dbbff6 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -6,7 +6,7 @@ github.com/klauspost/compress/s2
 # github.com/minio/highwayhash v1.0.1
 ## explicit
 github.com/minio/highwayhash
-# github.com/nats-io/jwt/v2 v2.0.2
+# github.com/nats-io/jwt/v2 v2.0.1
 ## explicit
 github.com/nats-io/jwt/v2
 # github.com/nats-io/nats.go v1.10.1-0.20210419223411-20527524c393


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 8dabc318..93ea9e5a 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -566,15 +566,12 @@ func TestFileStoreBytesLimit(t *testing.T) {
 }
 
 func TestFileStoreAgeLimit(t *testing.T) {
-	maxAge := 100 * time.Millisecond
+	maxAge := 10 * time.Millisecond
 
 	storeDir := createDir(t, JetStreamStoreDir)
 	defer removeDir(t, storeDir)
 
-	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: storeDir, BlockSize: 256},
-		StreamConfig{Name: "zzz", Storage: FileStorage, MaxAge: maxAge},
-	)
+	fs, err := newFileStore(FileStoreConfig{StoreDir: storeDir}, StreamConfig{Name: "zzz", Storage: FileStorage, MaxAge: maxAge})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -582,7 +579,7 @@ func TestFileStoreAgeLimit(t *testing.T) {
 
 	// Store some messages. Does not really matter how many.
 	subj, msg := "foo", []byte("Hello World")
-	toStore := 500
+	toStore := 100
 	for i := 0; i < toStore; i++ {
 		fs.StoreMsg(subj, nil, msg)
 	}
@@ -592,7 +589,7 @@ func TestFileStoreAgeLimit(t *testing.T) {
 	}
 	checkExpired := func(t *testing.T) {
 		t.Helper()
-		checkFor(t, 5*time.Second, maxAge, func() error {
+		checkFor(t, time.Second, maxAge, func() error {
 			state = fs.State()
 			if state.Msgs != 0 {
 				return fmt.Errorf("Expected no msgs, got %d", state.Msgs)
@@ -605,7 +602,6 @@ func TestFileStoreAgeLimit(t *testing.T) {
 	}
 	// Let them expire
 	checkExpired(t)
-
 	// Now add some more and make sure that timer will fire again.
 	for i := 0; i < toStore; i++ {
 		fs.StoreMsg(subj, nil, msg)
@@ -2833,52 +2829,3 @@ func TestFileStoreStreamPurgeAndDirtyRestartBug(t *testing.T) {
 		t.Fatalf("Unexpected state: %+v", state)
 	}
 }
-
-// rip
-func TestFileStoreStreamFailToRollBug(t *testing.T) {
-	storeDir := createDir(t, JetStreamStoreDir)
-	defer removeDir(t, storeDir)
-
-	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: storeDir, BlockSize: 512},
-		StreamConfig{Name: "zzz", Storage: FileStorage, MaxBytes: 300},
-	)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	defer fs.Stop()
-
-	// Make sure we properly roll underlying blocks.
-	n, msg := 200, bytes.Repeat([]byte("ABC"), 33) // ~100bytes
-	for i := 0; i < n; i++ {
-		if _, _, err := fs.StoreMsg("zzz", nil, msg); err != nil {
-			t.Fatalf("Unexpected error: %v", err)
-		}
-	}
-
-	// Grab some info for introspection.
-	fs.mu.RLock()
-	numBlks := len(fs.blks)
-	var index uint64
-	var blkSize int64
-	if numBlks > 0 {
-		mb := fs.blks[0]
-		mb.mu.RLock()
-		index = mb.index
-		if fi, _ := os.Stat(mb.mfn); fi != nil {
-			blkSize = fi.Size()
-		}
-		mb.mu.RUnlock()
-	}
-	fs.mu.RUnlock()
-
-	if numBlks != 1 {
-		t.Fatalf("Expected only one block, got %d", numBlks)
-	}
-	if index < 60 {
-		t.Fatalf("Expected a block index > 60, got %d", index)
-	}
-	if blkSize > 512 {
-		t.Fatalf("Expected block to be <= 512, got %d", blkSize)
-	}
-}
diff --git a/server/gateway_test.go b/server/gateway_test.go
index 7f153998..eb9e2a5c 100644
--- a/server/gateway_test.go
+++ b/server/gateway_test.go
@@ -5975,10 +5975,10 @@ func TestGatewayReplyMapTracking(t *testing.T) {
 	check := func(t *testing.T, expectedIndicator int32, expectLenMap int, expectedSrvMapEmpty bool) {
 		t.Helper()
 		bc.mu.Lock()
-		mapIndicator := atomic.LoadInt32(&bc.cgwrt)
+		mapIndicator := atomic.LoadInt32(&bc.gwReplyMapping.check)
 		var lenMap int
-		if bc.gwrm != nil {
-			lenMap = len(bc.gwrm)
+		if bc.gwReplyMapping.mapping != nil {
+			lenMap = len(bc.gwReplyMapping.mapping)
 		}
 		bc.mu.Unlock()
 		if mapIndicator != expectedIndicator {
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 5158acff..280daf41 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -114,36 +114,6 @@ func TestJetStreamClusterAccountInfo(t *testing.T) {
 	}
 }
 
-func TestJetStreamClusterStreamLimitWithAccountDefaults(t *testing.T) {
-	// 2MB memory, 8MB disk
-	c := createJetStreamClusterWithTemplate(t, jsClusterLimitsTempl, "R3L", 3)
-	defer c.shutdown()
-
-	// Client based API
-	s := c.randomNonLeader()
-	nc, js := jsClientConnect(t, s)
-	defer nc.Close()
-
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name:     "TEST",
-		Subjects: []string{"foo", "bar"},
-		Replicas: 2,
-		MaxBytes: 4 * 1024 * 1024,
-	})
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-
-	_, err = js.AddStream(&nats.StreamConfig{
-		Name:     "TEST2",
-		Replicas: 2,
-		MaxBytes: 15 * 1024 * 1024,
-	})
-	if err == nil || !strings.Contains(err.Error(), "insufficient storage") {
-		t.Fatalf("Expected %v but got %v", ErrStorageResourcesExceeded, err)
-	}
-}
-
 func TestJetStreamClusterSingleReplicaStreams(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R1S", 3)
 	defer c.shutdown()
diff --git a/server/mqtt_test.go b/server/mqtt_test.go
index 3077285e..19b8c48f 100644
--- a/server/mqtt_test.go
+++ b/server/mqtt_test.go
@@ -199,6 +199,7 @@ func TestMQTTWriter(t *testing.T) {
 
 func testMQTTDefaultOptions() *Options {
 	o := DefaultOptions()
+	o.ServerName = nuid.Next()
 	o.Cluster.Port = 0
 	o.Gateway.Name = ""
 	o.Gateway.Port = 0
@@ -224,6 +225,7 @@ func testMQTTRunServer(t testing.TB, o *Options) *Server {
 	s.SetLogger(l, true, true)
 	go s.Start()
 	if err := s.readyForConnections(3 * time.Second); err != nil {
+		testMQTTShutdownServer(s)
 		t.Fatal(err)
 	}
 	return s
@@ -261,8 +263,25 @@ func testMQTTDefaultTLSOptions(t *testing.T, verify bool) *Options {
 	return o
 }
 
+func TestMQTTServerNameRequired(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		mqtt {
+			port: -1
+		}
+	`))
+	defer removeFile(t, conf)
+	o, err := ProcessConfigFile(conf)
+	if err != nil {
+		t.Fatalf("Error processing config file: %v", err)
+	}
+	if _, err := NewServer(o); err == nil || err.Error() != errMQTTServerNameMustBeSet.Error() {
+		t.Fatalf("Expected error about requiring server name to be set, got %v", err)
+	}
+}
+
 func TestMQTTStandaloneRequiresJetStream(t *testing.T) {
 	conf := createConfFile(t, []byte(`
+		server_name: mqtt
 		mqtt {
 			port: -1
 			tls {
@@ -276,7 +295,7 @@ func TestMQTTStandaloneRequiresJetStream(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Error processing config file: %v", err)
 	}
-	if _, err := NewServer(o); err == nil || !strings.Contains(err.Error(), "standalone") {
+	if _, err := NewServer(o); err == nil || err.Error() != errMQTTStandaloneNeedsJetStream.Error() {
 		t.Fatalf("Expected error about requiring JetStream in standalone mode, got %v", err)
 	}
 }
@@ -284,6 +303,7 @@ func TestMQTTStandaloneRequiresJetStream(t *testing.T) {
 func TestMQTTConfig(t *testing.T) {
 	conf := createConfFile(t, []byte(`
 		jetstream: enabled
+		server_name: mqtt
 		mqtt {
 			port: -1
 			tls {
@@ -306,33 +326,33 @@ func TestMQTTValidateOptions(t *testing.T) {
 	for _, test := range []struct {
 		name    string
 		getOpts func() *Options
-		err     string
+		err     error
 	}{
-		{"mqtt disabled", func() *Options { return nmqtto.Clone() }, ""},
+		{"mqtt disabled", func() *Options { return nmqtto.Clone() }, nil},
 		{"mqtt username not allowed if users specified", func() *Options {
 			o := mqtto.Clone()
 			o.Users = []*User{{Username: "abc", Password: "pwd"}}
 			o.MQTT.Username = "b"
 			o.MQTT.Password = "pwd"
 			return o
-		}, "mqtt authentication username not compatible with presence of users/nkeys"},
+		}, errMQTTUserMixWithUsersNKeys},
 		{"mqtt token not allowed if users specified", func() *Options {
 			o := mqtto.Clone()
 			o.Nkeys = []*NkeyUser{{Nkey: "abc"}}
 			o.MQTT.Token = "mytoken"
 			return o
-		}, "mqtt authentication token not compatible with presence of users/nkeys"},
+		}, errMQTTTokenMixWIthUsersNKeys},
 		{"ack wait should be >=0", func() *Options {
 			o := mqtto.Clone()
 			o.MQTT.AckWait = -10 * time.Second
 			return o
-		}, "ack wait must be a positive value"},
+		}, errMQTTAckWaitMustBePositive},
 	} {
 		t.Run(test.name, func(t *testing.T) {
 			err := validateMQTTOptions(test.getOpts())
-			if test.err == "" && err != nil {
+			if test.err == nil && err != nil {
 				t.Fatalf("Unexpected error: %v", err)
-			} else if test.err != "" && (err == nil || !strings.Contains(err.Error(), test.err)) {
+			} else if test.err != nil && (err == nil || err.Error() != test.err.Error()) {
 				t.Fatalf("Expected error to contain %q, got %v", test.err, err)
 			}
 		})
@@ -1467,22 +1487,22 @@ func TestMQTTParseConnect(t *testing.T) {
 		{"error on protocol level", []byte{0, 4, 'M', 'Q', 'T', 'T'}, 6, eofr, "protocol level"},
 		{"unacceptable protocol version", []byte{0, 4, 'M', 'Q', 'T', 'T', 10}, 7, nil, "unacceptable protocol version"},
 		{"error on flags", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel}, 7, eofr, "flags"},
-		{"reserved flag", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, 1}, 8, nil, "connect flags reserved bit not set to 0"},
+		{"reserved flag", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, 1}, 8, nil, errMQTTConnFlagReserved.Error()},
 		{"will qos without will flag", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, 1 << 3}, 8, nil, "if Will flag is set to 0, Will QoS must be 0 too"},
-		{"will retain without will flag", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, 1 << 5}, 8, nil, "if Will flag is set to 0, Will Retain flag must be 0 too"},
+		{"will retain without will flag", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, 1 << 5}, 8, nil, errMQTTWillAndRetainFlag.Error()},
 		{"will qos", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, 3<<3 | 1<<2}, 8, nil, "if Will flag is set to 1, Will QoS can be 0, 1 or 2"},
-		{"no user but password", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, mqttConnFlagPasswordFlag}, 8, nil, "password flag set but username flag is not"},
+		{"no user but password", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, mqttConnFlagPasswordFlag}, 8, nil, errMQTTPasswordFlagAndNoUser.Error()},
 		{"missing keep alive", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, 0}, 8, nil, "keep alive"},
 		{"missing client ID", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, 0, 0, 1}, 10, nil, "client ID"},
-		{"empty client ID", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, 0, 0, 1, 0, 0}, 12, nil, "when client ID is empty, clean session flag must be set to 1"},
+		{"empty client ID", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, 0, 0, 1, 0, 0}, 12, nil, errMQTTCIDEmptyNeedsCleanFlag.Error()},
 		{"invalid utf8 client ID", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, 0, 0, 1, 0, 1, 241}, 13, nil, "invalid utf8 for client ID"},
 		{"missing will topic", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, mqttConnFlagWillFlag | mqttConnFlagCleanSession, 0, 0, 0, 0}, 12, nil, "Will topic"},
-		{"empty will topic", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, mqttConnFlagWillFlag | mqttConnFlagCleanSession, 0, 0, 0, 0, 0, 0}, 14, nil, "empty Will topic not allowed"},
-		{"invalid utf8 will topic", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, mqttConnFlagWillFlag | mqttConnFlagCleanSession, 0, 0, 0, 0, 0, 1, 241}, 15, nil, "invalide utf8 for Will topic"},
+		{"empty will topic", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, mqttConnFlagWillFlag | mqttConnFlagCleanSession, 0, 0, 0, 0, 0, 0}, 14, nil, errMQTTEmptyWillTopic.Error()},
+		{"invalid utf8 will topic", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, mqttConnFlagWillFlag | mqttConnFlagCleanSession, 0, 0, 0, 0, 0, 1, 241}, 15, nil, "invalid utf8 for Will topic"},
 		{"invalid wildcard will topic", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, mqttConnFlagWillFlag | mqttConnFlagCleanSession, 0, 0, 0, 0, 0, 1, '#'}, 15, nil, "wildcards not allowed"},
 		{"error on will message", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, mqttConnFlagWillFlag | mqttConnFlagCleanSession, 0, 0, 0, 0, 0, 1, 'a', 0, 3}, 17, eofr, "Will message"},
 		{"error on username", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, mqttConnFlagUsernameFlag | mqttConnFlagCleanSession, 0, 0, 0, 0}, 12, eofr, "user name"},
-		{"empty username", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, mqttConnFlagUsernameFlag | mqttConnFlagCleanSession, 0, 0, 0, 0, 0, 0}, 14, nil, "empty user name not allowed"},
+		{"empty username", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, mqttConnFlagUsernameFlag | mqttConnFlagCleanSession, 0, 0, 0, 0, 0, 0}, 14, nil, errMQTTEmptyUsername.Error()},
 		{"invalid utf8 username", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, mqttConnFlagUsernameFlag | mqttConnFlagCleanSession, 0, 0, 0, 0, 0, 1, 241}, 15, nil, "invalid utf8 for user name"},
 		{"error on password", []byte{0, 4, 'M', 'Q', 'T', 'T', mqttProtoLevel, mqttConnFlagUsernameFlag | mqttConnFlagPasswordFlag | mqttConnFlagCleanSession, 0, 0, 0, 0, 0, 1, 'a'}, 15, eofr, "password"},
 	} {
@@ -1989,10 +2009,10 @@ func TestMQTTParsePub(t *testing.T) {
 		{"qos not supported", 0x4, nil, 0, nil, "not supported"},
 		{"packet in buffer error", 0, nil, 10, eofr, "error ensuring protocol is loaded"},
 		{"error on topic", 0, []byte{0, 3, 'f', 'o'}, 4, eofr, "topic"},
-		{"empty topic", 0, []byte{0, 0}, 2, nil, "topic cannot be empty"},
+		{"empty topic", 0, []byte{0, 0}, 2, nil, errMQTTTopicIsEmpty.Error()},
 		{"wildcards topic", 0, []byte{0, 1, '#'}, 3, nil, "wildcards not allowed"},
 		{"error on packet identifier", mqttPubQos1, []byte{0, 3, 'f', 'o', 'o'}, 5, eofr, "packet identifier"},
-		{"invalid packet identifier", mqttPubQos1, []byte{0, 3, 'f', 'o', 'o', 0, 0}, 7, nil, "packet identifier cannot be 0"},
+		{"invalid packet identifier", mqttPubQos1, []byte{0, 3, 'f', 'o', 'o', 0, 0}, 7, nil, errMQTTPacketIdentifierIsZero.Error()},
 	} {
 		t.Run(test.name, func(t *testing.T) {
 			r := &mqttReader{reader: test.reader}
@@ -2018,7 +2038,7 @@ func TestMQTTParsePubAck(t *testing.T) {
 	}{
 		{"packet in buffer error", nil, 10, eofr, "error ensuring protocol is loaded"},
 		{"error reading packet identifier", []byte{0}, 1, eofr, "packet identifier"},
-		{"invalid packet identifier", []byte{0, 0}, 2, nil, "packet identifier cannot be 0"},
+		{"invalid packet identifier", []byte{0, 0}, 2, nil, errMQTTPacketIdentifierIsZero.Error()},
 	} {
 		t.Run(test.name, func(t *testing.T) {
 			r := &mqttReader{reader: test.reader}
@@ -2978,14 +2998,15 @@ func TestMQTTClusterReplicasCount(t *testing.T) {
 }
 
 func TestMQTTClusterPlacement(t *testing.T) {
-	c := createJetStreamClusterExplicit(t, "HUB", 3)
-	defer c.shutdown()
+	sc := createJetStreamSuperCluster(t, 3, 2)
+	defer sc.shutdown()
 
+	c := sc.randomCluster()
 	lnc := c.createLeafNodesWithStartPortAndMQTT("SPOKE", 3, 22111, `mqtt { listen: 127.0.0.1:-1 }`)
 	defer lnc.shutdown()
 
-	c.waitOnPeerCount(6)
-	c.waitOnLeader()
+	sc.waitOnPeerCount(9)
+	sc.waitOnLeader()
 
 	for i := 0; i < 10; i++ {
 		mc, rc := testMQTTConnect(t, &mqttConnInfo{cleanSess: true}, lnc.opts[i%3].MQTT.Host, lnc.opts[i%3].MQTT.Port)
@@ -3029,6 +3050,90 @@ func TestMQTTClusterPlacement(t *testing.T) {
 	}
 }
 
+func TestMQTTLeafnodeWithoutJSToClusterWithJS(t *testing.T) {
+	getClusterOpts := func(name string, i int) *Options {
+		o := testMQTTDefaultOptions()
+		o.ServerName = name
+		o.Cluster.Name = "hub"
+		o.Cluster.Host = "127.0.0.1"
+		o.Cluster.Port = 2790 + i
+		o.Routes = RoutesFromStr("nats://127.0.0.1:2791,nats://127.0.0.1:2792,nats://127.0.0.1:2793")
+		o.LeafNode.Host = "127.0.0.1"
+		o.LeafNode.Port = -1
+		return o
+	}
+	o1 := getClusterOpts("S1", 1)
+	s1 := testMQTTRunServer(t, o1)
+	defer testMQTTShutdownServer(s1)
+
+	o2 := getClusterOpts("S2", 2)
+	s2 := testMQTTRunServer(t, o2)
+	defer testMQTTShutdownServer(s2)
+
+	o3 := getClusterOpts("S3", 3)
+	s3 := testMQTTRunServer(t, o3)
+	defer testMQTTShutdownServer(s3)
+
+	cluster := []*Server{s1, s2, s3}
+	checkClusterFormed(t, cluster...)
+	checkFor(t, 10*time.Second, 50*time.Millisecond, func() error {
+		for _, s := range cluster {
+			if s.JetStreamIsLeader() {
+				return nil
+			}
+		}
+		return fmt.Errorf("no leader yet")
+	})
+
+	// Now define a leafnode that has mqtt enabled, but no JS. This should still work.
+	lno := testMQTTDefaultOptions()
+	// Make sure jetstream is not explicitly defined here.
+	lno.JetStream = false
+	// Use RoutesFromStr() to make an array of urls
+	urls := RoutesFromStr(fmt.Sprintf("nats://127.0.0.1:%d,nats://127.0.0.1:%d,nats://127.0.0.1:%d",
+		o1.LeafNode.Port, o2.LeafNode.Port, o3.LeafNode.Port))
+	lno.LeafNode.Remotes = []*RemoteLeafOpts{{URLs: urls}}
+	ln := RunServer(lno)
+	defer ln.Shutdown()
+
+	// Now connect to leafnode and subscribe
+	mc, rc := testMQTTConnect(t, &mqttConnInfo{clientID: "sub", cleanSess: true}, lno.MQTT.Host, lno.MQTT.Port)
+	defer mc.Close()
+	testMQTTCheckConnAck(t, rc, mqttConnAckRCConnectionAccepted, false)
+	testMQTTSub(t, 1, mc, rc, []*mqttFilter{{filter: "foo", qos: 1}}, []byte{1})
+	testMQTTFlush(t, mc, nil, rc)
+
+	connectAndPublish := func(o *Options) {
+		mp, rp := testMQTTConnect(t, &mqttConnInfo{cleanSess: true}, o.MQTT.Host, o.MQTT.Port)
+		defer mp.Close()
+		testMQTTCheckConnAck(t, rp, mqttConnAckRCConnectionAccepted, false)
+
+		testMQTTPublish(t, mp, rp, 1, false, false, "foo", 1, []byte("msg"))
+	}
+	// Connect a publisher from leafnode and publish, verify message is received.
+	connectAndPublish(lno)
+	testMQTTCheckPubMsg(t, mc, rc, "foo", mqttPubQos1, []byte("msg"))
+
+	// Connect from one server in the cluster check it works from there too.
+	connectAndPublish(o3)
+	testMQTTCheckPubMsg(t, mc, rc, "foo", mqttPubQos1, []byte("msg"))
+
+	// Connect from a server in the hub and subscribe
+	mc2, rc2 := testMQTTConnect(t, &mqttConnInfo{clientID: "sub2", cleanSess: true}, o2.MQTT.Host, o2.MQTT.Port)
+	defer mc2.Close()
+	testMQTTCheckConnAck(t, rc2, mqttConnAckRCConnectionAccepted, false)
+	testMQTTSub(t, 1, mc2, rc2, []*mqttFilter{{filter: "foo", qos: 1}}, []byte{1})
+	testMQTTFlush(t, mc2, nil, rc2)
+
+	// Connect a publisher from leafnode and publish, verify message is received.
+	connectAndPublish(lno)
+	testMQTTCheckPubMsg(t, mc2, rc2, "foo", mqttPubQos1, []byte("msg"))
+
+	// Connect from one server in the cluster check it works from there too.
+	connectAndPublish(o1)
+	testMQTTCheckPubMsg(t, mc2, rc2, "foo", mqttPubQos1, []byte("msg"))
+}
+
 func TestMQTTParseUnsub(t *testing.T) {
 	eofr := testNewEOFReader()
 	for _, test := range []struct {
@@ -3323,6 +3428,7 @@ func TestMQTTWillRetainPermViolation(t *testing.T) {
 	template := `
 		port: -1
 		jetstream: enabled
+		server_name: mqtt
 		authorization {
 			mqtt_perms = {
 				publish = ["%s"]
@@ -4446,12 +4552,12 @@ func TestMQTTMaxAckPending(t *testing.T) {
 
 func TestMQTTMaxAckPendingForMultipleSubs(t *testing.T) {
 	o := testMQTTDefaultOptions()
-	o.MQTT.AckWait = 500 * time.Millisecond
+	o.MQTT.AckWait = time.Second
 	o.MQTT.MaxAckPending = 1
 	s := testMQTTRunServer(t, o)
 	defer testMQTTShutdownServer(s)
 
-	cisub := &mqttConnInfo{clientID: "sub", cleanSess: false}
+	cisub := &mqttConnInfo{clientID: "sub", cleanSess: true}
 	c, r := testMQTTConnect(t, cisub, o.MQTT.Host, o.MQTT.Port)
 	defer c.Close()
 	testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, false)
@@ -4544,6 +4650,7 @@ func TestMQTTMaxAckPendingOverLimit(t *testing.T) {
 func TestMQTTConfigReload(t *testing.T) {
 	template := `
 		jetstream: true
+		server_name: mqtt
 		mqtt {
 			port: -1
 			ack_wait: %s
diff --git a/server/server_test.go b/server/server_test.go
index dfe76673..3ed4b2ad 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -1230,6 +1230,7 @@ func TestAcceptError(t *testing.T) {
 
 func TestServerShutdownDuringStart(t *testing.T) {
 	o := DefaultOptions()
+	o.ServerName = "server"
 	o.DisableShortFirstPing = true
 	o.Accounts = []*Account{NewAccount("$SYS")}
 	o.SystemAccount = "$SYS"