Repository: nats-io/nats-server
Pull request #6345 (2025-01-08 17:59:10)

Cherry-picks for 2.10.25-RC.1
#### Dependencies
- https://github.com/nats-io/nats-server/pull/6323
- https://github.com/nats-io/nats-server/pull/6324

####  Leafnode
- https://github.com/nats-io/nats-server/pull/6291

#### JetStream
- https://github.com/nats-io/nats-server/pull/6226
- https://github.com/nats-io/nats-server/pull/6235
- https://github.com/nats-io/nats-server/pull/6277
- https://github.com/nats-io/nats-server/pull/6279
- https://github.com/nats-io/nats-server/pull/6283
- https://github.com/nats-io/nats-server/pull/6289
- https://github.com/nats-io/nats-server/pull/6316
- https://github.com/nats-io/nats-server/pull/6317
- https://github.com/nats-io/nats-server/pull/6325
- https://github.com/nats-io/nats-server/pull/6326
- https://github.com/nats-io/nats-server/pull/6335
- https://github.com/nats-io/nats-server/pull/6338
- https://github.com/nats-io/nats-server/pull/6341
- https://github.com/nats-io/nats-server/pull/6344
- https://github.com/nats-io/nats-server/pull/6150
- https://github.com/nats-io/nats-server/pull/6351
- https://github.com/nats-io/nats-server/pull/6355

#### Tests
- https://github.com/nats-io/nats-server/pull/6278
- https://github.com/nats-io/nats-server/pull/6297
- https://github.com/nats-io/nats-server/pull/6300
- https://github.com/nats-io/nats-server/pull/6343
- https://github.com/nats-io/nats-server/pull/6329
- https://github.com/nats-io/nats-server/pull/6330
- https://github.com/nats-io/nats-server/pull/6331
- https://github.com/nats-io/nats-server/pull/6332
- https://github.com/nats-io/nats-server/pull/6334
- https://github.com/nats-io/nats-server/pull/6356


Patch:
diff --git a/go.mod b/go.mod
index 6ec47234..a056e0aa 100644
--- a/go.mod
+++ b/go.mod
@@ -13,6 +13,6 @@ require (
 	github.com/nats-io/nuid v1.0.1
 	go.uber.org/automaxprocs v1.6.0
 	golang.org/x/crypto v0.31.0
-	golang.org/x/sys v0.28.0
-	golang.org/x/time v0.8.0
+	golang.org/x/sys v0.29.0
+	golang.org/x/time v0.9.0
 )
diff --git a/go.sum b/go.sum
index 2406dda7..b74f4993 100644
--- a/go.sum
+++ b/go.sum
@@ -23,9 +23,9 @@ go.uber.org/automaxprocs v1.6.0/go.mod h1:ifeIMSnPZuznNm6jmdzmU3/bfk01Fe2fotchwE
 golang.org/x/crypto v0.31.0 h1:ihbySMvVjLAeSH1IbfcRTkD/iNscyz8rGzjF/E5hV6U=
 golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk=
 golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
-golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-golang.org/x/time v0.8.0 h1:9i3RxcPv3PZnitoVGMPDKZSq1xW1gK1Xy3ArNOGZfEg=
-golang.org/x/time v0.8.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
+golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU=
+golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/time v0.9.0 h1:EsRrnYcQiGH+5FfbgvV4AP7qEZstoyrHB0DzarOQ4ZY=
+golang.org/x/time v0.9.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/server/consumer.go b/server/consumer.go
index 438041ec..f2677183 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -1415,8 +1415,23 @@ func (o *consumer) unsubscribe(sub *subscription) {
 
 // We need to make sure we protect access to the outq.
 // Do all advisory sends here.
-func (o *consumer) sendAdvisory(subj string, msg []byte) {
-	o.outq.sendMsg(subj, msg)
+func (o *consumer) sendAdvisory(subject string, e any) {
+	if o.acc == nil {
+		return
+	}
+
+	// If there is no one listening for this advisory then save ourselves the effort
+	// and don't bother encoding the JSON or sending it.
+	if sl := o.acc.sl; (sl != nil && !sl.HasInterest(subject)) && !o.srv.hasGatewayInterest(o.acc.Name, subject) {
+		return
+	}
+
+	j, err := json.Marshal(e)
+	if err != nil {
+		return
+	}
+
+	o.outq.sendMsg(subject, j)
 }
 
 func (o *consumer) sendDeleteAdvisoryLocked() {
@@ -1432,13 +1447,8 @@ func (o *consumer) sendDeleteAdvisoryLocked() {
 		Domain:   o.srv.getOpts().JetStreamDomain,
 	}
 
-	j, err := json.Marshal(e)
-	if err != nil {
-		return
-	}
-
 	subj := JSAdvisoryConsumerDeletedPre + "." + o.stream + "." + o.name
-	o.sendAdvisory(subj, j)
+	o.sendAdvisory(subj, e)
 }
 
 func (o *consumer) sendCreateAdvisory() {
@@ -1457,13 +1467,8 @@ func (o *consumer) sendCreateAdvisory() {
 		Domain:   o.srv.getOpts().JetStreamDomain,
 	}
 
-	j, err := json.Marshal(e)
-	if err != nil {
-		return
-	}
-
 	subj := JSAdvisoryConsumerCreatedPre + "." + o.stream + "." + o.name
-	o.sendAdvisory(subj, j)
+	o.sendAdvisory(subj, e)
 }
 
 // Created returns created time.
@@ -1573,6 +1578,8 @@ var (
 	consumerNotActiveMaxInterval   = defaultConsumerNotActiveMaxInterval
 )
 
+// deleteNotActive must only be called from time.AfterFunc or in its own
+// goroutine, as it can block on clean-up.
 func (o *consumer) deleteNotActive() {
 	o.mu.Lock()
 	if o.mset == nil {
@@ -1613,8 +1620,25 @@ func (o *consumer) deleteNotActive() {
 
 	s, js := o.mset.srv, o.srv.js.Load()
 	acc, stream, name, isDirect := o.acc.Name, o.stream, o.name, o.cfg.Direct
+	var qch, cqch chan struct{}
+	if o.srv != nil {
+		qch = o.srv.quitCh
+	}
+	if o.js != nil {
+		cqch = o.js.clusterQuitC()
+	}
 	o.mu.Unlock()
 
+	// Useful for pprof.
+	setGoRoutineLabels(pprofLabels{
+		"account":  acc,
+		"stream":   stream,
+		"consumer": name,
+	})
+
+	// We will delete locally regardless.
+	defer o.delete()
+
 	// If we are clustered, check if we still have this consumer assigned.
 	// If we do forward a proposal to delete ourselves to the metacontroller leader.
 	if !isDirect && s.JetStreamIsClustered() {
@@ -1637,38 +1661,40 @@ func (o *consumer) deleteNotActive() {
 		if ca != nil && cc != nil {
 			// Check to make sure we went away.
 			// Don't think this needs to be a monitored go routine.
-			go func() {
-				jitter := time.Duration(rand.Int63n(int64(consumerNotActiveStartInterval)))
-				interval := consumerNotActiveStartInterval + jitter
-				ticker := time.NewTicker(interval)
-				defer ticker.Stop()
-				for range ticker.C {
-					js.mu.RLock()
-					if js.shuttingDown {
-						js.mu.RUnlock()
-						return
-					}
-					nca := js.consumerAssignment(acc, stream, name)
+			jitter := time.Duration(rand.Int63n(int64(consumerNotActiveStartInterval)))
+			interval := consumerNotActiveStartInterval + jitter
+			ticker := time.NewTicker(interval)
+			defer ticker.Stop()
+			for {
+				select {
+				case <-ticker.C:
+				case <-qch:
+					return
+				case <-cqch:
+					return
+				}
+				js.mu.RLock()
+				if js.shuttingDown {
 					js.mu.RUnlock()
-					// Make sure this is not a new consumer with the same name.
-					if nca != nil && nca == ca {
-						s.Warnf("Consumer assignment for '%s > %s > %s' not cleaned up, retrying", acc, stream, name)
-						meta.ForwardProposal(removeEntry)
-						if interval < consumerNotActiveMaxInterval {
-							interval *= 2
-							ticker.Reset(interval)
-						}
-						continue
-					}
-					// We saw that consumer has been removed, all done.
 					return
 				}
-			}()
+				nca := js.consumerAssignment(acc, stream, name)
+				js.mu.RUnlock()
+				// Make sure this is not a new consumer with the same name.
+				if nca != nil && nca == ca {
+					s.Warnf("Consumer assignment for '%s > %s > %s' not cleaned up, retrying", acc, stream, name)
+					meta.ForwardProposal(removeEntry)
+					if interval < consumerNotActiveMaxInterval {
+						interval *= 2
+						ticker.Reset(interval)
+					}
+					continue
+				}
+				// We saw that consumer has been removed, all done.
+				return
+			}
 		}
 	}
-
-	// We will delete here regardless.
-	o.delete()
 }
 
 func (o *consumer) watchGWinterest() {
@@ -2382,12 +2408,7 @@ func (o *consumer) processNak(sseq, dseq, dc uint64, nak []byte) {
 		Domain:      o.srv.getOpts().JetStreamDomain,
 	}
 
-	j, err := json.Marshal(e)
-	if err != nil {
-		return
-	}
-
-	o.sendAdvisory(o.nakEventT, j)
+	o.sendAdvisory(o.nakEventT, e)
 
 	// Check to see if we have delays attached.
 	if len(nak) > len(AckNak) {
@@ -2462,15 +2483,8 @@ func (o *consumer) processTerm(sseq, dseq, dc uint64, reason, reply string) bool
 		Domain:      o.srv.getOpts().JetStreamDomain,
 	}
 
-	j, err := json.Marshal(e)
-	if err != nil {
-		// We had an error during the marshal, so we can't send the advisory,
-		// but we still need to tell the caller that the ack was processed.
-		return ackedInPlace
-	}
-
 	subj := JSAdvisoryConsumerMsgTerminatedPre + "." + o.stream + "." + o.name
-	o.sendAdvisory(subj, j)
+	o.sendAdvisory(subj, e)
 	return ackedInPlace
 }
 
@@ -2765,12 +2779,7 @@ func (o *consumer) sampleAck(sseq, dseq, dc uint64) {
 		Domain:      o.srv.getOpts().JetStreamDomain,
 	}
 
-	j, err := json.Marshal(e)
-	if err != nil {
-		return
-	}
-
-	o.sendAdvisory(o.ackEventT, j)
+	o.sendAdvisory(o.ackEventT, e)
 }
 
 // Process an ACK.
@@ -3515,12 +3524,7 @@ func (o *consumer) notifyDeliveryExceeded(sseq, dc uint64) {
 		Domain:     o.srv.getOpts().JetStreamDomain,
 	}
 
-	j, err := json.Marshal(e)
-	if err != nil {
-		return
-	}
-
-	o.sendAdvisory(o.deliveryExcEventT, j)
+	o.sendAdvisory(o.deliveryExcEventT, e)
 }
 
 // Check if the candidate subject matches a filter if its present.
@@ -5379,6 +5383,7 @@ func (o *consumer) requestNextMsgSubject() string {
 
 func (o *consumer) decStreamPending(sseq uint64, subj string) {
 	o.mu.Lock()
+
 	// Update our cached num pending only if we think deliverMsg has not done so.
 	if sseq >= o.sseq && o.isFilteredMatch(subj) {
 		o.npc--
@@ -5390,6 +5395,7 @@ func (o *consumer) decStreamPending(sseq uint64, subj string) {
 	if o.rdc != nil {
 		rdc = o.rdc[sseq]
 	}
+
 	o.mu.Unlock()
 
 	// If it was pending process it like an ack.
diff --git a/server/events.go b/server/events.go
index 7c891b42..7cb9feb6 100644
--- a/server/events.go
+++ b/server/events.go
@@ -324,6 +324,28 @@ func (ci *ClientInfo) forAssignmentSnap() *ClientInfo {
 	}
 }
 
+// forProposal returns the minimum amount of ClientInfo we need for assignment proposals.
+func (ci *ClientInfo) forProposal() *ClientInfo {
+	if ci == nil {
+		return nil
+	}
+	cci := *ci
+	cci.Jwt = _EMPTY_
+	cci.IssuerKey = _EMPTY_
+	return &cci
+}
+
+// forAdvisory returns the minimum amount of ClientInfo we need for JS advisory events.
+func (ci *ClientInfo) forAdvisory() *ClientInfo {
+	if ci == nil {
+		return nil
+	}
+	cci := *ci
+	cci.Jwt = _EMPTY_
+	cci.Alternates = nil
+	return &cci
+}
+
 // ServerStats hold various statistics that we will periodically send out.
 type ServerStats struct {
 	Start            time.Time      `json:"start"`
diff --git a/server/filestore.go b/server/filestore.go
index c5920587..245c68a7 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -1376,14 +1376,14 @@ func (mb *msgBlock) rebuildStateLocked() (*LostStreamData, []uint64, error) {
 		}
 
 		hdr := buf[index : index+msgHdrSize]
-		rl, slen := le.Uint32(hdr[0:]), le.Uint16(hdr[20:])
+		rl, slen := le.Uint32(hdr[0:]), int(le.Uint16(hdr[20:]))
 
 		hasHeaders := rl&hbit != 0
 		// Clear any headers bit that could be set.
 		rl &^= hbit
 		dlen := int(rl) - msgHdrSize
 		// Do some quick sanity checks here.
-		if dlen < 0 || int(slen) > (dlen-recordHashSize) || dlen > int(rl) || index+rl > lbuf || rl > rlBadThresh {
+		if dlen < 0 || slen > (dlen-recordHashSize) || dlen > int(rl) || index+rl > lbuf || rl > rlBadThresh {
 			truncate(index)
 			return gatherLost(lbuf - index), tombstones, errBadMsg
 		}
@@ -2315,8 +2315,8 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *Stor
 		fseq = lseq + 1
 		for _, subj := range subs {
 			ss, _ := mb.fss.Find(stringToBytes(subj))
-			if ss != nil && ss.firstNeedsUpdate {
-				mb.recalculateFirstForSubj(subj, ss.First, ss)
+			if ss != nil && (ss.firstNeedsUpdate || ss.lastNeedsUpdate) {
+				mb.recalculateForSubj(subj, ss)
 			}
 			if ss == nil || start > ss.Last || ss.First >= fseq {
 				continue
@@ -2445,8 +2445,8 @@ func (mb *msgBlock) filteredPendingLocked(filter string, wc bool, sseq uint64) (
 			// If we already found a partial then don't do anything else.
 			return
 		}
-		if ss.firstNeedsUpdate {
-			mb.recalculateFirstForSubj(bytesToString(bsubj), ss.First, ss)
+		if ss.firstNeedsUpdate || ss.lastNeedsUpdate {
+			mb.recalculateForSubj(bytesToString(bsubj), ss)
 		}
 		if sseq <= ss.First {
 			update(ss)
@@ -2616,10 +2616,6 @@ func (fs *fileStore) numFilteredPendingWithLast(filter string, last bool, ss *Si
 	// Always reset.
 	ss.First, ss.Last, ss.Msgs = 0, 0, 0
 
-	if filter == _EMPTY_ {
-		filter = fwcs
-	}
-
 	// We do need to figure out the first and last sequences.
 	wc := subjectHasWildcard(filter)
 	start, stop := uint32(math.MaxUint32), uint32(0)
@@ -2749,8 +2745,8 @@ func (fs *fileStore) SubjectsState(subject string) map[string]SimpleState {
 		mb.lsts = time.Now().UnixNano()
 		mb.fss.Match(stringToBytes(subject), func(bsubj []byte, ss *SimpleState) {
 			subj := string(bsubj)
-			if ss.firstNeedsUpdate {
-				mb.recalculateFirstForSubj(subj, ss.First, ss)
+			if ss.firstNeedsUpdate || ss.lastNeedsUpdate {
+				mb.recalculateForSubj(subj, ss)
 			}
 			oss := fss[subj]
 			if oss.First == 0 { // New
@@ -2940,8 +2936,8 @@ func (fs *fileStore) NumPending(sseq uint64, filter string, lastPerSubject bool)
 					return
 				}
 				subj := bytesToString(bsubj)
-				if ss.firstNeedsUpdate {
-					mb.recalculateFirstForSubj(subj, ss.First, ss)
+				if ss.firstNeedsUpdate || ss.lastNeedsUpdate {
+					mb.recalculateForSubj(subj, ss)
 				}
 				if sseq <= ss.First {
 					t += ss.Msgs
@@ -3228,8 +3224,8 @@ func (fs *fileStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject bo
 					// If we already found a partial then don't do anything else.
 					return
 				}
-				if ss.firstNeedsUpdate {
-					mb.recalculateFirstForSubj(subj, ss.First, ss)
+				if ss.firstNeedsUpdate || ss.lastNeedsUpdate {
+					mb.recalculateForSubj(subj, ss)
 				}
 				if sseq <= ss.First {
 					t += ss.Msgs
@@ -3902,8 +3898,8 @@ func (fs *fileStore) firstSeqForSubj(subj string) (uint64, error) {
 					info.fblk = i
 				}
 			}
-			if ss.firstNeedsUpdate {
-				mb.recalculateFirstForSubj(subj, ss.First, ss)
+			if ss.firstNeedsUpdate || ss.lastNeedsUpdate {
+				mb.recalculateForSubj(subj, ss)
 			}
 			mb.mu.Unlock()
 			// Re-acquire fs lock
@@ -4034,8 +4030,8 @@ func (fs *fileStore) enforceMsgPerSubjectLimit(fireCallback bool) {
 			mb.mu.Lock()
 			mb.ensurePerSubjectInfoLoaded()
 			ss, ok := mb.fss.Find(stringToBytes(subj))
-			if ok && ss != nil && ss.firstNeedsUpdate {
-				mb.recalculateFirstForSubj(subj, ss.First, ss)
+			if ok && ss != nil && (ss.firstNeedsUpdate || ss.lastNeedsUpdate) {
+				mb.recalculateForSubj(subj, ss)
 			}
 			mb.mu.Unlock()
 			if ss == nil {
@@ -4364,12 +4360,12 @@ func (mb *msgBlock) compactWithFloor(floor uint64) {
 			return
 		}
 		hdr := buf[index : index+msgHdrSize]
-		rl, slen := le.Uint32(hdr[0:]), le.Uint16(hdr[20:])
+		rl, slen := le.Uint32(hdr[0:]), int(le.Uint16(hdr[20:]))
 		// Clear any headers bit that could be set.
 		rl &^= hbit
 		dlen := int(rl) - msgHdrSize
 		// Do some quick sanity checks here.
-		if dlen < 0 || int(slen) > dlen || dlen > int(rl) || rl > rlBadThresh || index+rl > lbuf {
+		if dlen < 0 || slen > (dlen-recordHashSize) || dlen > int(rl) || index+rl > lbuf || rl > rlBadThresh {
 			return
 		}
 		// Only need to process non-deleted messages.
@@ -6488,7 +6484,7 @@ func (mb *msgBlock) msgFromBuf(buf []byte, sm *StoreMsg, hh hash.Hash64) (*Store
 	dlen := int(rl) - msgHdrSize
 	slen := int(le.Uint16(hdr[20:]))
 	// Simple sanity check.
-	if dlen < 0 || slen > (dlen-recordHashSize) || dlen > int(rl) || int(rl) > len(buf) {
+	if dlen < 0 || slen > (dlen-recordHashSize) || dlen > int(rl) || int(rl) > len(buf) || rl > rlBadThresh {
 		return nil, errBadMsg
 	}
 	data := buf[msgHdrSize : msgHdrSize+dlen]
@@ -7836,24 +7832,14 @@ func (mb *msgBlock) removeSeqPerSubject(subj string, seq uint64) {
 
 	ss.Msgs--
 
-	// Only one left.
-	if ss.Msgs == 1 {
-		if seq == ss.Last {
-			ss.Last = ss.First
-		} else {
-			ss.First = ss.Last
-		}
-		ss.firstNeedsUpdate = false
-		return
-	}
-
-	// We can lazily calculate the first sequence when needed.
+	// We can lazily calculate the first/last sequence when needed.
 	ss.firstNeedsUpdate = seq == ss.First || ss.firstNeedsUpdate
+	ss.lastNeedsUpdate = seq == ss.Last || ss.lastNeedsUpdate
 }
 
-// Will recalulate the first sequence for this subject in this block.
+// Will recalculate the first and/or last sequence for this subject in this block.
 // Will avoid slower path message lookups and scan the cache directly instead.
-func (mb *msgBlock) recalculateFirstForSubj(subj string, startSeq uint64, ss *SimpleState) {
+func (mb *msgBlock) recalculateForSubj(subj string, ss *SimpleState) {
 	// Need to make sure messages are loaded.
 	if mb.cacheNotLoaded() {
 		if err := mb.loadMsgsWithLock(); err != nil {
@@ -7861,42 +7847,100 @@ func (mb *msgBlock) recalculateFirstForSubj(subj string, startSeq uint64, ss *Si
 		}
 	}
 
-	// Mark first as updated.
-	ss.firstNeedsUpdate = false
-
-	startSlot := int(startSeq - mb.cache.fseq)
+	startSlot := int(ss.First - mb.cache.fseq)
+	if startSlot < 0 {
+		startSlot = 0
+	}
 	if startSlot >= len(mb.cache.idx) {
 		ss.First = ss.Last
 		return
-	} else if startSlot < 0 {
-		startSlot = 0
+	}
+	endSlot := int(ss.Last - mb.cache.fseq)
+	if endSlot < 0 {
+		endSlot = 0
+	}
+	if endSlot >= len(mb.cache.idx) || startSlot > endSlot {
+		return
 	}
 
 	var le = binary.LittleEndian
-	for slot, fseq := startSlot, atomic.LoadUint64(&mb.first.seq); slot < len(mb.cache.idx); slot++ {
-		bi := mb.cache.idx[slot] &^ hbit
-		if bi == dbit {
-			// delete marker so skip.
-			continue
+	if ss.firstNeedsUpdate {
+		// Mark first as updated.
+		ss.firstNeedsUpdate = false
+
+		fseq := ss.First + 1
+		if mbFseq := atomic.LoadUint64(&mb.first.seq); fseq < mbFseq {
+			fseq = mbFseq
 		}
-		li := int(bi) - mb.cache.off
-		if li >= len(mb.cache.buf) {
-			ss.First = ss.Last
-			return
+		for slot := startSlot; slot < len(mb.cache.idx); slot++ {
+			bi := mb.cache.idx[slot] &^ hbit
+			if bi == dbit {
+				// delete marker so skip.
+				continue
+			}
+			li := int(bi) - mb.cache.off
+			if li >= len(mb.cache.buf) {
+				ss.First = ss.Last
+				return
+			}
+			buf := mb.cache.buf[li:]
+			hdr := buf[:msgHdrSize]
+			slen := int(le.Uint16(hdr[20:]))
+			if subj == bytesToString(buf[msgHdrSize:msgHdrSize+slen]) {
+				seq := le.Uint64(hdr[4:])
+				if seq < fseq || seq&ebit != 0 || mb.dmap.Exists(seq) {
+					continue
+				}
+				ss.First = seq
+				if ss.Msgs == 1 {
+					ss.Last = seq
+					ss.lastNeedsUpdate = false
+					return
+				}
+				// Skip the start slot ahead, if we need to recalculate last we can stop early.
+				startSlot = slot
+				break
+			}
+		}
+	}
+	if ss.lastNeedsUpdate {
+		// Mark last as updated.
+		ss.lastNeedsUpdate = false
+
+		lseq := ss.Last - 1
+		if mbLseq := atomic.LoadUint64(&mb.last.seq); lseq > mbLseq {
+			lseq = mbLseq
 		}
-		buf := mb.cache.buf[li:]
-		hdr := buf[:msgHdrSize]
-		slen := int(le.Uint16(hdr[20:]))
-		if subj == bytesToString(buf[msgHdrSize:msgHdrSize+slen]) {
-			seq := le.Uint64(hdr[4:])
-			if seq < fseq || seq&ebit != 0 || mb.dmap.Exists(seq) {
+		for slot := endSlot; slot >= startSlot; slot-- {
+			bi := mb.cache.idx[slot] &^ hbit
+			if bi == dbit {
+				// delete marker so skip.
 				continue
 			}
-			ss.First = seq
-			if ss.Msgs == 1 {
+			li := int(bi) - mb.cache.off
+			if li >= len(mb.cache.buf) {
+				// Can't overwrite ss.Last, just skip.
+				return
+			}
+			buf := mb.cache.buf[li:]
+			hdr := buf[:msgHdrSize]
+			slen := int(le.Uint16(hdr[20:]))
+			if subj == bytesToString(buf[msgHdrSize:msgHdrSize+slen]) {
+				seq := le.Uint64(hdr[4:])
+				if seq > lseq || seq&ebit != 0 || mb.dmap.Exists(seq) {
+					continue
+				}
+				// Sequence should never be lower, but guard against it nonetheless.
+				if seq < ss.First {
+					seq = ss.First
+				}
 				ss.Last = seq
+				if ss.Msgs == 1 {
+					ss.First = seq
+					ss.firstNeedsUpdate = false
+				}
+				return
 			}
-			return
 		}
 	}
 }
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index de014e74..d47b6dcc 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -864,8 +864,10 @@ func (s *Server) processJSAPIRoutedRequests() {
 	for {
 		select {
 		case <-queue.ch:
-			reqs := queue.pop()
-			for _, r := range reqs {
+			// Only pop one item at a time here, otherwise if the system is recovering
+			// from queue buildup, then one worker will pull off all the tasks and the
+			// others will be starved of work.
+			for r, ok := queue.popOne(); ok && r != nil; r, ok = queue.popOne() {
 				client.pa = r.pa
 				start := time.Now()
 				r.jsub.icb(r.sub, client, r.acc, r.subject, r.reply, r.msg)
@@ -874,7 +876,6 @@ func (s *Server) processJSAPIRoutedRequests() {
 				}
 				atomic.AddInt64(&js.apiInflight, -1)
 			}
-			queue.recycle(&reqs)
 		case <-s.quitCh:
 			return
 		}
@@ -3416,7 +3417,7 @@ func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, cfg *StreamC
 			Time: start,
 		},
 		Stream: streamName,
-		Client: ci,
+		Client: ci.forAdvisory(),
 		Domain: domain,
 	})
 
@@ -3548,7 +3549,7 @@ func (s *Server) processStreamRestore(ci *ClientInfo, acc *Account, cfg *StreamC
 					Start:  start,
 					End:    end,
 					Bytes:  int64(total),
-					Client: ci,
+					Client: ci.forAdvisory(),
 					Domain: domain,
 				})
 
@@ -3681,7 +3682,7 @@ func (s *Server) jsStreamSnapshotRequest(sub *subscription, c *client, _ *Accoun
 			},
 			Stream: mset.name(),
 			State:  sr.State,
-			Client: ci,
+			Client: ci.forAdvisory(),
 			Domain: s.getOpts().JetStreamDomain,
 		})
 
@@ -3699,7 +3700,7 @@ func (s *Server) jsStreamSnapshotRequest(sub *subscription, c *client, _ *Accoun
 			Stream: mset.name(),
 			Start:  start,
 			End:    end,
-			Client: ci,
+			Client: ci.forAdvisory(),
 			Domain: s.getOpts().JetStreamDomain,
 		})
 
@@ -4263,9 +4264,17 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, _ *Account,
 			return
 		}
 
+		js.mu.RLock()
+		meta := cc.meta
+		js.mu.RUnlock()
+
+		// Since these could wait on the Raft group lock, don't do so under the JS lock.
+		ourID := meta.ID()
+		groupLeader := meta.GroupLeader()
+		groupCreated := meta.Created()
+
 		js.mu.RLock()
 		isLeader, sa, ca := cc.isLeader(), js.streamAssignment(acc.Name, streamName), js.consumerAssignment(acc.Name, streamName, consumerName)
-		ourID := cc.meta.ID()
 		var rg *raftGroup
 		var offline, isMember bool
 		if ca != nil {
@@ -4279,7 +4288,7 @@ func (s *Server) jsConsumerInfoRequest(sub *subscription, c *client, _ *Account,
 		// Also capture if we think there is no meta leader.
 		var isLeaderLess bool
 		if !isLeader {
-			isLeaderLess = cc.meta.GroupLeader() == _EMPTY_ && time.Since(cc.meta.Created()) > lostQuorumIntervalDefault
+			isLeaderLess = groupLeader == _EMPTY_ && time.Since(groupCreated) > lostQuorumIntervalDefault
 		}
 		js.mu.RUnlock()
 
@@ -4489,7 +4498,7 @@ func (s *Server) sendJetStreamAPIAuditAdvisory(ci *ClientInfo, acc *Account, sub
 			Time: time.Now().UTC(),
 		},
 		Server:   s.Name(),
-		Client:   ci,
+		Client:   ci.forAdvisory(),
 		Subject:  subject,
 		Request:  request,
 		Response: response,
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 8f08b1e5..dcd54ed7 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -901,15 +901,17 @@ func (js *jetStream) server() *Server {
 // Will respond if we do not think we have a metacontroller leader.
 func (js *jetStream) isLeaderless() bool {
 	js.mu.RLock()
-	defer js.mu.RUnlock()
-
 	cc := js.cluster
 	if cc == nil || cc.meta == nil {
+		js.mu.RUnlock()
 		return false
 	}
+	meta := cc.meta
+	js.mu.RUnlock()
+
 	// If we don't have a leader.
 	// Make sure we have been running for enough time.
-	if cc.meta.GroupLeader() == _EMPTY_ && time.Since(cc.meta.Created()) > lostQuorumIntervalDefault {
+	if meta.GroupLeader() == _EMPTY_ && time.Since(meta.Created()) > lostQuorumIntervalDefault {
 		return true
 	}
 	return false
@@ -921,34 +923,38 @@ func (js *jetStream) isGroupLeaderless(rg *raftGroup) bool {
 		return false
 	}
 	js.mu.RLock()
-	defer js.mu.RUnlock()
-
 	cc := js.cluster
+	started := js.started
 
 	// If we are not a member we can not say..
 	if cc.meta == nil {
+		js.mu.RUnlock()
 		return false
 	}
 	if !rg.isMember(cc.meta.ID()) {
+		js.mu.RUnlock()
 		return false
 	}
 	// Single peer groups always have a leader if we are here.
 	if rg.node == nil {
+		js.mu.RUnlock()
 		return false
 	}
+	node := rg.node
+	js.mu.RUnlock()
 	// If we don't have a leader.
-	if rg.node.GroupLeader() == _EMPTY_ {
+	if node.GroupLeader() == _EMPTY_ {
 		// Threshold for jetstream startup.
 		const startupThreshold = 10 * time.Second
 
-		if rg.node.HadPreviousLeader() {
+		if node.HadPreviousLeader() {
 			// Make sure we have been running long enough to intelligently determine this.
-			if time.Since(js.started) > startupThreshold {
+			if time.Since(started) > startupThreshold {
 				return true
 			}
 		}
 		// Make sure we have been running for enough time.
-		if time.Since(rg.node.Created()) > lostQuorumIntervalDefault {
+		if time.Since(node.Created()) > lostQuorumIntervalDefault {
 			return true
 		}
 	}
@@ -2411,7 +2417,6 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 		compactInterval = 2 * time.Minute
 		compactSizeMin  = 8 * 1024 * 1024
 		compactNumMin   = 65536
-		minSnapDelta    = 10 * time.Second
 	)
 
 	// Spread these out for large numbers on server restart.
@@ -2435,16 +2440,15 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 	// a complete and detailed state which could be costly in terms of memory, cpu and GC.
 	// This only entails how many messages, and the first and last sequence of the stream.
 	// This is all that is needed to detect a change, and we can get this from FilteredState()
-	// with and empty filter.
+	// with an empty filter.
 	var lastState SimpleState
-	var lastSnapTime time.Time
 
 	// Don't allow the upper layer to install snapshots until we have
 	// fully recovered from disk.
 	isRecovering := true
 
 	doSnapshot := func() {
-		if mset == nil || isRecovering || isRestore || time.Since(lastSnapTime) < minSnapDelta {
+		if mset == nil || isRecovering || isRestore {
 			return
 		}
 
@@ -2462,7 +2466,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 		}
 
 		if err := n.InstallSnapshot(mset.stateSnapshot()); err == nil {
-			lastState, lastSnapTime = curState, time.Now()
+			lastState = curState
 		} else if err != errNoSnapAvailable && err != errNodeClosed && err != errCatchupsRunning {
 			s.RateLimitWarnf("Failed to install snapshot for '%s > %s' [%s]: %v", mset.acc.Name, mset.name(), n.Group(), err)
 		}
@@ -2541,10 +2545,14 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 	for {
 		select {
 		case <-s.quitCh:
+			// Server shutting down, but we might receive this before qch, so try to snapshot.
+			doSnapshot()
 			return
 		case <-mqch:
 			return
 		case <-qch:
+			// Clean signal from shutdown routine so do best effort attempt to snapshot.
+			doSnapshot()
 			return
 		case <-aq.ch:
 			var ne, nb uint64
@@ -2603,12 +2611,6 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 			// Check about snapshotting
 			// If we have at least min entries to compact, go ahead and try to snapshot/compact.
 			if ne >= compactNumMin || nb > compactSizeMin || mset.getCLFS() > pclfs {
-				// We want to make sure we do not short circuit if transistioning from no clfs.
-				if pclfs == 0 {
-					// This is always false by default.
-					lastState.firstNeedsUpdate = true
-					lastSnapTime = time.Time{}
-				}
 				doSnapshot()
 			}
 
@@ -2711,8 +2713,6 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 			// keep stream assignment current
 			sa = mset.streamAssignment()
 
-			// keep peer list up to date with config
-			js.checkPeers(mset.raftGroup())
 			// We get this when we have a new stream assignment caused by an update.
 			// We want to know if we are migrating.
 			if migrating := mset.isMigrating(); migrating {
@@ -2800,7 +2800,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 				// Check if we have a quorom.
 				if current >= neededCurrent {
 					s.Noticef("Transfer of stream leader for '%s > %s' to '%s'", accName, sa.Config.Name, newLeader)
-					n.UpdateKnownPeers(newPeers)
+					n.ProposeKnownPeers(newPeers)
 					n.StepDown(newLeaderPeer)
 				}
 			}
@@ -3090,8 +3090,10 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 				if subject == _EMPTY_ && ts == 0 && len(msg) == 0 && len(hdr) == 0 {
 					// Skip and update our lseq.
 					last := mset.store.SkipMsg()
+					mset.mu.Lock()
 					mset.setLastSeq(last)
 					mset.clearAllPreAcks(last)
+					mset.mu.Unlock()
 					continue
 				}
 
@@ -3318,22 +3320,6 @@ func (s *Server) replicas(node RaftNode) []*PeerInfo {
 	return replicas
 }
 
-// Will check our node peers and see if we should remove a peer.
-func (js *jetStream) checkPeers(rg *raftGroup) {
-	js.mu.Lock()
-	defer js.mu.Unlock()
-
-	// FIXME(dlc) - Single replicas?
-	if rg == nil || rg.node == nil {
-		return
-	}
-	for _, peer := range rg.node.Peers() {
-		if !rg.isMember(peer.ID) {
-			rg.node.ProposeRemovePeer(peer.ID)
-		}
-	}
-}
-
 // Process a leader change for the clustered stream.
 func (js *jetStream) processStreamLeaderChange(mset *stream, isLeader bool) {
 	if mset == nil {
@@ -3362,8 +3348,6 @@ func (js *jetStream) processStreamLeaderChange(mset *stream, isLeader bool) {
 	if isLeader {
 		s.Noticef("JetStream cluster new stream leader for '%s > %s'", account, streamName)
 		s.sendStreamLeaderElectAdvisory(mset)
-		// Check for peer removal and process here if needed.
-		js.checkPeers(sa.Group)
 		mset.checkAllowMsgCompress(peers)
 	} else {
 		// We are stepping down.
@@ -3579,7 +3563,7 @@ func (js *jetStream) processStreamAssignment(sa *streamAssignment) bool {
 		js.processClusterCreateStream(acc, sa)
 	} else if mset, _ := acc.lookupStream(sa.Config.Name); mset != nil {
 		// We have one here even though we are not a member. This can happen on re-assignment.
-		s.removeStream(ourID, mset, sa)
+		s.removeStream(mset, sa)
 	}
 
 	// If this stream assignment does not have a sync subject (bug) set that the meta-leader should check when elected.
@@ -3667,13 +3651,13 @@ func (js *jetStream) processUpdateStreamAssignment(sa *streamAssignment) {
 		js.processClusterUpdateStream(acc, osa, sa)
 	} else if mset, _ := acc.lookupStream(sa.Config.Name); mset != nil {
 		// We have one here even though we are not a member. This can happen on re-assignment.
-		s.removeStream(ourID, mset, sa)
+		s.removeStream(mset, sa)
 	}
 }
 
-// Common function to remove ourself from this server.
+// Common function to remove ourselves from this server.
 // This can happen on re-assignment, move, etc
-func (s *Server) removeStream(ourID string, mset *stream, nsa *streamAssignment) {
+func (s *Server) removeStream(mset *stream, nsa *streamAssignment) {
 	if mset == nil {
 		return
 	}
@@ -3683,7 +3667,6 @@ func (s *Server) removeStream(ourID string, mset *stream, nsa *streamAssignment)
 		if node.Leader() {
 			node.StepDown(nsa.Group.Preferred)
 		}
-		node.ProposeRemovePeer(ourID)
 		// shutdown monitor by shutting down raft.
 		node.Delete()
 	}
@@ -4472,10 +4455,11 @@ func (js *jetStream) processClusterCreateConsumer(ca *consumerAssignment, state
 	} else {
 		// If we are clustered update the known peers.
 		js.mu.RLock()
-		if node := rg.node; node != nil {
+		node := rg.node
+		js.mu.RUnlock()
+		if node != nil {
 			node.UpdateKnownPeers(ca.Group.Peers)
 		}
-		js.mu.RUnlock()
 	}
 
 	// Check if we already have this consumer running.
@@ -4943,8 +4927,12 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 	for {
 		select {
 		case <-s.quitCh:
+			// Server shutting down, but we might receive this before qch, so try to snapshot.
+			doSnapshot(false)
 			return
 		case <-qch:
+			// Clean signal from shutdown routine so do best effort attempt to snapshot.
+			doSnapshot(false)
 			return
 		case <-aq.ch:
 			ces := aq.pop()
@@ -5009,8 +4997,6 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 			// We get this when we have a new consumer assignment caused by an update.
 			// We want to know if we are migrating.
 			rg := o.raftGroup()
-			// keep peer list up to date with config
-			js.checkPeers(rg)
 			// If we are migrating, monitor for the new peers to be caught up.
 			replicas, err := o.replica()
 			if err != nil {
@@ -5327,8 +5313,6 @@ func (js *jetStream) processConsumerLeaderChange(o *consumer, isLeader bool) err
 	if isLeader {
 		s.Noticef("JetStream cluster new consumer leader for '%s > %s > %s'", ca.Client.serviceAccount(), streamName, consumerName)
 		s.sendConsumerLeaderElectAdvisory(o)
-		// Check for peer removal and process here if needed.
-		js.checkPeers(ca.Group)
 	} else {
 		// We are stepping down.
 		// Make sure if we are doing so because we have lost quorum that we send the appropriate advisories.
@@ -7267,23 +7251,29 @@ func (s *Server) jsClusteredMsgDeleteRequest(ci *ClientInfo, acc *Account, mset
 }
 
 func encodeAddStreamAssignment(sa *streamAssignment) []byte {
+	csa := *sa
+	csa.Client = csa.Client.forProposal()
 	var bb bytes.Buffer
 	bb.WriteByte(byte(assignStreamOp))
-	json.NewEncoder(&bb).Encode(sa)
+	json.NewEncoder(&bb).Encode(csa)
 	return bb.Bytes()
 }
 
 func encodeUpdateStreamAssignment(sa *streamAssignment) []byte {
+	csa := *sa
+	csa.Client = csa.Client.forProposal()
 	var bb bytes.Buffer
 	bb.WriteByte(byte(updateStreamOp))
-	json.NewEncoder(&bb).Encode(sa)
+	json.NewEncoder(&bb).Encode(csa)
 	return bb.Bytes()
 }
 
 func encodeDeleteStreamAssignment(sa *streamAssignment) []byte {
+	csa := *sa
+	csa.Client = csa.Client.forProposal()
 	var bb bytes.Buffer
 	bb.WriteByte(byte(removeStreamOp))
-	json.NewEncoder(&bb).Encode(sa)
+	json.NewEncoder(&bb).Encode(csa)
 	return bb.Bytes()
 }
 
@@ -7671,16 +7661,20 @@ func (s *Server) jsClusteredConsumerRequest(ci *ClientInfo, acc *Account, subjec
 }
 
 func encodeAddConsumerAssignment(ca *consumerAssignment) []byte {
+	cca := *ca
+	cca.Client = cca.Client.forProposal()
 	var bb bytes.Buffer
 	bb.WriteByte(byte(assignConsumerOp))
-	json.NewEncoder(&bb).Encode(ca)
+	json.NewEncoder(&bb).Encode(cca)
 	return bb.Bytes()
 }
 
 func encodeDeleteConsumerAssignment(ca *consumerAssignment) []byte {
+	cca := *ca
+	cca.Client = cca.Client.forProposal()
 	var bb bytes.Buffer
 	bb.WriteByte(byte(removeConsumerOp))
-	json.NewEncoder(&bb).Encode(ca)
+	json.NewEncoder(&bb).Encode(cca)
 	return bb.Bytes()
 }
 
@@ -7691,25 +7685,21 @@ func decodeConsumerAssignment(buf []byte) (*consumerAssignment, error) {
 }
 
 func encodeAddConsumerAssignmentCompressed(ca *consumerAssignment) []byte {
-	b, err := json.Marshal(ca)
-	if err != nil {
-		return nil
-	}
-	// TODO(dlc) - Streaming better approach here probably.
+	cca := *ca
+	cca.Client = cca.Client.forProposal()
 	var bb bytes.Buffer
 	bb.WriteByte(byte(assignCompressedConsumerOp))
-	bb.Write(s2.Encode(nil, b))
+	s2e := s2.NewWriter(&bb)
+	json.NewEncoder(s2e).Encode(cca)
+	s2e.Close()
 	return bb.Bytes()
 }
 
 func decodeConsumerAssignmentCompressed(buf []byte) (*consumerAssignment, error) {
 	var ca consumerAssignment
-	js, err := s2.Decode(nil, buf)
-	if err != nil {
-		return nil, err
-	}
-	err = json.Unmarshal(js, &ca)
-	return &ca, err
+	bb := bytes.NewBuffer(buf)
+	s2d := s2.NewReader(bb)
+	return &ca, json.NewDecoder(s2d).Decode(&ca)
 }
 
 var errBadStreamMsg = errors.New("jetstream cluster bad replicated stream msg")
@@ -8654,6 +8644,8 @@ func (mset *stream) processCatchupMsg(msg []byte) (uint64, error) {
 		return 0, err
 	}
 
+	mset.mu.Lock()
+	defer mset.mu.Unlock()
 	// Update our lseq.
 	mset.setLastSeq(seq)
 
@@ -8661,11 +8653,9 @@ func (mset *stream) processCatchupMsg(msg []byte) (uint64, error) {
 	if len(hdr) > 0 {
 		if msgId := getMsgId(hdr); msgId != _EMPTY_ {
 			if !ddloaded {
-				mset.mu.Lock()
 				mset.rebuildDedupe()
-				mset.mu.Unlock()
 			}
-			mset.storeMsgId(&ddentry{msgId, seq, ts})
+			mset.storeMsgIdLocked(&ddentry{msgId, seq, ts})
 		}
 	}
 
diff --git a/server/jetstream_events.go b/server/jetstream_events.go
index 8302fcc4..8c099c7a 100644
--- a/server/jetstream_events.go
+++ b/server/jetstream_events.go
@@ -18,13 +18,22 @@ import (
 	"time"
 )
 
-func (s *Server) publishAdvisory(acc *Account, subject string, adv any) {
+// publishAdvisory sends the given advisory into the account. Returns true if
+// it was sent, false if not (i.e. due to lack of interest or a marshal error).
+func (s *Server) publishAdvisory(acc *Account, subject string, adv any) bool {
 	if acc == nil {
 		acc = s.SystemAccount()
 		if acc == nil {
-			return
+			return false
 		}
 	}
+
+	// If there is no one listening for this advisory then save ourselves the effort
+	// and don't bother encoding the JSON or sending it.
+	if sl := acc.sl; (sl != nil && !sl.HasInterest(subject)) && !s.hasGatewayInterest(acc.Name, subject) {
+		return false
+	}
+
 	ej, err := json.Marshal(adv)
 	if err == nil {
 		err = s.sendInternalAccountMsg(acc, subject, ej)
@@ -34,6 +43,7 @@ func (s *Server) publishAdvisory(acc *Account, subject string, adv any) {
 	} else {
 		s.Warnf("Advisory could not be serialized for account %q: %v", acc.Name, err)
 	}
+	return err == nil
 }
 
 // JSAPIAudit is an advisory about administrative actions taken on JetStream
diff --git a/server/leafnode.go b/server/leafnode.go
index 26a3f6ec..5f3fa458 100644
--- a/server/leafnode.go
+++ b/server/leafnode.go
@@ -2247,8 +2247,16 @@ func (c *client) sendLeafNodeSubUpdate(key string, n int32) {
 				checkPerms = false
 			}
 		}
-		if checkPerms && !c.canSubscribe(key) {
-			return
+		if checkPerms {
+			var subject string
+			if sep := strings.IndexByte(key, ' '); sep != -1 {
+				subject = key[:sep]
+			} else {
+				subject = key
+			}
+			if !c.canSubscribe(subject) {
+				return
+			}
 		}
 	}
 	// If we are here we can send over to the other side.
diff --git a/server/memstore.go b/server/memstore.go
index e2ca1cae..350cfa38 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -143,8 +143,8 @@ func (ms *memStore) storeRawMsg(subj string, hdr, msg []byte, seq uint64, ts int
 					return ErrMaxBytes
 				}
 				// If we are here we are at a subject maximum, need to determine if dropping last message gives us enough room.
-				if ss.firstNeedsUpdate {
-					ms.recalculateFirstForSubj(subj, ss.First, ss)
+				if ss.firstNeedsUpdate || ss.lastNeedsUpdate {
+					ms.recalculateForSubj(subj, ss)
 				}
 				sm, ok := ms.msgs[ss.First]
 				if !ok || memStoreMsgSize(sm.subj, sm.hdr, sm.msg) < memStoreMsgSize(subj, hdr, msg) {
@@ -430,8 +430,8 @@ func (ms *memStore) filteredStateLocked(sseq uint64, filter string, lastPerSubje
 	var totalSkipped uint64
 	// We will track start and end sequences as we go.
 	ms.fss.Match(stringToBytes(filter), func(subj []byte, fss *SimpleState) {
-		if fss.firstNeedsUpdate {
-			ms.recalculateFirstForSubj(bytesToString(subj), fss.First, fss)
+		if fss.firstNeedsUpdate || fss.lastNeedsUpdate {
+			ms.recalculateForSubj(bytesToString(subj), fss)
 		}
 		if sseq <= fss.First {
 			update(fss)
@@ -585,8 +585,8 @@ func (ms *memStore) SubjectsState(subject string) map[string]SimpleState {
 	fss := make(map[string]SimpleState)
 	ms.fss.Match(stringToBytes(subject), func(subj []byte, ss *SimpleState) {
 		subjs := string(subj)
-		if ss.firstNeedsUpdate {
-			ms.recalculateFirstForSubj(subjs, ss.First, ss)
+		if ss.firstNeedsUpdate || ss.lastNeedsUpdate {
+			ms.recalculateForSubj(subjs, ss)
 		}
 		oss := fss[subjs]
 		if oss.First == 0 { // New
@@ -675,8 +675,8 @@ func (ms *memStore) NumPendingMulti(sseq uint64, sl *Sublist, lastPerSubject boo
 	var totalSkipped uint64
 	// We will track start and end sequences as we go.
 	IntersectStree[SimpleState](ms.fss, sl, func(subj []byte, fss *SimpleState) {
-		if fss.firstNeedsUpdate {
-			ms.recalculateFirstForSubj(bytesToString(subj), fss.First, fss)
+		if fss.firstNeedsUpdate || fss.lastNeedsUpdate {
+			ms.recalculateForSubj(bytesToString(subj), fss)
 		}
 		if sseq <= fss.First {
 			update(fss)
@@ -793,8 +793,8 @@ func (ms *memStore) enforcePerSubjectLimit(subj string, ss *SimpleState) {
 		return
 	}
 	for nmsgs := ss.Msgs; nmsgs > uint64(ms.maxp); nmsgs = ss.Msgs {
-		if ss.firstNeedsUpdate {
-			ms.recalculateFirstForSubj(subj, ss.First, ss)
+		if ss.firstNeedsUpdate || ss.lastNeedsUpdate {
+			ms.recalculateForSubj(subj, ss)
 		}
 		if !ms.removeMsg(ss.First, false) {
 			break
@@ -1009,8 +1009,9 @@ func (ms *memStore) Compact(seq uint64) (uint64, error) {
 			if sm := ms.msgs[seq]; sm != nil {
 				bytes += memStoreMsgSize(sm.subj, sm.hdr, sm.msg)
 				purged++
-				delete(ms.msgs, seq)
 				ms.removeSeqPerSubject(sm.subj, seq)
+				// Must delete message after updating per-subject info, to be consistent with file store.
+				delete(ms.msgs, seq)
 			}
 		}
 		if purged > ms.state.Msgs {
@@ -1098,8 +1099,9 @@ func (ms *memStore) Truncate(seq uint64) error {
 		if sm := ms.msgs[i]; sm != nil {
 			purged++
 			bytes += memStoreMsgSize(sm.subj, sm.hdr, sm.msg)
-			delete(ms.msgs, i)
 			ms.removeSeqPerSubject(sm.subj, i)
+			// Must delete message after updating per-subject info, to be consistent with file store.
+			delete(ms.msgs, i)
 		}
 	}
 	// Reset last.
@@ -1265,8 +1267,8 @@ func (ms *memStore) LoadNextMsg(filter string, wc bool, start uint64, smp *Store
 			if !ok {
 				continue
 			}
-			if ss.firstNeedsUpdate {
-				ms.recalculateFirstForSubj(subj, ss.First, ss)
+			if ss.firstNeedsUpdate || ss.lastNeedsUpdate {
+				ms.recalculateForSubj(subj, ss)
 			}
 			if ss.First < fseq {
 				fseq = ss.First
@@ -1360,34 +1362,47 @@ func (ms *memStore) removeSeqPerSubject(subj string, seq uint64) {
 	}
 	ss.Msgs--
 
-	// If we know we only have 1 msg left don't need to search for next first.
-	if ss.Msgs == 1 {
-		if seq == ss.Last {
-			ss.Last = ss.First
-		} else {
-			ss.First = ss.Last
-		}
-		ss.firstNeedsUpdate = false
-	} else {
-		ss.firstNeedsUpdate = seq == ss.First || ss.firstNeedsUpdate
-	}
+	// We can lazily calculate the first/last sequence when needed.
+	ss.firstNeedsUpdate = seq == ss.First || ss.firstNeedsUpdate
+	ss.lastNeedsUpdate = seq == ss.Last || ss.lastNeedsUpdate
 }
 
-// Will recalculate the first sequence for this subject in this block.
+// Will recalculate the first and/or last sequence for this subject.
 // Lock should be held.
-func (ms *memStore) recalculateFirstForSubj(subj string, startSeq uint64, ss *SimpleState) {
-	tseq := startSeq + 1
-	if tseq < ms.state.FirstSeq {
-		tseq = ms.state.FirstSeq
-	}
-	for ; tseq <= ss.Last; tseq++ {
-		if sm := ms.msgs[tseq]; sm != nil && sm.subj == subj {
-			ss.First = tseq
-			if ss.Msgs == 1 {
+func (ms *memStore) recalculateForSubj(subj string, ss *SimpleState) {
+	if ss.firstNeedsUpdate {
+		tseq := ss.First + 1
+		if tseq < ms.state.FirstSeq {
+			tseq = ms.state.FirstSeq
+		}
+		for ; tseq <= ss.Last; tseq++ {
+			if sm := ms.msgs[tseq]; sm != nil && sm.subj == subj {
+				ss.First = tseq
+				ss.firstNeedsUpdate = false
+				if ss.Msgs == 1 {
+					ss.Last = tseq
+					ss.lastNeedsUpdate = false
+					return
+				}
+				break
+			}
+		}
+	}
+	if ss.lastNeedsUpdate {
+		tseq := ss.Last - 1
+		if tseq > ms.state.LastSeq {
+			tseq = ms.state.LastSeq
+		}
+		for ; tseq >= ss.First; tseq-- {
+			if sm := ms.msgs[tseq]; sm != nil && sm.subj == subj {
 				ss.Last = tseq
+				ss.lastNeedsUpdate = false
+				if ss.Msgs == 1 {
+					ss.First = tseq
+					ss.firstNeedsUpdate = false
+				}
+				return
 			}
-			ss.firstNeedsUpdate = false
-			return
 		}
 	}
 }
@@ -1403,7 +1418,6 @@ func (ms *memStore) removeMsg(seq uint64, secure bool) bool {
 
 	ss = memStoreMsgSize(sm.subj, sm.hdr, sm.msg)
 
-	delete(ms.msgs, seq)
 	if ms.state.Msgs > 0 {
 		ms.state.Msgs--
 		if ss > ms.state.Bytes {
@@ -1428,6 +1442,8 @@ func (ms *memStore) removeMsg(seq uint64, secure bool) bool {
 
 	// Remove any per subject tracking.
 	ms.removeSeqPerSubject(sm.subj, seq)
+	// Must delete message after updating per-subject info, to be consistent with file store.
+	delete(ms.msgs, seq)
 
 	if ms.scb != nil {
 		// We do not want to hold any locks here.
diff --git a/server/raft.go b/server/raft.go
index 64d8e4df..fb70ee77 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -61,6 +61,7 @@ type RaftNode interface {
 	ID() string
 	Group() string
 	Peers() []*Peer
+	ProposeKnownPeers(knownPeers []string)
 	UpdateKnownPeers(knownPeers []string)
 	ProposeAddPeer(peer string) error
 	ProposeRemovePeer(peer string) error
@@ -1326,6 +1327,12 @@ func (n *raft) isCurrent(includeForwardProgress bool) bool {
 		return false
 	}
 
+	if n.paused && n.hcommit > n.commit {
+		// We're currently paused, waiting to be resumed to apply pending commits.
+		n.debug("Not current, waiting to resume applies commit=%d, hcommit=%d", n.commit, n.hcommit)
+		return false
+	}
+
 	if n.commit == n.applied {
 		// At this point if we are current, we can return saying so.
 		clearBehindState()
@@ -1556,14 +1563,12 @@ func (n *raft) ID() string {
 	if n == nil {
 		return _EMPTY_
 	}
-	n.RLock()
-	defer n.RUnlock()
+	// Lock not needed as n.id is never changed after creation.
 	return n.id
 }
 
 func (n *raft) Group() string {
-	n.RLock()
-	defer n.RUnlock()
+	// Lock not needed as n.group is never changed after creation.
 	return n.group
 }
 
@@ -1588,19 +1593,23 @@ func (n *raft) Peers() []*Peer {
 	return peers
 }
 
+// Update and propose our known set of peers.
+func (n *raft) ProposeKnownPeers(knownPeers []string) {
+	// If we are the leader update and send this update out.
+	if n.State() != Leader {
+		return
+	}
+	n.UpdateKnownPeers(knownPeers)
+	n.sendPeerState()
+}
+
 // Update our known set of peers.
 func (n *raft) UpdateKnownPeers(knownPeers []string) {
 	n.Lock()
 	// Process like peer state update.
 	ps := &peerState{knownPeers, len(knownPeers), n.extSt}
 	n.processPeerState(ps)
-	isLeader := n.State() == Leader
 	n.Unlock()
-
-	// If we are the leader send this update out as well.
-	if isLeader {
-		n.sendPeerState()
-	}
 }
 
 // ApplyQ returns the apply queue that new commits will be sent to for the
@@ -1615,8 +1624,7 @@ func (n *raft) LeadChangeC() <-chan bool { return n.leadc }
 func (n *raft) QuitC() <-chan struct{} { return n.quit }
 
 func (n *raft) Created() time.Time {
-	n.RLock()
-	defer n.RUnlock()
+	// Lock not needed as n.created is never changed after creation.
 	return n.created
 }
 
diff --git a/server/store.go b/server/store.go
index 72e03981..1c8f7f7e 100644
--- a/server/store.go
+++ b/server/store.go
@@ -166,6 +166,8 @@ type SimpleState struct {
 
 	// Internal usage for when the first needs to be updated before use.
 	firstNeedsUpdate bool
+	// Internal usage for when the last needs to be updated before use.
+	lastNeedsUpdate bool
 }
 
 // LostStreamData indicates msgs that have been lost.
diff --git a/server/stream.go b/server/stream.go
index a3a7c8fd..e34f8cd4 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -1039,10 +1039,10 @@ func (mset *stream) lastSeq() uint64 {
 	return mset.lseq
 }
 
+// Set last seq.
+// Write lock should be held.
 func (mset *stream) setLastSeq(lseq uint64) {
-	mset.mu.Lock()
 	mset.lseq = lseq
-	mset.mu.Unlock()
 }
 
 func (mset *stream) sendCreateAdvisory() {
@@ -2051,11 +2051,16 @@ func (mset *stream) purge(preq *JSApiStreamPurgeRequest) (purged uint64, err err
 	store.FastState(&state)
 	fseq, lseq := state.FirstSeq, state.LastSeq
 
+	mset.mu.Lock()
 	// Check if our last has moved past what our original last sequence was, if so reset.
 	if lseq > mlseq {
 		mset.setLastSeq(lseq)
 	}
 
+	// Clear any pending acks below first seq.
+	mset.clearAllPreAcksBelowFloor(fseq)
+	mset.mu.Unlock()
+
 	// Purge consumers.
 	// Check for filtered purge.
 	if preq != nil && preq.Subject != _EMPTY_ {
@@ -2102,7 +2107,14 @@ func (mset *stream) deleteMsg(seq uint64) (bool, error) {
 	if mset.closed.Load() {
 		return false, errStreamClosed
 	}
-	return mset.store.RemoveMsg(seq)
+	removed, err := mset.store.RemoveMsg(seq)
+	if err != nil {
+		return removed, err
+	}
+	mset.mu.Lock()
+	mset.clearAllPreAcks(seq)
+	mset.mu.Unlock()
+	return removed, err
 }
 
 // EraseMsg will securely remove a message and rewrite the data with random data.
@@ -2110,7 +2122,14 @@ func (mset *stream) eraseMsg(seq uint64) (bool, error) {
 	if mset.closed.Load() {
 		return false, errStreamClosed
 	}
-	return mset.store.EraseMsg(seq)
+	removed, err := mset.store.EraseMsg(seq)
+	if err != nil {
+		return removed, err
+	}
+	mset.mu.Lock()
+	mset.clearAllPreAcks(seq)
+	mset.mu.Unlock()
+	return removed, err
 }
 
 // Are we a mirror?
@@ -4000,15 +4019,8 @@ func (mset *stream) purgeMsgIds() {
 	}
 }
 
-// storeMsgId will store the message id for duplicate detection.
-func (mset *stream) storeMsgId(dde *ddentry) {
-	mset.mu.Lock()
-	defer mset.mu.Unlock()
-	mset.storeMsgIdLocked(dde)
-}
-
 // storeMsgIdLocked will store the message id for duplicate detection.
-// Lock should he held.
+// Lock should be held.
 func (mset *stream) storeMsgIdLocked(dde *ddentry) {
 	if mset.ddmap == nil {
 		mset.ddmap = make(map[string]*ddentry)


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 1be968f8..458cef7a 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -5030,7 +5030,7 @@ func TestFileStoreRecaluclateFirstForSubjBug(t *testing.T) {
 	mb.clearCacheAndOffset()
 	// Now call with start sequence of 1, the old one
 	// This will panic without the fix.
-	mb.recalculateFirstForSubj("foo", 1, ss)
+	mb.recalculateForSubj("foo", ss)
 	// Make sure it was update properly.
 	require_True(t, *ss == SimpleState{Msgs: 1, First: 3, Last: 3, firstNeedsUpdate: false})
 }
diff --git a/server/jetstream_cluster_1_test.go b/server/jetstream_cluster_1_test.go
index 89283b5a..fefbc25d 100644
--- a/server/jetstream_cluster_1_test.go
+++ b/server/jetstream_cluster_1_test.go
@@ -6857,6 +6857,153 @@ func TestJetStreamClusterConsumerInfoAfterCreate(t *testing.T) {
 	require_NoError(t, err)
 }
 
+func TestJetStreamClusterStreamUpscalePeersAfterDownscale(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R5S", 5)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	checkPeerSet := func() {
+		t.Helper()
+		checkFor(t, 5*time.Second, 500*time.Millisecond, func() error {
+			for _, s := range c.servers {
+				acc, err := s.lookupAccount(globalAccountName)
+				if err != nil {
+					return err
+				}
+				mset, err := acc.lookupStream("TEST")
+				if err != nil {
+					return err
+				}
+				peers := mset.raftNode().Peers()
+				if len(peers) != 5 {
+					return fmt.Errorf("expected 5 peers, got %d", len(peers))
+				}
+			}
+			return nil
+		})
+	}
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 5,
+	})
+	require_NoError(t, err)
+
+	checkPeerSet()
+
+	_, err = js.UpdateStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	_, err = js.UpdateStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Replicas: 5,
+	})
+	require_NoError(t, err)
+
+	checkPeerSet()
+}
+
+func TestJetStreamClusterClearAllPreAcksOnRemoveMsg(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"},
+		Replicas:  3,
+		Retention: nats.WorkQueuePolicy,
+	})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+		Durable:   "CONSUMER",
+		AckPolicy: nats.AckExplicitPolicy,
+	})
+	require_NoError(t, err)
+
+	for i := 0; i < 3; i++ {
+		_, err = js.Publish("foo", nil)
+		require_NoError(t, err)
+	}
+
+	// Wait for all servers to converge on the same state.
+	checkFor(t, 5*time.Second, 500*time.Millisecond, func() error {
+		return checkState(t, c, globalAccountName, "TEST")
+	})
+
+	// Register pre-acks on all servers.
+	// Normally this can't happen as the stream leader will have the message that's acked available, just for testing.
+	for _, s := range c.servers {
+		acc, err := s.lookupAccount(globalAccountName)
+		require_NoError(t, err)
+		mset, err := acc.lookupStream("TEST")
+		require_NoError(t, err)
+		o := mset.lookupConsumer("CONSUMER")
+		require_NotNil(t, o)
+
+		// Register pre-acks for the 3 messages.
+		mset.registerPreAckLock(o, 1)
+		mset.registerPreAckLock(o, 2)
+		mset.registerPreAckLock(o, 3)
+	}
+
+	// Check there's an expected amount of pre-acks, and there are no pre-acks for the given sequence.
+	checkPreAcks := func(seq uint64, expected int) {
+		t.Helper()
+		checkFor(t, 5*time.Second, time.Second, func() error {
+			for _, s := range c.servers {
+				acc, err := s.lookupAccount(globalAccountName)
+				if err != nil {
+					return err
+				}
+				mset, err := acc.lookupStream("TEST")
+				if err != nil {
+					return err
+				}
+				mset.mu.RLock()
+				numPreAcks := len(mset.preAcks)
+				numSeqPreAcks := len(mset.preAcks[seq])
+				mset.mu.RUnlock()
+				if numPreAcks != expected {
+					return fmt.Errorf("expected %d pre-acks, got %d", expected, numPreAcks)
+				}
+				if seq > 0 && numSeqPreAcks != 0 {
+					return fmt.Errorf("expected 0 pre-acks for seq %d, got %d", seq, numSeqPreAcks)
+				}
+			}
+			return nil
+		})
+	}
+	// Check all pre-acks were registered.
+	checkPreAcks(0, 3)
+
+	// Deleting the message should clear the pre-ack.
+	err = js.DeleteMsg("TEST", 1)
+	require_NoError(t, err)
+	checkPreAcks(1, 2)
+
+	// Erasing the message should clear the pre-ack.
+	err = js.SecureDeleteMsg("TEST", 2)
+	require_NoError(t, err)
+	checkPreAcks(2, 1)
+
+	// Purging should clear all pre-acks below the purged floor.
+	err = js.PurgeStream("TEST", &nats.StreamPurgeRequest{Sequence: 4})
+	require_NoError(t, err)
+	checkPreAcks(3, 0)
+}
+
 //
 // DO NOT ADD NEW TESTS IN THIS FILE (unless to balance test times)
 // Add at the end of jetstream_cluster_<n>_test.go, with <n> being the highest value.
diff --git a/server/jetstream_cluster_2_test.go b/server/jetstream_cluster_2_test.go
index 5807786e..f4a52042 100644
--- a/server/jetstream_cluster_2_test.go
+++ b/server/jetstream_cluster_2_test.go
@@ -4219,7 +4219,7 @@ func TestJetStreamClusterRedeliverBackoffs(t *testing.T) {
 		d := tr.Sub(start)
 		// Adjust start for next calcs.
 		start = start.Add(d)
-		if d < expected[i] || d > expected[i]*2 {
+		if d < expected[i]-5*time.Millisecond || d > expected[i]*2+5*time.Millisecond {
 			t.Fatalf("Timing is off for %d, expected ~%v, but got %v", i, expected[i], d)
 		}
 	}
@@ -6238,8 +6238,12 @@ func TestJetStreamClusterStreamResetOnExpirationDuringPeerDownAndRestartWithLead
 
 	// Wait for all messages to expire.
 	checkFor(t, 5*time.Second, time.Second, func() error {
-		si, err := js.StreamInfo("TEST")
-		require_NoError(t, err)
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		si, err := js.StreamInfo("TEST", nats.Context(ctx))
+		if err != nil {
+			return err
+		}
 		if si.State.Msgs == 0 {
 			return nil
 		}
@@ -6277,9 +6281,11 @@ func TestJetStreamClusterStreamResetOnExpirationDuringPeerDownAndRestartWithLead
 	}
 
 	// Now move the leader there and double check, but above test is sufficient.
-	checkFor(t, 10*time.Second, 250*time.Millisecond, func() error {
+	checkFor(t, 30*time.Second, 250*time.Millisecond, func() error {
 		_, err = nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, "TEST"), nil, time.Second)
-		require_NoError(t, err)
+		if err != nil {
+			return err
+		}
 		c.waitOnStreamLeader("$G", "TEST")
 		if c.streamLeader("$G", "TEST") == nsl {
 			return nil
diff --git a/server/jetstream_cluster_3_test.go b/server/jetstream_cluster_3_test.go
index 25d1ebea..0898da0e 100644
--- a/server/jetstream_cluster_3_test.go
+++ b/server/jetstream_cluster_3_test.go
@@ -1600,10 +1600,12 @@ func TestJetStreamClusterParallelConsumerCreation(t *testing.T) {
 }
 
 func TestJetStreamClusterGhostEphemeralsAfterRestart(t *testing.T) {
-	consumerNotActiveStartInterval = time.Second * 5
-	defer func() {
+	consumerNotActiveStartInterval = time.Second
+	consumerNotActiveMaxInterval = time.Second
+	t.Cleanup(func() {
 		consumerNotActiveStartInterval = defaultConsumerNotActiveStartInterval
-	}()
+		consumerNotActiveMaxInterval = defaultConsumerNotActiveMaxInterval
+	})
 
 	c := createJetStreamClusterExplicit(t, "R3S", 3)
 	defer c.shutdown()
diff --git a/server/jetstream_cluster_4_test.go b/server/jetstream_cluster_4_test.go
index f0145d21..4988c30f 100644
--- a/server/jetstream_cluster_4_test.go
+++ b/server/jetstream_cluster_4_test.go
@@ -669,13 +669,6 @@ func TestJetStreamClusterStreamOrphanMsgsAndReplicasDrifting(t *testing.T) {
 		// 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})
@@ -738,7 +731,7 @@ func TestJetStreamClusterStreamOrphanMsgsAndReplicasDrifting(t *testing.T) {
 		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.
+			// Gather the stream mset from each replica.
 			var msets []*stream
 			for _, s := range c.servers {
 				acc, err := s.LookupAccount("js")
@@ -748,14 +741,30 @@ func TestJetStreamClusterStreamOrphanMsgsAndReplicasDrifting(t *testing.T) {
 				msets = append(msets, mset)
 			}
 			for seq := state.FirstSeq; seq <= state.LastSeq; seq++ {
+				var expectedErr error
 				var msgId string
 				var smv StoreMsg
-				for _, mset := range msets {
+				for i, mset := range msets {
 					mset.mu.RLock()
 					sm, err := mset.store.LoadMsg(seq, &smv)
 					mset.mu.RUnlock()
-					require_NoError(t, err)
-					if msgId == _EMPTY_ {
+					if err != nil || expectedErr != nil {
+						// If one of the msets reports an error for LoadMsg for this
+						// particular sequence, then the same error should be reported
+						// by all msets for that seq to prove consistency across replicas.
+						// If one of the msets either returns no error or doesn't return
+						// the same error, then that replica has drifted.
+						if msgId != _EMPTY_ {
+							t.Fatalf("Expected MsgId %q for seq %d, but got error: %v", msgId, seq, err)
+						} else if expectedErr == nil {
+							expectedErr = err
+						} else {
+							require_Error(t, err, expectedErr)
+						}
+						continue
+					}
+					// Only set expected msg ID if it's for the very first time.
+					if msgId == _EMPTY_ && i == 0 {
 						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)
@@ -764,22 +773,13 @@ func TestJetStreamClusterStreamOrphanMsgsAndReplicasDrifting(t *testing.T) {
 			}
 		}
 
-		// 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)
-			}
-		}
+		// Wait for test to finish before checking state.
+		wg.Wait()
 
 		// 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 {
+			// If we have drifted do not have to wait too long, usually it's stuck for good.
+			checkFor(t, 5*time.Minute, time.Second, func() error {
 				return checkState(t)
 			})
 			// If we succeeded now let's check that all messages are also the same.
@@ -788,7 +788,39 @@ func TestJetStreamClusterStreamOrphanMsgsAndReplicasDrifting(t *testing.T) {
 			checkMsgsEqual(t)
 		}
 
-		wg.Wait()
+		err = checkForErr(2*time.Minute, time.Second, func() error {
+			var consumerPending int
+			consumers := make(map[string]int)
+			for i := 0; i < 10; i++ {
+				consumerName := fmt.Sprintf("consumer:EEEEE:%d", i)
+				ci, err := js.ConsumerInfo(sc.Name, consumerName)
+				if err != nil {
+					return err
+				}
+				pending := int(ci.NumPending)
+				consumers[consumerName] = pending
+				consumerPending += pending
+			}
+
+			// Only check if there are any pending messages.
+			if consumerPending > 0 {
+				// Check state of streams and consumers.
+				si, err := js.StreamInfo(sc.Name, &nats.StreamInfoRequest{SubjectsFilter: ">"})
+				if err != nil {
+					return err
+				}
+				streamPending := int(si.State.Msgs)
+				// FIXME: Num pending can be out of sync from the number of stream messages in the subject.
+				if streamPending != consumerPending {
+					return fmt.Errorf("Unexpected number of pending messages, stream=%d, consumers=%d \n subjects: %+v\nconsumers: %+v",
+						streamPending, consumerPending, si.State.Subjects, consumers)
+				}
+			}
+			return nil
+		})
+		if err != nil {
+			t.Logf("WRN: %v", err)
+		}
 	}
 
 	// Setting up test variations below:
@@ -3571,7 +3603,9 @@ func TestJetStreamClusterDesyncAfterErrorDuringCatchup(t *testing.T) {
 				for _, n := range server.raftNodes {
 					rn := n.(*raft)
 					if rn.accName == "$G" {
+						rn.Lock()
 						rn.updateLeader(noLeader)
+						rn.Unlock()
 					}
 				}
 
@@ -4392,3 +4426,100 @@ func TestJetStreamClusterStreamConsumerStateResetAfterRecreate(t *testing.T) {
 	}
 	wg.Wait()
 }
+
+func TestJetStreamClusterOnlyPublishAdvisoriesWhenInterest(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	subj := "$JS.ADVISORY.TEST"
+	s1 := c.servers[0]
+	s2 := c.servers[1]
+
+	// On the first server, see if we think the advisory will be published.
+	require_False(t, s1.publishAdvisory(s1.GlobalAccount(), subj, "test"))
+
+	// On the second server, subscribe to the advisory subject.
+	nc, _ := jsClientConnect(t, s2)
+	defer nc.Close()
+
+	_, err := nc.Subscribe(subj, func(_ *nats.Msg) {})
+	require_NoError(t, err)
+
+	// Wait for the interest to propagate to the first server.
+	checkFor(t, time.Second, 25*time.Millisecond, func() error {
+		if !s1.GlobalAccount().sl.HasInterest(subj) {
+			return fmt.Errorf("expected interest in %q, not yet found", subj)
+		}
+		return nil
+	})
+
+	// On the first server, try and publish the advisory again. THis time
+	// it should succeed.
+	require_True(t, s1.publishAdvisory(s1.GlobalAccount(), subj, "test"))
+}
+
+func TestJetStreamClusterRoutedAPIRecoverPerformance(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, _ := jsClientConnect(t, c.randomNonLeader())
+	defer nc.Close()
+
+	// We only run 16 JetStream API workers.
+	mp := runtime.GOMAXPROCS(0)
+	if mp > 16 {
+		mp = 16
+	}
+
+	leader := c.leader()
+	ljs := leader.js.Load()
+
+	// Take the JS lock, which allows the JS API queue to build up.
+	ljs.mu.Lock()
+	defer ljs.mu.Unlock()
+
+	count := JSDefaultRequestQueueLimit - 1
+	ch := make(chan *nats.Msg, count)
+
+	inbox := nc.NewRespInbox()
+	_, err := nc.ChanSubscribe(inbox, ch)
+	require_NoError(t, err)
+
+	// To ensure a fair starting line, we need to submit as many tasks as
+	// there are JS workers whilst holding the JS lock. This will ensure that
+	// each JS API worker is properly wedged.
+	msg := &nats.Msg{
+		Subject: fmt.Sprintf(JSApiConsumerInfoT, "Doesnt", "Exist"),
+		Reply:   "no_one_here",
+	}
+	for i := 0; i < mp; i++ {
+		require_NoError(t, nc.PublishMsg(msg))
+	}
+
+	// Then we want to submit a fixed number of tasks, big enough to fill
+	// the queue, so that we can measure them.
+	msg = &nats.Msg{
+		Subject: fmt.Sprintf(JSApiConsumerInfoT, "Doesnt", "Exist"),
+		Reply:   inbox,
+	}
+	for i := 0; i < count; i++ {
+		require_NoError(t, nc.PublishMsg(msg))
+	}
+	checkFor(t, 5*time.Second, 25*time.Millisecond, func() error {
+		if queued := leader.jsAPIRoutedReqs.len(); queued != count {
+			return fmt.Errorf("expected %d queued requests, got %d", count, queued)
+		}
+		return nil
+	})
+
+	// Now we're going to release the lock and start timing. The workers
+	// will now race to clear the queues and we'll wait to see how long
+	// it takes for them all to respond.
+	start := time.Now()
+	ljs.mu.Unlock()
+	for i := 0; i < count; i++ {
+		<-ch
+	}
+	ljs.mu.Lock()
+	t.Logf("Took %s to clear %d items", time.Since(start), count)
+}
diff --git a/server/jetstream_helpers_test.go b/server/jetstream_helpers_test.go
index eb21057a..622c5894 100644
--- a/server/jetstream_helpers_test.go
+++ b/server/jetstream_helpers_test.go
@@ -556,6 +556,27 @@ func (sc *supercluster) waitOnLeader() {
 	sc.t.Fatalf("Expected a cluster leader, got none")
 }
 
+func (sc *supercluster) waitOnAccount(account string) {
+	sc.t.Helper()
+	expires := time.Now().Add(40 * time.Second)
+	for time.Now().Before(expires) {
+		found := true
+		for _, c := range sc.clusters {
+			for _, s := range c.servers {
+				acc, err := s.fetchAccount(account)
+				found = found && err == nil && acc != nil
+			}
+		}
+		if found {
+			return
+		}
+		time.Sleep(100 * time.Millisecond)
+		continue
+	}
+
+	sc.t.Fatalf("Expected account %q to exist but didn't", account)
+}
+
 func (sc *supercluster) waitOnAllCurrent() {
 	sc.t.Helper()
 	for _, c := range sc.clusters {
diff --git a/server/jetstream_jwt_test.go b/server/jetstream_jwt_test.go
index f8f6466f..480964d4 100644
--- a/server/jetstream_jwt_test.go
+++ b/server/jetstream_jwt_test.go
@@ -325,6 +325,8 @@ func TestJetStreamJWTMove(t *testing.T) {
 		require_False(t, s.JetStreamEnabled())
 		updateJwt(t, s.ClientURL(), sysCreds, accJwt, 10)
 
+		sc.waitOnAccount(aExpPub)
+
 		s = sc.serverByName("C2-S1")
 		require_False(t, s.JetStreamEnabled())
 
@@ -609,6 +611,8 @@ func TestJetStreamJWTClusteredTiersChange(t *testing.T) {
 	updateJwt(t, c.randomServer().ClientURL(), sysCreds, sysJwt, 3)
 	updateJwt(t, c.randomServer().ClientURL(), sysCreds, accJwt1, 3)
 
+	c.waitOnAccount(aExpPub)
+
 	nc := natsConnect(t, c.randomServer().ClientURL(), nats.UserCredentials(accCreds))
 	defer nc.Close()
 
@@ -693,6 +697,8 @@ func TestJetStreamJWTClusteredDeleteTierWithStreamAndMove(t *testing.T) {
 	updateJwt(t, c.randomServer().ClientURL(), sysCreds, sysJwt, 3)
 	updateJwt(t, c.randomServer().ClientURL(), sysCreds, accJwt1, 3)
 
+	c.waitOnAccount(aExpPub)
+
 	nc := natsConnect(t, c.randomServer().ClientURL(), nats.UserCredentials(accCreds))
 	defer nc.Close()
 
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index c38d44be..f5a56247 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -8536,10 +8536,14 @@ func TestJetStreamNextMsgNoInterest(t *testing.T) {
 				}
 			}
 			nc.Flush()
-			ostate := o.info()
-			if ostate.AckFloor.Stream != 11 || ostate.NumAckPending > 0 {
-				t.Fatalf("Inconsistent ack state: %+v", ostate)
-			}
+
+			checkFor(t, time.Second, 10*time.Millisecond, func() error {
+				ostate := o.info()
+				if ostate.AckFloor.Stream != 11 || ostate.NumAckPending > 0 {
+					return fmt.Errorf("Inconsistent ack state: %+v", ostate)
+				}
+				return nil
+			})
 		})
 	}
 }
@@ -22997,3 +23001,131 @@ func TestJetStreamWouldExceedLimits(t *testing.T) {
 	require_True(t, js.wouldExceedLimits(MemoryStorage, int(js.config.MaxMemory)+1))
 	require_True(t, js.wouldExceedLimits(FileStorage, int(js.config.MaxStore)+1))
 }
+
+func TestJetStreamConsumerDontDecrementPendingCountOnSkippedMsg(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "TEST", Subjects: []string{"foo"}})
+	require_NoError(t, err)
+
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{Durable: "CONSUMER"})
+	require_NoError(t, err)
+
+	acc, err := s.lookupAccount(globalAccountName)
+	require_NoError(t, err)
+	mset, err := acc.lookupStream("TEST")
+	require_NoError(t, err)
+	o := mset.lookupConsumer("CONSUMER")
+
+	requireExpected := func(expected int64) {
+		t.Helper()
+		checkFor(t, time.Second, 10*time.Millisecond, func() error {
+			o.mu.RLock()
+			npc := o.npc
+			o.mu.RUnlock()
+			if npc != expected {
+				return fmt.Errorf("expected npc=%d, got %d", expected, npc)
+			}
+			return nil
+		})
+	}
+
+	// Should initially report no messages available.
+	requireExpected(0)
+
+	// A new message is available, should report in pending.
+	_, err = js.Publish("foo", nil)
+	require_NoError(t, err)
+	requireExpected(1)
+
+	// Pending count should decrease when the message is deleted.
+	err = js.DeleteMsg("TEST", 1)
+	require_NoError(t, err)
+	requireExpected(0)
+
+	// Make more messages available, should report in pending.
+	_, err = js.Publish("foo", nil)
+	require_NoError(t, err)
+	_, err = js.Publish("foo", nil)
+	require_NoError(t, err)
+	requireExpected(2)
+
+	// Simulate getNextMsg being called and the starting sequence to skip over a deleted message.
+	// Also simulate one pending message.
+	o.mu.Lock()
+	o.sseq = 100
+	o.npc--
+	o.pending = make(map[uint64]*Pending)
+	o.pending[2] = &Pending{}
+	o.mu.Unlock()
+
+	// Since this message is pending we should not decrement pending count as we've done so already.
+	o.decStreamPending(2, "foo")
+	requireExpected(1)
+
+	// This is the deleted message that was skipped, we've hit the race condition and are not able to
+	// fix it at this point. If we decrement then we could have decremented it twice if the message
+	// was removed as a result of an Ack with Interest or WorkQueue retention, instead of due to contention.
+	o.decStreamPending(3, "foo")
+	requireExpected(1)
+}
+
+func TestJetStreamConsumerPendingCountAfterMsgAckAboveFloor(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:      "TEST",
+		Subjects:  []string{"foo"},
+		Retention: nats.WorkQueuePolicy,
+	})
+	require_NoError(t, err)
+
+	for i := 0; i < 2; i++ {
+		_, err = js.Publish("foo", nil)
+		require_NoError(t, err)
+	}
+
+	sub, err := js.PullSubscribe("foo", "CONSUMER")
+	require_NoError(t, err)
+
+	acc, err := s.lookupAccount(globalAccountName)
+	require_NoError(t, err)
+	mset, err := acc.lookupStream("TEST")
+	require_NoError(t, err)
+	o := mset.lookupConsumer("CONSUMER")
+
+	requireExpected := func(expected int64) {
+		t.Helper()
+		checkFor(t, time.Second, 10*time.Millisecond, func() error {
+			o.mu.RLock()
+			npc := o.npc
+			o.mu.RUnlock()
+			if npc != expected {
+				return fmt.Errorf("expected npc=%d, got %d", expected, npc)
+			}
+			return nil
+		})
+	}
+
+	// Expect 2 messages pending.
+	requireExpected(2)
+
+	// Fetch 2 messages and ack the last.
+	msgs, err := sub.Fetch(2)
+	require_NoError(t, err)
+	require_Len(t, len(msgs), 2)
+	msg := msgs[1]
+	err = msg.AckSync()
+	require_NoError(t, err)
+
+	// We've fetched 2 message so should report 0 pending.
+	requireExpected(0)
+}
diff --git a/server/leafnode_test.go b/server/leafnode_test.go
index 9fb8743d..47d84eb2 100644
--- a/server/leafnode_test.go
+++ b/server/leafnode_test.go
@@ -8928,3 +8928,59 @@ func TestLeafCredFormatting(t *testing.T) {
 	runLeaf(t, creds)
 	runLeaf(t, bytes.ReplaceAll(creds, []byte{'\n'}, []byte{'\r', '\n'}))
 }
+
+func TestLeafNodePermissionWithLiteralSubjectAndQueueInterest(t *testing.T) {
+	hconf := createConfFile(t, []byte(`
+		server_name: "HUB"
+		listen: "127.0.0.1:-1"
+		leafnodes {
+			listen: "127.0.0.1:-1"
+		}
+		accounts {
+			A {
+				users: [
+					{ user: "user", password: "pwd",
+						permissions: {
+							subscribe: { allow: ["_INBOX.>", "my.subject"] }
+							publish: {allow: [">"]}
+						}
+					}
+				]
+			}
+		}
+	`))
+	hub, ohub := RunServerWithConfig(hconf)
+	defer hub.Shutdown()
+
+	lconf := createConfFile(t, []byte(fmt.Sprintf(`
+		server_name: "LEAF"
+		listen: "127.0.0.1:-1"
+		leafnodes {
+			remotes: [
+				{url: "nats://user:pwd@127.0.0.1:%d", account: A}
+			]
+		}
+		accounts {
+			A { users: [{user: user, password: pwd}] }
+		}
+	`, ohub.LeafNode.Port)))
+	leaf, _ := RunServerWithConfig(lconf)
+	defer leaf.Shutdown()
+
+	checkLeafNodeConnected(t, hub)
+	checkLeafNodeConnected(t, leaf)
+
+	ncLeaf := natsConnect(t, leaf.ClientURL(), nats.UserInfo("user", "pwd"))
+	defer ncLeaf.Close()
+	natsQueueSub(t, ncLeaf, "my.subject", "queue", func(m *nats.Msg) {
+		m.Respond([]byte("OK"))
+	})
+	natsFlush(t, ncLeaf)
+
+	ncHub := natsConnect(t, hub.ClientURL(), nats.UserInfo("user", "pwd"))
+	defer ncHub.Close()
+
+	resp, err := ncHub.Request("my.subject", []byte("hello"), time.Second)
+	require_NoError(t, err)
+	require_Equal(t, "OK", string(resp.Data))
+}
diff --git a/server/norace_test.go b/server/norace_test.go
index 5d54dfb1..cec10d7c 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -7728,32 +7728,47 @@ func TestNoRaceJetStreamClusterUnbalancedInterestMultipleConsumers(t *testing.T)
 	// make sure we do not remove prematurely.
 	msgs, err := sub.Fetch(100, nats.MaxWait(time.Second))
 	require_NoError(t, err)
-	require_True(t, len(msgs) == 100)
+	require_Len(t, len(msgs), 100)
 	for _, m := range msgs {
 		m.AckSync()
 	}
 
 	ci, err := js.ConsumerInfo("EVENTS", "D")
 	require_NoError(t, err)
-	require_True(t, ci.NumPending == uint64(numToSend-100))
-	require_True(t, ci.NumAckPending == 0)
-	require_True(t, ci.Delivered.Stream == 100)
-	require_True(t, ci.AckFloor.Stream == 100)
+	require_Equal(t, ci.NumPending, uint64(numToSend-100))
+	require_Equal(t, ci.NumAckPending, 0)
+	require_Equal(t, ci.Delivered.Stream, 100)
+	require_Equal(t, ci.AckFloor.Stream, 100)
 
 	// Check stream state on all servers.
-	for _, s := range c.servers {
-		mset, err := s.GlobalAccount().lookupStream("EVENTS")
-		require_NoError(t, err)
-		state := mset.state()
-		require_True(t, state.Msgs == 900)
-		require_True(t, state.FirstSeq == 101)
-		require_True(t, state.LastSeq == 1000)
-		require_True(t, state.Consumers == 2)
-	}
+	// Since acks result in messages to be removed through proposals,
+	// it could take some time to be reflected in the stream state.
+	checkFor(t, 5*time.Second, 500*time.Millisecond, func() error {
+		for _, s := range c.servers {
+			mset, err := s.GlobalAccount().lookupStream("EVENTS")
+			if err != nil {
+				return err
+			}
+			state := mset.state()
+			if state.Msgs != 900 {
+				return fmt.Errorf("expected state.Msgs=900, got %d", state.Msgs)
+			}
+			if state.FirstSeq != 101 {
+				return fmt.Errorf("expected state.FirstSeq=101, got %d", state.FirstSeq)
+			}
+			if state.LastSeq != 1000 {
+				return fmt.Errorf("expected state.LastSeq=1000, got %d", state.LastSeq)
+			}
+			if state.Consumers != 2 {
+				return fmt.Errorf("expected state.Consumers=2, got %d", state.Consumers)
+			}
+		}
+		return nil
+	})
 
 	msgs, err = sub.Fetch(900, nats.MaxWait(time.Second))
 	require_NoError(t, err)
-	require_True(t, len(msgs) == 900)
+	require_Len(t, len(msgs), 900)
 	for _, m := range msgs {
 		m.AckSync()
 	}
@@ -7766,15 +7781,15 @@ func TestNoRaceJetStreamClusterUnbalancedInterestMultipleConsumers(t *testing.T)
 		mset, err := s.GlobalAccount().lookupStream("EVENTS")
 		require_NoError(t, err)
 		state := mset.state()
-		require_True(t, state.Msgs == 0)
-		require_True(t, state.FirstSeq == 1001)
-		require_True(t, state.LastSeq == 1000)
-		require_True(t, state.Consumers == 2)
+		require_Equal(t, state.Msgs, 0)
+		require_Equal(t, state.FirstSeq, 1001)
+		require_Equal(t, state.LastSeq, 1000)
+		require_Equal(t, state.Consumers, 2)
 		// Now check preAcks
 		mset.mu.RLock()
 		numPreAcks := len(mset.preAcks)
 		mset.mu.RUnlock()
-		require_True(t, numPreAcks == 0)
+		require_Len(t, numPreAcks, 0)
 	}
 }
 
@@ -7862,27 +7877,27 @@ func TestNoRaceJetStreamClusterUnbalancedInterestMultipleFilteredConsumers(t *te
 
 	ci, err := js.ConsumerInfo("EVENTS", "D")
 	require_NoError(t, err)
-	require_True(t, ci.NumPending == 0)
-	require_True(t, ci.NumAckPending == 0)
-	require_True(t, ci.Delivered.Consumer == 500)
-	require_True(t, ci.Delivered.Stream == 1000)
-	require_True(t, ci.AckFloor.Consumer == 500)
-	require_True(t, ci.AckFloor.Stream == 1000)
+	require_Equal(t, ci.NumPending, 0)
+	require_Equal(t, ci.NumAckPending, 0)
+	require_Equal(t, ci.Delivered.Consumer, 500)
+	require_Equal(t, ci.Delivered.Stream, 1000)
+	require_Equal(t, ci.AckFloor.Consumer, 500)
+	require_Equal(t, ci.AckFloor.Stream, 1000)
 
 	// Check final stream state on all servers.
 	for _, s := range c.servers {
 		mset, err := s.GlobalAccount().lookupStream("EVENTS")
 		require_NoError(t, err)
 		state := mset.state()
-		require_True(t, state.Msgs == 0)
-		require_True(t, state.FirstSeq == 1001)
-		require_True(t, state.LastSeq == 1000)
-		require_True(t, state.Consumers == 2)
+		require_Equal(t, state.Msgs, 0)
+		require_Equal(t, state.FirstSeq, 1001)
+		require_Equal(t, state.LastSeq, 1000)
+		require_Equal(t, state.Consumers, 2)
 		// Now check preAcks
 		mset.mu.RLock()
 		numPreAcks := len(mset.preAcks)
 		mset.mu.RUnlock()
-		require_True(t, numPreAcks == 0)
+		require_Len(t, numPreAcks, 0)
 	}
 }
 
@@ -10717,27 +10732,42 @@ func TestNoRaceJetStreamClusterMemoryWorkQueueLastSequenceResetAfterRestart(t *t
 			nc, js := jsClientConnect(t, c.randomServer())
 			defer nc.Close()
 
-			_, err := js.AddStream(&nats.StreamConfig{
-				Name:      fmt.Sprintf("TEST:%d", n),
-				Storage:   nats.MemoryStorage,
-				Retention: nats.WorkQueuePolicy,
-				Subjects:  []string{fmt.Sprintf("foo.%d.*", n)},
-				Replicas:  3,
-			}, nats.MaxWait(30*time.Second))
-			require_NoError(t, err)
+			checkFor(t, 5*time.Second, time.Second, func() error {
+				_, err := js.AddStream(&nats.StreamConfig{
+					Name:      fmt.Sprintf("TEST:%d", n),
+					Storage:   nats.MemoryStorage,
+					Retention: nats.WorkQueuePolicy,
+					Subjects:  []string{fmt.Sprintf("foo.%d.*", n)},
+					Replicas:  3,
+				}, nats.MaxWait(time.Second))
+				return err
+			})
+
 			subj := fmt.Sprintf("foo.%d.bar", n)
 			for i := 0; i < 22; i++ {
-				js.Publish(subj, nil)
+				checkFor(t, 5*time.Second, time.Second, func() error {
+					_, err := js.Publish(subj, nil)
+					return err
+				})
 			}
-			// Now consumer them all as well.
-			sub, err := js.PullSubscribe(subj, "wq")
-			require_NoError(t, err)
-			msgs, err := sub.Fetch(22, nats.MaxWait(20*time.Second))
-			require_NoError(t, err)
+			// Now consume them all as well.
+			var err error
+			var sub *nats.Subscription
+			checkFor(t, 5*time.Second, time.Second, func() error {
+				sub, err = js.PullSubscribe(subj, "wq")
+				return err
+			})
+
+			var msgs []*nats.Msg
+			checkFor(t, 5*time.Second, time.Second, func() error {
+				msgs, err = sub.Fetch(22, nats.MaxWait(time.Second))
+				return err
+			})
 			require_Equal(t, len(msgs), 22)
 			for _, m := range msgs {
-				err := m.AckSync()
-				require_NoError(t, err)
+				checkFor(t, 5*time.Second, time.Second, func() error {
+					return m.AckSync()
+				})
 			}
 		}(i)
 	}
diff --git a/server/raft_test.go b/server/raft_test.go
index 30c3c188..fd4837f2 100644
--- a/server/raft_test.go
+++ b/server/raft_test.go
@@ -1896,3 +1896,52 @@ func TestNRGHealthCheckWaitForDoubleCatchup(t *testing.T) {
 	n.Applied(3)
 	require_True(t, n.Healthy())
 }
+
+func TestNRGHealthCheckWaitForPendingCommitsWhenPaused(t *testing.T) {
+	n, cleanup := initSingleMemRaftNode(t)
+	defer cleanup()
+
+	// Create a sample entry, the content doesn't matter, just that it's stored.
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	entries := []*Entry{newEntry(EntryNormal, esm)}
+
+	nats0 := "S1Nunr6R" // "nats-0"
+
+	// Timeline
+	aeMsg1 := encode(t, &appendEntry{leader: nats0, term: 1, commit: 0, pterm: 0, pindex: 0, entries: entries})
+	aeMsg2 := encode(t, &appendEntry{leader: nats0, term: 1, commit: 1, pterm: 1, pindex: 1, entries: entries})
+	aeHeartbeat := encode(t, &appendEntry{leader: nats0, term: 1, commit: 2, pterm: 1, pindex: 2, entries: nil})
+
+	// Process first message.
+	n.processAppendEntry(aeMsg1, n.aesub)
+	require_Equal(t, n.pindex, 1)
+	require_False(t, n.Healthy())
+
+	// Process second message, moves commit up.
+	n.processAppendEntry(aeMsg2, n.aesub)
+	require_Equal(t, n.pindex, 2)
+	require_False(t, n.Healthy())
+
+	// We're healthy once we've applied the first message.
+	n.Applied(1)
+	require_True(t, n.Healthy())
+
+	// If we're paused we still are healthy if there are no pending commits.
+	err := n.PauseApply()
+	require_NoError(t, err)
+	require_True(t, n.Healthy())
+
+	// Heartbeat marks second message to be committed.
+	n.processAppendEntry(aeHeartbeat, n.aesub)
+	require_Equal(t, n.pindex, 2)
+	require_False(t, n.Healthy())
+
+	// Resuming apply commits the message.
+	n.ResumeApply()
+	require_NoError(t, err)
+	require_False(t, n.Healthy())
+
+	// But still waiting for it to be applied before marking healthy.
+	n.Applied(2)
+	require_True(t, n.Healthy())
+}
diff --git a/server/store_test.go b/server/store_test.go
index e4470178..a916ceed 100644
--- a/server/store_test.go
+++ b/server/store_test.go
@@ -142,6 +142,123 @@ func TestStoreDeleteRange(t *testing.T) {
 	require_Equal(t, num, 1)
 }
 
+func TestStoreSubjectStateConsistency(t *testing.T) {
+	testAllStoreAllPermutations(
+		t, false,
+		StreamConfig{Name: "TEST", Subjects: []string{"foo"}},
+		func(t *testing.T, fs StreamStore) {
+			getSubjectState := func() SimpleState {
+				t.Helper()
+				ss := fs.SubjectsState("foo")
+				return ss["foo"]
+			}
+			var smp StoreMsg
+			expectFirstSeq := func(eseq uint64) {
+				t.Helper()
+				sm, _, err := fs.LoadNextMsg("foo", false, 0, &smp)
+				require_NoError(t, err)
+				require_Equal(t, sm.seq, eseq)
+			}
+			expectLastSeq := func(eseq uint64) {
+				t.Helper()
+				sm, err := fs.LoadLastMsg("foo", &smp)
+				require_NoError(t, err)
+				require_Equal(t, sm.seq, eseq)
+			}
+
+			// Publish an initial batch of messages.
+			for i := 0; i < 4; i++ {
+				_, _, err := fs.StoreMsg("foo", nil, nil)
+				require_NoError(t, err)
+			}
+
+			// Expect 4 msgs, with first=1, last=4.
+			ss := getSubjectState()
+			require_Equal(t, ss.Msgs, 4)
+			require_Equal(t, ss.First, 1)
+			expectFirstSeq(1)
+			require_Equal(t, ss.Last, 4)
+			expectLastSeq(4)
+
+			// Remove first message, ss.First is lazy so will only mark ss.firstNeedsUpdate.
+			removed, err := fs.RemoveMsg(1)
+			require_NoError(t, err)
+			require_True(t, removed)
+
+			// Will update first, so corrects to seq 2.
+			ss = getSubjectState()
+			require_Equal(t, ss.Msgs, 3)
+			require_Equal(t, ss.First, 2)
+			expectFirstSeq(2)
+			require_Equal(t, ss.Last, 4)
+			expectLastSeq(4)
+
+			// Remove last message, ss.Last is lazy so will only mark ss.lastNeedsUpdate.
+			removed, err = fs.RemoveMsg(4)
+			require_NoError(t, err)
+			require_True(t, removed)
+
+			// Will update last, so corrects to 3.
+			ss = getSubjectState()
+			require_Equal(t, ss.Msgs, 2)
+			require_Equal(t, ss.First, 2)
+			expectFirstSeq(2)
+			require_Equal(t, ss.Last, 3)
+			expectLastSeq(3)
+
+			// Remove first message again.
+			removed, err = fs.RemoveMsg(2)
+			require_NoError(t, err)
+			require_True(t, removed)
+
+			// Since we only have one message left, must update ss.First and ensure ss.Last equals.
+			ss = getSubjectState()
+			require_Equal(t, ss.Msgs, 1)
+			require_Equal(t, ss.First, 3)
+			expectFirstSeq(3)
+			require_Equal(t, ss.Last, 3)
+			expectLastSeq(3)
+
+			// Publish some more messages so we can test another scenario.
+			for i := 0; i < 3; i++ {
+				_, _, err := fs.StoreMsg("foo", nil, nil)
+				require_NoError(t, err)
+			}
+
+			// Just check the state is complete again.
+			ss = getSubjectState()
+			require_Equal(t, ss.Msgs, 4)
+			require_Equal(t, ss.First, 3)
+			expectFirstSeq(3)
+			require_Equal(t, ss.Last, 7)
+			expectLastSeq(7)
+
+			// Remove last sequence, ss.Last is lazy so doesn't get updated.
+			removed, err = fs.RemoveMsg(7)
+			require_NoError(t, err)
+			require_True(t, removed)
+
+			// Remove first sequence, ss.First is lazy so doesn't get updated.
+			removed, err = fs.RemoveMsg(3)
+			require_NoError(t, err)
+			require_True(t, removed)
+
+			// Remove (now) first sequence. Both ss.First and ss.Last are lazy and both need to be recalculated later.
+			removed, err = fs.RemoveMsg(5)
+			require_NoError(t, err)
+			require_True(t, removed)
+
+			// ss.First and ss.Last should both be recalculated and equal each other.
+			ss = getSubjectState()
+			require_Equal(t, ss.Msgs, 1)
+			require_Equal(t, ss.First, 6)
+			expectFirstSeq(6)
+			require_Equal(t, ss.Last, 6)
+			expectLastSeq(6)
+		},
+	)
+}
+
 func TestStoreMaxMsgsPerUpdateBug(t *testing.T) {
 	config := func() StreamConfig {
 		return StreamConfig{Name: "TEST", Subjects: []string{"foo"}, MaxMsgsPer: 0}