Repository: nats-io/nats-server
Pull request #4592 (2023-09-27 07:56:37)

[FIXED] Race condition in some leader failover scenarios leading to messages being potentially sourced more than once.

 - [X] Tests added
 - [X] Branch rebased on top of current main (`git pull --rebase origin main`)
 - [X] Changes squashed to a single commit (described [here](http://gitready.com/advanced/2009/02/10/squashing-commits-with-rebase.html))
 - [x] Build is green in Travis CI
 - [X] You have certified that the contribution is your original work and that you license the work to the project under the [Apache 2 license](https://github.com/nats-io/nats-server/blob/main/LICENSE)

### Changes proposed in this pull request:

Fixes a race condition in some leader failover scenarios leading to messages being potentially sourced more than once.

In some failure scenarios where the current leader of a stream sourcing from other stream(s) gets shutdown while publications are happening on the stream(s) being sourced leads to `setLeader(true)` being called on the new leader for the sourcing stream before all the messages having been sourced by the previous leader are completely processed such that when the new leader does it's reverse scan from the last message in it's view of the stream in order to know what sequence number to start the consumer for the stream being sourced from, such that the last message(s) sourced by the previous leader get sourced again, leading to some messages being sourced more than once.

The existing `TestNoRaceJetStreamSuperClusterSources` test would sidestep the issue by relying on the deduplication window in the sourcing stream. Without deduplication the test is a flapper.

This avoid the race condition by adding a small delay before scanning for the last message(s) having been sourced and starting the sources' consumer(s). Now the test (without using the deduplication window) never fails because more messages than expected have been received in the sourcing stream.

(Also adds a guard to give up if `setupSourceConsumers()` is called and we are no longer the leader for the stream (that check was already present in `setupMirrorConsumer()` so assuming it was forgotten for `setupSourceConsumers()`)


Patch:
diff --git a/go.mod b/go.mod
index 62917a58..038c514b 100644
--- a/go.mod
+++ b/go.mod
@@ -6,7 +6,7 @@ require (
 	github.com/klauspost/compress v1.17.0
 	github.com/minio/highwayhash v1.0.2
 	github.com/nats-io/jwt/v2 v2.5.2
-	github.com/nats-io/nats.go v1.30.2
+	github.com/nats-io/nats.go v1.29.0
 	github.com/nats-io/nkeys v0.4.5
 	github.com/nats-io/nuid v1.0.1
 	go.uber.org/automaxprocs v1.5.3
diff --git a/go.sum b/go.sum
index 72369f26..bc84b5c9 100644
--- a/go.sum
+++ b/go.sum
@@ -15,8 +15,8 @@ github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA
 github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
 github.com/nats-io/jwt/v2 v2.5.2 h1:DhGH+nKt+wIkDxM6qnVSKjokq5t59AZV5HRcFW0zJwU=
 github.com/nats-io/jwt/v2 v2.5.2/go.mod h1:24BeQtRwxRV8ruvC4CojXlx/WQ/VjuwlYiH+vu/+ibI=
-github.com/nats-io/nats.go v1.30.2 h1:aloM0TGpPorZKQhbAkdCzYDj+ZmsJDyeo3Gkbr72NuY=
-github.com/nats-io/nats.go v1.30.2/go.mod h1:dcfhUgmQNN4GJEfIb2f9R7Fow+gzBF4emzDHrVBd5qM=
+github.com/nats-io/nats.go v1.29.0 h1:dSXZ+SZeGyTdHVYeXimeq12FsIpb9dM8CJ2IZFiHcyE=
+github.com/nats-io/nats.go v1.29.0/go.mod h1:XpbWUlOElGwTYbMR7imivs7jJj9GtK7ypv321Wp6pjc=
 github.com/nats-io/nkeys v0.4.5 h1:Zdz2BUlFm4fJlierwvGK+yl20IAKUm7eV6AAZXEhkPk=
 github.com/nats-io/nkeys v0.4.5/go.mod h1:XUkxdLPTufzlihbamfzQ7mw/VGx6ObUs+0bN5sNvt64=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
diff --git a/server/accounts.go b/server/accounts.go
index c26c09a1..e7808551 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -1702,19 +1702,21 @@ func (a *Account) addReverseRespMapEntry(acc *Account, reply, from string) {
 // This will be called from checkForReverseEntry when the reply arg is a wildcard subject.
 // This will usually be called in a go routine since we need to walk all the entries.
 func (a *Account) checkForReverseEntries(reply string, checkInterest, recursed bool) {
-	if subjectIsLiteral(reply) {
-		a._checkForReverseEntry(reply, nil, checkInterest, recursed)
+	a.mu.RLock()
+	if len(a.imports.rrMap) == 0 {
+		a.mu.RUnlock()
 		return
 	}
 
-	a.mu.RLock()
-	if len(a.imports.rrMap) == 0 {
+	if subjectIsLiteral(reply) {
 		a.mu.RUnlock()
+		a._checkForReverseEntry(reply, nil, checkInterest, recursed)
 		return
 	}
 
 	var _rs [64]string
 	rs := _rs[:0]
+
 	if n := len(a.imports.rrMap); n > cap(rs) {
 		rs = make([]string, 0, n)
 	}
@@ -1724,14 +1726,9 @@ func (a *Account) checkForReverseEntries(reply string, checkInterest, recursed b
 	}
 	a.mu.RUnlock()
 
-	tsa := [32]string{}
-	tts := tokenizeSubjectIntoSlice(tsa[:0], reply)
-
-	rsa := [32]string{}
+	// subjectIsSubsetMatch is heavy so make sure we do this without the lock.
 	for _, r := range rs {
-		rts := tokenizeSubjectIntoSlice(rsa[:0], r)
-		//  isSubsetMatchTokenized is heavy so make sure we do this without the lock.
-		if isSubsetMatchTokenized(rts, tts) {
+		if subjectIsSubsetMatch(r, reply) {
 			a._checkForReverseEntry(r, nil, checkInterest, recursed)
 		}
 	}
diff --git a/server/client.go b/server/client.go
index e3364c8a..7a521047 100644
--- a/server/client.go
+++ b/server/client.go
@@ -3014,10 +3014,6 @@ func queueMatches(queue string, qsubs [][]*subscription) bool {
 
 // Low level unsubscribe for a given client.
 func (c *client) unsubscribe(acc *Account, sub *subscription, force, remove bool) {
-	if s := c.srv; s != nil && s.isShuttingDown() {
-		return
-	}
-
 	c.mu.Lock()
 	if !force && sub.max > 0 && sub.nm < sub.max {
 		c.Debugf(
@@ -3071,8 +3067,7 @@ func (c *client) unsubscribe(acc *Account, sub *subscription, force, remove bool
 	}
 
 	// Now check to see if this was part of a respMap entry for service imports.
-	// We can skip subscriptions on reserved replies.
-	if acc != nil && !isReservedReply(sub.subject) {
+	if acc != nil {
 		acc.checkForReverseEntry(string(sub.subject), nil, true)
 	}
 }
@@ -5082,23 +5077,6 @@ func (c *client) closeConnection(reason ClosedState) {
 		c.out.stc = nil
 	}
 
-	// If we have remote latency tracking running shut that down.
-	if c.rrTracking != nil {
-		c.rrTracking.ptmr.Stop()
-		c.rrTracking = nil
-	}
-
-	// If we are shutting down, no need to do all the accounting on subs, etc.
-	if reason == ServerShutdown {
-		s := c.srv
-		c.mu.Unlock()
-		if s != nil {
-			// Unregister
-			s.removeClient(c)
-		}
-		return
-	}
-
 	var (
 		kind        = c.kind
 		srv         = c.srv
@@ -5123,6 +5101,12 @@ func (c *client) closeConnection(reason ClosedState) {
 		spoke = c.isSpokeLeafNode()
 	}
 
+	// If we have remote latency tracking running shut that down.
+	if c.rrTracking != nil {
+		c.rrTracking.ptmr.Stop()
+		c.rrTracking = nil
+	}
+
 	c.mu.Unlock()
 
 	// Remove client's or leaf node or jetstream subscriptions.
diff --git a/server/const.go b/server/const.go
index 4d5d5e70..2de2835a 100644
--- a/server/const.go
+++ b/server/const.go
@@ -41,7 +41,7 @@ var (
 
 const (
 	// VERSION is the current version for the server.
-	VERSION = "2.10.2-RC.9"
+	VERSION = "2.10.2-RC.7"
 
 	// PROTO is the currently supported protocol.
 	// 0 was the original
diff --git a/server/filestore.go b/server/filestore.go
index b7674985..eb4c6940 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -60,9 +60,6 @@ type FileStoreConfig struct {
 	Cipher StoreCipher
 	// Compression is the algorithm to use when compressing.
 	Compression StoreCompression
-
-	// Internal reference to our server.
-	srv *Server
 }
 
 // FileStreamInfo allows us to remember created time.
@@ -390,7 +387,6 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 		qch:    make(chan struct{}),
 		fch:    make(chan struct{}, 1),
 		fsld:   make(chan struct{}),
-		srv:    fcfg.srv,
 	}
 
 	// Set flush in place to AsyncFlush which by default is false.
@@ -531,6 +527,12 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 	return fs, nil
 }
 
+func (fs *fileStore) registerServer(s *Server) {
+	fs.mu.Lock()
+	defer fs.mu.Unlock()
+	fs.srv = s
+}
+
 // Lock all existing message blocks.
 // Lock held on entry.
 func (fs *fileStore) lockAllMsgBlocks() {
@@ -1434,16 +1436,6 @@ func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, []uint64, error) {
 	return nil, tombstones, nil
 }
 
-// For doing warn logging.
-// Lock should be held.
-func (fs *fileStore) warn(format string, args ...any) {
-	// No-op if no server configured.
-	if fs.srv == nil {
-		return
-	}
-	fs.srv.Warnf(fmt.Sprintf("Filestore [%s] %s", fs.cfg.Name, format), args...)
-}
-
 // recoverFullState will attempt to receover our last full state and re-process any state changes
 // that happened afterwards.
 func (fs *fileStore) recoverFullState() (rerr error) {
@@ -1463,16 +1455,12 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 	dios <- struct{}{}
 
 	if err != nil {
-		if !os.IsNotExist(err) {
-			fs.warn("Could not read stream state file: %v", err)
-		}
 		return err
 	}
 
 	const minLen = 32
 	if len(buf) < minLen {
 		os.Remove(fn)
-		fs.warn("Stream state too short (%d bytes)", len(buf))
 		return errCorruptState
 	}
 
@@ -1483,7 +1471,6 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 	fs.hh.Write(buf)
 	if !bytes.Equal(h, fs.hh.Sum(nil)) {
 		os.Remove(fn)
-		fs.warn("Stream state checksum did not match")
 		return errCorruptState
 	}
 
@@ -1495,7 +1482,6 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 			ns := fs.aek.NonceSize()
 			buf, err = fs.aek.Open(nil, buf[:ns], buf[ns:], nil)
 			if err != nil {
-				fs.warn("Stream state error reading encryption key: %v", err)
 				return err
 			}
 		}
@@ -1503,7 +1489,6 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 
 	if buf[0] != fullStateMagic || buf[1] != fullStateVersion {
 		os.Remove(fn)
-		fs.warn("Stream state magic and version mismatch")
 		return errCorruptState
 	}
 
@@ -1558,7 +1543,6 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 			if lsubj := int(readU64()); lsubj > 0 {
 				if bi+lsubj > len(buf) {
 					os.Remove(fn)
-					fs.warn("Stream state bad subject len (%d)", lsubj)
 					return errCorruptState
 				}
 				subj := fs.subjString(buf[bi : bi+lsubj])
@@ -1589,7 +1573,6 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 				dmap, n, err := avl.Decode(buf[bi:])
 				if err != nil {
 					os.Remove(fn)
-					fs.warn("Stream state error decoding avl dmap: %v", err)
 					return errCorruptState
 				}
 				mb.dmap = *dmap
@@ -1622,7 +1605,6 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 	// Check if we had any errors.
 	if bi < 0 {
 		os.Remove(fn)
-		fs.warn("Stream state has no checksum present")
 		return errCorruptState
 	}
 
@@ -1639,9 +1621,9 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 			if ld, _, _ := mb.rebuildState(); ld != nil {
 				fs.addLostData(ld)
 			}
-			fs.warn("Stream state detected prior state, could not locate msg block %d", blkIndex)
 			return errPriorState
 		}
+
 		if matched = bytes.Equal(mb.lastChecksum(), lchk[:]); !matched {
 			// Remove the last message block since we will re-process below.
 			fs.removeMsgBlockFromList(mb)
@@ -1662,14 +1644,12 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 				return nil
 			}
 			os.Remove(fn)
-			fs.warn("Stream state could not recover msg block %d", bi)
 			return err
 		}
 		if nmb != nil {
 			// Check if we have to account for a partial message block.
 			if !matched && mb != nil && mb.index == nmb.index {
 				if err := fs.adjustAccounting(mb, nmb); err != nil {
-					fs.warn("Stream state could not adjust accounting: %v", err)
 					return err
 				}
 			}
@@ -2058,12 +2038,6 @@ func (fs *fileStore) expireMsgsOnRecover() {
 		// Clear any global subject state.
 		fs.psim = make(map[string]*psi)
 	}
-
-	// If we purged anything, make sure we kick flush state loop.
-	if purged > 0 {
-		fs.dirty++
-		fs.kickFlushStateLoop()
-	}
 }
 
 func copyMsgBlocks(src []*msgBlock) []*msgBlock {
@@ -3510,7 +3484,6 @@ func (fs *fileStore) removeMsg(seq uint64, secure, viaLimits, needFSLock bool) (
 			rbytes := mb.rbytes - uint64(mb.dmap.Size()*emptyRecordLen)
 			if rbytes>>2 > mb.bytes {
 				mb.compact()
-				fs.kickFlushStateLoop()
 			}
 		}
 	}
@@ -3662,7 +3635,7 @@ func (mb *msgBlock) compact() {
 	}
 
 	// Remove index file and wipe delete map, then rebuild.
-	mb.dmap.Empty()
+	mb.deleteDmap()
 	mb.rebuildStateLocked()
 
 	// If we entered with the msgs loaded make sure to reload them.
@@ -3671,6 +3644,11 @@ func (mb *msgBlock) compact() {
 	}
 }
 
+// Empty out our dmap.
+func (mb *msgBlock) deleteDmap() {
+	mb.dmap.Empty()
+}
+
 // Grab info from a slot.
 // Lock should be held.
 func (mb *msgBlock) slotInfo(slot int) (uint32, uint32, bool, error) {
@@ -6381,12 +6359,6 @@ func (fs *fileStore) reset() error {
 	fs.psim = make(map[string]*psi)
 	fs.bim = make(map[uint32]*msgBlock)
 
-	// If we purged anything, make sure we kick flush state loop.
-	if purged > 0 {
-		fs.dirty++
-		fs.kickFlushStateLoop()
-	}
-
 	fs.mu.Unlock()
 
 	if cb != nil {
@@ -6510,7 +6482,6 @@ func (fs *fileStore) removeMsgBlockFromList(mb *msgBlock) {
 	// Remove from list.
 	for i, omb := range fs.blks {
 		if mb == omb {
-			fs.dirty++
 			blks := append(fs.blks[:i], fs.blks[i+1:]...)
 			fs.blks = copyMsgBlocks(blks)
 			if fs.bim != nil {
diff --git a/server/gateway.go b/server/gateway.go
index 5c3b86f3..b7307e47 100644
--- a/server/gateway.go
+++ b/server/gateway.go
@@ -473,10 +473,6 @@ func (s *Server) startGateways() {
 // This starts the gateway accept loop in a go routine, unless it
 // is detected that the server has already been shutdown.
 func (s *Server) startGatewayAcceptLoop() {
-	if s.isShuttingDown() {
-		return
-	}
-
 	// Snapshot server options.
 	opts := s.getOpts()
 
@@ -486,6 +482,10 @@ func (s *Server) startGatewayAcceptLoop() {
 	}
 
 	s.mu.Lock()
+	if s.shutdown {
+		s.mu.Unlock()
+		return
+	}
 	hp := net.JoinHostPort(opts.Gateway.Host, strconv.Itoa(port))
 	l, e := natsListen("tcp", hp)
 	s.gatewayListenerErr = e
@@ -1575,7 +1575,7 @@ func (s *Server) addGatewayURL(urlStr string) bool {
 // Returns true if the URL has been removed, false otherwise.
 // Server lock held on entry
 func (s *Server) removeGatewayURL(urlStr string) bool {
-	if s.isShuttingDown() {
+	if s.shutdown {
 		return false
 	}
 	s.gateway.Lock()
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index dfade67d..53736a1b 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -1731,13 +1731,14 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, a *Account, s
 
 	var clusterWideConsCount int
 
-	js, cc := s.getJetStreamCluster()
-	if js == nil {
-		return
-	}
 	// If we are in clustered mode we need to be the stream leader to proceed.
-	if cc != nil {
+	if s.JetStreamIsClustered() {
 		// Check to make sure the stream is assigned.
+		js, cc := s.getJetStreamCluster()
+		if js == nil || cc == nil {
+			return
+		}
+
 		js.mu.RLock()
 		isLeader, sa := cc.isLeader(), js.streamAssignment(acc.Name, streamName)
 		var offline bool
@@ -1832,23 +1833,15 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, a *Account, s
 	}
 
 	mset, err := acc.lookupStream(streamName)
-	// Error is not to be expected at this point, but could happen if same stream trying to be created.
 	if err != nil {
-		if cc != nil {
-			// This could be inflight, pause for a short bit and try again.
-			// This will not be inline, so ok.
-			time.Sleep(10 * time.Millisecond)
-			mset, err = acc.lookupStream(streamName)
-		}
-		// Check again.
-		if err != nil {
-			resp.Error = NewJSStreamNotFoundError(Unless(err))
-			s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
-			return
-		}
+		resp.Error = NewJSStreamNotFoundError(Unless(err))
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
 	}
 	config := mset.config()
 
+	js, _ := s.getJetStreamCluster()
+
 	resp.StreamInfo = &StreamInfo{
 		Created:    mset.createdTime(),
 		State:      mset.stateWithDetail(details),
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index b5c02bd5..4c2a7a1d 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -198,15 +198,11 @@ func (s *Server) trackedJetStreamServers() (js, total int) {
 }
 
 func (s *Server) getJetStreamCluster() (*jetStream, *jetStreamCluster) {
-	if s.isShuttingDown() {
-		return nil, nil
-	}
-
 	s.mu.RLock()
-	js := s.js
+	shutdown, js := s.shutdown, s.js
 	s.mu.RUnlock()
 
-	if js == nil {
+	if shutdown || js == nil {
 		return nil, nil
 	}
 
@@ -755,7 +751,7 @@ func (js *jetStream) setupMetaGroup() error {
 	storeDir := filepath.Join(js.config.StoreDir, sysAcc.Name, defaultStoreDirName, defaultMetaGroupName)
 
 	fs, err := newFileStoreWithCreated(
-		FileStoreConfig{StoreDir: storeDir, BlockSize: defaultMetaFSBlkSize, AsyncFlush: false, srv: s},
+		FileStoreConfig{StoreDir: storeDir, BlockSize: defaultMetaFSBlkSize, AsyncFlush: false},
 		StreamConfig{Name: defaultMetaGroupName, Storage: FileStorage},
 		time.Now().UTC(),
 		s.jsKeyGen(s.getOpts().JetStreamKey, defaultMetaGroupName),
@@ -766,6 +762,9 @@ func (js *jetStream) setupMetaGroup() error {
 		return err
 	}
 
+	// Register our server.
+	fs.registerServer(s)
+
 	cfg := &RaftConfig{Name: defaultMetaGroupName, Store: storeDir, Log: fs}
 
 	// If we are soliciting leafnode connections and we are sharing a system account and do not disable it with a hint,
@@ -2029,7 +2028,7 @@ func (js *jetStream) createRaftGroup(accName string, rg *raftGroup, storage Stor
 	var store StreamStore
 	if storage == FileStorage {
 		fs, err := newFileStoreWithCreated(
-			FileStoreConfig{StoreDir: storeDir, BlockSize: defaultMediumBlockSize, AsyncFlush: false, SyncInterval: 5 * time.Minute, srv: s},
+			FileStoreConfig{StoreDir: storeDir, BlockSize: defaultMediumBlockSize, AsyncFlush: false, SyncInterval: 5 * time.Minute},
 			StreamConfig{Name: rg.Name, Storage: FileStorage},
 			time.Now().UTC(),
 			s.jsKeyGen(s.getOpts().JetStreamKey, rg.Name),
@@ -2039,6 +2038,8 @@ func (js *jetStream) createRaftGroup(accName string, rg *raftGroup, storage Stor
 			s.Errorf("Error creating filestore WAL: %v", err)
 			return err
 		}
+		// Register our server.
+		fs.registerServer(s)
 		store = fs
 	} else {
 		ms, err := newMemStore(&StreamConfig{Name: rg.Name, Storage: MemoryStorage})
diff --git a/server/leafnode.go b/server/leafnode.go
index d5d41c53..47bb8087 100644
--- a/server/leafnode.go
+++ b/server/leafnode.go
@@ -681,11 +681,11 @@ func (s *Server) startLeafNodeAcceptLoop() {
 		port = 0
 	}
 
-	if s.isShuttingDown() {
+	s.mu.Lock()
+	if s.shutdown {
+		s.mu.Unlock()
 		return
 	}
-
-	s.mu.Lock()
 	hp := net.JoinHostPort(opts.LeafNode.Host, strconv.Itoa(port))
 	l, e := natsListen("tcp", hp)
 	s.leafNodeListenerErr = e
@@ -878,7 +878,7 @@ func (s *Server) addLeafNodeURL(urlStr string) bool {
 func (s *Server) removeLeafNodeURL(urlStr string) bool {
 	// Don't need to do this if we are removing the route connection because
 	// we are shuting down...
-	if s.isShuttingDown() {
+	if s.shutdown {
 		return false
 	}
 	if s.leafURLsMap.removeUrl(urlStr) {
diff --git a/server/mqtt.go b/server/mqtt.go
index e3333339..b285af35 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -425,10 +425,6 @@ type mqttParsedPublishNATSHeader struct {
 }
 
 func (s *Server) startMQTT() {
-	if s.isShuttingDown() {
-		return
-	}
-
 	sopts := s.getOpts()
 	o := &sopts.MQTT
 
@@ -441,6 +437,10 @@ func (s *Server) startMQTT() {
 	}
 	hp := net.JoinHostPort(o.Host, strconv.Itoa(port))
 	s.mu.Lock()
+	if s.shutdown {
+		s.mu.Unlock()
+		return
+	}
 	s.mqtt.sessmgr.sessions = make(map[string]*mqttAccountSessionManager)
 	hl, err = net.Listen("tcp", hp)
 	s.mqtt.listenerErr = err
@@ -500,7 +500,7 @@ func (s *Server) createMQTTClient(conn net.Conn, ws *websocket) *client {
 
 	s.mu.Lock()
 	if !s.isRunning() || s.ldm {
-		if s.isShuttingDown() {
+		if s.shutdown {
 			conn.Close()
 		}
 		s.mu.Unlock()
diff --git a/server/route.go b/server/route.go
index 493f03e4..45732fda 100644
--- a/server/route.go
+++ b/server/route.go
@@ -653,14 +653,11 @@ func (c *client) processRouteInfo(info *Info) {
 		// We receive an INFO from a server that informs us about another server,
 		// so the info.ID in the INFO protocol does not match the ID of this route.
 		if remoteID != _EMPTY_ && remoteID != info.ID {
-			// We want to know if the existing route supports pooling/pinned-account
-			// or not when processing the implicit route.
-			noPool := c.route.noPool
 			c.mu.Unlock()
 
 			// Process this implicit route. We will check that it is not an explicit
 			// route and/or that it has not been connected already.
-			s.processImplicitRoute(info, noPool)
+			s.processImplicitRoute(info)
 			return
 		}
 
@@ -815,14 +812,10 @@ func (c *client) processRouteInfo(info *Info) {
 		}
 	}
 	// For accounts that are configured to have their own route:
-	// If this is a solicited route, we already have c.route.accName set in createRoute.
+	// If this is a solicit route, we already have c.route.accName set in createRoute.
 	// For non solicited route (the accept side), we will set the account name that
 	// is present in the INFO protocol.
-	if didSolicit && len(c.route.accName) > 0 {
-		// Set it in the info.RouteAccount so that addRoute can use that
-		// and we properly gossip that this is a route for an account.
-		info.RouteAccount = string(c.route.accName)
-	} else if !didSolicit && info.RouteAccount != _EMPTY_ {
+	if !didSolicit {
 		c.route.accName = []byte(info.RouteAccount)
 	}
 	accName := string(c.route.accName)
@@ -984,7 +977,7 @@ func (s *Server) updateRemoteRoutePerms(c *client, info *Info) {
 func (s *Server) sendAsyncInfoToClients(regCli, wsCli bool) {
 	// If there are no clients supporting async INFO protocols, we are done.
 	// Also don't send if we are shutting down...
-	if s.cproto == 0 || s.isShuttingDown() {
+	if s.cproto == 0 || s.shutdown {
 		return
 	}
 	info := s.copyInfo()
@@ -1009,7 +1002,7 @@ func (s *Server) sendAsyncInfoToClients(regCli, wsCli bool) {
 // This will process implicit route information received from another server.
 // We will check to see if we have configured or are already connected,
 // and if so we will ignore. Otherwise we will attempt to connect.
-func (s *Server) processImplicitRoute(info *Info, routeNoPool bool) {
+func (s *Server) processImplicitRoute(info *Info) {
 	remoteID := info.ID
 
 	s.mu.Lock()
@@ -1019,16 +1012,8 @@ func (s *Server) processImplicitRoute(info *Info, routeNoPool bool) {
 	if remoteID == s.info.ID {
 		return
 	}
-
-	// Snapshot server options.
-	opts := s.getOpts()
-
 	// Check if this route already exists
 	if accName := info.RouteAccount; accName != _EMPTY_ {
-		// If we don't support pooling/pinned account, bail.
-		if opts.Cluster.PoolSize <= 0 {
-			return
-		}
 		if remotes, ok := s.accRoutes[accName]; ok {
 			if r := remotes[remoteID]; r != nil {
 				return
@@ -1049,22 +1034,13 @@ func (s *Server) processImplicitRoute(info *Info, routeNoPool bool) {
 		return
 	}
 
+	// Snapshot server options.
+	opts := s.getOpts()
+
 	if info.AuthRequired {
 		r.User = url.UserPassword(opts.Cluster.Username, opts.Cluster.Password)
 	}
 	s.startGoRoutine(func() { s.connectToRoute(r, false, true, info.RouteAccount) })
-	// If we are processing an implicit route from a route that does not
-	// support pooling/pinned-accounts, we won't receive an INFO for each of
-	// the pinned-accounts that we would normally receive. In that case, just
-	// initiate routes for all our configured pinned accounts.
-	if routeNoPool && info.RouteAccount == _EMPTY_ && len(opts.Cluster.PinnedAccounts) > 0 {
-		// Copy since we are going to pass as closure to a go routine.
-		rURL := r
-		for _, an := range opts.Cluster.PinnedAccounts {
-			accName := an
-			s.startGoRoutine(func() { s.connectToRoute(rURL, false, true, accName) })
-		}
-	}
 }
 
 // hasThisRouteConfigured returns true if info.Host:info.Port is present
@@ -1095,10 +1071,7 @@ func (s *Server) forwardNewRouteInfoToKnownServers(info *Info) {
 
 	s.forEachRemote(func(r *client) {
 		r.mu.Lock()
-		// If this is a new route for a given account, do not send to a server
-		// that does not support pooling/pinned-accounts.
-		if r.route.remoteID != info.ID &&
-			(info.RouteAccount == _EMPTY_ || (info.RouteAccount != _EMPTY_ && !r.route.noPool)) {
+		if r.route.remoteID != info.ID {
 			r.enqueueProto(infoJSON)
 		}
 		r.mu.Unlock()
@@ -1882,7 +1855,7 @@ func (s *Server) addRoute(c *client, didSolicit bool, info *Info, accName string
 	// server and need to handle things differently.
 	if info.RoutePoolSize <= 0 || opts.Cluster.PoolSize < 0 {
 		if accName != _EMPTY_ {
-			invProtoErr = fmt.Sprintf("Not possible to have a dedicated route for account %q between those servers", accName)
+			invProtoErr = fmt.Sprintf("Not possible to have a dedicate route for account %q between those servers", accName)
 			// In this case, make sure this route does not attempt to reconnect
 			c.setNoReconnect()
 		} else {
@@ -2329,10 +2302,6 @@ func (s *Server) updateRouteSubscriptionMap(acc *Account, sub *subscription, del
 // is detected that the server has already been shutdown.
 // It will also start soliciting explicit routes.
 func (s *Server) startRouteAcceptLoop() {
-	if s.isShuttingDown() {
-		return
-	}
-
 	// Snapshot server options.
 	opts := s.getOpts()
 
@@ -2347,6 +2316,10 @@ func (s *Server) startRouteAcceptLoop() {
 	clusterName := s.ClusterName()
 
 	s.mu.Lock()
+	if s.shutdown {
+		s.mu.Unlock()
+		return
+	}
 	s.Noticef("Cluster name is %s", clusterName)
 	if s.isClusterNameDynamic() {
 		s.Warnf("Cluster name was dynamically generated, consider setting one")
@@ -2758,7 +2731,6 @@ func (s *Server) removeRoute(c *client) {
 		opts          = s.getOpts()
 		rURL          *url.URL
 		noPool        bool
-		didSolicit    bool
 	)
 	c.mu.Lock()
 	cid := c.cid
@@ -2777,7 +2749,6 @@ func (s *Server) removeRoute(c *client) {
 		connectURLs = r.connectURLs
 		wsConnectURLs = r.wsConnURLs
 		rURL = r.url
-		didSolicit = r.didSolicit
 	}
 	c.mu.Unlock()
 	if accName != _EMPTY_ {
@@ -2836,18 +2807,10 @@ func (s *Server) removeRoute(c *client) {
 			if lnURL != _EMPTY_ && s.removeLeafNodeURL(lnURL) {
 				s.sendAsyncLeafNodeInfo()
 			}
-			// If this server has pooling/pinned accounts and the route for
-			// this remote was a "no pool" route, attempt to reconnect.
-			if noPool {
-				if s.routesPoolSize > 1 {
-					s.startGoRoutine(func() { s.connectToRoute(rURL, didSolicit, true, _EMPTY_) })
-				}
-				if len(opts.Cluster.PinnedAccounts) > 0 {
-					for _, an := range opts.Cluster.PinnedAccounts {
-						accName := an
-						s.startGoRoutine(func() { s.connectToRoute(rURL, didSolicit, true, accName) })
-					}
-				}
+			// If this server has pooling and the route for this remote
+			// was a "no pool" route, attempt to reconnect.
+			if s.routesPoolSize > 1 && noPool {
+				s.startGoRoutine(func() { s.connectToRoute(rURL, true, true, _EMPTY_) })
 			}
 		}
 		// This is for gateway code. Remove this route from a map that uses
diff --git a/server/server.go b/server/server.go
index 5db5ca49..8a340961 100644
--- a/server/server.go
+++ b/server/server.go
@@ -132,7 +132,7 @@ type Server struct {
 	optsMu              sync.RWMutex
 	opts                *Options
 	running             atomic.Bool
-	shutdown            atomic.Bool
+	shutdown            bool
 	listener            net.Listener
 	listenerErr         error
 	gacc                *Account
@@ -2350,10 +2350,6 @@ func (s *Server) Start() {
 	s.startOCSPResponseCache()
 }
 
-func (s *Server) isShuttingDown() bool {
-	return s.shutdown.Load()
-}
-
 // Shutdown will shutdown the server instance by kicking out the AcceptLoop
 // and closing all associated clients.
 func (s *Server) Shutdown() {
@@ -2373,19 +2369,19 @@ func (s *Server) Shutdown() {
 	// eventing items associated with accounts.
 	s.shutdownEventing()
 
+	s.mu.Lock()
 	// Prevent issues with multiple calls.
-	if s.isShuttingDown() {
+	if s.shutdown {
+		s.mu.Unlock()
 		return
 	}
-
-	s.mu.Lock()
 	s.Noticef("Initiating Shutdown...")
 
 	accRes := s.accResolver
 
 	opts := s.getOpts()
 
-	s.shutdown.Store(true)
+	s.shutdown = true
 	s.running.Store(false)
 	s.grMu.Lock()
 	s.grRunning = false
@@ -2549,15 +2545,16 @@ func (s *Server) AcceptLoop(clr chan struct{}) {
 		}
 	}()
 
-	if s.isShuttingDown() {
-		return
-	}
-
 	// Snapshot server options.
 	opts := s.getOpts()
 
 	// Setup state that can enable shutdown
 	s.mu.Lock()
+	if s.shutdown {
+		s.mu.Unlock()
+		return
+	}
+
 	hp := net.JoinHostPort(opts.Host, strconv.Itoa(opts.Port))
 	l, e := natsListen("tcp", hp)
 	s.listenerErr = e
@@ -2680,10 +2677,6 @@ func (s *Server) setInfoHostPort() error {
 
 // StartProfiler is called to enable dynamic profiling.
 func (s *Server) StartProfiler() {
-	if s.isShuttingDown() {
-		return
-	}
-
 	// Snapshot server options.
 	opts := s.getOpts()
 
@@ -2695,7 +2688,12 @@ func (s *Server) StartProfiler() {
 	}
 
 	s.mu.Lock()
+	if s.shutdown {
+		s.mu.Unlock()
+		return
+	}
 	hp := net.JoinHostPort(opts.Host, strconv.Itoa(port))
+
 	l, err := net.Listen("tcp", hp)
 
 	if err != nil {
@@ -2719,7 +2717,10 @@ func (s *Server) StartProfiler() {
 		// if this errors out, it's probably because the server is being shutdown
 		err := srv.Serve(l)
 		if err != nil {
-			if !s.isShuttingDown() {
+			s.mu.Lock()
+			shutdown := s.shutdown
+			s.mu.Unlock()
+			if !shutdown {
 				s.Fatalf("error starting profiler: %s", err)
 			}
 		}
@@ -2826,10 +2827,6 @@ func (s *Server) getMonitoringTLSConfig(_ *tls.ClientHelloInfo) (*tls.Config, er
 
 // Start the monitoring server
 func (s *Server) startMonitoring(secure bool) error {
-	if s.isShuttingDown() {
-		return nil
-	}
-
 	// Snapshot server options.
 	opts := s.getOpts()
 
@@ -2911,6 +2908,11 @@ func (s *Server) startMonitoring(secure bool) error {
 		ErrorLog:       log.New(&captureHTTPServerLog{s, "monitoring: "}, _EMPTY_, 0),
 	}
 	s.mu.Lock()
+	if s.shutdown {
+		httpListener.Close()
+		s.mu.Unlock()
+		return nil
+	}
 	s.http = httpListener
 	s.httpHandler = mux
 	s.monitoringServer = srv
@@ -2918,7 +2920,10 @@ func (s *Server) startMonitoring(secure bool) error {
 
 	go func() {
 		if err := srv.Serve(httpListener); err != nil {
-			if !s.isShuttingDown() {
+			s.mu.Lock()
+			shutdown := s.shutdown
+			s.mu.Unlock()
+			if !shutdown {
 				s.Fatalf("Error starting monitor on %q: %v", hp, err)
 			}
 		}
@@ -3060,7 +3065,7 @@ func (s *Server) createClientEx(conn net.Conn, inProcess bool) *client {
 		// clients would branch here (since server is not running). However,
 		// when a server was really running and has been shutdown, we must
 		// close this connection.
-		if s.isShuttingDown() {
+		if s.shutdown {
 			conn.Close()
 		}
 		s.mu.Unlock()
@@ -3967,7 +3972,7 @@ func (s *Server) isLameDuckMode() bool {
 func (s *Server) lameDuckMode() {
 	s.mu.Lock()
 	// Check if there is actually anything to do
-	if s.isShuttingDown() || s.ldm || s.listener == nil {
+	if s.shutdown || s.ldm || s.listener == nil {
 		s.mu.Unlock()
 		return
 	}
@@ -4018,7 +4023,7 @@ func (s *Server) lameDuckMode() {
 
 	s.mu.Lock()
 	// Need to recheck few things
-	if s.isShuttingDown() || len(s.clients) == 0 {
+	if s.shutdown || len(s.clients) == 0 {
 		s.mu.Unlock()
 		// If there is no client, we need to call Shutdown() to complete
 		// the LDMode. If server has been shutdown while lock was released,
diff --git a/server/stream.go b/server/stream.go
index 80afc26c..4ec73f5b 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -246,8 +246,9 @@ type stream struct {
 	mirror *sourceInfo
 
 	// Sources
-	sources       map[string]*sourceInfo
-	sourceRetries map[string]*time.Timer
+	sources              map[string]*sourceInfo
+	sourceRetries        map[string]*time.Timer
+	sourcesConsumerSetup *time.Timer
 
 	// Indicates we have direct consumers.
 	directs int
@@ -821,6 +822,11 @@ func (mset *stream) setLeader(isLeader bool) error {
 			return err
 		}
 	} else {
+		// cancel timer to create the source consumers if not fired yet
+		if mset.sourcesConsumerSetup != nil {
+			mset.sourcesConsumerSetup.Stop()
+			mset.sourcesConsumerSetup = nil
+		}
 		// Stop responding to sync requests.
 		mset.stopClusterSubs()
 		// Unsubscribe from direct stream.
@@ -2389,7 +2395,7 @@ func (mset *stream) scheduleSetupMirrorConsumerRetryAsap() {
 	}
 	// To make *sure* that the next request will not fail, add a bit of buffer
 	// and some randomness.
-	next += time.Duration(rand.Intn(50)) + 10*time.Millisecond
+	next += time.Duration(rand.Intn(int(10*time.Millisecond))) + 10*time.Millisecond
 	time.AfterFunc(next, func() {
 		mset.mu.Lock()
 		mset.setupMirrorConsumer()
@@ -2736,7 +2742,7 @@ func (mset *stream) scheduleSetSourceConsumerRetryAsap(si *sourceInfo, seq uint6
 	}
 	// To make *sure* that the next request will not fail, add a bit of buffer
 	// and some randomness.
-	next += time.Duration(rand.Intn(50)) + 10*time.Millisecond
+	next += time.Duration(rand.Intn(int(10*time.Millisecond))) + 10*time.Millisecond
 	mset.scheduleSetSourceConsumerRetry(si.iname, seq, next, startTime)
 }
 
@@ -3295,16 +3301,9 @@ func (mset *stream) setStartingSequenceForSource(iName string, external *Externa
 	}
 }
 
-// Lock should be held.
-// This will do a reverse scan on startup or leader election
-// searching for the starting sequence number.
-// This can be slow in degenerative cases.
-// Lock should be held.
-func (mset *stream) startingSequenceForSources() {
-	if len(mset.cfg.Sources) == 0 {
-		return
-	}
-	// Always reset here.
+// lock should be held.
+// Resets the SourceInfo for all the sources
+func (mset *stream) resetSourceInfo() {
 	mset.sources = make(map[string]*sourceInfo)
 
 	for _, ssi := range mset.cfg.Sources {
@@ -3331,6 +3330,20 @@ func (mset *stream) startingSequenceForSources() {
 		}
 		mset.sources[ssi.iname] = si
 	}
+}
+
+// Lock should be held.
+// This will do a reverse scan on startup or leader election
+// searching for the starting sequence number.
+// This can be slow in degenerative cases.
+// Lock should be held.
+func (mset *stream) startingSequenceForSources() {
+	if len(mset.cfg.Sources) == 0 {
+		return
+	}
+
+	// Always reset here.
+	mset.resetSourceInfo()
 
 	var state StreamState
 	mset.store.FastState(&state)
@@ -3414,6 +3427,11 @@ func (mset *stream) setupSourceConsumers() error {
 		}
 	}
 
+	// If we are no longer the leader, give up
+	if !mset.isLeader() {
+		return nil
+	}
+
 	mset.startingSequenceForSources()
 
 	// Setup our consumers at the proper starting position.
@@ -3439,13 +3457,35 @@ func (mset *stream) subscribeToStream() error {
 	}
 	// Check if we need to setup mirroring.
 	if mset.cfg.Mirror != nil {
-		if err := mset.setupMirrorConsumer(); err != nil {
-			return err
+		// setup the initial mirror sourceInfo
+		mset.mirror = &sourceInfo{name: mset.cfg.Mirror.Name}
+		sfs := make([]string, len(mset.cfg.Mirror.SubjectTransforms))
+		trs := make([]*subjectTransform, len(mset.cfg.Mirror.SubjectTransforms))
+
+		for i, tr := range mset.cfg.Mirror.SubjectTransforms {
+			// will not fail as already checked before that the transform will work
+			subjectTransform, err := NewSubjectTransform(tr.Source, tr.Destination)
+			if err != nil {
+				mset.srv.Errorf("Unable to get transform for mirror consumer: %v", err)
+			}
+
+			sfs[i] = tr.Source
+			trs[i] = subjectTransform
 		}
+		mset.mirror.sfs = sfs
+		mset.mirror.trs = trs
+		// delay the actual mirror consumer creation for after a delay
+		mset.scheduleSetupMirrorConsumerRetryAsap()
 	} else if len(mset.cfg.Sources) > 0 {
-		if err := mset.setupSourceConsumers(); err != nil {
-			return err
-		}
+		// Setup the initial source infos for the sources
+		mset.resetSourceInfo()
+		// Delay the actual source consumer(s) creation(s) for after a delay
+
+		mset.sourcesConsumerSetup = time.AfterFunc(time.Duration(rand.Intn(int(10*time.Millisecond)))+10*time.Millisecond, func() {
+			mset.mu.Lock()
+			mset.setupSourceConsumers()
+			mset.mu.Unlock()
+		})
 	}
 	// Check for direct get access.
 	// We spin up followers for clustered streams in monitorStream().
@@ -3665,14 +3705,14 @@ func (mset *stream) setupStore(fsCfg *FileStoreConfig) error {
 			fsCfg.Cipher = s.getOpts().JetStreamCipher
 		}
 		oldprf := s.jsKeyGen(s.getOpts().JetStreamOldKey, mset.acc.Name)
-		cfg := *fsCfg
-		cfg.srv = s
-		fs, err := newFileStoreWithCreated(cfg, mset.cfg, mset.created, prf, oldprf)
+		fs, err := newFileStoreWithCreated(*fsCfg, mset.cfg, mset.created, prf, oldprf)
 		if err != nil {
 			mset.mu.Unlock()
 			return err
 		}
 		mset.store = fs
+		// Register our server.
+		fs.registerServer(s)
 	}
 	// This will fire the callback but we do not require the lock since md will be 0 here.
 	mset.store.RegisterStorageUpdates(mset.storeUpdates)
diff --git a/server/sublist.go b/server/sublist.go
index a78ff86c..6f12f891 100644
--- a/server/sublist.go
+++ b/server/sublist.go
@@ -540,7 +540,6 @@ func (s *Sublist) match(subject string, doLock bool) *SublistResult {
 	if doLock {
 		s.RLock()
 	}
-	cacheEnabled := s.cache != nil
 	r, ok := s.cache[subject]
 	if doLock {
 		s.RUnlock()
@@ -575,11 +574,7 @@ func (s *Sublist) match(subject string, doLock bool) *SublistResult {
 	var n int
 
 	if doLock {
-		if cacheEnabled {
-			s.Lock()
-		} else {
-			s.RLock()
-		}
+		s.Lock()
 	}
 
 	matchLevel(s.root, tokens, result)
@@ -587,20 +582,16 @@ func (s *Sublist) match(subject string, doLock bool) *SublistResult {
 	if len(result.psubs) == 0 && len(result.qsubs) == 0 {
 		result = emptyResult
 	}
-	if cacheEnabled {
+	if s.cache != nil {
 		s.cache[subject] = result
 		n = len(s.cache)
 	}
 	if doLock {
-		if cacheEnabled {
-			s.Unlock()
-		} else {
-			s.RUnlock()
-		}
+		s.Unlock()
 	}
 
 	// Reduce the cache count if we have exceeded our set maximum.
-	if cacheEnabled && n > slCacheMax && atomic.CompareAndSwapInt32(&s.ccSweep, 0, 1) {
+	if n > slCacheMax && atomic.CompareAndSwapInt32(&s.ccSweep, 0, 1) {
 		go s.reduceCacheCount()
 	}
 
diff --git a/server/websocket.go b/server/websocket.go
index 1afa3084..4eb93977 100644
--- a/server/websocket.go
+++ b/server/websocket.go
@@ -1050,10 +1050,6 @@ func (s *Server) wsConfigAuth(opts *WebsocketOpts) {
 }
 
 func (s *Server) startWebsocketServer() {
-	if s.isShuttingDown() {
-		return
-	}
-
 	sopts := s.getOpts()
 	o := &sopts.Websocket
 
@@ -1075,6 +1071,10 @@ func (s *Server) startWebsocketServer() {
 	// avoid the possibility of it being "intercepted".
 
 	s.mu.Lock()
+	if s.shutdown {
+		s.mu.Unlock()
+		return
+	}
 	// Do not check o.NoTLS here. If a TLS configuration is available, use it,
 	// regardless of NoTLS. If we don't have a TLS config, it means that the
 	// user has configured NoTLS because otherwise the server would have failed
@@ -1221,7 +1221,7 @@ func (s *Server) createWSClient(conn net.Conn, ws *websocket) *client {
 
 	s.mu.Lock()
 	if !s.isRunning() || s.ldm {
-		if s.isShuttingDown() {
+		if s.shutdown {
 			conn.Close()
 		}
 		s.mu.Unlock()


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 05e08204..a961ef9e 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -6096,75 +6096,6 @@ func TestFileStoreRemoveLastNoDoubleTombstones(t *testing.T) {
 	require_Equal(t, rbytes, emptyRecordLen)
 }
 
-func TestFileStoreFullStateMultiBlockPastWAL(t *testing.T) {
-	testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
-		fcfg.BlockSize = 100
-		scfg := StreamConfig{Name: "zzz", Subjects: []string{"*"}, Storage: FileStorage}
-
-		prf := func(context []byte) ([]byte, error) {
-			h := hmac.New(sha256.New, []byte("dlc22"))
-			if _, err := h.Write(context); err != nil {
-				return nil, err
-			}
-			return h.Sum(nil), nil
-		}
-		if fcfg.Cipher == NoCipher {
-			prf = nil
-		}
-
-		fs, err := newFileStoreWithCreated(fcfg, scfg, time.Now(), prf, nil)
-		require_NoError(t, err)
-		defer fs.Stop()
-
-		// This yields an internal record length of 50 bytes. So 2 msgs per blk.
-		msgLen := 19
-		msgA := bytes.Repeat([]byte("A"), msgLen)
-		msgZ := bytes.Repeat([]byte("Z"), msgLen)
-
-		// Store 2 msgs
-		fs.StoreMsg("A", nil, msgA)
-		fs.StoreMsg("B", nil, msgZ)
-		require_Equal(t, fs.numMsgBlocks(), 1)
-		fs.Stop()
-
-		// Grab the state from this stop.
-		sfile := filepath.Join(fcfg.StoreDir, msgDir, streamStreamStateFile)
-		buf, err := os.ReadFile(sfile)
-		require_NoError(t, err)
-
-		fs, err = newFileStoreWithCreated(fcfg, scfg, time.Now(), prf, nil)
-		require_NoError(t, err)
-		defer fs.Stop()
-
-		// Store 2 more msgs and delete 2 & 4, then another 2 msgs.
-		fs.StoreMsg("C", nil, msgA)
-		fs.StoreMsg("D", nil, msgZ)
-		fs.StoreMsg("E", nil, msgA)
-		fs.StoreMsg("F", nil, msgZ)
-		fs.StoreMsg("G", nil, msgA)
-		fs.StoreMsg("H", nil, msgZ)
-		require_Equal(t, fs.numMsgBlocks(), 4)
-		state := fs.State()
-		fs.Stop()
-
-		// Put back old stream state.
-		// This will test that we properly walk multiple blocks past where we snapshotted state.
-		fs.Stop()
-		err = os.WriteFile(sfile, buf, defaultFilePerms)
-		require_NoError(t, err)
-
-		fs, err = newFileStoreWithCreated(fcfg, scfg, time.Now(), prf, nil)
-		require_NoError(t, err)
-		defer fs.Stop()
-
-		if newState := fs.State(); !reflect.DeepEqual(state, newState) {
-			t.Fatalf("Restore state does not match:\n%+v\n%+v",
-				state, newState)
-		}
-		require_True(t, !state.FirstTime.IsZero())
-	})
-}
-
 ///////////////////////////////////////////////////////////////////////////
 // Benchmarks
 ///////////////////////////////////////////////////////////////////////////
diff --git a/server/jetstream_benchmark_test.go b/server/jetstream_benchmark_test.go
index 5d6ec571..466b98dc 100644
--- a/server/jetstream_benchmark_test.go
+++ b/server/jetstream_benchmark_test.go
@@ -29,12 +29,11 @@ import (
 func BenchmarkJetStreamConsume(b *testing.B) {
 
 	const (
-		verbose          = false
-		streamName       = "S"
-		subject          = "s"
-		seed             = 12345
-		publishTimeout   = 30 * time.Second
-		PublishBatchSize = 10000
+		verbose        = false
+		streamName     = "S"
+		subject        = "s"
+		seed           = 12345
+		publishTimeout = 30 * time.Second
 	)
 
 	runSyncPushConsumer := func(b *testing.B, js nats.JetStreamContext, streamName, subject string) (int, int, int) {
@@ -82,6 +81,7 @@ func BenchmarkJetStreamConsume(b *testing.B) {
 
 			uniqueConsumed++
 			bitset.set(index, true)
+			b.SetBytes(int64(len(msg.Data)))
 
 			if verbose && uniqueConsumed%1000 == 0 {
 				b.Logf("Consumed: %d/%d", bitset.count(), b.N)
@@ -127,6 +127,7 @@ func BenchmarkJetStreamConsume(b *testing.B) {
 
 			uniqueConsumed++
 			bitset.set(index, true)
+			b.SetBytes(int64(len(msg.Data)))
 
 			if uniqueConsumed == b.N {
 				msg.Sub.Unsubscribe()
@@ -222,6 +223,7 @@ func BenchmarkJetStreamConsume(b *testing.B) {
 
 				uniqueConsumed++
 				bitset.set(index, true)
+				b.SetBytes(int64(len(msg.Data)))
 
 				if uniqueConsumed == b.N {
 					msg.Sub.Unsubscribe()
@@ -305,9 +307,23 @@ func BenchmarkJetStreamConsume(b *testing.B) {
 							if verbose {
 								b.Logf("Setting up %d nodes", bc.clusterSize)
 							}
+							var (
+								connectURL string
+								cl         *cluster
+							)
+							if bc.clusterSize == 1 {
+								s := RunBasicJetStreamServer(b)
+								defer s.Shutdown()
+								connectURL = s.ClientURL()
+							} else {
+								cl = createJetStreamClusterExplicit(b, "BENCH_PUB", bc.clusterSize)
+								defer cl.shutdown()
+								cl.waitOnClusterReadyWithNumPeers(bc.clusterSize)
+								cl.waitOnLeader()
+								connectURL = cl.leader().ClientURL()
+							}
 
-							cl, _, shutdown, nc, js := startJSClusterAndConnect(b, bc.clusterSize)
-							defer shutdown()
+							nc, js := jsClientConnectURL(b, connectURL)
 							defer nc.Close()
 
 							if verbose {
@@ -322,38 +338,34 @@ func BenchmarkJetStreamConsume(b *testing.B) {
 								b.Fatalf("Error creating stream: %v", err)
 							}
 
-							// If replicated resource, connect to stream leader for lower variability
-							if bc.replicas > 1 {
-								connectURL := cl.streamLeader("$G", streamName).ClientURL()
+							// cluster_size > 1, connect to stream leader
+							if cl != nil {
+								connectURL = cl.streamLeader("$G", streamName).ClientURL()
 								nc.Close()
 								_, js = jsClientConnectURL(b, connectURL)
 							}
 
 							rng := rand.New(rand.NewSource(int64(seed)))
 							message := make([]byte, bc.messageSize)
-
-							// Publish b.N messages to the stream (in batches)
-							for i := 1; i <= b.N; i++ {
+							publishedCount := 0
+							for publishedCount < b.N {
 								rng.Read(message)
 								_, err := js.PublishAsync(subject, message)
 								if err != nil {
-									b.Fatalf("Failed to publish: %s", err)
-								}
-								// Limit outstanding published messages to PublishBatchSize
-								if i%PublishBatchSize == 0 || i == b.N {
-									select {
-									case <-js.PublishAsyncComplete():
-										if verbose {
-											b.Logf("Published %d/%d messages", i, b.N)
-										}
-									case <-time.After(publishTimeout):
-										b.Fatalf("Publish timed out")
-									}
+									continue
+								} else {
+									publishedCount++
 								}
 							}
 
-							// Set size of each operation, for throughput calculation
-							b.SetBytes(int64(bc.messageSize))
+							select {
+							case <-js.PublishAsyncComplete():
+								if verbose {
+									b.Logf("Published %d messages", b.N)
+								}
+							case <-time.After(publishTimeout):
+								b.Fatalf("Publish timed out")
+							}
 
 							// Discard time spent during setup
 							// Consumer may reset again further in
@@ -425,6 +437,7 @@ func BenchmarkJetStreamPublish(b *testing.B) {
 				errors++
 			} else {
 				published++
+				b.SetBytes(int64(messageSize))
 			}
 
 			if verbose && i%1000 == 0 {
@@ -441,54 +454,61 @@ func BenchmarkJetStreamPublish(b *testing.B) {
 		const publishCompleteMaxWait = 30 * time.Second
 		rng := rand.New(rand.NewSource(int64(seed)))
 		message := make([]byte, messageSize)
-
+		pending := make([]nats.PubAckFuture, 0, asyncWindow)
 		published, errors := 0, 0
 
 		b.ResetTimer()
 
-		for published < b.N {
-
-			// Normally publish a full batch (of size `asyncWindow`)
-			publishBatchSize := asyncWindow
-			// Unless fewer are left to complete the benchmark
-			if b.N-published < asyncWindow {
-				publishBatchSize = b.N - published
+		for i := 1; i <= b.N; i++ {
+			rng.Read(message) // TODO may skip this?
+			subject := subjects[rng.Intn(len(subjects))]
+			pubAckFuture, err := js.PublishAsync(subject, message)
+			if err != nil {
+				errors++
+				continue
 			}
-
-			pending := make([]nats.PubAckFuture, 0, publishBatchSize)
-
-			for i := 0; i < publishBatchSize; i++ {
-				rng.Read(message) // TODO may skip this?
-				subject := subjects[rng.Intn(len(subjects))]
-				pubAckFuture, err := js.PublishAsync(subject, message)
-				if err != nil {
-					errors++
-					continue
+			pending = append(pending, pubAckFuture)
+
+			// Regularly trim the list of pending
+			if i%asyncWindow == 0 {
+				newPending := make([]nats.PubAckFuture, 0, asyncWindow)
+				for _, pubAckFuture := range pending {
+					select {
+					case <-pubAckFuture.Ok():
+						published++
+						b.SetBytes(int64(messageSize))
+					case <-pubAckFuture.Err():
+						errors++
+					default:
+						// This pubAck is still pending, keep it
+						newPending = append(newPending, pubAckFuture)
+					}
 				}
-				pending = append(pending, pubAckFuture)
+				pending = newPending
 			}
 
-			// All in this batch published, wait for completed
-			select {
-			case <-js.PublishAsyncComplete():
-			case <-time.After(publishCompleteMaxWait):
-				b.Fatalf("Publish timed out")
+			if verbose && i%1000 == 0 {
+				b.Logf("Published %d/%d, %d errors", i, b.N, errors)
 			}
+		}
 
-			// Verify one by one if they were published successfully
-			for _, pubAckFuture := range pending {
-				select {
-				case <-pubAckFuture.Ok():
-					published++
-				case <-pubAckFuture.Err():
-					errors++
-				default:
-					b.Fatalf("PubAck is still pending after publish completed")
-				}
-			}
+		// All published, wait for completed
+		select {
+		case <-js.PublishAsyncComplete():
+		case <-time.After(publishCompleteMaxWait):
+			b.Fatalf("Publish timed out")
+		}
 
-			if verbose {
-				b.Logf("Published %d/%d", published, b.N)
+		// Clear whatever is left pending
+		for _, pubAckFuture := range pending {
+			select {
+			case <-pubAckFuture.Ok():
+				published++
+				b.SetBytes(int64(messageSize))
+			case <-pubAckFuture.Err():
+				errors++
+			default:
+				b.Fatalf("PubAck is still pending after publish completed")
 			}
 		}
 
@@ -549,6 +569,11 @@ func BenchmarkJetStreamPublish(b *testing.B) {
 					b.Run(
 						name,
 						func(b *testing.B) {
+							// Skip short runs, benchmark gets re-executed with a larger N
+							if b.N < bc.minMessages {
+								b.ResetTimer()
+								return
+							}
 
 							subjects := make([]string, bc.numSubjects)
 							for i := 0; i < bc.numSubjects; i++ {
@@ -562,9 +587,27 @@ func BenchmarkJetStreamPublish(b *testing.B) {
 							if verbose {
 								b.Logf("Setting up %d nodes", bc.clusterSize)
 							}
+							var (
+								connectURL string
+								cl         *cluster
+							)
+
+							if bc.clusterSize == 1 {
+								s := RunBasicJetStreamServer(b)
+								defer s.Shutdown()
+								connectURL = s.ClientURL()
+							} else {
+								cl = createJetStreamClusterExplicit(b, "BENCH_PUB", bc.clusterSize)
+								defer cl.shutdown()
+								cl.waitOnClusterReadyWithNumPeers(bc.clusterSize)
+								cl.waitOnLeader()
+								connectURL = cl.leader().ClientURL()
+							}
 
-							cl, _, shutdown, nc, _ := startJSClusterAndConnect(b, bc.clusterSize)
-							defer shutdown()
+							nc, err := nats.Connect(connectURL)
+							if err != nil {
+								b.Fatalf("Failed to create client: %v", err)
+							}
 							defer nc.Close()
 
 							jsOpts := []nats.JSOpt{
@@ -592,9 +635,9 @@ func BenchmarkJetStreamPublish(b *testing.B) {
 								b.Fatalf("Error creating stream: %v", err)
 							}
 
-							// If replicated resource, connect to stream leader for lower variability
-							if bc.replicas > 1 {
-								connectURL := cl.streamLeader("$G", streamName).ClientURL()
+							// cluster_size > 1, connect to stream leader
+							if cl != nil {
+								connectURL = cl.streamLeader("$G", streamName).ClientURL()
 								nc.Close()
 								nc, err = nats.Connect(connectURL)
 								if err != nil {
@@ -605,14 +648,13 @@ func BenchmarkJetStreamPublish(b *testing.B) {
 								if err != nil {
 									b.Fatalf("Unexpected error getting JetStream context for stream leader: %v", err)
 								}
+
 							}
 
 							if verbose {
 								b.Logf("Running %v publisher with message size: %dB", pc.pType, bc.messageSize)
 							}
 
-							b.SetBytes(int64(bc.messageSize))
-
 							// Benchmark starts here
 							b.ResetTimer()
 
@@ -705,6 +747,47 @@ func BenchmarkJetStreamInterestStreamWithLimit(b *testing.B) {
 		},
 	}
 
+	// Helper: Stand up in-process single node or cluster
+	setupCluster := func(b *testing.B, clusterSize int) (string, func()) {
+		var connectURL string
+		var shutdownFunc func()
+
+		if clusterSize == 1 {
+			s := RunBasicJetStreamServer(b)
+			shutdownFunc = s.Shutdown
+			connectURL = s.ClientURL()
+		} else {
+			cl := createJetStreamClusterExplicit(b, "BENCH_PUB", clusterSize)
+			shutdownFunc = cl.shutdown
+			cl.waitOnClusterReadyWithNumPeers(clusterSize)
+			cl.waitOnLeader()
+			connectURL = cl.leader().ClientURL()
+		}
+
+		return connectURL, shutdownFunc
+	}
+
+	// Helper: Create the stream
+	setupStream := func(b *testing.B, connectURL string, streamConfig *nats.StreamConfig) {
+		// Connect
+		nc, err := nats.Connect(connectURL)
+		if err != nil {
+			b.Fatalf("Failed to create client: %v", err)
+		}
+		defer nc.Close()
+
+		jsOpts := []nats.JSOpt{}
+
+		js, err := nc.JetStream(jsOpts...)
+		if err != nil {
+			b.Fatalf("Unexpected error getting JetStream context: %v", err)
+		}
+
+		if _, err := js.AddStream(streamConfig); err != nil {
+			b.Fatalf("Error creating stream: %v", err)
+		}
+	}
+
 	// Context shared by publishers routines
 	type PublishersContext = struct {
 		readyWg      sync.WaitGroup
@@ -790,6 +873,12 @@ func BenchmarkJetStreamInterestStreamWithLimit(b *testing.B) {
 								b.Run(
 									limitDescription,
 									func(b *testing.B) {
+										// Stop timer during setup
+										b.StopTimer()
+										b.ResetTimer()
+
+										// Set per-iteration bytes to calculate throughput (a.k.a. speed)
+										b.SetBytes(messageSize)
 
 										// Print benchmark parameters
 										if verbose {
@@ -803,9 +892,8 @@ func BenchmarkJetStreamInterestStreamWithLimit(b *testing.B) {
 										}
 
 										// Setup server or cluster
-										cl, ls, shutdown, nc, js := startJSClusterAndConnect(b, benchmarkCase.clusterSize)
-										defer shutdown()
-										defer nc.Close()
+										connectURL, shutdownFunc := setupCluster(b, benchmarkCase.clusterSize)
+										defer shutdownFunc()
 
 										// Common stream configuration
 										streamConfig := &nats.StreamConfig{
@@ -818,11 +906,8 @@ func BenchmarkJetStreamInterestStreamWithLimit(b *testing.B) {
 										}
 										// Configure stream limit
 										limitConfigFunc(streamConfig)
-
 										// Create stream
-										if _, err := js.AddStream(streamConfig); err != nil {
-											b.Fatalf("Error creating stream: %v", err)
-										}
+										setupStream(b, connectURL, streamConfig)
 
 										// Set up publishers shared context
 										var pubCtx PublishersContext
@@ -833,12 +918,6 @@ func BenchmarkJetStreamInterestStreamWithLimit(b *testing.B) {
 										pubCtx.lock.Lock()
 										pubCtx.messagesLeft = b.N
 
-										connectURL := ls.ClientURL()
-										// If replicated resource, connect to stream leader for lower variability
-										if benchmarkCase.replicas > 1 {
-											connectURL = cl.streamLeader("$G", "S").ClientURL()
-										}
-
 										// Spawn publishers routines, each with its own connection and JS context
 										for i := 0; i < numPublishers; i++ {
 											nc, err := nats.Connect(connectURL)
@@ -856,11 +935,8 @@ func BenchmarkJetStreamInterestStreamWithLimit(b *testing.B) {
 										// Wait for all publishers to be ready
 										pubCtx.readyWg.Wait()
 
-										// Set size of each operation, for throughput calculation
-										b.SetBytes(messageSize)
-
 										// Benchmark starts here
-										b.ResetTimer()
+										b.StartTimer()
 
 										// Unblock the publishers
 										pubCtx.lock.Unlock()
@@ -895,6 +971,7 @@ func BenchmarkJetStreamKV(b *testing.B) {
 		kvName    = "BUCKET"
 		keyPrefix = "K_"
 		seed      = 12345
+		minOps    = 1_000
 	)
 
 	runKVGet := func(b *testing.B, kv nats.KeyValue, keys []string) int {
@@ -905,12 +982,14 @@ func BenchmarkJetStreamKV(b *testing.B) {
 
 		for i := 1; i <= b.N; i++ {
 			key := keys[rng.Intn(len(keys))]
-			_, err := kv.Get(key)
+			kve, err := kv.Get(key)
 			if err != nil {
 				errors++
 				continue
 			}
 
+			b.SetBytes(int64(len(kve.Value())))
+
 			if verbose && i%1000 == 0 {
 				b.Logf("Completed %d/%d Get ops", i, b.N)
 			}
@@ -936,6 +1015,8 @@ func BenchmarkJetStreamKV(b *testing.B) {
 				continue
 			}
 
+			b.SetBytes(int64(valueSize))
+
 			if verbose && i%1000 == 0 {
 				b.Logf("Completed %d/%d Put ops", i, b.N)
 			}
@@ -968,6 +1049,8 @@ func BenchmarkJetStreamKV(b *testing.B) {
 				continue
 			}
 
+			b.SetBytes(int64(valueSize))
+
 			if verbose && i%1000 == 0 {
 				b.Logf("Completed %d/%d Update ops", i, b.N)
 			}
@@ -1021,6 +1104,11 @@ func BenchmarkJetStreamKV(b *testing.B) {
 					b.Run(
 						wName,
 						func(b *testing.B) {
+							// Skip short runs, benchmark gets re-executed with a larger N
+							if b.N < minOps {
+								b.ResetTimer()
+								return
+							}
 
 							if verbose {
 								b.Logf("Running %s workload %s with %d messages", wName, bName, b.N)
@@ -1029,6 +1117,24 @@ func BenchmarkJetStreamKV(b *testing.B) {
 							if verbose {
 								b.Logf("Setting up %d nodes", bc.clusterSize)
 							}
+							var (
+								connectURL string
+								cl         *cluster
+							)
+							if bc.clusterSize == 1 {
+								s := RunBasicJetStreamServer(b)
+								defer s.Shutdown()
+								connectURL = s.ClientURL()
+							} else {
+								cl = createJetStreamClusterExplicit(b, "BENCH_KV", bc.clusterSize)
+								defer cl.shutdown()
+								cl.waitOnClusterReadyWithNumPeers(bc.clusterSize)
+								cl.waitOnLeader()
+								connectURL = cl.leader().ClientURL()
+							}
+
+							nc, js := jsClientConnectURL(b, connectURL)
+							defer nc.Close()
 
 							// Pre-generate all keys
 							keys := make([]string, 0, bc.numKeys)
@@ -1037,11 +1143,6 @@ func BenchmarkJetStreamKV(b *testing.B) {
 								keys = append(keys, key)
 							}
 
-							// Setup server or cluster
-							cl, _, shutdown, nc, js := startJSClusterAndConnect(b, bc.clusterSize)
-							defer shutdown()
-							defer nc.Close()
-
 							// Create bucket
 							if verbose {
 								b.Logf("Creating KV %s with R=%d", kvName, bc.replicas)
@@ -1066,22 +1167,18 @@ func BenchmarkJetStreamKV(b *testing.B) {
 								}
 							}
 
-							// If replicated resource, connect to stream leader for lower variability
-							if bc.replicas > 1 {
+							// if cluster_size > 1, connect to stream leader of bucket
+							if cl != nil {
 								nc.Close()
-								connectURL := cl.streamLeader("$G", fmt.Sprintf("KV_%s", kvName)).ClientURL()
-								nc, js = jsClientConnectURL(b, connectURL)
-								defer nc.Close()
-							}
 
+								connectURL = cl.streamLeader("$G", fmt.Sprintf("KV_%s", kvName)).ClientURL()
+								_, js = jsClientConnectURL(b, connectURL)
+							}
 							kv, err = js.KeyValue(kv.Bucket())
 							if err != nil {
 								b.Fatalf("Error binding to KV: %v", err)
 							}
 
-							// Set size of each operation, for throughput calculation
-							b.SetBytes(int64(bc.valueSize))
-
 							// Discard time spent during setup
 							// May reset again further in
 							b.ResetTimer()
@@ -1188,6 +1285,7 @@ func BenchmarkJetStreamObjStore(b *testing.B) {
 		minObjSz int
 		maxObjSz int
 	}{
+		// TODO remove duplicates and fix comments
 		{nats.MemoryStorage, 100, 1024, 102400},     // mem storage, 100 objects sized (1KB-100KB)
 		{nats.MemoryStorage, 100, 102400, 1048576},  // mem storage, 100 objects sized (100KB-1MB)
 		{nats.MemoryStorage, 1000, 10240, 102400},   // mem storage, 1k objects of various size (10KB - 100KB)
@@ -1196,6 +1294,7 @@ func BenchmarkJetStreamObjStore(b *testing.B) {
 		{nats.FileStorage, 100, 102400, 1048576},    // file storage, 100 objects sized (100KB-1MB)
 		{nats.FileStorage, 100, 1048576, 10485760},  // file storage, 100 objects sized (1MB-10MB)
 		{nats.FileStorage, 10, 10485760, 104857600}, // file storage, 10 objects sized (10MB-100MB)
+
 	}
 
 	var (
@@ -1228,10 +1327,23 @@ func BenchmarkJetStreamObjStore(b *testing.B) {
 										if verbose {
 											b.Logf("Setting up %d nodes", replicas)
 										}
-
-										// Setup server or cluster
-										cl, _, shutdown, nc, js := startJSClusterAndConnect(b, clusterSize)
-										defer shutdown()
+										var (
+											connectURL string
+											cl         *cluster
+										)
+										if clusterSize == 1 {
+											s := RunBasicJetStreamServer(b)
+											defer s.Shutdown()
+											connectURL = s.ClientURL()
+										} else {
+											cl = createJetStreamClusterExplicit(b, "BENCH_OBJ_STORE", clusterSize)
+											defer cl.shutdown()
+											cl.waitOnClusterReadyWithNumPeers(replicas)
+											cl.waitOnLeader()
+											// connect to leader and not replicas
+											connectURL = cl.leader().ClientURL()
+										}
+										nc, js := jsClientConnectURL(b, connectURL)
 										defer nc.Close()
 
 										// Initialize object store
@@ -1248,10 +1360,10 @@ func BenchmarkJetStreamObjStore(b *testing.B) {
 											b.Fatalf("Error creating ObjectStore: %v", err)
 										}
 
-										// If replicated resource, connect to stream leader for lower variability
-										if clusterSize > 1 {
+										// if cluster_size > 1, connect to stream leader
+										if cl != nil {
 											nc.Close()
-											connectURL := cl.streamLeader("$G", fmt.Sprintf("OBJ_%s", objStoreName)).ClientURL()
+											connectURL = cl.streamLeader("$G", fmt.Sprintf("OBJ_%s", objStoreName)).ClientURL()
 											nc, js := jsClientConnectURL(b, connectURL)
 											defer nc.Close()
 											objStore, err = js.ObjectStore(objStoreName)
@@ -1289,41 +1401,9 @@ func BenchmarkJetStreamObjStore(b *testing.B) {
 							}
 						},
 					)
+
 				}
 			},
 		)
 	}
 }
-
-// Helper function to stand up a JS-enabled single server or cluster
-func startJSClusterAndConnect(b *testing.B, clusterSize int) (c *cluster, s *Server, shutdown func(), nc *nats.Conn, js nats.JetStreamContext) {
-	b.Helper()
-	var err error
-
-	if clusterSize == 1 {
-		s = RunBasicJetStreamServer(b)
-		shutdown = func() {
-			s.Shutdown()
-		}
-	} else {
-		c = createJetStreamClusterExplicit(b, "BENCH_PUB", clusterSize)
-		c.waitOnClusterReadyWithNumPeers(clusterSize)
-		c.waitOnLeader()
-		s = c.leader()
-		shutdown = func() {
-			c.shutdown()
-		}
-	}
-
-	nc, err = nats.Connect(s.ClientURL())
-	if err != nil {
-		b.Fatalf("failed to connect: %s", err)
-	}
-
-	js, err = nc.JetStream()
-	if err != nil {
-		b.Fatalf("failed to init jetstream: %s", err)
-	}
-
-	return c, s, shutdown, nc, js
-}
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index 095740e2..ad9b4978 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -3470,7 +3470,9 @@ func TestJetStreamClusterNoR1AssetsDuringLameDuck(t *testing.T) {
 	}()
 	defer close(qch)
 
-	gacc := s.GlobalAccount()
+	s.mu.RLock()
+	gacc := s.gacc
+	s.mu.RUnlock()
 	if gacc == nil {
 		t.Fatalf("No global account")
 	}
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 4c76f100..db2439ac 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -16617,6 +16617,8 @@ func TestJetStreamWorkQueueSourceRestart(t *testing.T) {
 	sub, err := js.PullSubscribe("foo", "dur", nats.BindStream("TEST"))
 	require_NoError(t, err)
 
+	time.Sleep(100 * time.Millisecond)
+
 	ci, err := js.ConsumerInfo("TEST", "dur")
 	require_NoError(t, err)
 	require_True(t, ci.NumPending == uint64(sent))
diff --git a/server/norace_test.go b/server/norace_test.go
index 65f32ebc..a6a8a2e6 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -1874,7 +1874,6 @@ func TestNoRaceJetStreamSuperClusterSources(t *testing.T) {
 			msg := fmt.Sprintf("R-MSG-%d", i+1)
 			for _, sname := range []string{"foo", "bar", "baz"} {
 				m := nats.NewMsg(sname)
-				m.Header.Set(nats.MsgIdHdr, sname+"-"+msg)
 				m.Data = []byte(msg)
 				if _, err := js.PublishMsg(m); err != nil {
 					t.Errorf("Unexpected publish error: %v", err)
@@ -1891,7 +1890,7 @@ func TestNoRaceJetStreamSuperClusterSources(t *testing.T) {
 	sc.clusterForName("C3").waitOnStreamLeader("$G", "MS2")
 	<-doneCh
 
-	checkFor(t, 15*time.Second, 100*time.Millisecond, func() error {
+	checkFor(t, 15*time.Second, time.Second, func() error {
 		si, err := js2.StreamInfo("MS2")
 		if err != nil {
 			return err
@@ -4227,7 +4226,7 @@ func TestNoRaceJetStreamStreamInfoSubjectDetailsLimits(t *testing.T) {
 		  default: {
 			jetstream: true
 			users: [ {user: me, password: pwd} ]
-			limits { max_payload: 512 }
+			limits { max_payload: 256 }
 		  }
 		}
 	`, t.TempDir())))
@@ -4244,11 +4243,11 @@ func TestNoRaceJetStreamStreamInfoSubjectDetailsLimits(t *testing.T) {
 	// Make sure to flush so we process the 2nd INFO.
 	nc.Flush()
 
-	// Make sure we cannot send larger than 512 bytes.
+	// Make sure we cannot send larger than 256 bytes.
 	// But we can receive larger.
 	sub, err := nc.SubscribeSync("foo")
 	require_NoError(t, err)
-	err = nc.Publish("foo", []byte(strings.Repeat("A", 600)))
+	err = nc.Publish("foo", []byte(strings.Repeat("A", 300)))
 	require_Error(t, err, nats.ErrMaxPayload)
 	sub.Unsubscribe()
 
diff --git a/server/routes_test.go b/server/routes_test.go
index 00fc4cc6..8385a6ad 100644
--- a/server/routes_test.go
+++ b/server/routes_test.go
@@ -2529,243 +2529,6 @@ func TestRoutePerAccountConnectRace(t *testing.T) {
 	}
 }
 
-func TestRoutePerAccountGossipWorks(t *testing.T) {
-	tmplA := `
-		port: -1
-		server_name: "A"
-		accounts {
-			A { users: [{user: "A", password: "pwd"}] }
-			B { users: [{user: "B", password: "pwd"}] }
-		}
-		cluster {
-			port: %d
-			name: "local"
-			accounts: ["A"]
-			%s
-		}
-	`
-	conf1 := createConfFile(t, []byte(fmt.Sprintf(tmplA, -1, _EMPTY_)))
-	s1, o1 := RunServerWithConfig(conf1)
-	defer s1.Shutdown()
-
-	tmplBC := `
-		port: -1
-		server_name: "%s"
-		accounts {
-			A { users: [{user: "A", password: "pwd"}] }
-			B { users: [{user: "B", password: "pwd"}] }
-		}
-		cluster {
-			port: -1
-			name: "local"
-			%s
-			accounts: ["A"]
-		}
-	`
-	conf2 := createConfFile(t, []byte(fmt.Sprintf(tmplBC, "B",
-		fmt.Sprintf("routes: [\"nats://127.0.0.1:%d\"]", o1.Cluster.Port))))
-	s2, _ := RunServerWithConfig(conf2)
-	defer s2.Shutdown()
-
-	checkClusterFormed(t, s1, s2)
-
-	// Now connect s3 to s1 and make sure that s2 connects properly to s3.
-	conf3 := createConfFile(t, []byte(fmt.Sprintf(tmplBC, "C",
-		fmt.Sprintf("routes: [\"nats://127.0.0.1:%d\"]", o1.Cluster.Port))))
-	s3, _ := RunServerWithConfig(conf3)
-	defer s3.Shutdown()
-
-	checkClusterFormed(t, s1, s2, s3)
-
-	s3.Shutdown()
-	s2.Shutdown()
-	s1.Shutdown()
-
-	// Slightly different version where s2 is connecting to s1, while s1
-	// connects to s3 (and s3 does not solicit connections).
-
-	conf1 = createConfFile(t, []byte(fmt.Sprintf(tmplA, -1, _EMPTY_)))
-	s1, o1 = RunServerWithConfig(conf1)
-	defer s1.Shutdown()
-
-	conf2 = createConfFile(t, []byte(fmt.Sprintf(tmplBC, "B",
-		fmt.Sprintf("routes: [\"nats://127.0.0.1:%d\"]", o1.Cluster.Port))))
-	s2, _ = RunServerWithConfig(conf2)
-	defer s2.Shutdown()
-
-	checkClusterFormed(t, s1, s2)
-
-	// Start s3 first that will simply accept connections
-	conf3 = createConfFile(t, []byte(fmt.Sprintf(tmplBC, "C", _EMPTY_)))
-	s3, o3 := RunServerWithConfig(conf3)
-	defer s3.Shutdown()
-
-	// Now config reload s1 so that it points to s3.
-	reloadUpdateConfig(t, s1, conf1,
-		fmt.Sprintf(tmplA, o1.Cluster.Port, fmt.Sprintf("routes: [\"nats://127.0.0.1:%d\"]", o3.Cluster.Port)))
-
-	checkClusterFormed(t, s1, s2, s3)
-}
-
-func TestRoutePerAccountGossipWorksWithOldServerNotSeed(t *testing.T) {
-	tmplA := `
-		port: -1
-		server_name: "A"
-		accounts {
-			A { users: [{user: "A", password: "pwd"}] }
-			B { users: [{user: "B", password: "pwd"}] }
-		}
-		cluster {
-			port: %d
-			name: "local"
-			accounts: ["A"]
-		}
-	`
-	conf1 := createConfFile(t, []byte(fmt.Sprintf(tmplA, -1)))
-	s1, o1 := RunServerWithConfig(conf1)
-	defer s1.Shutdown()
-
-	// Here, server "B" will have no pooling/accounts.
-	tmplB := `
-		port: -1
-		server_name: "B"
-		accounts {
-			A { users: [{user: "A", password: "pwd"}] }
-			B { users: [{user: "B", password: "pwd"}] }
-		}
-		cluster {
-			port: -1
-			name: "local"
-			routes: ["nats://127.0.0.1:%d"]
-			pool_size: -1
-		}
-	`
-	conf2 := createConfFile(t, []byte(fmt.Sprintf(tmplB, o1.Cluster.Port)))
-	s2, _ := RunServerWithConfig(conf2)
-	defer s2.Shutdown()
-
-	checkClusterFormed(t, s1, s2)
-
-	l := &captureErrorLogger{errCh: make(chan string, 100)}
-	s2.SetLogger(l, false, false)
-
-	// Now connect s3 to s1. Server s1 should not gossip to s2 information
-	// about pinned-account routes or extra routes from the pool.
-	tmplC := `
-		port: -1
-		server_name: "C"
-		accounts {
-			A { users: [{user: "A", password: "pwd"}] }
-			B { users: [{user: "B", password: "pwd"}] }
-		}
-		cluster {
-			port: -1
-			name: "local"
-			routes: ["nats://127.0.0.1:%d"]
-			accounts: ["A"]
-		}
-	`
-	conf3 := createConfFile(t, []byte(fmt.Sprintf(tmplC, o1.Cluster.Port)))
-	s3, _ := RunServerWithConfig(conf3)
-	defer s3.Shutdown()
-
-	checkClusterFormed(t, s1, s2, s3)
-
-	// We should not have had s2 try to create dedicated routes for "A" or "$SYS"
-	tm := time.NewTimer(time.Second)
-	defer tm.Stop()
-	for {
-		select {
-		case err := <-l.errCh:
-			if strings.Contains(err, "dedicated route") {
-				t.Fatalf("Server s2 should not have tried to create a dedicated route: %s", err)
-			}
-		case <-tm.C:
-			return
-		}
-	}
-}
-
-func TestRoutePerAccountGossipWorksWithOldServerSeed(t *testing.T) {
-	tmplA := `
-		port: -1
-		server_name: "A"
-		accounts {
-			A { users: [{user: "A", password: "pwd"}] }
-			B { users: [{user: "B", password: "pwd"}] }
-		}
-		cluster {
-			port: %d
-			name: "local"
-			pool_size: %d
-			%s
-		}
-	`
-	// Start with s1 being an "old" server, which does not support pooling/pinned-accounts.
-	conf1 := createConfFile(t, []byte(fmt.Sprintf(tmplA, -1, -1, _EMPTY_)))
-	s1, o1 := RunServerWithConfig(conf1)
-	defer s1.Shutdown()
-
-	tmplBC := `
-		port: -1
-		server_name: "%s"
-		accounts {
-			A { users: [{user: "A", password: "pwd"}] }
-			B { users: [{user: "B", password: "pwd"}] }
-		}
-		cluster {
-			port: -1
-			name: "local"
-			pool_size: 3
-			routes: ["nats://127.0.0.1:%d"]
-			accounts: ["A"]
-		}
-	`
-	conf2 := createConfFile(t, []byte(fmt.Sprintf(tmplBC, "B", o1.Cluster.Port)))
-	s2, _ := RunServerWithConfig(conf2)
-	defer s2.Shutdown()
-
-	checkClusterFormed(t, s1, s2)
-
-	// Now connect s3 to s1 and make sure that s2 connects properly to s3.
-	conf3 := createConfFile(t, []byte(fmt.Sprintf(tmplBC, "C", o1.Cluster.Port)))
-	s3, _ := RunServerWithConfig(conf3)
-	defer s3.Shutdown()
-
-	checkClusterFormed(t, s1, s2, s3)
-
-	checkRoutes := func(s *Server, expected int) {
-		t.Helper()
-		checkFor(t, 2*time.Second, 50*time.Millisecond, func() error {
-			if nr := s.NumRoutes(); nr != expected {
-				return fmt.Errorf("Server %q should have %v routes, got %v", s.Name(), expected, nr)
-			}
-			return nil
-		})
-	}
-	// Since s1 has no pooling/pinned-accounts, there should be only 2 routes,
-	// one to s2 and one to s3.
-	checkRoutes(s1, 2)
-	// s2 and s3 should have 1 route to s1 and 3(pool)+"A"+"$SYS" == 6
-	checkRoutes(s2, 6)
-	checkRoutes(s3, 6)
-
-	s1.Shutdown()
-
-	// The server s1 will now support pooling and accounts pinning.
-	// Restart the server s1 with the same cluster port otherwise
-	// s2/s3 would not be able to reconnect.
-	conf1 = createConfFile(t, []byte(fmt.Sprintf(tmplA, o1.Cluster.Port, 3, "accounts: [\"A\"]")))
-	s1, _ = RunServerWithConfig(conf1)
-	defer s1.Shutdown()
-	// Make sure reconnect occurs. We should now have 5 routes.
-	checkClusterFormed(t, s1, s2, s3)
-	// Now all servers should have 3+2 to each other, so 10 total.
-	checkRoutes(s1, 10)
-	checkRoutes(s2, 10)
-	checkRoutes(s3, 10)
-}
-
 func TestRoutePoolPerAccountSubUnsubProtoParsing(t *testing.T) {
 	for _, test := range []struct {
 		name  string
@@ -3014,7 +2777,9 @@ func TestRoutePoolAndPerAccountWithServiceLatencyNoDataRace(t *testing.T) {
 			defer nc.Close()
 
 			// The service listener.
+			// serviceTime := 25 * time.Millisecond
 			natsSub(t, nc, "req.echo", func(msg *nats.Msg) {
+				// time.Sleep(serviceTime)
 				msg.Respond(msg.Data)
 			})
 
@@ -3167,11 +2932,9 @@ func TestRoutePoolAndPerAccountOperatorMode(t *testing.T) {
 			return nil
 		})
 	}
-	// Route for accounts "apub" and "spub" should be a dedicated route
+	// Route for account "apub" should be a dedicated route
 	checkRoute(s1, apub, true)
 	checkRoute(s2, apub, true)
-	checkRoute(s1, spub, true)
-	checkRoute(s2, spub, true)
 	// Route for account "bpub" should not
 	checkRoute(s1, bpub, false)
 	checkRoute(s2, bpub, false)
@@ -3198,12 +2961,9 @@ func TestRoutePoolAndPerAccountOperatorMode(t *testing.T) {
 	// before doing the config reload on s2.
 	checkRoute(s1, bpub, true)
 	checkRoute(s2, bpub, true)
-	// Account "apub" should still have its dedicated route
+	// Account "aoub" should still have its dedicated route
 	checkRoute(s1, apub, true)
 	checkRoute(s2, apub, true)
-	// So the system account
-	checkRoute(s1, spub, true)
-	checkRoute(s2, spub, true)
 	// Let's complete the config reload on srvb
 	reloadUpdateConfig(t, s2, conf2, fmt.Sprintf(tmpl, "B", fmt.Sprintf("routes: [\"nats://127.0.0.1:%d\"]", o1.Cluster.Port),
 		fmt.Sprintf(",\"%s\"", bpub)))
diff --git a/server/server_test.go b/server/server_test.go
index 3defd412..4030aafa 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -773,7 +773,10 @@ func TestLameDuckMode(t *testing.T) {
 
 	// Check that if there is no client, server is shutdown
 	srvA.lameDuckMode()
-	if !srvA.isShuttingDown() {
+	srvA.mu.Lock()
+	shutdown := srvA.shutdown
+	srvA.mu.Unlock()
+	if !shutdown {
 		t.Fatalf("Server should have shutdown")
 	}