Repository: nats-io/nats-server (go)

Issue:
Flood of STATSZ system events starting nodes in a large cluster
### Observed behavior

The number of `STATSZ` system event messages sent by a newly-started (or restarted) server appears to increase quickly with the total size of the (super)cluster. When the number of nodes grows large, restarting too many at once can trigger a massive flood of system events that can overwhelm capacity and cause an outage across the entire system.

On a production supercluster of ~900 total nodes and a single-cluster size of 100 nodes, restarting a single server generates ~3 GB of total data and ~150k messages (as measured by `out_bytes` and `out_msgs` fields on the `/varz` endpoint) on startup.

Based on my reading of the code, `STATSZ` events are published periodically by a timer (waiting 250ms for the first event, then 500ms, etc until a max interval of 30s)

`STATSZ` events are _also_ published:
- When establishing an outbound gateway connection (`createGateway`)
- When we receive an update from any new server in the supercluster (`processNewServer`)

It's the latter that causes the total number of `STATSZ` messages to increase quickly with the number of nodes, since each update is sent to every other connected client.

### Expected behavior

A much smaller number of `STATSZ` updates when starting/restarting a server that does not increase as quickly with the total number of nodes in the cluster. A rate-limit of 1 update per second for `STATSZ` messages sent to each client subscription would prevent system-event floods while still allowing timely updates to be published across the cluster.

### Server and client version

I observed this issue in production with v2.7.4, but I can also reproduce this issue in an isolated test using the latest nats-server version.

### Host environment

_No response_

### Steps to reproduce

The following test case demonstrates the issue. In a cluster of `totalServers` servers, we would expect the total number of messages to be close to `totalServers`<sup>2</sup>, or a statsz update sent from each server to every other server. However, the result is closer to `totalServers`<sup>3</sup>.

```go
package server

import (
	"testing"
	"time"
)

func TestClusterSetupMsgs(t *testing.T) {
	numServers := 50
	c := createClusterEx(t, true, 100*time.Millisecond, true, "cluster", numServers)
	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)
	}
}
```

Sample output, with `numServers` set to 10, 50, 70:
```
Total outMsgs is 759, expected < 100
Total outMsgs is 84573, expected < 2500
Total outMsgs is 226932, expected < 4900
```



Pull request description:
Limit statsz updates
This PR adds a filter to `sendStatsz` that limits statsz updates to the current heartbeat interval (max once per second), adding a `time.Time` field to track the time the last statsz update was sent. This limit should reduce overall `STATSZ` system event load in large clusters while still allowing initial statsz update to quickly reach newly-discovered nodes.

Fixes #5469.

Signed-off-by: Will Jordan <will.jordan@gmail.com>


Patch:
diff --git a/server/consumer.go b/server/consumer.go
index ce32df2d..8b5d17ad 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -2232,22 +2232,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:
 	}
 }
 
@@ -2834,6 +2830,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()
diff --git a/server/events.go b/server/events.go
index 9c214cc6..24bc0f06 100644
--- a/server/events.go
+++ b/server/events.go
@@ -98,6 +98,8 @@ const (
 // FIXME(dlc) - make configurable.
 var eventsHBInterval = 30 * time.Second
 
+var statszRateLimit = 1 * time.Second
+
 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 +136,7 @@ type internal struct {
 	shash          string
 	inboxPre       string
 	remoteStatsSub *subscription
+	lastStatsz     time.Time
 }
 
 // ServerStatsMsg is sent periodically with stats updates.
@@ -808,6 +811,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 +956,35 @@ func (s *Server) sendStatsz(subj string) {
 	s.sendInternalMsg(subj, _EMPTY_, &m.Server, &m)
 }
 
+// Limit updates to the heartbeat interval, max one second.
+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 +1002,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 +1911,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 99fe6ad9..0a63fdbb 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -2344,7 +2344,8 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 	// If we do not think we should do a linear scan check how many fss we
 	// would need to scan vs the full range of the linear walk. Optimize for
 	// 25th quantile of a match in a linear walk. Filter should be a wildcard.
-	if !doLinearScan && wc {
+	// We should consult fss if our cache is not loaded and we only have fss loaded.
+	if !doLinearScan && wc && mb.cacheAlreadyLoaded() {
 		doLinearScan = len(mb.fss)*4 > int(lseq-fseq)
 	}
 
@@ -6341,6 +6342,11 @@ func (fs *fileStore) LoadNextMsg(filter string, wc bool, start uint64, sm *Store
 	if start <= fs.state.FirstSeq {
 		var ss SimpleState
 		fs.numFilteredPending(filter, &ss)
+		// Nothing available.
+		if ss.Msgs == 0 {
+			return nil, fs.state.LastSeq, ErrStoreEOF
+		}
+		// We can skip ahead.
 		if ss.First > start {
 			start = ss.First
 		}
@@ -6356,8 +6362,28 @@ func (fs *fileStore) LoadNextMsg(filter string, wc bool, start uint64, sm *Store
 				return sm, sm.seq, nil
 			} else if err != ErrStoreMsgNotFound {
 				return nil, 0, err
-			} else if expireOk {
-				mb.tryForceExpireCache()
+			} else {
+				// Nothing found in this block. We missed, if first block (bi) check psim.
+				// Similar to above if start <= first seq.
+				// TODO(dlc) - For v2 track these by filter subject since they will represent filtered consumers.
+				if i == bi {
+					var ss SimpleState
+					fs.numFilteredPending(filter, &ss)
+					// Nothing available.
+					if ss.Msgs == 0 {
+						return nil, fs.state.LastSeq, ErrStoreEOF
+					}
+					// See if we can jump ahead here.
+					// Right now we can only spin on first, so if we have interior sparseness need to favor checking per block fss if loaded.
+					// For v2 will track all blocks that have matches for psim.
+					if nbi, _ := fs.selectMsgBlockWithIndex(ss.First); nbi > i {
+						i = nbi - 1 // For the iterator condition i++
+					}
+				}
+				// Check is we can expire.
+				if expireOk {
+					mb.tryForceExpireCache()
+				}
 			}
 		}
 	}
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 14ba811d..f95758be 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -2496,7 +2496,8 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 			}
 
 		case <-cistc:
-			mset.checkInterestState()
+			// We may be adjusting some things with consumers so do this in its own go routine.
+			go mset.checkInterestState()
 
 		case <-datc:
 			if mset == nil || isRecovering {
diff --git a/server/memstore.go b/server/memstore.go
index 61cd8094..4d9b184b 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -465,14 +465,28 @@ func (ms *memStore) filteredStateLocked(sseq uint64, filter string, lastPerSubje
 	}
 	if toScan < toExclude {
 		ss.Msgs, ss.First = 0, 0
-		for seq := first; seq <= last; seq++ {
-			if sm, ok := ms.msgs[seq]; ok && !seen[sm.subj] && isMatch(sm.subj) {
-				ss.Msgs++
-				if ss.First == 0 {
-					ss.First = seq
+
+		update := func(sm *StoreMsg) {
+			ss.Msgs++
+			if ss.First == 0 {
+				ss.First = sm.seq
+			}
+			if seen != nil {
+				seen[sm.subj] = true
+			}
+		}
+		// Check if easier to just scan msgs vs the sequence range.
+		// This can happen with lots of interior deletes.
+		if last-first > uint64(len(ms.msgs)) {
+			for _, sm := range ms.msgs {
+				if sm.seq >= first && sm.seq <= last && !seen[sm.subj] && isMatch(sm.subj) {
+					update(sm)
 				}
-				if seen != nil {
-					seen[sm.subj] = true
+			}
+		} else {
+			for seq := first; seq <= last; seq++ {
+				if sm, ok := ms.msgs[seq]; ok && !seen[sm.subj] && isMatch(sm.subj) {
+					update(sm)
 				}
 			}
 		}
@@ -482,17 +496,29 @@ func (ms *memStore) filteredStateLocked(sseq uint64, filter string, lastPerSubje
 		var adjust uint64
 		var tss *SimpleState
 
-		for seq := ms.state.FirstSeq; seq < first; seq++ {
-			if sm, ok := ms.msgs[seq]; ok && !seen[sm.subj] && isMatch(sm.subj) {
-				if lastPerSubject {
-					tss, _ = ms.fss.Find(stringToBytes(sm.subj))
-				}
-				// If we are last per subject, make sure to only adjust if all messages are before our first.
-				if tss == nil || tss.Last < first {
-					adjust++
+		update := func(sm *StoreMsg) {
+			if lastPerSubject {
+				tss, _ = ms.fss.Find(stringToBytes(sm.subj))
+			}
+			// If we are last per subject, make sure to only adjust if all messages are before our first.
+			if tss == nil || tss.Last < first {
+				adjust++
+			}
+			if seen != nil {
+				seen[sm.subj] = true
+			}
+		}
+		// Check if easier to just scan msgs vs the sequence range.
+		if first-ms.state.FirstSeq > uint64(len(ms.msgs)) {
+			for _, sm := range ms.msgs {
+				if sm.seq < first && !seen[sm.subj] && isMatch(sm.subj) {
+					update(sm)
 				}
-				if seen != nil {
-					seen[sm.subj] = true
+			}
+		} else {
+			for seq := ms.state.FirstSeq; seq < first; seq++ {
+				if sm, ok := ms.msgs[seq]; ok && !seen[sm.subj] && isMatch(sm.subj) {
+					update(sm)
 				}
 			}
 		}
@@ -507,10 +533,27 @@ func (ms *memStore) filteredStateLocked(sseq uint64, filter string, lastPerSubje
 		}
 		ss.Msgs -= adjust
 		if needScanFirst {
-			for seq := first; seq < last; seq++ {
-				if sm, ok := ms.msgs[seq]; ok && isMatch(sm.subj) {
-					ss.First = seq
-					break
+			// Check if easier to just scan msgs vs the sequence range.
+			// Since we will need to scan all of the msgs vs below where we break on the first match,
+			// we will only do so if a few orders of magnitude lower.
+			if last-first > 100*uint64(len(ms.msgs)) {
+				low := ms.state.LastSeq
+				for _, sm := range ms.msgs {
+					if sm.seq >= first && sm.seq < last && isMatch(sm.subj) {
+						if sm.seq < low {
+							low = sm.seq
+						}
+					}
+				}
+				if low < ms.state.LastSeq {
+					ss.First = low
+				}
+			} else {
+				for seq := first; seq < last; seq++ {
+					if sm, ok := ms.msgs[seq]; ok && isMatch(sm.subj) {
+						ss.First = seq
+						break
+					}
 				}
 			}
 		}
diff --git a/server/raft.go b/server/raft.go
index 8d12afdc..82bbf9f0 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -3159,15 +3159,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)
 		}
 	}
diff --git a/server/reload.go b/server/reload.go
index b5951cbc..63e9367a 100644
--- a/server/reload.go
+++ b/server/reload.go
@@ -995,9 +995,11 @@ func (s *Server) Reload() error {
 	return s.ReloadOptions(newOpts)
 }
 
-// ReloadOptions applies any supported options from the provided Option
+// ReloadOptions applies any supported options from the provided Options
 // type. This returns an error if an option which doesn't support
 // hot-swapping was changed.
+// The provided Options type should not be re-used afterwards.
+// Either use Options.Clone() to pass a copy, or make a new one.
 func (s *Server) ReloadOptions(newOpts *Options) error {
 	s.reloadMu.Lock()
 	defer s.reloadMu.Unlock()
diff --git a/server/server.go b/server/server.go
index 33cd155e..10718c10 100644
--- a/server/server.go
+++ b/server/server.go
@@ -643,6 +643,8 @@ func New(opts *Options) *Server {
 
 // NewServer will setup a new server struct after parsing the options.
 // Could return an error if options can not be validated.
+// The provided Options type should not be re-used afterwards.
+// Either use Options.Clone() to pass a copy, or make a new one.
 func NewServer(opts *Options) (*Server, error) {
 	setBaselineOptions(opts)
 
diff --git a/server/stream.go b/server/stream.go
index 5f7066cb..de378651 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -5500,9 +5500,8 @@ func (mset *stream) checkInterestState() {
 
 	var zeroAcks []*consumer
 	var lowAckFloor uint64 = math.MaxUint64
-	consumers := mset.getConsumers()
 
-	for _, o := range consumers {
+	for _, o := range mset.getConsumers() {
 		o.checkStateForInterestStream()
 
 		o.mu.Lock()
@@ -5541,39 +5540,45 @@ func (mset *stream) checkInterestState() {
 		return
 	}
 
-	// Hold stream write lock in case we need to purge.
-	mset.mu.Lock()
-	defer mset.mu.Unlock()
-
 	// Capture our current state.
+	// ok to do so without lock.
 	var state StreamState
 	mset.store.FastState(&state)
 
-	if lowAckFloor < math.MaxUint64 && lowAckFloor > state.FirstSeq {
-		// Check if we had any zeroAcks, we will need to check them.
-		for _, o := range zeroAcks {
-			var np uint64
-			o.mu.RLock()
-			if o.isLeader() {
-				np = uint64(o.numPending())
-			} else {
-				np, _ = o.calculateNumPending()
-			}
-			o.mu.RUnlock()
-			// This means we have pending and can not remove anything at this time.
-			if np > 0 {
-				return
-			}
-		}
-		if lowAckFloor <= state.LastSeq {
-			// Purge the stream to lowest ack floor + 1
-			mset.store.PurgeEx(_EMPTY_, lowAckFloor+1, 0)
+	if lowAckFloor <= state.FirstSeq {
+		return
+	}
+
+	// Do not want to hold stream lock if calculating numPending.
+	// Check if we had any zeroAcks, we will need to check them.
+	for _, o := range zeroAcks {
+		var np uint64
+		o.mu.RLock()
+		if o.isLeader() {
+			np = uint64(o.numPending())
 		} else {
-			// Here we have a low ack floor higher then our last seq.
-			// So we will just do normal purge.
-			mset.store.Purge()
+			np, _ = o.calculateNumPending()
+		}
+		o.mu.RUnlock()
+		// This means we have pending and can not remove anything at this time.
+		if np > 0 {
+			return
 		}
 	}
+
+	mset.mu.Lock()
+	defer mset.mu.Unlock()
+
+	// Check which purge we need to perform.
+	if lowAckFloor <= state.LastSeq {
+		// Purge the stream to lowest ack floor + 1
+		mset.store.PurgeEx(_EMPTY_, lowAckFloor+1, 0)
+	} else {
+		// Here we have a low ack floor higher then our last seq.
+		// So we will just do normal purge.
+		mset.store.Purge()
+	}
+
 	// Make sure to reset our local lseq.
 	mset.store.FastState(&state)
 	mset.lseq = state.LastSeq


Test patch:
diff --git a/server/events_test.go b/server/events_test.go
index 296a49a8..8aed50fe 100644
--- a/server/events_test.go
+++ b/server/events_test.go
@@ -3538,3 +3538,17 @@ func Benchmark_GetHash(b *testing.B) {
 	default:
 	}
 }
+
+func TestClusterSetupMsgs(t *testing.T) {
+	numServers := 10
+	c := createClusterEx(t, false, 0, false, "cluster", numServers)
+	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 68dd1f90..4b492fec 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -6953,9 +6953,7 @@ func Benchmark_FileStoreSelectMsgBlock(b *testing.B) {
 	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: b.TempDir(), BlockSize: 128},
 		StreamConfig{Name: "zzz", Subjects: []string{"*"}, Storage: FileStorage})
-	if err != nil {
-		b.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(b, err)
 	defer fs.Stop()
 
 	subj, msg := "A", bytes.Repeat([]byte("ABC"), 33) // ~100bytes
@@ -6985,9 +6983,7 @@ func Benchmark_FileStoreLoadNextMsgSameFilterAsStream(b *testing.B) {
 	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: b.TempDir()},
 		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
-	if err != nil {
-		b.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(b, err)
 	defer fs.Stop()
 
 	// Small om purpose.
@@ -7013,9 +7009,7 @@ func Benchmark_FileStoreLoadNextMsgLiteralSubject(b *testing.B) {
 	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: b.TempDir()},
 		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
-	if err != nil {
-		b.Fatalf("Unexpected error: %v", err)
-	}
+	require_NoError(b, err)
 	defer fs.Stop()
 
 	// Small om purpose.
@@ -7039,3 +7033,171 @@ func Benchmark_FileStoreLoadNextMsgLiteralSubject(b *testing.B) {
 		require_NoError(b, err)
 	}
 }
+
+func Benchmark_FileStoreLoadNextMsgNoMsgsFirstSeq(b *testing.B) {
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: b.TempDir(), BlockSize: 8192},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(b, err)
+	defer fs.Stop()
+
+	// Small om purpose.
+	msg := []byte("ok")
+
+	// Add in a bunch of msgs
+	for i := 0; i < 1_000_000; i++ {
+		fs.StoreMsg("foo.bar", nil, msg)
+	}
+
+	b.ResetTimer()
+
+	var smv StoreMsg
+	for i := 0; i < b.N; i++ {
+		// This should error with EOF
+		_, _, err := fs.LoadNextMsg("foo.baz", false, 1, &smv)
+		if err != ErrStoreEOF {
+			b.Fatalf("Wrong error, expected EOF got %v", err)
+		}
+	}
+}
+
+func Benchmark_FileStoreLoadNextMsgNoMsgsNotFirstSeq(b *testing.B) {
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: b.TempDir(), BlockSize: 8192},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(b, err)
+	defer fs.Stop()
+
+	// Small om purpose.
+	msg := []byte("ok")
+
+	// Add in a bunch of msgs
+	for i := 0; i < 1_000_000; i++ {
+		fs.StoreMsg("foo.bar", nil, msg)
+	}
+
+	b.ResetTimer()
+
+	var smv StoreMsg
+	for i := 0; i < b.N; i++ {
+		// This should error with EOF
+		// Make sure the sequence is not first seq of 1.
+		_, _, err := fs.LoadNextMsg("foo.baz", false, 10, &smv)
+		if err != ErrStoreEOF {
+			b.Fatalf("Wrong error, expected EOF got %v", err)
+		}
+	}
+}
+
+func Benchmark_FileStoreLoadNextMsgVerySparseMsgsFirstSeq(b *testing.B) {
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: b.TempDir(), BlockSize: 8192},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(b, err)
+	defer fs.Stop()
+
+	// Small om purpose.
+	msg := []byte("ok")
+
+	// Add in a bunch of msgs
+	for i := 0; i < 1_000_000; i++ {
+		fs.StoreMsg("foo.bar", nil, msg)
+	}
+	// Make last msg one that would match.
+	fs.StoreMsg("foo.baz", nil, msg)
+
+	b.ResetTimer()
+
+	var smv StoreMsg
+	for i := 0; i < b.N; i++ {
+		_, _, err := fs.LoadNextMsg("foo.baz", false, 1, &smv)
+		require_NoError(b, err)
+	}
+}
+
+func Benchmark_FileStoreLoadNextMsgVerySparseMsgsNotFirstSeq(b *testing.B) {
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: b.TempDir(), BlockSize: 8192},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(b, err)
+	defer fs.Stop()
+
+	// Small om purpose.
+	msg := []byte("ok")
+
+	// Add in a bunch of msgs
+	for i := 0; i < 1_000_000; i++ {
+		fs.StoreMsg("foo.bar", nil, msg)
+	}
+	// Make last msg one that would match.
+	fs.StoreMsg("foo.baz", nil, msg)
+
+	b.ResetTimer()
+
+	var smv StoreMsg
+	for i := 0; i < b.N; i++ {
+		// Make sure not first seq.
+		_, _, err := fs.LoadNextMsg("foo.baz", false, 10, &smv)
+		require_NoError(b, err)
+	}
+}
+
+func Benchmark_FileStoreLoadNextMsgVerySparseMsgsInBetween(b *testing.B) {
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: b.TempDir(), BlockSize: 8192},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(b, err)
+	defer fs.Stop()
+
+	// Small om purpose.
+	msg := []byte("ok")
+
+	// Make first msg one that would match as well.
+	fs.StoreMsg("foo.baz", nil, msg)
+	// Add in a bunch of msgs
+	for i := 0; i < 1_000_000; i++ {
+		fs.StoreMsg("foo.bar", nil, msg)
+	}
+	// Make last msg one that would match as well.
+	fs.StoreMsg("foo.baz", nil, msg)
+
+	b.ResetTimer()
+
+	var smv StoreMsg
+	for i := 0; i < b.N; i++ {
+		// Make sure not first seq.
+		_, _, err := fs.LoadNextMsg("foo.baz", false, 2, &smv)
+		require_NoError(b, err)
+	}
+}
+
+func Benchmark_FileStoreLoadNextMsgVerySparseMsgsInBetweenWithWildcard(b *testing.B) {
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: b.TempDir()},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*.*"}, Storage: FileStorage})
+	require_NoError(b, err)
+	defer fs.Stop()
+
+	// Small om purpose.
+	msg := []byte("ok")
+
+	// Make first msg one that would match as well.
+	fs.StoreMsg("foo.1.baz", nil, msg)
+	// Add in a bunch of msgs.
+	// We need to make sure we have a range of subjects that could kick in a linear scan.
+	for i := 0; i < 1_000_000; i++ {
+		subj := fmt.Sprintf("foo.%d.bar", rand.Intn(100_000)+2)
+		fs.StoreMsg(subj, nil, msg)
+	}
+	// Make last msg one that would match as well.
+	fs.StoreMsg("foo.1.baz", nil, msg)
+
+	b.ResetTimer()
+
+	var smv StoreMsg
+	for i := 0; i < b.N; i++ {
+		// Make sure not first seq.
+		_, _, err := fs.LoadNextMsg("foo.*.baz", true, 2, &smv)
+		require_NoError(b, err)
+	}
+}
diff --git a/server/jetstream_cluster_4_test.go b/server/jetstream_cluster_4_test.go
index 0013e5db..4a90d161 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
+					}
 				}
 			}
 		}
diff --git a/server/jetstream_super_cluster_test.go b/server/jetstream_super_cluster_test.go
index ffd7c122..9c2c1fd1 100644
--- a/server/jetstream_super_cluster_test.go
+++ b/server/jetstream_super_cluster_test.go
@@ -1355,6 +1355,10 @@ func TestJetStreamSuperClusterPushConsumerInterest(t *testing.T) {
 }
 
 func TestJetStreamSuperClusterOverflowPlacement(t *testing.T) {
+	orgStatszRateLimit := statszRateLimit
+	statszRateLimit = time.Millisecond * 100
+	defer func() { statszRateLimit = orgStatszRateLimit }()
+
 	sc := createJetStreamSuperClusterWithTemplate(t, jsClusterMaxBytesTempl, 3, 3)
 	defer sc.shutdown()
 
diff --git a/server/memstore_test.go b/server/memstore_test.go
index 3c049233..e74976c4 100644
--- a/server/memstore_test.go
+++ b/server/memstore_test.go
@@ -1051,3 +1051,61 @@ func TestMemStorePurgeExWithDeletedMsgs(t *testing.T) {
 	require_Equal(t, state.LastSeq, 10)
 	require_Equal(t, state.Msgs, 1)
 }
+
+///////////////////////////////////////////////////////////////////////////
+// Benchmarks
+///////////////////////////////////////////////////////////////////////////
+
+func Benchmark_MemStoreNumPendingWithLargeInteriorDeletesScan(b *testing.B) {
+	cfg := &StreamConfig{
+		Name:     "zzz",
+		Subjects: []string{"foo.*.*"},
+		Storage:  MemoryStorage,
+	}
+	ms, err := newMemStore(cfg)
+	require_NoError(b, err)
+	defer ms.Stop()
+
+	msg := []byte("abc")
+	ms.StoreMsg("foo.bar.baz", nil, msg)
+	for i := 1; i <= 1_000_000; i++ {
+		ms.SkipMsg()
+	}
+	ms.StoreMsg("foo.bar.baz", nil, msg)
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		total, _ := ms.NumPending(600_000, "foo.*.baz", false)
+		if total != 1 {
+			b.Fatalf("Expected total of 2 got %d", total)
+		}
+	}
+}
+
+func Benchmark_MemStoreNumPendingWithLargeInteriorDeletesExclude(b *testing.B) {
+	cfg := &StreamConfig{
+		Name:     "zzz",
+		Subjects: []string{"foo.*.*"},
+		Storage:  MemoryStorage,
+	}
+	ms, err := newMemStore(cfg)
+	require_NoError(b, err)
+	defer ms.Stop()
+
+	msg := []byte("abc")
+	ms.StoreMsg("foo.bar.baz", nil, msg)
+	for i := 1; i <= 1_000_000; i++ {
+		ms.SkipMsg()
+	}
+	ms.StoreMsg("foo.bar.baz", nil, msg)
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		total, _ := ms.NumPending(400_000, "foo.*.baz", false)
+		if total != 1 {
+			b.Fatalf("Expected total of 2 got %d", total)
+		}
+	}
+}
diff --git a/server/raft_test.go b/server/raft_test.go
index 90046bcd..b12af87b 100644
--- a/server/raft_test.go
+++ b/server/raft_test.go
@@ -14,6 +14,7 @@
 package server
 
 import (
+	"fmt"
 	"math"
 	"math/rand"
 	"os"
@@ -428,3 +429,42 @@ 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
+	})
+}