Repository: nats-io/nats-server
Pull request #3187 (2022-06-13 21:13:49)

[ADD] account specific in/out msgs/bytes stats to CONNS
This subject $SYS.ACCOUNT.%s.SERVER.CONNS will now respond with account
specific datastats for Received and sent messages as well as number of slow
consumers for the account.

Signed-off-by: Matthias Hanel <mh@synadia.com>

Benchmarks ran are `PubSub` and `Pub0b_Payload` 3 times each.
Without then with this change.
The impact seems small enough to prefer the straight forward solution. 
If deemed not so, I can try to also delete the server counter and aggregate its value from account based ones.

First set of 3 is without this change.
```
> go test -v -vet=off -bench='PubSub$' -run=X
goos: darwin
goarch: amd64
pkg: github.com/nats-io/nats-server/v2/test
cpu: Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz
Benchmark_____________PubSub
Benchmark_____________PubSub-16    	 5629795	       203.8 ns/op
Benchmark__DenyMsgNoWCPubSub
Benchmark__DenyMsgNoWCPubSub-16    	 5474841	       203.9 ns/op
Benchmark_DenyMsgYesWCPubSub
Benchmark_DenyMsgYesWCPubSub-16    	 5560723	       206.4 ns/op
PASS
ok  	github.com/nats-io/nats-server/v2/test	4.631s
> go test -v -vet=off -bench='PubSub$' -run=X
goos: darwin
goarch: amd64
pkg: github.com/nats-io/nats-server/v2/test
cpu: Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz
Benchmark_____________PubSub
Benchmark_____________PubSub-16    	 5586711	       203.5 ns/op
Benchmark__DenyMsgNoWCPubSub
Benchmark__DenyMsgNoWCPubSub-16    	 5466132	       205.7 ns/op
Benchmark_DenyMsgYesWCPubSub
Benchmark_DenyMsgYesWCPubSub-16    	 5449238	       206.3 ns/op
PASS
ok  	github.com/nats-io/nats-server/v2/test	4.617s
> go test -v -vet=off -bench='PubSub$' -run=X
goos: darwin
goarch: amd64
pkg: github.com/nats-io/nats-server/v2/test
cpu: Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz
Benchmark_____________PubSub
Benchmark_____________PubSub-16    	 5619054	       204.3 ns/op
Benchmark__DenyMsgNoWCPubSub
Benchmark__DenyMsgNoWCPubSub-16    	 5584328	       205.3 ns/op
Benchmark_DenyMsgYesWCPubSub
Benchmark_DenyMsgYesWCPubSub-16    	 5423282	       208.1 ns/op
PASS
ok  	github.com/nats-io/nats-server/v2/test	4.647s
> go test -v -vet=off -bench="Pub0b_Payload" -run=X
goos: darwin
goarch: amd64
pkg: github.com/nats-io/nats-server/v2/test
cpu: Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz
Benchmark______Pub0b_Payload
Benchmark______Pub0b_Payload-16    	15054192	        67.94 ns/op	 161.91 MB/s
Benchmark__AuthPub0b_Payload
Benchmark__AuthPub0b_Payload-16    	 7577354	       149.2 ns/op	  73.72 MB/s
PASS
ok  	github.com/nats-io/nats-server/v2/test	2.818s
> go test -v -vet=off -bench="Pub0b_Payload" -run=X
goos: darwin
goarch: amd64
pkg: github.com/nats-io/nats-server/v2/test
cpu: Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz
Benchmark______Pub0b_Payload
Benchmark______Pub0b_Payload-16    	15259664	        67.75 ns/op	 162.37 MB/s
Benchmark__AuthPub0b_Payload
Benchmark__AuthPub0b_Payload-16    	 7556667	       149.9 ns/op	  73.39 MB/s
PASS
ok  	github.com/nats-io/nats-server/v2/test	2.838s
> go test -v -vet=off -bench="Pub0b_Payload" -run=X
goos: darwin
goarch: amd64
pkg: github.com/nats-io/nats-server/v2/test
cpu: Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz
Benchmark______Pub0b_Payload
Benchmark______Pub0b_Payload-16    	15028556	        68.19 ns/op	 161.31 MB/s
Benchmark__AuthPub0b_Payload
Benchmark__AuthPub0b_Payload-16    	 7491639	       149.5 ns/op	  73.56 MB/s
PASS
ok  	github.com/nats-io/nats-server/v2/test	2.814s
```

Second set of 3 is with this change.
```
> git stash apply
Auto-merging server/events_test.go
Auto-merging server/client.go
On branch main
Your branch is up to date with 'origin/main'.

Changes not staged for commit:
  (use "git add <file>..." to update what will be committed)
  (use "git restore <file>..." to discard changes in working directory)
	modified:   ../server/accounts.go
	modified:   ../server/client.go
	modified:   ../server/events.go
	modified:   ../server/events_test.go

Untracked files:
  (use "git add <file>..." to include in what will be committed)
	../dist/
	../vendor/

no changes added to commit (use "git add" and/or "git commit -a")
> go test -v -vet=off -bench='PubSub$' -run=X
goos: darwin
goarch: amd64
pkg: github.com/nats-io/nats-server/v2/test
cpu: Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz
Benchmark_____________PubSub
Benchmark_____________PubSub-16    	 5527941	       203.8 ns/op
Benchmark__DenyMsgNoWCPubSub
Benchmark__DenyMsgNoWCPubSub-16    	 5517493	       204.6 ns/op
Benchmark_DenyMsgYesWCPubSub
Benchmark_DenyMsgYesWCPubSub-16    	 5486712	       210.7 ns/op
PASS
ok  	github.com/nats-io/nats-server/v2/test	4.712s
> go test -v -vet=off -bench='PubSub$' -run=X
goos: darwin
goarch: amd64
pkg: github.com/nats-io/nats-server/v2/test
cpu: Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz
Benchmark_____________PubSub
Benchmark_____________PubSub-16    	 5561923	       204.8 ns/op
Benchmark__DenyMsgNoWCPubSub
Benchmark__DenyMsgNoWCPubSub-16    	 5458404	       204.9 ns/op
Benchmark_DenyMsgYesWCPubSub
Benchmark_DenyMsgYesWCPubSub-16    	 5397844	       205.2 ns/op
PASS
ok  	github.com/nats-io/nats-server/v2/test	4.597s
> go test -v -vet=off -bench='PubSub$' -run=X
goos: darwin
goarch: amd64
pkg: github.com/nats-io/nats-server/v2/test
cpu: Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz
Benchmark_____________PubSub
Benchmark_____________PubSub-16    	 5502734	       204.0 ns/op
Benchmark__DenyMsgNoWCPubSub
Benchmark__DenyMsgNoWCPubSub-16    	 5579556	       205.4 ns/op
Benchmark_DenyMsgYesWCPubSub
Benchmark_DenyMsgYesWCPubSub-16    	 5502544	       215.5 ns/op
PASS
ok  	github.com/nats-io/nats-server/v2/test	4.682s
> go test -v -vet=off -bench="Pub0b_Payload" -run=X
goos: darwin
goarch: amd64
pkg: github.com/nats-io/nats-server/v2/test
cpu: Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz
Benchmark______Pub0b_Payload
Benchmark______Pub0b_Payload-16    	15086301	        67.56 ns/op	 162.83 MB/s
Benchmark__AuthPub0b_Payload
Benchmark__AuthPub0b_Payload-16    	 7361306	       148.4 ns/op	  74.13 MB/s
PASS
ok  	github.com/nats-io/nats-server/v2/test	2.779s
> go test -v -vet=off -bench="Pub0b_Payload" -run=X
goos: darwin
goarch: amd64
pkg: github.com/nats-io/nats-server/v2/test
cpu: Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz
Benchmark______Pub0b_Payload
Benchmark______Pub0b_Payload-16    	15156819	        67.78 ns/op	 162.30 MB/s
Benchmark__AuthPub0b_Payload
Benchmark__AuthPub0b_Payload-16    	 7549778	       148.5 ns/op	  74.09 MB/s
PASS
ok  	github.com/nats-io/nats-server/v2/test	2.810s
> go test -v -vet=off -bench="Pub0b_Payload" -run=X
goos: darwin
goarch: amd64
pkg: github.com/nats-io/nats-server/v2/test
cpu: Intel(R) Core(TM) i9-9980HK CPU @ 2.40GHz
Benchmark______Pub0b_Payload
Benchmark______Pub0b_Payload-16    	15042363	        67.78 ns/op	 162.28 MB/s
Benchmark__AuthPub0b_Payload
Benchmark__AuthPub0b_Payload-16    	 7392249	       149.7 ns/op	  73.48 MB/s
PASS
ok  	github.com/nats-io/nats-server/v2/test	2.797s
>
```

derekcollison: ok looks good. ping me when you any changes have been pushed.

Patch:
diff --git a/.travis.yml b/.travis.yml
index 45d96dc9..61f67cd5 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -7,7 +7,7 @@ vm:
 
 language: go
 go:
-- 1.17.x
+- 1.18.x
 
 addons:
   apt:
@@ -32,10 +32,10 @@ jobs:
     - name: "Run all tests from all other packages"
       env: TEST_SUITE=non_srv_pkg_tests
     - name: "Compile and various checks with older Go release"
-      go: 1.16.x
-      env: TEST_SUITE=compile
+      go: 1.17.x
+      env: TEST_SUITE=build_only
 
-script: ./runTestsOnTravis.sh $TEST_SUITE
+script: ./scripts/runTestsOnTravis.sh $TEST_SUITE
 
 deploy:
   provider: script
@@ -43,4 +43,4 @@ deploy:
   script: curl -sL http://git.io/goreleaser | bash
   on:
     tags: true
-    condition: ($TRAVIS_GO_VERSION =~ 1.17) && ($TEST_SUITE = "compile")
+    condition: ($TRAVIS_GO_VERSION =~ 1.18) && ($TEST_SUITE = "compile")
diff --git a/server/accounts.go b/server/accounts.go
index 1e415fd7..edf6ea14 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -49,6 +49,7 @@ var maxSubLimitReportThreshold = defaultMaxSubLimitReportThreshold
 // 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 {
+	stats
 	gwReplyMapping
 	Name         string
 	Nkey         string
@@ -1812,20 +1813,8 @@ func (a *Account) addServiceImport(dest *Account, from, to string, claim *jwt.Im
 		rt = se.respType
 		lat = se.latency
 	}
-	s := dest.srv
 	dest.mu.RUnlock()
 
-	// Track if this maps us to the system account.
-	// We will always share information with them.
-	var isSysAcc bool
-	if s != nil {
-		s.mu.Lock()
-		if s.sys != nil && dest == s.sys.account {
-			isSysAcc = true
-		}
-		s.mu.Unlock()
-	}
-
 	a.mu.Lock()
 	if a.imports.services == nil {
 		a.imports.services = make(map[string]*serviceImport)
@@ -1862,8 +1851,7 @@ func (a *Account) addServiceImport(dest *Account, from, to string, claim *jwt.Im
 			}
 		}
 	}
-	// Turn on sharing by default if importing from system services.
-	share := isSysAcc
+	var share bool
 	if claim != nil {
 		share = claim.Share
 	}
diff --git a/server/client.go b/server/client.go
index 37add98e..64af0411 100644
--- a/server/client.go
+++ b/server/client.go
@@ -1257,6 +1257,10 @@ func (c *client) readLoop(pre []byte) {
 		if c.in.msgs > 0 {
 			atomic.AddInt64(&c.inMsgs, int64(c.in.msgs))
 			atomic.AddInt64(&c.inBytes, int64(c.in.bytes))
+			if acc != nil {
+				atomic.AddInt64(&acc.inMsgs, int64(c.in.msgs))
+				atomic.AddInt64(&acc.inBytes, int64(c.in.bytes))
+			}
 			atomic.AddInt64(&s.inMsgs, int64(c.in.msgs))
 			atomic.AddInt64(&s.inBytes, int64(c.in.bytes))
 		}
@@ -1518,6 +1522,9 @@ func (c *client) handleWriteTimeout(written, attempted int64, numChunks int) boo
 
 	// Slow consumer here..
 	atomic.AddInt64(&c.srv.slowConsumers, 1)
+	if c.acc != nil {
+		atomic.AddInt64(&c.acc.slowConsumers, 1)
+	}
 	c.Noticef("Slow Consumer Detected: WriteDeadline of %v exceeded with %d chunks of %d total bytes.",
 		c.out.wdl, numChunks, attempted)
 
@@ -1980,6 +1987,9 @@ func (c *client) queueOutbound(data []byte) {
 		// checking current pb+len(data) and then add to pb.
 		c.out.pb -= int64(len(data))
 		atomic.AddInt64(&c.srv.slowConsumers, 1)
+		if c.acc != nil {
+			atomic.AddInt64(&c.acc.slowConsumers, 1)
+		}
 		c.Noticef("Slow Consumer Detected: MaxPending of %d Exceeded", c.out.mp)
 		c.markConnAsClosed(SlowConsumerPendingBytes)
 		return
@@ -3178,6 +3188,10 @@ func (c *client) deliverMsg(sub *subscription, acc *Account, subject, reply, mh,
 	}
 
 	// We don't count internal deliveries so we update server statistics here.
+	if acc != nil {
+		atomic.AddInt64(&acc.outMsgs, 1)
+		atomic.AddInt64(&acc.outBytes, msgSize)
+	}
 	atomic.AddInt64(&srv.outMsgs, 1)
 	atomic.AddInt64(&srv.outBytes, msgSize)
 
@@ -3848,12 +3862,7 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 	// However the GetNext for consumers is a no-op and causes buildups of service imports,
 	// response service imports and rrMap entries which all will need to simply expire.
 	// TODO(dlc) - Come up with something better.
-	if checkJSGetNext && si.se != nil && si.se.acc == c.srv.SystemAccount() {
-		shouldReturn = true
-	}
-
-	// Check for short circuit return.
-	if shouldReturn {
+	if shouldReturn || (checkJSGetNext && si.se != nil && si.se.acc == c.srv.SystemAccount()) {
 		return
 	}
 
@@ -3920,6 +3929,7 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 	// Place our client info for the request in the original message.
 	// This will survive going across routes, etc.
 	if !isResponse {
+		isSysImport := si.acc == c.srv.SystemAccount()
 		var ci *ClientInfo
 		if hadPrevSi && c.pa.hdr >= 0 {
 			var cis ClientInfo
@@ -3928,15 +3938,22 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 				ci.Service = acc.Name
 				// Check if we are moving into a share details account from a non-shared
 				// and add in server and cluster details.
-				if !share && si.share {
+				if !share && (si.share || isSysImport) {
 					c.addServerAndClusterInfo(ci)
 				}
 			}
 		} else if c.kind != LEAF || c.pa.hdr < 0 || len(getHeader(ClientInfoHdr, msg[:c.pa.hdr])) == 0 {
 			ci = c.getClientInfo(share)
-		} else if c.kind == LEAF && si.share {
+			// If we did not share but the imports destination is the system account add in the server and cluster info.
+			if !share && isSysImport {
+				c.addServerAndClusterInfo(ci)
+			}
+		} else if c.kind == LEAF && (si.share || isSysImport) {
 			// We have a leaf header here for ci, augment as above.
 			ci = c.getClientInfo(si.share)
+			if !si.share && isSysImport {
+				c.addServerAndClusterInfo(ci)
+			}
 		}
 		// Set clientInfo if present.
 		if ci != nil {
diff --git a/server/const.go b/server/const.go
index 7f9a4f68..3de703be 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.8.5-beta.4"
+	VERSION = "2.8.5-beta.6"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/consumer.go b/server/consumer.go
index b801f199..e40058f0 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -1382,9 +1382,6 @@ func (acc *Account) checkNewConsumerConfig(cfg, ncfg *ConsumerConfig) error {
 		return nil
 	}
 	// Something different, so check since we only allow certain things to be updated.
-	if cfg.FilterSubject != ncfg.FilterSubject {
-		return errors.New("filter subject can not be updated")
-	}
 	if cfg.DeliverPolicy != ncfg.DeliverPolicy {
 		return errors.New("deliver policy can not be updated")
 	}
@@ -1963,6 +1960,7 @@ func (o *consumer) readStoredState(slseq uint64) error {
 }
 
 // Apply the consumer stored state.
+// Lock should be held.
 func (o *consumer) applyState(state *ConsumerState) {
 	if state == nil {
 		return
@@ -1993,6 +1991,7 @@ func (o *consumer) applyState(state *ConsumerState) {
 }
 
 // Sets our store state from another source. Used in clustered mode on snapshot restore.
+// Lock should be held.
 func (o *consumer) setStoreState(state *ConsumerState) error {
 	if state == nil || o.store == nil {
 		return nil
diff --git a/server/errors.json b/server/errors.json
index 1192377b..e0ba3cf2 100644
--- a/server/errors.json
+++ b/server/errors.json
@@ -1248,5 +1248,25 @@
     "help": "",
     "url": "",
     "deprecates": ""
+  },
+  {
+    "constant": "JSConsumerNameContainsPathSeparatorsErr",
+    "code": 400,
+    "error_code": 10127,
+    "description": "Consumer name can not contain path separators",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
+  },
+  {
+    "constant": "JSStreamNameContainsPathSeparatorsErr",
+    "code": 400,
+    "error_code": 10128,
+    "description": "Stream name can not contain path separators",
+    "comment": "",
+    "help": "",
+    "url": "",
+    "deprecates": ""
   }
 ]
\ No newline at end of file
diff --git a/server/events.go b/server/events.go
index bec38180..e160da54 100644
--- a/server/events.go
+++ b/server/events.go
@@ -138,11 +138,14 @@ const DisconnectEventMsgType = "io.nats.server.advisory.v1.client_disconnect"
 // updates in the absence of any changes.
 type AccountNumConns struct {
 	TypedEvent
-	Server     ServerInfo `json:"server"`
-	Account    string     `json:"acc"`
-	Conns      int        `json:"conns"`
-	LeafNodes  int        `json:"leafnodes"`
-	TotalConns int        `json:"total_conns"`
+	Server        ServerInfo `json:"server"`
+	Account       string     `json:"acc"`
+	Conns         int        `json:"conns"`
+	LeafNodes     int        `json:"leafnodes"`
+	TotalConns    int        `json:"total_conns"`
+	Sent          DataStats  `json:"sent"`
+	Received      DataStats  `json:"received"`
+	SlowConsumers int64      `json:"slow_consumers"`
 }
 
 const AccountNumConnsMsgType = "io.nats.server.advisory.v1.account_connections"
@@ -296,9 +299,9 @@ func (s *Server) internalSendLoop(wg *sync.WaitGroup) {
 	defer wg.Done()
 
 RESET:
-	s.mu.Lock()
+	s.mu.RLock()
 	if s.sys == nil || s.sys.sendq == nil {
-		s.mu.Unlock()
+		s.mu.RUnlock()
 		return
 	}
 	sysc := s.sys.client
@@ -314,7 +317,7 @@ RESET:
 	if s.gateway.enabled {
 		cluster = s.getGatewayName()
 	}
-	s.mu.Unlock()
+	s.mu.RUnlock()
 
 	// Grab tags.
 	tags := s.getOpts().Tags
@@ -541,17 +544,17 @@ func (s *Server) eventsRunning() bool {
 	if s == nil {
 		return false
 	}
-	s.mu.Lock()
+	s.mu.RLock()
 	er := s.running && s.eventsEnabled()
-	s.mu.Unlock()
+	s.mu.RUnlock()
 	return er
 }
 
 // EventsEnabled will report if the server has internal events enabled via
 // a defined system account.
 func (s *Server) EventsEnabled() bool {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.eventsEnabled()
 }
 
@@ -564,8 +567,8 @@ func (s *Server) eventsEnabled() bool {
 // TrackedRemoteServers returns how many remote servers we are tracking
 // from a system events perspective.
 func (s *Server) TrackedRemoteServers() int {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	if !s.running || !s.eventsEnabled() {
 		return -1
 	}
@@ -764,8 +767,8 @@ func getHash(name string) []byte {
 
 // Returns the node name for this server which is a hash of the server name.
 func (s *Server) Node() string {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	if s.sys != nil {
 		return s.sys.shash
 	}
@@ -977,9 +980,9 @@ func (s *Server) initEventTracking() {
 func (s *Server) registerSystemImportsForExisting() {
 	var accounts []*Account
 
-	s.mu.Lock()
+	s.mu.RLock()
 	if s.sys == nil {
-		s.mu.Unlock()
+		s.mu.RUnlock()
 		return
 	}
 	sacc := s.sys.account
@@ -990,7 +993,7 @@ func (s *Server) registerSystemImportsForExisting() {
 		}
 		return true
 	})
-	s.mu.Unlock()
+	s.mu.RUnlock()
 
 	for _, a := range accounts {
 		s.registerSystemImports(a)
@@ -1290,9 +1293,9 @@ func (s *Server) leafNodeConnected(sub *subscription, _ *client, _ *Account, sub
 		return
 	}
 
-	s.mu.Lock()
+	s.mu.RLock()
 	na := m.Account == _EMPTY_ || !s.eventsEnabled() || !s.gateway.enabled
-	s.mu.Unlock()
+	s.mu.RUnlock()
 
 	if na {
 		return
@@ -1382,10 +1385,7 @@ func (s *Server) filterRequest(fOpts *EventFilterOptions) bool {
 		return true
 	}
 	if fOpts.Cluster != _EMPTY_ {
-		s.mu.Lock()
-		cluster := s.info.Cluster
-		s.mu.Unlock()
-		if !strings.Contains(cluster, fOpts.Cluster) {
+		if !strings.Contains(s.ClusterName(), fOpts.Cluster) {
 			return true
 		}
 	}
@@ -1644,6 +1644,7 @@ func (s *Server) sendAccConnsUpdate(a *Account, subj ...string) {
 	eid := s.nextEventID()
 	a.mu.Lock()
 	localConns := a.numLocalConnections()
+	leafConns := a.numLocalLeafNodes()
 	m := &AccountNumConns{
 		TypedEvent: TypedEvent{
 			Type: AccountNumConnsMsgType,
@@ -1652,8 +1653,15 @@ func (s *Server) sendAccConnsUpdate(a *Account, subj ...string) {
 		},
 		Account:    a.Name,
 		Conns:      localConns,
-		LeafNodes:  a.numLocalLeafNodes(),
-		TotalConns: localConns + a.numLocalLeafNodes(),
+		LeafNodes:  leafConns,
+		TotalConns: localConns + leafConns,
+		Received: DataStats{
+			Msgs:  atomic.LoadInt64(&a.inMsgs),
+			Bytes: atomic.LoadInt64(&a.inBytes)},
+		Sent: DataStats{
+			Msgs:  atomic.LoadInt64(&a.outMsgs),
+			Bytes: atomic.LoadInt64(&a.outBytes)},
+		SlowConsumers: atomic.LoadInt64(&a.slowConsumers),
 	}
 	// Set timer to fire again unless we are at zero.
 	if localConns == 0 {
@@ -1808,6 +1816,7 @@ func (s *Server) sendAuthErrorEvent(c *client) {
 	}
 	eid := s.nextEventID()
 	s.mu.Unlock()
+
 	now := time.Now().UTC()
 	c.mu.Lock()
 	m := DisconnectEventMsg{
@@ -1908,13 +1917,13 @@ func (s *Server) sysUnsubscribe(sub *subscription) {
 	if sub == nil {
 		return
 	}
-	s.mu.Lock()
+	s.mu.RLock()
 	if !s.eventsEnabled() {
-		s.mu.Unlock()
+		s.mu.RUnlock()
 		return
 	}
 	c := sub.client
-	s.mu.Unlock()
+	s.mu.RUnlock()
 
 	if c != nil {
 		c.processUnsub(sub.sid)
@@ -1996,13 +2005,13 @@ func (s *Server) remoteLatencyUpdate(sub *subscription, _ *client, _ *Account, s
 // This is used for all inbox replies so that we do not send supercluster wide interest
 // updates for every request. Same trick used in modern NATS clients.
 func (s *Server) inboxReply(sub *subscription, c *client, acc *Account, subject, reply string, msg []byte) {
-	s.mu.Lock()
+	s.mu.RLock()
 	if !s.eventsEnabled() || s.sys.replies == nil {
-		s.mu.Unlock()
+		s.mu.RUnlock()
 		return
 	}
 	cb, ok := s.sys.replies[subject]
-	s.mu.Unlock()
+	s.mu.RUnlock()
 
 	if ok && cb != nil {
 		cb(sub, c, acc, subject, reply, msg)
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 349e5aea..ac719b61 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -187,6 +187,16 @@ const (
 	// Will return JSON response.
 	JSApiRemoveServer = "$JS.API.SERVER.REMOVE"
 
+	// JSApiServerStreamMove is the endpoint to move streams off a server
+	// Only works from system account.
+	// Will return JSON response.
+	JSApiServerStreamMove = "$JS.API.SERVER.STREAM.MOVE"
+
+	// JSApiServerInfo is the endpoint to abtain which stream a server contains
+	// Only works from system account.
+	// Will return JSON response.
+	//JSApiServerStreamInfo = "$JS.API.SERVER.STREAM.INFO"
+
 	// jsAckT is the template for the ack message stream coming back from a consumer
 	// when they ACK/NAK, etc a message.
 	jsAckT      = "$JS.ACK.%s.%s"
@@ -558,6 +568,36 @@ type JSApiMetaServerRemoveResponse struct {
 
 const JSApiMetaServerRemoveResponseType = "io.nats.jetstream.api.v1.meta_server_remove_response"
 
+/*
+// JSApiMetaServerStreamInfoRequest will provide which streams are located on a particular server
+type JSApiMetaServerStreamInfoRequest struct {
+	// Server name of the peer to be evacuated.
+	Server string `json:"peer"`
+}
+
+// JSApiMetaServerStreamInfoResponse is the response to a peer evacuation request in the meta group.
+type JSApiMetaServerStreamInfoResponse struct {
+	ApiResponse
+	Success bool                      `json:"success,omitempty"`
+	Content map[string][]StreamConfig `json:"content,omitempty"`
+}
+
+const JSApiMetaServerStreamInfoType = "io.nats.jetstream.api.v1.meta_server_stream_info"
+*/
+
+// JSApiMetaServerStreamMoveRequest will move a stream on a server to another
+// response to this will come as JSApiStreamUpdateResponse/JSApiStreamUpdateResponseType
+type JSApiMetaServerStreamMoveRequest struct {
+	// Server name of the peer to be evacuated.
+	Server string `json:"peer"`
+	// Account name the stream to move is in
+	Account string `json:"account"`
+	// Name of stream to move
+	Stream string `json:"string"`
+	// Ephemeral placement tags for the move
+	Tags []string `json:"tags,omitempty"`
+}
+
 // JSApiMsgGetRequest get a message request.
 type JSApiMsgGetRequest struct {
 	Seq     uint64 `json:"seq,omitempty"`
@@ -1251,6 +1291,13 @@ func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, _ *Account,
 		return
 	}
 
+	// Check for path like separators in the name.
+	if strings.ContainsAny(streamName, `\/`) {
+		resp.Error = NewJSStreamNameContainsPathSeparatorsError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
 	// Can't create a stream with a sealed state.
 	if cfg.Sealed {
 		resp.Error = NewJSStreamInvalidConfigError(fmt.Errorf("stream configuration for create can not be sealed"))
@@ -1353,9 +1400,9 @@ func (s *Server) jsStreamUpdateRequest(sub *subscription, c *client, _ *Account,
 		// If we are inline with client, we still may need to do a callout for stream info
 		// during this call, so place in Go routine to not block client.
 		if c.kind != ROUTER && c.kind != GATEWAY {
-			go s.jsClusteredStreamUpdateRequest(ci, acc, subject, reply, rmsg, &cfg)
+			go s.jsClusteredStreamUpdateRequest(ci, acc, subject, reply, rmsg, &cfg, nil)
 		} else {
-			s.jsClusteredStreamUpdateRequest(ci, acc, subject, reply, rmsg, &cfg)
+			s.jsClusteredStreamUpdateRequest(ci, acc, subject, reply, rmsg, &cfg, nil)
 		}
 		return
 	}
@@ -2149,6 +2196,262 @@ func (s *Server) jsLeaderServerRemoveRequest(sub *subscription, c *client, _ *Ac
 	s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 }
 
+/*
+// Request to have the metaleader provide which streams a server contains
+func (s *Server) jsLeaderServerStreamInfoRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
+	if c == nil || !s.JetStreamEnabled() {
+		return
+	}
+
+	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	if err != nil {
+		s.Warnf(badAPIRequestT, msg)
+		return
+	}
+
+	js, cc := s.getJetStreamCluster()
+	if js == nil && (cc != nil && cc.meta == nil) {
+		return
+	}
+
+	// Extra checks here but only leader is listening.
+	js.mu.RLock()
+	isLeader := cc.isLeader()
+	js.mu.RUnlock()
+
+	if !isLeader {
+		return
+	}
+
+	var resp = JSApiMetaServerStreamInfoResponse{ApiResponse: ApiResponse{Type: JSApiMetaServerStreamInfoType}}
+
+	if isEmptyRequest(msg) {
+		resp.Error = NewJSBadRequestError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	var req JSApiMetaServerStreamInfoRequest
+	if err := json.Unmarshal(msg, &req); err != nil {
+		resp.Error = NewJSInvalidJSONError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	if cc == nil {
+		// single server logic
+		if req.Server != s.info.Name {
+			resp.Error = NewJSClusterServerNotMemberError()
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+
+		resp.Content = make(map[string][]StreamConfig)
+		resp.Success = true
+		js.mu.RLock()
+		for aName, jsa := range js.accounts {
+			jsa.mu.RLock()
+			for _, s := range jsa.streams {
+				resp.Content[aName] = append(resp.Content[aName], s.cfg)
+			}
+			jsa.mu.RUnlock()
+		}
+		js.mu.RUnlock()
+		return
+	}
+	// clustered code
+
+	var found string
+	js.mu.RLock()
+	for _, p := range cc.meta.Peers() {
+		si, ok := s.nodeToInfo.Load(p.ID)
+		if ok && si.(nodeInfo).name == req.Server {
+			found = p.ID
+			break
+		}
+	}
+	js.mu.RUnlock()
+
+	if found == _EMPTY_ {
+		resp.Error = NewJSClusterServerNotMemberError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	resp.Content = make(map[string][]StreamConfig)
+	js.mu.Lock()
+	for aName, a := range cc.streams {
+		for _, s := range a {
+			contains := false
+			for _, p := range s.Group.Peers {
+				if p == found {
+					contains = true
+					break
+				}
+			}
+			if contains {
+				resp.Content[aName] = append(resp.Content[aName], *s.Config)
+			}
+		}
+	}
+	js.mu.Unlock()
+
+	resp.Success = true
+	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+}
+*/
+
+// Request to have the metaleader move a stream on a peer to another
+func (s *Server) jsLeaderServerStreamMoveRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
+	if c == nil || !s.JetStreamEnabled() {
+		return
+	}
+
+	ci, acc, _, msg, err := s.getRequestInfo(c, rmsg)
+	if err != nil {
+		s.Warnf(badAPIRequestT, msg)
+		return
+	}
+
+	js, cc := s.getJetStreamCluster()
+	if js == nil || cc == nil || cc.meta == nil {
+		return
+	}
+
+	// Extra checks here but only leader is listening.
+	js.mu.RLock()
+	isLeader := cc.isLeader()
+	js.mu.RUnlock()
+
+	if !isLeader {
+		return
+	}
+
+	var resp = JSApiStreamUpdateResponse{ApiResponse: ApiResponse{Type: JSApiStreamUpdateResponseType}}
+
+	if isEmptyRequest(msg) {
+		resp.Error = NewJSBadRequestError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	var req JSApiMetaServerStreamMoveRequest
+	if err := json.Unmarshal(msg, &req); err != nil {
+		resp.Error = NewJSInvalidJSONError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	var srcPeer string
+	js.mu.RLock()
+	for _, p := range cc.meta.Peers() {
+		si, ok := s.nodeToInfo.Load(p.ID)
+		if ok && si.(nodeInfo).name == req.Server {
+			srcPeer = p.ID
+			break
+		}
+	}
+	js.mu.RUnlock()
+
+	if srcPeer == _EMPTY_ {
+		resp.Error = NewJSClusterServerNotMemberError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	targetAcc, ok := s.accounts.Load(req.Account)
+	if !ok {
+		resp.Error = NewJSNoAccountError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	streamFound := false
+	cfg := StreamConfig{}
+	currPeers := []string{}
+	currCluster := _EMPTY_
+	js.mu.Lock()
+	streams, ok := cc.streams[req.Account]
+	if ok {
+		sa, ok := streams[req.Stream]
+		if ok {
+			cfg = *sa.Config
+			streamFound = true
+			currPeers = sa.Group.Peers
+			currCluster = sa.Group.Cluster
+		}
+	}
+	js.mu.Unlock()
+
+	// make sure src peer is first. Removal will drop peers from the left
+	for i := 0; i < len(currPeers); i++ {
+		if currPeers[i] == srcPeer {
+			currPeers[i] = currPeers[0]
+			currPeers[0] = srcPeer
+			break
+		}
+	}
+
+	if !streamFound {
+		resp.Error = NewJSStreamNotFoundError()
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
+	// make sure client is scoped to requested account
+	ciNew := *(ci)
+	ciNew.Account = req.Account
+
+	// backup placement such that peers can be looked up with modified tag list
+	var origPlacement *Placement
+	if cfg.Placement != nil {
+		tmp := *cfg.Placement
+		origPlacement = &tmp
+	}
+
+	if len(req.Tags) > 0 {
+		if cfg.Placement == nil {
+			cfg.Placement = &Placement{}
+		}
+		cfg.Placement.Tags = append(cfg.Placement.Tags, req.Tags...)
+	}
+
+	peers := cc.selectPeerGroup(cfg.Replicas+1, currCluster, &cfg, currPeers)
+	if len(peers) <= cfg.Replicas {
+		// since expanding in the same cluster did not yield a result, try in different cluster
+		peers = nil
+
+		clusters := map[string]struct{}{}
+		s.nodeToInfo.Range(func(_, ni interface{}) bool {
+			if currCluster != ni.(nodeInfo).cluster {
+				clusters[ni.(nodeInfo).cluster] = struct{}{}
+			}
+			return true
+		})
+		for cluster := range clusters {
+			newPeers := cc.selectPeerGroup(cfg.Replicas, cluster, &cfg, nil)
+			if len(newPeers) >= cfg.Replicas {
+				peers = append([]string{}, currPeers...)
+				peers = append(peers, newPeers[:cfg.Replicas]...)
+				break
+			}
+		}
+		if peers == nil {
+			resp.Error = NewJSClusterNoPeersError()
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+	}
+
+	cfg.Placement = origPlacement
+
+	if c.kind != ROUTER && c.kind != GATEWAY {
+		go s.jsClusteredStreamUpdateRequest(&ciNew, targetAcc.(*Account), subject, reply, rmsg, &cfg, peers)
+	} else {
+		s.jsClusteredStreamUpdateRequest(&ciNew, targetAcc.(*Account), subject, reply, rmsg, &cfg, peers)
+	}
+}
+
 // Request to have the meta leader stepdown.
 // These will only be received the the meta leaders, so less checking needed.
 func (s *Server) jsLeaderStepDownRequest(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
@@ -3263,6 +3566,13 @@ func (s *Server) jsConsumerCreate(sub *subscription, c *client, a *Account, subj
 			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
 			return
 		}
+		// Check for path like separators in the name.
+		if strings.ContainsAny(consumerName, `\/`) {
+			resp.Error = NewJSConsumerNameContainsPathSeparatorsError()
+			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+			return
+		}
+
 	} else {
 		if numTokens(subject) != 5 {
 			resp.Error = NewJSConsumerEphemeralWithDurableInSubjectError()
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index cf3f877e..fa2791bb 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -52,6 +52,10 @@ type jetStreamCluster struct {
 	stepdown *subscription
 	// System level requests to remove a peer.
 	peerRemove *subscription
+	// System level request to evacuate a peer.
+	peerStreamMove *subscription
+	// System level request to obtain which streams a server
+	//peerStreamInfo *subscription
 }
 
 // Used to guide placement of streams and meta controllers in clustered JetStream.
@@ -158,6 +162,7 @@ const (
 	defaultStoreDirName  = "_js_"
 	defaultMetaGroupName = "_meta_"
 	defaultMetaFSBlkSize = 1024 * 1024
+	jsExcludePlacement   = "!jetstream"
 )
 
 // Returns information useful in mixed mode.
@@ -1316,7 +1321,7 @@ func (js *jetStream) processRemovePeer(peer string) {
 
 // Remove old peers after the new peers are caught up.
 // We are the old stream leader here.
-func (js *jetStream) removeOldPeers(mset *stream, newPreferred string) {
+func (js *jetStream) removeOldPeers(mset *stream, newPreferred string, newPeerSet []string) {
 	// Make sure still valid.
 	mset.mu.Lock()
 	isValid := mset.qch != nil
@@ -1327,7 +1332,6 @@ func (js *jetStream) removeOldPeers(mset *stream, newPreferred string) {
 	}
 
 	sa := mset.streamAssignment()
-	ci := js.clusterInfo(mset.raftGroup())
 
 	js.mu.Lock()
 	defer js.mu.Unlock()
@@ -1339,13 +1343,7 @@ func (js *jetStream) removeOldPeers(mset *stream, newPreferred string) {
 
 	cc, csa := js.cluster, sa.copyGroup()
 	numExpandedPeers := len(csa.Group.Peers)
-	csa.Group.Peers = csa.Group.Peers[:0]
-
-	for _, r := range ci.Replicas {
-		if r.cluster != ci.Name {
-			csa.Group.Peers = append(csa.Group.Peers, r.peer)
-		}
-	}
+	csa.Group.Peers = newPeerSet
 
 	// Now do consumers actually first here, followed by the owning stream.
 	for _, ca := range csa.consumers {
@@ -1360,6 +1358,9 @@ func (js *jetStream) removeOldPeers(mset *stream, newPreferred string) {
 		cc.meta.ForwardProposal(encodeAddConsumerAssignment(cca))
 	}
 
+	si, _ := js.srv.nodeToInfo.Load(newPreferred)
+	csa.Group.Cluster = si.(nodeInfo).cluster
+
 	csa.Group.Preferred = newPreferred
 	cc.meta.ForwardProposal(encodeUpdateStreamAssignment(csa))
 }
@@ -1797,7 +1798,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 
 			// We may receive a leader change after the stream assignment which would cancel us
 			// monitoring for this closely. So re-assess our state here as well.
-			migrating, peerGroup = mset.isMigrating()
+			migrating, peerGroup = mset.isMigrating(cc.meta.ID())
 
 			// Check for migrations here. We set the state on the stream assignment update below.
 			if isLeader && migrating {
@@ -1812,7 +1813,11 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 		case <-uch:
 			// We get this when we have a new stream assignment caused by an update. We want
 			// to know if we are migrating.
-			migrating, peerGroup = mset.isMigrating()
+			if cc == nil || cc.meta == nil {
+				migrating = false
+			} else {
+				migrating, peerGroup = mset.isMigrating(cc.meta.ID())
+			}
 			// If we are migrating and in the old peer group and we are leader, monitor for the
 			// new peers to be caught up. We could not be leader yet, so we will do same check below
 			// on leadership change.
@@ -1829,26 +1834,50 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 			// Check to see that we have someone caught up.
 			// TODO(dlc) - For now start checking after a second in order to give proper time to kick in any catchup logic needed.
 			// What we really need to do longer term is know if we need catchup and make sure that process has kicked off and/or completed.
-			ci := js.clusterInfo(mset.raftGroup())
+			rg := mset.raftGroup()
+			ci := js.clusterInfo(rg)
 			// The polling interval of one second allows this to be kicked in if needed.
 			if mset.hasCatchupPeers() {
 				mset.checkClusterInfo(ci)
 			}
 			// Track the new peers and check the ones that are current.
-			var newPeers []*PeerInfo
-			quorum := mset.cfg.Replicas/2 + 1
-			for _, r := range ci.Replicas {
-				if r.cluster != ci.Name {
-					if r.Current {
-						newPeers = append(newPeers, r)
+			if len(ci.Replicas)+1 != len(rg.Peers) {
+				continue
+			}
+			replicas := mset.cfg.Replicas
+			if len(rg.Peers) <= replicas {
+				stopMigrationMonitoring()
+				continue
+			}
+			//
+			toSkip := len(rg.Peers) - replicas
+			newPeerSet := rg.Peers[toSkip:]
+			currentCount := 0
+			firstPeer := _EMPTY_
+			for _, peer := range newPeerSet {
+				foundCurrent := peer == mset.leader
+				if !foundCurrent {
+					for _, p := range ci.Replicas {
+						if peer == string(getHash(p.Name)) {
+							if p.Current {
+								foundCurrent = true
+							}
+							break
+						}
+					}
+				}
+				if foundCurrent {
+					currentCount++
+					if firstPeer == _EMPTY_ {
+						firstPeer = peer
 					}
 				}
 			}
 			// If all are current we are good, or if we have some offline and we have a quorum.
-			if lnp := len(newPeers); lnp >= quorum {
+			if quorum := mset.cfg.Replicas/2 + 1; currentCount >= quorum {
 				stopMigrationMonitoring()
 				// Remove the old peers and transfer leadership.
-				time.AfterFunc(2*time.Second, func() { js.removeOldPeers(mset, newPeers[0].peer) })
+				time.AfterFunc(2*time.Second, func() { js.removeOldPeers(mset, firstPeer, newPeerSet) })
 			}
 		case err := <-restoreDoneCh:
 			// We have completed a restore from snapshot on this server. The stream assignment has
@@ -1960,9 +1989,9 @@ const (
 )
 
 // Determine if we are migrating and if so if we are part of the old or new set.
-func (mset *stream) isMigrating() (bool, peerMigrateType) {
+func (mset *stream) isMigrating(selfPeer string) (bool, peerMigrateType) {
 	mset.mu.RLock()
-	s, js, sa := mset.srv, mset.js, mset.sa
+	js, sa := mset.js, mset.sa
 	mset.mu.RUnlock()
 
 	js.mu.RLock()
@@ -1978,10 +2007,15 @@ func (mset *stream) isMigrating() (bool, peerMigrateType) {
 		return false, oldPeerGroup
 	}
 	// So we believe we are migrating here, need to determine if we are the old set or new set.
-	// We can shor circuit this based on our group assigned cluster vs our own.
-	if sa.Group.Cluster == s.cachedClusterName() {
-		return true, newPeerGroup
+	for i, peer := range sa.Group.Peers {
+		if peer == selfPeer {
+			if i >= sa.Config.Replicas {
+				return true, newPeerGroup
+			}
+			break
+		}
 	}
+
 	return true, oldPeerGroup
 }
 
@@ -3246,7 +3280,9 @@ func (js *jetStream) processClusterCreateConsumer(ca *consumerAssignment, state
 
 	// If we have an initial state set apply that now.
 	if state != nil && o != nil {
+		o.mu.Lock()
 		err = o.setStoreState(state)
+		o.mu.Unlock()
 	}
 	if err != nil {
 		if IsNatsErr(err, JSConsumerStoreFailedErrF) {
@@ -4018,6 +4054,14 @@ func (js *jetStream) startUpdatesSub() {
 	if cc.peerRemove == nil {
 		cc.peerRemove, _ = s.systemSubscribe(JSApiRemoveServer, _EMPTY_, false, c, s.jsLeaderServerRemoveRequest)
 	}
+	/*
+		if cc.peerStreamInfo == nil {
+			cc.peerStreamInfo, _ = s.systemSubscribe(JSApiServerStreamInfo, _EMPTY_, false, c, s.jsLeaderServerStreamInfoRequest)
+		}
+	*/
+	if cc.peerStreamMove == nil {
+		cc.peerStreamMove, _ = s.systemSubscribe(JSApiServerStreamMove, _EMPTY_, false, c, s.jsLeaderServerStreamMoveRequest)
+	}
 }
 
 // Lock should be held.
@@ -4039,6 +4083,16 @@ func (js *jetStream) stopUpdatesSub() {
 		cc.s.sysUnsubscribe(cc.peerRemove)
 		cc.peerRemove = nil
 	}
+	/*
+		if cc.peerStreamInfo != nil {
+			cc.s.sysUnsubscribe(cc.peerStreamInfo)
+			cc.peerStreamInfo = nil
+		}
+	*/
+	if cc.peerStreamMove != nil {
+		cc.s.sysUnsubscribe(cc.peerStreamMove)
+		cc.peerStreamMove = nil
+	}
 }
 
 func (js *jetStream) processLeaderChange(isLeader bool) {
@@ -4158,6 +4212,7 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 	type wn struct {
 		id    string
 		avail uint64
+		ha    int
 	}
 
 	var nodes []wn
@@ -4180,7 +4235,7 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 	if uniqueTagPrefix != _EMPTY_ {
 		for _, tag := range tags {
 			if strings.HasPrefix(tag, uniqueTagPrefix) {
-				// disable uniqueness check of explicitly listed in tags
+				// disable uniqueness check if explicitly listed in tags
 				uniqueTagPrefix = _EMPTY_
 				break
 			}
@@ -4189,6 +4244,8 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 	var uniqueTags = make(map[string]struct{})
 	maxHaAssets := s.getOpts().JetStreamLimits.MaxHAAssets
 
+	// Shuffle them up.
+	rand.Shuffle(len(peers), func(i, j int) { peers[i], peers[j] = peers[j], peers[i] })
 	for _, p := range peers {
 		si, ok := s.nodeToInfo.Load(p.ID)
 		if !ok || si == nil {
@@ -4208,6 +4265,10 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 			}
 		}
 
+		if ni.tags.Contains(jsExcludePlacement) {
+			continue
+		}
+
 		if len(tags) > 0 {
 			matched := true
 			for _, t := range tags {
@@ -4222,23 +4283,27 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 		}
 
 		var available uint64
-		switch cfg.Storage {
-		case MemoryStorage:
-			used := ni.stats.ReservedMemory
-			if ni.stats.Memory > used {
-				used = ni.stats.Memory
-			}
-			if ni.cfg.MaxMemory > int64(used) {
-				available = uint64(ni.cfg.MaxMemory) - used
-			}
-		case FileStorage:
-			used := ni.stats.ReservedStore
-			if ni.stats.Store > used {
-				used = ni.stats.Store
-			}
-			if ni.cfg.MaxStore > int64(used) {
-				available = uint64(ni.cfg.MaxStore) - used
+		var ha int
+		if ni.stats != nil {
+			switch cfg.Storage {
+			case MemoryStorage:
+				used := ni.stats.ReservedMemory
+				if ni.stats.Memory > used {
+					used = ni.stats.Memory
+				}
+				if ni.cfg.MaxMemory > int64(used) {
+					available = uint64(ni.cfg.MaxMemory) - used
+				}
+			case FileStorage:
+				used := ni.stats.ReservedStore
+				if ni.stats.Store > used {
+					used = ni.stats.Store
+				}
+				if ni.cfg.MaxStore > int64(used) {
+					available = uint64(ni.cfg.MaxStore) - used
+				}
 			}
+			ha = ni.stats.HAAssets
 		}
 
 		// Otherwise check if we have enough room if maxBytes set.
@@ -4271,7 +4336,7 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 			}
 		}
 		// Add to our list of potential nodes.
-		nodes = append(nodes, wn{p.ID, available})
+		nodes = append(nodes, wn{p.ID, available, ha})
 	}
 
 	// If we could not select enough peers, fail.
@@ -4281,6 +4346,11 @@ func (cc *jetStreamCluster) selectPeerGroup(r int, cluster string, cfg *StreamCo
 	// Sort based on available from most to least.
 	sort.Slice(nodes, func(i, j int) bool { return nodes[i].avail > nodes[j].avail })
 
+	// If we are placing a replicated stream, let's sort based in haAssets, as that is more important to balance.
+	if cfg.Replicas > 1 {
+		sort.SliceStable(nodes, func(i, j int) bool { return nodes[i].ha < nodes[j].ha })
+	}
+
 	var results []string
 	if len(existing) > 0 {
 		results = append(results, existing...)
@@ -4491,7 +4561,7 @@ func (s *Server) jsClusteredStreamRequest(ci *ClientInfo, acc *Account, subject,
 	cc.meta.Propose(encodeAddStreamAssignment(sa))
 }
 
-func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, subject, reply string, rmsg []byte, cfg *StreamConfig) {
+func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, subject, reply string, rmsg []byte, cfg *StreamConfig, peerSet []string) {
 	js, cc := s.getJetStreamCluster()
 	if js == nil || cc == nil {
 		return
@@ -4555,7 +4625,12 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 
 	// Check for a move update.
 	// TODO(dlc) - Should add a resolve from Tags to cluster and check that vs reflect.
-	isMoveRequest := newCfg.Placement != nil && !reflect.DeepEqual(osa.Config.Placement, newCfg.Placement)
+	var isMoveRequest bool
+	if len(peerSet) > 0 {
+		isMoveRequest = true
+	} else {
+		isMoveRequest = newCfg.Placement != nil && !reflect.DeepEqual(osa.Config.Placement, newCfg.Placement)
+	}
 
 	// Check for replica changes.
 	isReplicaChange := newCfg.Replicas != osa.Config.Replicas
@@ -4688,42 +4763,51 @@ func (s *Server) jsClusteredStreamUpdateRequest(ci *ClientInfo, acc *Account, su
 				consumers = append(consumers, cca)
 			}
 		}
-
 	} else if isMoveRequest {
-		nrg := js.createGroupForStream(ci, newCfg)
-		if nrg == nil {
-			resp.Error = NewJSInsufficientResourcesError()
-			s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
-			return
-		}
-		// Only change if resolved clusters are different.
-		if rg.Cluster != nrg.Cluster {
-			// If we are R1, make sure original is leader during scale up for move.
-			if len(rg.Peers) == 1 {
-				rg.Preferred = rg.Peers[0]
+		if len(peerSet) == 0 {
+			nrg := js.createGroupForStream(ci, newCfg)
+			if nrg == nil {
+				resp.Error = NewJSInsufficientResourcesError()
+				s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
+				return
 			}
-			// Add in new peers since we will extend the peer group to straddle both clusters.
-			rg.Peers = append(rg.Peers, nrg.Peers...)
-			rg.Cluster = nrg.Cluster
-
-			for _, ca := range osa.consumers {
-				cca := ca.copyGroup()
-				// Ephemerals are R=1, so only auto-remap if consumer peer count == nrg peer count.
-				numPeers := len(ca.Group.Peers)
-				if numPeers == len(nrg.Peers) {
-					cca.Group.Peers = append(cca.Group.Peers, nrg.Peers...)
-				} else {
-					// This is an ephemeral, so R1. Just randomly pick a single peer from the new set.
-					pi := rand.Int31n(int32(len(nrg.Peers)))
-					cca.Group.Peers = append(cca.Group.Peers, nrg.Peers[pi])
+			// filter peers present in both sets
+			for _, peer := range rg.Peers {
+				found := false
+				for _, newPeer := range nrg.Peers {
+					if peer == newPeer {
+						found = true
+						break
+					}
 				}
-				// Make sure to set if not already set.
-				if cca.Group.Preferred == _EMPTY_ {
-					cca.Group.Preferred = cca.Group.Peers[0]
+				if !found {
+					peerSet = append(peerSet, peer)
 				}
-				// We can not propose here before the stream itself so we collect them.
-				consumers = append(consumers, cca)
 			}
+			peerSet = append(peerSet, nrg.Peers...)
+		}
+		origPeers := len(rg.Peers)
+		if origPeers == 1 {
+			rg.Preferred = peerSet[0]
+		}
+		rg.Peers = peerSet
+
+		for _, ca := range osa.consumers {
+			cca := ca.copyGroup()
+			// Ephemerals are R=1, so only auto-remap if consumer peer count == nrg peer count.
+			if len(ca.Group.Peers) == newCfg.Replicas {
+				cca.Group.Peers = peerSet
+			} else {
+				// This is an ephemeral, so R1. Just randomly pick a single peer from the new set.
+				randPeer := peerSet[len(peerSet)-newCfg.Replicas:][rand.Int31n(int32(newCfg.Replicas))]
+				cca.Group.Peers = append(cca.Group.Peers, randPeer)
+			}
+			// Make sure to set if not already set.
+			if cca.Group.Preferred == _EMPTY_ {
+				cca.Group.Preferred = cca.Group.Peers[0]
+			}
+			// We can not propose here before the stream itself so we collect them.
+			consumers = append(consumers, cca)
 		}
 	} else {
 		// All other updates make sure no preferred is set.
diff --git a/server/jetstream_errors_generated.go b/server/jetstream_errors_generated.go
index be507d1a..7bffd96e 100644
--- a/server/jetstream_errors_generated.go
+++ b/server/jetstream_errors_generated.go
@@ -125,6 +125,9 @@ const (
 	// JSConsumerMaxWaitingNegativeErr consumer max waiting needs to be positive
 	JSConsumerMaxWaitingNegativeErr ErrorIdentifier = 10087
 
+	// JSConsumerNameContainsPathSeparatorsErr Consumer name can not contain path separators
+	JSConsumerNameContainsPathSeparatorsErr ErrorIdentifier = 10127
+
 	// JSConsumerNameExistErr consumer name already in use
 	JSConsumerNameExistErr ErrorIdentifier = 10013
 
@@ -314,6 +317,9 @@ const (
 	// JSStreamMsgDeleteFailedF Generic message deletion failure error string ({err})
 	JSStreamMsgDeleteFailedF ErrorIdentifier = 10057
 
+	// JSStreamNameContainsPathSeparatorsErr Stream name can not contain path separators
+	JSStreamNameContainsPathSeparatorsErr ErrorIdentifier = 10128
+
 	// JSStreamNameExistErr stream name already in use
 	JSStreamNameExistErr ErrorIdentifier = 10058
 
@@ -423,6 +429,7 @@ var (
 		JSConsumerMaxRequestBatchNegativeErr:       {Code: 400, ErrCode: 10114, Description: "consumer max request batch needs to be > 0"},
 		JSConsumerMaxRequestExpiresToSmall:         {Code: 400, ErrCode: 10115, Description: "consumer max request expires needs to be >= 1ms"},
 		JSConsumerMaxWaitingNegativeErr:            {Code: 400, ErrCode: 10087, Description: "consumer max waiting needs to be positive"},
+		JSConsumerNameContainsPathSeparatorsErr:    {Code: 400, ErrCode: 10127, Description: "Consumer name can not contain path separators"},
 		JSConsumerNameExistErr:                     {Code: 400, ErrCode: 10013, Description: "consumer name already in use"},
 		JSConsumerNameTooLongErrF:                  {Code: 400, ErrCode: 10102, Description: "consumer name is too long, maximum allowed is {max}"},
 		JSConsumerNotFoundErr:                      {Code: 404, ErrCode: 10014, Description: "consumer not found"},
@@ -486,6 +493,7 @@ var (
 		JSStreamMoveAndScaleErr:                    {Code: 400, ErrCode: 10123, Description: "can not move and scale a stream in a single update"},
 		JSStreamMoveInProgress:                     {Code: 400, ErrCode: 10124, Description: "stream move already in progress"},
 		JSStreamMsgDeleteFailedF:                   {Code: 500, ErrCode: 10057, Description: "{err}"},
+		JSStreamNameContainsPathSeparatorsErr:      {Code: 400, ErrCode: 10128, Description: "Stream name can not contain path separators"},
 		JSStreamNameExistErr:                       {Code: 400, ErrCode: 10058, Description: "stream name already in use"},
 		JSStreamNotFoundErr:                        {Code: 404, ErrCode: 10059, Description: "stream not found"},
 		JSStreamNotMatchErr:                        {Code: 400, ErrCode: 10060, Description: "expected stream does not match"},
@@ -969,6 +977,16 @@ func NewJSConsumerMaxWaitingNegativeError(opts ...ErrorOption) *ApiError {
 	return ApiErrors[JSConsumerMaxWaitingNegativeErr]
 }
 
+// NewJSConsumerNameContainsPathSeparatorsError creates a new JSConsumerNameContainsPathSeparatorsErr error: "Consumer name can not contain path separators"
+func NewJSConsumerNameContainsPathSeparatorsError(opts ...ErrorOption) *ApiError {
+	eopts := parseOpts(opts)
+	if ae, ok := eopts.err.(*ApiError); ok {
+		return ae
+	}
+
+	return ApiErrors[JSConsumerNameContainsPathSeparatorsErr]
+}
+
 // NewJSConsumerNameExistError creates a new JSConsumerNameExistErr error: "consumer name already in use"
 func NewJSConsumerNameExistError(opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
@@ -1701,6 +1719,16 @@ func NewJSStreamMsgDeleteFailedError(err error, opts ...ErrorOption) *ApiError {
 	}
 }
 
+// NewJSStreamNameContainsPathSeparatorsError creates a new JSStreamNameContainsPathSeparatorsErr error: "Stream name can not contain path separators"
+func NewJSStreamNameContainsPathSeparatorsError(opts ...ErrorOption) *ApiError {
+	eopts := parseOpts(opts)
+	if ae, ok := eopts.err.(*ApiError); ok {
+		return ae
+	}
+
+	return ApiErrors[JSStreamNameContainsPathSeparatorsErr]
+}
+
 // NewJSStreamNameExistError creates a new JSStreamNameExistErr error: "stream name already in use"
 func NewJSStreamNameExistError(opts ...ErrorOption) *ApiError {
 	eopts := parseOpts(opts)
diff --git a/server/mqtt.go b/server/mqtt.go
index 30f23b62..1c903b56 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -3622,6 +3622,9 @@ func (sess *mqttSession) processJSConsumer(c *client, subject, sid string,
 		if r := opts.MQTT.ConsumerReplicas; r > 0 {
 			cc.Replicas = r
 		}
+		if opts.MQTT.ConsumerInactiveThreshold > 0 {
+			cc.InactiveThreshold = opts.MQTT.ConsumerInactiveThreshold
+		}
 		if err := sess.createConsumer(cc); err != nil {
 			c.Errorf("Unable to add JetStream consumer for subscription on %q: err=%v", subject, err)
 			return nil, nil, err
diff --git a/server/opts.go b/server/opts.go
index 3a32d5ea..db5a1a56 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -431,6 +431,10 @@ type MQTTOpts struct {
 	// Note that existing consumers are not modified.
 	ConsumerMemoryStorage bool
 
+	// If specified will have the system auto-cleanup the consumers after being
+	// inactive for the specified amount of time.
+	ConsumerInactiveThreshold time.Duration
+
 	// Timeout for the authentication process.
 	AuthTimeout float64
 
@@ -4191,6 +4195,9 @@ func parseMQTT(v interface{}, o *Options, errors *[]error, warnings *[]error) er
 			o.MQTT.ConsumerReplicas = int(mv.(int64))
 		case "consumer_memory_storage":
 			o.MQTT.ConsumerMemoryStorage = mv.(bool)
+		case "consumer_inactive_threshold", "consumer_auto_cleanup":
+			o.MQTT.ConsumerInactiveThreshold = parseDuration("consumer_inactive_threshold", tk, mv, errors, warnings)
+
 		default:
 			if !tk.IsUsedVariable() {
 				err := &unknownConfigFieldErr{
diff --git a/server/reload.go b/server/reload.go
index bce1e3f7..372414e1 100644
--- a/server/reload.go
+++ b/server/reload.go
@@ -1,4 +1,4 @@
-// Copyright 2017-2021 The NATS Authors
+// Copyright 2017-2022 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
@@ -60,6 +60,9 @@ type option interface {
 	// IsJetStreamChange inidicates a change in the servers config for JetStream.
 	// Account changes will be handled separately in reloadAuthorization.
 	IsJetStreamChange() bool
+
+	// Indicates a change in the server that requires publishing the server's statz
+	IsStatszChange() bool
 }
 
 // noopOption is a base struct that provides default no-op behaviors.
@@ -89,6 +92,10 @@ func (n noopOption) IsJetStreamChange() bool {
 	return false
 }
 
+func (n noopOption) IsStatszChange() bool {
+	return false
+}
+
 // loggingOption is a base struct that provides default option behaviors for
 // logging-related options.
 type loggingOption struct {
@@ -293,6 +300,19 @@ func (a *authTimeoutOption) Apply(server *Server) {
 	server.Noticef("Reloaded: authorization timeout = %v", a.newValue)
 }
 
+// tagsOption implements the option interface for the `tags` setting.
+type tagsOption struct {
+	noopOption // Not authOption because this is a no-op; will be reloaded with options.
+}
+
+func (u *tagsOption) Apply(server *Server) {
+	server.Noticef("Reloaded: tags")
+}
+
+func (u *tagsOption) IsStatszChange() bool {
+	return true
+}
+
 // usersOption implements the option interface for the authorization `users`
 // setting.
 type usersOption struct {
@@ -582,6 +602,10 @@ func (jso jetStreamOption) IsJetStreamChange() bool {
 	return true
 }
 
+func (jso jetStreamOption) IsStatszChange() bool {
+	return true
+}
+
 type ocspOption struct {
 	noopOption
 	newValue *OCSPConfig
@@ -675,6 +699,15 @@ func (o *mqttConsumerMemoryStorageReload) Apply(s *Server) {
 	s.Noticef("Reloaded: MQTT consumer_memory_storage = %v", o.newValue)
 }
 
+type mqttInactiveThresholdReload struct {
+	noopOption
+	newValue time.Duration
+}
+
+func (o *mqttInactiveThresholdReload) Apply(s *Server) {
+	s.Noticef("Reloaded: MQTT consumer_inactive_threshold = %v", o.newValue)
+}
+
 // Compares options and disconnects clients that are no longer listed in pinned certs. Lock must not be held.
 func (s *Server) recheckPinnedCerts(curOpts *Options, newOpts *Options) {
 	s.mu.Lock()
@@ -957,16 +990,22 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 			return nil, err
 		}
 
-		if changed := !reflect.DeepEqual(oldValue, newValue); !changed {
-			// Check to make sure we are running JetStream if we think we should be.
-			if strings.ToLower(field.Name) == "jetstream" && newValue.(bool) {
-				if !jsEnabled {
-					diffOpts = append(diffOpts, &jetStreamOption{newValue: true})
+		optName := strings.ToLower(field.Name)
+		// accounts and users (referencing accounts) will always differ as accounts
+		// contain internal state, say locks etc..., so we don't bother here.
+		// This also avoids races with atomic stats counters
+		if optName != "accounts" && optName != "users" {
+			if changed := !reflect.DeepEqual(oldValue, newValue); !changed {
+				// Check to make sure we are running JetStream if we think we should be.
+				if optName == "jetstream" && newValue.(bool) {
+					if !jsEnabled {
+						diffOpts = append(diffOpts, &jetStreamOption{newValue: true})
+					}
 				}
+				continue
 			}
-			continue
 		}
-		switch optName := strings.ToLower(field.Name); optName {
+		switch optName {
 		case "traceverbose":
 			diffOpts = append(diffOpts, &traceVerboseOption{newValue: newValue.(bool)})
 		case "trace":
@@ -991,6 +1030,8 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 			diffOpts = append(diffOpts, &usernameOption{})
 		case "password":
 			diffOpts = append(diffOpts, &passwordOption{})
+		case "tags":
+			diffOpts = append(diffOpts, &tagsOption{})
 		case "authorization":
 			diffOpts = append(diffOpts, &authorizationOption{})
 		case "authtimeout":
@@ -1088,19 +1129,19 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 
 			// Special check for leafnode remotes changes which are not supported right now.
 			leafRemotesChanged := func(a, b LeafNodeOpts) bool {
-				if len(tmpOld.Remotes) != len(tmpNew.Remotes) {
+				if len(a.Remotes) != len(b.Remotes) {
 					return true
 				}
 
 				// Check whether all remotes URLs are still the same.
-				for _, oldRemote := range tmpOld.Remotes {
+				for _, oldRemote := range a.Remotes {
 					var found bool
 
 					if oldRemote.LocalAccount == _EMPTY_ {
 						oldRemote.LocalAccount = globalAccountName
 					}
 
-					for _, newRemote := range tmpNew.Remotes {
+					for _, newRemote := range b.Remotes {
 						// Bind to global account in case not defined.
 						if newRemote.LocalAccount == _EMPTY_ {
 							newRemote.LocalAccount = globalAccountName
@@ -1111,7 +1152,6 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 							break
 						}
 					}
-
 					if !found {
 						return true
 					}
@@ -1127,6 +1167,41 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 				tmpNew.Remotes = nil
 			}
 
+			// Special check for auth users to detect changes.
+			// If anything is off will fall through and fail below.
+			// If we detect they are semantically the same we nil them out
+			// to pass the check below.
+			if tmpOld.Users != nil || tmpNew.Users != nil {
+				if len(tmpOld.Users) == len(tmpNew.Users) {
+					oua := make(map[string]*User, len(tmpOld.Users))
+					nua := make(map[string]*User, len(tmpOld.Users))
+					for _, u := range tmpOld.Users {
+						oua[u.Username] = u
+					}
+					for _, u := range tmpNew.Users {
+						nua[u.Username] = u
+					}
+					same := true
+					for uname, u := range oua {
+						// If we can not find new one with same name, drop through to fail.
+						nu, ok := nua[uname]
+						if !ok {
+							same = false
+							break
+						}
+						// If username or password or account different break.
+						if u.Username != nu.Username || u.Password != nu.Password || u.Account.GetName() != nu.Account.GetName() {
+							same = false
+							break
+						}
+					}
+					// We can nil out here.
+					if same {
+						tmpOld.Users, tmpNew.Users = nil, nil
+					}
+				}
+			}
+
 			// If there is really a change prevents reload.
 			if !reflect.DeepEqual(tmpOld, tmpNew) {
 				// See TODO(ik) note below about printing old/new values.
@@ -1208,12 +1283,17 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 			diffOpts = append(diffOpts, &mqttStreamReplicasReload{newValue: newValue.(MQTTOpts).StreamReplicas})
 			diffOpts = append(diffOpts, &mqttConsumerReplicasReload{newValue: newValue.(MQTTOpts).ConsumerReplicas})
 			diffOpts = append(diffOpts, &mqttConsumerMemoryStorageReload{newValue: newValue.(MQTTOpts).ConsumerMemoryStorage})
+			diffOpts = append(diffOpts, &mqttInactiveThresholdReload{newValue: newValue.(MQTTOpts).ConsumerInactiveThreshold})
+
 			// Nil out/set to 0 the options that we allow to be reloaded so that
 			// we only fail reload if some that we don't support are changed.
 			tmpOld := oldValue.(MQTTOpts)
 			tmpNew := newValue.(MQTTOpts)
 			tmpOld.TLSConfig, tmpOld.AckWait, tmpOld.MaxAckPending, tmpOld.StreamReplicas, tmpOld.ConsumerReplicas, tmpOld.ConsumerMemoryStorage = nil, 0, 0, 0, 0, false
+			tmpOld.ConsumerInactiveThreshold = 0
 			tmpNew.TLSConfig, tmpNew.AckWait, tmpNew.MaxAckPending, tmpNew.StreamReplicas, tmpNew.ConsumerReplicas, tmpNew.ConsumerMemoryStorage = nil, 0, 0, 0, 0, false
+			tmpNew.ConsumerInactiveThreshold = 0
+
 			if !reflect.DeepEqual(tmpOld, tmpNew) {
 				// See TODO(ik) note below about printing old/new values.
 				return nil, fmt.Errorf("config reload not supported for %s: old=%v, new=%v",
@@ -1224,6 +1304,7 @@ func (s *Server) diffOptions(newOpts *Options) ([]option, error) {
 			tmpNew.StreamReplicas = newValue.(MQTTOpts).StreamReplicas
 			tmpNew.ConsumerReplicas = newValue.(MQTTOpts).ConsumerReplicas
 			tmpNew.ConsumerMemoryStorage = newValue.(MQTTOpts).ConsumerMemoryStorage
+			tmpNew.ConsumerInactiveThreshold = newValue.(MQTTOpts).ConsumerInactiveThreshold
 		case "connecterrorreports":
 			diffOpts = append(diffOpts, &connectErrorReports{newValue: newValue.(int)})
 		case "reconnecterrorreports":
@@ -1333,6 +1414,7 @@ func (s *Server) applyOptions(ctx *reloadContext, opts []option) {
 		reloadJetstream    = false
 		jsEnabled          = false
 		reloadTLS          = false
+		isStatszChange     = false
 	)
 	for _, opt := range opts {
 		opt.Apply(s)
@@ -1355,6 +1437,9 @@ func (s *Server) applyOptions(ctx *reloadContext, opts []option) {
 			reloadJetstream = true
 			jsEnabled = opt.(*jetStreamOption).newValue
 		}
+		if opt.IsStatszChange() {
+			isStatszChange = true
+		}
 	}
 
 	if reloadLogging {
@@ -1380,6 +1465,8 @@ func (s *Server) applyOptions(ctx *reloadContext, opts []option) {
 		}
 		// Make sure to reset the internal loop's version of JS.
 		s.resetInternalLoopInfo()
+	}
+	if isStatszChange {
 		s.sendStatszUpdate()
 	}
 
@@ -1519,7 +1606,12 @@ func (s *Server) reloadAuthorization() {
 				newAcc.lleafs = append([]*client(nil), acc.lleafs...)
 
 				newAcc.sl = acc.sl
-				newAcc.rm = acc.rm
+				if acc.rm != nil {
+					newAcc.rm = make(map[string]int32)
+				}
+				for k, v := range acc.rm {
+					newAcc.rm[k] = v
+				}
 				// Transfer internal client state. The configureAccounts call from above may have set up a new one.
 				// We need to use the old one, and the isid to not confuse internal subs.
 				newAcc.ic, newAcc.isid = acc.ic, acc.isid
diff --git a/server/server.go b/server/server.go
index 09fc9c92..e9a3e879 100644
--- a/server/server.go
+++ b/server/server.go
@@ -112,7 +112,7 @@ type Server struct {
 	// How often user logon fails due to the issuer account not being pinned.
 	pinnedAccFail uint64
 	stats
-	mu                  sync.Mutex
+	mu                  sync.RWMutex
 	kp                  nkeys.KeyPair
 	prand               *rand.Rand
 	info                Info
@@ -571,9 +571,9 @@ func (s *Server) logRejectedTLSConns() {
 
 // clusterName returns our cluster name which could be dynamic.
 func (s *Server) ClusterName() string {
-	s.mu.Lock()
+	s.mu.RLock()
 	cn := s.info.Cluster
-	s.mu.Unlock()
+	s.mu.RUnlock()
 	return cn
 }
 
@@ -726,9 +726,9 @@ func (s *Server) setOpts(opts *Options) {
 }
 
 func (s *Server) globalAccount() *Account {
-	s.mu.Lock()
+	s.mu.RLock()
 	gacc := s.gacc
-	s.mu.Unlock()
+	s.mu.RUnlock()
 	return gacc
 }
 
@@ -936,7 +936,7 @@ func (s *Server) globalAccountOnly() bool {
 		return false
 	}
 
-	s.mu.Lock()
+	s.mu.RLock()
 	s.accounts.Range(func(k, v interface{}) bool {
 		acc := v.(*Account)
 		// Ignore global and system
@@ -946,7 +946,7 @@ func (s *Server) globalAccountOnly() bool {
 		hasOthers = true
 		return false
 	})
-	s.mu.Unlock()
+	s.mu.RUnlock()
 
 	return !hasOthers
 }
@@ -976,10 +976,10 @@ func (s *Server) ActivePeers() (peers []string) {
 // isTrustedIssuer will check that the issuer is a trusted public key.
 // This is used to make sure an account was signed by a trusted operator.
 func (s *Server) isTrustedIssuer(issuer string) bool {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	// If we are not running in trusted mode and there is no issuer, that is ok.
-	if s.trustedKeys == nil && issuer == "" {
+	if s.trustedKeys == nil && issuer == _EMPTY_ {
 		return true
 	}
 	for _, tk := range s.trustedKeys {
@@ -1085,9 +1085,9 @@ func (s *Server) Running() bool {
 
 // Protected check on running state
 func (s *Server) isRunning() bool {
-	s.mu.Lock()
+	s.mu.RLock()
 	running := s.running
-	s.mu.Unlock()
+	s.mu.RUnlock()
 	return running
 }
 
@@ -1121,12 +1121,12 @@ func (s *Server) decActiveAccounts() {
 // range over all accounts in the sync.Map to count.
 func (s *Server) numAccounts() int {
 	count := 0
-	s.mu.Lock()
+	s.mu.RLock()
 	s.accounts.Range(func(k, v interface{}) bool {
 		count++
 		return true
 	})
-	s.mu.Unlock()
+	s.mu.RUnlock()
 	return count
 }
 
@@ -1187,19 +1187,19 @@ func (s *Server) SetSystemAccount(accName string) error {
 // SystemAccount returns the system account if set.
 func (s *Server) SystemAccount() *Account {
 	var sacc *Account
-	s.mu.Lock()
+	s.mu.RLock()
 	if s.sys != nil {
 		sacc = s.sys.account
 	}
-	s.mu.Unlock()
+	s.mu.RUnlock()
 	return sacc
 }
 
 // GlobalAccount returns the global account.
 // Default clients will use the global account.
 func (s *Server) GlobalAccount() *Account {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.gacc
 }
 
@@ -2801,30 +2801,30 @@ func (s *Server) addToTempClients(cid uint64, c *client) bool {
 
 // NumRoutes will report the number of registered routes.
 func (s *Server) NumRoutes() int {
-	s.mu.Lock()
+	s.mu.RLock()
 	nr := len(s.routes)
-	s.mu.Unlock()
+	s.mu.RUnlock()
 	return nr
 }
 
 // NumRemotes will report number of registered remotes.
 func (s *Server) NumRemotes() int {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return len(s.remotes)
 }
 
 // NumLeafNodes will report number of leaf node connections.
 func (s *Server) NumLeafNodes() int {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return len(s.leafs)
 }
 
 // NumClients will report the number of registered clients.
 func (s *Server) NumClients() int {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return len(s.clients)
 }
 
@@ -2835,22 +2835,22 @@ func (s *Server) GetClient(cid uint64) *client {
 
 // getClient will return the client associated with cid.
 func (s *Server) getClient(cid uint64) *client {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.clients[cid]
 }
 
 // GetLeafNode returns the leafnode associated with the cid.
 func (s *Server) GetLeafNode(cid uint64) *client {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.leafs[cid]
 }
 
 // NumSubscriptions will report how many subscriptions are active.
 func (s *Server) NumSubscriptions() uint32 {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.numSubscriptions()
 }
 
@@ -2875,15 +2875,15 @@ func (s *Server) NumSlowConsumers() int64 {
 
 // ConfigTime will report the last time the server configuration was loaded.
 func (s *Server) ConfigTime() time.Time {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.configTime
 }
 
 // Addr will return the net.Addr object for the current listener.
 func (s *Server) Addr() net.Addr {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	if s.listener == nil {
 		return nil
 	}
@@ -2892,8 +2892,8 @@ func (s *Server) Addr() net.Addr {
 
 // MonitorAddr will return the net.Addr object for the monitoring listener.
 func (s *Server) MonitorAddr() *net.TCPAddr {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	if s.http == nil {
 		return nil
 	}
@@ -2902,8 +2902,8 @@ func (s *Server) MonitorAddr() *net.TCPAddr {
 
 // ClusterAddr returns the net.Addr object for the route listener.
 func (s *Server) ClusterAddr() *net.TCPAddr {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	if s.routeListener == nil {
 		return nil
 	}
@@ -2912,8 +2912,8 @@ func (s *Server) ClusterAddr() *net.TCPAddr {
 
 // ProfilerAddr returns the net.Addr object for the profiler listener.
 func (s *Server) ProfilerAddr() *net.TCPAddr {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	if s.profiler == nil {
 		return nil
 	}
@@ -2932,14 +2932,14 @@ func (s *Server) readyForConnections(d time.Duration) error {
 
 	end := time.Now().Add(d)
 	for time.Now().Before(end) {
-		s.mu.Lock()
+		s.mu.RLock()
 		chk["server"] = info{ok: s.listener != nil, err: s.listenerErr}
 		chk["route"] = info{ok: (opts.Cluster.Port == 0 || s.routeListener != nil), err: s.routeListenerErr}
 		chk["gateway"] = info{ok: (opts.Gateway.Name == _EMPTY_ || s.gatewayListener != nil), err: s.gatewayListenerErr}
 		chk["leafNode"] = info{ok: (opts.LeafNode.Port == 0 || s.leafNodeListener != nil), err: s.leafNodeListenerErr}
 		chk["websocket"] = info{ok: (opts.Websocket.Port == 0 || s.websocket.listener != nil), err: s.websocket.listenerErr}
 		chk["mqtt"] = info{ok: (opts.MQTT.Port == 0 || s.mqtt.listener != nil), err: s.mqtt.listenerErr}
-		s.mu.Unlock()
+		s.mu.RUnlock()
 
 		var numOK int
 		for _, inf := range chk {
@@ -3021,20 +3021,20 @@ func (s *Server) startGoRoutine(f func()) bool {
 }
 
 func (s *Server) numClosedConns() int {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.closed.len()
 }
 
 func (s *Server) totalClosedConns() uint64 {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.closed.totalConns()
 }
 
 func (s *Server) closedClients() []*closedClient {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.closed.closedClients()
 }
 
@@ -3186,7 +3186,7 @@ func (s *Server) PortsInfo(maxWait time.Duration) *Ports {
 	if s.readyForListeners(maxWait) {
 		opts := s.getOpts()
 
-		s.mu.Lock()
+		s.mu.RLock()
 		tls := s.info.TLSRequired
 		listener := s.listener
 		httpListener := s.http
@@ -3194,7 +3194,7 @@ func (s *Server) PortsInfo(maxWait time.Duration) *Ports {
 		profileListener := s.profiler
 		wsListener := s.websocket.listener
 		wss := s.websocket.tls
-		s.mu.Unlock()
+		s.mu.RUnlock()
 
 		ports := Ports{}
 
@@ -3295,9 +3295,9 @@ func (s *Server) logPorts() {
 func (s *Server) readyForListeners(dur time.Duration) bool {
 	end := time.Now().Add(dur)
 	for time.Now().Before(end) {
-		s.mu.Lock()
+		s.mu.RLock()
 		listeners := s.serviceListeners()
-		s.mu.Unlock()
+		s.mu.RUnlock()
 		if len(listeners) == 0 {
 			return false
 		}
@@ -3345,8 +3345,8 @@ func (s *Server) serviceListeners() []net.Listener {
 
 // Returns true if in lame duck mode.
 func (s *Server) isLameDuckMode() bool {
-	s.mu.Lock()
-	defer s.mu.Unlock()
+	s.mu.RLock()
+	defer s.mu.RUnlock()
 	return s.ldm
 }
 
@@ -3529,6 +3529,7 @@ func (s *Server) acceptError(acceptName string, err error, tmpDelay time.Duratio
 	if !s.isRunning() {
 		return -1
 	}
+	//lint:ignore SA1019 We want to retry on a bunch of errors here.
 	if ne, ok := err.(net.Error); ok && ne.Temporary() {
 		s.Errorf("Temporary %s Accept Error(%v), sleeping %dms", acceptName, ne, tmpDelay/time.Millisecond)
 		select {
diff --git a/server/store.go b/server/store.go
index ed5786eb..36d57627 100644
--- a/server/store.go
+++ b/server/store.go
@@ -372,11 +372,11 @@ const (
 func (st StorageType) String() string {
 	switch st {
 	case MemoryStorage:
-		return strings.Title(memoryStorageString)
+		return "Memory"
 	case FileStorage:
-		return strings.Title(fileStorageString)
+		return "File"
 	case AnyStorage:
-		return strings.Title(anyStorageString)
+		return "Any"
 	default:
 		return "Unknown Storage Type"
 	}
diff --git a/server/stream.go b/server/stream.go
index 40801312..f243ebf6 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -4041,6 +4041,7 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 		mset.outq.unregister()
 	}
 
+	store := mset.store
 	// Clustered cleanup.
 	mset.mu.Unlock()
 
@@ -4050,16 +4051,16 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 		sysc.closeConnection(ClientClosed)
 	}
 
-	if mset.store == nil {
+	if store == nil {
 		return nil
 	}
 
 	if deleteFlag {
-		if err := mset.store.Delete(); err != nil {
+		if err := store.Delete(); err != nil {
 			return err
 		}
 		js.releaseStreamResources(&mset.cfg)
-	} else if err := mset.store.Stop(); err != nil {
+	} else if err := store.Stop(); err != nil {
 		return err
 	}
 


Test patch:
diff --git a/runTestsOnTravis.sh b/scripts/runTestsOnTravis.sh
similarity index 86%
rename from runTestsOnTravis.sh
rename to scripts/runTestsOnTravis.sh
index 0fb4e450..8f42cc62 100755
--- a/runTestsOnTravis.sh
+++ b/scripts/runTestsOnTravis.sh
@@ -6,7 +6,7 @@ if [ "$1" = "compile" ]; then
 
     # We will compile and run some vet, spelling and some other checks.
 
-    go install honnef.co/go/tools/cmd/staticcheck@v0.2.2;
+    go install honnef.co/go/tools/cmd/staticcheck@latest;
     go install github.com/client9/misspell/cmd/misspell@latest;
     GO_LIST=$(go list ./...);
     go build;
@@ -15,14 +15,19 @@ if [ "$1" = "compile" ]; then
     find . -type f -name "*.go" | xargs misspell -error -locale US;
     staticcheck $GO_LIST
     if [ "$TRAVIS_TAG" != "" ]; then
+	export GODEBUG="x509sha1=1"
         go test -race -v -run=TestVersionMatchesTag ./server -count=1 -vet=off
     fi
 
+elif [ "$1" = "build_only" ]; then
+    go build;
+
 elif [ "$1" = "no_race_tests" ]; then
 
     # Run tests without the `-race` flag. By convention, those tests start
     # with `TestNoRace`.
 
+    export GODEBUG="x509sha1=1"
     go test -v -p=1 -run=TestNoRace ./... -count=1 -vet=off -timeout=30m -failfast
 
 elif [ "$1" = "js_tests" ]; then
@@ -32,6 +37,7 @@ elif [ "$1" = "js_tests" ]; then
     # tests by using the `skip_js_cluster_tests` and `skip_js_super_cluster_tests`
     # build tags.
 
+    export GODEBUG="x509sha1=1"
     go test -race -v -run=TestJetStream ./server -tags=skip_js_cluster_tests,skip_js_super_cluster_tests -count=1 -vet=off -timeout=30m -failfast
 
 elif [ "$1" = "js_cluster_tests" ]; then
@@ -39,6 +45,7 @@ elif [ "$1" = "js_cluster_tests" ]; then
     # Run JetStream clustered tests. By convention, all JS cluster tests
     # start with `TestJetStreamCluster`.
 
+    export GODEBUG="x509sha1=1"
     go test -race -v -run=TestJetStreamCluster ./server -count=1 -vet=off -timeout=30m -failfast
 
 elif [ "$1" = "js_super_cluster_tests" ]; then
@@ -46,6 +53,7 @@ elif [ "$1" = "js_super_cluster_tests" ]; then
     # Run JetStream super clustered tests. By convention, all JS super cluster
     # tests with `TestJetStreamSuperCluster`.
 
+    export GODEBUG="x509sha1=1"
     go test -race -v -run=TestJetStreamSuperCluster ./server -count=1 -vet=off -timeout=30m -failfast
 
 elif [ "$1" = "srv_pkg_non_js_tests" ]; then
@@ -53,12 +61,14 @@ elif [ "$1" = "srv_pkg_non_js_tests" ]; then
     # Run all non JetStream tests in the server package. We exclude the
     # JS tests by using the `skip_js_tests` build tag.
 
+    export GODEBUG="x509sha1=1"
     go test -race -v -p=1 ./server/... -tags=skip_js_tests -count=1 -vet=off -timeout=30m -failfast
 
 elif [ "$1" = "non_srv_pkg_tests" ]; then
 
     # Run all tests of all non server package.
 
+    export GODEBUG="x509sha1=1"
     go test -race -v -p=1 $(go list ./... | grep -v "/server") -count=1 -vet=off -timeout=30m -failfast
 
 fi
diff --git a/server/events_test.go b/server/events_test.go
index 987510f6..5445808e 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -1721,6 +1721,51 @@ func TestSystemAccountNoAuthUser(t *testing.T) {
 	}
 }
 
+func TestServerAccountConns(t *testing.T) {
+	// speed up hb
+	eventsHBInterval = time.Millisecond * 100
+	conf := createConfFile(t, []byte(`
+	   host: 127.0.0.1
+	   port: -1
+	   system_account: SYS
+	   accounts: {
+			   SYS: {users: [{user: s, password: s}]}
+			   ACC: {users: [{user: a, password: a}]}
+	   }`))
+	defer removeFile(t, conf)
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	nc := natsConnect(t, s.ClientURL(), nats.UserInfo("a", "a"))
+	defer nc.Close()
+
+	subOut, err := nc.SubscribeSync("foo")
+	require_NoError(t, err)
+	hw := "HELLO WORLD"
+	nc.Publish("foo", []byte(hw))
+	nc.Publish("bar", []byte(hw)) // will only count towards received
+	nc.Flush()
+	m, err := subOut.NextMsg(time.Second)
+	require_NoError(t, err)
+	require_Equal(t, string(m.Data), hw)
+
+	ncs := natsConnect(t, s.ClientURL(), nats.UserInfo("s", "s"))
+	defer ncs.Close()
+	subs, err := ncs.SubscribeSync("$SYS.ACCOUNT.ACC.SERVER.CONNS")
+	require_NoError(t, err)
+
+	m, err = subs.NextMsg(time.Second)
+	require_NoError(t, err)
+	accConns := &AccountNumConns{}
+	err = json.Unmarshal(m.Data, accConns)
+	require_NoError(t, err)
+
+	require_True(t, accConns.Received.Msgs == 2)
+	require_True(t, accConns.Received.Bytes == 2*int64(len(hw)))
+	require_True(t, accConns.Sent.Msgs == 1)
+	require_True(t, accConns.Sent.Bytes == int64(len(hw)))
+}
+
 func TestServerEventsStatsZ(t *testing.T) {
 	serverStatsReqSubj := "$SYS.REQ.SERVER.%s.STATSZ"
 	preStart := time.Now().UTC()
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index fe41ed9f..e57f1693 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -3599,6 +3599,258 @@ func TestJetStreamClusterPeerRemovalAndStreamReassignmentWithoutSpace(t *testing
 	streamCurrent(2)
 }
 
+func TestJetStreamClusterPeerExclusionTag(t *testing.T) {
+	c := createJetStreamClusterWithTemplateAndModHook(t, jsClusterTempl, "C", 3,
+		func(serverName, clusterName, storeDir, conf string) string {
+			switch serverName {
+			case "S-1":
+				return fmt.Sprintf("%s\nserver_tags: [server:%s, intersect, %s]", conf, serverName, jsExcludePlacement)
+			case "S-2":
+				return fmt.Sprintf("%s\nserver_tags: [server:%s, intersect]", conf, serverName)
+			default:
+				return fmt.Sprintf("%s\nserver_tags: [server:%s]", conf, serverName)
+			}
+		})
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	for i, c := range []nats.StreamConfig{
+		{Replicas: 1, Placement: &nats.Placement{Tags: []string{"server:S-1"}}},
+		{Replicas: 2, Placement: &nats.Placement{Tags: []string{"intersect"}}},
+		{Replicas: 3}, // not enough server without !jetstream
+	} {
+		c.Name = fmt.Sprintf("TEST%d", i)
+		c.Subjects = []string{c.Name}
+		_, err := js.AddStream(&c)
+		require_Error(t, err)
+		require_Contains(t, err.Error(), "insufficient resources")
+	}
+
+	// Test update failure
+	cfg := &nats.StreamConfig{Name: "TEST", Subjects: []string{"foo"}, Replicas: 2}
+	_, err := js.AddStream(cfg)
+	require_NoError(t, err)
+
+	cfg.Replicas = 3
+	_, err = js.UpdateStream(cfg)
+	require_Error(t, err)
+	require_Contains(t, err.Error(), "insufficient resources")
+
+	// Test tag reload removing !jetstream tag, and allowing placement again
+
+	srv := c.serverByName("S-1")
+
+	v, err := srv.Varz(nil)
+	require_NoError(t, err)
+	require_True(t, v.Tags.Contains(jsExcludePlacement))
+	content, err := os.ReadFile(srv.configFile)
+	require_NoError(t, err)
+	newContent := strings.ReplaceAll(string(content), fmt.Sprintf(", %s]", jsExcludePlacement), "]")
+	changeCurrentConfigContentWithNewContent(t, srv.configFile, []byte(newContent))
+
+	ncSys := natsConnect(t, c.randomServer().ClientURL(), nats.UserInfo("admin", "s3cr3t!"))
+	sub, err := ncSys.SubscribeSync(fmt.Sprintf("$SYS.SERVER.%s.STATSZ", srv.ID()))
+	require_NoError(t, err)
+
+	require_NoError(t, srv.Reload())
+	v, err = srv.Varz(nil)
+	require_NoError(t, err)
+	require_True(t, !v.Tags.Contains(jsExcludePlacement))
+
+	// it is possible that sub already received a stasz message prior to reload, retry once
+	cmp := false
+	for i := 0; i < 2 && !cmp; i++ {
+		m, err := sub.NextMsg(time.Second)
+		require_NoError(t, err)
+		cmp = strings.Contains(string(m.Data), `"tags":["server:s-1","intersect"]`)
+	}
+	require_True(t, cmp)
+
+	cfg.Replicas = 3
+	_, err = js.UpdateStream(cfg)
+	require_NoError(t, err)
+}
+
+func TestJetStreamClusterPeerEvacuationAndStreamReassignment(t *testing.T) {
+	s := createJetStreamSuperClusterWithTemplateAndModHook(t, jsClusterTempl, 4, 2,
+		func(serverName, clusterName, storeDir, conf string) string {
+			return fmt.Sprintf("%s\nserver_tags: [cluster:%s, server:%s]", conf, clusterName, serverName)
+		})
+	defer s.shutdown()
+
+	c := s.clusterForName("C1")
+
+	// Client based API
+	srv := c.randomNonLeader()
+	nc, js := jsClientConnect(t, srv)
+	defer nc.Close()
+
+	test := func(r int, moveTags []string, targetCluster string, testMigrateTo bool) {
+		si, err := js.AddStream(&nats.StreamConfig{
+			Name:     "TEST",
+			Subjects: []string{"foo"},
+			Replicas: r,
+		})
+		require_NoError(t, err)
+		defer js.DeleteStream("TEST")
+		toMoveFrom := si.Cluster.Leader
+
+		_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{Durable: "DUR", AckPolicy: nats.AckExplicitPolicy})
+		require_NoError(t, err)
+
+		sub, err := js.SubscribeSync("foo")
+		require_NoError(t, err)
+
+		for i := 0; i < 100; i++ {
+			_, err = js.Publish("foo", nil)
+			require_NoError(t, err)
+		}
+
+		sEmpty := c.serverByName(toMoveFrom)
+		jszBefore, err := sEmpty.Jsz(nil)
+		require_NoError(t, err)
+		require_True(t, jszBefore.Streams == 1)
+		require_True(t, jszBefore.Consumers >= 1)
+		require_True(t, jszBefore.Store != 0)
+
+		migrateToServer := _EMPTY_
+		if testMigrateTo {
+			// find an empty server
+			for _, s := range c.servers {
+				name := s.Name()
+				found := si.Cluster.Leader == name
+				if !found {
+					for _, r := range si.Cluster.Replicas {
+						if r.Name == name {
+							found = true
+							break
+						}
+					}
+				}
+				if !found {
+					migrateToServer = name
+					break
+				}
+			}
+			jszAfter, err := c.serverByName(migrateToServer).Jsz(nil)
+			require_NoError(t, err)
+			require_True(t, jszAfter.Streams == 0)
+
+			moveTags = append(moveTags, fmt.Sprintf("server:%s", migrateToServer))
+		}
+
+		ncsys, err := nats.Connect(srv.ClientURL(), nats.UserInfo("admin", "s3cr3t!"))
+		require_NoError(t, err)
+		defer ncsys.Close()
+		/*
+			infoReq, err := json.Marshal(&JSApiMetaServerStreamInfoRequest{Server: toMoveFrom})
+			require_NoError(t, err)
+			rmsg, err := ncsys.Request(JSApiServerStreamInfo, infoReq, time.Second)
+			require_NoError(t, err)
+			var infoResp JSApiMetaServerStreamInfoResponse
+			require_NoError(t, json.Unmarshal(rmsg.Data, &infoResp))
+			require_True(t, infoResp.Error == nil)
+
+			found := false
+			for _, sCfg := range infoResp.Content["$G"] {
+				if sCfg.Name == "TEST" {
+					found = true
+					break
+				}
+			}
+			require_True(t, found)
+		*/
+		moveReq, err := json.Marshal(&JSApiMetaServerStreamMoveRequest{
+			Server: toMoveFrom, Account: "$G", Stream: "TEST", Tags: moveTags})
+		require_NoError(t, err)
+		rmsg, err := ncsys.Request(JSApiServerStreamMove, moveReq, 100*time.Second)
+		require_NoError(t, err)
+		var moveResp JSApiStreamUpdateResponse
+		require_NoError(t, json.Unmarshal(rmsg.Data, &moveResp))
+		require_True(t, moveResp.Error == nil)
+
+		// test draining
+		checkFor(t, 20*time.Second, 1000*time.Millisecond, func() error {
+			jszAfter, err := sEmpty.Jsz(nil)
+			if err != nil {
+				return fmt.Errorf("could not fetch JS info for server: %v", err)
+			}
+			if jszAfter.Streams != 0 {
+				return fmt.Errorf("empty server still has %d streams", jszAfter.Streams)
+			}
+			if jszAfter.Consumers != 0 {
+				return fmt.Errorf("empty server still has %d consumers", jszAfter.Consumers)
+			}
+			if jszAfter.Store != 0 {
+				return fmt.Errorf("empty server still has %d storage", jszAfter.Store)
+			}
+			return nil
+		})
+		// test move to particular server
+		if testMigrateTo {
+			toSrv := c.serverByName(migrateToServer)
+			checkFor(t, 20*time.Second, 1000*time.Millisecond, func() error {
+				jszAfter, err := toSrv.Jsz(nil)
+				if err != nil {
+					return fmt.Errorf("could not fetch JS info for server: %v", err)
+				}
+				if jszAfter.Streams != 1 {
+					return fmt.Errorf("server expected to have one stream, has %d", jszAfter.Streams)
+				}
+				return nil
+			})
+		}
+		// Now wait until the stream is now current.
+		checkFor(t, 20*time.Second, 100*time.Millisecond, func() error {
+			si, err := js.StreamInfo("TEST", nats.MaxWait(time.Second))
+			if err != nil {
+				return fmt.Errorf("could not fetch stream info: %v", err)
+			}
+			if si.Cluster.Leader == toMoveFrom {
+				return fmt.Errorf("peer not removed yet: %+v", toMoveFrom)
+			}
+			if len(si.Cluster.Replicas) != r-1 {
+				return fmt.Errorf("not yet downsized replica should be empty has: %d", len(si.Cluster.Replicas))
+			}
+			if si.Config.Replicas != r {
+				return fmt.Errorf("bad replica count %d", si.Config.Replicas)
+			}
+			if si.Cluster.Name != targetCluster {
+				return fmt.Errorf("stream expected in %s but found in %s", si.Cluster.Name, targetCluster)
+			}
+			sNew := s.serverByName(si.Cluster.Leader)
+			if jszNew, err := sNew.Jsz(nil); err != nil {
+				return err
+			} else if jszNew.Streams != 1 {
+				return fmt.Errorf("new leader has %d streams, not one", jszNew.Streams)
+			} else if jszNew.Store != jszBefore.Store {
+				return fmt.Errorf("new leader has %d storage, should have %d", jszNew.Store, jszBefore.Store)
+			}
+			return nil
+		})
+		// consume messages from ephemeral consumer
+		for i := 0; i < 100; i++ {
+			_, err := sub.NextMsg(time.Second)
+			require_NoError(t, err)
+		}
+	}
+
+	for i := 1; i <= 3; i++ {
+		t.Run(fmt.Sprintf("r%d", i), func(t *testing.T) {
+			test(i, nil, "C1", false)
+		})
+		t.Run(fmt.Sprintf("r%d-explicit", i), func(t *testing.T) {
+			test(i, nil, "C1", true)
+		})
+	}
+
+	t.Run("r3-cluster-move", func(t *testing.T) {
+		test(3, []string{"cluster:C2"}, "C2", false)
+	})
+}
+
 func TestJetStreamClusterPeerOffline(t *testing.T) {
 	c := createJetStreamClusterExplicit(t, "R5S", 5)
 	defer c.shutdown()
@@ -8622,11 +8874,6 @@ func TestJetStreamClusterConsumerUpdates(t *testing.T) {
 		require_NoError(t, err)
 
 		// These all should fail.
-		ncfg = *cfg
-		ncfg.FilterSubject = "bar"
-		_, err = js.AddConsumer("TEST", &ncfg)
-		require_Error(t, err)
-
 		ncfg = *cfg
 		ncfg.DeliverPolicy = nats.DeliverLastPolicy
 		_, err = js.AddConsumer("TEST", &ncfg)
@@ -10875,3 +11122,50 @@ func TestJetStreamClusterNoRestartAdvisories(t *testing.T) {
 
 	checkSubsPending(t, sub, 0)
 }
+
+func TestJetStreamClusterR1StreamPlacementNoReservation(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "JSC", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	sp := make(map[string]int)
+	for i := 0; i < 100; i++ {
+		sname := fmt.Sprintf("T-%d", i)
+		_, err := js.AddStream(&nats.StreamConfig{
+			Name: sname,
+		})
+		require_NoError(t, err)
+		sp[c.streamLeader("$G", sname).Name()]++
+	}
+
+	for serverName, num := range sp {
+		if num > 60 {
+			t.Fatalf("Streams not distributed, expected ~30-35 but got %d for server %q", num, serverName)
+		}
+	}
+}
+
+func TestJetStreamClusterConsumerAndStreamNamesWithPathSeparators(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "JSC", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "usr/bin"})
+	require_Error(t, err, NewJSStreamNameContainsPathSeparatorsError(), nats.ErrInvalidStreamName)
+	_, err = js.AddStream(&nats.StreamConfig{Name: `Documents\readme.txt`})
+	require_Error(t, err, NewJSStreamNameContainsPathSeparatorsError(), nats.ErrInvalidStreamName)
+
+	// Now consumers.
+	_, err = js.AddStream(&nats.StreamConfig{Name: "T"})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("T", &nats.ConsumerConfig{Durable: "a/b", AckPolicy: nats.AckExplicitPolicy})
+	require_Error(t, err, NewJSConsumerNameContainsPathSeparatorsError(), nats.ErrInvalidConsumerName)
+
+	_, err = js.AddConsumer("T", &nats.ConsumerConfig{Durable: `a\b`, AckPolicy: nats.AckExplicitPolicy})
+	require_Error(t, err, NewJSConsumerNameContainsPathSeparatorsError(), nats.ErrInvalidConsumerName)
+}
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index d5905099..eac38846 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -13573,12 +13573,8 @@ func TestJetStreamLongStreamNamesAndPubAck(t *testing.T) {
 	nc, js := jsClientConnect(t, s)
 	defer nc.Close()
 
-	data := make([]byte, 255)
-	rand.Read(data)
-	stream := base64.StdEncoding.EncodeToString(data)[:255]
-
 	cfg := &nats.StreamConfig{
-		Name:     stream,
+		Name:     strings.Repeat("ZABC", 256/4)[:255],
 		Subjects: []string{"foo"},
 	}
 	if _, err := js.AddStream(cfg); err != nil {
@@ -17911,6 +17907,97 @@ func TestJetStreamConsumerInactiveThreshold(t *testing.T) {
 	waitOnCleanup(ci)
 }
 
+func TestJetStreamConsumerAndStreamNamesWithPathSeparators(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "usr/bin"})
+	require_Error(t, err, NewJSStreamNameContainsPathSeparatorsError(), nats.ErrInvalidStreamName)
+	_, err = js.AddStream(&nats.StreamConfig{Name: `Documents\readme.txt`})
+	require_Error(t, err, NewJSStreamNameContainsPathSeparatorsError(), nats.ErrInvalidStreamName)
+
+	// Now consumers.
+	_, err = js.AddStream(&nats.StreamConfig{Name: "T"})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("T", &nats.ConsumerConfig{Durable: "a/b", AckPolicy: nats.AckExplicitPolicy})
+	require_Error(t, err, NewJSConsumerNameContainsPathSeparatorsError(), nats.ErrInvalidConsumerName)
+
+	_, err = js.AddConsumer("T", &nats.ConsumerConfig{Durable: `a\b`, AckPolicy: nats.AckExplicitPolicy})
+	require_Error(t, err, NewJSConsumerNameContainsPathSeparatorsError(), nats.ErrInvalidConsumerName)
+}
+
+func TestJetStreamConsumerUpdateFilterSubject(t *testing.T) {
+	s := RunBasicJetStreamServer()
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "T", Subjects: []string{"foo", "bar", "baz"}})
+	require_NoError(t, err)
+
+	// 10 foo
+	for i := 0; i < 10; i++ {
+		js.PublishAsync("foo", []byte("OK"))
+	}
+	// 20 bar
+	for i := 0; i < 20; i++ {
+		js.PublishAsync("bar", []byte("OK"))
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	sub, err := js.PullSubscribe("foo", "d")
+	require_NoError(t, err)
+
+	// Consume 5 msgs
+	msgs, err := sub.Fetch(5)
+	require_NoError(t, err)
+	require_True(t, len(msgs) == 5)
+
+	// Now update to different filter subject.
+	_, err = js.UpdateConsumer("T", &nats.ConsumerConfig{
+		Durable:       "d",
+		FilterSubject: "bar",
+		AckPolicy:     nats.AckExplicitPolicy,
+	})
+	require_NoError(t, err)
+
+	sub, err = js.PullSubscribe("bar", "d")
+	require_NoError(t, err)
+
+	msgs, err = sub.Fetch(1)
+	require_NoError(t, err)
+
+	// Make sure meta and pending etc are all correct.
+	m := msgs[0]
+	meta, err := m.Metadata()
+	require_NoError(t, err)
+
+	if meta.Sequence.Consumer != 6 || meta.Sequence.Stream != 11 {
+		t.Fatalf("Sequence incorrect %+v", meta.Sequence)
+	}
+	if meta.NumDelivered != 1 {
+		t.Fatalf("Expected NumDelivered to be 1, got %d", meta.NumDelivered)
+	}
+	if meta.NumPending != 19 {
+		t.Fatalf("Expected NumPending to be 19, got %d", meta.NumPending)
+	}
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Simple JetStream Benchmarks
 ///////////////////////////////////////////////////////////////////////////
diff --git a/server/leafnode_test.go b/server/leafnode_test.go
index 381adc2f..33519bd1 100644
--- a/server/leafnode_test.go
+++ b/server/leafnode_test.go
@@ -2610,7 +2610,7 @@ func TestLeafNodeTLSConfigReload(t *testing.T) {
 	// Wait for the error
 	select {
 	case err := <-lg.errCh:
-		if !strings.Contains(err, "unknown") {
+		if !strings.Contains(err, "unknown") && !strings.Contains(err, "broken") {
 			t.Fatalf("Unexpected error: %v", err)
 		}
 	case <-time.After(2 * time.Second):
@@ -4349,3 +4349,33 @@ func TestLeafNodeStreamAndShadowSubs(t *testing.T) {
 	// Check again
 	subPubAndCheck()
 }
+
+func TestLeafNodeAuthConfigReload(t *testing.T) {
+	template := `
+		listen: 127.0.0.1:-1
+		accounts { test: {} }
+		leaf {
+			listen: "127.0.0.1:7422"
+			tls {
+				cert_file: "../test/configs/certs/server-cert.pem"
+				key_file:  "../test/configs/certs/server-key.pem"
+				ca_file:   "../test/configs/certs/ca.pem"
+			}
+			authorization {
+				# These are only fields allowed atm.
+				users = [ { user: test, password: "s3cret1", account: "test"  } ]
+			}
+		}
+	`
+	conf := createConfFile(t, []byte(template))
+	defer removeFile(t, conf)
+
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	lg := &captureErrorLogger{errCh: make(chan string, 10)}
+	s.SetLogger(lg, false, false)
+
+	// Reload here should work ok.
+	reloadUpdateConfig(t, s, conf, template)
+}
diff --git a/server/mqtt_test.go b/server/mqtt_test.go
index 465abcef..3913844a 100644
--- a/server/mqtt_test.go
+++ b/server/mqtt_test.go
@@ -6001,6 +6001,54 @@ func TestMQTTSessionNotDeletedOnDeleteConsumerError(t *testing.T) {
 	require_True(t, si.State.Msgs == 1)
 }
 
+// Test for auto-cleanup of consumers.
+func TestMQTTConsumerInactiveThreshold(t *testing.T) {
+	tmpl := `
+		listen: 127.0.0.1:-1
+		server_name: mqtt
+		jetstream: enabled
+
+		mqtt {
+			listen: 127.0.0.1:-1
+			consumer_inactive_threshold: %q
+		}
+
+		# For access to system account.
+		accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
+	`
+	conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, "0.2s")))
+	defer removeFile(t, conf)
+	s, o := RunServerWithConfig(conf)
+	defer testMQTTShutdownServer(s)
+
+	mc, r := testMQTTConnectRetry(t, &mqttConnInfo{clientID: "test", cleanSess: true}, o.MQTT.Host, o.MQTT.Port, 5)
+	defer mc.Close()
+	testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, false)
+	testMQTTSub(t, 1, mc, r, []*mqttFilter{{filter: "foo", qos: 1}}, []byte{1})
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	ci := <-js.ConsumersInfo("$MQTT_msgs")
+
+	// Make sure we clean up this consumer.
+	mc.Close()
+	checkFor(t, 2*time.Second, 50*time.Millisecond, func() error {
+		_, err := js.ConsumerInfo(ci.Stream, ci.Name)
+		if err == nil {
+			return fmt.Errorf("Consumer still present")
+		}
+		return nil
+	})
+
+	// Check reload.
+	// We will not redo existing consumers however.
+	reloadUpdateConfig(t, s, conf, fmt.Sprintf(tmpl, "22s"))
+	if opts := s.getOpts(); opts.MQTT.ConsumerInactiveThreshold != 22*time.Second {
+		t.Fatalf("Expected reloaded value of %v but got %v", 22*time.Second, opts.MQTT.ConsumerInactiveThreshold)
+	}
+}
+
 //////////////////////////////////////////////////////////////////////////
 //
 // Benchmarks