Repository: nats-io/nats-server
Pull request #5345 (2024-04-24 11:17:31)

Cherry-pick PRs for 2.10.15-RC.2
Includes the following:

* https://github.com/nats-io/nats-server/pull/5328
* https://github.com/nats-io/nats-server/pull/5324
* https://github.com/nats-io/nats-server/pull/5332
* https://github.com/nats-io/nats-server/pull/5314
* https://github.com/nats-io/nats-server/pull/5333
* https://github.com/nats-io/nats-server/pull/5330
* https://github.com/nats-io/nats-server/pull/5329
* https://github.com/nats-io/nats-server/pull/5338
* https://github.com/nats-io/nats-server/pull/5315
* https://github.com/nats-io/nats-server/pull/5339
* https://github.com/nats-io/nats-server/pull/5340
* https://github.com/nats-io/nats-server/pull/5341
* https://github.com/nats-io/nats-server/pull/5342
* https://github.com/nats-io/nats-server/pull/5347

Signed-off-by: Neil Twigg <neil@nats.io>


Patch:
diff --git a/.travis.yml b/.travis.yml
index 5a80db61..4c72ed8f 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -34,6 +34,8 @@ jobs:
       env: TEST_SUITE=js_cluster_tests_2
     - name: "Run JetStream cluster tests (3)"
       env: TEST_SUITE=js_cluster_tests_3
+    - name: "Run JetStream cluster tests (4)"
+      env: TEST_SUITE=js_cluster_tests_4
     - name: "Run JetStream super cluster tests"
       env: TEST_SUITE=js_super_cluster_tests
     - name: "Run MQTT tests"
diff --git a/go.mod b/go.mod
index d877f236..fc38a41b 100644
--- a/go.mod
+++ b/go.mod
@@ -5,7 +5,7 @@ go 1.20
 require (
 	github.com/klauspost/compress v1.17.7
 	github.com/minio/highwayhash v1.0.2
-	github.com/nats-io/jwt/v2 v2.5.5
+	github.com/nats-io/jwt/v2 v2.5.6
 	github.com/nats-io/nats.go v1.34.1
 	github.com/nats-io/nkeys v0.4.7
 	github.com/nats-io/nuid v1.0.1
diff --git a/go.sum b/go.sum
index e03ef82f..3d32c316 100644
--- a/go.sum
+++ b/go.sum
@@ -3,8 +3,8 @@ github.com/klauspost/compress v1.17.7 h1:ehO88t2UGzQK66LMdE8tibEd1ErmzZjNEqWkjLA
 github.com/klauspost/compress v1.17.7/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
 github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA/g=
 github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
-github.com/nats-io/jwt/v2 v2.5.5 h1:ROfXb50elFq5c9+1ztaUbdlrArNFl2+fQWP6B8HGEq4=
-github.com/nats-io/jwt/v2 v2.5.5/go.mod h1:ZdWS1nZa6WMZfFwwgpEaqBV8EPGVgOTDHN/wTbz0Y5A=
+github.com/nats-io/jwt/v2 v2.5.6 h1:Cp618+z4q042sWqHiSoIHFT08OZtAskui0hTmRfmGGQ=
+github.com/nats-io/jwt/v2 v2.5.6/go.mod h1:ZdWS1nZa6WMZfFwwgpEaqBV8EPGVgOTDHN/wTbz0Y5A=
 github.com/nats-io/nats.go v1.34.1 h1:syWey5xaNHZgicYBemv0nohUPPmaLteiBEUT6Q5+F/4=
 github.com/nats-io/nats.go v1.34.1/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8=
 github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
diff --git a/server/consumer.go b/server/consumer.go
index adc81d4b..ce9ba01a 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -3356,15 +3356,9 @@ func (o *consumer) incDeliveryCount(sseq uint64) uint64 {
 // Lock should be held.
 func (o *consumer) decDeliveryCount(sseq uint64) {
 	if o.rdc == nil {
-		return
-	}
-	if dc, ok := o.rdc[sseq]; ok {
-		if dc == 1 {
-			delete(o.rdc, sseq)
-		} else {
-			o.rdc[sseq] -= 1
-		}
+		o.rdc = make(map[uint64]uint64)
 	}
+	o.rdc[sseq] -= 1
 }
 
 // send a delivery exceeded advisory.
@@ -3512,21 +3506,23 @@ func (o *consumer) getNextMsg() (*jsPubMsg, uint64, error) {
 
 	// Grab next message applicable to us.
 	// We will unlock here in case lots of contention, e.g. WQ.
+	filters, subjf, fseq := o.filters, o.subjf, o.sseq
 	o.mu.Unlock()
 	// Check if we are multi-filtered or not.
-	if o.filters != nil {
-		sm, sseq, err = store.LoadNextMsgMulti(o.filters, o.sseq, &pmsg.StoreMsg)
-	} else if o.subjf != nil { // Means single filtered subject since o.filters means > 1.
-		filter, wc := o.subjf[0].subject, o.subjf[0].hasWildcard
-		sm, sseq, err = store.LoadNextMsg(filter, wc, o.sseq, &pmsg.StoreMsg)
+	if filters != nil {
+		sm, sseq, err = store.LoadNextMsgMulti(filters, fseq, &pmsg.StoreMsg)
+	} else if subjf != nil { // Means single filtered subject since o.filters means > 1.
+		filter, wc := subjf[0].subject, subjf[0].hasWildcard
+		sm, sseq, err = store.LoadNextMsg(filter, wc, fseq, &pmsg.StoreMsg)
 	} else {
 		// No filter here.
-		sm, sseq, err = store.LoadNextMsg(_EMPTY_, false, o.sseq, &pmsg.StoreMsg)
+		sm, sseq, err = store.LoadNextMsg(_EMPTY_, false, fseq, &pmsg.StoreMsg)
 	}
 	if sm == nil {
 		pmsg.returnToPool()
 		pmsg = nil
 	}
+	// Re-acquire lock.
 	o.mu.Lock()
 	// Check if we should move our o.sseq.
 	if sseq >= o.sseq {
@@ -3948,9 +3944,17 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 			}
 		} else {
 			// We will redo this one as long as this is not a redelivery.
-			if dc == 1 {
+			// Need to also test that this is not going backwards since if
+			// we fail to deliver we can end up here from rdq but we do not
+			// want to decrement o.sseq if that is the case.
+			if dc == 1 && pmsg.seq == o.sseq-1 {
 				o.sseq--
 				o.npc++
+			} else if !o.onRedeliverQueue(pmsg.seq) {
+				// We are not on the rdq so decrement the delivery count
+				// and add it back.
+				o.decDeliveryCount(pmsg.seq)
+				o.addToRedeliverQueue(pmsg.seq)
 			}
 			pmsg.returnToPool()
 			goto waitForMsgs
@@ -4386,6 +4390,9 @@ func (o *consumer) didNotDeliver(seq uint64, subj string) {
 		o.active = false
 		checkDeliveryInterest = true
 	} else if o.pending != nil {
+		// Good chance we did not deliver because no interest so force a check.
+		o.processWaiting(false)
+		// If it is still there credit it back.
 		o.creditWaitingRequest(subj)
 		// pull mode and we have pending.
 		if _, ok := o.pending[seq]; ok {
@@ -4394,7 +4401,9 @@ func (o *consumer) didNotDeliver(seq uint64, subj string) {
 			// we know it was not delivered
 			if !o.onRedeliverQueue(seq) {
 				o.addToRedeliverQueue(seq)
-				o.signalNewMessages()
+				if !o.waiting.isEmpty() {
+					o.signalNewMessages()
+				}
 			}
 		}
 	}
diff --git a/server/filestore.go b/server/filestore.go
index e56d7cd2..b0285d45 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -182,7 +182,6 @@ type fileStore struct {
 	adml        int
 	hh          hash.Hash64
 	qch         chan struct{}
-	fch         chan struct{}
 	fsld        chan struct{}
 	cmu         sync.RWMutex
 	cfs         []ConsumerStore
@@ -390,7 +389,6 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 		prf:    prf,
 		oldprf: oldprf,
 		qch:    make(chan struct{}),
-		fch:    make(chan struct{}, 1),
 		fsld:   make(chan struct{}),
 		srv:    fcfg.srv,
 	}
@@ -451,7 +449,6 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 		}
 		// Since we recovered here, make sure to kick ourselves to write out our stream state.
 		fs.dirty++
-		defer fs.kickFlushStateLoop()
 	}
 
 	// Also make sure we get rid of old idx and fss files on return.
@@ -525,8 +522,8 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 	// Setup our sync timer.
 	fs.setSyncTimer()
 
-	// Spin up the go routine that will write out or full state stream index.
-	go fs.flushStreamStateLoop(fs.fch, fs.qch, fs.fsld)
+	// Spin up the go routine that will write out our full state stream index.
+	go fs.flushStreamStateLoop(fs.qch, fs.fsld)
 
 	return fs, nil
 }
@@ -548,6 +545,13 @@ func (fs *fileStore) unlockAllMsgBlocks() {
 }
 
 func (fs *fileStore) UpdateConfig(cfg *StreamConfig) error {
+	start := time.Now()
+	defer func() {
+		if took := time.Since(start); took > time.Minute {
+			fs.warn("UpdateConfig took %v", took.Round(time.Millisecond))
+		}
+	}()
+
 	if fs.isClosed() {
 		return ErrStoreClosed
 	}
@@ -955,16 +959,6 @@ func (mb *msgBlock) ensureLastChecksumLoaded() {
 	copy(mb.lchk[0:], mb.lastChecksum())
 }
 
-// Perform a recover but do not update PSIM.
-// Lock should be held.
-func (fs *fileStore) recoverMsgBlockNoSubjectUpdates(index uint32) (*msgBlock, error) {
-	psim, tsl := fs.psim, fs.tsl
-	fs.psim = nil
-	mb, err := fs.recoverMsgBlock(index)
-	fs.psim, fs.tsl = psim, tsl
-	return mb, err
-}
-
 // Lock held on entry
 func (fs *fileStore) recoverMsgBlock(index uint32) (*msgBlock, error) {
 	mb := fs.initMsgBlock(index)
@@ -1002,6 +996,7 @@ func (fs *fileStore) recoverMsgBlock(index uint32) (*msgBlock, error) {
 	file.Close()
 
 	// Read our index file. Use this as source of truth if possible.
+	// This not applicable in >= 2.10 servers. Here for upgrade paths from < 2.10.
 	if err := mb.readIndexInfo(); err == nil {
 		// Quick sanity check here.
 		// Note this only checks that the message blk file is not newer then this file, or is empty and we expect empty.
@@ -1733,13 +1728,13 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 		// Reverse update of tracking state for this mb, will add new state in below.
 		mstate.Msgs -= mb.msgs
 		mstate.Bytes -= mb.bytes
-		if nmb, err := fs.recoverMsgBlockNoSubjectUpdates(mb.index); err != nil && !os.IsNotExist(err) {
+		if nmb, err := fs.recoverMsgBlock(mb.index); err != nil && !os.IsNotExist(err) {
 			fs.warn("Stream state could not recover last msg block")
 			os.Remove(fn)
 			return errCorruptState
 		} else if nmb != nil {
 			fs.adjustAccounting(mb, nmb)
-			updateTrackingState(&mstate, mb)
+			updateTrackingState(&mstate, nmb)
 		}
 	}
 
@@ -1796,10 +1791,26 @@ func (fs *fileStore) adjustAccounting(mb, nmb *msgBlock) {
 	}
 	nmb.ensurePerSubjectInfoLoaded()
 
+	var smv StoreMsg
+
+	// Need to walk previous messages and undo psim stats.
+	// We already undid msgs and bytes accounting.
+	for seq, lseq := atomic.LoadUint64(&mb.first.seq), atomic.LoadUint64(&mb.last.seq); seq <= lseq; seq++ {
+		// Lookup the message. If an error will be deleted, so can skip.
+		sm, err := nmb.cacheLookup(seq, &smv)
+		if err != nil {
+			continue
+		}
+		if len(sm.subj) > 0 && fs.psim != nil {
+			if info, ok := fs.psim.Find(stringToBytes(sm.subj)); ok {
+				info.total--
+			}
+		}
+	}
+
 	// Walk only new messages and update accounting at fs level. Any messages that should have
 	// triggered limits exceeded will be handled after the recovery and prior to the stream
 	// being available to the system.
-	var smv StoreMsg
 	for seq, lseq := atomic.LoadUint64(&mb.last.seq)+1, atomic.LoadUint64(&nmb.last.seq); seq <= lseq; seq++ {
 		// Lookup the message. If an error will be deleted, so can skip.
 		sm, err := nmb.cacheLookup(seq, &smv)
@@ -1809,17 +1820,6 @@ func (fs *fileStore) adjustAccounting(mb, nmb *msgBlock) {
 		// Since we found it we just need to adjust fs totals and psim.
 		fs.state.Msgs++
 		fs.state.Bytes += fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
-		if len(sm.subj) > 0 && fs.psim != nil {
-			if info, ok := fs.psim.Find(stringToBytes(sm.subj)); ok {
-				info.total++
-				if nmb.index > info.lblk {
-					info.lblk = nmb.index
-				}
-			} else {
-				fs.psim.Insert(stringToBytes(sm.subj), psi{total: 1, fblk: nmb.index, lblk: nmb.index})
-				fs.tsl += len(sm.subj)
-			}
-		}
 	}
 
 	// Now check to see if we had a higher first for the recovered state mb vs nmb.
@@ -2191,7 +2191,6 @@ func (fs *fileStore) expireMsgsOnRecover() {
 	// If we purged anything, make sure we kick flush state loop.
 	if purged > 0 {
 		fs.dirty++
-		fs.kickFlushStateLoop()
 	}
 }
 
@@ -2265,7 +2264,6 @@ func (mb *msgBlock) firstMatchingMulti(sl *Sublist, start uint64, sm *StoreMsg)
 		sm = new(StoreMsg)
 	}
 
-	var result SublistResult
 	for seq := start; seq <= lseq; seq++ {
 		llseq := mb.llseq
 		fsm, err := mb.cacheLookup(seq, sm)
@@ -2274,7 +2272,7 @@ func (mb *msgBlock) firstMatchingMulti(sl *Sublist, start uint64, sm *StoreMsg)
 		}
 		expireOk := seq == lseq && mb.llseq == seq
 
-		if r := sl.MatchWithResult(fsm.subj, &result); len(r.psubs) > 0 {
+		if sl.HasInterest(fsm.subj) {
 			return fsm, expireOk, nil
 		}
 		// If we are here we did not match, so put the llseq back.
@@ -3089,10 +3087,6 @@ func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 	// Add to our list of blocks and mark as last.
 	fs.addMsgBlock(mb)
 
-	if fs.dirty > 0 {
-		fs.kickFlushStateLoop()
-	}
-
 	return mb, nil
 }
 
@@ -3548,6 +3542,13 @@ func (fs *fileStore) enforceBytesLimit() {
 // will most likely only be the last one, so can take a more conservative approach.
 // Lock should be held.
 func (fs *fileStore) enforceMsgPerSubjectLimit(fireCallback bool) {
+	start := time.Now()
+	defer func() {
+		if took := time.Since(start); took > time.Minute {
+			fs.warn("enforceMsgPerSubjectLimit took %v", took.Round(time.Millisecond))
+		}
+	}()
+
 	maxMsgsPer := uint64(fs.cfg.MaxMsgsPer)
 
 	// We may want to suppress callbacks from remove during this process
@@ -3829,7 +3830,6 @@ func (fs *fileStore) removeMsg(seq uint64, secure, viaLimits, needFSLock bool) (
 		// TODO(dlc) - This should not be inline, should kick the sync routine.
 		if mb.rbytes > compactMinimum && mb.bytes*2 < mb.rbytes && !isLastBlock {
 			mb.compact()
-			fs.kickFlushStateLoop()
 		}
 	}
 
@@ -5813,6 +5813,7 @@ var (
 	errUnknownCipher = errors.New("unknown cipher")
 	errNoMainKey     = errors.New("encrypted store encountered with no main key")
 	errNoBlkData     = errors.New("message block data missing")
+	errStateTooBig   = errors.New("store state too big for optional write")
 )
 
 const (
@@ -6634,8 +6635,6 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 	cb := fs.scb
 	fs.mu.Unlock()
 
-	fs.kickFlushStateLoop()
-
 	if cb != nil {
 		cb(-int64(purged), -int64(bytes), 0, _EMPTY_)
 	}
@@ -6916,7 +6915,6 @@ SKIP:
 	fs.state.Bytes -= bytes
 
 	fs.dirty++
-	fs.kickFlushStateLoop()
 
 	cb := fs.scb
 	fs.mu.Unlock()
@@ -6970,7 +6968,6 @@ func (fs *fileStore) reset() error {
 	// If we purged anything, make sure we kick flush state loop.
 	if purged > 0 {
 		fs.dirty++
-		fs.kickFlushStateLoop()
 	}
 
 	fs.mu.Unlock()
@@ -7057,7 +7054,6 @@ func (fs *fileStore) Truncate(seq uint64) error {
 	fs.resetGlobalPerSubjectInfo()
 
 	fs.dirty++
-	fs.kickFlushStateLoop()
 
 	cb := fs.scb
 	fs.mu.Unlock()
@@ -7165,8 +7161,6 @@ func (mb *msgBlock) dirtyCloseWithRemove(remove bool) {
 		if mb.kfn != _EMPTY_ {
 			os.Remove(mb.kfn)
 		}
-		// Since we are removing a block kick the state flusher.
-		mb.fs.kickFlushStateLoop()
 	}
 }
 
@@ -7485,50 +7479,25 @@ const (
 // This go routine runs and receives kicks to write out our full stream state index.
 // This will get kicked when we create a new block or when we delete a block in general.
 // This is also called during Stop().
-func (fs *fileStore) flushStreamStateLoop(fch, qch, done chan struct{}) {
-	// Make sure we do not try to write these out too fast.
-	const writeThreshold = time.Minute
-	lastWrite := time.Time{}
-
-	// We will use these to complete the full state write while not doing them too fast.
-	var dt *time.Timer
-	var dtc <-chan time.Time
-
+func (fs *fileStore) flushStreamStateLoop(qch, done chan struct{}) {
+	// Signal we are done on exit.
 	defer close(done)
 
+	// Make sure we do not try to write these out too fast.
+	const writeThreshold = 2 * time.Minute
+	t := time.NewTicker(writeThreshold)
+	defer t.Stop()
+
 	for {
 		select {
-		case <-fch:
-			if elapsed := time.Since(lastWrite); elapsed > writeThreshold {
-				fs.writeFullState()
-				lastWrite = time.Now()
-				if dt != nil {
-					dt.Stop()
-					dt, dtc = nil, nil
-				}
-			} else if dtc == nil {
-				fireIn := time.Until(lastWrite.Add(writeThreshold))
-				if fireIn < 0 {
-					fireIn = 100 * time.Millisecond
-				}
-				dt = time.NewTimer(fireIn)
-				dtc = dt.C
-			}
-		case <-dtc:
+		case <-t.C:
 			fs.writeFullState()
-			lastWrite = time.Now()
-			dt, dtc = nil, nil
 		case <-qch:
 			return
 		}
 	}
 }
 
-// Kick the flusher.
-func (fs *fileStore) kickFlushStateLoop() {
-	kickFlusher(fs.fch)
-}
-
 // Helper since unixnano of zero time undefined.
 func timestampNormalized(t time.Time) int64 {
 	if t.IsZero() {
@@ -7537,6 +7506,17 @@ func timestampNormalized(t time.Time) int64 {
 	return t.UnixNano()
 }
 
+// writeFullState will proceed to write the full meta state iff not complex and time consuming.
+// Since this is for quick recovery it is optional and should not block/stall normal operations.
+func (fs *fileStore) writeFullState() error {
+	return fs._writeFullState(false)
+}
+
+// forceWriteFullState will proceed to write the full meta state. This should only be called by stop()
+func (fs *fileStore) forceWriteFullState() error {
+	return fs._writeFullState(true)
+}
+
 // This will write the full binary state for the stream.
 // This plus everything new since last hash will be the total recovered state.
 // This state dump will have the following.
@@ -7544,13 +7524,34 @@ func timestampNormalized(t time.Time) int64 {
 // 2. PSIM - Per Subject Index Map - Tracks first and last blocks with subjects present.
 // 3. MBs - Index, Bytes, First and Last Sequence and Timestamps, and the deleted map (avl.seqset).
 // 4. Last block index and hash of record inclusive to this stream state.
-func (fs *fileStore) writeFullState() error {
+func (fs *fileStore) _writeFullState(force bool) error {
+	start := time.Now()
 	fs.mu.Lock()
 	if fs.closed || fs.dirty == 0 {
 		fs.mu.Unlock()
 		return nil
 	}
 
+	// For calculating size and checking time costs for non forced calls.
+	numSubjects := fs.numSubjects()
+
+	// If we are not being forced to write out our state, check the complexity for time costs as to not
+	// block or stall normal operations.
+	// We will base off of number of subjects and interior deletes. A very large number of msg blocks could also
+	// be used, but for next server version will redo all meta handling to be disk based. So this is temporary.
+	if !force {
+		const numThreshold = 1_000_000
+		// Calculate interior deletes.
+		var numDeleted int
+		if fs.state.LastSeq > fs.state.FirstSeq {
+			numDeleted = int((fs.state.LastSeq - fs.state.FirstSeq + 1) - fs.state.Msgs)
+		}
+		if numSubjects > numThreshold || numDeleted > numThreshold {
+			fs.mu.Unlock()
+			return errStateTooBig
+		}
+	}
+
 	// We track this through subsequent runs to get an avg per blk used for subsequent runs.
 	avgDmapLen := fs.adml
 	// If first time through could be 0
@@ -7558,11 +7559,8 @@ func (fs *fileStore) writeFullState() error {
 		avgDmapLen = 1024
 	}
 
-	// For calculating size.
-	numSubjects := fs.psim.Size()
-
 	// Calculate and estimate of the uper bound on the  size to avoid multiple allocations.
-	sz := 2 + // Magic and Version
+	sz := hdrLen + // Magic and Version
 		(binary.MaxVarintLen64 * 6) + // FS data
 		binary.MaxVarintLen64 + fs.tsl + // NumSubjects + total subject length
 		numSubjects*(binary.MaxVarintLen64*4) + // psi record
@@ -7576,7 +7574,7 @@ func (fs *fileStore) writeFullState() error {
 
 	if sz <= ssz {
 		var _buf [ssz]byte
-		buf, sz = _buf[0:2:ssz], ssz
+		buf, sz = _buf[0:hdrLen:ssz], ssz
 	} else {
 		buf = make([]byte, hdrLen, sz)
 	}
@@ -7685,8 +7683,6 @@ func (fs *fileStore) writeFullState() error {
 		fs.warn("Stream state encountered internal inconsistency on write")
 		// Rebuild our fs state from the mb state.
 		fs.rebuildState(nil)
-		// Make sure to reprocess.
-		fs.kickFlushStateLoop()
 		return errCorruptState
 	}
 
@@ -7721,6 +7717,9 @@ func (fs *fileStore) writeFullState() error {
 	}
 	fs.mu.Unlock()
 
+	if took := time.Since(start); took > time.Minute {
+		fs.warn("WriteFullState took %v (%d bytes)", took.Round(time.Millisecond), len(buf))
+	}
 	return nil
 }
 
@@ -7761,7 +7760,7 @@ func (fs *fileStore) stop(writeState bool) error {
 		fs.mu.Unlock()
 		<-fsld
 		// Write full state if needed. If not dirty this is a no-op.
-		fs.writeFullState()
+		fs.forceWriteFullState()
 		fs.mu.Lock()
 	}
 
diff --git a/server/memstore.go b/server/memstore.go
index 4fbbc197..3da696bc 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -1007,13 +1007,12 @@ func (ms *memStore) LoadNextMsgMulti(sl *Sublist, start uint64, smp *StoreMsg) (
 	// Initial setup.
 	fseq, lseq := start, ms.state.LastSeq
 
-	var result SublistResult
 	for nseq := fseq; nseq <= lseq; nseq++ {
 		sm, ok := ms.msgs[nseq]
 		if !ok {
 			continue
 		}
-		if r := sl.MatchWithResult(sm.subj, &result); len(r.psubs) > 0 {
+		if sl.HasInterest(sm.subj) {
 			if smp == nil {
 				smp = new(StoreMsg)
 			}
diff --git a/server/raft.go b/server/raft.go
index 83157f6e..f610e570 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -150,6 +150,7 @@ type raft struct {
 	pae     map[uint64]*appendEntry        // Pending append entries
 
 	elect  *time.Timer // Election timer, normally accessed via electTimer
+	etlr   time.Time   // Election timer last reset time, for unit tests only
 	active time.Time   // Last activity time, i.e. for heartbeats
 	llqrt  time.Time   // Last quorum lost time
 	lsut   time.Time   // Last scale-up time
@@ -1766,6 +1767,7 @@ func (n *raft) resetElectionTimeoutWithLock() {
 
 // Lock should be held.
 func (n *raft) resetElect(et time.Duration) {
+	n.etlr = time.Now()
 	if n.elect == nil {
 		n.elect = time.NewTimer(et)
 	} else {
@@ -2682,15 +2684,11 @@ func (n *raft) applyCommit(index uint64) error {
 		n.debug("Ignoring apply commit for %d, already processed", index)
 		return nil
 	}
-	original := n.commit
-	n.commit = index
 
 	if n.State() == Leader {
 		delete(n.acks, index)
 	}
 
-	var fpae bool
-
 	ae := n.pae[index]
 	if ae == nil {
 		var state StreamState
@@ -2708,15 +2706,14 @@ func (n *raft) applyCommit(index uint64) error {
 				// Reset and cancel any catchup.
 				n.resetWAL()
 				n.cancelCatchup()
-			} else {
-				n.commit = original
 			}
 			return errEntryLoadFailed
 		}
 	} else {
-		fpae = true
+		defer delete(n.pae, index)
 	}
 
+	n.commit = index
 	ae.buf = nil
 
 	var committed []*Entry
@@ -2791,9 +2788,6 @@ func (n *raft) applyCommit(index uint64) error {
 			committed = append(committed, e)
 		}
 	}
-	if fpae {
-		delete(n.pae, index)
-	}
 	// Pass to the upper layers if we have normal entries. It is
 	// entirely possible that 'committed' might be an empty slice here,
 	// which will happen if we've processed updates inline (like peer
@@ -3121,9 +3115,13 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	if n.State() == Leader {
 		// If we are the same we should step down to break the tie.
 		if ae.term >= n.term {
-			n.term = ae.term
-			n.vote = noVote
-			n.writeTermVote()
+			// 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 from another leader, stepping down to %q", ae.leader)
 			n.stepdown.push(ae.leader)
 		} else {
@@ -3142,13 +3140,18 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 	// another node has taken on the leader role already, so we should convert
 	// to a follower of that node instead.
 	if n.State() == Candidate {
-		n.debug("Received append entry in candidate state from %q, converting to follower", ae.leader)
-		if n.term < ae.term {
-			n.term = ae.term
-			n.vote = noVote
-			n.writeTermVote()
+		// Ignore old terms, otherwise we might end up stepping down incorrectly.
+		if ae.term >= n.term {
+			// 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.stepdown.push(ae.leader)
 		}
-		n.stepdown.push(ae.leader)
 	}
 
 	// Catching up state.
@@ -3258,6 +3261,15 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 		// so make sure this is a snapshot entry. If it is not start the catchup process again since it
 		// means we may have missed additional messages.
 		if catchingUp {
+			// This means we already entered into a catchup state but what the leader sent us did not match what we expected.
+			// Snapshots and peerstate will always be together when a leader is catching us up in this fashion.
+			if len(ae.entries) != 2 || ae.entries[0].Type != EntrySnapshot || ae.entries[1].Type != EntryPeerState {
+				n.warn("Expected first catchup entry to be a snapshot and peerstate, will retry")
+				n.cancelCatchup()
+				n.Unlock()
+				return
+			}
+
 			// Check if only our terms do not match here.
 			if ae.pindex == n.pindex {
 				// Make sure pterms match and we take on the leader's.
@@ -3267,14 +3279,6 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 				n.Unlock()
 				return
 			}
-			// This means we already entered into a catchup state but what the leader sent us did not match what we expected.
-			// Snapshots and peerstate will always be together when a leader is catching us up in this fashion.
-			if len(ae.entries) != 2 || ae.entries[0].Type != EntrySnapshot || ae.entries[1].Type != EntryPeerState {
-				n.warn("Expected first catchup entry to be a snapshot and peerstate, will retry")
-				n.cancelCatchup()
-				n.Unlock()
-				return
-			}
 
 			if ps, err := decodePeerState(ae.entries[1].Data); err == nil {
 				n.processPeerState(ps)
@@ -3875,7 +3879,6 @@ func (n *raft) processVoteRequest(vr *voteRequest) error {
 	}
 
 	n.Lock()
-	n.resetElectionTimeout()
 
 	vresp := &voteResponse{n.term, n.id, false}
 	defer n.debug("Sending a voteResponse %+v -> %q", vresp, vr.reply)
@@ -3907,6 +3910,7 @@ func (n *raft) processVoteRequest(vr *voteRequest) error {
 		n.term = vr.term
 		n.vote = vr.candidate
 		n.writeTermVote()
+		n.resetElectionTimeout()
 	} else {
 		if vr.term >= n.term && n.vote == noVote {
 			n.term = vr.term
diff --git a/server/stree/dump.go b/server/stree/dump.go
index 79b6fd99..4a7d76fb 100644
--- a/server/stree/dump.go
+++ b/server/stree/dump.go
@@ -38,7 +38,7 @@ func (t *SubjectTree[T]) dump(w io.Writer, n node, depth int) {
 	} else {
 		// We are a node type here, grab meta portion.
 		bn := n.base()
-		fmt.Fprintf(w, "%s %s Prefix: %q\n", dumpPre(depth), n.kind(), bn.prefix[:bn.prefixLen])
+		fmt.Fprintf(w, "%s %s Prefix: %q\n", dumpPre(depth), n.kind(), bn.prefix)
 		depth++
 		n.iter(func(n node) bool {
 			t.dump(w, n, depth)
diff --git a/server/stree/node.go b/server/stree/node.go
index 6285ba3c..c8edfe3e 100644
--- a/server/stree/node.go
+++ b/server/stree/node.go
@@ -32,13 +32,22 @@ type node interface {
 	path() []byte
 }
 
-// Maximum prefix len
-// We expect the most savings to come from long shared prefixes.
-const maxPrefixLen = 24
-
-// 64 bytes total - an L1 cache line.
 type meta struct {
-	prefix    [maxPrefixLen]byte
-	prefixLen uint16
-	size      uint16
+	prefix []byte
+	size   uint16
+}
+
+func (n *meta) isLeaf() bool { return false }
+func (n *meta) base() *meta  { return n }
+
+func (n *meta) setPrefix(pre []byte) {
+	n.prefix = append([]byte(nil), pre...)
+}
+
+func (n *meta) numChildren() uint16 { return n.size }
+func (n *meta) path() []byte        { return n.prefix }
+
+// Will match parts against our prefix.
+func (n *meta) matchParts(parts [][]byte) ([][]byte, bool) {
+	return matchParts(parts, n.prefix)
 }
diff --git a/server/stree/node16.go b/server/stree/node16.go
index b61f286d..2d206afd 100644
--- a/server/stree/node16.go
+++ b/server/stree/node16.go
@@ -26,16 +26,6 @@ func newNode16(prefix []byte) *node16 {
 	return nn
 }
 
-func (n *node16) isLeaf() bool { return false }
-func (n *node16) base() *meta  { return &n.meta }
-
-func (n *node16) setPrefix(pre []byte) {
-	n.prefixLen = uint16(min(len(pre), maxPrefixLen))
-	for i := uint16(0); i < n.prefixLen; i++ {
-		n.prefix[i] = pre[i]
-	}
-}
-
 // Currently we do not keep node16 sorted or use bitfields for traversal so just add to the end.
 // TODO(dlc) - We should revisit here with more detailed benchmarks.
 func (n *node16) addChild(c byte, nn node) {
@@ -47,9 +37,6 @@ func (n *node16) addChild(c byte, nn node) {
 	n.size++
 }
 
-func (n *node16) numChildren() uint16 { return n.size }
-func (n *node16) path() []byte        { return n.prefix[:n.prefixLen] }
-
 func (n *node16) findChild(c byte) *node {
 	for i := uint16(0); i < n.size; i++ {
 		if n.key[i] == c {
@@ -62,7 +49,7 @@ func (n *node16) findChild(c byte) *node {
 func (n *node16) isFull() bool { return n.size >= 16 }
 
 func (n *node16) grow() node {
-	nn := newNode256(n.prefix[:n.prefixLen])
+	nn := newNode256(n.prefix)
 	for i := 0; i < 16; i++ {
 		nn.addChild(n.key[i], n.child[i])
 	}
@@ -101,11 +88,6 @@ func (n *node16) shrink() node {
 	return nn
 }
 
-// Will match parts against our prefix.no
-func (n *node16) matchParts(parts [][]byte) ([][]byte, bool) {
-	return matchParts(parts, n.prefix[:n.prefixLen])
-}
-
 // Iterate over all children calling func f.
 func (n *node16) iter(f func(node) bool) {
 	for i := uint16(0); i < n.size; i++ {
diff --git a/server/stree/node256.go b/server/stree/node256.go
index c7bcd336..fdadde0b 100644
--- a/server/stree/node256.go
+++ b/server/stree/node256.go
@@ -25,24 +25,11 @@ func newNode256(prefix []byte) *node256 {
 	return nn
 }
 
-func (n *node256) isLeaf() bool { return false }
-func (n *node256) base() *meta  { return &n.meta }
-
-func (n *node256) setPrefix(pre []byte) {
-	n.prefixLen = uint16(min(len(pre), maxPrefixLen))
-	for i := uint16(0); i < n.prefixLen; i++ {
-		n.prefix[i] = pre[i]
-	}
-}
-
 func (n *node256) addChild(c byte, nn node) {
 	n.child[c] = nn
 	n.size++
 }
 
-func (n *node256) numChildren() uint16 { return n.size }
-func (n *node256) path() []byte        { return n.prefix[:n.prefixLen] }
-
 func (n *node256) findChild(c byte) *node {
 	if n.child[c] != nil {
 		return &n.child[c]
@@ -75,11 +62,6 @@ func (n *node256) shrink() node {
 	return nn
 }
 
-// Will match parts against our prefix.
-func (n *node256) matchParts(parts [][]byte) ([][]byte, bool) {
-	return matchParts(parts, n.prefix[:n.prefixLen])
-}
-
 // Iterate over all children calling func f.
 func (n *node256) iter(f func(node) bool) {
 	for i := 0; i < 256; i++ {
diff --git a/server/stree/node4.go b/server/stree/node4.go
index 195299e2..2d489625 100644
--- a/server/stree/node4.go
+++ b/server/stree/node4.go
@@ -26,16 +26,6 @@ func newNode4(prefix []byte) *node4 {
 	return nn
 }
 
-func (n *node4) isLeaf() bool { return false }
-func (n *node4) base() *meta  { return &n.meta }
-
-func (n *node4) setPrefix(pre []byte) {
-	n.prefixLen = uint16(min(len(pre), maxPrefixLen))
-	for i := uint16(0); i < n.prefixLen; i++ {
-		n.prefix[i] = pre[i]
-	}
-}
-
 // Currently we do not need to keep sorted for traversal so just add to the end.
 func (n *node4) addChild(c byte, nn node) {
 	if n.size >= 4 {
@@ -46,9 +36,6 @@ func (n *node4) addChild(c byte, nn node) {
 	n.size++
 }
 
-func (n *node4) numChildren() uint16 { return n.size }
-func (n *node4) path() []byte        { return n.prefix[:n.prefixLen] }
-
 func (n *node4) findChild(c byte) *node {
 	for i := uint16(0); i < n.size; i++ {
 		if n.key[i] == c {
@@ -61,7 +48,7 @@ func (n *node4) findChild(c byte) *node {
 func (n *node4) isFull() bool { return n.size >= 4 }
 
 func (n *node4) grow() node {
-	nn := newNode16(n.prefix[:n.prefixLen])
+	nn := newNode16(n.prefix)
 	for i := 0; i < 4; i++ {
 		nn.addChild(n.key[i], n.child[i])
 	}
@@ -96,11 +83,6 @@ func (n *node4) shrink() node {
 	return nil
 }
 
-// Will match parts against our prefix.
-func (n *node4) matchParts(parts [][]byte) ([][]byte, bool) {
-	return matchParts(parts, n.prefix[:n.prefixLen])
-}
-
 // Iterate over all children calling func f.
 func (n *node4) iter(f func(node) bool) {
 	for i := uint16(0); i < n.size; i++ {
diff --git a/server/stree/stree.go b/server/stree/stree.go
index ad75eb5f..b6531924 100644
--- a/server/stree/stree.go
+++ b/server/stree/stree.go
@@ -60,7 +60,7 @@ func (t *SubjectTree[T]) Insert(subject []byte, value T) (*T, bool) {
 
 // Find will find the value and return it or false if it was not found.
 func (t *SubjectTree[T]) Find(subject []byte) (*T, bool) {
-	var si uint16
+	var si int
 	for n := t.root; n != nil; {
 		if n.isLeaf() {
 			if ln := n.(*leaf[T]); ln.match(subject[si:]) {
@@ -69,13 +69,13 @@ func (t *SubjectTree[T]) Find(subject []byte) (*T, bool) {
 			return nil, false
 		}
 		// We are a node type here, grab meta portion.
-		if bn := n.base(); bn.prefixLen > 0 {
-			end := min(int(si+bn.prefixLen), len(subject))
-			if !bytes.Equal(subject[si:end], bn.prefix[:bn.prefixLen]) {
+		if bn := n.base(); len(bn.prefix) > 0 {
+			end := min(si+len(bn.prefix), len(subject))
+			if !bytes.Equal(subject[si:end], bn.prefix) {
 				return nil, false
 			}
 			// Increment our subject index.
-			si += bn.prefixLen
+			si += len(bn.prefix)
 		}
 		if an := n.findChild(pivot(subject, si)); an != nil {
 			n = *an
@@ -157,9 +157,9 @@ func (t *SubjectTree[T]) insert(np *node, subject []byte, value T, si int) (*T,
 
 	// Non-leaf nodes.
 	bn := n.base()
-	if bn.prefixLen > 0 {
-		cpi := commonPrefixLen(bn.prefix[:bn.prefixLen], subject[si:])
-		if pli := int(bn.prefixLen); cpi >= pli {
+	if len(bn.prefix) > 0 {
+		cpi := commonPrefixLen(bn.prefix, subject[si:])
+		if pli := len(bn.prefix); cpi >= pli {
 			// Move past this node. We look for an existing child node to recurse into.
 			// If one does not exist we can create a new leaf node.
 			si += pli
@@ -180,7 +180,7 @@ func (t *SubjectTree[T]) insert(np *node, subject []byte, value T, si int) (*T,
 			// We will insert a new node4 and attach our current node below after adjusting prefix.
 			nn := newNode4(prefix)
 			// Shift the prefix for our original node.
-			n.setPrefix(bn.prefix[cpi:bn.prefixLen])
+			n.setPrefix(bn.prefix[cpi:])
 			nn.addChild(pivot(bn.prefix[:], 0), n)
 			// Add in our new leaf.
 			nn.addChild(pivot(subject[si:], 0), newLeaf(subject[si:], value))
@@ -203,7 +203,7 @@ func (t *SubjectTree[T]) insert(np *node, subject []byte, value T, si int) (*T,
 }
 
 // internal function to recursively find the leaf to delete. Will do compaction if the item is found and removed.
-func (t *SubjectTree[T]) delete(np *node, subject []byte, si uint16) (*T, bool) {
+func (t *SubjectTree[T]) delete(np *node, subject []byte, si int) (*T, bool) {
 	if t == nil || np == nil || *np == nil || len(subject) == 0 {
 		return nil, false
 	}
@@ -217,12 +217,12 @@ func (t *SubjectTree[T]) delete(np *node, subject []byte, si uint16) (*T, bool)
 		return nil, false
 	}
 	// Not a leaf node.
-	if bn := n.base(); bn.prefixLen > 0 {
-		if !bytes.Equal(subject[si:si+bn.prefixLen], bn.prefix[:bn.prefixLen]) {
+	if bn := n.base(); len(bn.prefix) > 0 {
+		if !bytes.Equal(subject[si:si+len(bn.prefix)], bn.prefix) {
 			return nil, false
 		}
 		// Increment our subject index.
-		si += bn.prefixLen
+		si += len(bn.prefix)
 	}
 	p := pivot(subject, si)
 	nna := n.findChild(p)
@@ -238,7 +238,7 @@ func (t *SubjectTree[T]) delete(np *node, subject []byte, si uint16) (*T, bool)
 			if sn := n.shrink(); sn != nil {
 				bn := n.base()
 				// Make sure to set cap so we force an append to copy below.
-				pre := bn.prefix[:bn.prefixLen:bn.prefixLen]
+				pre := bn.prefix[:len(bn.prefix):len(bn.prefix)]
 				// Need to fix up prefixes/suffixes.
 				if sn.isLeaf() {
 					ln := sn.(*leaf[T])
@@ -248,7 +248,7 @@ func (t *SubjectTree[T]) delete(np *node, subject []byte, si uint16) (*T, bool)
 					// We are a node here, we need to add in the old prefix.
 					if len(pre) > 0 {
 						bsn := sn.base()
-						sn.setPrefix(append(pre, bsn.prefix[:bsn.prefixLen]...))
+						sn.setPrefix(append(pre, bsn.prefix...))
 					}
 				}
 				*np = sn
@@ -287,9 +287,9 @@ func (t *SubjectTree[T]) match(n node, parts [][]byte, pre []byte, cb func(subje
 		// We have normal nodes here.
 		// We need to append our prefix
 		bn := n.base()
-		if bn.prefixLen > 0 {
+		if len(bn.prefix) > 0 {
 			// Note that this append may reallocate, but it doesn't modify "pre" at the "match" callsite.
-			pre = append(pre, bn.prefix[:bn.prefixLen]...)
+			pre = append(pre, bn.prefix...)
 		}
 
 		// Check our remaining parts.
@@ -359,7 +359,7 @@ func (t *SubjectTree[T]) iter(n node, pre []byte, cb func(subject []byte, val *T
 	// We are normal node here.
 	bn := n.base()
 	// Note that this append may reallocate, but it doesn't modify "pre" at the "iter" callsite.
-	pre = append(pre, bn.prefix[:bn.prefixLen]...)
+	pre = append(pre, bn.prefix...)
 	// Collect nodes since unsorted.
 	var _nodes [256]node
 	nodes := _nodes[:0]
diff --git a/server/stree/util.go b/server/stree/util.go
index 5ced54bc..585800ab 100644
--- a/server/stree/util.go
+++ b/server/stree/util.go
@@ -29,7 +29,7 @@ func commonPrefixLen(s1, s2 []byte) int {
 			break
 		}
 	}
-	return min(i, maxPrefixLen)
+	return i
 }
 
 // Helper to copy bytes.
diff --git a/server/sublist.go b/server/sublist.go
index 8dcc3ff9..71c256e4 100644
--- a/server/sublist.go
+++ b/server/sublist.go
@@ -527,21 +527,20 @@ var emptyResult = &SublistResult{}
 // Match will match all entries to the literal subject.
 // It will return a set of results for both normal and queue subscribers.
 func (s *Sublist) Match(subject string) *SublistResult {
-	return s.match(subject, true, &SublistResult{})
+	return s.match(subject, true)
 }
 
-// MatchWithResult will match all entries to the literal subject, reusing the
-// supplied SublistResult to reduce allocations on hot paths.
-// It will return a set of results for both normal and queue subscribers.
-func (s *Sublist) MatchWithResult(subject string, result *SublistResult) *SublistResult {
-	return s.match(subject, true, result)
+// HasInterest will return whether or not there is any interest in the subject.
+// In cases where more detail is not required, this may be faster than Match.
+func (s *Sublist) HasInterest(subject string) bool {
+	return s.hasInterest(subject, true)
 }
 
 func (s *Sublist) matchNoLock(subject string) *SublistResult {
-	return s.match(subject, false, &SublistResult{})
+	return s.match(subject, false)
 }
 
-func (s *Sublist) match(subject string, doLock bool, result *SublistResult) *SublistResult {
+func (s *Sublist) match(subject string, doLock bool) *SublistResult {
 	atomic.AddUint64(&s.matches, 1)
 
 	// Check cache first.
@@ -576,8 +575,7 @@ func (s *Sublist) match(subject string, doLock bool, result *SublistResult) *Sub
 	tokens = append(tokens, subject[start:])
 
 	// FIXME(dlc) - Make shared pool between sublist and client readLoop?
-	result.psubs = result.psubs[:0]
-	result.qsubs = result.qsubs[:0]
+	result := &SublistResult{}
 
 	// Get result from the main structure and place into the shared cache.
 	// Hold the read lock to avoid race between match and store.
@@ -616,6 +614,47 @@ func (s *Sublist) match(subject string, doLock bool, result *SublistResult) *Sub
 	return result
 }
 
+func (s *Sublist) hasInterest(subject string, doLock bool) bool {
+	// Check cache first.
+	if doLock {
+		s.RLock()
+	}
+	var matched bool
+	if s.cache != nil {
+		_, matched = s.cache[subject]
+	}
+	if doLock {
+		s.RUnlock()
+	}
+	if matched {
+		atomic.AddUint64(&s.cacheHits, 1)
+		return true
+	}
+
+	tsa := [32]string{}
+	tokens := tsa[:0]
+	start := 0
+	for i := 0; i < len(subject); i++ {
+		if subject[i] == btsep {
+			if i-start == 0 {
+				return false
+			}
+			tokens = append(tokens, subject[start:i])
+			start = i + 1
+		}
+	}
+	if start >= len(subject) {
+		return false
+	}
+	tokens = append(tokens, subject[start:])
+
+	if doLock {
+		s.RLock()
+		defer s.RUnlock()
+	}
+	return matchLevelForAny(s.root, tokens)
+}
+
 // Remove entries in the cache until we are under the maximum.
 // TODO(dlc) this could be smarter now that its not inline.
 func (s *Sublist) reduceCacheCount() {
@@ -728,6 +767,36 @@ func matchLevel(l *level, toks []string, results *SublistResult) {
 	}
 }
 
+func matchLevelForAny(l *level, toks []string) bool {
+	var pwc, n *node
+	for i, t := range toks {
+		if l == nil {
+			return false
+		}
+		if l.fwc != nil {
+			return true
+		}
+		if pwc = l.pwc; pwc != nil {
+			if match := matchLevelForAny(pwc.next, toks[i+1:]); match {
+				return true
+			}
+		}
+		n = l.nodes[t]
+		if n != nil {
+			l = n.next
+		} else {
+			l = nil
+		}
+	}
+	if n != nil {
+		return true
+	}
+	if pwc != nil {
+		return true
+	}
+	return false
+}
+
 // lnt is used to track descent into levels for a removal for pruning.
 type lnt struct {
 	l *level


Test patch:
diff --git a/scripts/runTestsOnTravis.sh b/scripts/runTestsOnTravis.sh
index 62cd059e..84ebe2f3 100755
--- a/scripts/runTestsOnTravis.sh
+++ b/scripts/runTestsOnTravis.sh
@@ -38,7 +38,7 @@ elif [ "$1" = "js_tests" ]; then
     # with `TestJetStream`. We exclude the clustered and super-clustered
     # tests by using the appropriate tags.
 
-    go test -race -v -run=TestJetStream ./server -tags=skip_js_cluster_tests,skip_js_cluster_tests_2,skip_js_cluster_tests_3,skip_js_super_cluster_tests -count=1 -vet=off -timeout=30m -failfast
+    go test -race -v -run=TestJetStream ./server -tags=skip_js_cluster_tests,skip_js_cluster_tests_2,skip_js_cluster_tests_3,skip_js_cluster_tests_4,skip_js_super_cluster_tests -count=1 -vet=off -timeout=30m -failfast
 
 elif [ "$1" = "js_cluster_tests_1" ]; then
 
@@ -46,7 +46,7 @@ elif [ "$1" = "js_cluster_tests_1" ]; then
     # start with `TestJetStreamCluster`. Will run the first batch of tests,
     # excluding others with use of proper tags.
 
-    go test -race -v -run=TestJetStreamCluster ./server -tags=skip_js_cluster_tests_2,skip_js_cluster_tests_3 -count=1 -vet=off -timeout=30m -failfast
+    go test -race -v -run=TestJetStreamCluster ./server -tags=skip_js_cluster_tests_2,skip_js_cluster_tests_3,skip_js_cluster_tests_4 -count=1 -vet=off -timeout=30m -failfast
 
 elif [ "$1" = "js_cluster_tests_2" ]; then
 
@@ -54,7 +54,7 @@ elif [ "$1" = "js_cluster_tests_2" ]; then
     # start with `TestJetStreamCluster`. Will run the second batch of tests,
     # excluding others with use of proper tags.
 
-    go test -race -v -run=TestJetStreamCluster ./server -tags=skip_js_cluster_tests,skip_js_cluster_tests_3 -count=1 -vet=off -timeout=30m -failfast
+    go test -race -v -run=TestJetStreamCluster ./server -tags=skip_js_cluster_tests,skip_js_cluster_tests_3,skip_js_cluster_tests_4 -count=1 -vet=off -timeout=30m -failfast
 
 elif [ "$1" = "js_cluster_tests_3" ]; then
 
@@ -63,7 +63,16 @@ elif [ "$1" = "js_cluster_tests_3" ]; then
     # excluding others with use of proper tags.
     #
 
-    go test -race -v -run=TestJetStreamCluster ./server -tags=skip_js_cluster_tests,skip_js_cluster_tests_2 -count=1 -vet=off -timeout=30m -failfast
+    go test -race -v -run=TestJetStreamCluster ./server -tags=skip_js_cluster_tests,skip_js_cluster_tests_2,skip_js_cluster_tests_4 -count=1 -vet=off -timeout=30m -failfast
+
+elif [ "$1" = "js_cluster_tests_4" ]; then
+
+    # Run JetStream clustered tests. By convention, all JS cluster tests
+    # start with `TestJetStreamCluster`. Will run the third batch of tests,
+    # excluding others with use of proper tags.
+    #
+
+    go test -race -v -run=TestJetStreamCluster ./server -tags=skip_js_cluster_tests,skip_js_cluster_tests_2,skip_js_cluster_tests_3 -count=1 -vet=off -timeout=30m -failfast
 
 elif [ "$1" = "js_super_cluster_tests" ]; then
 
diff --git a/server/filestore_test.go b/server/filestore_test.go
index bf704559..17bf5833 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -5844,8 +5844,8 @@ func TestFileStoreFullStateMultiBlockPastWAL(t *testing.T) {
 }
 
 // This tests we can successfully recover without having to rebuild the whole stream from a mid block index.db marker
-// when they updated block has a removed entry.
-// TODO(dlc) - This test will force a rebuild atm, leaving here for later.
+// when the updated block has a removed entry.
+// Make sure this does not cause a recover of the full state.
 func TestFileStoreFullStateMidBlockPastWAL(t *testing.T) {
 	testFileStoreAllPermutations(t, func(t *testing.T, fcfg FileStoreConfig) {
 		cfg := StreamConfig{Name: "zzz", Subjects: []string{"*"}, Storage: FileStorage, MaxMsgsPer: 1}
@@ -5900,6 +5900,9 @@ func TestFileStoreFullStateMidBlockPastWAL(t *testing.T) {
 			t.Fatalf("Restore state does not match:\n%+v\n%+v",
 				state, newState)
 		}
+		// Check that index.db is still there. If we recover by raw data on a corrupt state we delete this.
+		_, err = os.Stat(sfile)
+		require_NoError(t, err)
 	})
 }
 
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index f62ff0bf..d0b36c18 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -35,7 +35,6 @@ import (
 
 	"github.com/nats-io/jwt/v2"
 	"github.com/nats-io/nats.go"
-	"github.com/nats-io/nuid"
 )
 
 func TestJetStreamClusterRemovePeerByID(t *testing.T) {
@@ -6353,920 +6352,3 @@ Consume3:
 		t.Errorf("Consumers to same stream are at different sequences: %d vs %d", a, b)
 	}
 }
-
-func TestJetStreamClusterWorkQueueStreamDiscardNewDesync(t *testing.T) {
-	t.Run("max msgs", func(t *testing.T) {
-		testJetStreamClusterWorkQueueStreamDiscardNewDesync(t, &nats.StreamConfig{
-			Name:      "WQTEST_MM",
-			Subjects:  []string{"messages.*"},
-			Replicas:  3,
-			MaxAge:    10 * time.Minute,
-			MaxMsgs:   100,
-			Retention: nats.WorkQueuePolicy,
-			Discard:   nats.DiscardNew,
-		})
-	})
-	t.Run("max bytes", func(t *testing.T) {
-		testJetStreamClusterWorkQueueStreamDiscardNewDesync(t, &nats.StreamConfig{
-			Name:      "WQTEST_MB",
-			Subjects:  []string{"messages.*"},
-			Replicas:  3,
-			MaxAge:    10 * time.Minute,
-			MaxBytes:  1 * 1024 * 1024,
-			Retention: nats.WorkQueuePolicy,
-			Discard:   nats.DiscardNew,
-		})
-	})
-}
-
-func testJetStreamClusterWorkQueueStreamDiscardNewDesync(t *testing.T, sc *nats.StreamConfig) {
-	conf := `
-	listen: 127.0.0.1:-1
-	server_name: %s
-	jetstream: {
-		store_dir: '%s',
-	}
-	cluster {
-		name: %s
-		listen: 127.0.0.1:%d
-		routes = [%s]
-	}
-        system_account: sys
-        no_auth_user: js
-	accounts {
-	  sys {
-	    users = [
-	      { user: sys, pass: sys }
-	    ]
-	  }
-	  js {
-	    jetstream = enabled
-	    users = [
-	      { user: js, pass: js }
-	    ]
-	  }
-	}`
-	c := createJetStreamClusterWithTemplate(t, conf, sc.Name, 3)
-	defer c.shutdown()
-
-	nc, js := jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	cnc, cjs := jsClientConnect(t, c.randomServer())
-	defer cnc.Close()
-
-	_, err := js.AddStream(sc)
-	require_NoError(t, err)
-
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	psub, err := cjs.PullSubscribe("messages.*", "consumer")
-	require_NoError(t, err)
-
-	stepDown := func() {
-		_, err = nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, sc.Name), nil, time.Second)
-	}
-
-	// Messages will be produced and consumed in parallel, then once there are
-	// enough errors a leader election will be triggered.
-	var (
-		wg          sync.WaitGroup
-		received    uint64
-		errCh       = make(chan error, 100_000)
-		receivedMap = make(map[string]*nats.Msg)
-	)
-	wg.Add(1)
-	go func() {
-		tick := time.NewTicker(20 * time.Millisecond)
-		for {
-			select {
-			case <-ctx.Done():
-				wg.Done()
-				return
-			case <-tick.C:
-				msgs, err := psub.Fetch(10, nats.MaxWait(200*time.Millisecond))
-				if err != nil {
-					// The consumer will continue to timeout here eventually.
-					continue
-				}
-				for _, msg := range msgs {
-					received++
-					receivedMap[msg.Subject] = msg
-					msg.Ack()
-				}
-			}
-		}
-	}()
-
-	shouldDrop := make(map[string]error)
-	wg.Add(1)
-	go func() {
-		payload := []byte(strings.Repeat("A", 1024))
-		tick := time.NewTicker(1 * time.Millisecond)
-		for i := 1; ; i++ {
-			select {
-			case <-ctx.Done():
-				wg.Done()
-				return
-			case <-tick.C:
-				subject := fmt.Sprintf("messages.%d", i)
-				_, err := js.Publish(subject, payload, nats.RetryAttempts(0))
-				if err != nil {
-					errCh <- err
-				}
-				// Capture the messages that have failed.
-				if err != nil {
-					shouldDrop[subject] = err
-				}
-			}
-		}
-	}()
-
-	// Collect enough errors to cause things to get out of sync.
-	var errCount int
-Setup:
-	for {
-		select {
-		case err = <-errCh:
-			errCount++
-			if errCount%500 == 0 {
-				stepDown()
-			} else if errCount >= 2000 {
-				// Stop both producing and consuming.
-				cancel()
-				break Setup
-			}
-		case <-time.After(5 * time.Second):
-			// Unblock the test and continue.
-			cancel()
-			break Setup
-		}
-	}
-
-	// Both goroutines should be exiting now..
-	wg.Wait()
-
-	// Let acks propagate for stream checks.
-	time.Sleep(250 * time.Millisecond)
-
-	// Check messages that ought to have been dropped.
-	for subject := range receivedMap {
-		found, ok := shouldDrop[subject]
-		if ok {
-			t.Errorf("Should have dropped message published on %q since got error: %v", subject, found)
-		}
-	}
-}
-
-// https://github.com/nats-io/nats-server/issues/5071
-func TestJetStreamClusterStreamPlacementDistribution(t *testing.T) {
-	c := createJetStreamClusterExplicit(t, "R3S", 5)
-	defer c.shutdown()
-
-	s := c.randomNonLeader()
-	nc, js := jsClientConnect(t, s)
-	defer nc.Close()
-
-	for i := 1; i <= 10; i++ {
-		_, err := js.AddStream(&nats.StreamConfig{
-			Name:     fmt.Sprintf("TEST:%d", i),
-			Subjects: []string{fmt.Sprintf("foo.%d.*", i)},
-			Replicas: 3,
-		})
-		require_NoError(t, err)
-	}
-
-	// 10 streams, 3 replicas div 5 servers.
-	expectedStreams := 10 * 3 / 5
-	for _, s := range c.servers {
-		jsz, err := s.Jsz(nil)
-		require_NoError(t, err)
-		require_Equal(t, jsz.Streams, expectedStreams)
-	}
-}
-
-func TestJetStreamClusterSourceWorkingQueueWithLimit(t *testing.T) {
-	c := createJetStreamClusterExplicit(t, "WQ3", 3)
-	defer c.shutdown()
-
-	nc, js := jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	_, err := js.AddStream(&nats.StreamConfig{Name: "test", Subjects: []string{"test"}, Replicas: 3})
-	require_NoError(t, err)
-
-	_, err = js.AddStream(&nats.StreamConfig{Name: "wq", MaxMsgs: 100, Discard: nats.DiscardNew, Retention: nats.WorkQueuePolicy,
-		Sources: []*nats.StreamSource{{Name: "test"}}, Replicas: 3})
-	require_NoError(t, err)
-
-	sendBatch := func(subject string, n int) {
-		for i := 0; i < n; i++ {
-			_, err = js.Publish(subject, []byte("OK"))
-			require_NoError(t, err)
-		}
-	}
-	// Populate each one.
-	sendBatch("test", 300)
-
-	checkFor(t, 3*time.Second, 250*time.Millisecond, func() error {
-		si, err := js.StreamInfo("wq")
-		require_NoError(t, err)
-		if si.State.Msgs != 100 {
-			return fmt.Errorf("Expected 100 msgs, got state: %+v", si.State)
-		}
-		return nil
-	})
-
-	_, err = js.AddConsumer("wq", &nats.ConsumerConfig{Durable: "wqc", FilterSubject: "test", AckPolicy: nats.AckExplicitPolicy})
-	require_NoError(t, err)
-
-	ss, err := js.PullSubscribe("test", "wqc", nats.Bind("wq", "wqc"))
-	require_NoError(t, err)
-	// we must have at least one message on the transformed subject name (ie no timeout)
-	f := func(done chan bool) {
-		for i := 0; i < 300; i++ {
-			m, err := ss.Fetch(1, nats.MaxWait(3*time.Second))
-			require_NoError(t, err)
-			time.Sleep(11 * time.Millisecond)
-			err = m[0].Ack()
-			require_NoError(t, err)
-		}
-		done <- true
-	}
-
-	var doneChan = make(chan bool)
-	go f(doneChan)
-
-	checkFor(t, 6*time.Second, 100*time.Millisecond, func() error {
-		si, err := js.StreamInfo("wq")
-		require_NoError(t, err)
-		if si.State.Msgs > 0 && si.State.Msgs <= 100 {
-			return fmt.Errorf("Expected 0 msgs, got: %d", si.State.Msgs)
-		} else if si.State.Msgs > 100 {
-			t.Fatalf("Got more than our 100 message limit: %+v", si.State)
-		}
-		return nil
-	})
-
-	select {
-	case <-doneChan:
-		ss.Drain()
-	case <-time.After(5 * time.Second):
-		t.Fatalf("Did not receive completion signal")
-	}
-}
-
-func TestJetStreamClusterStreamOrphanMsgsAndReplicasDrifting(t *testing.T) {
-	type testParams struct {
-		restartAny       bool
-		restartLeader    bool
-		rolloutRestart   bool
-		ldmRestart       bool
-		restarts         int
-		checkHealthz     bool
-		reconnectRoutes  bool
-		reconnectClients bool
-	}
-	test := func(t *testing.T, params *testParams, sc *nats.StreamConfig) {
-		conf := `
-		listen: 127.0.0.1:-1
-		server_name: %s
-		jetstream: {
-			store_dir: '%s',
-		}
-		cluster {
-			name: %s
-			listen: 127.0.0.1:%d
-			routes = [%s]
-		}
-		server_tags: ["test"]
-		system_account: sys
-		no_auth_user: js
-		accounts {
-			sys { users = [ { user: sys, pass: sys } ] }
-			js {
-				jetstream = enabled
-				users = [ { user: js, pass: js } ]
-		    }
-		}`
-		c := createJetStreamClusterWithTemplate(t, conf, sc.Name, 3)
-		defer c.shutdown()
-
-		// Update lame duck duration for all servers.
-		for _, s := range c.servers {
-			s.optsMu.Lock()
-			s.opts.LameDuckDuration = 5 * time.Second
-			s.opts.LameDuckGracePeriod = -5 * time.Second
-			s.optsMu.Unlock()
-		}
-
-		nc, js := jsClientConnect(t, c.randomServer())
-		defer nc.Close()
-
-		cnc, cjs := jsClientConnect(t, c.randomServer())
-		defer cnc.Close()
-
-		_, err := js.AddStream(sc)
-		require_NoError(t, err)
-
-		pctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-		defer cancel()
-
-		// Start producers
-		var wg sync.WaitGroup
-
-		// First call is just to create the pull subscribers.
-		mp := nats.MaxAckPending(10000)
-		mw := nats.PullMaxWaiting(1000)
-		aw := nats.AckWait(5 * time.Second)
-
-		for i := 0; i < 10; i++ {
-			for _, partition := range []string{"EEEEE"} {
-				subject := fmt.Sprintf("MSGS.%s.*.H.100XY.*.*.WQ.00000000000%d", partition, i)
-				consumer := fmt.Sprintf("consumer:%s:%d", partition, i)
-				_, err := cjs.PullSubscribe(subject, consumer, mp, mw, aw)
-				require_NoError(t, err)
-			}
-		}
-
-		// Create a single consumer that does no activity.
-		// Make sure we still calculate low ack properly and cleanup etc.
-		_, err = cjs.PullSubscribe("MSGS.ZZ.>", "consumer:ZZ:0", mp, mw, aw)
-		require_NoError(t, err)
-
-		subjects := []string{
-			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000000",
-			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000001",
-			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000002",
-			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000003",
-			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000004",
-			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000005",
-			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000006",
-			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000007",
-			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000008",
-			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000009",
-		}
-		payload := []byte(strings.Repeat("A", 1024))
-
-		for i := 0; i < 50; i++ {
-			wg.Add(1)
-			go func() {
-				pnc, pjs := jsClientConnect(t, c.randomServer())
-				defer pnc.Close()
-
-				for i := 1; i < 200_000; i++ {
-					select {
-					case <-pctx.Done():
-						wg.Done()
-						return
-					default:
-					}
-					for _, subject := range subjects {
-						// Send each message a few times.
-						msgID := nats.MsgId(nuid.Next())
-						pjs.PublishAsync(subject, payload, msgID)
-						pjs.Publish(subject, payload, msgID, nats.AckWait(250*time.Millisecond))
-						pjs.Publish(subject, payload, msgID, nats.AckWait(250*time.Millisecond))
-					}
-				}
-			}()
-		}
-
-		// Rogue publisher that sends the same msg ID everytime.
-		for i := 0; i < 10; i++ {
-			wg.Add(1)
-			go func() {
-				pnc, pjs := jsClientConnect(t, c.randomServer())
-				defer pnc.Close()
-
-				msgID := nats.MsgId("1234567890")
-				for i := 1; ; i++ {
-					select {
-					case <-pctx.Done():
-						wg.Done()
-						return
-					default:
-					}
-					for _, subject := range subjects {
-						// Send each message a few times.
-						pjs.PublishAsync(subject, payload, msgID, nats.RetryAttempts(0), nats.RetryWait(0))
-						pjs.Publish(subject, payload, msgID, nats.AckWait(1*time.Millisecond), nats.RetryAttempts(0), nats.RetryWait(0))
-						pjs.Publish(subject, payload, msgID, nats.AckWait(1*time.Millisecond), nats.RetryAttempts(0), nats.RetryWait(0))
-					}
-				}
-			}()
-		}
-
-		// Let enough messages into the stream then start consumers.
-		time.Sleep(15 * time.Second)
-
-		ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
-		defer cancel()
-
-		for i := 0; i < 10; i++ {
-			subject := fmt.Sprintf("MSGS.EEEEE.*.H.100XY.*.*.WQ.00000000000%d", i)
-			consumer := fmt.Sprintf("consumer:EEEEE:%d", i)
-			for n := 0; n < 5; n++ {
-				cpnc, cpjs := jsClientConnect(t, c.randomServer())
-				defer cpnc.Close()
-
-				psub, err := cpjs.PullSubscribe(subject, consumer, mp, mw, aw)
-				require_NoError(t, err)
-
-				time.AfterFunc(15*time.Second, func() {
-					cpnc.Close()
-				})
-
-				wg.Add(1)
-				go func() {
-					tick := time.NewTicker(1 * time.Millisecond)
-					for {
-						if cpnc.IsClosed() {
-							wg.Done()
-							return
-						}
-						select {
-						case <-ctx.Done():
-							wg.Done()
-							return
-						case <-tick.C:
-							// Fetch 1 first, then if no errors Fetch 100.
-							msgs, err := psub.Fetch(1, nats.MaxWait(200*time.Millisecond))
-							if err != nil {
-								continue
-							}
-							for _, msg := range msgs {
-								msg.Ack()
-							}
-							msgs, err = psub.Fetch(100, nats.MaxWait(200*time.Millisecond))
-							if err != nil {
-								continue
-							}
-							for _, msg := range msgs {
-								msg.Ack()
-							}
-							msgs, err = psub.Fetch(1000, nats.MaxWait(200*time.Millisecond))
-							if err != nil {
-								continue
-							}
-							for _, msg := range msgs {
-								msg.Ack()
-							}
-						}
-					}
-				}()
-			}
-		}
-
-		for i := 0; i < 10; i++ {
-			subject := fmt.Sprintf("MSGS.EEEEE.*.H.100XY.*.*.WQ.00000000000%d", i)
-			consumer := fmt.Sprintf("consumer:EEEEE:%d", i)
-			for n := 0; n < 10; n++ {
-				cpnc, cpjs := jsClientConnect(t, c.randomServer())
-				defer cpnc.Close()
-
-				psub, err := cpjs.PullSubscribe(subject, consumer, mp, mw, aw)
-				if err != nil {
-					t.Logf("ERROR: %v", err)
-					continue
-				}
-
-				wg.Add(1)
-				go func() {
-					tick := time.NewTicker(1 * time.Millisecond)
-					for {
-						select {
-						case <-ctx.Done():
-							wg.Done()
-							return
-						case <-tick.C:
-							// Fetch 1 first, then if no errors Fetch 100.
-							msgs, err := psub.Fetch(1, nats.MaxWait(200*time.Millisecond))
-							if err != nil {
-								continue
-							}
-							for _, msg := range msgs {
-								msg.Ack()
-							}
-							msgs, err = psub.Fetch(100, nats.MaxWait(200*time.Millisecond))
-							if err != nil {
-								continue
-							}
-							for _, msg := range msgs {
-								msg.Ack()
-							}
-
-							msgs, err = psub.Fetch(1000, nats.MaxWait(200*time.Millisecond))
-							if err != nil {
-								continue
-							}
-							for _, msg := range msgs {
-								msg.Ack()
-							}
-						}
-					}
-				}()
-			}
-		}
-
-		// Periodically disconnect routes from one of the servers.
-		if params.reconnectRoutes {
-			wg.Add(1)
-			go func() {
-				for range time.NewTicker(10 * time.Second).C {
-					select {
-					case <-ctx.Done():
-						wg.Done()
-						return
-					default:
-					}
-
-					// Force disconnecting routes from one of the servers.
-					s := c.servers[rand.Intn(3)]
-					var routes []*client
-					t.Logf("Disconnecting routes from %v", s.Name())
-					s.mu.Lock()
-					for _, conns := range s.routes {
-						routes = append(routes, conns...)
-					}
-					s.mu.Unlock()
-					for _, r := range routes {
-						r.closeConnection(ClientClosed)
-					}
-				}
-			}()
-		}
-
-		// Periodically reconnect clients.
-		if params.reconnectClients {
-			reconnectClients := func(s *Server) {
-				for _, client := range s.clients {
-					client.closeConnection(Kicked)
-				}
-			}
-
-			wg.Add(1)
-			go func() {
-				for range time.NewTicker(10 * time.Second).C {
-					select {
-					case <-ctx.Done():
-						wg.Done()
-						return
-					default:
-					}
-					// Force reconnect clients from one of the servers.
-					s := c.servers[rand.Intn(len(c.servers))]
-					reconnectClients(s)
-				}
-			}()
-		}
-
-		// Restarts
-		time.AfterFunc(10*time.Second, func() {
-			for i := 0; i < params.restarts; i++ {
-				switch {
-				case params.restartLeader:
-					// Find server leader of the stream and restart it.
-					s := c.streamLeader("js", sc.Name)
-					if params.ldmRestart {
-						s.lameDuckMode()
-					} else {
-						s.Shutdown()
-					}
-					s.WaitForShutdown()
-					c.restartServer(s)
-				case params.restartAny:
-					s := c.servers[rand.Intn(len(c.servers))]
-					if params.ldmRestart {
-						s.lameDuckMode()
-					} else {
-						s.Shutdown()
-					}
-					s.WaitForShutdown()
-					c.restartServer(s)
-				case params.rolloutRestart:
-					for _, s := range c.servers {
-						if params.ldmRestart {
-							s.lameDuckMode()
-						} else {
-							s.Shutdown()
-						}
-						s.WaitForShutdown()
-						c.restartServer(s)
-
-						if params.checkHealthz {
-							hctx, hcancel := context.WithTimeout(ctx, 15*time.Second)
-							defer hcancel()
-
-							for range time.NewTicker(2 * time.Second).C {
-								select {
-								case <-hctx.Done():
-								default:
-								}
-
-								status := s.healthz(nil)
-								if status.StatusCode == 200 {
-									break
-								}
-							}
-						}
-					}
-				}
-				c.waitOnClusterReady()
-			}
-		})
-
-		// Wait until context is done then check state.
-		<-ctx.Done()
-
-		var consumerPending int
-		for i := 0; i < 10; i++ {
-			ci, err := js.ConsumerInfo(sc.Name, fmt.Sprintf("consumer:EEEEE:%d", i))
-			require_NoError(t, err)
-			consumerPending += int(ci.NumPending)
-		}
-
-		getStreamDetails := func(t *testing.T, srv *Server) *StreamDetail {
-			t.Helper()
-			jsz, err := srv.Jsz(&JSzOptions{Accounts: true, Streams: true, Consumer: true})
-			require_NoError(t, err)
-			if len(jsz.AccountDetails) > 0 && len(jsz.AccountDetails[0].Streams) > 0 {
-				stream := jsz.AccountDetails[0].Streams[0]
-				return &stream
-			}
-			t.Error("Could not find account details")
-			return nil
-		}
-
-		checkState := func(t *testing.T) error {
-			t.Helper()
-
-			leaderSrv := c.streamLeader("js", sc.Name)
-			if leaderSrv == nil {
-				return fmt.Errorf("no leader found for stream")
-			}
-			streamLeader := getStreamDetails(t, leaderSrv)
-			var errs []error
-			for _, srv := range c.servers {
-				if srv == leaderSrv {
-					// Skip self
-					continue
-				}
-				stream := getStreamDetails(t, srv)
-				if stream == nil {
-					return fmt.Errorf("stream not found")
-				}
-
-				if stream.State.Msgs != streamLeader.State.Msgs {
-					err := fmt.Errorf("Leader %v has %d messages, Follower %v has %d messages",
-						stream.Cluster.Leader, streamLeader.State.Msgs,
-						srv, stream.State.Msgs,
-					)
-					errs = append(errs, err)
-				}
-				if stream.State.FirstSeq != streamLeader.State.FirstSeq {
-					err := fmt.Errorf("Leader %v FirstSeq is %d, Follower %v is at %d",
-						stream.Cluster.Leader, streamLeader.State.FirstSeq,
-						srv, stream.State.FirstSeq,
-					)
-					errs = append(errs, err)
-				}
-				if stream.State.LastSeq != streamLeader.State.LastSeq {
-					err := fmt.Errorf("Leader %v LastSeq is %d, Follower %v is at %d",
-						stream.Cluster.Leader, streamLeader.State.LastSeq,
-						srv, stream.State.LastSeq,
-					)
-					errs = append(errs, err)
-				}
-			}
-			if len(errs) > 0 {
-				return errors.Join(errs...)
-			}
-			return nil
-		}
-
-		checkMsgsEqual := func(t *testing.T) {
-			// These have already been checked to be the same for all streams.
-			state := getStreamDetails(t, c.streamLeader("js", sc.Name)).State
-			// Gather all the streams.
-			var msets []*stream
-			for _, s := range c.servers {
-				acc, err := s.LookupAccount("js")
-				require_NoError(t, err)
-				mset, err := acc.lookupStream(sc.Name)
-				require_NoError(t, err)
-				msets = append(msets, mset)
-			}
-			for seq := state.FirstSeq; seq <= state.LastSeq; seq++ {
-				var msgId string
-				var smv StoreMsg
-				for _, mset := range msets {
-					mset.mu.RLock()
-					sm, err := mset.store.LoadMsg(seq, &smv)
-					mset.mu.RUnlock()
-					require_NoError(t, err)
-					if msgId == _EMPTY_ {
-						msgId = string(sm.hdr)
-					} else if msgId != string(sm.hdr) {
-						t.Fatalf("MsgIds do not match for seq %d: %q vs %q", seq, msgId, sm.hdr)
-					}
-				}
-			}
-		}
-
-		// Check state of streams and consumers.
-		si, err := js.StreamInfo(sc.Name)
-		require_NoError(t, err)
-
-		// Only check if there are any pending messages.
-		if consumerPending > 0 {
-			streamPending := int(si.State.Msgs)
-			if streamPending != consumerPending {
-				t.Errorf("Unexpected number of pending messages, stream=%d, consumers=%d", streamPending, consumerPending)
-			}
-		}
-
-		// If clustered, check whether leader and followers have drifted.
-		if sc.Replicas > 1 {
-			// If we have drifted do not have to wait too long, usually its stuck for good.
-			checkFor(t, time.Minute, time.Second, func() error {
-				return checkState(t)
-			})
-			// If we succeeded now let's check that all messages are also the same.
-			// We may have no messages but for tests that do we make sure each msg is the same
-			// across all replicas.
-			checkMsgsEqual(t)
-		}
-
-		wg.Wait()
-	}
-
-	// Setting up test variations below:
-	//
-	// File based with single replica and discard old policy.
-	t.Run("R1F", func(t *testing.T) {
-		params := &testParams{
-			restartAny:     true,
-			ldmRestart:     false,
-			rolloutRestart: false,
-			restarts:       1,
-		}
-		test(t, params, &nats.StreamConfig{
-			Name:        "OWQTEST_R1F",
-			Subjects:    []string{"MSGS.>"},
-			Replicas:    1,
-			MaxAge:      30 * time.Minute,
-			Duplicates:  5 * time.Minute,
-			Retention:   nats.WorkQueuePolicy,
-			Discard:     nats.DiscardOld,
-			AllowRollup: true,
-			Placement: &nats.Placement{
-				Tags: []string{"test"},
-			},
-		})
-	})
-
-	// Clustered memory based with discard new policy and max msgs limit.
-	t.Run("R3M", func(t *testing.T) {
-		params := &testParams{
-			restartAny:     true,
-			ldmRestart:     true,
-			rolloutRestart: false,
-			restarts:       1,
-			checkHealthz:   false,
-		}
-		test(t, params, &nats.StreamConfig{
-			Name:        "OWQTEST_R3M",
-			Subjects:    []string{"MSGS.>"},
-			Replicas:    3,
-			MaxAge:      30 * time.Minute,
-			MaxMsgs:     100_000,
-			Duplicates:  5 * time.Minute,
-			Retention:   nats.WorkQueuePolicy,
-			Discard:     nats.DiscardNew,
-			AllowRollup: true,
-			Storage:     nats.MemoryStorage,
-			Placement: &nats.Placement{
-				Tags: []string{"test"},
-			},
-		})
-	})
-
-	// Clustered file based with discard new policy and max msgs limit.
-	t.Run("R3F_DN", func(t *testing.T) {
-		params := &testParams{
-			restartAny:     true,
-			ldmRestart:     true,
-			rolloutRestart: false,
-			restarts:       1,
-		}
-		test(t, params, &nats.StreamConfig{
-			Name:        "OWQTEST_R3F_DN",
-			Subjects:    []string{"MSGS.>"},
-			Replicas:    3,
-			MaxAge:      30 * time.Minute,
-			MaxMsgs:     100_000,
-			Duplicates:  5 * time.Minute,
-			Retention:   nats.WorkQueuePolicy,
-			Discard:     nats.DiscardNew,
-			AllowRollup: true,
-			Placement: &nats.Placement{
-				Tags: []string{"test"},
-			},
-		})
-	})
-
-	// Clustered file based with discard old policy and max msgs limit.
-	t.Run("R3F_DO", func(t *testing.T) {
-		params := &testParams{
-			restartAny:     true,
-			ldmRestart:     true,
-			rolloutRestart: false,
-			restarts:       1,
-		}
-		test(t, params, &nats.StreamConfig{
-			Name:        "OWQTEST_R3F_DO",
-			Subjects:    []string{"MSGS.>"},
-			Replicas:    3,
-			MaxAge:      30 * time.Minute,
-			MaxMsgs:     100_000,
-			Duplicates:  5 * time.Minute,
-			Retention:   nats.WorkQueuePolicy,
-			Discard:     nats.DiscardOld,
-			AllowRollup: true,
-			Placement: &nats.Placement{
-				Tags: []string{"test"},
-			},
-		})
-	})
-
-	// Clustered file based with discard old policy and no limits.
-	t.Run("R3F_DO_NOLIMIT", func(t *testing.T) {
-		params := &testParams{
-			restartAny:       false,
-			ldmRestart:       true,
-			rolloutRestart:   true,
-			restarts:         3,
-			checkHealthz:     true,
-			reconnectRoutes:  true,
-			reconnectClients: true,
-		}
-		test(t, params, &nats.StreamConfig{
-			Name:       "OWQTEST_R3F_DO_NOLIMIT",
-			Subjects:   []string{"MSGS.>"},
-			Replicas:   3,
-			Duplicates: 30 * time.Second,
-			Discard:    nats.DiscardOld,
-			Placement: &nats.Placement{
-				Tags: []string{"test"},
-			},
-		})
-	})
-}
-
-func TestJetStreamClusterConsumerNRGCleanup(t *testing.T) {
-	c := createJetStreamClusterExplicit(t, "R3S", 3)
-	defer c.shutdown()
-
-	nc, js := jsClientConnect(t, c.randomServer())
-	defer nc.Close()
-
-	_, err := js.AddStream(&nats.StreamConfig{
-		Name:      "TEST",
-		Subjects:  []string{"foo"},
-		Storage:   nats.MemoryStorage,
-		Retention: nats.WorkQueuePolicy,
-		Replicas:  3,
-	})
-	require_NoError(t, err)
-
-	// First call is just to create the pull subscribers.
-	_, err = js.PullSubscribe("foo", "dlc")
-	require_NoError(t, err)
-
-	require_NoError(t, js.DeleteConsumer("TEST", "dlc"))
-
-	// Now delete the stream.
-	require_NoError(t, js.DeleteStream("TEST"))
-
-	// Now make sure we cleaned up the NRG directories for the stream and consumer.
-	var numConsumers, numStreams int
-	for _, s := range c.servers {
-		sd := s.JetStreamConfig().StoreDir
-		nd := filepath.Join(sd, "$SYS", "_js_")
-		f, err := os.Open(nd)
-		require_NoError(t, err)
-		dirs, err := f.ReadDir(-1)
-		require_NoError(t, err)
-		for _, fi := range dirs {
-			if strings.HasPrefix(fi.Name(), "C-") {
-				numConsumers++
-			} else if strings.HasPrefix(fi.Name(), "S-") {
-				numStreams++
-			}
-		}
-	}
-	require_Equal(t, numConsumers, 0)
-	require_Equal(t, numStreams, 0)
-}
diff --git a/server/jetstream_cluster_4_test.go b/server/jetstream_cluster_4_test.go
new file mode 100644
index 00000000..218562d3
--- /dev/null
+++ b/server/jetstream_cluster_4_test.go
@@ -0,0 +1,950 @@
+// Copyright 2022-2024 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build !skip_js_tests && !skip_js_cluster_tests_4
+// +build !skip_js_tests,!skip_js_cluster_tests_4
+
+package server
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math/rand"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nuid"
+)
+
+func TestJetStreamClusterWorkQueueStreamDiscardNewDesync(t *testing.T) {
+	t.Run("max msgs", func(t *testing.T) {
+		testJetStreamClusterWorkQueueStreamDiscardNewDesync(t, &nats.StreamConfig{
+			Name:      "WQTEST_MM",
+			Subjects:  []string{"messages.*"},
+			Replicas:  3,
+			MaxAge:    10 * time.Minute,
+			MaxMsgs:   100,
+			Retention: nats.WorkQueuePolicy,
+			Discard:   nats.DiscardNew,
+		})
+	})
+	t.Run("max bytes", func(t *testing.T) {
+		testJetStreamClusterWorkQueueStreamDiscardNewDesync(t, &nats.StreamConfig{
+			Name:      "WQTEST_MB",
+			Subjects:  []string{"messages.*"},
+			Replicas:  3,
+			MaxAge:    10 * time.Minute,
+			MaxBytes:  1 * 1024 * 1024,
+			Retention: nats.WorkQueuePolicy,
+			Discard:   nats.DiscardNew,
+		})
+	})
+}
+
+func testJetStreamClusterWorkQueueStreamDiscardNewDesync(t *testing.T, sc *nats.StreamConfig) {
+	conf := `
+	listen: 127.0.0.1:-1
+	server_name: %s
+	jetstream: {
+		store_dir: '%s',
+	}
+	cluster {
+		name: %s
+		listen: 127.0.0.1:%d
+		routes = [%s]
+	}
+        system_account: sys
+        no_auth_user: js
+	accounts {
+	  sys {
+	    users = [
+	      { user: sys, pass: sys }
+	    ]
+	  }
+	  js {
+	    jetstream = enabled
+	    users = [
+	      { user: js, pass: js }
+	    ]
+	  }
+	}`
+	c := createJetStreamClusterWithTemplate(t, conf, sc.Name, 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	cnc, cjs := jsClientConnect(t, c.randomServer())
+	defer cnc.Close()
+
+	_, err := js.AddStream(sc)
+	require_NoError(t, err)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	psub, err := cjs.PullSubscribe("messages.*", "consumer")
+	require_NoError(t, err)
+
+	stepDown := func() {
+		_, err = nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, sc.Name), nil, time.Second)
+	}
+
+	// Messages will be produced and consumed in parallel, then once there are
+	// enough errors a leader election will be triggered.
+	var (
+		wg          sync.WaitGroup
+		received    uint64
+		errCh       = make(chan error, 100_000)
+		receivedMap = make(map[string]*nats.Msg)
+	)
+	wg.Add(1)
+	go func() {
+		tick := time.NewTicker(20 * time.Millisecond)
+		for {
+			select {
+			case <-ctx.Done():
+				wg.Done()
+				return
+			case <-tick.C:
+				msgs, err := psub.Fetch(10, nats.MaxWait(200*time.Millisecond))
+				if err != nil {
+					// The consumer will continue to timeout here eventually.
+					continue
+				}
+				for _, msg := range msgs {
+					received++
+					receivedMap[msg.Subject] = msg
+					msg.Ack()
+				}
+			}
+		}
+	}()
+
+	shouldDrop := make(map[string]error)
+	wg.Add(1)
+	go func() {
+		payload := []byte(strings.Repeat("A", 1024))
+		tick := time.NewTicker(1 * time.Millisecond)
+		for i := 1; ; i++ {
+			select {
+			case <-ctx.Done():
+				wg.Done()
+				return
+			case <-tick.C:
+				subject := fmt.Sprintf("messages.%d", i)
+				_, err := js.Publish(subject, payload, nats.RetryAttempts(0))
+				if err != nil {
+					errCh <- err
+				}
+				// Capture the messages that have failed.
+				if err != nil {
+					shouldDrop[subject] = err
+				}
+			}
+		}
+	}()
+
+	// Collect enough errors to cause things to get out of sync.
+	var errCount int
+Setup:
+	for {
+		select {
+		case err = <-errCh:
+			errCount++
+			if errCount%500 == 0 {
+				stepDown()
+			} else if errCount >= 2000 {
+				// Stop both producing and consuming.
+				cancel()
+				break Setup
+			}
+		case <-time.After(5 * time.Second):
+			// Unblock the test and continue.
+			cancel()
+			break Setup
+		}
+	}
+
+	// Both goroutines should be exiting now..
+	wg.Wait()
+
+	// Let acks propagate for stream checks.
+	time.Sleep(250 * time.Millisecond)
+
+	// Check messages that ought to have been dropped.
+	for subject := range receivedMap {
+		found, ok := shouldDrop[subject]
+		if ok {
+			t.Errorf("Should have dropped message published on %q since got error: %v", subject, found)
+		}
+	}
+}
+
+// https://github.com/nats-io/nats-server/issues/5071
+func TestJetStreamClusterStreamPlacementDistribution(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 5)
+	defer c.shutdown()
+
+	s := c.randomNonLeader()
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	for i := 1; i <= 10; i++ {
+		_, err := js.AddStream(&nats.StreamConfig{
+			Name:     fmt.Sprintf("TEST:%d", i),
+			Subjects: []string{fmt.Sprintf("foo.%d.*", i)},
+			Replicas: 3,
+		})
+		require_NoError(t, err)
+	}
+
+	// 10 streams, 3 replicas div 5 servers.
+	expectedStreams := 10 * 3 / 5
+	for _, s := range c.servers {
+		jsz, err := s.Jsz(nil)
+		require_NoError(t, err)
+		require_Equal(t, jsz.Streams, expectedStreams)
+	}
+}
+
+func TestJetStreamClusterSourceWorkingQueueWithLimit(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "WQ3", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "test", Subjects: []string{"test"}, Replicas: 3})
+	require_NoError(t, err)
+
+	_, err = js.AddStream(&nats.StreamConfig{Name: "wq", MaxMsgs: 100, Discard: nats.DiscardNew, Retention: nats.WorkQueuePolicy,
+		Sources: []*nats.StreamSource{{Name: "test"}}, Replicas: 3})
+	require_NoError(t, err)
+
+	sendBatch := func(subject string, n int) {
+		for i := 0; i < n; i++ {
+			_, err = js.Publish(subject, []byte("OK"))
+			require_NoError(t, err)
+		}
+	}
+	// Populate each one.
+	sendBatch("test", 300)
+
+	checkFor(t, 3*time.Second, 250*time.Millisecond, func() error {
+		si, err := js.StreamInfo("wq")
+		require_NoError(t, err)
+		if si.State.Msgs != 100 {
+			return fmt.Errorf("Expected 100 msgs, got state: %+v", si.State)
+		}
+		return nil
+	})
+
+	_, err = js.AddConsumer("wq", &nats.ConsumerConfig{Durable: "wqc", FilterSubject: "test", AckPolicy: nats.AckExplicitPolicy})
+	require_NoError(t, err)
+
+	ss, err := js.PullSubscribe("test", "wqc", nats.Bind("wq", "wqc"))
+	require_NoError(t, err)
+	// we must have at least one message on the transformed subject name (ie no timeout)
+	f := func(done chan bool) {
+		for i := 0; i < 300; i++ {
+			m, err := ss.Fetch(1, nats.MaxWait(3*time.Second))
+			require_NoError(t, err)
+			time.Sleep(11 * time.Millisecond)
+			err = m[0].Ack()
+			require_NoError(t, err)
+		}
+		done <- true
+	}
+
+	var doneChan = make(chan bool)
+	go f(doneChan)
+
+	checkFor(t, 6*time.Second, 100*time.Millisecond, func() error {
+		si, err := js.StreamInfo("wq")
+		require_NoError(t, err)
+		if si.State.Msgs > 0 && si.State.Msgs <= 100 {
+			return fmt.Errorf("Expected 0 msgs, got: %d", si.State.Msgs)
+		} else if si.State.Msgs > 100 {
+			t.Fatalf("Got more than our 100 message limit: %+v", si.State)
+		}
+		return nil
+	})
+
+	select {
+	case <-doneChan:
+		ss.Drain()
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+}
+
+func TestJetStreamClusterStreamOrphanMsgsAndReplicasDrifting(t *testing.T) {
+	type testParams struct {
+		restartAny       bool
+		restartLeader    bool
+		rolloutRestart   bool
+		ldmRestart       bool
+		restarts         int
+		checkHealthz     bool
+		reconnectRoutes  bool
+		reconnectClients bool
+	}
+	test := func(t *testing.T, params *testParams, sc *nats.StreamConfig) {
+		conf := `
+		listen: 127.0.0.1:-1
+		server_name: %s
+		jetstream: {
+			store_dir: '%s',
+		}
+		cluster {
+			name: %s
+			listen: 127.0.0.1:%d
+			routes = [%s]
+		}
+		server_tags: ["test"]
+		system_account: sys
+		no_auth_user: js
+		accounts {
+			sys { users = [ { user: sys, pass: sys } ] }
+			js {
+				jetstream = enabled
+				users = [ { user: js, pass: js } ]
+		    }
+		}`
+		c := createJetStreamClusterWithTemplate(t, conf, sc.Name, 3)
+		defer c.shutdown()
+
+		// Update lame duck duration for all servers.
+		for _, s := range c.servers {
+			s.optsMu.Lock()
+			s.opts.LameDuckDuration = 5 * time.Second
+			s.opts.LameDuckGracePeriod = -5 * time.Second
+			s.optsMu.Unlock()
+		}
+
+		nc, js := jsClientConnect(t, c.randomServer())
+		defer nc.Close()
+
+		cnc, cjs := jsClientConnect(t, c.randomServer())
+		defer cnc.Close()
+
+		_, err := js.AddStream(sc)
+		require_NoError(t, err)
+
+		pctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+		defer cancel()
+
+		// Start producers
+		var wg sync.WaitGroup
+
+		// First call is just to create the pull subscribers.
+		mp := nats.MaxAckPending(10000)
+		mw := nats.PullMaxWaiting(1000)
+		aw := nats.AckWait(5 * time.Second)
+
+		for i := 0; i < 10; i++ {
+			for _, partition := range []string{"EEEEE"} {
+				subject := fmt.Sprintf("MSGS.%s.*.H.100XY.*.*.WQ.00000000000%d", partition, i)
+				consumer := fmt.Sprintf("consumer:%s:%d", partition, i)
+				_, err := cjs.PullSubscribe(subject, consumer, mp, mw, aw)
+				require_NoError(t, err)
+			}
+		}
+
+		// Create a single consumer that does no activity.
+		// Make sure we still calculate low ack properly and cleanup etc.
+		_, err = cjs.PullSubscribe("MSGS.ZZ.>", "consumer:ZZ:0", mp, mw, aw)
+		require_NoError(t, err)
+
+		subjects := []string{
+			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000000",
+			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000001",
+			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000002",
+			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000003",
+			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000004",
+			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000005",
+			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000006",
+			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000007",
+			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000008",
+			"MSGS.EEEEE.P.H.100XY.1.100Z.WQ.000000000009",
+		}
+		payload := []byte(strings.Repeat("A", 1024))
+
+		for i := 0; i < 50; i++ {
+			wg.Add(1)
+			go func() {
+				pnc, pjs := jsClientConnect(t, c.randomServer())
+				defer pnc.Close()
+
+				for i := 1; i < 200_000; i++ {
+					select {
+					case <-pctx.Done():
+						wg.Done()
+						return
+					default:
+					}
+					for _, subject := range subjects {
+						// Send each message a few times.
+						msgID := nats.MsgId(nuid.Next())
+						pjs.PublishAsync(subject, payload, msgID)
+						pjs.Publish(subject, payload, msgID, nats.AckWait(250*time.Millisecond))
+						pjs.Publish(subject, payload, msgID, nats.AckWait(250*time.Millisecond))
+					}
+				}
+			}()
+		}
+
+		// Rogue publisher that sends the same msg ID everytime.
+		for i := 0; i < 10; i++ {
+			wg.Add(1)
+			go func() {
+				pnc, pjs := jsClientConnect(t, c.randomServer())
+				defer pnc.Close()
+
+				msgID := nats.MsgId("1234567890")
+				for i := 1; ; i++ {
+					select {
+					case <-pctx.Done():
+						wg.Done()
+						return
+					default:
+					}
+					for _, subject := range subjects {
+						// Send each message a few times.
+						pjs.PublishAsync(subject, payload, msgID, nats.RetryAttempts(0), nats.RetryWait(0))
+						pjs.Publish(subject, payload, msgID, nats.AckWait(1*time.Millisecond), nats.RetryAttempts(0), nats.RetryWait(0))
+						pjs.Publish(subject, payload, msgID, nats.AckWait(1*time.Millisecond), nats.RetryAttempts(0), nats.RetryWait(0))
+					}
+				}
+			}()
+		}
+
+		// Let enough messages into the stream then start consumers.
+		time.Sleep(15 * time.Second)
+
+		ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
+		defer cancel()
+
+		for i := 0; i < 10; i++ {
+			subject := fmt.Sprintf("MSGS.EEEEE.*.H.100XY.*.*.WQ.00000000000%d", i)
+			consumer := fmt.Sprintf("consumer:EEEEE:%d", i)
+			for n := 0; n < 5; n++ {
+				cpnc, cpjs := jsClientConnect(t, c.randomServer())
+				defer cpnc.Close()
+
+				psub, err := cpjs.PullSubscribe(subject, consumer, mp, mw, aw)
+				require_NoError(t, err)
+
+				time.AfterFunc(15*time.Second, func() {
+					cpnc.Close()
+				})
+
+				wg.Add(1)
+				go func() {
+					tick := time.NewTicker(1 * time.Millisecond)
+					for {
+						if cpnc.IsClosed() {
+							wg.Done()
+							return
+						}
+						select {
+						case <-ctx.Done():
+							wg.Done()
+							return
+						case <-tick.C:
+							// Fetch 1 first, then if no errors Fetch 100.
+							msgs, err := psub.Fetch(1, nats.MaxWait(200*time.Millisecond))
+							if err != nil {
+								continue
+							}
+							for _, msg := range msgs {
+								msg.Ack()
+							}
+							msgs, err = psub.Fetch(100, nats.MaxWait(200*time.Millisecond))
+							if err != nil {
+								continue
+							}
+							for _, msg := range msgs {
+								msg.Ack()
+							}
+							msgs, err = psub.Fetch(1000, nats.MaxWait(200*time.Millisecond))
+							if err != nil {
+								continue
+							}
+							for _, msg := range msgs {
+								msg.Ack()
+							}
+						}
+					}
+				}()
+			}
+		}
+
+		for i := 0; i < 10; i++ {
+			subject := fmt.Sprintf("MSGS.EEEEE.*.H.100XY.*.*.WQ.00000000000%d", i)
+			consumer := fmt.Sprintf("consumer:EEEEE:%d", i)
+			for n := 0; n < 10; n++ {
+				cpnc, cpjs := jsClientConnect(t, c.randomServer())
+				defer cpnc.Close()
+
+				psub, err := cpjs.PullSubscribe(subject, consumer, mp, mw, aw)
+				if err != nil {
+					t.Logf("ERROR: %v", err)
+					continue
+				}
+
+				wg.Add(1)
+				go func() {
+					tick := time.NewTicker(1 * time.Millisecond)
+					for {
+						select {
+						case <-ctx.Done():
+							wg.Done()
+							return
+						case <-tick.C:
+							// Fetch 1 first, then if no errors Fetch 100.
+							msgs, err := psub.Fetch(1, nats.MaxWait(200*time.Millisecond))
+							if err != nil {
+								continue
+							}
+							for _, msg := range msgs {
+								msg.Ack()
+							}
+							msgs, err = psub.Fetch(100, nats.MaxWait(200*time.Millisecond))
+							if err != nil {
+								continue
+							}
+							for _, msg := range msgs {
+								msg.Ack()
+							}
+
+							msgs, err = psub.Fetch(1000, nats.MaxWait(200*time.Millisecond))
+							if err != nil {
+								continue
+							}
+							for _, msg := range msgs {
+								msg.Ack()
+							}
+						}
+					}
+				}()
+			}
+		}
+
+		// Periodically disconnect routes from one of the servers.
+		if params.reconnectRoutes {
+			wg.Add(1)
+			go func() {
+				for range time.NewTicker(10 * time.Second).C {
+					select {
+					case <-ctx.Done():
+						wg.Done()
+						return
+					default:
+					}
+
+					// Force disconnecting routes from one of the servers.
+					s := c.servers[rand.Intn(3)]
+					var routes []*client
+					t.Logf("Disconnecting routes from %v", s.Name())
+					s.mu.Lock()
+					for _, conns := range s.routes {
+						routes = append(routes, conns...)
+					}
+					s.mu.Unlock()
+					for _, r := range routes {
+						r.closeConnection(ClientClosed)
+					}
+				}
+			}()
+		}
+
+		// Periodically reconnect clients.
+		if params.reconnectClients {
+			reconnectClients := func(s *Server) {
+				for _, client := range s.clients {
+					client.closeConnection(Kicked)
+				}
+			}
+
+			wg.Add(1)
+			go func() {
+				for range time.NewTicker(10 * time.Second).C {
+					select {
+					case <-ctx.Done():
+						wg.Done()
+						return
+					default:
+					}
+					// Force reconnect clients from one of the servers.
+					s := c.servers[rand.Intn(len(c.servers))]
+					reconnectClients(s)
+				}
+			}()
+		}
+
+		// Restarts
+		time.AfterFunc(10*time.Second, func() {
+			for i := 0; i < params.restarts; i++ {
+				switch {
+				case params.restartLeader:
+					// Find server leader of the stream and restart it.
+					s := c.streamLeader("js", sc.Name)
+					if params.ldmRestart {
+						s.lameDuckMode()
+					} else {
+						s.Shutdown()
+					}
+					s.WaitForShutdown()
+					c.restartServer(s)
+				case params.restartAny:
+					s := c.servers[rand.Intn(len(c.servers))]
+					if params.ldmRestart {
+						s.lameDuckMode()
+					} else {
+						s.Shutdown()
+					}
+					s.WaitForShutdown()
+					c.restartServer(s)
+				case params.rolloutRestart:
+					for _, s := range c.servers {
+						if params.ldmRestart {
+							s.lameDuckMode()
+						} else {
+							s.Shutdown()
+						}
+						s.WaitForShutdown()
+						c.restartServer(s)
+
+						if params.checkHealthz {
+							hctx, hcancel := context.WithTimeout(ctx, 15*time.Second)
+							defer hcancel()
+
+							for range time.NewTicker(2 * time.Second).C {
+								select {
+								case <-hctx.Done():
+								default:
+								}
+
+								status := s.healthz(nil)
+								if status.StatusCode == 200 {
+									break
+								}
+							}
+						}
+					}
+				}
+				c.waitOnClusterReady()
+			}
+		})
+
+		// Wait until context is done then check state.
+		<-ctx.Done()
+
+		var consumerPending int
+		for i := 0; i < 10; i++ {
+			ci, err := js.ConsumerInfo(sc.Name, fmt.Sprintf("consumer:EEEEE:%d", i))
+			require_NoError(t, err)
+			consumerPending += int(ci.NumPending)
+		}
+
+		getStreamDetails := func(t *testing.T, srv *Server) *StreamDetail {
+			t.Helper()
+			jsz, err := srv.Jsz(&JSzOptions{Accounts: true, Streams: true, Consumer: true})
+			require_NoError(t, err)
+			if len(jsz.AccountDetails) > 0 && len(jsz.AccountDetails[0].Streams) > 0 {
+				stream := jsz.AccountDetails[0].Streams[0]
+				return &stream
+			}
+			t.Error("Could not find account details")
+			return nil
+		}
+
+		checkState := func(t *testing.T) error {
+			t.Helper()
+
+			leaderSrv := c.streamLeader("js", sc.Name)
+			if leaderSrv == nil {
+				return fmt.Errorf("no leader found for stream")
+			}
+			streamLeader := getStreamDetails(t, leaderSrv)
+			var errs []error
+			for _, srv := range c.servers {
+				if srv == leaderSrv {
+					// Skip self
+					continue
+				}
+				stream := getStreamDetails(t, srv)
+				if stream == nil {
+					return fmt.Errorf("stream not found")
+				}
+
+				if stream.State.Msgs != streamLeader.State.Msgs {
+					err := fmt.Errorf("Leader %v has %d messages, Follower %v has %d messages",
+						stream.Cluster.Leader, streamLeader.State.Msgs,
+						srv, stream.State.Msgs,
+					)
+					errs = append(errs, err)
+				}
+				if stream.State.FirstSeq != streamLeader.State.FirstSeq {
+					err := fmt.Errorf("Leader %v FirstSeq is %d, Follower %v is at %d",
+						stream.Cluster.Leader, streamLeader.State.FirstSeq,
+						srv, stream.State.FirstSeq,
+					)
+					errs = append(errs, err)
+				}
+				if stream.State.LastSeq != streamLeader.State.LastSeq {
+					err := fmt.Errorf("Leader %v LastSeq is %d, Follower %v is at %d",
+						stream.Cluster.Leader, streamLeader.State.LastSeq,
+						srv, stream.State.LastSeq,
+					)
+					errs = append(errs, err)
+				}
+			}
+			if len(errs) > 0 {
+				return errors.Join(errs...)
+			}
+			return nil
+		}
+
+		checkMsgsEqual := func(t *testing.T) {
+			// These have already been checked to be the same for all streams.
+			state := getStreamDetails(t, c.streamLeader("js", sc.Name)).State
+			// Gather all the streams.
+			var msets []*stream
+			for _, s := range c.servers {
+				acc, err := s.LookupAccount("js")
+				require_NoError(t, err)
+				mset, err := acc.lookupStream(sc.Name)
+				require_NoError(t, err)
+				msets = append(msets, mset)
+			}
+			for seq := state.FirstSeq; seq <= state.LastSeq; seq++ {
+				var msgId string
+				var smv StoreMsg
+				for _, mset := range msets {
+					mset.mu.RLock()
+					sm, err := mset.store.LoadMsg(seq, &smv)
+					mset.mu.RUnlock()
+					require_NoError(t, err)
+					if msgId == _EMPTY_ {
+						msgId = string(sm.hdr)
+					} else if msgId != string(sm.hdr) {
+						t.Fatalf("MsgIds do not match for seq %d: %q vs %q", seq, msgId, sm.hdr)
+					}
+				}
+			}
+		}
+
+		// Check state of streams and consumers.
+		si, err := js.StreamInfo(sc.Name)
+		require_NoError(t, err)
+
+		// Only check if there are any pending messages.
+		if consumerPending > 0 {
+			streamPending := int(si.State.Msgs)
+			if streamPending != consumerPending {
+				t.Errorf("Unexpected number of pending messages, stream=%d, consumers=%d", streamPending, consumerPending)
+			}
+		}
+
+		// If clustered, check whether leader and followers have drifted.
+		if sc.Replicas > 1 {
+			// If we have drifted do not have to wait too long, usually its stuck for good.
+			checkFor(t, time.Minute, time.Second, func() error {
+				return checkState(t)
+			})
+			// If we succeeded now let's check that all messages are also the same.
+			// We may have no messages but for tests that do we make sure each msg is the same
+			// across all replicas.
+			checkMsgsEqual(t)
+		}
+
+		wg.Wait()
+	}
+
+	// Setting up test variations below:
+	//
+	// File based with single replica and discard old policy.
+	t.Run("R1F", func(t *testing.T) {
+		params := &testParams{
+			restartAny:     true,
+			ldmRestart:     false,
+			rolloutRestart: false,
+			restarts:       1,
+		}
+		test(t, params, &nats.StreamConfig{
+			Name:        "OWQTEST_R1F",
+			Subjects:    []string{"MSGS.>"},
+			Replicas:    1,
+			MaxAge:      30 * time.Minute,
+			Duplicates:  5 * time.Minute,
+			Retention:   nats.WorkQueuePolicy,
+			Discard:     nats.DiscardOld,
+			AllowRollup: true,
+			Placement: &nats.Placement{
+				Tags: []string{"test"},
+			},
+		})
+	})
+
+	// Clustered memory based with discard new policy and max msgs limit.
+	t.Run("R3M", func(t *testing.T) {
+		params := &testParams{
+			restartAny:     true,
+			ldmRestart:     true,
+			rolloutRestart: false,
+			restarts:       1,
+			checkHealthz:   false,
+		}
+		test(t, params, &nats.StreamConfig{
+			Name:        "OWQTEST_R3M",
+			Subjects:    []string{"MSGS.>"},
+			Replicas:    3,
+			MaxAge:      30 * time.Minute,
+			MaxMsgs:     100_000,
+			Duplicates:  5 * time.Minute,
+			Retention:   nats.WorkQueuePolicy,
+			Discard:     nats.DiscardNew,
+			AllowRollup: true,
+			Storage:     nats.MemoryStorage,
+			Placement: &nats.Placement{
+				Tags: []string{"test"},
+			},
+		})
+	})
+
+	// Clustered file based with discard new policy and max msgs limit.
+	t.Run("R3F_DN", func(t *testing.T) {
+		params := &testParams{
+			restartAny:     true,
+			ldmRestart:     true,
+			rolloutRestart: false,
+			restarts:       1,
+		}
+		test(t, params, &nats.StreamConfig{
+			Name:        "OWQTEST_R3F_DN",
+			Subjects:    []string{"MSGS.>"},
+			Replicas:    3,
+			MaxAge:      30 * time.Minute,
+			MaxMsgs:     100_000,
+			Duplicates:  5 * time.Minute,
+			Retention:   nats.WorkQueuePolicy,
+			Discard:     nats.DiscardNew,
+			AllowRollup: true,
+			Placement: &nats.Placement{
+				Tags: []string{"test"},
+			},
+		})
+	})
+
+	// Clustered file based with discard old policy and max msgs limit.
+	t.Run("R3F_DO", func(t *testing.T) {
+		params := &testParams{
+			restartAny:     true,
+			ldmRestart:     true,
+			rolloutRestart: false,
+			restarts:       1,
+		}
+		test(t, params, &nats.StreamConfig{
+			Name:        "OWQTEST_R3F_DO",
+			Subjects:    []string{"MSGS.>"},
+			Replicas:    3,
+			MaxAge:      30 * time.Minute,
+			MaxMsgs:     100_000,
+			Duplicates:  5 * time.Minute,
+			Retention:   nats.WorkQueuePolicy,
+			Discard:     nats.DiscardOld,
+			AllowRollup: true,
+			Placement: &nats.Placement{
+				Tags: []string{"test"},
+			},
+		})
+	})
+
+	// Clustered file based with discard old policy and no limits.
+	t.Run("R3F_DO_NOLIMIT", func(t *testing.T) {
+		params := &testParams{
+			restartAny:       false,
+			ldmRestart:       true,
+			rolloutRestart:   true,
+			restarts:         3,
+			checkHealthz:     true,
+			reconnectRoutes:  true,
+			reconnectClients: true,
+		}
+		test(t, params, &nats.StreamConfig{
+			Name:       "OWQTEST_R3F_DO_NOLIMIT",
+			Subjects:   []string{"MSGS.>"},
+			Replicas:   3,
+			Duplicates: 30 * time.Second,
+			Discard:    nats.DiscardOld,
+			Placement: &nats.Placement{
+				Tags: []string{"test"},
+			},
+		})
+	})
+}
+
+func TestJetStreamClusterConsumerNRGCleanup(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "TEST",
+		Subjects:  []string{"foo"},
+		Storage:   nats.MemoryStorage,
+		Retention: nats.WorkQueuePolicy,
+		Replicas:  3,
+	})
+	require_NoError(t, err)
+
+	// First call is just to create the pull subscribers.
+	_, err = js.PullSubscribe("foo", "dlc")
+	require_NoError(t, err)
+
+	require_NoError(t, js.DeleteConsumer("TEST", "dlc"))
+
+	// Now delete the stream.
+	require_NoError(t, js.DeleteStream("TEST"))
+
+	// Now make sure we cleaned up the NRG directories for the stream and consumer.
+	var numConsumers, numStreams int
+	for _, s := range c.servers {
+		sd := s.JetStreamConfig().StoreDir
+		nd := filepath.Join(sd, "$SYS", "_js_")
+		f, err := os.Open(nd)
+		require_NoError(t, err)
+		dirs, err := f.ReadDir(-1)
+		require_NoError(t, err)
+		for _, fi := range dirs {
+			if strings.HasPrefix(fi.Name(), "C-") {
+				numConsumers++
+			} else if strings.HasPrefix(fi.Name(), "S-") {
+				numStreams++
+			}
+		}
+	}
+	require_Equal(t, numConsumers, 0)
+	require_Equal(t, numStreams, 0)
+}
diff --git a/server/jetstream_consumer_test.go b/server/jetstream_consumer_test.go
index a7d02200..a908783e 100644
--- a/server/jetstream_consumer_test.go
+++ b/server/jetstream_consumer_test.go
@@ -1079,7 +1079,7 @@ func TestJetStreamConsumerDelete(t *testing.T) {
 	}
 }
 
-func TestFetchWithDrain(t *testing.T) {
+func TestJetStreamConsumerFetchWithDrain(t *testing.T) {
 	s := RunBasicJetStreamServer(t)
 	defer s.Shutdown()
 
@@ -1116,27 +1116,29 @@ func TestFetchWithDrain(t *testing.T) {
 	msgs := make(map[int]int)
 
 	processMsg := func(t *testing.T, sub *nats.Subscription, msgs map[int]int) bool {
-		msg, err := sub.NextMsg(time.Second * 1)
+		msg, err := sub.NextMsg(time.Second)
 		if err != nil {
 			return false
 		}
 		metadata, err := msg.Metadata()
 		require_NoError(t, err)
-		err = msg.Ack()
-		require_NoError(t, err)
+		require_NoError(t, msg.Ack())
 
 		v, err := strconv.Atoi(string(msg.Data))
 		require_NoError(t, err)
 		require_Equal(t, uint64(v), metadata.Sequence.Stream)
 
-		_, ok := msgs[int(metadata.Sequence.Stream)]
 		if _, ok := msgs[int(metadata.Sequence.Stream-1)]; !ok && len(msgs) > 0 {
 			t.Logf("Stream Sequence gap detected: current %d", metadata.Sequence.Stream)
 		}
-		if ok {
-			t.Fatalf("Message has been seen before")
+		if _, ok := msgs[int(metadata.Sequence.Stream)]; ok {
+			t.Fatalf("Message for seq %d has been seen before", metadata.Sequence.Stream)
+		}
+		// We do not expect official redeliveries here so this should always be 1.
+		if metadata.NumDelivered != 1 {
+			t.Fatalf("Expected NumDelivered of 1, got %d for seq %d",
+				metadata.NumDelivered, metadata.Sequence.Stream)
 		}
-
 		msgs[int(metadata.Sequence.Stream)] = int(metadata.NumDelivered)
 
 		require_NoError(t, err)
@@ -1166,6 +1168,50 @@ func TestFetchWithDrain(t *testing.T) {
 	}
 }
 
+func TestJetStreamConsumerLongSubjectHang(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	readSubj := "a1."
+	purgeSubj := "a2."
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:        "TEST",
+		Subjects:    []string{readSubj + ">", purgeSubj + ">"},
+		AllowRollup: true,
+	})
+	require_NoError(t, err)
+
+	prefix := strings.Repeat("a", 22)
+	for i := 0; i < 2; i++ {
+		subj := readSubj + prefix + fmt.Sprintf("%d", i)
+		_, err = js.Publish(subj, []byte("hello"))
+		require_NoError(t, err)
+		chunkSubj := purgeSubj + fmt.Sprintf("%d", i)
+		_, err = js.Publish(chunkSubj, []byte("contents"))
+		require_NoError(t, err)
+	}
+	err = js.PurgeStream("TEST", &nats.StreamPurgeRequest{Subject: purgeSubj + ">"})
+	require_NoError(t, err)
+
+	si, err := js.StreamInfo("TEST")
+	require_NoError(t, err)
+	// we should have 2 msgs left after purge
+	require_Equal(t, si.State.Msgs, 2)
+
+	sub, err := js.SubscribeSync(readSubj+">", nats.OrderedConsumer(), nats.DeliverLastPerSubject())
+	require_NoError(t, err)
+	defer sub.Unsubscribe()
+
+	for i := 0; i < 2; i++ {
+		m, err := sub.NextMsg(500 * time.Millisecond)
+		require_NoError(t, err)
+		require_True(t, string(m.Data) == "hello")
+	}
+}
+
 func Benchmark____JetStreamConsumerIsFilteredMatch(b *testing.B) {
 	subject := "foo.bar.do.not.match.any.filter.subject"
 	for n := 1; n <= 1024; n *= 2 {
diff --git a/server/norace_test.go b/server/norace_test.go
index 2e6edef4..81cb2325 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -31,6 +31,7 @@ import (
 	"net"
 	"net/http"
 	"net/url"
+	"os"
 	"path/filepath"
 	"reflect"
 	"runtime"
@@ -10349,3 +10350,59 @@ func TestNoRaceWQAndMultiSubjectFiltersRace(t *testing.T) {
 		return nil
 	})
 }
+
+func TestNoRaceFileStoreWriteFullStateUniqueSubjects(t *testing.T) {
+	fcfg := FileStoreConfig{StoreDir: t.TempDir()}
+	fs, err := newFileStore(fcfg,
+		StreamConfig{Name: "zzz", Subjects: []string{"records.>"}, Storage: FileStorage, MaxMsgsPer: 1, MaxBytes: 15 * 1024 * 1024 * 1024})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	qch := make(chan struct{})
+	defer close(qch)
+
+	go func() {
+		const numThreshold = 1_000_000
+		tick := time.NewTicker(1 * time.Second)
+		for {
+			select {
+			case <-qch:
+				return
+			case <-tick.C:
+				err := fs.writeFullState()
+				var state StreamState
+				fs.FastState(&state)
+				if state.Msgs > numThreshold && err != nil {
+					require_Error(t, err, errStateTooBig)
+				}
+			}
+		}
+	}()
+
+	labels := []string{"AAAAA", "BBBB", "CCCC", "DD", "EEEEE"}
+	msg := []byte(strings.Repeat("Z", 128))
+
+	for i := 0; i < 100; i++ {
+		partA := nuid.Next()
+		for j := 0; j < 100; j++ {
+			partB := nuid.Next()
+			for k := 0; k < 500; k++ {
+				partC := nuid.Next()
+				partD := labels[rand.Intn(len(labels)-1)]
+				subject := fmt.Sprintf("records.%s.%s.%s.%s.%s", partA, partB, partC, partD, nuid.Next())
+				start := time.Now()
+				fs.StoreMsg(subject, nil, msg)
+				elapsed := time.Since(start)
+				if elapsed > 500*time.Millisecond {
+					t.Fatalf("Slow store for %q: %v\n", subject, elapsed)
+				}
+			}
+		}
+	}
+	// Make sure we do write the full state on stop.
+	fs.Stop()
+	fi, err := os.Stat(filepath.Join(fcfg.StoreDir, msgDir, streamStreamStateFile))
+	require_NoError(t, err)
+	// ~500MB, could change if we tweak encodings..
+	require_True(t, fi.Size() > 500*1024*1024)
+}
diff --git a/server/raft_helpers_test.go b/server/raft_helpers_test.go
index e7a7ef88..f8f12eef 100644
--- a/server/raft_helpers_test.go
+++ b/server/raft_helpers_test.go
@@ -83,6 +83,20 @@ func (sg smGroup) nonLeader() stateMachine {
 	return nil
 }
 
+// Take out the lock on all nodes.
+func (sg smGroup) lockAll() {
+	for _, sm := range sg {
+		sm.node().(*raft).Lock()
+	}
+}
+
+// Release the lock on all nodes.
+func (sg smGroup) unlockAll() {
+	for _, sm := range sg {
+		sm.node().(*raft).Unlock()
+	}
+}
+
 // Create a raft group and place on numMembers servers at random.
 func (c *cluster) createRaftGroup(name string, numMembers int, smf smFactory) smGroup {
 	c.t.Helper()
@@ -157,6 +171,7 @@ type stateAdder struct {
 	n   RaftNode
 	cfg *RaftConfig
 	sum int64
+	lch chan bool
 }
 
 // Simple getters for server and the raft node.
@@ -196,7 +211,12 @@ func (a *stateAdder) applyEntry(ce *CommittedEntry) {
 	a.n.Applied(ce.Index)
 }
 
-func (a *stateAdder) leaderChange(isLeader bool) {}
+func (a *stateAdder) leaderChange(isLeader bool) {
+	select {
+	case a.lch <- isLeader:
+	default:
+	}
+}
 
 // Adder specific to change the total.
 func (a *stateAdder) proposeDelta(delta int64) {
@@ -272,5 +292,5 @@ func (rg smGroup) waitOnTotal(t *testing.T, expected int64) {
 
 // Factory function.
 func newStateAdder(s *Server, cfg *RaftConfig, n RaftNode) stateMachine {
-	return &stateAdder{s: s, n: n, cfg: cfg}
+	return &stateAdder{s: s, n: n, cfg: cfg, lch: make(chan bool, 1)}
 }
diff --git a/server/raft_test.go b/server/raft_test.go
index 5a8d0b86..75addc15 100644
--- a/server/raft_test.go
+++ b/server/raft_test.go
@@ -291,3 +291,117 @@ func TestNRGSimpleElection(t *testing.T) {
 		require_Equal(t, rn.vote, vr.candidate)
 	}
 }
+
+func TestNRGStepDownOnSameTermDoesntClearVote(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()
+
+	lsm := rg.leader().(*stateAdder)
+	leader := lsm.node().(*raft)
+	follower := rg.nonLeader().node().(*raft)
+
+	// Make sure we handle the leader change notification from above.
+	require_ChanRead(t, lsm.lch, time.Second)
+
+	// Subscribe to the append entry subject.
+	sub, err := nc.SubscribeSync(leader.asubj)
+	require_NoError(t, err)
+
+	// Get the first append entry that we receive.
+	msg, err := sub.NextMsg(time.Second)
+	require_NoError(t, err)
+	require_NoError(t, sub.Unsubscribe())
+
+	// We're going to modify the append entry that we received so that
+	// we can send it again with modifications.
+	ae, err := leader.decodeAppendEntry(msg.Data, nil, msg.Reply)
+	require_NoError(t, err)
+
+	// First of all we're going to try sending an append entry that
+	// has an old term and a fake leader.
+	msg.Reply = follower.areply
+	ae.leader = follower.id
+	ae.term = leader.term - 1
+	msg.Data, err = ae.encode(msg.Data[:0])
+	require_NoError(t, err)
+	require_NoError(t, nc.PublishMsg(msg))
+
+	// Because the term was old, the fake leader shouldn't matter as
+	// the current leader should ignore it.
+	require_NoChanRead(t, lsm.lch, time.Second)
+
+	// Now we're going to send it on the same term that the current leader
+	// is on. What we expect to happen is that the leader will step down
+	// but it *shouldn't* clear the vote.
+	ae.term = leader.term
+	msg.Data, err = ae.encode(msg.Data[:0])
+	require_NoError(t, err)
+	require_NoError(t, nc.PublishMsg(msg))
+
+	// Wait for the leader transition and ensure that the vote wasn't
+	// cleared.
+	require_ChanRead(t, lsm.lch, time.Second)
+	require_NotEqual(t, leader.vote, noVote)
+}
+
+func TestNRGUnsuccessfulVoteRequestDoesntResetElectionTimer(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()
+	leader := rg.leader().node().(*raft)
+	follower := rg.nonLeader().node().(*raft)
+
+	// Set up a new inbox for the vote responses to go to.
+	vsubj, vreply := leader.vsubj, nc.NewInbox()
+	ch := make(chan *nats.Msg, 3)
+	_, err := nc.ChanSubscribe(vreply, ch)
+	require_NoError(t, err)
+
+	// Keep a track of the last time the election timer was reset before this.
+	// Also build up a vote request that's obviously behind so that the other
+	// nodes should not do anything with it. All locks are taken at the same
+	// time so that it guarantees that both the leader and the follower aren't
+	// operating at the time we take the etlr snapshots.
+	rg.lockAll()
+	leaderOriginal := leader.etlr
+	followerOriginal := follower.etlr
+	vr := &voteRequest{
+		term:      follower.term,
+		lastTerm:  follower.term - 1,
+		lastIndex: 0,
+		candidate: follower.id,
+	}
+	rg.unlockAll()
+
+	// Now send a vote request that's obviously behind.
+	require_NoError(t, nc.PublishMsg(&nats.Msg{
+		Subject: vsubj,
+		Reply:   vreply,
+		Data:    vr.encode(),
+	}))
+
+	// Wait for everyone to respond.
+	require_ChanRead(t, ch, time.Second)
+	require_ChanRead(t, ch, time.Second)
+	require_ChanRead(t, ch, time.Second)
+
+	// Neither the leader nor our chosen follower should have updated their
+	// election timer as a result of this.
+	rg.lockAll()
+	defer rg.unlockAll()
+	require_True(t, leaderOriginal.Equal(leader.etlr))
+	require_True(t, followerOriginal.Equal(follower.etlr))
+}
diff --git a/server/stree/stree_test.go b/server/stree/stree_test.go
index 1dcc741b..7421bcf6 100644
--- a/server/stree/stree_test.go
+++ b/server/stree/stree_test.go
@@ -22,7 +22,6 @@ import (
 	"strings"
 	"testing"
 	"time"
-	"unsafe"
 )
 
 // Print Results: go test -v  --args --results
@@ -363,24 +362,6 @@ func TestSubjectTreePartialTerminalWildcardBugMatch(t *testing.T) {
 	match(t, st, "STATE.GLOBAL.CELL1.7PDSGAALXNN000010.*", 3)
 }
 
-func TestSubjectTreeMaxPrefix(t *testing.T) {
-	st := NewSubjectTree[int]()
-	longPre := strings.Repeat("Z", maxPrefixLen+2)
-	for i := 0; i < 2; i++ {
-		subj := b(fmt.Sprintf("%s.%c", longPre, 'A'+i))
-		old, updated := st.Insert(subj, 22)
-		require_True(t, old == nil)
-		require_False(t, updated)
-	}
-	v, found := st.Find(b(fmt.Sprintf("%s.B", longPre)))
-	require_True(t, found)
-	require_Equal(t, *v, 22)
-	v, found = st.Delete(b(fmt.Sprintf("%s.A", longPre)))
-	require_True(t, found)
-	require_Equal(t, *v, 22)
-	require_Equal(t, st.root.numChildren(), 1)
-}
-
 func TestSubjectTreeMatchSubjectParam(t *testing.T) {
 	st := NewSubjectTree[int]()
 	st.Insert(b("foo.bar.A"), 1)
@@ -602,9 +583,24 @@ func TestSubjectTreeRandomTrackEntries(t *testing.T) {
 	}
 }
 
-func TestSubjectTreeMetaSize(t *testing.T) {
-	var base meta
-	require_Equal(t, unsafe.Sizeof(base), 28)
+// Needs to be longer then internal node prefix, which currently is 24.
+func TestSubjectTreeLongTokens(t *testing.T) {
+	st := NewSubjectTree[int]()
+	st.Insert(b("a1.aaaaaaaaaaaaaaaaaaaaaa0"), 1)
+	st.Insert(b("a2.0"), 2)
+	st.Insert(b("a1.aaaaaaaaaaaaaaaaaaaaaa1"), 3)
+	st.Insert(b("a2.1"), 4)
+	// Simulate purge of a2.>
+	// This required to show bug.
+	st.Delete(b("a2.0"))
+	st.Delete(b("a2.1"))
+	require_Equal(t, st.Size(), 2)
+	v, found := st.Find(b("a1.aaaaaaaaaaaaaaaaaaaaaa0"))
+	require_True(t, found)
+	require_Equal(t, *v, 1)
+	v, found = st.Find(b("a1.aaaaaaaaaaaaaaaaaaaaaa1"))
+	require_True(t, found)
+	require_Equal(t, *v, 3)
 }
 
 func b(s string) []byte {
diff --git a/server/sublist_test.go b/server/sublist_test.go
index 1e2e0288..16444b92 100644
--- a/server/sublist_test.go
+++ b/server/sublist_test.go
@@ -1602,6 +1602,28 @@ func TestMain(m *testing.M) {
 	os.Exit(m.Run())
 }
 
+func TestSublistHasInterest(t *testing.T) {
+	sl := NewSublistWithCache()
+	fooSub := newSub("foo")
+	sl.Insert(fooSub)
+
+	// Expect to find that "foo" matches but "bar" doesn't.
+	// At this point nothing should be in the cache.
+	require_True(t, sl.HasInterest("foo"))
+	require_False(t, sl.HasInterest("bar"))
+	require_Equal(t, sl.cacheHits, 0)
+
+	// Now call Match(), which will populate the cache.
+	sl.Match("foo")
+	require_Equal(t, sl.cacheHits, 0)
+
+	// Future calls to HasInterest() should hit the cache now.
+	for i := uint64(1); i <= 5; i++ {
+		require_True(t, sl.HasInterest("foo"))
+		require_Equal(t, sl.cacheHits, i)
+	}
+}
+
 func subsInit(pre string, toks []string) {
 	var sub string
 	for _, t := range toks {
diff --git a/server/test_test.go b/server/test_test.go
index 4af8d0e8..58af5d76 100644
--- a/server/test_test.go
+++ b/server/test_test.go
@@ -122,7 +122,7 @@ func require_Equal[T comparable](t *testing.T, a, b T) {
 func require_NotEqual[T comparable](t *testing.T, a, b T) {
 	t.Helper()
 	if a == b {
-		t.Fatalf("require %T not equal, but got: %v != %v", a, a, b)
+		t.Fatalf("require %T not equal, but got: %v == %v", a, a, b)
 	}
 }
 
@@ -151,6 +151,15 @@ func require_ChanRead[T any](t *testing.T, ch chan T, timeout time.Duration) T {
 	panic("this shouldn't be possible")
 }
 
+func require_NoChanRead[T any](t *testing.T, ch chan T, timeout time.Duration) {
+	t.Helper()
+	select {
+	case <-ch:
+		t.Fatalf("require no read from channel within %v but got something", timeout)
+	case <-time.After(timeout):
+	}
+}
+
 func checkNatsError(t *testing.T, e *ApiError, id ErrorIdentifier) {
 	t.Helper()
 	ae, ok := ApiErrors[id]