Repository: nats-io/nats-server
Pull request #5082 (2024-02-14 13:34:10)

Added tests for MQTT retained messages in various cluster/domain conf…
This is a Test- and CI-only PR. No server changes.

See also https://github.com/ConnectEverything/mqtt-test/pull/2.

levb: > Still need to make sure that the cluster is really restarted. 

@kozlovic sorry my commit message was non-descriptive, I squashed from another branch. Anyway, I am still working on the restart a another potential issue I ran into, will mark you for review when done.

Patch:
diff --git a/.travis.yml b/.travis.yml
index 3e2362c4..efd426ac 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -8,8 +8,8 @@ language: go
 go:
   # This should be quoted or use .x, but should not be unquoted.
   # Remember that a YAML bare float drops trailing zeroes.
-  - "1.22.3"
-  - "1.21.10"
+  - "1.22.4"
+  - "1.21.11"
 
 go_import_path: github.com/nats-io/nats-server
 
@@ -55,7 +55,7 @@ script: ./scripts/runTestsOnTravis.sh $TEST_SUITE
 deploy:
   provider: script
   cleanup: true
-  script: curl -sL http://git.io/goreleaser | bash
+  script: curl -o goreleaser.tar.gz -sLf https://github.com/goreleaser/goreleaser/releases/download/v1.26.2/goreleaser_Linux_x86_64.tar.gz && tar -xvf goreleaser.tar.gz && ./goreleaser
   on:
     tags: true
     condition: ($TRAVIS_GO_VERSION =~ 1.22) && ($TEST_SUITE = "compile")
diff --git a/DEPENDENCIES.md b/DEPENDENCIES.md
index d34a5cf7..66ee9f67 100644
--- a/DEPENDENCIES.md
+++ b/DEPENDENCIES.md
@@ -6,16 +6,14 @@ This file lists the dependencies used in this repository.
 |-|-|
 | Go | BSD 3-Clause "New" or "Revised" License |
 | github.com/nats-io/nats-server/v2 | Apache License 2.0 |
-| github.com/golang/protobuf v1.4.2 | BSD 3-Clause "New" or "Revised" License |
-| github.com/nats-io/jwt/v2 v2.5.2 | Apache License 2.0 |
-| github.com/nats-io/nats.go v1.30.2 | Apache License 2.0 |
-| github.com/nats-io/nkeys v0.4.5 | Apache License 2.0 |
-| github.com/nats-io/nuid v1.0.1 | Apache License 2.0 |
-| golang.org/x/crypto v0.13.0 | BSD 3-Clause "New" or "Revised" License |
-| golang.org/x/exp v0.0.0-20230905200255-921286631fa9 | BSD 3-Clause "New" or "Revised" License |
-| golang.org/x/sys v0.12.0 | BSD 3-Clause "New" or "Revised" License |
-| golang.org/x/time v0.3.0 | BSD 3-Clause "New" or "Revised" License |
-| google.golang.org/protobuf v1.23.0 | BSD 3-Clause "New" or "Revised" License |
-| github.com/klauspost/compress/s2 v1.17.0 | BSD 3-Clause "New" or "Revised" License |
-| github.com/minio/highwayhash v1.0.2 | Apache License 2.0 |
-| go.uber.org/automaxprocs v1.5.3 | MIT License |
+| github.com/google/go-tpm | Apache License 2.0 |
+| github.com/klauspost/compress | BSD 3-Clause "New" or "Revised" License |
+| github.com/minio/highwayhash | Apache License 2.0 |
+| github.com/nats-io/jwt/v2 | Apache License 2.0 |
+| github.com/nats-io/nats.go | Apache License 2.0 |
+| github.com/nats-io/nkeys | Apache License 2.0 |
+| github.com/nats-io/nuid  | Apache License 2.0 |
+| go.uber.org/automaxprocs | MIT License |
+| golang.org/x/crypto | BSD 3-Clause "New" or "Revised" License |
+| golang.org/x/sys | BSD 3-Clause "New" or "Revised" License |
+| golang.org/x/time | BSD 3-Clause "New" or "Revised" License |
diff --git a/docker/Dockerfile.nightly b/docker/Dockerfile.nightly
index dbb2c093..3030badd 100644
--- a/docker/Dockerfile.nightly
+++ b/docker/Dockerfile.nightly
@@ -1,4 +1,4 @@
-FROM golang:1.21-alpine AS builder
+FROM golang:1.22-alpine AS builder
 
 ARG VERSION="nightly"
 
diff --git a/go.mod b/go.mod
index abad5d77..19bc49ed 100644
--- a/go.mod
+++ b/go.mod
@@ -1,17 +1,17 @@
 module github.com/nats-io/nats-server/v2
 
-go 1.20
+go 1.21
 
 require (
-	github.com/google/go-tpm v0.9.0
-	github.com/klauspost/compress v1.17.8
+	github.com/google/go-tpm v0.9.1
+	github.com/klauspost/compress v1.17.9
 	github.com/minio/highwayhash v1.0.2
 	github.com/nats-io/jwt/v2 v2.5.7
-	github.com/nats-io/nats.go v1.35.0
+	github.com/nats-io/nats.go v1.36.0
 	github.com/nats-io/nkeys v0.4.7
 	github.com/nats-io/nuid v1.0.1
 	go.uber.org/automaxprocs v1.5.3
-	golang.org/x/crypto v0.23.0
-	golang.org/x/sys v0.20.0
+	golang.org/x/crypto v0.24.0
+	golang.org/x/sys v0.21.0
 	golang.org/x/time v0.5.0
 )
diff --git a/go.sum b/go.sum
index b894d9bc..0aa6f867 100644
--- a/go.sum
+++ b/go.sum
@@ -1,14 +1,18 @@
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
-github.com/google/go-tpm v0.9.0 h1:sQF6YqWMi+SCXpsmS3fd21oPy/vSddwZry4JnmltHVk=
-github.com/google/go-tpm v0.9.0/go.mod h1:FkNVkc6C+IsvDI9Jw1OveJmxGZUUaKxtrpOS47QWKfU=
+github.com/google/go-tpm v0.9.1 h1:0pGc4X//bAlmZzMKf8iz6IsDo1nYTbYJ6FZN/rg4zdM=
+github.com/google/go-tpm v0.9.1/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
 github.com/klauspost/compress v1.17.8 h1:YcnTYrq7MikUT7k0Yb5eceMmALQPYBW/Xltxn0NAMnU=
 github.com/klauspost/compress v1.17.8/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
+github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
+github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
 github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA/g=
 github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
 github.com/nats-io/jwt/v2 v2.5.7 h1:j5lH1fUXCnJnY8SsQeB/a/z9Azgu2bYIDvtPVNdxe2c=
 github.com/nats-io/jwt/v2 v2.5.7/go.mod h1:ZdWS1nZa6WMZfFwwgpEaqBV8EPGVgOTDHN/wTbz0Y5A=
 github.com/nats-io/nats.go v1.35.0 h1:XFNqNM7v5B+MQMKqVGAyHwYhyKb48jrenXNxIU20ULk=
 github.com/nats-io/nats.go v1.35.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
+github.com/nats-io/nats.go v1.36.0 h1:suEUPuWzTSse/XhESwqLxXGuj8vGRuPRoG7MoRN/qyU=
+github.com/nats-io/nats.go v1.36.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
 github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
 github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
@@ -18,11 +22,11 @@ github.com/prashantv/gostub v1.1.0 h1:BTyx3RfQjRHnUWaGF9oQos79AlQ5k8WNktv7VGvVH4
 github.com/stretchr/testify v1.7.1 h1:5TQK59W5E3v0r2duFAb7P95B6hEeOyEnHRa8MjYSMTY=
 go.uber.org/automaxprocs v1.5.3 h1:kWazyxZUrS3Gs4qUpbwo5kEIMGe/DAvi5Z4tl2NW4j8=
 go.uber.org/automaxprocs v1.5.3/go.mod h1:eRbA25aqJrxAbsLO0xy5jVwPt7FQnRgjW+efnwa1WM0=
-golang.org/x/crypto v0.23.0 h1:dIJU/v2J8Mdglj/8rJ6UUOM3Zc9zLZxVZwwxMooUSAI=
-golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8=
+golang.org/x/crypto v0.24.0 h1:mnl8DM0o513X8fdIkmyFE/5hTYxbwYOjDS/+rK6qpRI=
+golang.org/x/crypto v0.24.0/go.mod h1:Z1PMYSOR5nyMcyAVAIQSKCDwalqy85Aqn1x3Ws4L5DM=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y=
-golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.21.0 h1:rF+pYz3DAGSQAxAu1CbC7catZg4ebC4UIeIhKxBZvws=
+golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
 golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
diff --git a/locksordering.txt b/locksordering.txt
index e57655df..4a2c9e71 100644
--- a/locksordering.txt
+++ b/locksordering.txt
@@ -21,4 +21,12 @@ A reloadMu lock was added to prevent newly connecting clients racing with the co
 This must be taken out as soon as a reload is about to happen before any other locks:
 
     reloadMu -> Server
-    reloadMu -> optsMu
\ No newline at end of file
+    reloadMu -> optsMu
+
+The "jscmMu" lock in the Account is used to serialise calls to checkJetStreamMigrate and
+clearObserverState so that they cannot interleave which would leave Raft nodes in
+inconsistent observer states.
+
+    jscmMu -> Account -> jsAccount
+    jscmMu -> stream.clsMu 
+    jscmMu -> RaftNode
\ No newline at end of file
diff --git a/server/accounts.go b/server/accounts.go
index 4576c9fe..6f5a1a70 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -103,6 +103,9 @@ type Account struct {
 	// and if it falls between 0 and that value, message tracing will be triggered.
 	traceDest         string
 	traceDestSampling int
+	// Guarantee that only one goroutine can be running either checkJetStreamMigrate
+	// or clearObserverState at a given time for this account to prevent interleaving.
+	jscmMu sync.Mutex
 }
 
 const (
@@ -1515,6 +1518,10 @@ func (a *Account) addServiceImportWithClaim(destination *Account, from, to strin
 		return err
 	}
 
+	if err := a.serviceImportFormsCycle(destination, to); err != nil {
+		return err
+	}
+
 	_, err := a.addServiceImport(destination, from, to, imClaim)
 
 	return err
@@ -2530,6 +2537,10 @@ func (a *Account) addMappedStreamImportWithClaim(account *Account, from, to stri
 		return err
 	}
 
+	if err := a.streamImportFormsCycle(account, from); err != nil {
+		return err
+	}
+
 	var (
 		usePub bool
 		tr     *subjectTransform
@@ -3675,6 +3686,14 @@ func (s *Server) updateAccountClaimsWithRefresh(a *Account, ac *jwt.AccountClaim
 		}
 		c.mu.Lock()
 		c.applyAccountLimits()
+		// if we have an nkey user we are a callout user - save
+		// the issuedAt, and nkey user id to honor revocations
+		var nkeyUserID string
+		var issuedAt int64
+		if c.user != nil {
+			issuedAt = c.user.Issued
+			nkeyUserID = c.user.Nkey
+		}
 		theJWT := c.opts.JWT
 		c.mu.Unlock()
 		// Check for being revoked here. We use ac one to avoid the account lock.
@@ -3693,6 +3712,27 @@ func (s *Server) updateAccountClaimsWithRefresh(a *Account, ac *jwt.AccountClaim
 				continue
 			}
 		}
+
+		// if we extracted nkeyUserID and issuedAt we are a callout type
+		// calloutIAT should only be set if we are in callout scenario as
+		// the user JWT is _NOT_ associated with the client for callouts,
+		// so we rely on the calloutIAT to know when the JWT was issued
+		// revocations simply state that JWT issued before or by that date
+		// are not valid
+		if ac.Revocations != nil && nkeyUserID != _EMPTY_ && issuedAt > 0 {
+			seconds, ok := ac.Revocations[jwt.All]
+			if ok && seconds >= issuedAt {
+				c.sendErrAndDebug("User Authentication Revoked")
+				c.closeConnection(Revocation)
+				continue
+			}
+			seconds, ok = ac.Revocations[nkeyUserID]
+			if ok && seconds >= issuedAt {
+				c.sendErrAndDebug("User Authentication Revoked")
+				c.closeConnection(Revocation)
+				continue
+			}
+		}
 	}
 
 	// Check if the signing keys changed, might have to evict
@@ -3803,7 +3843,7 @@ func buildPermissionsFromJwt(uc *jwt.Permissions) *Permissions {
 
 // Helper to build internal NKeyUser.
 func buildInternalNkeyUser(uc *jwt.UserClaims, acts map[string]struct{}, acc *Account) *NkeyUser {
-	nu := &NkeyUser{Nkey: uc.Subject, Account: acc, AllowedConnectionTypes: acts}
+	nu := &NkeyUser{Nkey: uc.Subject, Account: acc, AllowedConnectionTypes: acts, Issued: uc.IssuedAt}
 	if uc.IssuerAccount != _EMPTY_ {
 		nu.SigningKey = uc.Issuer
 	}
diff --git a/server/auth.go b/server/auth.go
index 700e5741..865faeb9 100644
--- a/server/auth.go
+++ b/server/auth.go
@@ -60,6 +60,7 @@ type ClientAuthentication interface {
 // NkeyUser is for multiple nkey based users
 type NkeyUser struct {
 	Nkey                   string              `json:"user"`
+	Issued                 int64               `json:"issued,omitempty"` // this is a copy of the issued at (iat) field in the jwt
 	Permissions            *Permissions        `json:"permissions,omitempty"`
 	Account                *Account            `json:"account,omitempty"`
 	SigningKey             string              `json:"signing_key,omitempty"`
diff --git a/server/client.go b/server/client.go
index 97cfc179..5f41b4fb 100644
--- a/server/client.go
+++ b/server/client.go
@@ -2995,8 +2995,10 @@ func (c *client) addShadowSub(sub *subscription, ime *ime, enact bool) (*subscri
 		return nil, fmt.Errorf(errs)
 	}
 
-	// Update our route map here.
-	c.srv.updateRemoteSubscription(im.acc, &nsub, 1)
+	// Update our route map here. But only if we are not a leaf node or a hub leafnode.
+	if c.kind != LEAF || c.isHubLeafNode() {
+		c.srv.updateRemoteSubscription(im.acc, &nsub, 1)
+	}
 
 	return &nsub, nil
 }
@@ -5455,48 +5457,51 @@ func (c *client) closeConnection(reason ClosedState) {
 		// Unregister
 		srv.removeClient(c)
 
-		// Update remote subscriptions.
-		if acc != nil && (kind == CLIENT || kind == LEAF || kind == JETSTREAM) {
-			qsubs := map[string]*qsub{}
-			for _, sub := range subs {
-				// Call unsubscribe here to cleanup shadow subscriptions and such.
-				c.unsubscribe(acc, sub, true, false)
-				// Update route as normal for a normal subscriber.
-				if sub.queue == nil {
-					if !spoke {
-						srv.updateRouteSubscriptionMap(acc, sub, -1)
-						if srv.gateway.enabled {
-							srv.gatewayUpdateSubInterest(acc.Name, sub, -1)
+		if acc != nil {
+			// Update remote subscriptions.
+			if kind == CLIENT || kind == LEAF || kind == JETSTREAM {
+				qsubs := map[string]*qsub{}
+				for _, sub := range subs {
+					// Call unsubscribe here to cleanup shadow subscriptions and such.
+					c.unsubscribe(acc, sub, true, false)
+					// Update route as normal for a normal subscriber.
+					if sub.queue == nil {
+						if !spoke {
+							srv.updateRouteSubscriptionMap(acc, sub, -1)
+							if srv.gateway.enabled {
+								srv.gatewayUpdateSubInterest(acc.Name, sub, -1)
+							}
 						}
-					}
-					acc.updateLeafNodes(sub, -1)
-				} else {
-					// We handle queue subscribers special in case we
-					// have a bunch we can just send one update to the
-					// connected routes.
-					num := int32(1)
-					if kind == LEAF {
-						num = sub.qw
-					}
-					// TODO(dlc) - Better to use string builder?
-					key := bytesToString(sub.subject) + " " + bytesToString(sub.queue)
-					if esub, ok := qsubs[key]; ok {
-						esub.n += num
+						acc.updateLeafNodes(sub, -1)
 					} else {
-						qsubs[key] = &qsub{sub, num}
+						// We handle queue subscribers special in case we
+						// have a bunch we can just send one update to the
+						// connected routes.
+						num := int32(1)
+						if kind == LEAF {
+							num = sub.qw
+						}
+						key := keyFromSub(sub)
+						if esub, ok := qsubs[key]; ok {
+							esub.n += num
+						} else {
+							qsubs[key] = &qsub{sub, num}
+						}
 					}
 				}
-			}
-			// Process any qsubs here.
-			for _, esub := range qsubs {
-				if !spoke {
-					srv.updateRouteSubscriptionMap(acc, esub.sub, -(esub.n))
-					if srv.gateway.enabled {
-						srv.gatewayUpdateSubInterest(acc.Name, esub.sub, -(esub.n))
+				// Process any qsubs here.
+				for _, esub := range qsubs {
+					if !spoke {
+						srv.updateRouteSubscriptionMap(acc, esub.sub, -(esub.n))
+						if srv.gateway.enabled {
+							srv.gatewayUpdateSubInterest(acc.Name, esub.sub, -(esub.n))
+						}
 					}
+					acc.updateLeafNodes(esub.sub, -(esub.n))
 				}
-				acc.updateLeafNodes(esub.sub, -(esub.n))
 			}
+			// Always remove from the account, otherwise we can leak clients.
+			// Note that SYSTEM and ACCOUNT types from above cleanup their own subs.
 			if prev := acc.removeClient(c); prev == 1 {
 				srv.decActiveAccounts()
 			}
diff --git a/server/consumer.go b/server/consumer.go
index ce32df2d..846cd788 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -717,7 +717,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	}
 
 	mset.mu.RLock()
-	s, jsa, tierName, cfg, acc := mset.srv, mset.jsa, mset.tier, mset.cfg, mset.acc
+	s, jsa, cfg, acc := mset.srv, mset.jsa, mset.cfg, mset.acc
 	retention := cfg.Retention
 	mset.mu.RUnlock()
 
@@ -732,10 +732,8 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 		return nil, NewJSConsumerConfigRequiredError()
 	}
 
-	jsa.usageMu.RLock()
-	selectedLimits, limitsFound := jsa.limits[tierName]
-	jsa.usageMu.RUnlock()
-	if !limitsFound {
+	selectedLimits, _, _, _ := acc.selectLimits(config.replicas(&cfg))
+	if selectedLimits == nil {
 		return nil, NewJSNoLimitsError()
 	}
 
@@ -743,10 +741,10 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	// Make sure we have sane defaults. Do so with the JS lock, otherwise a
 	// badly timed meta snapshot can result in a race condition.
 	mset.js.mu.Lock()
-	setConsumerConfigDefaults(config, &mset.cfg, srvLim, &selectedLimits)
+	setConsumerConfigDefaults(config, &mset.cfg, srvLim, selectedLimits)
 	mset.js.mu.Unlock()
 
-	if err := checkConsumerCfg(config, srvLim, &cfg, acc, &selectedLimits, isRecovering); err != nil {
+	if err := checkConsumerCfg(config, srvLim, &cfg, acc, selectedLimits, isRecovering); err != nil {
 		return nil, err
 	}
 	sampleFreq := 0
@@ -2197,19 +2195,17 @@ func (o *consumer) loopAndForwardProposals(qch chan struct{}) {
 		const maxBatch = 256 * 1024
 		var entries []*Entry
 		for sz := 0; proposal != nil; proposal = proposal.next {
-			entry := entryPool.Get().(*Entry)
-			entry.Type, entry.Data = EntryNormal, proposal.data
-			entries = append(entries, entry)
+			entries = append(entries, newEntry(EntryNormal, proposal.data))
 			sz += len(proposal.data)
 			if sz > maxBatch {
-				node.ProposeDirect(entries)
+				node.ProposeMulti(entries)
 				// We need to re-create `entries` because there is a reference
 				// to it in the node's pae map.
 				sz, entries = 0, nil
 			}
 		}
 		if len(entries) > 0 {
-			node.ProposeDirect(entries)
+			node.ProposeMulti(entries)
 		}
 		return nil
 	}
@@ -2232,22 +2228,18 @@ func (o *consumer) loopAndForwardProposals(qch chan struct{}) {
 
 // Lock should be held.
 func (o *consumer) propose(entry []byte) {
-	var notify bool
 	p := &proposal{data: entry}
 	if o.phead == nil {
 		o.phead = p
-		notify = true
 	} else {
 		o.ptail.next = p
 	}
 	o.ptail = p
 
-	// Kick our looper routine if needed.
-	if notify {
-		select {
-		case o.pch <- struct{}{}:
-		default:
-		}
+	// Kick our looper routine.
+	select {
+	case o.pch <- struct{}{}:
+	default:
 	}
 }
 
@@ -2725,17 +2717,24 @@ func (o *consumer) infoWithSnapAndReply(snap bool, reply string) *ConsumerInfo {
 		}
 	}
 
-	// If we are replicated and we are not the leader we need to pull certain data from our store.
-	if rg != nil && rg.node != nil && !o.isLeader() && o.store != nil {
+	// If we are replicated and we are not the leader or we are filtered, we need to pull certain data from our store.
+	isLeader := o.isLeader()
+	if rg != nil && rg.node != nil && o.store != nil && (!isLeader || o.isFiltered()) {
 		state, err := o.store.BorrowState()
 		if err != nil {
 			o.mu.Unlock()
 			return nil
 		}
-		info.Delivered.Consumer, info.Delivered.Stream = state.Delivered.Consumer, state.Delivered.Stream
-		info.AckFloor.Consumer, info.AckFloor.Stream = state.AckFloor.Consumer, state.AckFloor.Stream
-		info.NumAckPending = len(state.Pending)
-		info.NumRedelivered = len(state.Redelivered)
+		if !isLeader {
+			info.Delivered.Consumer, info.Delivered.Stream = state.Delivered.Consumer, state.Delivered.Stream
+			info.AckFloor.Consumer, info.AckFloor.Stream = state.AckFloor.Consumer, state.AckFloor.Stream
+			info.NumAckPending = len(state.Pending)
+			info.NumRedelivered = len(state.Redelivered)
+		} else {
+			// Since we are filtered and we are the leader we could have o.sseq that is skipped ahead.
+			// To maintain consistency in reporting (e.g. jsz) we take the state for our delivered stream sequence.
+			info.Delivered.Stream = state.Delivered.Stream
+		}
 	}
 
 	// Adjust active based on non-zero etc. Also make UTC here.
@@ -2834,6 +2833,12 @@ func (o *consumer) processAckMsg(sseq, dseq, dc uint64, reply string, doSample b
 		return
 	}
 
+	// Check if this ack is above the current pointer to our next to deliver.
+	// This could happen on a cooperative takeover with high speed deliveries.
+	if sseq >= o.sseq {
+		o.sseq = sseq + 1
+	}
+
 	mset := o.mset
 	if mset == nil || mset.closed.Load() {
 		o.mu.Unlock()
@@ -2855,8 +2860,12 @@ func (o *consumer) processAckMsg(sseq, dseq, dc uint64, reply string, doSample b
 			delete(o.pending, sseq)
 			// Use the original deliver sequence from our pending record.
 			dseq = p.Sequence
+
 			// Only move floors if we matched an existing pending.
-			if dseq == o.adflr+1 {
+			if len(o.pending) == 0 {
+				o.adflr = o.dseq - 1
+				o.asflr = o.sseq - 1
+			} else if dseq == o.adflr+1 {
 				o.adflr, o.asflr = dseq, sseq
 				for ss := sseq + 1; ss < o.sseq; ss++ {
 					if p, ok := o.pending[ss]; ok {
@@ -2867,11 +2876,6 @@ func (o *consumer) processAckMsg(sseq, dseq, dc uint64, reply string, doSample b
 					}
 				}
 			}
-			// If nothing left set consumer to current delivered.
-			// Do not update stream.
-			if len(o.pending) == 0 {
-				o.adflr = o.dseq - 1
-			}
 		}
 		delete(o.rdc, sseq)
 		o.removeFromRedeliverQueue(sseq)
@@ -5319,18 +5323,19 @@ func (o *consumer) stopWithFlags(dflag, sdflag, doSignal, advisory bool) error {
 // ignoreInterest marks whether the consumer should be ignored when determining interest.
 // No lock held on entry.
 func (o *consumer) cleanupNoInterestMessages(mset *stream, ignoreInterest bool) {
-	state := mset.state()
-	stop := state.LastSeq
 	o.mu.Lock()
 	if !o.isLeader() {
-		o.readStoredState(stop)
+		o.readStoredState(0)
 	}
 	start := o.asflr
 	o.mu.Unlock()
+
 	// Make sure we start at worst with first sequence in the stream.
+	state := mset.state()
 	if start < state.FirstSeq {
 		start = state.FirstSeq
 	}
+	stop := state.LastSeq
 
 	// Consumer's interests are ignored by default. If we should not ignore interest, unset.
 	co := o
@@ -5339,13 +5344,37 @@ func (o *consumer) cleanupNoInterestMessages(mset *stream, ignoreInterest bool)
 	}
 
 	var rmseqs []uint64
-	mset.mu.Lock()
+	mset.mu.RLock()
+
+	// If over this amount of messages to check, defer to checkInterestState() which
+	// will do the right thing since we are now removed.
+	// TODO(dlc) - Better way?
+	const bailThresh = 100_000
+
+	// Check if we would be spending too much time here and defer to separate go routine.
+	if len(mset.consumers) == 0 {
+		mset.mu.RUnlock()
+		mset.mu.Lock()
+		defer mset.mu.Unlock()
+		mset.store.Purge()
+		var state StreamState
+		mset.store.FastState(&state)
+		mset.lseq = state.LastSeq
+		// Also make sure we clear any pending acks.
+		mset.clearAllPreAcksBelowFloor(state.FirstSeq)
+		return
+	} else if stop-start > bailThresh {
+		mset.mu.RUnlock()
+		go mset.checkInterestState()
+		return
+	}
+
 	for seq := start; seq <= stop; seq++ {
 		if mset.noInterest(seq, co) {
 			rmseqs = append(rmseqs, seq)
 		}
 	}
-	mset.mu.Unlock()
+	mset.mu.RUnlock()
 
 	// These can be removed.
 	for _, seq := range rmseqs {
diff --git a/server/events.go b/server/events.go
index 9c214cc6..ff940571 100644
--- a/server/events.go
+++ b/server/events.go
@@ -98,6 +98,12 @@ const (
 // FIXME(dlc) - make configurable.
 var eventsHBInterval = 30 * time.Second
 
+// Default minimum wait time for sending statsz
+const defaultStatszRateLimit = 1 * time.Second
+
+// Variable version so we can set in tests.
+var statszRateLimit = defaultStatszRateLimit
+
 type sysMsgHandler func(sub *subscription, client *client, acc *Account, subject, reply string, hdr, msg []byte)
 
 // Used if we have to queue things internally to avoid the route/gw path.
@@ -134,6 +140,7 @@ type internal struct {
 	shash          string
 	inboxPre       string
 	remoteStatsSub *subscription
+	lastStatsz     time.Time
 }
 
 // ServerStatsMsg is sent periodically with stats updates.
@@ -808,6 +815,10 @@ func (s *Server) sendStatsz(subj string) {
 	var m ServerStatsMsg
 	s.updateServerUsage(&m.Stats)
 
+	if s.limitStatsz(subj) {
+		return
+	}
+
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 
@@ -949,6 +960,35 @@ func (s *Server) sendStatsz(subj string) {
 	s.sendInternalMsg(subj, _EMPTY_, &m.Server, &m)
 }
 
+// Limit updates to the heartbeat interval, max one second by default.
+func (s *Server) limitStatsz(subj string) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.sys == nil {
+		return true
+	}
+
+	// Only limit the normal broadcast subject.
+	if subj != fmt.Sprintf(serverStatsSubj, s.ID()) {
+		return false
+	}
+
+	interval := statszRateLimit
+	if s.sys.cstatsz < interval {
+		interval = s.sys.cstatsz
+	}
+	if time.Since(s.sys.lastStatsz) < interval {
+		// Reschedule heartbeat for the next interval.
+		if s.sys.stmr != nil {
+			s.sys.stmr.Reset(time.Until(s.sys.lastStatsz.Add(interval)))
+		}
+		return true
+	}
+	s.sys.lastStatsz = time.Now()
+	return false
+}
+
 // Send out our statz update.
 // This should be wrapChk() to setup common locking.
 func (s *Server) heartbeatStatsz() {
@@ -966,6 +1006,12 @@ func (s *Server) heartbeatStatsz() {
 	go s.sendStatszUpdate()
 }
 
+// Reset statsz rate limit for the next broadcast.
+// This should be wrapChk() to setup common locking.
+func (s *Server) resetLastStatsz() {
+	s.sys.lastStatsz = time.Time{}
+}
+
 func (s *Server) sendStatszUpdate() {
 	s.sendStatsz(fmt.Sprintf(serverStatsSubj, s.ID()))
 }
@@ -1869,6 +1915,7 @@ func (s *Server) statszReq(sub *subscription, c *client, _ *Account, subject, re
 	// No reply is a signal that we should use our normal broadcast subject.
 	if reply == _EMPTY_ {
 		reply = fmt.Sprintf(serverStatsSubj, s.info.ID)
+		s.wrapChk(s.resetLastStatsz)
 	}
 
 	opts := StatszEventOptions{}
diff --git a/server/filestore.go b/server/filestore.go
index 0a63fdbb..67e3a3e3 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -295,13 +295,13 @@ const (
 	// Maximum size of a write buffer we may consider for re-use.
 	maxBufReuse = 2 * 1024 * 1024
 	// default cache buffer expiration
-	defaultCacheBufferExpiration = 2 * time.Second
+	defaultCacheBufferExpiration = 10 * time.Second
 	// default sync interval
 	defaultSyncInterval = 2 * time.Minute
 	// default idle timeout to close FDs.
 	closeFDsIdle = 30 * time.Second
 	// default expiration time for mb.fss when idle.
-	defaultFssExpiration = 10 * time.Second
+	defaultFssExpiration = 2 * time.Minute
 	// coalesceMinimum
 	coalesceMinimum = 16 * 1024
 	// maxFlushWait is maximum we will wait to gather messages to flush.
@@ -1869,7 +1869,7 @@ func (mb *msgBlock) lastChecksum() []byte {
 		mb.rbytes = uint64(fi.Size())
 	}
 	if mb.rbytes < checksumSize {
-		return nil
+		return lchk[:]
 	}
 	// Encrypted?
 	// Check for encryption, we do not load keys on startup anymore so might need to load them here.
@@ -2603,28 +2603,44 @@ func (fs *fileStore) numFilteredPending(filter string, ss *SimpleState) {
 		return
 	}
 
-	start, stop := uint32(math.MaxUint32), uint32(0)
-	fs.psim.Match(stringToBytes(filter), func(_ []byte, psi *psi) {
-		ss.Msgs += psi.total
-		// Keep track of start and stop indexes for this subject.
-		if psi.fblk < start {
-			start = psi.fblk
-		}
-		if psi.lblk > stop {
-			stop = psi.lblk
-		}
-	})
 	// We do need to figure out the first and last sequences.
 	wc := subjectHasWildcard(filter)
+	start, stop := uint32(math.MaxUint32), uint32(0)
+
+	if wc {
+		fs.psim.Match(stringToBytes(filter), func(_ []byte, psi *psi) {
+			ss.Msgs += psi.total
+			// Keep track of start and stop indexes for this subject.
+			if psi.fblk < start {
+				start = psi.fblk
+			}
+			if psi.lblk > stop {
+				stop = psi.lblk
+			}
+		})
+	} else if psi, ok := fs.psim.Find(stringToBytes(filter)); ok {
+		ss.Msgs += psi.total
+		start, stop = psi.fblk, psi.lblk
+	}
+
+	// Did not find anything.
+	if stop == 0 {
+		ss.First, ss.Last, ss.Msgs = 0, 0, 0
+		return
+	}
+
 	// Do start
 	mb := fs.bim[start]
 	if mb != nil {
 		_, f, _ := mb.filteredPending(filter, wc, 0)
 		ss.First = f
 	}
+
+	// Hold this outside loop for psim fblk updates on misses.
+	i := start + 1
 	if ss.First == 0 {
 		// This is a miss. This can happen since psi.fblk is lazy, but should be very rare.
-		for i := start + 1; i <= stop; i++ {
+		for ; i <= stop; i++ {
 			mb := fs.bim[i]
 			if mb == nil {
 				continue
@@ -2635,6 +2651,20 @@ func (fs *fileStore) numFilteredPending(filter string, ss *SimpleState) {
 			}
 		}
 	}
+	// Update fblk if we missed matching some blocks, meaning fblk was outdated.
+	if i > start+1 {
+		if !wc {
+			if info, ok := fs.psim.Find(stringToBytes(filter)); ok {
+				info.fblk = i
+			}
+		} else {
+			fs.psim.Match(stringToBytes(filter), func(_ []byte, psi *psi) {
+				if i > psi.fblk {
+					psi.fblk = i
+				}
+			})
+		}
+	}
 	// Now last
 	if mb = fs.bim[stop]; mb != nil {
 		_, _, l := mb.filteredPending(filter, wc, 0)
@@ -3453,6 +3483,17 @@ func (mb *msgBlock) skipMsg(seq uint64, now time.Time) {
 		mb.last.ts = nowts
 		atomic.StoreUint64(&mb.first.seq, seq+1)
 		mb.first.ts = nowts
+		needsRecord = mb == mb.fs.lmb
+		if needsRecord && mb.rbytes > 0 {
+			// We want to make sure since we have no messages
+			// that we write to the beginning since we only need last one.
+			mb.rbytes, mb.cache = 0, &cache{}
+			// If encrypted we need to reset counter since we just keep one.
+			if mb.bek != nil {
+				// Recreate to reset counter.
+				mb.bek, _ = genBlockEncryptionKey(mb.fs.fcfg.Cipher, mb.seed, mb.nonce)
+			}
+		}
 	} else {
 		needsRecord = true
 		mb.dmap.Insert(seq)
@@ -6219,15 +6260,29 @@ func (fs *fileStore) loadLast(subj string, sm *StoreMsg) (lsm *StoreMsg, err err
 		return nil, ErrStoreMsgNotFound
 	}
 
-	start, stop := fs.lmb.index, fs.blks[0].index
 	wc := subjectHasWildcard(subj)
+	var start, stop uint32
+
 	// If literal subject check for presence.
-	if !wc {
-		if info, ok := fs.psim.Find(stringToBytes(subj)); !ok {
+	if wc {
+		start = fs.lmb.index
+		fs.psim.Match(stringToBytes(subj), func(_ []byte, psi *psi) {
+			// Keep track of start and stop indexes for this subject.
+			if psi.fblk < start {
+				start = psi.fblk
+			}
+			if psi.lblk > stop {
+				stop = psi.lblk
+			}
+		})
+		// None matched.
+		if stop == 0 {
 			return nil, ErrStoreMsgNotFound
-		} else {
-			start, stop = info.lblk, info.fblk
 		}
+	} else if info, ok := fs.psim.Find(stringToBytes(subj)); ok {
+		start, stop = info.lblk, info.fblk
+	} else {
+		return nil, ErrStoreMsgNotFound
 	}
 
 	// Walk blocks backwards.
@@ -8689,7 +8744,8 @@ func (o *consumerFileStore) UpdateDelivered(dseq, sseq, dc uint64, ts int64) err
 		// Check for an update to a message already delivered.
 		if sseq <= o.state.Delivered.Stream {
 			if p = o.state.Pending[sseq]; p != nil {
-				p.Sequence, p.Timestamp = dseq, ts
+				// Do not update p.Sequence, that should be the original delivery sequence.
+				p.Timestamp = ts
 			}
 		} else {
 			// Add to pending.
@@ -8747,7 +8803,14 @@ func (o *consumerFileStore) UpdateAcks(dseq, sseq uint64) error {
 		return nil
 	}
 
+	// Match leader logic on checking if ack is ahead of delivered.
+	// This could happen on a cooperative takeover with high speed deliveries.
+	if sseq > o.state.Delivered.Stream {
+		o.state.Delivered.Stream = sseq + 1
+	}
+
 	if len(o.state.Pending) == 0 || o.state.Pending[sseq] == nil {
+		delete(o.state.Redelivered, sseq)
 		return ErrStoreMsgNotFound
 	}
 
@@ -8778,7 +8841,9 @@ func (o *consumerFileStore) UpdateAcks(dseq, sseq uint64) error {
 	// First delete from our pending state.
 	if p, ok := o.state.Pending[sseq]; ok {
 		delete(o.state.Pending, sseq)
-		dseq = p.Sequence // Use the original.
+		if dseq > p.Sequence && p.Sequence > 0 {
+			dseq = p.Sequence // Use the original.
+		}
 	}
 	if len(o.state.Pending) == 0 {
 		o.state.AckFloor.Consumer = o.state.Delivered.Consumer
diff --git a/server/jetstream.go b/server/jetstream.go
index cf10e743..7627417d 100644
--- a/server/jetstream.go
+++ b/server/jetstream.go
@@ -1483,7 +1483,11 @@ func (a *Account) maxBytesLimits(cfg *StreamConfig) (bool, int64) {
 		return false, 0
 	}
 	jsa.usageMu.RLock()
-	selectedLimits, _, ok := jsa.selectLimits(cfg)
+	var replicas int
+	if cfg != nil {
+		replicas = cfg.Replicas
+	}
+	selectedLimits, _, ok := jsa.selectLimits(replicas)
 	jsa.usageMu.RUnlock()
 	if !ok {
 		return false, 0
@@ -1633,7 +1637,7 @@ func diffCheckedLimits(a, b map[string]JetStreamAccountLimits) map[string]JetStr
 func (jsa *jsAccount) reservedStorage(tier string) (mem, store uint64) {
 	for _, mset := range jsa.streams {
 		cfg := &mset.cfg
-		if tier == _EMPTY_ || tier == tierName(cfg) && cfg.MaxBytes > 0 {
+		if tier == _EMPTY_ || tier == tierName(cfg.Replicas) && cfg.MaxBytes > 0 {
 			switch cfg.Storage {
 			case FileStorage:
 				store += uint64(cfg.MaxBytes)
@@ -1650,7 +1654,7 @@ func (jsa *jsAccount) reservedStorage(tier string) (mem, store uint64) {
 func reservedStorage(sas map[string]*streamAssignment, tier string) (mem, store uint64) {
 	for _, sa := range sas {
 		cfg := sa.Config
-		if tier == _EMPTY_ || tier == tierName(cfg) && cfg.MaxBytes > 0 {
+		if tier == _EMPTY_ || tier == tierName(cfg.Replicas) && cfg.MaxBytes > 0 {
 			switch cfg.Storage {
 			case FileStorage:
 				store += uint64(cfg.MaxBytes)
@@ -1738,17 +1742,29 @@ func (a *Account) JetStreamUsage() JetStreamAccountStats {
 				stats.ReservedMemory, stats.ReservedStore = reservedStorage(sas, _EMPTY_)
 			}
 			for _, sa := range sas {
-				stats.Consumers += len(sa.consumers)
-				if !defaultTier {
-					tier := tierName(sa.Config)
-					u, ok := stats.Tiers[tier]
+				if defaultTier {
+					stats.Consumers += len(sa.consumers)
+				} else {
+					stats.Streams++
+					streamTier := tierName(sa.Config.Replicas)
+					su, ok := stats.Tiers[streamTier]
 					if !ok {
-						u = JetStreamTier{}
+						su = JetStreamTier{}
+					}
+					su.Streams++
+					stats.Tiers[streamTier] = su
+
+					// Now consumers, check each since could be different tiers.
+					for _, ca := range sa.consumers {
+						stats.Consumers++
+						consumerTier := tierName(ca.Config.replicas(sa.Config))
+						cu, ok := stats.Tiers[consumerTier]
+						if !ok {
+							cu = JetStreamTier{}
+						}
+						cu.Consumers++
+						stats.Tiers[consumerTier] = cu
 					}
-					u.Streams++
-					stats.Streams++
-					u.Consumers += len(sa.consumers)
-					stats.Tiers[tier] = u
 				}
 			}
 		} else {
@@ -2132,9 +2148,8 @@ func (js *jetStream) limitsExceeded(storeType StorageType) bool {
 	return js.wouldExceedLimits(storeType, 0)
 }
 
-func tierName(cfg *StreamConfig) string {
+func tierName(replicas int) string {
 	// TODO (mh) this is where we could select based off a placement tag as well "qos:tier"
-	replicas := cfg.Replicas
 	if replicas == 0 {
 		replicas = 1
 	}
@@ -2154,11 +2169,11 @@ func (jsa *jsAccount) jetStreamAndClustered() (*jetStream, bool) {
 }
 
 // jsa.usageMu read lock should be held.
-func (jsa *jsAccount) selectLimits(cfg *StreamConfig) (JetStreamAccountLimits, string, bool) {
+func (jsa *jsAccount) selectLimits(replicas int) (JetStreamAccountLimits, string, bool) {
 	if selectedLimits, ok := jsa.limits[_EMPTY_]; ok {
 		return selectedLimits, _EMPTY_, true
 	}
-	tier := tierName(cfg)
+	tier := tierName(replicas)
 	if selectedLimits, ok := jsa.limits[tier]; ok {
 		return selectedLimits, tier, true
 	}
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index cc75715d..7cb2c27c 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -3319,7 +3319,11 @@ func (s *Server) jsStreamPurgeRequest(sub *subscription, c *client, _ *Account,
 }
 
 func (acc *Account) jsNonClusteredStreamLimitsCheck(cfg *StreamConfig) *ApiError {
-	selectedLimits, tier, jsa, apiErr := acc.selectLimits(cfg)
+	var replicas int
+	if cfg != nil {
+		replicas = cfg.Replicas
+	}
+	selectedLimits, tier, jsa, apiErr := acc.selectLimits(replicas)
 	if apiErr != nil {
 		return apiErr
 	}
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index f95758be..6552410f 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -534,12 +534,18 @@ func (js *jetStream) isStreamHealthy(acc *Account, sa *streamAssignment) bool {
 		return false
 	}
 
-	// If we are catching up return false.
-	if mset.isCatchingUp() {
+	// If R1 we are good.
+	if node == nil {
+		return true
+	}
+
+	// Here we are a replicated stream.
+	// First make sure our monitor routine is running.
+	if !mset.isMonitorRunning() {
 		return false
 	}
 
-	if node == nil || node.Healthy() {
+	if node.Healthy() {
 		// Check if we are processing a snapshot and are catching up.
 		if !mset.isCatchingUp() {
 			return true
@@ -553,7 +559,6 @@ func (js *jetStream) isStreamHealthy(acc *Account, sa *streamAssignment) bool {
 			js.restartStream(acc, sa)
 		}
 	}
-
 	return false
 }
 
@@ -863,6 +868,8 @@ func (js *jetStream) setupMetaGroup() error {
 	atomic.StoreInt32(&js.clustered, 1)
 	c.registerWithAccount(sacc)
 
+	// Set to true before we start.
+	js.metaRecovering = true
 	js.srv.startGoRoutine(
 		js.monitorCluster,
 		pprofLabels{
@@ -2164,7 +2171,7 @@ func genPeerInfo(peers []string, split int) (newPeers, oldPeers []string, newPee
 // Should only be called from monitorStream.
 func (mset *stream) waitOnConsumerAssignments() {
 	mset.mu.RLock()
-	s, js, acc, sa, name := mset.srv, mset.js, mset.acc, mset.sa, mset.cfg.Name
+	s, js, acc, sa, name, replicas := mset.srv, mset.js, mset.acc, mset.sa, mset.cfg.Name, mset.cfg.Replicas
 	mset.mu.RUnlock()
 
 	if s == nil || js == nil || acc == nil || sa == nil {
@@ -2186,6 +2193,9 @@ func (mset *stream) waitOnConsumerAssignments() {
 		for _, o := range mset.getConsumers() {
 			// Make sure we are registered with our consumer assignment.
 			if ca := o.consumerAssignment(); ca != nil {
+				if replicas > 1 && !o.isMonitorRunning() {
+					break
+				}
 				numReady++
 			} else {
 				break
@@ -2373,7 +2383,8 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 		// since we process streams first then consumers as an asset class.
 		mset.waitOnConsumerAssignments()
 		// Setup a periodic check here.
-		cist = time.NewTicker(30 * time.Second)
+		// We will fire in 5s the first time then back off to 30s
+		cist = time.NewTicker(5 * time.Second)
 		cistc = cist.C
 	}
 
@@ -2496,6 +2507,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 			}
 
 		case <-cistc:
+			cist.Reset(30 * time.Second)
 			// We may be adjusting some things with consumers so do this in its own go routine.
 			go mset.checkInterestState()
 
@@ -4431,11 +4443,15 @@ func (js *jetStream) processClusterCreateConsumer(ca *consumerAssignment, state
 			s.sendInternalMsgLocked(consumerAssignmentSubj, _EMPTY_, nil, b)
 		}
 	} else {
+		js.mu.RLock()
+		node := rg.node
+		js.mu.RUnlock()
+
 		if didCreate {
 			o.setCreatedTime(ca.Created)
 		} else {
 			// Check for scale down to 1..
-			if rg.node != nil && len(rg.Peers) == 1 {
+			if node != nil && len(rg.Peers) == 1 {
 				o.clearNode()
 				o.setLeader(true)
 				// Need to clear from rg too.
@@ -4450,7 +4466,7 @@ func (js *jetStream) processClusterCreateConsumer(ca *consumerAssignment, state
 			}
 		}
 
-		if rg.node == nil {
+		if node == nil {
 			// Single replica consumer, process manually here.
 			js.mu.Lock()
 			// Force response in case we think this is an update.
@@ -4920,7 +4936,22 @@ func (js *jetStream) applyConsumerEntries(o *consumer, ce *CommittedEntry, isLea
 					}
 				}
 				// Check our interest state if applicable.
-				o.checkStateForInterestStream()
+				if err := o.checkStateForInterestStream(); err == errAckFloorHigherThanLastSeq {
+					o.mu.RLock()
+					mset := o.mset
+					o.mu.RUnlock()
+					// Register pre-acks unless no state at all for the stream and we would create alot of pre-acks.
+					mset.mu.Lock()
+					var ss StreamState
+					mset.store.FastState(&ss)
+					// Only register if we have a valid FirstSeq.
+					if ss.FirstSeq > 0 {
+						for seq := ss.FirstSeq; seq < state.AckFloor.Stream; seq++ {
+							mset.registerPreAck(o, seq)
+						}
+					}
+					mset.mu.Unlock()
+				}
 			}
 
 		} else if e.Type == EntryRemovePeer {
@@ -5969,7 +6000,7 @@ func (js *jetStream) createGroupForStream(ci *ClientInfo, cfg *StreamConfig) (*r
 	return nil, errs
 }
 
-func (acc *Account) selectLimits(cfg *StreamConfig) (*JetStreamAccountLimits, string, *jsAccount, *ApiError) {
+func (acc *Account) selectLimits(replicas int) (*JetStreamAccountLimits, string, *jsAccount, *ApiError) {
 	// Grab our jetstream account info.
 	acc.mu.RLock()
 	jsa := acc.js
@@ -5980,7 +6011,7 @@ func (acc *Account) selectLimits(cfg *StreamConfig) (*JetStreamAccountLimits, st
 	}
 
 	jsa.usageMu.RLock()
-	selectedLimits, tierName, ok := jsa.selectLimits(cfg)
+	selectedLimits, tierName, ok := jsa.selectLimits(replicas)
 	jsa.usageMu.RUnlock()
 
 	if !ok {
@@ -5991,7 +6022,11 @@ func (acc *Account) selectLimits(cfg *StreamConfig) (*JetStreamAccountLimits, st
 
 // Read lock needs to be held
 func (js *jetStream) jsClusteredStreamLimitsCheck(acc *Account, cfg *StreamConfig) *ApiError {
-	selectedLimits, tier, _, apiErr := acc.selectLimits(cfg)
+	var replicas int
+	if cfg != nil {
+		replicas = cfg.Replicas
+	}
+	selectedLimits, tier, _, apiErr := acc.selectLimits(replicas)
 	if apiErr != nil {
 		return apiErr
 	}
@@ -7128,7 +7163,7 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
 		return
 	}
-	selectedLimits, _, _, apiErr := acc.selectLimits(&streamCfg)
+	selectedLimits, _, _, apiErr := acc.selectLimits(cfg.replicas(&streamCfg))
 	if apiErr != nil {
 		resp.Error = apiErr
 		s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
@@ -7159,25 +7194,45 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 		return
 	}
 
+	// Was a consumer name provided?
+	var oname string
+	if isDurableConsumer(cfg) || cfg.Name != _EMPTY_ {
+		if cfg.Name != _EMPTY_ {
+			oname = cfg.Name
+		} else {
+			oname = cfg.Durable
+		}
+	}
+
 	// Check for max consumers here to short circuit if possible.
 	// Start with limit on a stream, but if one is defined at the level of the account
 	// and is lower, use that limit.
-	maxc := sa.Config.MaxConsumers
-	if maxc <= 0 || (selectedLimits.MaxConsumers > 0 && selectedLimits.MaxConsumers < maxc) {
-		maxc = selectedLimits.MaxConsumers
-	}
-	if maxc > 0 {
-		// Don't count DIRECTS.
-		total := 0
-		for _, ca := range sa.consumers {
-			if ca.Config != nil && !ca.Config.Direct {
-				total++
+	if action == ActionCreate || action == ActionCreateOrUpdate {
+		maxc := sa.Config.MaxConsumers
+		if maxc <= 0 || (selectedLimits.MaxConsumers > 0 && selectedLimits.MaxConsumers < maxc) {
+			maxc = selectedLimits.MaxConsumers
+		}
+		if maxc > 0 {
+			// Don't count DIRECTS.
+			total := 0
+			for cn, ca := range sa.consumers {
+				if action == ActionCreateOrUpdate {
+					// If the consumer name is specified and we think it already exists, then
+					// we're likely updating an existing consumer, so don't count it. Otherwise
+					// we will incorrectly return NewJSMaximumConsumersLimitError for an update.
+					if oname != _EMPTY_ && cn == oname && sa.consumers[oname] != nil {
+						continue
+					}
+				}
+				if ca.Config != nil && !ca.Config.Direct {
+					total++
+				}
+			}
+			if total >= maxc {
+				resp.Error = NewJSMaximumConsumersLimitError()
+				s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
+				return
 			}
-		}
-		if total >= maxc {
-			resp.Error = NewJSMaximumConsumersLimitError()
-			s.sendAPIErrResponse(ci, acc, subject, reply, string(rmsg), s.jsonResponse(&resp))
-			return
 		}
 	}
 
@@ -7204,16 +7259,10 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 	}
 
 	var ca *consumerAssignment
-	var oname string
 
 	// See if we have an existing one already under same durable name or
 	// if name was set by the user.
-	if isDurableConsumer(cfg) || cfg.Name != _EMPTY_ {
-		if cfg.Name != _EMPTY_ {
-			oname = cfg.Name
-		} else {
-			oname = cfg.Durable
-		}
+	if oname != _EMPTY_ {
 		if ca = sa.consumers[oname]; ca != nil && !ca.deleted {
 			if action == ActionCreate && !reflect.DeepEqual(cfg, ca.Config) {
 				resp.Error = NewJSConsumerAlreadyExistsError()
@@ -7634,7 +7683,10 @@ func (mset *stream) stateSnapshot() []byte {
 func (mset *stream) stateSnapshotLocked() []byte {
 	// Decide if we can support the new style of stream snapshots.
 	if mset.supportsBinarySnapshotLocked() {
-		snap, _ := mset.store.EncodedStreamState(mset.getCLFS())
+		snap, err := mset.store.EncodedStreamState(mset.getCLFS())
+		if err != nil {
+			return nil
+		}
 		return snap
 	}
 
@@ -8144,8 +8196,11 @@ func (mset *stream) processSnapshot(snap *StreamReplicatedState) (e error) {
 	var sub *subscription
 	var err error
 
-	const activityInterval = 30 * time.Second
-	notActive := time.NewTimer(activityInterval)
+	const (
+		startInterval    = 5 * time.Second
+		activityInterval = 30 * time.Second
+	)
+	notActive := time.NewTimer(startInterval)
 	defer notActive.Stop()
 
 	defer func() {
@@ -8228,7 +8283,7 @@ RETRY:
 		default:
 		}
 	}
-	notActive.Reset(activityInterval)
+	notActive.Reset(startInterval)
 
 	// Grab sync request again on failures.
 	if sreq == nil {
@@ -8273,8 +8328,10 @@ RETRY:
 	// Send our sync request.
 	b, _ := json.Marshal(sreq)
 	s.sendInternalMsgLocked(subject, reply, nil, b)
+
 	// Remember when we sent this out to avoid loop spins on errors below.
 	reqSendTime := time.Now()
+
 	// Clear our sync request.
 	sreq = nil
 
@@ -8823,7 +8880,7 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 					done = maxOutMsgs-atomic.LoadInt32(&outm) > minBatchWait
 					if !done {
 						// Wait for a small bit.
-						time.Sleep(50 * time.Millisecond)
+						time.Sleep(100 * time.Millisecond)
 					} else {
 						// GC friendly.
 						mw.Stop()
@@ -8912,7 +8969,9 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 							mset.account(), mset.name(), seq, state)
 						// Try our best to redo our invalidated snapshot as well.
 						if n := mset.raftNode(); n != nil {
-							n.InstallSnapshot(mset.stateSnapshot())
+							if snap := mset.stateSnapshot(); snap != nil {
+								n.InstallSnapshot(snap)
+							}
 						}
 						// If we allow gap markers check if we have one pending.
 						if drOk && dr.First > 0 {
diff --git a/server/leafnode.go b/server/leafnode.go
index 23503c2b..f8624406 100644
--- a/server/leafnode.go
+++ b/server/leafnode.go
@@ -584,6 +584,9 @@ func (s *Server) clearObserverState(remote *leafNodeCfg) {
 		return
 	}
 
+	acc.jscmMu.Lock()
+	defer acc.jscmMu.Unlock()
+
 	// Walk all streams looking for any clustered stream, skip otherwise.
 	for _, mset := range acc.streams() {
 		node := mset.raftNode()
@@ -619,6 +622,9 @@ func (s *Server) checkJetStreamMigrate(remote *leafNodeCfg) {
 		return
 	}
 
+	acc.jscmMu.Lock()
+	defer acc.jscmMu.Unlock()
+
 	// Walk all streams looking for any clustered stream, skip otherwise.
 	// If we are the leader force stepdown.
 	for _, mset := range acc.streams() {
diff --git a/server/memstore.go b/server/memstore.go
index 4d9b184b..df134e7d 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -261,7 +261,7 @@ func (ms *memStore) SkipMsg() uint64 {
 	ms.state.LastSeq = seq
 	ms.state.LastTime = now
 	if ms.state.Msgs == 0 {
-		ms.state.FirstSeq = seq
+		ms.state.FirstSeq = seq + 1
 		ms.state.FirstTime = now
 	} else {
 		ms.dmap.Insert(seq)
@@ -1600,7 +1600,8 @@ func (o *consumerMemStore) UpdateDelivered(dseq, sseq, dc uint64, ts int64) erro
 		// Check for an update to a message already delivered.
 		if sseq <= o.state.Delivered.Stream {
 			if p = o.state.Pending[sseq]; p != nil {
-				p.Sequence, p.Timestamp = dseq, ts
+				// Do not update p.Sequence, that should be the original delivery sequence.
+				p.Timestamp = ts
 			}
 		} else {
 			// Add to pending.
@@ -1649,23 +1650,38 @@ func (o *consumerMemStore) UpdateAcks(dseq, sseq uint64) error {
 	if o.cfg.AckPolicy == AckNone {
 		return ErrNoAckPolicy
 	}
-	if len(o.state.Pending) == 0 || o.state.Pending[sseq] == nil {
-		return ErrStoreMsgNotFound
-	}
 
 	// On restarts the old leader may get a replay from the raft logs that are old.
 	if dseq <= o.state.AckFloor.Consumer {
 		return nil
 	}
 
+	// Match leader logic on checking if ack is ahead of delivered.
+	// This could happen on a cooperative takeover with high speed deliveries.
+	if sseq > o.state.Delivered.Stream {
+		o.state.Delivered.Stream = sseq + 1
+	}
+
+	if len(o.state.Pending) == 0 || o.state.Pending[sseq] == nil {
+		delete(o.state.Redelivered, sseq)
+		return ErrStoreMsgNotFound
+	}
+
 	// Check for AckAll here.
 	if o.cfg.AckPolicy == AckAll {
 		sgap := sseq - o.state.AckFloor.Stream
 		o.state.AckFloor.Consumer = dseq
 		o.state.AckFloor.Stream = sseq
-		for seq := sseq; seq > sseq-sgap; seq-- {
-			delete(o.state.Pending, seq)
-			if len(o.state.Redelivered) > 0 {
+		if sgap > uint64(len(o.state.Pending)) {
+			for seq := range o.state.Pending {
+				if seq <= sseq {
+					delete(o.state.Pending, seq)
+					delete(o.state.Redelivered, seq)
+				}
+			}
+		} else {
+			for seq := sseq; seq > sseq-sgap && len(o.state.Pending) > 0; seq-- {
+				delete(o.state.Pending, seq)
 				delete(o.state.Redelivered, seq)
 			}
 		}
@@ -1677,23 +1693,20 @@ func (o *consumerMemStore) UpdateAcks(dseq, sseq uint64) error {
 	// First delete from our pending state.
 	if p, ok := o.state.Pending[sseq]; ok {
 		delete(o.state.Pending, sseq)
-		dseq = p.Sequence // Use the original.
-	}
-	// Now remove from redelivered.
-	if len(o.state.Redelivered) > 0 {
-		delete(o.state.Redelivered, sseq)
+		if dseq > p.Sequence && p.Sequence > 0 {
+			dseq = p.Sequence // Use the original.
+		}
 	}
 
 	if len(o.state.Pending) == 0 {
 		o.state.AckFloor.Consumer = o.state.Delivered.Consumer
 		o.state.AckFloor.Stream = o.state.Delivered.Stream
 	} else if dseq == o.state.AckFloor.Consumer+1 {
-		first := o.state.AckFloor.Consumer == 0
 		o.state.AckFloor.Consumer = dseq
 		o.state.AckFloor.Stream = sseq
 
-		if !first && o.state.Delivered.Consumer > dseq {
-			for ss := sseq + 1; ss < o.state.Delivered.Stream; ss++ {
+		if o.state.Delivered.Consumer > dseq {
+			for ss := sseq + 1; ss <= o.state.Delivered.Stream; ss++ {
 				if p, ok := o.state.Pending[ss]; ok {
 					if p.Sequence > 0 {
 						o.state.AckFloor.Consumer = p.Sequence - 1
@@ -1704,6 +1717,8 @@ func (o *consumerMemStore) UpdateAcks(dseq, sseq uint64) error {
 			}
 		}
 	}
+	// We do these regardless.
+	delete(o.state.Redelivered, sseq)
 
 	return nil
 }
diff --git a/server/monitor.go b/server/monitor.go
index e4023cba..c0c2bc08 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -1421,6 +1421,7 @@ func (s *Server) HandleRoot(w http.ResponseWriter, r *http.Request) {
 	<a href=.%s>Routes</a>
 	<a href=.%s>LeafNodes</a>
 	<a href=.%s>Gateways</a>
+	<a href=.%s>Raft Groups</a>
 	<a href=.%s class=last>Health Probe</a>
     <a href=https://docs.nats.io/running-a-nats-service/nats_admin/monitoring class="help">Help</a>
   </body>
@@ -1436,6 +1437,7 @@ func (s *Server) HandleRoot(w http.ResponseWriter, r *http.Request) {
 		s.basePath(RoutezPath),
 		s.basePath(LeafzPath),
 		s.basePath(GatewayzPath),
+		s.basePath(RaftzPath),
 		s.basePath(HealthzPath),
 	)
 }
@@ -3490,6 +3492,23 @@ func (s *Server) healthz(opts *HealthzOptions) *HealthStatus {
 		return health
 	}
 
+	// Are we still recovering meta layer?
+	if js.isMetaRecovering() {
+		if !details {
+			health.Status = na
+			health.Error = "JetStream is still recovering meta layer"
+
+		} else {
+			health.Errors = []HealthzError{
+				{
+					Type:  HealthzErrorJetStream,
+					Error: "JetStream is still recovering meta layer",
+				},
+			}
+		}
+		return health
+	}
+
 	// Range across all accounts, the streams assigned to them, and the consumers.
 	// If they are assigned to this server check their status.
 	ourID := meta.ID()
@@ -3724,3 +3743,138 @@ func (s *Server) profilez(opts *ProfilezOptions) *ProfilezStatus {
 		Profile: buffer.Bytes(),
 	}
 }
+
+type RaftzGroup struct {
+	ID            string                    `json:"id"`
+	State         string                    `json:"state"`
+	Size          int                       `json:"size"`
+	QuorumNeeded  int                       `json:"quorum_needed"`
+	Observer      bool                      `json:"observer,omitempty"`
+	Paused        bool                      `json:"paused,omitempty"`
+	Committed     uint64                    `json:"committed"`
+	Applied       uint64                    `json:"applied"`
+	CatchingUp    bool                      `json:"catching_up,omitempty"`
+	Leader        string                    `json:"leader,omitempty"`
+	EverHadLeader bool                      `json:"ever_had_leader"`
+	Term          uint64                    `json:"term"`
+	Vote          string                    `json:"voted_for,omitempty"`
+	PTerm         uint64                    `json:"pterm"`
+	PIndex        uint64                    `json:"pindex"`
+	IPQPropLen    int                       `json:"ipq_proposal_len"`
+	IPQEntryLen   int                       `json:"ipq_entry_len"`
+	IPQRespLen    int                       `json:"ipq_resp_len"`
+	IPQApplyLen   int                       `json:"ipq_apply_len"`
+	WAL           StreamState               `json:"wal"`
+	WALError      error                     `json:"wal_error,omitempty"`
+	Peers         map[string]RaftzGroupPeer `json:"peers"`
+}
+
+type RaftzGroupPeer struct {
+	Name                string `json:"name"`
+	Known               bool   `json:"known"`
+	LastReplicatedIndex uint64 `json:"last_replicated_index,omitempty"`
+	LastSeen            string `json:"last_seen,omitempty"`
+}
+
+func (s *Server) HandleRaftz(w http.ResponseWriter, r *http.Request) {
+	if s.raftNodes == nil {
+		w.WriteHeader(404)
+		w.Write([]byte("No Raft nodes registered"))
+		return
+	}
+
+	gfilter := r.URL.Query().Get("group")
+	afilter := r.URL.Query().Get("acc")
+	if afilter == "" {
+		afilter = s.SystemAccount().Name
+	}
+
+	groups := map[string]RaftNode{}
+	infos := map[string]map[string]RaftzGroup{} // account -> group ID
+
+	s.rnMu.RLock()
+	if gfilter != _EMPTY_ {
+		if rg, ok := s.raftNodes[gfilter]; ok && rg != nil {
+			if n, ok := rg.(*raft); ok {
+				if n.accName == afilter {
+					groups[gfilter] = rg
+				}
+			}
+		}
+	} else {
+		for name, rg := range s.raftNodes {
+			if rg == nil {
+				continue
+			}
+			if n, ok := rg.(*raft); ok {
+				if n.accName != afilter {
+					continue
+				}
+				groups[name] = rg
+			}
+		}
+	}
+	s.rnMu.RUnlock()
+
+	if len(groups) == 0 {
+		w.WriteHeader(404)
+		w.Write([]byte("No Raft nodes found, does the specified account/group exist?"))
+		return
+	}
+
+	for name, rg := range groups {
+		n, ok := rg.(*raft)
+		if n == nil || !ok {
+			continue
+		}
+		if _, ok := infos[n.accName]; !ok {
+			infos[n.accName] = map[string]RaftzGroup{}
+		}
+		// Only take the lock once, using the public RaftNode functions would
+		// cause us to take and release the locks over and over again.
+		n.RLock()
+		info := RaftzGroup{
+			ID:            n.id,
+			State:         RaftState(n.state.Load()).String(),
+			Size:          n.csz,
+			QuorumNeeded:  n.qn,
+			Observer:      n.observer,
+			Paused:        n.paused,
+			Committed:     n.commit,
+			Applied:       n.applied,
+			CatchingUp:    n.catchup != nil,
+			Leader:        n.leader,
+			EverHadLeader: n.pleader,
+			Term:          n.term,
+			Vote:          n.vote,
+			PTerm:         n.pterm,
+			PIndex:        n.pindex,
+			IPQPropLen:    n.prop.len(),
+			IPQEntryLen:   n.entry.len(),
+			IPQRespLen:    n.resp.len(),
+			IPQApplyLen:   n.apply.len(),
+			WALError:      n.werr,
+			Peers:         map[string]RaftzGroupPeer{},
+		}
+		n.wal.FastState(&info.WAL)
+		for id, p := range n.peers {
+			if id == n.id {
+				continue
+			}
+			peer := RaftzGroupPeer{
+				Name:                s.serverNameForNode(id),
+				Known:               p.kp,
+				LastReplicatedIndex: p.li,
+			}
+			if p.ts > 0 {
+				peer.LastSeen = time.Since(time.Unix(0, p.ts)).String()
+			}
+			info.Peers[id] = peer
+		}
+		n.RUnlock()
+		infos[n.accName][name] = info
+	}
+
+	b, _ := json.MarshalIndent(infos, "", "   ")
+	ResponseHandler(w, r, b)
+}
diff --git a/server/mqtt.go b/server/mqtt.go
index b69de16f..68ae6e70 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -1979,9 +1979,13 @@ func (as *mqttAccountSessionManager) processRetainedMsg(_ *subscription, c *clie
 	}
 	// If lastSeq is 0 (nothing to recover, or done doing it) and this is
 	// from our own server, ignore.
+	as.mu.RLock()
 	if as.rrmLastSeq == 0 && rm.Origin == as.jsa.id {
+		as.mu.RUnlock()
 		return
 	}
+	as.mu.RUnlock()
+
 	// At this point we either recover from our own server, or process a remote retained message.
 	seq, _, _ := ackReplyInfo(reply)
 
@@ -1989,11 +1993,13 @@ func (as *mqttAccountSessionManager) processRetainedMsg(_ *subscription, c *clie
 	as.handleRetainedMsg(rm.Subject, &mqttRetainedMsgRef{sseq: seq}, rm, false)
 
 	// If we were recovering (lastSeq > 0), then check if we are done.
+	as.mu.Lock()
 	if as.rrmLastSeq > 0 && seq >= as.rrmLastSeq {
 		as.rrmLastSeq = 0
 		close(as.rrmDoneCh)
 		as.rrmDoneCh = nil
 	}
+	as.mu.Unlock()
 }
 
 func (as *mqttAccountSessionManager) processRetainedMsgDel(_ *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
diff --git a/server/raft.go b/server/raft.go
index 8d12afdc..291c3c80 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -38,7 +38,7 @@ import (
 
 type RaftNode interface {
 	Propose(entry []byte) error
-	ProposeDirect(entries []*Entry) error
+	ProposeMulti(entries []*Entry) error
 	ForwardProposal(entry []byte) error
 	InstallSnapshot(snap []byte) error
 	SendSnapshot(snap []byte) error
@@ -85,6 +85,7 @@ type WAL interface {
 	RemoveMsg(index uint64) (bool, error)
 	Compact(index uint64) (uint64, error)
 	Purge() (uint64, error)
+	PurgeEx(subject string, seq, keep uint64) (uint64, error)
 	Truncate(seq uint64) error
 	State() StreamState
 	FastState(*StreamState)
@@ -155,25 +156,27 @@ type raft struct {
 	llqrt  time.Time   // Last quorum lost time
 	lsut   time.Time   // Last scale-up time
 
-	term     uint64 // The current vote term
-	pterm    uint64 // Previous term from the last snapshot
-	pindex   uint64 // Previous index from the last snapshot
-	commit   uint64 // Sequence number of the most recent commit
-	applied  uint64 // Sequence number of the most recently applied commit
-	hcbehind bool   // Were we falling behind at the last health check? (see: isCurrent)
+	term    uint64 // The current vote term
+	pterm   uint64 // Previous term from the last snapshot
+	pindex  uint64 // Previous index from the last snapshot
+	commit  uint64 // Index of the most recent commit
+	applied uint64 // Index of the most recently applied commit
 
 	leader string // The ID of the leader
 	vote   string // Our current vote state
 	lxfer  bool   // Are we doing a leadership transfer?
 
+	hcbehind bool // Were we falling behind at the last health check? (see: isCurrent)
+
 	s  *Server    // Reference to top-level server
 	c  *client    // Internal client for subscriptions
 	js *jetStream // JetStream, if running, to see if we are out of resources
 
-	dflag    bool           // Debug flag
-	pleader  bool           // Has the group ever had a leader?
-	observer bool           // The node is observing, i.e. not participating in voting
-	extSt    extensionState // Extension state
+	dflag    bool // Debug flag
+	pleader  bool // Has the group ever had a leader?
+	observer bool // The node is observing, i.e. not participating in voting
+
+	extSt extensionState // Extension state
 
 	psubj  string // Proposals subject
 	rpsubj string // Remove peers subject
@@ -232,16 +235,18 @@ const (
 	hbIntervalDefault              = 1 * time.Second
 	lostQuorumIntervalDefault      = hbIntervalDefault * 10 // 10 seconds
 	lostQuorumCheckIntervalDefault = hbIntervalDefault * 10 // 10 seconds
+	observerModeIntervalDefault    = 48 * time.Hour
 )
 
 var (
-	minElectionTimeout = minElectionTimeoutDefault
-	maxElectionTimeout = maxElectionTimeoutDefault
-	minCampaignTimeout = minCampaignTimeoutDefault
-	maxCampaignTimeout = maxCampaignTimeoutDefault
-	hbInterval         = hbIntervalDefault
-	lostQuorumInterval = lostQuorumIntervalDefault
-	lostQuorumCheck    = lostQuorumCheckIntervalDefault
+	minElectionTimeout   = minElectionTimeoutDefault
+	maxElectionTimeout   = maxElectionTimeoutDefault
+	minCampaignTimeout   = minCampaignTimeoutDefault
+	maxCampaignTimeout   = maxCampaignTimeoutDefault
+	hbInterval           = hbIntervalDefault
+	lostQuorumInterval   = lostQuorumIntervalDefault
+	lostQuorumCheck      = lostQuorumCheckIntervalDefault
+	observerModeInterval = observerModeIntervalDefault
 )
 
 type RaftConfig struct {
@@ -387,7 +392,7 @@ func (s *Server) startRaftNode(accName string, cfg *RaftConfig, labels pprofLabe
 		apply:    newIPQueue[*CommittedEntry](s, qpfx+"committedEntry"),
 		stepdown: newIPQueue[string](s, qpfx+"stepdown"),
 		accName:  accName,
-		leadc:    make(chan bool, 1),
+		leadc:    make(chan bool, 32),
 		observer: cfg.Observer,
 		extSt:    ps.domainExt,
 	}
@@ -414,7 +419,8 @@ func (s *Server) startRaftNode(accName string, cfg *RaftConfig, labels pprofLabe
 		return nil, fmt.Errorf("could not create snapshots directory - %v", err)
 	}
 
-	// Can't recover snapshots if memory based.
+	// Can't recover snapshots if memory based since wal will be reset.
+	// We will inherit from the current leader.
 	if _, ok := n.wal.(*memStore); ok {
 		os.Remove(filepath.Join(n.sd, snapshotsDir, "*"))
 	} else {
@@ -692,36 +698,34 @@ func (n *raft) Propose(data []byte) error {
 		n.debug("Proposal ignored, not leader (state: %v)", state)
 		return errNotLeader
 	}
-	n.RLock()
+	n.Lock()
+	defer n.Unlock()
+
 	// Error if we had a previous write error.
 	if werr := n.werr; werr != nil {
-		n.RUnlock()
 		return werr
 	}
-	prop := n.prop
-	n.RUnlock()
-
-	prop.push(newEntry(EntryNormal, data))
+	n.prop.push(newEntry(EntryNormal, data))
 	return nil
 }
 
-// ProposeDirect will propose entries directly by skipping the Raft state
-// machine and sending them straight to the wire instead.
+// ProposeDirect will propose multiple entries at once.
 // This should only be called on the leader.
-func (n *raft) ProposeDirect(entries []*Entry) error {
+func (n *raft) ProposeMulti(entries []*Entry) error {
 	if state := n.State(); state != Leader {
 		n.debug("Direct proposal ignored, not leader (state: %v)", state)
 		return errNotLeader
 	}
-	n.RLock()
+	n.Lock()
+	defer n.Unlock()
+
 	// Error if we had a previous write error.
 	if werr := n.werr; werr != nil {
-		n.RUnlock()
 		return werr
 	}
-	n.RUnlock()
-
-	n.sendAppendEntry(entries)
+	for _, e := range entries {
+		n.prop.push(e)
+	}
 	return nil
 }
 
@@ -871,7 +875,7 @@ func (n *raft) PauseApply() error {
 	n.hcommit = n.commit
 	// Also prevent us from trying to become a leader while paused and catching up.
 	n.pobserver, n.observer = n.observer, true
-	n.resetElect(48 * time.Hour)
+	n.resetElect(observerModeInterval)
 
 	return nil
 }
@@ -1012,25 +1016,20 @@ func (n *raft) InstallSnapshot(data []byte) error {
 	}
 
 	n.Lock()
+	defer n.Unlock()
 
 	// If a write error has occurred already then stop here.
 	if werr := n.werr; werr != nil {
-		n.Unlock()
 		return werr
 	}
 
 	// Check that a catchup isn't already taking place. If it is then we won't
 	// allow installing snapshots until it is done.
 	if len(n.progress) > 0 {
-		n.Unlock()
 		return errCatchupsRunning
 	}
 
-	var state StreamState
-	n.wal.FastState(&state)
-
 	if n.applied == 0 {
-		n.Unlock()
 		return errNoSnapAvailable
 	}
 
@@ -1055,6 +1054,12 @@ func (n *raft) InstallSnapshot(data []byte) error {
 		data:      data,
 	}
 
+	return n.installSnapshot(snap)
+}
+
+// Install the snapshot.
+// Lock should be held.
+func (n *raft) installSnapshot(snap *snapshot) error {
 	snapDir := filepath.Join(n.sd, snapshotsDir)
 	sn := fmt.Sprintf(snapFileT, snap.lastTerm, snap.lastIndex)
 	sfile := filepath.Join(snapDir, sn)
@@ -1064,29 +1069,21 @@ func (n *raft) InstallSnapshot(data []byte) error {
 	dios <- struct{}{}
 
 	if err != nil {
-		n.Unlock()
 		// We could set write err here, but if this is a temporary situation, too many open files etc.
 		// we want to retry and snapshots are not fatal.
 		return err
 	}
 
+	// Delete our previous snapshot file if it exists.
+	if n.snapfile != _EMPTY_ && n.snapfile != sfile {
+		os.Remove(n.snapfile)
+	}
 	// Remember our latest snapshot file.
 	n.snapfile = sfile
 	if _, err := n.wal.Compact(snap.lastIndex + 1); err != nil {
 		n.setWriteErrLocked(err)
-		n.Unlock()
 		return err
 	}
-	n.Unlock()
-
-	psnaps, _ := os.ReadDir(snapDir)
-	// Remove any old snapshots.
-	for _, fi := range psnaps {
-		pn := fi.Name()
-		if pn != sn {
-			os.Remove(filepath.Join(snapDir, pn))
-		}
-	}
 
 	return nil
 }
@@ -1888,8 +1885,16 @@ func (n *raft) SetObserver(isObserver bool) {
 func (n *raft) setObserver(isObserver bool, extSt extensionState) {
 	n.Lock()
 	defer n.Unlock()
+
+	wasObserver := n.observer
 	n.observer = isObserver
 	n.extSt = extSt
+
+	// If we're leaving observer state then reset the election timer or
+	// we might end up waiting for up to the observerModeInterval.
+	if wasObserver && !isObserver {
+		n.resetElect(randCampaignTimeout())
+	}
 }
 
 // processAppendEntries is called by the Raft state machine when there are
@@ -1939,7 +1944,7 @@ func (n *raft) runAsFollower() {
 				n.resetElectionTimeoutWithLock()
 				n.debug("Not switching to candidate, no resources")
 			} else if n.IsObserver() {
-				n.resetElectWithLock(48 * time.Hour)
+				n.resetElectWithLock(observerModeInterval)
 				n.debug("Not switching to candidate, observer only")
 			} else if n.isCatchingUp() {
 				n.debug("Not switching to candidate, catching up")
@@ -2450,8 +2455,10 @@ func (n *raft) lostQuorum() bool {
 }
 
 func (n *raft) lostQuorumLocked() bool {
-	// Make sure we let any scale up actions settle before deciding.
-	if !n.lsut.IsZero() && time.Since(n.lsut) < lostQuorumInterval {
+	// In order to avoid false positives that can happen in heavily loaded systems
+	// make sure nothing is queued up that we have not processed yet.
+	// Also make sure we let any scale up actions settle before deciding.
+	if n.resp.len() != 0 || (!n.lsut.IsZero() && time.Since(n.lsut) < lostQuorumInterval) {
 		return false
 	}
 
@@ -3080,17 +3087,20 @@ func (n *raft) truncateWAL(term, index uint64) {
 
 	if err := n.wal.Truncate(index); err != nil {
 		// If we get an invalid sequence, reset our wal all together.
+		// We will not have holes, so this means we do not have this message stored anymore.
 		if err == ErrInvalidSequence {
 			n.debug("Resetting WAL")
 			n.wal.Truncate(0)
-			index, n.term, n.pterm, n.pindex = 0, 0, 0, 0
+			// If our index is non-zero use PurgeEx to set us to the correct next index.
+			if index > 0 {
+				n.wal.PurgeEx(fwcs, index+1, 0)
+			}
 		} else {
 			n.warn("Error truncating WAL: %v", err)
 			n.setWriteErrLocked(err)
+			return
 		}
-		return
 	}
-
 	// Set after we know we have truncated properly.
 	n.term, n.pterm, n.pindex = term, term, index
 }
@@ -3159,15 +3169,17 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	// to a follower of that node instead.
 	if n.State() == Candidate {
 		// Ignore old terms, otherwise we might end up stepping down incorrectly.
-		if ae.term >= n.term {
+		// Needs to be ahead of our pterm (last log index), as an isolated node
+		// could have bumped its vote term up considerably past this point.
+		if ae.term >= n.pterm {
 			// If the append entry term is newer than the current term, erase our
 			// vote.
 			if ae.term > n.term {
-				n.term = ae.term
 				n.vote = noVote
-				n.writeTermVote()
 			}
 			n.debug("Received append entry in candidate state from %q, converting to follower", ae.leader)
+			n.term = ae.term
+			n.writeTermVote()
 			n.stepdown.push(ae.leader)
 		}
 	}
@@ -3262,7 +3274,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 				// If terms mismatched, or we got an error loading, delete that entry and all others past it.
 				// Make sure to cancel any catchups in progress.
 				// Truncate will reset our pterm and pindex. Only do so if we have an entry.
-				n.truncateWAL(ae.pterm, ae.pindex)
+				n.truncateWAL(eae.pterm, eae.pindex)
 			}
 			// Cancel regardless.
 			n.cancelCatchup()
@@ -3309,6 +3321,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 				return
 			}
 
+			// Inherit state from appendEntry with the leader's snapshot.
 			n.pindex = ae.pindex
 			n.pterm = ae.pterm
 			n.commit = ae.pindex
@@ -3319,6 +3332,19 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 				return
 			}
 
+			snap := &snapshot{
+				lastTerm:  n.pterm,
+				lastIndex: n.pindex,
+				peerstate: encodePeerState(&peerState{n.peerNames(), n.csz, n.extSt}),
+				data:      ae.entries[0].Data,
+			}
+			// Install the leader's snapshot as our own.
+			if err := n.installSnapshot(snap); err != nil {
+				n.setWriteErrLocked(err)
+				n.Unlock()
+				return
+			}
+
 			// Now send snapshot to upper levels. Only send the snapshot, not the peerstate entry.
 			n.apply.push(newCommittedEntry(n.commit, ae.entries[:1]))
 			n.Unlock()
diff --git a/server/server.go b/server/server.go
index 10718c10..fc8c8863 100644
--- a/server/server.go
+++ b/server/server.go
@@ -2892,6 +2892,7 @@ const (
 	JszPath          = "/jsz"
 	HealthzPath      = "/healthz"
 	IPQueuesPath     = "/ipqueuesz"
+	RaftzPath        = "/raftz"
 )
 
 func (s *Server) basePath(p string) string {
@@ -3006,6 +3007,8 @@ func (s *Server) startMonitoring(secure bool) error {
 	mux.HandleFunc(s.basePath(HealthzPath), s.HandleHealthz)
 	// IPQueuesz
 	mux.HandleFunc(s.basePath(IPQueuesPath), s.HandleIPQueuesz)
+	// Raftz
+	mux.HandleFunc(s.basePath(RaftzPath), s.HandleRaftz)
 
 	// Do not set a WriteTimeout because it could cause cURL/browser
 	// to return empty response or unable to display page if the
@@ -4276,6 +4279,7 @@ func (s *Server) lameDuckMode() {
 		}
 	}
 	s.Shutdown()
+	s.WaitForShutdown()
 }
 
 // Send an INFO update to routes with the indication that this server is in LDM mode.
diff --git a/server/signal.go b/server/signal.go
index aa133b4f..aad65e82 100644
--- a/server/signal.go
+++ b/server/signal.go
@@ -51,6 +51,7 @@ func (s *Server) handleSignals() {
 				switch sig {
 				case syscall.SIGINT:
 					s.Shutdown()
+					s.WaitForShutdown()
 					os.Exit(0)
 				case syscall.SIGTERM:
 					// Shutdown unless graceful shutdown already in progress.
@@ -60,6 +61,7 @@ func (s *Server) handleSignals() {
 
 					if !ldm {
 						s.Shutdown()
+						s.WaitForShutdown()
 						os.Exit(1)
 					}
 				case syscall.SIGUSR1:
diff --git a/server/stream.go b/server/stream.go
index de378651..65e0a574 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -468,7 +468,7 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 		}
 	}
 	jsa.usageMu.RLock()
-	selected, tier, hasTier := jsa.selectLimits(&cfg)
+	selected, tier, hasTier := jsa.selectLimits(cfg.Replicas)
 	jsa.usageMu.RUnlock()
 	reserved := int64(0)
 	if !isClustered {
@@ -864,7 +864,11 @@ func (mset *stream) setLeader(isLeader bool) error {
 		if mset.sourcesConsumerSetup != nil {
 			mset.sourcesConsumerSetup.Stop()
 			mset.sourcesConsumerSetup = nil
+		} else {
+			// Stop any source consumers
+			mset.stopSourceConsumers()
 		}
+
 		// Stop responding to sync requests.
 		mset.stopClusterSubs()
 		// Unsubscribe from direct stream.
@@ -1488,19 +1492,38 @@ func (s *Server) checkStreamCfg(config *StreamConfig, acc *Account) (StreamConfi
 		}
 
 		// Check for literal duplication of subject interest in config
-		// and no overlap with any JS API subject space
+		// and no overlap with any JS or SYS API subject space.
 		dset := make(map[string]struct{}, len(cfg.Subjects))
 		for _, subj := range cfg.Subjects {
+			// Make sure the subject is valid. Check this first.
+			if !IsValidSubject(subj) {
+				return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("invalid subject"))
+			}
 			if _, ok := dset[subj]; ok {
 				return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("duplicate subjects detected"))
 			}
+			// Check for trying to capture everything.
+			if subj == fwcs {
+				if !cfg.NoAck {
+					return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("capturing all subjects requires no-ack to be true"))
+				}
+				// Capturing everything also will require R1.
+				if cfg.Replicas != 1 {
+					return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("capturing all subjects requires replicas of 1"))
+				}
+			}
 			// Also check to make sure we do not overlap with our $JS API subjects.
-			if subjectIsSubsetMatch(subj, "$JS.API.>") {
-				return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("subjects overlap with jetstream api"))
+			if !cfg.NoAck && (subjectIsSubsetMatch(subj, "$JS.>") || subjectIsSubsetMatch(subj, "$JSC.>")) {
+				// We allow an exception for $JS.EVENT.> since these could have been created in the past.
+				if !subjectIsSubsetMatch(subj, "$JS.EVENT.>") {
+					return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("subjects that overlap with jetstream api require no-ack to be true"))
+				}
 			}
-			// Make sure the subject is valid.
-			if !IsValidSubject(subj) {
-				return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("invalid subject"))
+			// And the $SYS subjects.
+			if !cfg.NoAck && subjectIsSubsetMatch(subj, "$SYS.>") {
+				if !subjectIsSubsetMatch(subj, "$SYS.ACCOUNT.>") {
+					return StreamConfig{}, NewJSStreamInvalidConfigError(fmt.Errorf("subjects that overlap with system api require no-ack to be true"))
+				}
 			}
 			// Mark for duplicate check.
 			dset[subj] = struct{}{}
@@ -1668,9 +1691,9 @@ func (jsa *jsAccount) configUpdateCheck(old, new *StreamConfig, s *Server) (*Str
 	jsa.mu.RLock()
 	acc := jsa.account
 	jsa.usageMu.RLock()
-	selected, tier, hasTier := jsa.selectLimits(&cfg)
+	selected, tier, hasTier := jsa.selectLimits(cfg.Replicas)
 	if !hasTier && old.Replicas != cfg.Replicas {
-		selected, tier, hasTier = jsa.selectLimits(old)
+		selected, tier, hasTier = jsa.selectLimits(old.Replicas)
 	}
 	jsa.usageMu.RUnlock()
 	reserved := int64(0)
@@ -1823,7 +1846,6 @@ func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool)
 							var err error
 							si.trs[i], err = NewSubjectTransform(s.SubjectTransforms[i].Source, s.SubjectTransforms[i].Destination)
 							if err != nil {
-								mset.mu.Unlock()
 								mset.srv.Errorf("Unable to get subject transform for source: %v", err)
 							}
 						}
@@ -1905,7 +1927,7 @@ func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool)
 
 	js := mset.js
 
-	if targetTier := tierName(cfg); mset.tier != targetTier {
+	if targetTier := tierName(cfg.Replicas); mset.tier != targetTier {
 		// In cases such as R1->R3, only one update is needed
 		jsa.usageMu.RLock()
 		_, ok := jsa.limits[targetTier]
@@ -2193,9 +2215,11 @@ func (mset *stream) processMirrorMsgs(mirror *sourceInfo, ready *sync.WaitGroup)
 			msgs.recycle(&ims)
 		case <-t.C:
 			mset.mu.RLock()
+			var stalled bool
+			if mset.mirror != nil {
+				stalled = time.Since(time.Unix(0, mset.mirror.last.Load())) > sourceHealthCheckInterval
+			}
 			isLeader := mset.isLeader()
-			last := time.Unix(0, mset.mirror.last.Load())
-			stalled := mset.mirror != nil && time.Since(last) > sourceHealthCheckInterval
 			mset.mu.RUnlock()
 			// No longer leader.
 			if !isLeader {
@@ -2412,14 +2436,14 @@ func (mset *stream) skipMsgs(start, end uint64) {
 		return
 	}
 
-	// FIXME (dlc) - We should allow proposals of DeleteEange, but would need to make sure all peers support.
+	// FIXME (dlc) - We should allow proposals of DeleteRange, but would need to make sure all peers support.
 	// With syncRequest was easy to add bool into request.
 	var entries []*Entry
 	for seq := start; seq <= end; seq++ {
-		entries = append(entries, &Entry{EntryNormal, encodeStreamMsg(_EMPTY_, _EMPTY_, nil, nil, seq-1, 0)})
+		entries = append(entries, newEntry(EntryNormal, encodeStreamMsg(_EMPTY_, _EMPTY_, nil, nil, seq-1, 0)))
 		// So a single message does not get too big.
 		if len(entries) > 10_000 {
-			node.ProposeDirect(entries)
+			node.ProposeMulti(entries)
 			// We need to re-create `entries` because there is a reference
 			// to it in the node's pae map.
 			entries = entries[:0]
@@ -2427,7 +2451,7 @@ func (mset *stream) skipMsgs(start, end uint64) {
 	}
 	// Send all at once.
 	if len(entries) > 0 {
-		node.ProposeDirect(entries)
+		node.ProposeMulti(entries)
 	}
 }
 
@@ -5570,7 +5594,7 @@ func (mset *stream) checkInterestState() {
 	defer mset.mu.Unlock()
 
 	// Check which purge we need to perform.
-	if lowAckFloor <= state.LastSeq {
+	if lowAckFloor <= state.LastSeq || state.Msgs == 0 {
 		// Purge the stream to lowest ack floor + 1
 		mset.store.PurgeEx(_EMPTY_, lowAckFloor+1, 0)
 	} else {
@@ -6096,6 +6120,8 @@ func (a *Account) RestoreStream(ncfg *StreamConfig, r io.Reader) (*stream, error
 	}
 	mset, err := a.addStream(&cfg)
 	if err != nil {
+		// Make sure to clean up after ourselves here.
+		os.RemoveAll(ndir)
 		return nil, err
 	}
 	if !fcfg.Created.IsZero() {
@@ -6231,3 +6257,10 @@ func (mset *stream) clearMonitorRunning() {
 	defer mset.mu.Unlock()
 	mset.inMonitor = false
 }
+
+// Check if our monitor is running.
+func (mset *stream) isMonitorRunning() bool {
+	mset.mu.RLock()
+	defer mset.mu.RUnlock()
+	return mset.inMonitor
+}
diff --git a/server/stree/leaf.go b/server/stree/leaf.go
index 839450f2..119837ec 100644
--- a/server/stree/leaf.go
+++ b/server/stree/leaf.go
@@ -18,16 +18,17 @@ import (
 )
 
 // Leaf node
+// Order of struct fields for best memory alignment (as per govet/fieldalignment)
 type leaf[T any] struct {
+	value T
 	// This could be the whole subject, but most likely just the suffix portion.
 	// We will only store the suffix here and assume all prior prefix paths have
 	// been checked once we arrive at this leafnode.
 	suffix []byte
-	value  T
 }
 
 func newLeaf[T any](suffix []byte, value T) *leaf[T] {
-	return &leaf[T]{copyBytes(suffix), value}
+	return &leaf[T]{value, copyBytes(suffix)}
 }
 
 func (n *leaf[T]) isLeaf() bool                               { return true }
diff --git a/server/stree/node16.go b/server/stree/node16.go
index 2d206afd..7da5df89 100644
--- a/server/stree/node16.go
+++ b/server/stree/node16.go
@@ -14,10 +14,11 @@
 package stree
 
 // Node with 16 children
+// Order of struct fields for best memory alignment (as per govet/fieldalignment)
 type node16 struct {
-	meta
 	child [16]node
-	key   [16]byte
+	meta
+	key [16]byte
 }
 
 func newNode16(prefix []byte) *node16 {
diff --git a/server/stree/node256.go b/server/stree/node256.go
index fdadde0b..f5bf69bc 100644
--- a/server/stree/node256.go
+++ b/server/stree/node256.go
@@ -14,9 +14,10 @@
 package stree
 
 // Node with 256 children
+// Order of struct fields for best memory alignment (as per govet/fieldalignment)
 type node256 struct {
-	meta
 	child [256]node
+	meta
 }
 
 func newNode256(prefix []byte) *node256 {
diff --git a/server/stree/node4.go b/server/stree/node4.go
index 2d489625..6aeb024a 100644
--- a/server/stree/node4.go
+++ b/server/stree/node4.go
@@ -14,10 +14,11 @@
 package stree
 
 // Node with 4 children
+// Order of struct fields for best memory alignment (as per govet/fieldalignment)
 type node4 struct {
-	meta
 	child [4]node
-	key   [4]byte
+	meta
+	key [4]byte
 }
 
 func newNode4(prefix []byte) *node4 {
diff --git a/util/nats-server-hardened.service b/util/nats-server-hardened.service
index 79b88b0d..0a1ad182 100644
--- a/util/nats-server-hardened.service
+++ b/util/nats-server-hardened.service
@@ -81,4 +81,6 @@ ReadWritePaths=/var/lib/nats
 
 [Install]
 WantedBy=multi-user.target
-Alias=nats.service
+# If you install this service as nats-server.service and want 'nats'
+# to work as an alias, then uncomment this next line:
+#Alias=nats.service


Test patch:
diff --git a/.github/workflows/MQTT_test.yaml b/.github/workflows/MQTT_test.yaml
deleted file mode 100644
index 2bfd0bed..00000000
--- a/.github/workflows/MQTT_test.yaml
+++ /dev/null
@@ -1,67 +0,0 @@
-name: MQTTEx
-on: [push, pull_request]
-
-permissions:
-  pull-requests: write # to comment on PRs
-  contents: write # to comment on commits (to upload artifacts?)
-
-jobs:
-  test:
-    env:
-      GOPATH: /home/runner/work/nats-server
-      GO111MODULE: "on"
-    runs-on: ubuntu-latest
-    steps:
-      - name: Checkout code
-        uses: actions/checkout@v4
-        with:
-          path: src/github.com/nats-io/nats-server
-
-      - name: Setup Go
-        uses: actions/setup-go@v5
-        with:
-          go-version-file: src/github.com/nats-io/nats-server/go.mod
-          cache-dependency-path: src/github.com/nats-io/nats-server/go.sum
-
-      - name: Set up testing tools and environment
-        shell: bash --noprofile --norc -eo pipefail {0}
-        id: setup
-        run: |
-          wget https://github.com/hivemq/mqtt-cli/releases/download/v4.20.0/mqtt-cli-4.20.0.deb
-          sudo apt install ./mqtt-cli-4.20.0.deb
-          go install github.com/ConnectEverything/mqtt-test@4dd571c31318dcfebe5443242f53f262403ceafb
-
-      # - name: Download benchmark result for ${{ github.base_ref || github.ref_name }}
-      #   uses: dawidd6/action-download-artifact@v2
-      #   continue-on-error: true
-      #   with:
-      #     path: src/github.com/nats-io/nats-server/bench
-      #     name: bench-output-${{ runner.os }}
-      #     branch: ${{ github.base_ref || github.ref }}
-
-      - name: Run tests and benchmarks
-        shell: bash --noprofile --norc -eo pipefail {0}
-        run: |
-          cd src/github.com/nats-io/nats-server
-          go test -v --run='MQTTEx' ./server
-          # go test --run='-' --count=10 --bench 'MQTT_' ./server | tee output.txt
-          # go test --run='-' --count=10 --bench 'MQTTEx' --timeout=20m --benchtime=100x ./server | tee -a output.txt
-          go test --run='-' --count=3 --bench 'MQTTEx' --benchtime=100x ./server
-
-      # - name: Compare benchmarks
-      #   uses: benchmark-action/github-action-benchmark@v1
-      #   with:
-      #     tool: go
-      #     output-file-path: src/github.com/nats-io/nats-server/output.txt
-      #     github-token: ${{ secrets.GITHUB_TOKEN }}
-      #     alert-threshold: 140%
-      #     comment-on-alert: true
-      #     # fail-on-alert: true
-      #     external-data-json-path: src/github.com/nats-io/nats-server/bench/benchmark-data.json
-
-      # - name: Store benchmark result for ${{ github.ref_name }}
-      #   if: always()
-      #   uses: actions/upload-artifact@v3
-      #   with:
-      #     path: src/github.com/nats-io/nats-server/bench
-      #     name: bench-output-${{ runner.os }}
diff --git a/.github/workflows/mqtt-test.yaml b/.github/workflows/mqtt-test.yaml
new file mode 100644
index 00000000..893837e6
--- /dev/null
+++ b/.github/workflows/mqtt-test.yaml
@@ -0,0 +1,48 @@
+name: MQTT external test
+on: [pull_request]
+
+jobs:
+  test:
+    env:
+      GOPATH: /home/runner/work/nats-server
+      GO111MODULE: "on"
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+        with:
+          path: src/github.com/nats-io/nats-server
+
+      - name: Setup Go
+        uses: actions/setup-go@v5
+        with:
+          go-version-file: src/github.com/nats-io/nats-server/go.mod
+          cache-dependency-path: src/github.com/nats-io/nats-server/go.sum
+
+      - name: Set up testing tools and environment
+        shell: bash --noprofile --norc -eo pipefail {0}
+        id: setup
+        run: |
+          wget https://github.com/hivemq/mqtt-cli/releases/download/v4.20.0/mqtt-cli-4.20.0.deb
+          sudo apt install ./mqtt-cli-4.20.0.deb
+          go install github.com/ConnectEverything/mqtt-test@v0.1.0
+
+      - name: Run tests (3 times to detect flappers)
+        shell: bash --noprofile --norc -eo pipefail {0}
+        run: |
+          cd src/github.com/nats-io/nats-server
+          go test -v --count=3 --run='TestXMQTT' ./server
+
+      - name: Run tests with --race
+        shell: bash --noprofile --norc -eo pipefail {0}
+        run: |
+          cd src/github.com/nats-io/nats-server
+          go test -v --race --failfast --run='TestXMQTT' ./server
+
+      - name: Run benchmarks
+        shell: bash --noprofile --norc -eo pipefail {0}
+        run: |
+          cd src/github.com/nats-io/nats-server
+          go test --run='-' --count=3 --bench 'BenchmarkXMQTT' --benchtime=100x ./server
+
+      # TODO: compare benchmarks
diff --git a/server/auth_callout_test.go b/server/auth_callout_test.go
index d8634d60..1458dca2 100644
--- a/server/auth_callout_test.go
+++ b/server/auth_callout_test.go
@@ -20,9 +20,11 @@ import (
 	"encoding/pem"
 	"errors"
 	"fmt"
+	"os"
 	"reflect"
 	"sort"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -1827,3 +1829,174 @@ func TestAuthCallout_ClientAuthErrorOperatorMode(t *testing.T) {
 	testAuthCall_ClientAuthErrorOperatorMode(t, true)
 	testAuthCall_ClientAuthErrorOperatorMode(t, false)
 }
+
+func TestOperatorModeUserRevocation(t *testing.T) {
+	skp, spub := createKey(t)
+	sysClaim := jwt.NewAccountClaims(spub)
+	sysClaim.Name = "$SYS"
+	sysJwt, err := sysClaim.Encode(oKp)
+	require_NoError(t, err)
+
+	// TEST account.
+	tkp, tpub := createKey(t)
+	accClaim := jwt.NewAccountClaims(tpub)
+	accClaim.Name = "TEST"
+	accJwt, err := accClaim.Encode(oKp)
+	require_NoError(t, err)
+
+	// AUTH service account.
+	akp, err := nkeys.FromSeed([]byte(authCalloutIssuerSeed))
+	require_NoError(t, err)
+
+	apub, err := akp.PublicKey()
+	require_NoError(t, err)
+
+	// The authorized user for the service.
+	upub, creds := createAuthServiceUser(t, akp)
+	defer removeFile(t, creds)
+
+	authClaim := jwt.NewAccountClaims(apub)
+	authClaim.Name = "AUTH"
+	authClaim.EnableExternalAuthorization(upub)
+	authClaim.Authorization.AllowedAccounts.Add(tpub)
+	authJwt, err := authClaim.Encode(oKp)
+	require_NoError(t, err)
+
+	jwtDir, err := os.MkdirTemp("", "")
+	require_NoError(t, err)
+	defer func() {
+		_ = os.RemoveAll(jwtDir)
+	}()
+
+	conf := fmt.Sprintf(`
+		listen: 127.0.0.1:-1
+		operator: %s
+		system_account: %s
+		resolver: {
+			type: "full"
+			dir: %s
+			allow_delete: false
+			interval: "2m"
+			timeout: "1.9s"
+		}
+		resolver_preload: {
+			%s: %s
+			%s: %s
+			%s: %s
+		}
+    `, ojwt, spub, jwtDir, apub, authJwt, tpub, accJwt, spub, sysJwt)
+
+	const token = "--secret--"
+
+	users := make(map[string]string)
+	handler := func(m *nats.Msg) {
+		user, si, _, opts, _ := decodeAuthRequest(t, m.Data)
+		if opts.Token == token {
+			// must have no limits set
+			ujwt := createAuthUser(t, user, "user", tpub, tpub, tkp, 0, &jwt.UserPermissionLimits{})
+			users[opts.Name] = ujwt
+			m.Respond(serviceResponse(t, user, si.ID, ujwt, "", 0))
+		} else {
+			m.Respond(nil)
+		}
+	}
+
+	ac := NewAuthTest(t, conf, handler, nats.UserCredentials(creds))
+	defer ac.Cleanup()
+
+	// create a system user
+	_, sysCreds := createAuthServiceUser(t, skp)
+	defer removeFile(t, sysCreds)
+	// connect the system user
+	sysNC, err := ac.NewClient(nats.UserCredentials(sysCreds))
+	require_NoError(t, err)
+
+	// Bearer token etc..
+	// This is used by all users, and the customization will be in other connect args.
+	// This needs to also be bound to the authorization account.
+	creds = createBasicAccountUser(t, akp)
+	defer removeFile(t, creds)
+
+	var fwg sync.WaitGroup
+
+	// connect three clients
+	nc := ac.Connect(nats.UserCredentials(creds), nats.Name("first"), nats.Token(token), nats.NoReconnect(), nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) {
+		if err != nil && strings.Contains(err.Error(), "authentication revoked") {
+			fwg.Done()
+		}
+	}))
+	fwg.Add(1)
+
+	var swg sync.WaitGroup
+	// connect another user
+	ncA := ac.Connect(nats.UserCredentials(creds), nats.Token(token), nats.Name("second"), nats.NoReconnect(), nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) {
+		if err != nil && strings.Contains(err.Error(), "authentication revoked") {
+			swg.Done()
+		}
+	}))
+	swg.Add(1)
+
+	ncB := ac.Connect(nats.UserCredentials(creds), nats.Token(token), nats.NoReconnect(), nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) {
+		if err != nil && strings.Contains(err.Error(), "authentication revoked") {
+			swg.Done()
+		}
+	}))
+	swg.Add(1)
+
+	require_NoError(t, err)
+
+	// revoke the user first - look at the JWT we issued
+	uc, err := jwt.DecodeUserClaims(users["first"])
+	require_NoError(t, err)
+	// revoke the user in account
+	accClaim.Revocations = make(map[string]int64)
+	accClaim.Revocations.Revoke(uc.Subject, time.Now().Add(time.Minute))
+	accJwt, err = accClaim.Encode(oKp)
+	require_NoError(t, err)
+	// send the update request
+	updateAccount(t, sysNC, accJwt)
+
+	// wait for the user to be disconnected with the error we expect
+	fwg.Wait()
+	require_Equal(t, nc.IsConnected(), false)
+
+	// update the account to remove any revocations
+	accClaim.Revocations = make(map[string]int64)
+	accJwt, err = accClaim.Encode(oKp)
+	require_NoError(t, err)
+	updateAccount(t, sysNC, accJwt)
+	// we should still be connected on the other 2 clients
+	require_Equal(t, ncA.IsConnected(), true)
+	require_Equal(t, ncB.IsConnected(), true)
+
+	// update the jwt and revoke all users
+	accClaim.Revocations.Revoke(jwt.All, time.Now().Add(time.Minute))
+	accJwt, err = accClaim.Encode(oKp)
+	require_NoError(t, err)
+	updateAccount(t, sysNC, accJwt)
+
+	swg.Wait()
+	require_Equal(t, ncA.IsConnected(), false)
+	require_Equal(t, ncB.IsConnected(), false)
+}
+
+func updateAccount(t *testing.T, sys *nats.Conn, jwtToken string) {
+	ac, err := jwt.DecodeAccountClaims(jwtToken)
+	require_NoError(t, err)
+	r, err := sys.Request(fmt.Sprintf(`$SYS.REQ.ACCOUNT.%s.CLAIMS.UPDATE`, ac.Subject), []byte(jwtToken), time.Second*2)
+	require_NoError(t, err)
+
+	type data struct {
+		Account string `json:"account"`
+		Code    int    `json:"code"`
+	}
+	type serverResponse struct {
+		Data data `json:"data"`
+	}
+
+	var response serverResponse
+	err = json.Unmarshal(r.Data, &response)
+	require_NoError(t, err)
+	require_NotNil(t, response.Data)
+	require_Equal(t, response.Data.Code, int(200))
+}
diff --git a/server/events_test.go b/server/events_test.go
index 296a49a8..0d52ca20 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -1,4 +1,4 @@
-// Copyright 2018-2023 The NATS Authors
+// Copyright 2018-2024 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -3538,3 +3538,23 @@ func Benchmark_GetHash(b *testing.B) {
 	default:
 	}
 }
+
+func TestClusterSetupMsgs(t *testing.T) {
+	// Tests will set this general faster, but here we want default for production.
+	original := statszRateLimit
+	statszRateLimit = defaultStatszRateLimit
+	defer func() { statszRateLimit = original }()
+
+	numServers := 10
+	c := createClusterEx(t, false, 0, false, "cluster", numServers)
+	defer shutdownCluster(c)
+
+	var totalOut int
+	for _, server := range c.servers {
+		totalOut += int(server.outMsgs)
+	}
+	totalExpected := numServers * numServers
+	if totalOut >= totalExpected {
+		t.Fatalf("Total outMsgs is %d, expected < %d\n", totalOut, totalExpected)
+	}
+}
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 4b492fec..302dce59 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -6944,6 +6944,228 @@ func TestFileStoreRecoverWithRemovesAndNoIndexDB(t *testing.T) {
 	require_Equal(t, ss.Msgs, 7)
 }
 
+func TestFileStoreReloadAndLoseLastSequence(t *testing.T) {
+	sd := t.TempDir()
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: sd},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	for i := 0; i < 22; i++ {
+		fs.SkipMsg()
+	}
+
+	// Restart 5 times.
+	for i := 0; i < 5; i++ {
+		fs.Stop()
+		fs, err = newFileStore(
+			FileStoreConfig{StoreDir: sd},
+			StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+		require_NoError(t, err)
+		defer fs.Stop()
+		var ss StreamState
+		fs.FastState(&ss)
+		require_Equal(t, ss.FirstSeq, 23)
+		require_Equal(t, ss.LastSeq, 22)
+	}
+}
+
+func TestFileStoreReloadAndLoseLastSequenceWithSkipMsgs(t *testing.T) {
+	sd := t.TempDir()
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: sd},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	// Make sure same works with SkipMsgs which can kick in from delete blocks to replicas.
+	require_NoError(t, fs.SkipMsgs(0, 22))
+
+	// Restart 5 times.
+	for i := 0; i < 5; i++ {
+		fs.Stop()
+		fs, err = newFileStore(
+			FileStoreConfig{StoreDir: sd},
+			StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+		require_NoError(t, err)
+		defer fs.Stop()
+		var ss StreamState
+		fs.FastState(&ss)
+		require_Equal(t, ss.FirstSeq, 23)
+		require_Equal(t, ss.LastSeq, 22)
+	}
+}
+
+func TestFileStoreLoadLastWildcard(t *testing.T) {
+	sd := t.TempDir()
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: sd, BlockSize: 512},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	msg := []byte("hello")
+	fs.StoreMsg("foo.22.baz", nil, msg)
+	fs.StoreMsg("foo.22.bar", nil, msg)
+	for i := 0; i < 1000; i++ {
+		fs.StoreMsg("foo.11.foo", nil, msg)
+	}
+
+	// Make sure we remove fss since that would mask the problem that we walk
+	// all the blocks.
+	fs.mu.RLock()
+	for _, mb := range fs.blks {
+		mb.mu.Lock()
+		mb.fss = nil
+		mb.mu.Unlock()
+	}
+	fs.mu.RUnlock()
+
+	// Attempt to load the last msg using a wildcarded subject.
+	sm, err := fs.LoadLastMsg("foo.22.*", nil)
+	require_NoError(t, err)
+	require_Equal(t, sm.seq, 2)
+
+	// Make sure that we took advantage of psim meta data and only load one block.
+	var cloads uint64
+	fs.mu.RLock()
+	for _, mb := range fs.blks {
+		mb.mu.RLock()
+		cloads += mb.cloads
+		mb.mu.RUnlock()
+	}
+	fs.mu.RUnlock()
+	require_Equal(t, cloads, 1)
+}
+
+// We want to make sure that we update psim correctly on a miss.
+func TestFileStoreFilteredPendingPSIMFirstBlockUpdate(t *testing.T) {
+	sd := t.TempDir()
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: sd, BlockSize: 512},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	// When PSIM detects msgs == 1 will catch up, so msgs needs to be > 1.
+	// Then create a huge block gap.
+	msg := []byte("hello")
+	fs.StoreMsg("foo.baz", nil, msg)
+	for i := 0; i < 1000; i++ {
+		fs.StoreMsg("foo.foo", nil, msg)
+	}
+	// Bookend with 2 more foo.baz
+	fs.StoreMsg("foo.baz", nil, msg)
+	fs.StoreMsg("foo.baz", nil, msg)
+	// Now remove first one.
+	removed, err := fs.RemoveMsg(1)
+	require_NoError(t, err)
+	require_True(t, removed)
+	// 84 blocks.
+	require_Equal(t, fs.numMsgBlocks(), 84)
+	fs.mu.RLock()
+	psi, ok := fs.psim.Find([]byte("foo.baz"))
+	fs.mu.RUnlock()
+	require_True(t, ok)
+	require_Equal(t, psi.total, 2)
+	require_Equal(t, psi.fblk, 1)
+	require_Equal(t, psi.lblk, 84)
+
+	// No make sure that a call to numFilterPending which will initially walk all blocks if starting from seq 1 updates psi.
+	var ss SimpleState
+	fs.mu.RLock()
+	fs.numFilteredPending("foo.baz", &ss)
+	fs.mu.RUnlock()
+	require_Equal(t, ss.Msgs, 2)
+	require_Equal(t, ss.First, 1002)
+	require_Equal(t, ss.Last, 1003)
+
+	// Check psi was updated.
+	fs.mu.RLock()
+	psi, ok = fs.psim.Find([]byte("foo.baz"))
+	fs.mu.RUnlock()
+	require_True(t, ok)
+	require_Equal(t, psi.total, 2)
+	require_Equal(t, psi.fblk, 84)
+	require_Equal(t, psi.lblk, 84)
+}
+
+func TestFileStoreWildcardFilteredPendingPSIMFirstBlockUpdate(t *testing.T) {
+	sd := t.TempDir()
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: sd, BlockSize: 512},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	// When PSIM detects msgs == 1 will catch up, so msgs needs to be > 1.
+	// Then create a huge block gap.
+	msg := []byte("hello")
+	fs.StoreMsg("foo.22.baz", nil, msg)
+	fs.StoreMsg("foo.22.bar", nil, msg)
+	for i := 0; i < 1000; i++ {
+		fs.StoreMsg("foo.1.foo", nil, msg)
+	}
+	// Bookend with 3 more,twoe foo.baz and two foo.bar.
+	fs.StoreMsg("foo.22.baz", nil, msg)
+	fs.StoreMsg("foo.22.baz", nil, msg)
+	fs.StoreMsg("foo.22.bar", nil, msg)
+	fs.StoreMsg("foo.22.bar", nil, msg)
+
+	// Now remove first one for foo.bar and foo.baz.
+	removed, err := fs.RemoveMsg(1)
+	require_NoError(t, err)
+	require_True(t, removed)
+	removed, err = fs.RemoveMsg(2)
+	require_NoError(t, err)
+	require_True(t, removed)
+
+	// 92 blocks.
+	require_Equal(t, fs.numMsgBlocks(), 92)
+	fs.mu.RLock()
+	psi, ok := fs.psim.Find([]byte("foo.22.baz"))
+	fs.mu.RUnlock()
+	require_True(t, ok)
+	require_Equal(t, psi.total, 2)
+	require_Equal(t, psi.fblk, 1)
+	require_Equal(t, psi.lblk, 92)
+
+	fs.mu.RLock()
+	psi, ok = fs.psim.Find([]byte("foo.22.bar"))
+	fs.mu.RUnlock()
+	require_True(t, ok)
+	require_Equal(t, psi.total, 2)
+	require_Equal(t, psi.fblk, 1)
+	require_Equal(t, psi.lblk, 92)
+
+	// No make sure that a call to numFilterPending which will initially walk all blocks if starting from seq 1 updates psi.
+	var ss SimpleState
+	fs.mu.RLock()
+	fs.numFilteredPending("foo.22.*", &ss)
+	fs.mu.RUnlock()
+	require_Equal(t, ss.Msgs, 4)
+	require_Equal(t, ss.First, 1003)
+	require_Equal(t, ss.Last, 1006)
+
+	// Check both psi were updated.
+	fs.mu.RLock()
+	psi, ok = fs.psim.Find([]byte("foo.22.baz"))
+	fs.mu.RUnlock()
+	require_True(t, ok)
+	require_Equal(t, psi.total, 2)
+	require_Equal(t, psi.fblk, 92)
+	require_Equal(t, psi.lblk, 92)
+
+	fs.mu.RLock()
+	psi, ok = fs.psim.Find([]byte("foo.22.bar"))
+	fs.mu.RUnlock()
+	require_True(t, ok)
+	require_Equal(t, psi.total, 2)
+	require_Equal(t, psi.fblk, 92)
+	require_Equal(t, psi.lblk, 92)
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Benchmarks
 ///////////////////////////////////////////////////////////////////////////
diff --git a/server/jetstream_cluster_1_test.go b/server/jetstream_cluster_1_test.go
index 6b6ade74..b759e557 100644
--- a/server/jetstream_cluster_1_test.go
+++ b/server/jetstream_cluster_1_test.go
@@ -5966,7 +5966,444 @@ func TestJetStreamClusterFailMirrorsAndSources(t *testing.T) {
 	})
 }
 
+func TestJetStreamClusterConsumerDeliveredSyncReporting(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "TEST",
+		Retention: nats.WorkQueuePolicy,
+		Subjects:  []string{"foo.*"},
+		Replicas:  3,
+	})
+	require_NoError(t, err)
+
+	sub, err := js.PullSubscribe("foo.bar", "mc")
+	require_NoError(t, err)
+
+	// Make us match first, but not next 10.
+	_, err = js.Publish("foo.bar", nil)
+	require_NoError(t, err)
+	for i := 0; i < 10; i++ {
+		_, err = js.Publish("foo.baz", nil)
+		require_NoError(t, err)
+	}
+
+	msgs, err := sub.Fetch(1)
+	require_NoError(t, err)
+	require_Equal(t, len(msgs), 1)
+
+	// Now we want to make sure that jsz reporting will show the same
+	// state, including delivered, which will have skipped to the end.
+	// The skip can happen on several factors, but for here we just send
+	// another pull request which we will let fail.
+	_, err = sub.Fetch(1, nats.MaxWait(200*time.Millisecond))
+	require_Error(t, err)
+
+	opts := &JSzOptions{Accounts: true, Streams: true, Consumer: true}
+	for _, s := range c.servers {
+		jsz, err := s.Jsz(opts)
+		require_NoError(t, err)
+		ci := jsz.AccountDetails[0].Streams[0].Consumer[0]
+		require_Equal(t, ci.Delivered.Consumer, 1)
+		require_Equal(t, ci.Delivered.Stream, 1)
+	}
+}
+
+// This is to test follower ack fill logic when pending not empty.
+// There was a bug that would update p.Sequence in the stores (mem & file)
+// that would cause the logic to fail. Redeliveries were required to trigger.
+func TestJetStreamClusterConsumerAckSyncReporting(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "TEST",
+		Retention: nats.WorkQueuePolicy,
+		Subjects:  []string{"foo.*"},
+		Replicas:  3,
+	})
+	require_NoError(t, err)
+
+	sub, err := js.PullSubscribe("foo.bar", "mc", nats.AckWait(250*time.Millisecond))
+	require_NoError(t, err)
+
+	for i := 0; i < 10; i++ {
+		_, err = js.Publish("foo.bar", nil)
+		require_NoError(t, err)
+	}
+
+	msgs, err := sub.Fetch(10)
+	require_NoError(t, err)
+	require_Equal(t, len(msgs), 10)
+	// Let redeliveries kick in.
+	time.Sleep(time.Second)
+	msgs, err = sub.Fetch(10)
+	require_NoError(t, err)
+	require_Equal(t, len(msgs), 10)
+
+	// Randomize
+	rand.Shuffle(len(msgs), func(i, j int) { msgs[i], msgs[j] = msgs[j], msgs[i] })
+
+	dontAck := uint64(7)
+	var skipped, last *nats.Msg
+	for _, m := range msgs {
+		meta, err := m.Metadata()
+		require_NoError(t, err)
+		if meta.Sequence.Stream == dontAck {
+			skipped = m
+			continue
+		}
+		if meta.Sequence.Stream == 10 {
+			last = m
+			continue
+		}
+		m.AckSync()
+	}
+
+	// Now we want to make sure that jsz reporting will show the same
+	// state for ack floor.
+	opts := &JSzOptions{Accounts: true, Streams: true, Consumer: true}
+	for _, s := range c.servers {
+		jsz, err := s.Jsz(opts)
+		require_NoError(t, err)
+		ci := jsz.AccountDetails[0].Streams[0].Consumer[0]
+		require_Equal(t, ci.AckFloor.Consumer, dontAck-1)
+		require_Equal(t, ci.AckFloor.Stream, dontAck-1)
+	}
+
+	// Now ack the skipped message
+	skipped.AckSync()
+	for _, s := range c.servers {
+		jsz, err := s.Jsz(opts)
+		require_NoError(t, err)
+		ci := jsz.AccountDetails[0].Streams[0].Consumer[0]
+		require_Equal(t, ci.AckFloor.Consumer, 9)
+		require_Equal(t, ci.AckFloor.Stream, 9)
+	}
+
+	// Now ack the last message
+	last.AckSync()
+	for _, s := range c.servers {
+		jsz, err := s.Jsz(opts)
+		require_NoError(t, err)
+		ci := jsz.AccountDetails[0].Streams[0].Consumer[0]
+		require_Equal(t, ci.AckFloor.Consumer, 20)
+		require_Equal(t, ci.AckFloor.Stream, 10)
+	}
+}
+
+func TestJetStreamClusterConsumerDeleteInterestPolicyMultipleConsumers(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "TEST",
+		Retention: nats.InterestPolicy,
+		Subjects:  []string{"foo.*"},
+		Replicas:  3,
+	})
+	require_NoError(t, err)
+
+	// Make trhe first sequence high. We already protect against it but for extra sanity.
+	err = js.PurgeStream("TEST", &nats.StreamPurgeRequest{Sequence: 100_000_000})
+	require_NoError(t, err)
+
+	// Create 2 consumers.
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:   "C1",
+		AckPolicy: nats.AckExplicitPolicy,
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:   "C2",
+		AckPolicy: nats.AckExplicitPolicy,
+	})
+	require_NoError(t, err)
+
+	for i := 0; i < 100; i++ {
+		js.PublishAsync("foo.bar", []byte("ok"))
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	si, err := js.StreamInfo("TEST")
+	require_NoError(t, err)
+	require_Equal(t, si.State.Msgs, 100)
+
+	sub, err := js.PullSubscribe("foo.bar", "C1")
+	require_NoError(t, err)
+	msgs, err := sub.Fetch(50)
+	require_NoError(t, err)
+	require_Equal(t, len(msgs), 50)
+	for _, m := range msgs {
+		m.AckSync()
+	}
+
+	// Now delete second one and make sure accounting correct.
+	err = js.DeleteConsumer("TEST", "C2")
+	require_NoError(t, err)
+	time.Sleep(100 * time.Millisecond)
+
+	si, err = js.StreamInfo("TEST")
+	require_NoError(t, err)
+	require_Equal(t, si.State.Msgs, 50)
+}
+
+func TestJetStreamClusterConsumerAckNoneInterestPolicyShouldNotRetainAfterDelivery(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "TEST",
+		Retention: nats.InterestPolicy,
+		Subjects:  []string{"foo.*"},
+		Replicas:  3,
+	})
+	require_NoError(t, err)
+
+	// Make trhe first sequence high. We already protect against it but for extra sanity.
+	err = js.PurgeStream("TEST", &nats.StreamPurgeRequest{Sequence: 100_000_000})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:   "C1",
+		AckPolicy: nats.AckNonePolicy,
+	})
+	require_NoError(t, err)
+
+	for i := 0; i < 100; i++ {
+		js.PublishAsync("foo.bar", []byte("ok"))
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	si, err := js.StreamInfo("TEST")
+	require_NoError(t, err)
+	require_Equal(t, si.State.Msgs, 100)
+
+	sub, err := js.PullSubscribe("foo.bar", "C1")
+	require_NoError(t, err)
+	msgs, err := sub.Fetch(100)
+	require_NoError(t, err)
+	require_Equal(t, len(msgs), 100)
+	for _, m := range msgs {
+		m.AckSync()
+	}
+	si, err = js.StreamInfo("TEST")
+	require_NoError(t, err)
+	require_Equal(t, si.State.Msgs, 0)
+}
+
+func TestJetStreamClusterConsumerDeleteAckNoneInterestPolicyWithOthers(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "TEST",
+		Retention: nats.InterestPolicy,
+		Subjects:  []string{"foo.*"},
+		Replicas:  3,
+	})
+	require_NoError(t, err)
+
+	// Make trhe first sequence high. We already protect against it but for extra sanity.
+	err = js.PurgeStream("TEST", &nats.StreamPurgeRequest{Sequence: 100_000_000})
+	require_NoError(t, err)
+
+	// Create 2 consumers.
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:   "C1",
+		AckPolicy: nats.AckExplicitPolicy,
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:   "C2",
+		AckPolicy: nats.AckNonePolicy,
+	})
+	require_NoError(t, err)
+
+	for i := 0; i < 100; i++ {
+		js.PublishAsync("foo.bar", []byte("ok"))
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	si, err := js.StreamInfo("TEST")
+	require_NoError(t, err)
+	require_Equal(t, si.State.Msgs, 100)
+
+	sub, err := js.PullSubscribe("foo.bar", "C1")
+	require_NoError(t, err)
+	msgs, err := sub.Fetch(100)
+	require_NoError(t, err)
+	require_Equal(t, len(msgs), 100)
+	for _, m := range msgs {
+		m.AckSync()
+	}
+	// AckNone will hold.
+	si, err = js.StreamInfo("TEST")
+	require_NoError(t, err)
+	require_Equal(t, si.State.Msgs, 100)
+
+	// Now delete second one and make sure accounting correct.
+	err = js.DeleteConsumer("TEST", "C2")
+	require_NoError(t, err)
+	time.Sleep(100 * time.Millisecond)
+
+	si, err = js.StreamInfo("TEST")
+	require_NoError(t, err)
+	require_Equal(t, si.State.Msgs, 0)
+}
+
+func TestJetStreamClusterConsumerDeleteInterestPolicyPerf(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "TEST",
+		Retention: nats.InterestPolicy,
+		Subjects:  []string{"foo.*"},
+		Replicas:  3,
+	})
+	require_NoError(t, err)
+
+	// Make trhe first sequence high. We already protect against it but for extra sanity.
+	err = js.PurgeStream("TEST", &nats.StreamPurgeRequest{Sequence: 100_000_000})
+	require_NoError(t, err)
+
+	// Create 3 consumers. 1 Ack explicit, 1 AckAll and 1 AckNone
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:   "C1",
+		AckPolicy: nats.AckExplicitPolicy,
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:   "C2",
+		AckPolicy: nats.AckAllPolicy,
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:   "C3",
+		AckPolicy: nats.AckNonePolicy,
+	})
+	require_NoError(t, err)
+
+	for i := 0; i < 500_000; i++ {
+		js.PublishAsync("foo.bar", []byte("ok"))
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	// For C1 grab 100 and ack evens.
+	sub, err := js.PullSubscribe("foo.bar", "C1")
+	require_NoError(t, err)
+	msgs, err := sub.Fetch(100)
+	require_NoError(t, err)
+	require_Equal(t, len(msgs), 100)
+	for _, m := range msgs {
+		meta, _ := m.Metadata()
+		if meta.Sequence.Stream%2 == 0 {
+			m.AckSync()
+		}
+	}
+
+	// For C2 grab 500 and ack 100.
+	sub, err = js.PullSubscribe("foo.bar", "C2")
+	require_NoError(t, err)
+	msgs, err = sub.Fetch(500, nats.MaxWait(10*time.Second))
+	require_NoError(t, err)
+	require_Equal(t, len(msgs), 500)
+	msgs[99].AckSync()
+
+	// Simulate stream viewer, get first 10 from C3
+	sub, err = js.PullSubscribe("foo.bar", "C3")
+	require_NoError(t, err)
+	msgs, err = sub.Fetch(10)
+	require_NoError(t, err)
+	require_Equal(t, len(msgs), 10)
+
+	time.Sleep(500 * time.Millisecond)
+	si, err := js.StreamInfo("TEST")
+	require_NoError(t, err)
+	require_Equal(t, si.State.Msgs, 499_995)
+
+	// Before fix this was in the seconds. All the while the stream is locked.
+	// This should be short now.
+	start := time.Now()
+	err = js.DeleteConsumer("TEST", "C3")
+	require_NoError(t, err)
+	if elapsed := time.Since(start); elapsed > 50*time.Millisecond {
+		t.Fatalf("Deleting AckNone consumer took too long: %v", elapsed)
+	}
+
+	time.Sleep(500 * time.Millisecond)
+	si, err = js.StreamInfo("TEST")
+	require_NoError(t, err)
+	require_Equal(t, si.State.Msgs, 499_950)
+
+	// Now do AckAll
+	start = time.Now()
+	err = js.DeleteConsumer("TEST", "C2")
+	require_NoError(t, err)
+	if elapsed := time.Since(start); elapsed > 50*time.Millisecond {
+		t.Fatalf("Deleting AckAll consumer took too long: %v", elapsed)
+	}
+
+	time.Sleep(500 * time.Millisecond)
+	si, err = js.StreamInfo("TEST")
+	require_NoError(t, err)
+	require_Equal(t, si.State.Msgs, 499_950)
+
+	// Now do AckExplicit
+	start = time.Now()
+	err = js.DeleteConsumer("TEST", "C1")
+	require_NoError(t, err)
+	if elapsed := time.Since(start); elapsed > 50*time.Millisecond {
+		t.Fatalf("Deleting AckExplicit consumer took too long: %v", elapsed)
+	}
+
+	time.Sleep(500 * time.Millisecond)
+	si, err = js.StreamInfo("TEST")
+	require_NoError(t, err)
+	require_Equal(t, si.State.Msgs, 0)
+}
+
 //
-// DO NOT ADD NEW TESTS IN THIS FILE
+// DO NOT ADD NEW TESTS IN THIS FILE (unless to balance test times)
 // Add at the end of jetstream_cluster_<n>_test.go, with <n> being the highest value.
 //
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index 89388b8c..d27a949a 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -4735,7 +4735,7 @@ func TestJetStreamClusterSnapshotAndRestoreWithHealthz(t *testing.T) {
 
 	rresp.Error = nil
 	json.Unmarshal(rmsg.Data, &rresp)
-	require_True(t, resp.Error == nil)
+	require_True(t, rresp.Error == nil)
 
 	checkHealth()
 
@@ -4757,7 +4757,7 @@ func TestJetStreamClusterSnapshotAndRestoreWithHealthz(t *testing.T) {
 	require_NoError(t, err)
 	rresp.Error = nil
 	json.Unmarshal(rmsg.Data, &rresp)
-	require_True(t, resp.Error == nil)
+	require_True(t, rresp.Error == nil)
 
 	si, err := js.StreamInfo("TEST")
 	require_NoError(t, err)
@@ -4770,6 +4770,40 @@ func TestJetStreamClusterSnapshotAndRestoreWithHealthz(t *testing.T) {
 	si, err = js.StreamInfo("TEST")
 	require_NoError(t, err)
 	require_True(t, si.State.Msgs == uint64(toSend))
+
+	// Now make sure if we try to restore to a single server that the artifact is cleaned up and the server returns ok for healthz.
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, _ = jsClientConnect(t, s)
+	defer nc.Close()
+
+	rmsg, err = nc.Request(fmt.Sprintf(JSApiStreamRestoreT, "TEST"), req, 5*time.Second)
+	require_NoError(t, err)
+
+	rresp.Error = nil
+	json.Unmarshal(rmsg.Data, &rresp)
+	require_True(t, rresp.Error == nil)
+
+	for i, r := 0, bytes.NewReader(snapshot); ; {
+		n, err := r.Read(chunk[:])
+		if err != nil {
+			break
+		}
+		_, err = nc.Request(rresp.DeliverSubject, chunk[:n], time.Second)
+		require_NoError(t, err)
+		i++
+	}
+	rmsg, err = nc.Request(rresp.DeliverSubject, nil, time.Second)
+	require_NoError(t, err)
+	rresp.Error = nil
+	json.Unmarshal(rmsg.Data, &rresp)
+
+	require_True(t, rresp.Error != nil)
+	require_Equal(t, rresp.ApiResponse.Error.ErrCode, 10074)
+
+	status := s.healthz(nil)
+	require_Equal(t, status.StatusCode, 200)
 }
 
 func TestJetStreamClusterBinaryStreamSnapshotCapability(t *testing.T) {
diff --git a/server/jetstream_cluster_4_test.go b/server/jetstream_cluster_4_test.go
index 0013e5db..8407187e 100644
--- a/server/jetstream_cluster_4_test.go
+++ b/server/jetstream_cluster_4_test.go
@@ -1540,25 +1540,28 @@ func TestJetStreamClusterDoubleAckRedelivery(t *testing.T) {
 				}
 
 				msgID := msg.Header.Get(nats.MsgIdHdr)
-				if meta.NumDelivered > 1 {
-					if err, ok := errors[msgID]; ok {
-						t.Logf("Redelivery after failed Ack Sync: %+v - %+v - error: %v", msg.Reply, msg.Header, err)
-					} else {
-						t.Logf("Redelivery: %+v - %+v", msg.Reply, msg.Header)
-					}
-					if resp, ok := acked[msgID]; ok {
-						t.Errorf("Redelivery after successful Ack Sync: msgID:%v - redelivered:%v - original:%+v - ack:%+v",
-							msgID, msg.Reply, resp.original.Reply, resp.ack)
-						resp.redelivered = msg
-						extraRedeliveries++
-					}
+				if err, ok := errors[msgID]; ok {
+					t.Logf("Redelivery (num_delivered: %v) after failed Ack Sync: %+v - %+v - error: %v", meta.NumDelivered, msg.Reply, msg.Header, err)
+				}
+				if resp, ok := acked[msgID]; ok {
+					t.Errorf("Redelivery (num_delivered: %v) after successful Ack Sync: msgID:%v - redelivered:%v - original:%+v - ack:%+v",
+						meta.NumDelivered, msgID, msg.Reply, resp.original.Reply, resp.ack)
+					resp.redelivered = msg
+					extraRedeliveries++
 				}
 				received[msgID]++
-				resp, err := nc.Request(msg.Reply, []byte("+ACK"), 500*time.Millisecond)
-				if err != nil {
-					errors[msgID] = err
-				} else {
-					acked[msgID] = &ackResult{resp, msg, nil}
+
+				// Retry quickly a few times after there is a failed ack.
+			Retries:
+				for i := 0; i < 10; i++ {
+					resp, err := nc.Request(msg.Reply, []byte("+ACK"), 500*time.Millisecond)
+					if err != nil {
+						t.Logf("Error: %v %v", msgID, err)
+						errors[msgID] = err
+					} else {
+						acked[msgID] = &ackResult{resp, msg, nil}
+						break Retries
+					}
 				}
 			}
 		}
@@ -2184,3 +2187,145 @@ func TestJetStreamClusterBusyStreams(t *testing.T) {
 		}
 	})
 }
+
+// https://github.com/nats-io/nats-server/issues/5488
+func TestJetStreamClusterSingleMaxConsumerUpdate(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:         "TEST",
+		MaxConsumers: 1,
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Name:          "test_consumer",
+		MaxAckPending: 1000,
+	})
+	require_NoError(t, err)
+
+	// This would previously return a "nats: maximum consumers limit
+	// reached" (10026) error.
+	_, err = js.UpdateConsumer("TEST", &nats.ConsumerConfig{
+		Name:          "test_consumer",
+		MaxAckPending: 1001,
+	})
+	require_NoError(t, err)
+}
+
+func TestJetStreamClusterStreamLastSequenceResetAfterStorageWipe(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	// After bug was found, number of streams and wiping store directory really did not affect.
+	numStreams := 50
+	var wg sync.WaitGroup
+	wg.Add(numStreams)
+
+	for i := 1; i <= numStreams; i++ {
+		go func(n int) {
+			defer wg.Done()
+			_, err := js.AddStream(&nats.StreamConfig{
+				Name:      fmt.Sprintf("TEST:%d", n),
+				Retention: nats.InterestPolicy,
+				Subjects:  []string{fmt.Sprintf("foo.%d.*", n)},
+				Replicas:  3,
+			}, nats.MaxWait(30*time.Second))
+			require_NoError(t, err)
+			subj := fmt.Sprintf("foo.%d.bar", n)
+			for i := 0; i < 222; i++ {
+				js.Publish(subj, nil)
+			}
+		}(i)
+	}
+	wg.Wait()
+
+	for i := 0; i < 5; i++ {
+		// Walk the servers and shut each down, and wipe the storage directory.
+		for _, s := range c.servers {
+			sd := s.JetStreamConfig().StoreDir
+			s.Shutdown()
+			s.WaitForShutdown()
+			os.RemoveAll(sd)
+			s = c.restartServer(s)
+			checkFor(t, 10*time.Second, 200*time.Millisecond, func() error {
+				hs := s.healthz(nil)
+				if hs.Error != _EMPTY_ {
+					return errors.New(hs.Error)
+				}
+				return nil
+			})
+		}
+
+		for _, s := range c.servers {
+			for i := 1; i <= numStreams; i++ {
+				stream := fmt.Sprintf("TEST:%d", i)
+				mset, err := s.GlobalAccount().lookupStream(stream)
+				require_NoError(t, err)
+				var state StreamState
+				checkFor(t, 10*time.Second, 200*time.Millisecond, func() error {
+					mset.store.FastState(&state)
+					if state.LastSeq != 222 {
+						return fmt.Errorf("%v Wrong last sequence %d for %q - State  %+v", s, state.LastSeq, stream, state)
+					}
+					return nil
+				})
+			}
+		}
+	}
+}
+
+func TestJetStreamClusterAckFloorBetweenLeaderAndFollowers(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "TEST",
+		Retention: nats.InterestPolicy,
+		Subjects:  []string{"foo.*"},
+		Replicas:  3,
+	})
+	require_NoError(t, err)
+
+	sub, err := js.PullSubscribe("foo.*", "consumer")
+	require_NoError(t, err)
+
+	// Do 25 rounds.
+	for i := 1; i <= 25; i++ {
+		// Send 50 msgs.
+		for x := 0; x < 50; x++ {
+			_, err := js.Publish("foo.bar", nil)
+			require_NoError(t, err)
+		}
+		msgs, err := sub.Fetch(50, nats.MaxWait(10*time.Second))
+		require_NoError(t, err)
+		require_Equal(t, len(msgs), 50)
+		// Randomize
+		rand.Shuffle(len(msgs), func(i, j int) { msgs[i], msgs[j] = msgs[j], msgs[i] })
+		for _, m := range msgs {
+			m.AckSync()
+		}
+
+		time.Sleep(100 * time.Millisecond)
+		for _, s := range c.servers {
+			mset, err := s.GlobalAccount().lookupStream("TEST")
+			require_NoError(t, err)
+			consumer := mset.lookupConsumer("consumer")
+			require_NotEqual(t, consumer, nil)
+			info := consumer.info()
+			require_Equal(t, info.NumAckPending, 0)
+			require_Equal(t, info.AckFloor.Consumer, uint64(i*50))
+			require_Equal(t, info.AckFloor.Stream, uint64(i*50))
+		}
+	}
+}
diff --git a/server/jetstream_consumer_test.go b/server/jetstream_consumer_test.go
index 8cf9ddfa..e408621d 100644
--- a/server/jetstream_consumer_test.go
+++ b/server/jetstream_consumer_test.go
@@ -1080,6 +1080,8 @@ func TestJetStreamConsumerDelete(t *testing.T) {
 }
 
 func TestJetStreamConsumerFetchWithDrain(t *testing.T) {
+	t.Skip()
+
 	test := func(t *testing.T, cc *nats.ConsumerConfig) {
 		s := RunBasicJetStreamServer(t)
 		defer s.Shutdown()
@@ -1137,8 +1139,6 @@ func TestJetStreamConsumerFetchWithDrain(t *testing.T) {
 					metadata.NumDelivered, metadata.Sequence.Stream)
 			}
 			msgs[int(metadata.Sequence.Stream)] = int(metadata.NumDelivered)
-
-			require_NoError(t, err)
 			return true
 		}
 
@@ -1169,14 +1169,14 @@ func TestJetStreamConsumerFetchWithDrain(t *testing.T) {
 		test(t, &nats.ConsumerConfig{
 			Durable:   "C",
 			AckPolicy: nats.AckExplicitPolicy,
-			AckWait:   10 * time.Second,
+			AckWait:   20 * time.Second,
 		})
 	})
 	t.Run("with-backoff", func(t *testing.T) {
 		test(t, &nats.ConsumerConfig{
 			Durable:   "C",
 			AckPolicy: nats.AckExplicitPolicy,
-			AckWait:   10 * time.Second,
+			AckWait:   20 * time.Second,
 			BackOff:   []time.Duration{25 * time.Millisecond, 100 * time.Millisecond, 250 * time.Millisecond},
 		})
 	})
diff --git a/server/jetstream_helpers_test.go b/server/jetstream_helpers_test.go
index f130b524..da2814e1 100644
--- a/server/jetstream_helpers_test.go
+++ b/server/jetstream_helpers_test.go
@@ -42,6 +42,9 @@ func init() {
 	maxElectionTimeout = 3500 * time.Millisecond
 	lostQuorumInterval = 2 * time.Second
 	lostQuorumCheck = 4 * hbInterval
+
+	// For statz and jetstream placement speedups as well.
+	statszRateLimit = 0
 }
 
 // Used to setup clusters of clusters for tests.
diff --git a/server/jetstream_jwt_test.go b/server/jetstream_jwt_test.go
index 4e261f41..f8f6466f 100644
--- a/server/jetstream_jwt_test.go
+++ b/server/jetstream_jwt_test.go
@@ -1460,3 +1460,75 @@ func TestJetStreamJWTHAStorageLimitsOnScaleAndUpdate(t *testing.T) {
 	require_Equal(t, r3.ReservedMemory, 22*1024)    // TEST9
 	require_Equal(t, r3.ReservedStore, 5*1024*1024) // TEST1-TEST6
 }
+
+func TestJetStreamJWTClusteredTiersR3StreamWithR1ConsumersAndAccounting(t *testing.T) {
+	sysKp, syspub := createKey(t)
+	sysJwt := encodeClaim(t, jwt.NewAccountClaims(syspub), syspub)
+	newUser(t, sysKp)
+
+	accKp, aExpPub := createKey(t)
+	accClaim := jwt.NewAccountClaims(aExpPub)
+	accClaim.Name = "acc"
+	accClaim.Limits.JetStreamTieredLimits["R1"] = jwt.JetStreamLimits{
+		DiskStorage: 1100, Consumer: 10, Streams: 1}
+	accClaim.Limits.JetStreamTieredLimits["R3"] = jwt.JetStreamLimits{
+		DiskStorage: 1100, Consumer: 1, Streams: 1}
+	accJwt := encodeClaim(t, accClaim, aExpPub)
+	accCreds := newUser(t, accKp)
+	tmlp := `
+		listen: 127.0.0.1:-1
+		server_name: %s
+		jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
+		leaf {
+			listen: 127.0.0.1:-1
+		}
+		cluster {
+			name: %s
+			listen: 127.0.0.1:%d
+			routes = [%s]
+		}
+	` + fmt.Sprintf(`
+		operator: %s
+		system_account: %s
+		resolver = MEMORY
+		resolver_preload = {
+			%s : %s
+			%s : %s
+		}
+	`, ojwt, syspub, syspub, sysJwt, aExpPub, accJwt)
+
+	c := createJetStreamClusterWithTemplate(t, tmlp, "cluster", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer(), nats.UserCredentials(accCreds))
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo.*"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	// Now make sure we can add in 10 R1 consumers.
+	for i := 1; i <= 10; i++ {
+		_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+			Name:      fmt.Sprintf("C-%d", i),
+			AckPolicy: nats.AckExplicitPolicy,
+			Replicas:  1,
+		})
+		require_NoError(t, err)
+	}
+
+	info, err := js.AccountInfo()
+	require_NoError(t, err)
+
+	// Make sure we account for these properly.
+	r1 := info.Tiers["R1"]
+	r3 := info.Tiers["R3"]
+
+	require_Equal(t, r1.Streams, 0)
+	require_Equal(t, r1.Consumers, 10)
+	require_Equal(t, r3.Streams, 1)
+	require_Equal(t, r3.Consumers, 0)
+}
diff --git a/server/jetstream_leafnode_test.go b/server/jetstream_leafnode_test.go
index bb2337d7..f0fa7888 100644
--- a/server/jetstream_leafnode_test.go
+++ b/server/jetstream_leafnode_test.go
@@ -1242,3 +1242,85 @@ func TestJetStreamLeafNodeSvcImportExportCycle(t *testing.T) {
 	_, err = js.Publish("foo", []byte("msg"))
 	require_NoError(t, err)
 }
+
+func TestJetStreamLeafNodeJSClusterMigrateRecovery(t *testing.T) {
+	tmpl := strings.Replace(jsClusterAccountsTempl, "store_dir:", "domain: hub, store_dir:", 1)
+	c := createJetStreamCluster(t, tmpl, "hub", _EMPTY_, 3, 12232, true)
+	defer c.shutdown()
+
+	tmpl = strings.Replace(jsClusterTemplWithLeafNode, "store_dir:", "domain: leaf, store_dir:", 1)
+	lnc := c.createLeafNodesWithTemplateAndStartPort(tmpl, "leaf", 3, 23913)
+	defer lnc.shutdown()
+
+	lnc.waitOnClusterReady()
+	for _, s := range lnc.servers {
+		s.setJetStreamMigrateOnRemoteLeaf()
+	}
+
+	nc, _ := jsClientConnect(t, lnc.randomServer())
+	defer nc.Close()
+
+	ljs, err := nc.JetStream(nats.Domain("leaf"))
+	require_NoError(t, err)
+
+	// Create an asset in the leafnode cluster.
+	si, err := ljs.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+	require_Equal(t, si.Cluster.Name, "leaf")
+	require_NotEqual(t, si.Cluster.Leader, noLeader)
+	require_Equal(t, len(si.Cluster.Replicas), 2)
+
+	// Count how many remotes each server in the leafnode cluster is
+	// supposed to have and then take them down.
+	remotes := map[*Server]int{}
+	for _, s := range lnc.servers {
+		remotes[s] += len(s.leafRemoteCfgs)
+		s.closeAndDisableLeafnodes()
+		checkLeafNodeConnectedCount(t, s, 0)
+	}
+
+	// The Raft nodes in the leafnode cluster now need some time to
+	// notice that they're no longer receiving AEs from a leader, as
+	// they should have been forced into observer mode. Check that
+	// this is the case.
+	time.Sleep(maxElectionTimeout)
+	for _, s := range lnc.servers {
+		s.rnMu.RLock()
+		for name, n := range s.raftNodes {
+			// We don't expect the metagroup to have turned into an
+			// observer but all other assets should have done.
+			if name == defaultMetaGroupName {
+				require_False(t, n.IsObserver())
+			} else {
+				require_True(t, n.IsObserver())
+			}
+		}
+		s.rnMu.RUnlock()
+	}
+
+	// Bring the leafnode connections back up.
+	for _, s := range lnc.servers {
+		s.reEnableLeafnodes()
+		checkLeafNodeConnectedCount(t, s, remotes[s])
+	}
+
+	// Wait for nodes to notice they are no longer in observer mode
+	// and to leave observer mode.
+	time.Sleep(maxElectionTimeout)
+	for _, s := range lnc.servers {
+		s.rnMu.RLock()
+		for _, n := range s.raftNodes {
+			require_False(t, n.IsObserver())
+		}
+		s.rnMu.RUnlock()
+	}
+
+	// Previously nodes would have left observer mode but then would
+	// have failed to elect a stream leader as they were stuck on a
+	// long election timer. Now this should work reliably.
+	lnc.waitOnStreamLeader(globalAccountName, "TEST")
+}
diff --git a/server/jetstream_sourcing_scaling_test.go b/server/jetstream_sourcing_scaling_test.go
index ba17e439..dbd6bbf6 100644
--- a/server/jetstream_sourcing_scaling_test.go
+++ b/server/jetstream_sourcing_scaling_test.go
@@ -14,231 +14,94 @@
 package server
 
 import (
-	"encoding/json"
 	"fmt"
+	"github.com/nats-io/nats.go"
+	"strconv"
 	"testing"
 	"time"
-
-	"github.com/nats-io/nats.go"
 )
 
-// This test is being skipped by CI as it takes too long to run and is meant to test the scalability of sourcing
-// rather than being a unit test.
-func TestStreamSourcingScalingManySourcing(t *testing.T) {
-	t.Skip()
-
-	var numSourcing = 10000 // fails at 10000
-	var numMessages = uint64(50)
+var serverConfig1 = `
+	server_name: server1
+	listen: 127.0.0.1:4222
+	http: 8222
 
-	c := createJetStreamClusterExplicit(t, "R3S", 3)
-	defer c.shutdown()
-	s := c.randomServer()
-	urls := s.clientConnectURLs
-	fmt.Printf("Connected to server %+v\n", urls)
+	prof_port = 18222
 
-	nc, js := jsClientConnect(t, s, nats.Timeout(20*time.Second))
-	defer nc.Close()
+	jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
 
-	// create a stream to source from
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name:                 "source",
-		Subjects:             []string{"foo.>"},
-		Retention:            nats.LimitsPolicy,
-		Storage:              nats.FileStorage,
-		Discard:              nats.DiscardOld,
-		Replicas:             1,
-		AllowDirect:          true,
-		MirrorDirect:         false,
-		DiscardNewPerSubject: false,
-		MaxAge:               time.Duration(60 * time.Minute),
-	})
-	require_NoError(t, err)
-
-	// create n streams that source from it
-	for i := 0; i < numSourcing; i++ {
-		_, err = js.AddStream(&nats.StreamConfig{
-			Name:                 fmt.Sprintf("sourcing-%d", i),
-			Sources:              []*nats.StreamSource{{Name: "source", FilterSubject: fmt.Sprintf("foo.%d", i)}},
-			Retention:            nats.LimitsPolicy,
-			Storage:              nats.FileStorage,
-			Discard:              nats.DiscardOld,
-			Replicas:             1,
-			AllowDirect:          true,
-			MirrorDirect:         false,
-			DiscardNewPerSubject: false,
-			MaxAge:               time.Duration(60 * time.Minute),
-		})
-		require_NoError(t, err)
-	}
-
-	fmt.Printf("Streams created\n")
-
-	// publish n messages for each sourcer in the source stream
-	for j := uint64(0); j < numMessages; j++ {
-		start := time.Now()
-		for i := 0; i < numSourcing; i++ {
-			_, err = js.Publish(fmt.Sprintf("foo.%d", i), []byte("hello"))
-			require_NoError(t, err)
-		}
-		end := time.Now()
-		fmt.Printf("[%v] Published round %d, avg pub latency %v\n", time.Now(), j, end.Sub(start)/time.Duration(numSourcing))
+	cluster {
+  		name: my_cluster
+  		listen: 127.0.0.1:4248
+  		routes: [nats://127.0.0.1:4249,nats://127.0.0.1:4250]
 	}
 
-	fmt.Printf("Messages published\n")
-	time.Sleep(100 * time.Millisecond)
-
-	// cause a leader stepdown
-	resp, err := nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, "source"), nil, time.Second)
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	var sdResp JSApiStreamLeaderStepDownResponse
-	if err := json.Unmarshal(resp.Data, &sdResp); err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	if sdResp.Error != nil {
-		t.Fatalf("Unexpected error: %+v", sdResp.Error)
-	}
+	accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }}
+`
 
-	fmt.Printf("Leader stepdown done\n")
+var serverConfig2 = `
+	server_name: server2
+	listen: 127.0.0.1:5222
+	http: 8223
 
-	for i := 0; i < numSourcing; i++ {
-		// publish one message for each sourcing stream into the source stream
-		_, err = js.Publish(fmt.Sprintf("foo.%d", i), []byte("hello"))
-		require_NoError(t, err)
-	}
+	prof_port = 18223
 
-	time.Sleep(500 * time.Millisecond)
+	jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
 
-	for i := 0; i < numSourcing; i++ {
-		// check that each sourcer has the message
-		state, err := js.StreamInfo(fmt.Sprintf("sourcing-%d", i))
-		require_NoError(t, err)
-		if state.State.Msgs != numMessages+1 {
-			t.Fatalf("Expected %d messages, got %d", numMessages, state.State.Msgs)
-		}
-		fmt.Printf("Stream sourcing-%d has %d messages\n", i, state.State.Msgs)
+	cluster {
+  		name: my_cluster
+  		listen: 127.0.0.1:4249
+  		routes: [nats://127.0.0.1:4248,nats://127.0.0.1:4250]
 	}
 
-}
+	accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }}
+`
 
-// This test is being skipped by CI as it takes too long to run and is meant to test the scalability of sourcing
-// rather than being a unit test.
-func TestStreamSourcingScalingSourcingMany(t *testing.T) {
-	t.Skip()
+var serverConfig3 = `
+	server_name: server3
+	listen: 127.0.0.1:6222
+	http: 8224
 
-	var numSourced = 10000
-	var numMsgPerSource = uint64(1000)
-	var batchSize = 1000
-	var retries int
+	prof_port = 18224
 
-	c := createJetStreamClusterExplicit(t, "R3S", 3)
-	defer c.shutdown()
-	s := c.randomServer()
-	urls := s.clientConnectURLs
-	fmt.Printf("Connected to server %+v\n", urls)
-
-	nc, js := jsClientConnect(t, s, nats.Timeout(20*time.Second))
-	defer nc.Close()
+	jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}
 
-	// create n streams to source from
-	for i := 0; i < numSourced; i++ {
-		_, err := js.AddStream(&nats.StreamConfig{
-			Name:                 fmt.Sprintf("sourced-%d", i),
-			Subjects:             []string{fmt.Sprintf("foo.%d", i)},
-			Retention:            nats.LimitsPolicy,
-			Storage:              nats.FileStorage,
-			Discard:              nats.DiscardOld,
-			Replicas:             1,
-			AllowDirect:          true,
-			MirrorDirect:         false,
-			DiscardNewPerSubject: false,
-		})
-		require_NoError(t, err)
+	cluster {
+  		name: my_cluster
+  		listen: 127.0.0.1:4250
+  		routes: [nats://127.0.0.1:4248,nats://127.0.0.1:4249]
 	}
 
-	// create the StreamSources
-	streamSources := make([]*nats.StreamSource, numSourced)
-
-	for i := 0; i < numSourced; i++ {
-		streamSources[i] = &nats.StreamSource{Name: fmt.Sprintf("sourced-%d", i), FilterSubject: fmt.Sprintf("foo.%d", i)}
-	}
-
-	// create a stream that sources from them
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name:                 "sourcing",
-		Sources:              streamSources,
-		Retention:            nats.LimitsPolicy,
-		Storage:              nats.FileStorage,
-		Discard:              nats.DiscardOld,
-		Replicas:             1,
-		AllowDirect:          true,
-		MirrorDirect:         false,
-		DiscardNewPerSubject: false,
-	})
-	require_NoError(t, err)
-
-	fmt.Printf("Streams created\n")
+	accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }}
+`
 
-	// publish n messages for each sourced stream
-	for j := uint64(0); j < numMsgPerSource; j++ {
-		start := time.Now()
-		var pafs = make([]nats.PubAckFuture, numSourced)
-		for i := 0; i < numSourced; i++ {
-			var err error
+var connectURL = "nats://127.0.0.1:4222,nats://127.0.0.1:4223,nats://127.0.0.1:4224"
 
-			for {
-				pafs[i], err = js.PublishAsync(fmt.Sprintf("foo.%d", i), []byte("hello"))
-				if err != nil {
-					fmt.Printf("Error async publishing: %v, retrying\n", err)
-					retries++
-					time.Sleep(10 * time.Millisecond)
-				} else {
-					break
-				}
-			}
-
-			if i != 0 && i%batchSize == 0 {
-				<-js.PublishAsyncComplete()
-			}
+func createMyLocalCluster(t *testing.T) *cluster {
+	c := &cluster{servers: make([]*Server, 0, 3), opts: make([]*Options, 0, 3), name: "C3"}
 
-		}
+	storeDir1 := t.TempDir()
+	s1, o := RunServerWithConfig(createConfFile(t, []byte(fmt.Sprintf(serverConfig1, storeDir1))))
+	c.servers = append(c.servers, s1)
+	c.opts = append(c.opts, o)
 
-		<-js.PublishAsyncComplete()
+	storeDir2 := t.TempDir()
+	s2, o := RunServerWithConfig(createConfFile(t, []byte(fmt.Sprintf(serverConfig2, storeDir2))))
+	c.servers = append(c.servers, s2)
+	c.opts = append(c.opts, o)
 
-		for i := 0; i < numSourced; i++ {
-			select {
-			case <-pafs[i].Ok():
-			case psae := <-pafs[i].Err():
-				fmt.Printf("Error on PubAckFuture: %v, retrying sync...\n", psae)
-				retries++
-				_, err = js.Publish(fmt.Sprintf("foo.%d", i), []byte("hello"))
-				require_NoError(t, err)
-			}
-		}
+	storeDir3 := t.TempDir()
+	s3, o := RunServerWithConfig(createConfFile(t, []byte(fmt.Sprintf(serverConfig3, storeDir3))))
+	c.servers = append(c.servers, s3)
+	c.opts = append(c.opts, o)
 
-		end := time.Now()
-		fmt.Printf("[%v] Published round %d, avg pub latency %v\n", time.Now(), j, end.Sub(start)/time.Duration(numSourced))
-	}
+	c.t = t
 
-	fmt.Printf("Messages published\n")
+	// Wait til we are formed and have a leader.
+	c.checkClusterFormed()
+	c.waitOnClusterReady()
 
-	checkFor(t, 10*time.Minute, 1*time.Second, func() error {
-		state, err := js.StreamInfo("sourcing")
-		if err != nil {
-			return err
-		}
-		if state.State.Msgs == numMsgPerSource*uint64(numSourced) {
-			fmt.Printf("👍 Test passed: expected %d messages, got %d\n", numMsgPerSource*uint64(numSourced), state.State.Msgs)
-			return nil
-		} else if state.State.Msgs < numMsgPerSource*uint64(numSourced) {
-			fmt.Printf("Expected %d messages, got %d\n", numMsgPerSource*uint64(numSourced), state.State.Msgs)
-			return fmt.Errorf("Expected %d messages, got %d", numMsgPerSource*uint64(numSourced), state.State.Msgs)
-		} else {
-			fmt.Printf("\nToo many messages! expected %d (retries=%d), got %d", numMsgPerSource*uint64(numSourced), retries, state.State.Msgs)
-			return fmt.Errorf("Too many messages: expected %d (retries=%d), got %d", numMsgPerSource*uint64(numSourced), retries, state.State.Msgs)
-		}
-	})
+	return c
 }
 
 // This test is being skipped by CI as it takes too long to run and is meant to test the scalability of sourcing
@@ -246,27 +109,31 @@ func TestStreamSourcingScalingSourcingMany(t *testing.T) {
 func TestStreamSourcingScalingSourcingManyBenchmark(t *testing.T) {
 	t.Skip()
 
-	var numSourced = 500
-	var numMsgPerSource = uint64(10_000)
-	var batchSize = 1000
+	var numSourced = 1000
+	var numMsgPerSource = 10_000
+	var batchSize = 200
 	var retries int
 
 	var err error
 
-	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	c := createMyLocalCluster(t)
 	defer c.shutdown()
-	s := c.randomServer()
-	urls := s.clientConnectURLs
-	fmt.Printf("Connected to server %+v\n", urls)
 
-	nc, js := jsClientConnect(t, s, nats.Timeout(20*time.Second))
+	nc, err := nats.Connect(connectURL)
+	if err != nil {
+		t.Fatalf("Failed to create client: %v", err)
+	}
+	js, err := nc.JetStream(nats.MaxWait(10 * time.Second))
+	if err != nil {
+		t.Fatalf("Unexpected error getting JetStream context: %v", err)
+	}
 	defer nc.Close()
 
 	// create n streams to source from
 	for i := 0; i < numSourced; i++ {
 		_, err := js.AddStream(&nats.StreamConfig{
 			Name:                 fmt.Sprintf("sourced-%d", i),
-			Subjects:             []string{fmt.Sprintf("foo.%d", i)},
+			Subjects:             []string{strconv.Itoa(i)},
 			Retention:            nats.LimitsPolicy,
 			Storage:              nats.FileStorage,
 			Discard:              nats.DiscardOld,
@@ -281,14 +148,14 @@ func TestStreamSourcingScalingSourcingManyBenchmark(t *testing.T) {
 	fmt.Printf("Streams created\n")
 
 	// publish n messages for each sourced stream
-	for j := uint64(0); j < numMsgPerSource; j++ {
+	for j := 0; j < numMsgPerSource; j++ {
 		start := time.Now()
 		var pafs = make([]nats.PubAckFuture, numSourced)
 		for i := 0; i < numSourced; i++ {
 			var err error
 
 			for {
-				pafs[i], err = js.PublishAsync(fmt.Sprintf("foo.%d", i), []byte("hello"))
+				pafs[i], err = js.PublishAsync(strconv.Itoa(i), []byte(strconv.Itoa(j)))
 				if err != nil {
 					fmt.Printf("Error async publishing: %v, retrying\n", err)
 					retries++
@@ -312,7 +179,7 @@ func TestStreamSourcingScalingSourcingManyBenchmark(t *testing.T) {
 			case psae := <-pafs[i].Err():
 				fmt.Printf("Error on PubAckFuture: %v, retrying sync...\n", psae)
 				retries++
-				_, err = js.Publish(fmt.Sprintf("foo.%d", i), []byte("hello"))
+				_, err = js.Publish(strconv.Itoa(i), []byte(strconv.Itoa(j)))
 				require_NoError(t, err)
 			}
 		}
@@ -329,12 +196,13 @@ func TestStreamSourcingScalingSourcingManyBenchmark(t *testing.T) {
 	streamSources := make([]*nats.StreamSource, numSourced)
 
 	for i := 0; i < numSourced; i++ {
-		streamSources[i] = &nats.StreamSource{Name: fmt.Sprintf("sourced-%d", i), FilterSubject: fmt.Sprintf("foo.%d", i)}
+		streamSources[i] = &nats.StreamSource{Name: fmt.Sprintf("sourced-%d", i), FilterSubject: strconv.Itoa(i)}
 	}
 
 	// create a stream that sources from them
 	_, err = js.AddStream(&nats.StreamConfig{
 		Name:                 "sourcing",
+		Subjects:             []string{"foo"},
 		Sources:              streamSources,
 		Retention:            nats.LimitsPolicy,
 		Storage:              nats.FileStorage,
@@ -345,45 +213,70 @@ func TestStreamSourcingScalingSourcingManyBenchmark(t *testing.T) {
 		DiscardNewPerSubject: false,
 	})
 	require_NoError(t, err)
-
+	c.waitOnStreamLeader(globalAccountName, "sourcing")
 	sl := c.streamLeader(globalAccountName, "sourcing")
-	mset, err := sl.GlobalAccount().lookupStream("sourcing")
-	require_NoError(t, err)
+	fmt.Printf("Sourcing stream created leader is *** %s ***\n", sl.Name())
 
-	start := time.Now()
+	expectedSeq := make([]int, numSourced)
 
-	fmt.Printf("Sourcing stream created\n")
+	start := time.Now()
 
 	var lastMsgs uint64
-	var steppedDown bool
+	mset, err := sl.GlobalAccount().lookupStream("sourcing")
+	require_NoError(t, err)
 
-	checkFor(t, 10*time.Minute, 1*time.Second, func() error {
+	checkFor(t, 5*time.Minute, 1000*time.Millisecond, func() error {
 		mset.mu.RLock()
 		var state StreamState
 		mset.store.FastState(&state)
 		mset.mu.RUnlock()
 
-		if state.Msgs == numMsgPerSource*uint64(numSourced) {
-			fmt.Printf("👍 Test passed: expected %d messages, got %d and took %v\n", numMsgPerSource*uint64(numSourced), state.Msgs, time.Since(start))
+		if state.Msgs == uint64(numMsgPerSource*numSourced) {
+			fmt.Printf("👍 Test passed: expected %d messages, got %d and took %v\n", uint64(numMsgPerSource*numSourced), state.Msgs, time.Since(start))
 			return nil
-		} else if state.Msgs < numMsgPerSource*uint64(numSourced) {
-			fmt.Printf("Current Rate %d/s - Received %d\n", state.Msgs-lastMsgs, state.Msgs)
+		} else if state.Msgs < uint64(numMsgPerSource*numSourced) {
+			fmt.Printf("Current Rate %d per second - Received %d\n", state.Msgs-lastMsgs, state.Msgs)
 			lastMsgs = state.Msgs
-
-			if !steppedDown && state.Msgs >= 200_000 {
-				fmt.Printf("\nStepping Down\n")
-				mset.node.StepDown()
-				steppedDown = true
-				// Need to repull our data.
-				c.waitOnStreamLeader(globalAccountName, "sourcing")
-				sl = c.streamLeader(globalAccountName, "sourcing")
-				mset, err = sl.GlobalAccount().lookupStream("sourcing")
-				require_NoError(t, err)
-			}
-			return fmt.Errorf("Expected %d messages, got %d", numMsgPerSource*uint64(numSourced), state.Msgs)
+			return fmt.Errorf("Expected %d messages, got %d", uint64(numMsgPerSource*numSourced), state.Msgs)
 		} else {
-			fmt.Printf("Too many messages! expected %d (retries=%d), got %d\n", numMsgPerSource*uint64(numSourced), retries, state.Msgs)
-			return fmt.Errorf("Too many messages: expected %d (retries=%d), got %d", numMsgPerSource*uint64(numSourced), retries, state.Msgs)
+			fmt.Printf("Too many messages! expected %d (retries=%d), got %d\n", uint64(numMsgPerSource*numSourced), retries, state.Msgs)
+			return fmt.Errorf("Too many messages: expected %d (retries=%d), got %d", uint64(numMsgPerSource*numSourced), retries, state.Msgs)
 		}
 	})
+
+	// Check that all the messages sourced in the stream are correct
+	// Note: expects to see exactly increasing matching sequence numbers, so could theoretically fail if some messages
+	// get recorded 'out of order' (according to the payload value) because asynchronous JS publication is used to
+	// publish the messages.
+	// However, that should not happen if the publish 'batch size' is not more than the number of streams being sourced.
+
+	// create a consumer on sourcing
+	_, err = js.AddConsumer("sourcing", &nats.ConsumerConfig{AckPolicy: nats.AckExplicitPolicy})
+	require_NoError(t, err)
+	syncSub, err := js.SubscribeSync("", nats.BindStream("sourcing"))
+	require_NoError(t, err)
+
+	start = time.Now()
+
+	print("Checking the messages\n")
+	for i := 0; i < numSourced*numMsgPerSource; i++ {
+		msg, err := syncSub.NextMsg(30 * time.Second)
+		require_NoError(t, err)
+		sId, err := strconv.Atoi(msg.Subject)
+		require_NoError(t, err)
+		seq, err := strconv.Atoi(string(msg.Data))
+		require_NoError(t, err)
+		if expectedSeq[sId] == seq {
+			expectedSeq[sId]++
+		} else {
+			t.Fatalf("Expected seq number %d got %d for source %d\n", expectedSeq[sId], seq, sId)
+		}
+		msg.Ack()
+		if i%100_000 == 0 {
+			now := time.Now()
+			fmt.Printf("[%v] Checked %d messages: %f msgs/sec \n", now, i, 100_000/now.Sub(start).Seconds())
+			start = now
+		}
+	}
+	print("👍 Done. \n")
 }
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index fb7be760..49adf5c4 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -969,7 +969,7 @@ func TestJetStreamAddStreamOverlappingSubjects(t *testing.T) {
 	// Test that any overlapping subjects will fail.
 	expectErr(acc.addStream(&StreamConfig{Name: "foo"}))
 	expectErr(acc.addStream(&StreamConfig{Name: "a", Subjects: []string{"baz", "bar"}}))
-	expectErr(acc.addStream(&StreamConfig{Name: "b", Subjects: []string{">"}}))
+	expectErr(acc.addStream(&StreamConfig{Name: "b", Subjects: []string{">"}, NoAck: true}))
 	expectErr(acc.addStream(&StreamConfig{Name: "c", Subjects: []string{"baz.33"}}))
 	expectErr(acc.addStream(&StreamConfig{Name: "d", Subjects: []string{"*.33"}}))
 	expectErr(acc.addStream(&StreamConfig{Name: "e", Subjects: []string{"*.>"}}))
@@ -984,7 +984,7 @@ func TestJetStreamAddStreamOverlapWithJSAPISubjects(t *testing.T) {
 
 	expectErr := func(_ *stream, err error) {
 		t.Helper()
-		if err == nil || !strings.Contains(err.Error(), "subjects overlap") {
+		if err == nil || !strings.Contains(err.Error(), "subjects that overlap with jetstream api") {
 			t.Fatalf("Expected error but got none")
 		}
 	}
@@ -995,7 +995,7 @@ func TestJetStreamAddStreamOverlapWithJSAPISubjects(t *testing.T) {
 	expectErr(acc.addStream(&StreamConfig{Name: "c", Subjects: []string{"$JS.API.*"}}))
 
 	// Events and Advisories etc should be ok.
-	if _, err := acc.addStream(&StreamConfig{Name: "a", Subjects: []string{"$JS.EVENT.>"}}); err != nil {
+	if _, err := acc.addStream(&StreamConfig{Name: "a", Subjects: []string{"$JS.EVENT.>"}, NoAck: true}); err != nil {
 		t.Fatalf("Expected this to work: %v", err)
 	}
 }
@@ -19537,9 +19537,7 @@ func TestJetStreamConsumerMultipleSubjectsLast(t *testing.T) {
 	require_NoError(t, err)
 
 	require_True(t, info.NumAckPending == 0)
-	// Should be 6 since we do not pull "other". We used to jump ack floor ahead
-	// but no longer do that.
-	require_True(t, info.AckFloor.Stream == 6)
+	require_True(t, info.AckFloor.Stream == 8)
 	require_True(t, info.AckFloor.Consumer == 1)
 	require_True(t, info.NumPending == 0)
 }
@@ -23613,3 +23611,154 @@ func TestJetStreamAckAllWithLargeFirstSequenceAndNoAckFloorWithInterestPolicy(t
 	_, err = js.StreamInfo("TEST", nats.MaxWait(100*time.Millisecond))
 	require_NoError(t, err)
 }
+
+// Allow streams with $JS or $SYS prefixes for audit purposes but require no pub ack be set.
+func TestJetStreamAuditStreams(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	// Client for API requests.
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	jsOverlap := errors.New("subjects that overlap with jetstream api require no-ack to be true")
+	sysOverlap := errors.New("subjects that overlap with system api require no-ack to be true")
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"$JS.>"},
+	})
+	require_Error(t, err, NewJSStreamInvalidConfigError(jsOverlap))
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"$JS.API.>"},
+	})
+	require_Error(t, err, NewJSStreamInvalidConfigError(jsOverlap))
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"$JSC.>"},
+	})
+	require_Error(t, err, NewJSStreamInvalidConfigError(jsOverlap))
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"$SYS.>"},
+	})
+	require_Error(t, err, NewJSStreamInvalidConfigError(sysOverlap))
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{">"},
+	})
+	require_Error(t, err, NewJSStreamInvalidConfigError(errors.New("capturing all subjects requires no-ack to be true")))
+
+	// These should all be ok if no pub ack.
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "TEST1",
+		Subjects: []string{"$JS.>"},
+		NoAck:    true,
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "TEST2",
+		Subjects: []string{"$JSC.>"},
+		NoAck:    true,
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "TEST3",
+		Subjects: []string{"$SYS.>"},
+		NoAck:    true,
+	})
+	require_NoError(t, err)
+
+	// Since prior behavior did allow $JS.EVENT to be captured without no-ack, these might break
+	// on a server upgrade so make sure they still work ok without --no-ack.
+
+	// To avoid overlap error.
+	err = js.DeleteStream("TEST1")
+	require_NoError(t, err)
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "TEST4",
+		Subjects: []string{"$JS.EVENT.>"},
+	})
+	require_NoError(t, err)
+
+	// Also allow $SYS.ACCOUNT to be captured without no-ack, these also might break
+	// on a server upgrade so make sure they still work ok without --no-ack.
+
+	// To avoid overlap error.
+	err = js.DeleteStream("TEST3")
+	require_NoError(t, err)
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "TEST5",
+		Subjects: []string{"$SYS.ACCOUNT.>"},
+	})
+	require_NoError(t, err)
+
+	// We will test handling of ">" on a cluster here.
+	// Specific test for capturing everything which will require both no-ack and replicas of 1.
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js = jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{">"},
+	})
+	require_Error(t, err, NewJSStreamInvalidConfigError(errors.New("capturing all subjects requires no-ack to be true")))
+
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{">"},
+		Replicas: 3,
+		NoAck:    true,
+	})
+	require_Error(t, err, NewJSStreamInvalidConfigError(errors.New("capturing all subjects requires replicas of 1")))
+
+	// Ths should work ok.
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{">"},
+		Replicas: 1,
+		NoAck:    true,
+	})
+	require_NoError(t, err)
+}
+
+// https://github.com/nats-io/nats-server/issues/5570
+func TestJetStreamBadSubjectMappingStream(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	// Client for API requests.
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "test"})
+	require_NoError(t, err)
+
+	_, err = js.UpdateStream(&nats.StreamConfig{
+		Name: "test",
+		Sources: []*nats.StreamSource{
+			{
+				Name: "mapping",
+				SubjectTransforms: []nats.SubjectTransformConfig{
+					{
+						Source:      "events.*",
+						Destination: "events.{{wildcard(1)}}{{split(3,1)}}",
+					},
+				},
+			},
+		},
+	})
+	require_NoError(t, err)
+}
diff --git a/server/leafnode_test.go b/server/leafnode_test.go
index c46067ed..a1e5de0d 100644
--- a/server/leafnode_test.go
+++ b/server/leafnode_test.go
@@ -7751,3 +7751,111 @@ func TestLeafNodeDetectsStaleConnectionIfNoInfo(t *testing.T) {
 		})
 	}
 }
+
+// https://github.com/nats-io/nats-server/issues/5473
+func TestLeafNodeDupeDeliveryQueueSubAndPlainSub(t *testing.T) {
+	clusterCommonConf := `
+		accounts: {
+			tenant: {
+				users: [ { user:t, password: t } ]
+				exports: [{stream: system-a.events.>}]
+			}
+			system-a: {
+				users: [ { user:sa, password: sa } ]
+				imports: [
+					{stream: {subject: system-a.events.>, account: tenant}, prefix: tenant}
+				]
+			}
+			$SYS: { users = [ {user: "s", password: "s"} ] }
+		}
+		leafnodes {
+			remotes: [{
+				urls: [ "nats-leaf://sa:sa@127.0.0.1:17422" ]
+				account: system-a
+			}]
+		}`
+
+	confCluster0 := createConfFile(t, []byte(fmt.Sprintf(`
+		server_name: a-0
+		port: -1
+		cluster: {
+			name: cluster-a
+			listen: 127.0.0.1:16122
+			routes = [ nats://127.0.0.1:16123 ]
+			pool_size: -1
+		}
+		%s`, clusterCommonConf)))
+
+	confCluster1 := createConfFile(t, []byte(fmt.Sprintf(`
+		server_name: a-1
+		port: -1
+		cluster: {
+			name: cluster-a
+			listen: 127.0.0.1:16123
+			routes = [ nats://127.0.0.1:16122 ]
+			pool_size: -1
+		}
+		%s`, clusterCommonConf)))
+
+	serverB := createConfFile(t, []byte(`
+		server_name: b
+		port: -1
+		leafnodes: { port: 17422 }
+		accounts: {
+			system-a: {
+				users: [ { user: sa, password: sa } ]
+				exports: [{stream: *.system-a.>}]
+			}
+			system-b: {
+				users: [ { user: sb, password: sb } ]
+				imports: [ {stream: {subject: *.system-a.>, account: system-a }}]
+			}
+			$SYS: { users = [ {user: "s", password: "s"} ] }
+		}`))
+
+	// Start server B
+	srvB, _ := RunServerWithConfig(serverB)
+	defer srvB.Shutdown()
+
+	// Start the cluster servers.
+	srvA0, _ := RunServerWithConfig(confCluster0)
+	defer srvA0.Shutdown()
+	// Make sure this is connected first before starting the second server in cluster A.
+	checkLeafNodeConnectedCount(t, srvB, 1)
+	// Start second A server.
+	srvA1, _ := RunServerWithConfig(confCluster1)
+	defer srvA1.Shutdown()
+	// Make sure they are routed together.
+	checkNumRoutes(t, srvA0, 1)
+	checkNumRoutes(t, srvA1, 1)
+	// Make sure each cluster server is connected to server B.
+	checkLeafNodeConnectedCount(t, srvB, 2)
+
+	// Create plain subscriber on server B attached to system-b account.
+	ncB := natsConnect(t, srvB.ClientURL(), nats.UserInfo("sb", "sb"))
+	defer ncB.Close()
+	sub, err := ncB.SubscribeSync("*.system-a.events.>")
+	require_NoError(t, err)
+	// Create a new sub that has a queue group as well.
+	subq, err := ncB.QueueSubscribeSync("*.system-a.events.objectnotfound", "SYSB")
+	require_NoError(t, err)
+	ncB.Flush()
+	time.Sleep(250 * time.Millisecond)
+
+	// Connect to cluster A
+	ncA := natsConnect(t, srvA0.ClientURL(), nats.UserInfo("t", "t"))
+	defer ncA.Close()
+
+	err = ncA.Publish("system-a.events.objectnotfound", []byte("EventA"))
+	require_NoError(t, err)
+	ncA.Flush()
+	// Wait for them to be received.
+	time.Sleep(250 * time.Millisecond)
+
+	n, _, err := sub.Pending()
+	require_NoError(t, err)
+	require_Equal(t, n, 1)
+	n, _, err = subq.Pending()
+	require_NoError(t, err)
+	require_Equal(t, n, 1)
+}
diff --git a/server/memstore_test.go b/server/memstore_test.go
index e74976c4..867249a5 100644
--- a/server/memstore_test.go
+++ b/server/memstore_test.go
@@ -1052,6 +1052,31 @@ func TestMemStorePurgeExWithDeletedMsgs(t *testing.T) {
 	require_Equal(t, state.Msgs, 1)
 }
 
+// When all messages are deleted we should have a state of first = last + 1.
+func TestMemStoreDeleteAllFirstSequenceCheck(t *testing.T) {
+	cfg := &StreamConfig{
+		Name:     "zzz",
+		Subjects: []string{"foo"},
+		Storage:  MemoryStorage,
+	}
+	ms, err := newMemStore(cfg)
+	require_NoError(t, err)
+	defer ms.Stop()
+
+	msg := []byte("abc")
+	for i := 1; i <= 10; i++ {
+		ms.StoreMsg("foo", nil, msg)
+	}
+	for seq := uint64(1); seq <= 10; seq++ {
+		ms.RemoveMsg(seq)
+	}
+	var state StreamState
+	ms.FastState(&state)
+	require_Equal(t, state.FirstSeq, 11)
+	require_Equal(t, state.LastSeq, 10)
+	require_Equal(t, state.Msgs, 0)
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Benchmarks
 ///////////////////////////////////////////////////////////////////////////
diff --git a/server/mqtt_ex_test.go b/server/mqtt_ex_bench_test.go
similarity index 70%
rename from server/mqtt_ex_test.go
rename to server/mqtt_ex_bench_test.go
index 7ddcf0aa..d81ab3cd 100644
--- a/server/mqtt_ex_test.go
+++ b/server/mqtt_ex_bench_test.go
@@ -17,51 +17,12 @@
 package server
 
 import (
-	"bytes"
-	"encoding/json"
 	"fmt"
-	"os"
-	"os/exec"
 	"strconv"
 	"testing"
 	"time"
 )
 
-func TestMQTTExCompliance(t *testing.T) {
-	mqttPath := os.Getenv("MQTT_CLI")
-	if mqttPath == "" {
-		if p, err := exec.LookPath("mqtt"); err == nil {
-			mqttPath = p
-		}
-	}
-	if mqttPath == "" {
-		t.Skip(`"mqtt" command is not found in $PATH nor $MQTT_CLI. See https://hivemq.github.io/mqtt-cli/docs/installation/#debian-package for installation instructions`)
-	}
-
-	conf := createConfFile(t, []byte(fmt.Sprintf(`
-		listen: 127.0.0.1:-1
-		server_name: mqtt
-		jetstream {
-			store_dir = %q
-		}
-		mqtt {
-			listen: 127.0.0.1:-1
-		}
-	`, t.TempDir())))
-	s, o := RunServerWithConfig(conf)
-	defer testMQTTShutdownServer(s)
-
-	cmd := exec.Command(mqttPath, "test", "-V", "3", "-p", strconv.Itoa(o.MQTT.Port))
-
-	output, err := cmd.CombinedOutput()
-	t.Log(string(output))
-	if err != nil {
-		if exitError, ok := err.(*exec.ExitError); ok {
-			t.Fatalf("mqtt cli exited with error: %v", exitError)
-		}
-	}
-}
-
 const (
 	KB = 1024
 )
@@ -83,9 +44,6 @@ type mqttBenchContext struct {
 
 	Host string
 	Port int
-
-	// full path to mqtt-test command
-	testCmdPath string
 }
 
 var mqttBenchDefaultMatrix = mqttBenchMatrix{
@@ -102,8 +60,12 @@ type MQTTBenchmarkResult struct {
 	Bytes int64                    `json:"bytes"`
 }
 
-func BenchmarkMQTTEx(b *testing.B) {
-	bc := mqttNewBenchEx(b)
+func BenchmarkXMQTT(b *testing.B) {
+	if mqttTestCommandPath == "" {
+		b.Skip(`"mqtt-test" command is not found in $PATH.`)
+	}
+
+	bc := mqttBenchContext{}
 	b.Run("Server", func(b *testing.B) {
 		b.Cleanup(bc.startServer(b, false))
 		bc.runAll(b)
@@ -142,11 +104,11 @@ func (bc mqttBenchContext) benchmarkPub(b *testing.B) {
 
 	b.Run("PUB", func(b *testing.B) {
 		m.runMatrix(b, bc, func(b *testing.B, bc *mqttBenchContext) {
-			bc.runCommand(b, "pub",
+			bc.runAndReport(b, "pub",
 				"--qos", strconv.Itoa(bc.QOS),
-				"--n", strconv.Itoa(b.N),
+				"--messages", strconv.Itoa(b.N),
 				"--size", strconv.Itoa(bc.MessageSize),
-				"--num-publishers", strconv.Itoa(bc.Publishers),
+				"--publishers", strconv.Itoa(bc.Publishers),
 			)
 		})
 	})
@@ -165,11 +127,11 @@ func (bc mqttBenchContext) benchmarkPubRetained(b *testing.B) {
 
 	b.Run("PUBRET", func(b *testing.B) {
 		m.runMatrix(b, bc, func(b *testing.B, bc *mqttBenchContext) {
-			bc.runCommand(b, "pub", "--retain",
+			bc.runAndReport(b, "pub", "--retain",
 				"--qos", strconv.Itoa(bc.QOS),
-				"--n", strconv.Itoa(b.N),
+				"--messages", strconv.Itoa(b.N),
 				"--size", strconv.Itoa(bc.MessageSize),
-				"--num-publishers", strconv.Itoa(bc.Publishers),
+				"--publishers", strconv.Itoa(bc.Publishers),
 			)
 		})
 	})
@@ -185,11 +147,11 @@ func (bc mqttBenchContext) benchmarkPubSub(b *testing.B) {
 
 	b.Run("PUBSUB", func(b *testing.B) {
 		m.runMatrix(b, bc, func(b *testing.B, bc *mqttBenchContext) {
-			bc.runCommand(b, "pubsub",
+			bc.runAndReport(b, "pubsub",
 				"--qos", strconv.Itoa(bc.QOS),
-				"--n", strconv.Itoa(b.N),
+				"--messages", strconv.Itoa(b.N),
 				"--size", strconv.Itoa(bc.MessageSize),
-				"--num-subscribers", strconv.Itoa(bc.Subscribers),
+				"--subscribers", strconv.Itoa(bc.Subscribers),
 			)
 		})
 	})
@@ -206,67 +168,23 @@ func (bc mqttBenchContext) benchmarkSubRet(b *testing.B) {
 
 	b.Run("SUBRET", func(b *testing.B) {
 		m.runMatrix(b, bc, func(b *testing.B, bc *mqttBenchContext) {
-			bc.runCommand(b, "subret",
+			bc.runAndReport(b, "subret",
 				"--qos", strconv.Itoa(bc.QOS),
-				"--n", strconv.Itoa(b.N), // number of subscribe requests
-				"--num-subscribers", strconv.Itoa(bc.Subscribers),
-				"--num-topics", strconv.Itoa(bc.Topics),
+				"--topics", strconv.Itoa(bc.Topics), // number of retained messages
+				"--subscribers", strconv.Itoa(bc.Subscribers),
 				"--size", strconv.Itoa(bc.MessageSize),
+				"--repeat", strconv.Itoa(b.N), // number of subscribe requests
 			)
 		})
 	})
 }
 
-func mqttBenchLookupCommand(b *testing.B, name string) string {
+func (bc mqttBenchContext) runAndReport(b *testing.B, name string, extraArgs ...string) {
 	b.Helper()
-	cmd, err := exec.LookPath(name)
-	if err != nil {
-		b.Skipf("%q command is not found in $PATH. Please `go install github.com/nats-io/meta-nats/apps/go/mqtt/...@latest` and try again.", name)
-	}
-	return cmd
-}
-
-func (bc mqttBenchContext) runCommand(b *testing.B, name string, extraArgs ...string) {
-	b.Helper()
-
-	args := append([]string{
-		name,
-		"-q",
-		"--servers", fmt.Sprintf("%s:%d", bc.Host, bc.Port),
-	}, extraArgs...)
-
-	cmd := exec.Command(bc.testCmdPath, args...)
-	stdout, err := cmd.StdoutPipe()
-	if err != nil {
-		b.Fatalf("Error executing %q: %v", cmd.String(), err)
-	}
-	defer stdout.Close()
-	errbuf := bytes.Buffer{}
-	cmd.Stderr = &errbuf
-	if err = cmd.Start(); err != nil {
-		b.Fatalf("Error executing %q: %v", cmd.String(), err)
-	}
-	r := &MQTTBenchmarkResult{}
-	if err = json.NewDecoder(stdout).Decode(r); err != nil {
-		b.Fatalf("failed to decode output of %q: %v\n\n%s", cmd.String(), err, errbuf.String())
-	}
-	if err = cmd.Wait(); err != nil {
-		b.Fatalf("Error executing %q: %v\n\n%s", cmd.String(), err, errbuf.String())
-	}
-
+	r := mqttRunExCommandTest(b, name, mqttNewDial("", "", bc.Host, bc.Port, ""), extraArgs...)
 	r.report(b)
 }
 
-func (bc mqttBenchContext) initServer(b *testing.B) {
-	b.Helper()
-	bc.runCommand(b, "pubsub",
-		"--id", "__init__",
-		"--qos", "0",
-		"--n", "1",
-		"--size", "100",
-		"--num-subscribers", "1")
-}
-
 func (bc *mqttBenchContext) startServer(b *testing.B, disableRMSCache bool) func() {
 	b.Helper()
 	b.StopTimer()
@@ -278,7 +196,7 @@ func (bc *mqttBenchContext) startServer(b *testing.B, disableRMSCache bool) func
 	o = s.getOpts()
 	bc.Host = o.MQTT.Host
 	bc.Port = o.MQTT.Port
-	bc.initServer(b)
+	mqttInitTestServer(b, mqttNewDial("", "", bc.Host, bc.Port, ""))
 	return func() {
 		testMQTTShutdownServer(s)
 		testDisableRMSCache = prevDisableRMSCache
@@ -314,7 +232,7 @@ func (bc *mqttBenchContext) startCluster(b *testing.B, disableRMSCache bool) fun
 	o := cl.randomNonLeader().getOpts()
 	bc.Host = o.MQTT.Host
 	bc.Port = o.MQTT.Port
-	bc.initServer(b)
+	mqttInitTestServer(b, mqttNewDial("", "", bc.Host, bc.Port, ""))
 	return func() {
 		cl.shutdown()
 		testDisableRMSCache = prevDisableRMSCache
@@ -410,15 +328,7 @@ func (r MQTTBenchmarkResult) report(b *testing.B) {
 		nsOp := float64(ns) / float64(r.Ops)
 		b.ReportMetric(nsOp/1000000, unit+"_ms/op")
 	}
-
 	// Diable ReportAllocs() since it confuses the github benchmarking action
 	// with the noise.
 	// b.ReportAllocs()
 }
-
-func mqttNewBenchEx(b *testing.B) *mqttBenchContext {
-	cmd := mqttBenchLookupCommand(b, "mqtt-test")
-	return &mqttBenchContext{
-		testCmdPath: cmd,
-	}
-}
diff --git a/server/mqtt_ex_test_test.go b/server/mqtt_ex_test_test.go
new file mode 100644
index 00000000..5ae007ea
--- /dev/null
+++ b/server/mqtt_ex_test_test.go
@@ -0,0 +1,599 @@
+// Copyright 2024 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build !skip_mqtt_tests
+// +build !skip_mqtt_tests
+
+package server
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/nats-io/nuid"
+)
+
+type mqttDial string
+
+type mqttTarget struct {
+	singleServers []*Server
+	clusters      []*cluster
+	configs       []mqttTestConfig
+	all           []mqttDial
+}
+
+type mqttTestConfig struct {
+	name string
+	pub  []mqttDial
+	sub  []mqttDial
+}
+
+func TestXMQTTCompliance(t *testing.T) {
+	if mqttCLICommandPath == _EMPTY_ {
+		t.Skip(`"mqtt" command is not found in $PATH nor $MQTT_CLI. See https://hivemq.github.io/mqtt-cli/docs/installation/#debian-package for installation instructions`)
+	}
+
+	o := testMQTTDefaultOptions()
+	s := testMQTTRunServer(t, o)
+	o = s.getOpts()
+	defer testMQTTShutdownServer(s)
+
+	cmd := exec.Command(mqttCLICommandPath, "test", "-V", "3", "-p", strconv.Itoa(o.MQTT.Port))
+
+	output, err := cmd.CombinedOutput()
+	t.Log(string(output))
+	if err != nil {
+		if exitError, ok := err.(*exec.ExitError); ok {
+			t.Fatalf("mqtt cli exited with error: %v", exitError)
+		}
+	}
+}
+
+func TestXMQTTRetainedMessages(t *testing.T) {
+	if mqttTestCommandPath == _EMPTY_ {
+		t.Skip(`"mqtt-test" command is not found in $PATH.`)
+	}
+
+	for _, topo := range []struct {
+		name  string
+		makef func(testing.TB) *mqttTarget
+	}{
+		{
+			name:  "single server",
+			makef: mqttMakeTestServer,
+		},
+		{
+			name:  "server with leafnode",
+			makef: mqttMakeTestServerWithLeafnode("HUBD", "LEAFD", true),
+		},
+		{
+			name:  "server with leafnode no domains",
+			makef: mqttMakeTestServerWithLeafnode("", "", true),
+		},
+		{
+			name:  "server with leafnode no system account",
+			makef: mqttMakeTestServerWithLeafnode("HUBD", "LEAFD", false),
+		},
+		{
+			name:  "cluster",
+			makef: mqttMakeTestCluster(4, ""),
+		},
+		{
+			name:  "cluster with leafnode cluster",
+			makef: mqttMakeTestClusterWithLeafnodeCluster("HUBD", "LEAFD", true),
+		},
+		{
+			name:  "cluster with leafnode cluster no system account",
+			makef: mqttMakeTestClusterWithLeafnodeCluster("HUBD", "LEAFD", false),
+		},
+	} {
+		t.Run(topo.name, func(t *testing.T) {
+			target := topo.makef(t)
+			t.Cleanup(target.Shutdown)
+
+			// initialize the MQTT assets by "touching" all nodes in the
+			// cluster, but then reload to start with fresh server state.
+			for _, dial := range target.all {
+				mqttInitTestServer(t, dial)
+			}
+
+			numRMS := 100
+			strNumRMS := strconv.Itoa(numRMS)
+			topics := make([]string, len(target.configs))
+
+			for i, tc := range target.configs {
+				// Publish numRMS retained messages one at a time,
+				// round-robin across pub nodes. Remember the topic for each
+				// test config to check the subs after reload.
+				topic := "subret_" + nuid.Next()
+				topics[i] = topic
+				iNode := 0
+				for i := 0; i < numRMS; i++ {
+					pubTopic := fmt.Sprintf("%s/%d", topic, i)
+					dial := tc.pub[iNode%len(tc.pub)]
+					mqttRunExCommandTest(t, "pub", dial,
+						"--retain",
+						"--topic", pubTopic,
+						"--qos", "0",
+						"--size", "128", // message size 128 bytes
+					)
+					iNode++
+				}
+			}
+
+			// Check all sub nodes for retained messages
+			for i, tc := range target.configs {
+				for _, dial := range tc.sub {
+					mqttRunExCommandTest(t, "sub", dial,
+						"--retained", strNumRMS,
+						"--qos", "0",
+						"--topic", topics[i],
+					)
+				}
+			}
+
+			// Reload the target
+			target.Reload(t)
+
+			// Now check again
+			for i, tc := range target.configs {
+				for _, dial := range tc.sub {
+					mqttRunExCommandTestRetry(t, 1, "sub", dial,
+						"--retained", strNumRMS,
+						"--qos", "0",
+						"--topic", topics[i],
+					)
+				}
+			}
+		})
+	}
+}
+
+func mqttInitTestServer(tb testing.TB, dial mqttDial) {
+	tb.Helper()
+	mqttRunExCommandTestRetry(tb, 5, "pub", dial)
+}
+
+func mqttNewDialForServer(s *Server, username, password string) mqttDial {
+	o := s.getOpts().MQTT
+	return mqttNewDial(username, password, o.Host, o.Port, s.Name())
+}
+
+func mqttNewDial(username, password, host string, port int, comment string) mqttDial {
+	d := ""
+	switch {
+	case username != "" && password != "":
+		d = fmt.Sprintf("%s:%s@%s:%d", username, password, host, port)
+	case username != "":
+		d = fmt.Sprintf("%s@%s:%d", username, host, port)
+	default:
+		d = fmt.Sprintf("%s:%d", host, port)
+	}
+	if comment != "" {
+		d += "#" + comment
+	}
+	return mqttDial(d)
+}
+
+func (d mqttDial) Get() (u, p, s, c string) {
+	if d == "" {
+		return "", "", "127.0.0.1:1883", ""
+	}
+	in := string(d)
+	if i := strings.LastIndex(in, "#"); i != -1 {
+		c = in[i+1:]
+		in = in[:i]
+	}
+	if i := strings.LastIndex(in, "@"); i != -1 {
+		up := in[:i]
+		in = in[i+1:]
+		u = up
+		if i := strings.Index(up, ":"); i != -1 {
+			u = up[:i]
+			p = up[i+1:]
+		}
+	}
+	s = in
+	return u, p, s, c
+}
+
+func (d mqttDial) Name() string {
+	_, _, _, c := d.Get()
+	return c
+}
+
+func (t *mqttTarget) Reload(tb testing.TB) {
+	tb.Helper()
+
+	for _, c := range t.clusters {
+		c.stopAll()
+		c.restartAllSamePorts()
+	}
+
+	for i, s := range t.singleServers {
+		o := s.getOpts()
+		s.Shutdown()
+		t.singleServers[i] = testMQTTRunServer(tb, o)
+	}
+
+	for _, dial := range t.all {
+		mqttInitTestServer(tb, dial)
+	}
+}
+
+func (t *mqttTarget) Shutdown() {
+	for _, c := range t.clusters {
+		c.shutdown()
+	}
+	for _, s := range t.singleServers {
+		testMQTTShutdownServer(s)
+	}
+}
+
+func mqttMakeTestServer(tb testing.TB) *mqttTarget {
+	tb.Helper()
+	o := testMQTTDefaultOptions()
+	s := testMQTTRunServer(tb, o)
+	all := []mqttDial{mqttNewDialForServer(s, "", "")}
+	return &mqttTarget{
+		singleServers: []*Server{s},
+		all:           all,
+		configs: []mqttTestConfig{
+			{
+				name: "single server",
+				pub:  all,
+				sub:  all,
+			},
+		},
+	}
+}
+
+func mqttMakeTestServerWithLeafnode(hubd, leafd string, connectSystemAccount bool) func(tb testing.TB) *mqttTarget {
+	return func(tb testing.TB) *mqttTarget {
+		tb.Helper()
+
+		if hubd != "" {
+			hubd = "domain: " + hubd + ", "
+		}
+		sconf := `
+listen: 127.0.0.1:-1
+
+server_name: HUB
+jetstream: {max_mem_store: 256MB, max_file_store: 2GB, ` + hubd + `store_dir: '` + tb.TempDir() + `'}
+
+leafnodes {
+	listen: 127.0.0.1:-1
+}
+
+accounts {
+	ONE { users = [ { user: "one", pass: "p" } ]; jetstream: enabled }
+	$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
+}
+
+mqtt {
+	listen: 127.0.0.1:-1
+}
+`
+		hubConf := createConfFile(tb, []byte(sconf))
+		hubServer, o := RunServerWithConfig(hubConf)
+		leafRemoteAddr := fmt.Sprintf("%s:%d", o.LeafNode.Host, o.LeafNode.Port)
+		tb.Cleanup(func() {
+			os.Remove(hubConf)
+		})
+
+		sysRemote := ""
+		if connectSystemAccount {
+			sysRemote = `{ url: "nats://admin:s3cr3t!@` + leafRemoteAddr + `", account: "$SYS" },` + "\n\t\t"
+		}
+		if leafd != "" {
+			leafd = "domain: " + leafd + ", "
+		}
+		leafconf := `
+listen: 127.0.0.1:-1
+
+server_name: SPOKE
+jetstream: {max_mem_store: 256MB, max_file_store: 2GB, ` + leafd + `store_dir: '` + tb.TempDir() + `'}
+
+leafnodes {
+	remotes = [
+		` + sysRemote + `{ url: "nats://one:p@` + leafRemoteAddr + `", account: "ONE" },
+	]
+}
+
+accounts {
+	ONE { users = [ { user: "one", pass: "p" } ]; jetstream: enabled }
+	$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
+}
+
+mqtt {
+	listen: 127.0.0.1:-1
+}
+`
+		leafConf := createConfFile(tb, []byte(leafconf))
+		leafServer, _ := RunServerWithConfig(leafConf)
+		tb.Cleanup(func() {
+			os.Remove(leafConf)
+		})
+
+		both := []mqttDial{
+			mqttNewDialForServer(hubServer, "one", "p"),
+			mqttNewDialForServer(leafServer, "one", "p"),
+		}
+		return &mqttTarget{
+			singleServers: []*Server{hubServer, leafServer},
+			all:           both,
+			configs: []mqttTestConfig{
+				{name: "pub to all", pub: both, sub: both},
+				{name: "pub to SPOKE", pub: both[1:], sub: both},
+				{name: "pub to HUB", pub: both[:1], sub: both},
+			},
+		}
+	}
+}
+
+func mqttMakeTestCluster(size int, domain string) func(tb testing.TB) *mqttTarget {
+	return func(tb testing.TB) *mqttTarget {
+		tb.Helper()
+		if size < 3 {
+			tb.Fatal("cluster size must be at least 3")
+		}
+
+		if domain != "" {
+			domain = "domain: " + domain + ", "
+		}
+		clusterConf := `
+	listen: 127.0.0.1:-1
+
+	server_name: %s
+	jetstream: {max_mem_store: 256MB, max_file_store: 2GB, ` + domain + `store_dir: '%s'}
+
+	leafnodes {
+		listen: 127.0.0.1:-1
+	}
+
+	cluster {
+		name: %s
+		listen: 127.0.0.1:%d
+		routes = [%s]
+	}
+
+	mqtt {
+		listen: 127.0.0.1:-1
+		stream_replicas: 3
+	}
+
+	accounts {
+		ONE { users = [ { user: "one", pass: "p" } ]; jetstream: enabled }
+		$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
+	}
+`
+		cl := createJetStreamClusterWithTemplate(tb, clusterConf, "MQTT", size)
+		cl.waitOnLeader()
+
+		all := []mqttDial{}
+		for _, s := range cl.servers {
+			all = append(all, mqttNewDialForServer(s, "one", "p"))
+		}
+
+		return &mqttTarget{
+			clusters: []*cluster{cl},
+			all:      all,
+			configs: []mqttTestConfig{
+				{
+					name: "publish to one",
+					pub: []mqttDial{
+						mqttNewDialForServer(cl.randomServer(), "one", "p"),
+					},
+					sub: all,
+				},
+				{
+					name: "publish to all",
+					pub:  all,
+					sub:  all,
+				},
+			},
+		}
+	}
+}
+
+func mqttMakeTestClusterWithLeafnodeCluster(hubd, leafd string, connectSystemAccount bool) func(tb testing.TB) *mqttTarget {
+	return func(tb testing.TB) *mqttTarget {
+		tb.Helper()
+
+		// Create HUB cluster.
+		if hubd != "" {
+			hubd = "domain: " + hubd + ", "
+		}
+		hubConf := `
+	listen: 127.0.0.1:-1
+
+	server_name: %s
+	jetstream: {max_mem_store: 256MB, max_file_store: 2GB, ` + hubd + `store_dir: '%s'}
+
+	leafnodes {
+		listen: 127.0.0.1:-1
+	}
+
+	cluster {
+		name: %s
+		listen: 127.0.0.1:%d
+		routes = [%s]
+	}
+
+	mqtt {
+		listen: 127.0.0.1:-1
+		stream_replicas: 3
+	}
+
+	accounts {
+		ONE { users = [ { user: "one", pass: "p" } ]; jetstream: enabled }
+		$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
+	}
+`
+		hub := createJetStreamClusterWithTemplate(tb, hubConf, "HUB", 3)
+		hub.waitOnLeader()
+
+		// Pick a host to connect leafnodes to
+		lno := hub.randomNonLeader().getOpts().LeafNode
+		leafRemoteAddr := fmt.Sprintf("%s:%d", lno.Host, lno.Port)
+		hubRandom := mqttNewDialForServer(hub.randomNonLeader(), "one", "p")
+		hubAll := []mqttDial{}
+		for _, s := range hub.servers {
+			hubAll = append(hubAll, mqttNewDialForServer(s, "one", "p"))
+		}
+
+		// Create SPOKE (leafnode) cluster.
+		sysRemote := ""
+		if connectSystemAccount {
+			sysRemote = `{ url: "nats://admin:s3cr3t!@` + leafRemoteAddr + `", account: "$SYS" },` + "\n\t\t\t"
+		}
+		if leafd != "" {
+			leafd = "domain: " + leafd + ", "
+		}
+		leafConf := `
+	listen: 127.0.0.1:-1
+
+	server_name: %s
+	jetstream: {max_mem_store: 256MB, max_file_store: 2GB, ` + leafd + `store_dir: '%s'}
+
+	leafnodes {
+		remotes = [
+			` + sysRemote + `{ url: "nats://one:p@` + leafRemoteAddr + `", account: "ONE" },
+		]
+	}
+
+	cluster {
+		name: %s
+		listen: 127.0.0.1:%d
+		routes = [%s]
+	}
+
+	mqtt {
+		listen: 127.0.0.1:-1
+		stream_replicas: 3
+	}
+
+	accounts {
+		ONE { users = [ { user: "one", pass: "p" } ]; jetstream: enabled }
+		$SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] }
+	}
+`
+		spoke := createJetStreamCluster(tb, leafConf, "SPOKE", "SPOKE-", 3, 22111, false)
+		expectedConnections := 2
+		if !connectSystemAccount {
+			expectedConnections = 1
+		}
+		for _, s := range spoke.servers {
+			checkLeafNodeConnectedCount(tb, s, expectedConnections)
+		}
+		spoke.waitOnPeerCount(3)
+		spokeRandom := mqttNewDialForServer(spoke.randomNonLeader(), "one", "p")
+		spokeAll := []mqttDial{}
+		for _, s := range spoke.servers {
+			spokeAll = append(spokeAll, mqttNewDialForServer(s, "one", "p"))
+		}
+
+		all := append(hubAll, spokeAll...)
+
+		return &mqttTarget{
+			clusters: []*cluster{hub, spoke},
+			all:      all,
+			configs: []mqttTestConfig{
+				{name: "publish to all", pub: all, sub: all},
+				{name: "publish to all hub", pub: hubAll, sub: all},
+				{name: "publish to random in hub", pub: []mqttDial{hubRandom}, sub: all},
+				{name: "publish to all spoke", pub: spokeAll, sub: all},
+				{name: "publish to random in spoke", pub: []mqttDial{spokeRandom}, sub: all},
+			},
+		}
+	}
+}
+
+var mqttCLICommandPath = func() string {
+	p := os.Getenv("MQTT_CLI")
+	if p == "" {
+		p, _ = exec.LookPath("mqtt")
+	}
+	return p
+}()
+
+var mqttTestCommandPath = func() string {
+	p, _ := exec.LookPath("mqtt-test")
+	return p
+}()
+
+func mqttRunExCommandTest(tb testing.TB, subCommand string, dial mqttDial, extraArgs ...string) *MQTTBenchmarkResult {
+	tb.Helper()
+	return mqttRunExCommandTestRetry(tb, 1, subCommand, dial, extraArgs...)
+}
+
+func mqttRunExCommandTestRetry(tb testing.TB, n int, subCommand string, dial mqttDial, extraArgs ...string) (r *MQTTBenchmarkResult) {
+	tb.Helper()
+	var err error
+	for i := 0; i < n; i++ {
+		if r, err = mqttTryExCommandTest(tb, subCommand, dial, extraArgs...); err == nil {
+			return r
+		}
+
+		if i < (n - 1) {
+			tb.Logf("failed to %q %s to %q on attempt %v, will retry.", subCommand, extraArgs, dial.Name(), i)
+		} else {
+			tb.Fatal(err)
+		}
+	}
+	return nil
+}
+
+func mqttTryExCommandTest(tb testing.TB, subCommand string, dial mqttDial, extraArgs ...string) (r *MQTTBenchmarkResult, err error) {
+	tb.Helper()
+	if mqttTestCommandPath == "" {
+		tb.Skip(`"mqtt-test" command is not found in $PATH.`)
+	}
+
+	args := []string{subCommand, // "-q",
+		"-s", string(dial),
+	}
+	args = append(args, extraArgs...)
+	cmd := exec.Command(mqttTestCommandPath, args...)
+
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return nil, fmt.Errorf("error executing %q: %v", cmd.String(), err)
+	}
+	defer stdout.Close()
+	errbuf := bytes.Buffer{}
+	cmd.Stderr = &errbuf
+	if err = cmd.Start(); err != nil {
+		return nil, fmt.Errorf("error executing %q: %v", cmd.String(), err)
+	}
+	out, err := io.ReadAll(stdout)
+	if err != nil {
+		return nil, fmt.Errorf("error executing %q: failed to read output: %v", cmd.String(), err)
+	}
+	if err = cmd.Wait(); err != nil {
+		return nil, fmt.Errorf("error executing %q: %v\n\n%s\n\n%s", cmd.String(), err, string(out), errbuf.String())
+	}
+
+	r = &MQTTBenchmarkResult{}
+	if err := json.Unmarshal(out, r); err != nil {
+		tb.Fatalf("error executing %q: failed to decode output: %v\n\n%s\n\n%s", cmd.String(), err, string(out), errbuf.String())
+	}
+	return r, nil
+}
diff --git a/server/norace_test.go b/server/norace_test.go
index 87d86431..d91caec7 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -10569,3 +10569,207 @@ func TestNoRaceLargeNumDeletesStreamCatchups(t *testing.T) {
 		return nil
 	})
 }
+
+func TestNoRaceJetStreamClusterMemoryStreamLastSequenceResetAfterRestart(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	numStreams := 250
+	var wg sync.WaitGroup
+	wg.Add(numStreams)
+
+	for i := 1; i <= numStreams; i++ {
+		go func(n int) {
+			defer wg.Done()
+			_, err := js.AddStream(&nats.StreamConfig{
+				Name:     fmt.Sprintf("TEST:%d", n),
+				Storage:  nats.MemoryStorage,
+				Subjects: []string{fmt.Sprintf("foo.%d.*", n)},
+				Replicas: 3,
+			}, nats.MaxWait(30*time.Second))
+			require_NoError(t, err)
+			subj := fmt.Sprintf("foo.%d.bar", n)
+			for i := 0; i < 222; i++ {
+				js.Publish(subj, nil)
+			}
+		}(i)
+	}
+	wg.Wait()
+
+	// Make sure all streams have a snapshot in place to stress the snapshot logic for memory based streams.
+	for _, s := range c.servers {
+		for i := 1; i <= numStreams; i++ {
+			stream := fmt.Sprintf("TEST:%d", i)
+			mset, err := s.GlobalAccount().lookupStream(stream)
+			require_NoError(t, err)
+			node := mset.raftNode()
+			require_NotNil(t, node)
+			node.InstallSnapshot(mset.stateSnapshot())
+		}
+	}
+
+	// Do 5 rolling restarts waiting on healthz in between.
+	for i := 0; i < 5; i++ {
+		// Walk the servers and shut each down, and wipe the storage directory.
+		for _, s := range c.servers {
+			s.Shutdown()
+			s.WaitForShutdown()
+			s = c.restartServer(s)
+			checkFor(t, 30*time.Second, time.Second, func() error {
+				hs := s.healthz(nil)
+				if hs.Error != _EMPTY_ {
+					return errors.New(hs.Error)
+				}
+				return nil
+			})
+			// Make sure all streams are current after healthz returns ok.
+			for i := 1; i <= numStreams; i++ {
+				stream := fmt.Sprintf("TEST:%d", i)
+				mset, err := s.GlobalAccount().lookupStream(stream)
+				require_NoError(t, err)
+				var state StreamState
+				checkFor(t, 30*time.Second, time.Second, func() error {
+					mset.store.FastState(&state)
+					if state.LastSeq != 222 {
+						return fmt.Errorf("%v Wrong last sequence %d for %q - State  %+v", s, state.LastSeq, stream, state)
+					}
+					return nil
+				})
+			}
+		}
+	}
+}
+
+func TestNoRaceJetStreamClusterMemoryWorkQueueLastSequenceResetAfterRestart(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	numStreams := 50
+	var wg sync.WaitGroup
+	wg.Add(numStreams)
+
+	for i := 1; i <= numStreams; i++ {
+		go func(n int) {
+			defer wg.Done()
+
+			nc, js := jsClientConnect(t, c.randomServer())
+			defer nc.Close()
+
+			_, err := js.AddStream(&nats.StreamConfig{
+				Name:      fmt.Sprintf("TEST:%d", n),
+				Storage:   nats.MemoryStorage,
+				Retention: nats.WorkQueuePolicy,
+				Subjects:  []string{fmt.Sprintf("foo.%d.*", n)},
+				Replicas:  3,
+			}, nats.MaxWait(30*time.Second))
+			require_NoError(t, err)
+			subj := fmt.Sprintf("foo.%d.bar", n)
+			for i := 0; i < 22; i++ {
+				js.Publish(subj, nil)
+			}
+			// Now consumer them all as well.
+			sub, err := js.PullSubscribe(subj, "wq")
+			require_NoError(t, err)
+			msgs, err := sub.Fetch(22, nats.MaxWait(20*time.Second))
+			require_NoError(t, err)
+			require_Equal(t, len(msgs), 22)
+			for _, m := range msgs {
+				err := m.AckSync()
+				require_NoError(t, err)
+			}
+		}(i)
+	}
+	wg.Wait()
+
+	// Do 2 rolling restarts waiting on healthz in between.
+	for i := 0; i < 2; i++ {
+		// Walk the servers and shut each down, and wipe the storage directory.
+		for _, s := range c.servers {
+			s.Shutdown()
+			s.WaitForShutdown()
+			s = c.restartServer(s)
+			checkFor(t, 30*time.Second, time.Second, func() error {
+				hs := s.healthz(nil)
+				if hs.Error != _EMPTY_ {
+					return errors.New(hs.Error)
+				}
+				return nil
+			})
+			// Make sure all streams are current after healthz returns ok.
+			for i := 1; i <= numStreams; i++ {
+				stream := fmt.Sprintf("TEST:%d", i)
+				mset, err := s.GlobalAccount().lookupStream(stream)
+				require_NoError(t, err)
+				var state StreamState
+				checkFor(t, 20*time.Second, time.Second, func() error {
+					mset.store.FastState(&state)
+					if state.LastSeq != 22 {
+						return fmt.Errorf("%v Wrong last sequence %d for %q - State  %+v", s, state.LastSeq, stream, state)
+					}
+					if state.FirstSeq != 23 {
+						return fmt.Errorf("%v Wrong first sequence %d for %q - State  %+v", s, state.FirstSeq, stream, state)
+					}
+					return nil
+				})
+			}
+		}
+	}
+}
+
+func TestNoRaceJetStreamClusterMirrorSkipSequencingBug(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R5S", 5)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:              "ORIGIN",
+		Subjects:          []string{"foo.*"},
+		MaxMsgsPerSubject: 1,
+		Replicas:          1,
+		Storage:           nats.MemoryStorage,
+	})
+	require_NoError(t, err)
+
+	// Create a mirror with R5 such that it will be much slower than the ORIGIN.
+	_, err = js.AddStream(&nats.StreamConfig{
+		Name:     "M",
+		Replicas: 5,
+		Mirror:   &nats.StreamSource{Name: "ORIGIN"},
+	})
+	require_NoError(t, err)
+
+	// Connect new directly to ORIGIN
+	s := c.streamLeader(globalAccountName, "ORIGIN")
+	nc, js = jsClientConnect(t, s)
+	defer nc.Close()
+
+	// We are going to send at a high rate and also delete some along the way
+	// via the max msgs per limit.
+	for i := 0; i < 500_000; i++ {
+		subj := fmt.Sprintf("foo.%d", i)
+		js.PublishAsync(subj, nil)
+		// Create sequence holes every 100k.
+		if i%100_000 == 0 {
+			js.PublishAsync(subj, nil)
+		}
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	checkFor(t, 20*time.Second, time.Second, func() error {
+		si, err := js.StreamInfo("M")
+		require_NoError(t, err)
+		if si.State.Msgs != 500_000 {
+			return fmt.Errorf("Expected 1M msgs, got state: %+v", si.State)
+		}
+		return nil
+	})
+}
diff --git a/server/raft_helpers_test.go b/server/raft_helpers_test.go
index f8f12eef..127a3642 100644
--- a/server/raft_helpers_test.go
+++ b/server/raft_helpers_test.go
@@ -98,17 +98,26 @@ func (sg smGroup) unlockAll() {
 }
 
 // Create a raft group and place on numMembers servers at random.
+// Filestore based.
 func (c *cluster) createRaftGroup(name string, numMembers int, smf smFactory) smGroup {
+	return c.createRaftGroupEx(name, numMembers, smf, FileStorage)
+}
+
+func (c *cluster) createMemRaftGroup(name string, numMembers int, smf smFactory) smGroup {
+	return c.createRaftGroupEx(name, numMembers, smf, MemoryStorage)
+}
+
+func (c *cluster) createRaftGroupEx(name string, numMembers int, smf smFactory, st StorageType) smGroup {
 	c.t.Helper()
 	if numMembers > len(c.servers) {
 		c.t.Fatalf("Members > Peers: %d vs  %d", numMembers, len(c.servers))
 	}
 	servers := append([]*Server{}, c.servers...)
 	rand.Shuffle(len(servers), func(i, j int) { servers[i], servers[j] = servers[j], servers[i] })
-	return c.createRaftGroupWithPeers(name, servers[:numMembers], smf)
+	return c.createRaftGroupWithPeers(name, servers[:numMembers], smf, st)
 }
 
-func (c *cluster) createRaftGroupWithPeers(name string, servers []*Server, smf smFactory) smGroup {
+func (c *cluster) createRaftGroupWithPeers(name string, servers []*Server, smf smFactory, st StorageType) smGroup {
 	c.t.Helper()
 
 	var sg smGroup
@@ -122,12 +131,19 @@ func (c *cluster) createRaftGroupWithPeers(name string, servers []*Server, smf s
 	}
 
 	for _, s := range servers {
-		fs, err := newFileStore(
-			FileStoreConfig{StoreDir: c.t.TempDir(), BlockSize: defaultMediumBlockSize, AsyncFlush: false, SyncInterval: 5 * time.Minute},
-			StreamConfig{Name: name, Storage: FileStorage},
-		)
-		require_NoError(c.t, err)
-		cfg := &RaftConfig{Name: name, Store: c.t.TempDir(), Log: fs}
+		var cfg *RaftConfig
+		if st == FileStorage {
+			fs, err := newFileStore(
+				FileStoreConfig{StoreDir: c.t.TempDir(), BlockSize: defaultMediumBlockSize, AsyncFlush: false, SyncInterval: 5 * time.Minute},
+				StreamConfig{Name: name, Storage: FileStorage},
+			)
+			require_NoError(c.t, err)
+			cfg = &RaftConfig{Name: name, Store: c.t.TempDir(), Log: fs}
+		} else {
+			ms, err := newMemStore(&StreamConfig{Name: name, Storage: MemoryStorage})
+			require_NoError(c.t, err)
+			cfg = &RaftConfig{Name: name, Store: c.t.TempDir(), Log: ms}
+		}
 		s.bootstrapRaftNode(cfg, peers, true)
 		n, err := s.startRaftNode(globalAccountName, cfg, pprofLabels{})
 		require_NoError(c.t, err)
@@ -243,13 +259,20 @@ func (a *stateAdder) restart() {
 
 	// The filestore is stopped as well, so need to extract the parts to recreate it.
 	rn := a.n.(*raft)
-	fs := rn.wal.(*fileStore)
-
 	var err error
-	a.cfg.Log, err = newFileStore(fs.fcfg, fs.cfg.StreamConfig)
+
+	switch rn.wal.(type) {
+	case *fileStore:
+		fs := rn.wal.(*fileStore)
+		a.cfg.Log, err = newFileStore(fs.fcfg, fs.cfg.StreamConfig)
+	case *memStore:
+		ms := rn.wal.(*memStore)
+		a.cfg.Log, err = newMemStore(&ms.cfg)
+	}
 	if err != nil {
 		panic(err)
 	}
+
 	a.n, err = a.s.startRaftNode(globalAccountName, a.cfg, pprofLabels{})
 	if err != nil {
 		panic(err)
diff --git a/server/raft_test.go b/server/raft_test.go
index 90046bcd..91226407 100644
--- a/server/raft_test.go
+++ b/server/raft_test.go
@@ -1,4 +1,4 @@
-// Copyright 2021-2023 The NATS Authors
+// Copyright 2021-2024 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -14,6 +14,7 @@
 package server
 
 import (
+	"fmt"
 	"math"
 	"math/rand"
 	"os"
@@ -167,7 +168,7 @@ func TestNRGRecoverFromFollowingNoLeader(t *testing.T) {
 	for _, n := range rg {
 		rn := n.node().(*raft)
 		rn.ApplyQ().drain()
-		rn.switchToFollower("")
+		rn.switchToFollower(noLeader)
 	}
 
 	// Resume the nodes.
@@ -261,7 +262,7 @@ func TestNRGSimpleElection(t *testing.T) {
 	msg := require_ChanRead(t, voteReqs, time.Second)
 	vr := decodeVoteRequest(msg.Data, msg.Reply)
 	require_True(t, vr != nil)
-	require_NotEqual(t, vr.candidate, "")
+	require_NotEqual(t, vr.candidate, _EMPTY_)
 
 	// The leader should have bumped their term in order to start
 	// an election.
@@ -428,3 +429,123 @@ func TestNRGInvalidTAVDoesntPanic(t *testing.T) {
 	// Before the fix, a crash would have happened before this point.
 	c.waitOnAllCurrent()
 }
+
+func TestNRGCandidateStepsDownAfterAE(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+	c.waitOnLeader()
+
+	nc, _ := jsClientConnect(t, c.leader(), nats.UserInfo("admin", "s3cr3t!"))
+	defer nc.Close()
+
+	rg := c.createRaftGroup("TEST", 3, newStateAdder)
+	rg.waitOnLeader()
+
+	// Pick a random follower node. Bump the term up by a considerable
+	// amount and force it into the candidate state. This is what happens
+	// after a period of time in isolation.
+	n := rg.nonLeader().node().(*raft)
+	n.Lock()
+	n.term += 100
+	n.switchState(Candidate)
+	n.Unlock()
+
+	// Have the leader push through something on the current term just
+	// for good measure, although the heartbeats probably work too.
+	rg.leader().(*stateAdder).proposeDelta(1)
+
+	// Wait for the leader to receive the next append entry from the
+	// current leader. What should happen is that the node steps down
+	// and starts following the leader, as nothing in the log of the
+	// follower is newer than the term of the leader.
+	checkFor(t, time.Second, 50*time.Millisecond, func() error {
+		if n.State() == Candidate {
+			return fmt.Errorf("shouldn't still be candidate state")
+		}
+		if nterm, lterm := n.Term(), rg.leader().node().Term(); nterm != lterm {
+			return fmt.Errorf("follower term %d should match leader term %d", nterm, lterm)
+		}
+		return nil
+	})
+}
+
+// Test to make sure this does not cause us to truncate our wal or enter catchup state.
+func TestNRGHeartbeatOnLeaderChange(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	rg := c.createMemRaftGroup("TEST", 3, newStateAdder)
+	rg.waitOnLeader()
+
+	for i := 0; i < 10; i++ {
+		// Restart the leader.
+		leader := rg.leader().(*stateAdder)
+		leader.proposeDelta(22)
+		leader.proposeDelta(-11)
+		leader.proposeDelta(-11)
+		rg.waitOnTotal(t, 0)
+		leader.stop()
+		leader.restart()
+		rg.waitOnLeader()
+	}
+}
+
+func TestNRGElectionTimerAfterObserver(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	rg := c.createMemRaftGroup("TEST", 3, newStateAdder)
+	rg.waitOnLeader()
+
+	for _, n := range rg {
+		n.node().SetObserver(true)
+	}
+
+	time.Sleep(maxElectionTimeout)
+	before := time.Now()
+
+	for _, n := range rg {
+		n.node().SetObserver(false)
+	}
+
+	time.Sleep(maxCampaignTimeout)
+
+	for _, n := range rg {
+		rn := n.node().(*raft)
+		rn.RLock()
+		etlr := rn.etlr
+		rn.RUnlock()
+		require_True(t, etlr.After(before))
+	}
+}
+
+func TestNRGSystemClientCleanupFromAccount(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	s := c.randomServer()
+	sacc := s.SystemAccount()
+
+	numClients := func() int {
+		sacc.mu.RLock()
+		defer sacc.mu.RUnlock()
+		return len(sacc.clients)
+	}
+
+	start := numClients()
+
+	var all []smGroup
+	for i := 0; i < 5; i++ {
+		rgName := fmt.Sprintf("TEST-%d", i)
+		rg := c.createRaftGroup(rgName, 3, newStateAdder)
+		all = append(all, rg)
+		rg.waitOnLeader()
+	}
+	for _, rg := range all {
+		for _, sm := range rg {
+			sm.node().Stop()
+		}
+	}
+	finish := numClients()
+	require_Equal(t, start, finish)
+}
diff --git a/server/test_test.go b/server/test_test.go
index 58af5d76..cf7c5e9b 100644
--- a/server/test_test.go
+++ b/server/test_test.go
@@ -52,14 +52,14 @@ func RunRandClientPortServer(t *testing.T) *Server {
 	return RunServer(&opts)
 }
 
-func require_True(t *testing.T, b bool) {
+func require_True(t testing.TB, b bool) {
 	t.Helper()
 	if !b {
 		t.Fatalf("require true, but got false")
 	}
 }
 
-func require_False(t *testing.T, b bool) {
+func require_False(t testing.TB, b bool) {
 	t.Helper()
 	if b {
 		t.Fatalf("require false, but got true")
@@ -89,7 +89,7 @@ func require_Contains(t *testing.T, s string, subStrs ...string) {
 	}
 }
 
-func require_Error(t *testing.T, err error, expected ...error) {
+func require_Error(t testing.TB, err error, expected ...error) {
 	t.Helper()
 	if err == nil {
 		t.Fatalf("require error, but got none")
@@ -112,21 +112,21 @@ func require_Error(t *testing.T, err error, expected ...error) {
 	t.Fatalf("Expected one of %v, got '%v'", expected, err)
 }
 
-func require_Equal[T comparable](t *testing.T, a, b T) {
+func require_Equal[T comparable](t testing.TB, a, b T) {
 	t.Helper()
 	if a != b {
 		t.Fatalf("require %T equal, but got: %v != %v", a, a, b)
 	}
 }
 
-func require_NotEqual[T comparable](t *testing.T, a, b T) {
+func require_NotEqual[T comparable](t testing.TB, a, b T) {
 	t.Helper()
 	if a == b {
 		t.Fatalf("require %T not equal, but got: %v == %v", a, a, b)
 	}
 }
 
-func require_Len(t *testing.T, a, b int) {
+func require_Len(t testing.TB, a, b int) {
 	t.Helper()
 	if a != b {
 		t.Fatalf("require len, but got: %v != %v", a, b)
diff --git a/test/accounts_cycles_test.go b/test/accounts_cycles_test.go
index 71a4ed93..0b3bac30 100644
--- a/test/accounts_cycles_test.go
+++ b/test/accounts_cycles_test.go
@@ -489,6 +489,41 @@ func TestAccountImportSubjectMapping(t *testing.T) {
 	}
 }
 
+// bug: https://github.com/nats-io/nats-server/issues/1789
+func TestAccountCycleWithRenaming(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		accounts {
+		  A {
+		    exports [ { service: * } ]
+			imports [ { service { subject: foo, account: B }, to: foo } ]
+		  }
+		  B {
+		    exports [ { service: foo } ]
+			imports [ { service { subject: *, account: A }, to: "$1" } ]
+		  }
+		}
+	`))
+	if _, err := server.ProcessConfigFile(conf); err == nil || !strings.Contains(err.Error(), server.ErrImportFormsCycle.Error()) {
+		t.Fatalf("Expected an error on cycle service import, got none")
+	}
+
+	conf = createConfFile(t, []byte(`
+		accounts {
+		  A {
+		    exports [ { stream: * } ]
+			imports [ { stream { subject: foo, account: B }, to: foo } ]
+		  }
+		  B {
+		    exports [ { stream: foo } ]
+			imports [ { stream { subject: *, account: A }, to: "$1" } ]
+		  }
+		}
+	`))
+	if _, err := server.ProcessConfigFile(conf); err == nil || !strings.Contains(err.Error(), server.ErrImportFormsCycle.Error()) {
+		t.Fatalf("Expected an error on cycle service import, got none")
+	}
+}
+
 func clientConnectToServer(t *testing.T, s *server.Server) *nats.Conn {
 	t.Helper()
 	nc, err := nats.Connect(s.ClientURL(),