Repository: nats-io/nats-server
Pull request #4931 (2024-01-09 15:21:49)

Cherry-pick PRs into v2.10.8 release branch
Cherry-pick the following PRs into the 2.10.8 release branch:

* #4918 
* #4923
* #4925
* #4926
* #4927
* #4929 
* #4933
* #4835

Signed-off-by: Neil Twigg <neil@nats.io>


Patch:
diff --git a/README.md b/README.md
index a81b5b63..4a3d59d4 100644
--- a/README.md
+++ b/README.md
@@ -4,7 +4,7 @@
 
 [NATS](https://nats.io) is a simple, secure and performant communications system for digital systems, services and devices. NATS is part of the Cloud Native Computing Foundation ([CNCF](https://cncf.io)). NATS has over [40 client language implementations](https://nats.io/download/), and its server can run on-premise, in the cloud, at the edge, and even on a Raspberry Pi. NATS can secure and simplify design and operation of modern distributed systems.
 
-[![License][License-Image]][License-Url] [![Build][Build-Status-Image]][Build-Status-Url] [![Release][Release-Image]][Release-Url] [![Slack][Slack-Image]][Slack-Url] [![Coverage][Coverage-Image]][Coverage-Url] [![Docker Downloads][Docker-Image]][Docker-Url] [![CII Best Practices](https://bestpractices.coreinfrastructure.org/projects/1895/badge)](https://bestpractices.coreinfrastructure.org/projects/1895)
+[![License][License-Image]][License-Url] [![Build][Build-Status-Image]][Build-Status-Url] [![Release][Release-Image]][Release-Url] [![Slack][Slack-Image]][Slack-Url] [![Coverage][Coverage-Image]][Coverage-Url] [![Docker Downloads][Docker-Image]][Docker-Url] [![CII Best Practices][CIIBestPractices-Image]][CIIBestPractices-Url] [![Artifact Hub][ArtifactHub-Image]][ArtifactHub-Url]
 
 ## Documentation
 
@@ -43,6 +43,10 @@ If you are interested in contributing to NATS, read about our...
 [Coverage-image]: https://coveralls.io/repos/github/nats-io/nats-server/badge.svg?branch=main
 [ReportCard-Url]: https://goreportcard.com/report/nats-io/nats-server
 [ReportCard-Image]: https://goreportcard.com/badge/github.com/nats-io/nats-server
+[CIIBestPractices-Url]: https://bestpractices.coreinfrastructure.org/projects/1895
+[CIIBestPractices-Image]: https://bestpractices.coreinfrastructure.org/projects/1895/badge
+[ArtifactHub-Url]: https://artifacthub.io/packages/helm/nats/nats
+[ArtifactHub-Image]: https://img.shields.io/endpoint?url=https://artifacthub.io/badge/repository/nats
 [github-release]: https://github.com/nats-io/nats-server/releases/
 
 ## Roadmap
diff --git a/go.mod b/go.mod
index 725c659c..b295e667 100644
--- a/go.mod
+++ b/go.mod
@@ -7,10 +7,10 @@ require (
 	github.com/minio/highwayhash v1.0.2
 	github.com/nats-io/jwt/v2 v2.5.3
 	github.com/nats-io/nats.go v1.31.0
-	github.com/nats-io/nkeys v0.4.6
+	github.com/nats-io/nkeys v0.4.7
 	github.com/nats-io/nuid v1.0.1
 	go.uber.org/automaxprocs v1.5.3
 	golang.org/x/crypto v0.17.0
-	golang.org/x/sys v0.15.0
+	golang.org/x/sys v0.16.0
 	golang.org/x/time v0.5.0
 )
diff --git a/go.sum b/go.sum
index 9331e2d8..fd94d9da 100644
--- a/go.sum
+++ b/go.sum
@@ -7,8 +7,8 @@ github.com/nats-io/jwt/v2 v2.5.3 h1:/9SWvzc6hTfamcgXJ3uYRpgj+QuY2aLNqRiqrKcrpEo=
 github.com/nats-io/jwt/v2 v2.5.3/go.mod h1:iysuPemFcc7p4IoYots3IuELSI4EDe9Y0bQMe+I3Bf4=
 github.com/nats-io/nats.go v1.31.0 h1:/WFBHEc/dOKBF6qf1TZhrdEfTmOZ5JzdJ+Y3m6Y/p7E=
 github.com/nats-io/nats.go v1.31.0/go.mod h1:di3Bm5MLsoB4Bx61CBTsxuarI36WbhAwOm8QrW39+i8=
-github.com/nats-io/nkeys v0.4.6 h1:IzVe95ru2CT6ta874rt9saQRkWfe2nFj1NtvYSLqMzY=
-github.com/nats-io/nkeys v0.4.6/go.mod h1:4DxZNzenSVd1cYQoAa8948QY3QDjrHfcfVADymtkpts=
+github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI=
+github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -19,8 +19,8 @@ go.uber.org/automaxprocs v1.5.3/go.mod h1:eRbA25aqJrxAbsLO0xy5jVwPt7FQnRgjW+efnw
 golang.org/x/crypto v0.17.0 h1:r8bRNjWL3GshPW3gkd+RpvzWrZAwPS49OmTGZ/uhM4k=
 golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
 golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.15.0 h1:h48lPFYpsTvQJZF4EKyI4aLHaev3CxivZmv7yZig9pc=
-golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
+golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
 golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
diff --git a/server/consumer.go b/server/consumer.go
index 68f55418..02a8443e 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -1,4 +1,4 @@
-// Copyright 2019-2023 The NATS Authors
+// Copyright 2019-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
@@ -128,30 +128,37 @@ const (
 )
 
 const (
-	actionUpdateString         = "update"
-	actionCreateString         = "create"
-	actionCreateOrUpdateString = ""
+	actionUpdateJSONString         = `"update"`
+	actionCreateJSONString         = `"create"`
+	actionCreateOrUpdateJSONString = `""`
+)
+
+var (
+	actionUpdateJSONBytes         = []byte(actionUpdateJSONString)
+	actionCreateJSONBytes         = []byte(actionCreateJSONString)
+	actionCreateOrUpdateJSONBytes = []byte(actionCreateOrUpdateJSONString)
 )
 
 func (a ConsumerAction) String() string {
 	switch a {
 	case ActionCreateOrUpdate:
-		return actionCreateOrUpdateString
+		return actionCreateOrUpdateJSONString
 	case ActionCreate:
-		return actionCreateString
+		return actionCreateJSONString
 	case ActionUpdate:
-		return actionUpdateString
+		return actionUpdateJSONString
 	}
-	return actionCreateOrUpdateString
+	return actionCreateOrUpdateJSONString
 }
+
 func (a ConsumerAction) MarshalJSON() ([]byte, error) {
 	switch a {
 	case ActionCreate:
-		return json.Marshal(actionCreateString)
+		return actionCreateJSONBytes, nil
 	case ActionUpdate:
-		return json.Marshal(actionUpdateString)
+		return actionUpdateJSONBytes, nil
 	case ActionCreateOrUpdate:
-		return json.Marshal(actionCreateOrUpdateString)
+		return actionCreateOrUpdateJSONBytes, nil
 	default:
 		return nil, fmt.Errorf("can not marshal %v", a)
 	}
@@ -159,11 +166,11 @@ func (a ConsumerAction) MarshalJSON() ([]byte, error) {
 
 func (a *ConsumerAction) UnmarshalJSON(data []byte) error {
 	switch string(data) {
-	case jsonString("create"):
+	case actionCreateJSONString:
 		*a = ActionCreate
-	case jsonString("update"):
+	case actionUpdateJSONString:
 		*a = ActionUpdate
-	case jsonString(""):
+	case actionCreateOrUpdateJSONString:
 		*a = ActionCreateOrUpdate
 	default:
 		return fmt.Errorf("unknown consumer action: %v", string(data))
@@ -249,9 +256,9 @@ const (
 func (r ReplayPolicy) String() string {
 	switch r {
 	case ReplayInstant:
-		return "instant"
+		return replayInstantPolicyJSONString
 	default:
-		return "original"
+		return replayOriginalPolicyJSONString
 	}
 }
 
@@ -700,16 +707,16 @@ func (mset *stream) addConsumer(config *ConsumerConfig) (*consumer, error) {
 }
 
 func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname string, ca *consumerAssignment, isRecovering bool, action ConsumerAction) (*consumer, error) {
-	mset.mu.RLock()
-	s, jsa, tierName, cfg, acc, closed := mset.srv, mset.jsa, mset.tier, mset.cfg, mset.acc, mset.closed
-	retention := cfg.Retention
-	mset.mu.RUnlock()
-
 	// Check if this stream has closed.
-	if closed {
+	if mset.closed.Load() {
 		return nil, NewJSStreamInvalidError()
 	}
 
+	mset.mu.RLock()
+	s, jsa, tierName, cfg, acc := mset.srv, mset.jsa, mset.tier, mset.cfg, mset.acc
+	retention := cfg.Retention
+	mset.mu.RUnlock()
+
 	// If we do not have the consumer currently assigned to us in cluster mode we will proceed but warn.
 	// This can happen on startup with restored state where on meta replay we still do not have
 	// the assignment. Running in single server mode this always returns true.
diff --git a/server/filestore.go b/server/filestore.go
index c7244c82..8b73d353 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -1,4 +1,4 @@
-// Copyright 2019-2023 The NATS Authors
+// Copyright 2019-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
@@ -290,7 +290,7 @@ const (
 	// Maximum size of a write buffer we may consider for re-use.
 	maxBufReuse = 2 * 1024 * 1024
 	// default cache buffer expiration
-	defaultCacheBufferExpiration = 5 * time.Second
+	defaultCacheBufferExpiration = 2 * time.Second
 	// default sync interval
 	defaultSyncInterval = 2 * time.Minute
 	// default idle timeout to close FDs.
@@ -3222,8 +3222,27 @@ func (fs *fileStore) SkipMsg() uint64 {
 	fs.mu.Lock()
 	defer fs.mu.Unlock()
 
+	// Grab our current last message block.
+	mb := fs.lmb
+	if mb == nil || mb.msgs > 0 && mb.blkSize()+emptyRecordLen > fs.fcfg.BlockSize {
+		if mb != nil && fs.fcfg.Compression != NoCompression {
+			// We've now reached the end of this message block, if we want
+			// to compress blocks then now's the time to do it.
+			go mb.recompressOnDiskIfNeeded()
+		}
+		var err error
+		if mb, err = fs.newMsgBlockForWrite(); err != nil {
+			return 0
+		}
+	}
+
 	// Grab time and last seq.
 	now, seq := time.Now().UTC(), fs.state.LastSeq+1
+
+	// Write skip msg.
+	mb.skipMsg(seq, now)
+
+	// Update fs state.
 	fs.state.LastSeq, fs.state.LastTime = seq, now
 	if fs.state.Msgs == 0 {
 		fs.state.FirstSeq, fs.state.FirstTime = seq, now
@@ -3231,11 +3250,84 @@ func (fs *fileStore) SkipMsg() uint64 {
 	if seq == fs.state.FirstSeq {
 		fs.state.FirstSeq, fs.state.FirstTime = seq+1, now
 	}
-	fs.lmb.skipMsg(seq, now)
+	// Mark as dirty for stream state.
+	fs.dirty++
 
 	return seq
 }
 
+// Skip multiple msgs. We will determine if we can fit into current lmb or we need to create a new block.
+func (fs *fileStore) SkipMsgs(seq uint64, num uint64) error {
+	fs.mu.Lock()
+	defer fs.mu.Unlock()
+
+	// Check sequence matches our last sequence.
+	if seq != fs.state.LastSeq+1 {
+		if seq > 0 {
+			return ErrSequenceMismatch
+		}
+		seq = fs.state.LastSeq + 1
+	}
+
+	// Limit number of dmap entries
+	const maxDeletes = 64 * 1024
+	mb := fs.lmb
+
+	numDeletes := int(num)
+	if mb != nil {
+		numDeletes += mb.dmap.Size()
+	}
+	if mb == nil || numDeletes > maxDeletes && mb.msgs > 0 || mb.msgs > 0 && mb.blkSize()+emptyRecordLen > fs.fcfg.BlockSize {
+		if mb != nil && fs.fcfg.Compression != NoCompression {
+			// We've now reached the end of this message block, if we want
+			// to compress blocks then now's the time to do it.
+			go mb.recompressOnDiskIfNeeded()
+		}
+		var err error
+		if mb, err = fs.newMsgBlockForWrite(); err != nil {
+			return err
+		}
+	}
+
+	// Insert into dmap all entries and place last as marker.
+	now := time.Now().UTC()
+	nowts := now.UnixNano()
+	lseq := seq + num - 1
+
+	mb.mu.Lock()
+	var needsRecord bool
+	// If we are empty update meta directly.
+	if mb.msgs == 0 {
+		atomic.StoreUint64(&mb.last.seq, lseq)
+		mb.last.ts = nowts
+		atomic.StoreUint64(&mb.first.seq, lseq+1)
+		mb.first.ts = nowts
+	} else {
+		needsRecord = true
+		for ; seq <= lseq; seq++ {
+			mb.dmap.Insert(seq)
+		}
+	}
+	mb.mu.Unlock()
+
+	// Write out our placeholder.
+	if needsRecord {
+		mb.writeMsgRecord(emptyRecordLen, lseq|ebit, _EMPTY_, nil, nil, nowts, true)
+	}
+
+	// Now update FS accounting.
+	// Update fs state.
+	fs.state.LastSeq, fs.state.LastTime = lseq, now
+	if fs.state.Msgs == 0 {
+		fs.state.FirstSeq, fs.state.FirstTime = lseq+1, now
+	}
+
+	// Mark as dirty for stream state.
+	fs.dirty++
+
+	return nil
+}
+
 // Lock should be held.
 func (fs *fileStore) rebuildFirst() {
 	if len(fs.blks) == 0 {
@@ -4203,7 +4295,7 @@ func (fs *fileStore) selectNextFirst() {
 // Lock should be held.
 func (mb *msgBlock) resetCacheExpireTimer(td time.Duration) {
 	if td == 0 {
-		td = mb.cexp
+		td = mb.cexp + 100*time.Millisecond
 	}
 	if mb.ctmr == nil {
 		mb.ctmr = time.AfterFunc(td, mb.expireCache)
@@ -5066,24 +5158,30 @@ func (mb *msgBlock) indexCacheBuf(buf []byte) error {
 
 	mbFirstSeq := atomic.LoadUint64(&mb.first.seq)
 
+	// Capture beginning size of dmap.
+	dms := uint64(mb.dmap.Size())
+	idxSz := atomic.LoadUint64(&mb.last.seq) - mbFirstSeq + 1
+
 	if mb.cache == nil {
 		// Approximation, may adjust below.
 		fseq = mbFirstSeq
-		idx = make([]uint32, 0, mb.msgs)
+		idx = make([]uint32, 0, idxSz)
 		mb.cache = &cache{}
 	} else {
 		fseq = mb.cache.fseq
 		idx = mb.cache.idx
 		if len(idx) == 0 {
-			idx = make([]uint32, 0, mb.msgs)
+			idx = make([]uint32, 0, idxSz)
 		}
 		index = uint32(len(mb.cache.buf))
 		buf = append(mb.cache.buf, buf...)
 	}
 
 	// Create FSS if we should track.
-	if !mb.noTrack {
+	var popFss bool
+	if mb.fssNotLoaded() {
 		mb.fss = make(map[string]*SimpleState)
+		popFss = true
 	}
 
 	lbuf := uint32(len(buf))
@@ -5123,7 +5221,9 @@ func (mb *msgBlock) indexCacheBuf(buf []byte) error {
 				// If we have a hole fill it.
 				for dseq := mbFirstSeq + uint64(len(idx)); dseq < seq; dseq++ {
 					idx = append(idx, dbit)
-					mb.dmap.Insert(dseq)
+					if dms == 0 {
+						mb.dmap.Insert(dseq)
+					}
 				}
 			}
 			// Add to our index.
@@ -5135,12 +5235,12 @@ func (mb *msgBlock) indexCacheBuf(buf []byte) error {
 			}
 
 			// Make sure our dmap has this entry if it was erased.
-			if erased {
+			if erased && dms == 0 {
 				mb.dmap.Insert(seq)
 			}
 
 			// Handle FSS inline here.
-			if slen > 0 && !mb.noTrack && !erased && !mb.dmap.Exists(seq) {
+			if popFss && slen > 0 && !mb.noTrack && !erased && !mb.dmap.Exists(seq) {
 				bsubj := buf[index+msgHdrSize : index+msgHdrSize+uint32(slen)]
 				if ss := mb.fss[string(bsubj)]; ss != nil {
 					ss.Msgs++
@@ -5345,14 +5445,26 @@ func (mb *msgBlock) fssLoaded() bool {
 // Used to load in the block contents.
 // Lock should be held and all conditionals satisfied prior.
 func (mb *msgBlock) loadBlock(buf []byte) ([]byte, error) {
-	f, err := os.Open(mb.mfn)
-	if err != nil {
-		if os.IsNotExist(err) {
-			err = errNoBlkData
+	var f *os.File
+	// Re-use if we have mfd open.
+	if mb.mfd != nil {
+		f = mb.mfd
+		if n, err := f.Seek(0, 0); n != 0 || err != nil {
+			f = nil
+			mb.closeFDsLockedNoCheck()
 		}
-		return nil, err
 	}
-	defer f.Close()
+	if f == nil {
+		var err error
+		f, err = os.Open(mb.mfn)
+		if err != nil {
+			if os.IsNotExist(err) {
+				err = errNoBlkData
+			}
+			return nil, err
+		}
+		defer f.Close()
+	}
 
 	var sz int
 	if info, err := f.Stat(); err == nil {
@@ -5495,17 +5607,34 @@ func (mb *msgBlock) fetchMsg(seq uint64, sm *StoreMsg) (*StoreMsg, bool, error)
 	mb.mu.Lock()
 	defer mb.mu.Unlock()
 
+	fseq, lseq := atomic.LoadUint64(&mb.first.seq), atomic.LoadUint64(&mb.last.seq)
+	if seq < fseq || seq > lseq {
+		return nil, false, ErrStoreMsgNotFound
+	}
+
+	// See if we can short circuit if we already know msg deleted.
+	if mb.dmap.Exists(seq) {
+		// Update for scanning like cacheLookup would have.
+		llseq := mb.llseq
+		if mb.llseq == 0 || seq < mb.llseq || seq == mb.llseq+1 || seq == mb.llseq-1 {
+			mb.llseq = seq
+		}
+		expireOk := (seq == lseq && llseq == seq-1) || (seq == fseq && llseq == seq+1)
+		return nil, expireOk, errDeletedMsg
+	}
+
 	if mb.cacheNotLoaded() {
 		if err := mb.loadMsgsWithLock(); err != nil {
 			return nil, false, err
 		}
 	}
 	llseq := mb.llseq
+
 	fsm, err := mb.cacheLookup(seq, sm)
 	if err != nil {
 		return nil, false, err
 	}
-	expireOk := (seq == atomic.LoadUint64(&mb.last.seq) && llseq == seq-1) || (seq == atomic.LoadUint64(&mb.first.seq) && llseq == seq+1)
+	expireOk := (seq == lseq && llseq == seq-1) || (seq == fseq && llseq == seq+1)
 	return fsm, expireOk, err
 }
 
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 6c282c26..76c1c2a7 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -3717,7 +3717,7 @@ func (s *Server) streamSnapshot(ci *ClientInfo, acc *Account, mset *stream, sr *
 
 	// We will place sequence number and size of chunk sent in the reply.
 	ackSubj := fmt.Sprintf(jsSnapshotAckT, mset.name(), nuid.Next())
-	ackSub, _ := mset.subscribeInternalUnlocked(ackSubj+".>", func(_ *subscription, _ *client, _ *Account, subject, _ string, _ []byte) {
+	ackSub, _ := mset.subscribeInternal(ackSubj+".>", func(_ *subscription, _ *client, _ *Account, subject, _ string, _ []byte) {
 		cs, _ := strconv.Atoi(tokenAt(subject, 6))
 		// This is very crude and simple, but ok for now.
 		// This only matters when sending multiple chunks.
@@ -3728,7 +3728,7 @@ func (s *Server) streamSnapshot(ci *ClientInfo, acc *Account, mset *stream, sr *
 			}
 		}
 	})
-	defer mset.unsubscribeUnlocked(ackSub)
+	defer mset.unsubscribe(ackSub)
 
 	// TODO(dlc) - Add in NATS-Chunked-Sequence header
 
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index bc1b5e28..770b9957 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -103,6 +103,8 @@ const (
 	removePendingRequest
 	// For sending compressed streams, either through RAFT or catchup.
 	compressedStreamMsgOp
+	// For sending deleted gaps on catchups for replicas.
+	deleteRangeOp
 )
 
 // raftGroups are controlled by the metagroup controller.
@@ -490,24 +492,24 @@ func (js *jetStream) restartStream(acc *Account, csa *streamAssignment) {
 // isStreamHealthy will determine if the stream is up to date or very close.
 // For R1 it will make sure the stream is present on this server.
 func (js *jetStream) isStreamHealthy(acc *Account, sa *streamAssignment) bool {
-	js.mu.Lock()
+	js.mu.RLock()
 	s, cc := js.srv, js.cluster
 	if cc == nil {
 		// Non-clustered mode
-		js.mu.Unlock()
+		js.mu.RUnlock()
 		return true
 	}
 
 	// Pull the group out.
 	rg := sa.Group
 	if rg == nil {
-		js.mu.Unlock()
+		js.mu.RUnlock()
 		return false
 	}
 
 	streamName := sa.Config.Name
 	node := rg.node
-	js.mu.Unlock()
+	js.mu.RUnlock()
 
 	// First lookup stream and make sure its there.
 	mset, err := acc.lookupStream(streamName)
@@ -516,6 +518,11 @@ func (js *jetStream) isStreamHealthy(acc *Account, sa *streamAssignment) bool {
 		return false
 	}
 
+	// If we are catching up return false.
+	if mset.isCatchingUp() {
+		return false
+	}
+
 	if node == nil || node.Healthy() {
 		// Check if we are processing a snapshot and are catching up.
 		if !mset.isCatchingUp() {
@@ -1239,7 +1246,8 @@ func (js *jetStream) monitorCluster() {
 	lt := time.NewTicker(leaderCheckInterval)
 	defer lt.Stop()
 
-	const healthCheckInterval = 2 * time.Minute
+	// Check the general health once an hour.
+	const healthCheckInterval = 1 * time.Hour
 	ht := time.NewTicker(healthCheckInterval)
 	defer ht.Stop()
 
@@ -1455,18 +1463,22 @@ func (js *jetStream) clusterStreamConfig(accName, streamName string) (StreamConf
 }
 
 func (js *jetStream) metaSnapshot() []byte {
-	var streams []writeableStreamAssignment
-
 	js.mu.RLock()
 	cc := js.cluster
+	nsa := 0
+	for _, asa := range cc.streams {
+		nsa += len(asa)
+	}
+	streams := make([]writeableStreamAssignment, 0, nsa)
 	for _, asa := range cc.streams {
 		for _, sa := range asa {
 			wsa := writeableStreamAssignment{
-				Client:  sa.Client,
-				Created: sa.Created,
-				Config:  sa.Config,
-				Group:   sa.Group,
-				Sync:    sa.Sync,
+				Client:    sa.Client,
+				Created:   sa.Created,
+				Config:    sa.Config,
+				Group:     sa.Group,
+				Sync:      sa.Sync,
+				Consumers: make([]*consumerAssignment, 0, len(sa.consumers)),
 			}
 			for _, ca := range sa.consumers {
 				wsa.Consumers = append(wsa.Consumers, ca)
@@ -1935,6 +1947,9 @@ func (js *jetStream) applyMetaEntries(entries []*Entry, ru *recoveryUpdates) (bo
 					delete(ru.removeStreams, key)
 				} else {
 					js.processUpdateStreamAssignment(sa)
+					// Since an update can be lowering replica count, we want upper layer to treat
+					// similar to a removal and snapshot to collapse old entries.
+					didRemoveStream = true
 				}
 			default:
 				panic(fmt.Sprintf("JetStream Cluster Unknown meta entry op type: %v", entryOp(buf[0])))
@@ -2193,9 +2208,13 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 	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
+
 	// Should only to be called from leader.
 	doSnapshot := func() {
-		if mset == nil || isRestore || time.Since(lastSnapTime) < minSnapDelta {
+		if mset == nil || isRecovering || isRestore || time.Since(lastSnapTime) < minSnapDelta {
 			return
 		}
 
@@ -2222,7 +2241,6 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 	// We will establish a restoreDoneCh no matter what. Will never be triggered unless
 	// we replace with the restore chan.
 	restoreDoneCh := make(<-chan error)
-	isRecovering := true
 
 	// For migration tracking.
 	var mmt *time.Ticker
@@ -4548,9 +4566,13 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 	var lastSnap []byte
 	var lastSnapTime time.Time
 
+	// Don't allow the upper layer to install snapshots until we have
+	// fully recovered from disk.
+	recovering := true
+
 	doSnapshot := func(force bool) {
 		// Bail if trying too fast and not in a forced situation.
-		if !force && time.Since(lastSnapTime) < minSnapDelta {
+		if recovering || (!force && time.Since(lastSnapTime) < minSnapDelta) {
 			return
 		}
 
@@ -4601,7 +4623,6 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 
 	// Track if we are leader.
 	var isLeader bool
-	recovering := true
 
 	for {
 		select {
@@ -6827,6 +6848,22 @@ func decodeStreamAssignment(buf []byte) (*streamAssignment, error) {
 	return &sa, err
 }
 
+func encodeDeleteRange(dr *DeleteRange) []byte {
+	var bb bytes.Buffer
+	bb.WriteByte(byte(deleteRangeOp))
+	json.NewEncoder(&bb).Encode(dr)
+	return bb.Bytes()
+}
+
+func decodeDeleteRange(buf []byte) (*DeleteRange, error) {
+	var dr DeleteRange
+	err := json.Unmarshal(buf, &dr)
+	if err != nil {
+		return nil, err
+	}
+	return &dr, err
+}
+
 // createGroupForConsumer will create a new group from same peer set as the stream.
 func (cc *jetStreamCluster) createGroupForConsumer(cfg *ConsumerConfig, sa *streamAssignment) *raftGroup {
 	if len(sa.Group.Peers) == 0 || cfg.Replicas > len(sa.Group.Peers) {
@@ -7597,9 +7634,10 @@ func (mset *stream) processClusteredInboundMsg(subject, reply string, hdr, msg [
 // For requesting messages post raft snapshot to catch up streams post server restart.
 // Any deleted msgs etc will be handled inline on catchup.
 type streamSyncRequest struct {
-	Peer     string `json:"peer,omitempty"`
-	FirstSeq uint64 `json:"first_seq"`
-	LastSeq  uint64 `json:"last_seq"`
+	Peer           string `json:"peer,omitempty"`
+	FirstSeq       uint64 `json:"first_seq"`
+	LastSeq        uint64 `json:"last_seq"`
+	DeleteRangesOk bool   `json:"delete_ranges"`
 }
 
 // Given a stream state that represents a snapshot, calculate the sync request based on our current state.
@@ -7608,7 +7646,7 @@ func (mset *stream) calculateSyncRequest(state *StreamState, snap *StreamReplica
 	if state.LastSeq >= snap.LastSeq {
 		return nil
 	}
-	return &streamSyncRequest{FirstSeq: state.LastSeq + 1, LastSeq: snap.LastSeq, Peer: mset.node.ID()}
+	return &streamSyncRequest{FirstSeq: state.LastSeq + 1, LastSeq: snap.LastSeq, Peer: mset.node.ID(), DeleteRangesOk: true}
 }
 
 // processSnapshotDeletes will update our current store based on the snapshot
@@ -7618,14 +7656,22 @@ func (mset *stream) processSnapshotDeletes(snap *StreamReplicatedState) {
 	var state StreamState
 	mset.store.FastState(&state)
 	// Always adjust if FirstSeq has moved beyond our state.
+	var didReset bool
 	if snap.FirstSeq > state.FirstSeq {
 		mset.store.Compact(snap.FirstSeq)
 		mset.store.FastState(&state)
 		mset.lseq = state.LastSeq
 		mset.clearAllPreAcksBelowFloor(state.FirstSeq)
+		didReset = true
 	}
+	s := mset.srv
 	mset.mu.Unlock()
 
+	if didReset {
+		s.Warnf("Catchup for stream '%s > %s' resetting first sequence: %d on catchup request",
+			mset.account(), mset.name(), snap.FirstSeq)
+	}
+
 	if len(snap.Deleted) > 0 {
 		mset.store.SyncDeleted(snap.Deleted)
 	}
@@ -7655,6 +7701,22 @@ func (mset *stream) updateCatchupPeer(peer string) {
 	mset.mu.Unlock()
 }
 
+func (mset *stream) decrementCatchupPeer(peer string, num uint64) {
+	if peer == _EMPTY_ {
+		return
+	}
+	mset.mu.Lock()
+	if lag := mset.catchups[peer]; lag > 0 {
+		if lag >= num {
+			lag -= num
+		} else {
+			lag = 0
+		}
+		mset.catchups[peer] = lag
+	}
+	mset.mu.Unlock()
+}
+
 func (mset *stream) clearCatchupPeer(peer string) {
 	mset.mu.Lock()
 	if mset.catchups != nil {
@@ -7686,21 +7748,15 @@ func (mset *stream) hasCatchupPeers() bool {
 }
 
 func (mset *stream) setCatchingUp() {
-	mset.mu.Lock()
-	mset.catchup = true
-	mset.mu.Unlock()
+	mset.catchup.Store(true)
 }
 
 func (mset *stream) clearCatchingUp() {
-	mset.mu.Lock()
-	mset.catchup = false
-	mset.mu.Unlock()
+	mset.catchup.Store(false)
 }
 
 func (mset *stream) isCatchingUp() bool {
-	mset.mu.RLock()
-	defer mset.mu.RUnlock()
-	return mset.catchup
+	return mset.catchup.Load()
 }
 
 // Determine if a non-leader is current.
@@ -7709,7 +7765,7 @@ func (mset *stream) isCurrent() bool {
 	if mset.node == nil {
 		return true
 	}
-	return mset.node.Current() && !mset.catchup
+	return mset.node.Current() && !mset.catchup.Load()
 }
 
 // Maximum requests for the whole server that can be in flight at the same time.
@@ -7733,7 +7789,6 @@ func (mset *stream) processSnapshot(snap *StreamReplicatedState) (e error) {
 	mset.store.FastState(&state)
 	mset.setCLFS(snap.Failed)
 	sreq := mset.calculateSyncRequest(&state, snap)
-
 	s, js, subject, n := mset.srv, mset.js, mset.sa.Sync, mset.node
 	qname := fmt.Sprintf("[ACC:%s] stream '%s' snapshot", mset.acc.Name, mset.cfg.Name)
 	mset.mu.Unlock()
@@ -7767,7 +7822,7 @@ func (mset *stream) processSnapshot(snap *StreamReplicatedState) (e error) {
 	var sub *subscription
 	var err error
 
-	const activityInterval = 10 * time.Second
+	const activityInterval = 30 * time.Second
 	notActive := time.NewTimer(activityInterval)
 	defer notActive.Stop()
 
@@ -7803,36 +7858,12 @@ func (mset *stream) processSnapshot(snap *StreamReplicatedState) (e error) {
 	defer releaseSyncOutSem()
 
 	// Check our final state when we exit cleanly.
-	// If this snapshot was for messages no longer held by the leader we want to make sure
-	// we are synched for the next message sequence properly.
-	lastRequested := sreq.LastSeq
+	// This will make sure we have interest consumers updated.
 	checkFinalState := func() {
 		// Bail if no stream.
 		if mset == nil {
 			return
 		}
-
-		mset.mu.Lock()
-		var state StreamState
-		mset.store.FastState(&state)
-		var didReset bool
-		firstExpected := lastRequested + 1
-		if state.FirstSeq != firstExpected {
-			// Reset our notion of first.
-			mset.store.Compact(firstExpected)
-			mset.store.FastState(&state)
-			// Make sure last is also correct in case this also moved.
-			mset.lseq = state.LastSeq
-			mset.clearAllPreAcksBelowFloor(state.FirstSeq)
-			didReset = true
-		}
-		mset.mu.Unlock()
-
-		if didReset {
-			s.Warnf("Catchup for stream '%s > %s' resetting first sequence: %d on catchup complete",
-				mset.account(), mset.name(), firstExpected)
-		}
-
 		mset.mu.RLock()
 		consumers := make([]*consumer, 0, len(mset.consumers))
 		for _, o := range mset.consumers {
@@ -7878,7 +7909,7 @@ RETRY:
 	<-s.syncOutSem
 	releaseSem = true
 
-	// We may have been blocked for a bit, so the reset need to ensure that we
+	// We may have been blocked for a bit, so the reset needs to ensure that we
 	// consume the already fired timer.
 	if !notActive.Stop() {
 		select {
@@ -7898,8 +7929,6 @@ RETRY:
 		if sreq == nil {
 			return nil
 		}
-		// Reset notion of lastRequested
-		lastRequested = sreq.LastSeq
 	}
 
 	// Used to transfer message from the wire to another Go routine internally.
@@ -7929,14 +7958,13 @@ RETRY:
 		s.Errorf("Could not subscribe to stream catchup: %v", err)
 		goto RETRY
 	}
+
 	// Send our sync request.
 	b, _ := json.Marshal(sreq)
 	s.sendInternalMsgLocked(subject, reply, nil, b)
-	// Remember when we sent this out to avoimd loop spins on errors below.
+	// Remember when we sent this out to avoid loop spins on errors below.
 	reqSendTime := time.Now()
-
-	// Clear our sync request and capture last.
-	last := sreq.LastSeq
+	// Clear our sync request.
 	sreq = nil
 
 	// Run our own select loop here.
@@ -7946,24 +7974,18 @@ RETRY:
 			notActive.Reset(activityInterval)
 
 			mrecs := msgsQ.pop()
-
 			for _, mrec := range mrecs {
 				msg := mrec.msg
-
 				// Check for eof signaling.
 				if len(msg) == 0 {
 					msgsQ.recycle(&mrecs)
 					checkFinalState()
 					return nil
 				}
-				if lseq, err := mset.processCatchupMsg(msg); err == nil {
+				if _, err := mset.processCatchupMsg(msg); err == nil {
 					if mrec.reply != _EMPTY_ {
 						s.sendInternalMsgLocked(mrec.reply, _EMPTY_, nil, nil)
 					}
-					if lseq >= last {
-						msgsQ.recycle(&mrecs)
-						return nil
-					}
 				} else if isOutOfSpaceErr(err) {
 					notifyLeaderStopCatchup(mrec, err)
 					return err
@@ -7996,6 +8018,7 @@ RETRY:
 					goto RETRY
 				}
 			}
+			notActive.Reset(activityInterval)
 			msgsQ.recycle(&mrecs)
 		case <-notActive.C:
 			if mrecs := msgsQ.pop(); len(mrecs) > 0 {
@@ -8024,11 +8047,34 @@ func (mset *stream) processCatchupMsg(msg []byte) (uint64, error) {
 		return 0, errCatchupBadMsg
 	}
 	op := entryOp(msg[0])
-	if op != streamMsgOp && op != compressedStreamMsgOp {
+	if op != streamMsgOp && op != compressedStreamMsgOp && op != deleteRangeOp {
 		return 0, errCatchupBadMsg
 	}
 
 	mbuf := msg[1:]
+	if op == deleteRangeOp {
+		dr, err := decodeDeleteRange(mbuf)
+		if err != nil {
+			return 0, errCatchupBadMsg
+		}
+		// Handle the delete range.
+		// Make sure the sequences match up properly.
+		mset.mu.Lock()
+		if len(mset.preAcks) > 0 {
+			for seq := dr.First; seq < dr.First+dr.Num; seq++ {
+				mset.clearAllPreAcks(seq)
+			}
+		}
+		if err = mset.store.SkipMsgs(dr.First, dr.Num); err != nil {
+			mset.mu.Unlock()
+			return 0, errCatchupWrongSeqForSkip
+		}
+		mset.lseq = dr.First + dr.Num - 1
+		lseq := mset.lseq
+		mset.mu.Unlock()
+		return lseq, nil
+	}
+
 	if op == compressedStreamMsgOp {
 		var err error
 		mbuf, err = s2.Decode(nil, mbuf)
@@ -8371,8 +8417,8 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 	s := mset.srv
 	defer s.grWG.Done()
 
-	const maxOutBytes = int64(8 * 1024 * 1024) // 8MB for now, these are all internal, from server to server
-	const maxOutMsgs = int32(32 * 1024)
+	const maxOutBytes = int64(64 * 1024 * 1024) // 64MB for now, these are all internal, from server to server
+	const maxOutMsgs = int32(256 * 1024)        // 256k in case we have lots of small messages or skip msgs.
 	outb := int64(0)
 	outm := int32(0)
 
@@ -8391,6 +8437,10 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 	nextBatchC <- struct{}{}
 	remoteQuitCh := make(chan struct{})
 
+	const activityInterval = 30 * time.Second
+	notActive := time.NewTimer(activityInterval)
+	defer notActive.Stop()
+
 	// Setup ackReply for flow control.
 	ackReply := syncAckSubject()
 	ackSub, _ := s.sysSubscribe(ackReply, func(sub *subscription, c *client, _ *Account, subject, reply string, msg []byte) {
@@ -8408,16 +8458,14 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 		// Kick ourselves and anyone else who might have stalled on global state.
 		select {
 		case nextBatchC <- struct{}{}:
+			// Reset our activity
+			notActive.Reset(activityInterval)
 		default:
 		}
 	})
 	defer s.sysUnsubscribe(ackSub)
 	ackReplyT := strings.ReplaceAll(ackReply, ".*", ".%d")
 
-	const activityInterval = 5 * time.Second
-	notActive := time.NewTimer(activityInterval)
-	defer notActive.Stop()
-
 	// Grab our state.
 	var state StreamState
 	mset.mu.RLock()
@@ -8440,10 +8488,9 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 	compressOk := mset.compressAllowed()
 
 	var spb int
-	sendNextBatchAndContinue := func(qch chan struct{}) bool {
-		// Update our activity timer.
-		notActive.Reset(activityInterval)
+	const minWait = 5 * time.Second
 
+	sendNextBatchAndContinue := func(qch chan struct{}) bool {
 		// Check if we know we will not enter the loop because we are done.
 		if seq > last {
 			s.Noticef("Catchup for stream '%s > %s' complete", mset.account(), mset.name())
@@ -8452,15 +8499,23 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 			return false
 		}
 
-		// If we already sent a batch, we will try to make sure we process around
-		// half the FC responses - or reach a certain amount of time - before sending
-		// the next batch.
+		// If we already sent a batch, we will try to make sure we can at least send a minimum
+		// batch before sending the next batch.
 		if spb > 0 {
-			mw := time.NewTimer(100 * time.Millisecond)
+			// Wait til we can send at least 4k
+			const minBatchWait = int32(4 * 1024)
+			mw := time.NewTimer(minWait)
 			for done := false; !done; {
 				select {
 				case <-nextBatchC:
-					done = int(atomic.LoadInt32(&outm)) <= spb/2
+					done = maxOutMsgs-atomic.LoadInt32(&outm) > minBatchWait
+					if !done {
+						// Wait for a small bit.
+						time.Sleep(50 * time.Millisecond)
+					} else {
+						// GC friendly.
+						mw.Stop()
+					}
 				case <-mw.C:
 					done = true
 				case <-s.quitCh:
@@ -8474,8 +8529,38 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 			spb = 0
 		}
 
-		var smv StoreMsg
+		// Send an encoded msg.
+		sendEM := func(em []byte) {
+			// Place size in reply subject for flow control.
+			l := int64(len(em))
+			reply := fmt.Sprintf(ackReplyT, l)
+			s.gcbAdd(&outb, l)
+			atomic.AddInt32(&outm, 1)
+			s.sendInternalMsgLocked(sendSubject, reply, nil, em)
+			spb++
+		}
+
+		// If we support gap markers.
+		var dr DeleteRange
+		drOk := sreq.DeleteRangesOk
 
+		// Will send our delete range.
+		// Should already be checked for being valid.
+		sendDR := func() {
+			if dr.Num == 1 {
+				// Send like a normal skip msg.
+				sendEM(encodeStreamMsg(_EMPTY_, _EMPTY_, nil, nil, dr.First, 0))
+			} else {
+				// We have a run, send a gap record. We send these without reply or tracking.
+				s.sendInternalMsgLocked(sendSubject, _EMPTY_, nil, encodeDeleteRange(&dr))
+				// Clear out the pending for catchup.
+				mset.decrementCatchupPeer(sreq.Peer, dr.Num)
+			}
+			// Reset always.
+			dr.First, dr.Num = 0, 0
+		}
+
+		var smv StoreMsg
 		for ; seq <= last && atomic.LoadInt64(&outb) <= maxOutBytes && atomic.LoadInt32(&outm) <= maxOutMsgs && s.gcbBelowMax(); seq++ {
 			sm, err := mset.store.LoadMsg(seq, &smv)
 			// if this is not a deleted msg, bail out.
@@ -8501,22 +8586,33 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 				s.Warnf("Error loading message for catchup '%s > %s': %v", mset.account(), mset.name(), err)
 				return false
 			}
-			var em []byte
+
 			if sm != nil {
-				em = encodeStreamMsgAllowCompress(sm.subj, _EMPTY_, sm.hdr, sm.msg, sm.seq, sm.ts, compressOk)
+				// If we allow gap markers check if we have one pending.
+				if drOk && dr.First > 0 {
+					sendDR()
+				}
+				// Send the normal message now.
+				sendEM(encodeStreamMsgAllowCompress(sm.subj, _EMPTY_, sm.hdr, sm.msg, sm.seq, sm.ts, compressOk))
 			} else {
-				// Skip record for deleted msg.
-				em = encodeStreamMsg(_EMPTY_, _EMPTY_, nil, nil, seq, 0)
+				if drOk {
+					if dr.First == 0 {
+						dr.First, dr.Num = seq, 1
+					} else {
+						dr.Num++
+					}
+				} else {
+					// Skip record for deleted msg.
+					sendEM(encodeStreamMsg(_EMPTY_, _EMPTY_, nil, nil, seq, 0))
+				}
 			}
 
-			// Place size in reply subject for flow control.
-			l := int64(len(em))
-			reply := fmt.Sprintf(ackReplyT, l)
-			s.gcbAdd(&outb, l)
-			atomic.AddInt32(&outm, 1)
-			s.sendInternalMsgLocked(sendSubject, reply, nil, em)
-			spb++
+			// Check if we are done.
 			if seq == last {
+				// Need to see if we have a pending delete range.
+				if drOk && dr.First > 0 {
+					sendDR()
+				}
 				s.Noticef("Catchup for stream '%s > %s' complete", mset.account(), mset.name())
 				// EOF
 				s.sendInternalMsgLocked(sendSubject, _EMPTY_, nil, nil)
@@ -8528,10 +8624,14 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 			default:
 			}
 		}
+		if drOk && dr.First > 0 {
+			sendDR()
+		}
+
 		return true
 	}
 
-	// Grab stream quit channel.
+	// Check is this stream got closed.
 	mset.mu.RLock()
 	qch := mset.qch
 	mset.mu.RUnlock()
@@ -8555,6 +8655,7 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 			return
 		case <-notActive.C:
 			s.Warnf("Catchup for stream '%s > %s' stalled", mset.account(), mset.name())
+			mset.clearCatchupPeer(sreq.Peer)
 			return
 		case <-nextBatchC:
 			if !sendNextBatchAndContinue(qch) {
diff --git a/server/memstore.go b/server/memstore.go
index 9853dd31..8d324d01 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -256,12 +256,42 @@ func (ms *memStore) SkipMsg() uint64 {
 	if ms.state.Msgs == 0 {
 		ms.state.FirstSeq = seq
 		ms.state.FirstTime = now
+	} else {
+		ms.dmap.Insert(seq)
 	}
-	ms.updateFirstSeq(seq)
 	ms.mu.Unlock()
 	return seq
 }
 
+// Skip multiple msgs.
+func (ms *memStore) SkipMsgs(seq uint64, num uint64) error {
+	// Grab time.
+	now := time.Now().UTC()
+
+	ms.mu.Lock()
+	defer ms.mu.Unlock()
+
+	// Check sequence matches our last sequence.
+	if seq != ms.state.LastSeq+1 {
+		if seq > 0 {
+			return ErrSequenceMismatch
+		}
+		seq = ms.state.LastSeq + 1
+	}
+	lseq := seq + num - 1
+
+	ms.state.LastSeq = lseq
+	ms.state.LastTime = now
+	if ms.state.Msgs == 0 {
+		ms.state.FirstSeq, ms.state.FirstTime = lseq+1, now
+	} else {
+		for ; seq <= lseq; seq++ {
+			ms.dmap.Insert(seq)
+		}
+	}
+	return nil
+}
+
 // RegisterStorageUpdates registers a callback for updates to storage changes.
 // It will present number of messages and bytes as a signed integer and an
 // optional sequence number of the message if a single.
@@ -1058,9 +1088,6 @@ func (ms *memStore) updateFirstSeq(seq uint64) {
 			ms.dmap.Delete(seq)
 		}
 	}
-	if ms.dmap.IsEmpty() {
-		ms.dmap.SetInitialMin(ms.state.FirstSeq)
-	}
 }
 
 // Remove a seq from the fss and select new first.
@@ -1186,13 +1213,16 @@ func (ms *memStore) State() StreamState {
 
 	// Calculate interior delete details.
 	if numDeleted := int((state.LastSeq - state.FirstSeq + 1) - state.Msgs); numDeleted > 0 {
-		state.Deleted = make([]uint64, 0, state.NumDeleted)
-		// TODO(dlc) - Too Simplistic, once state is updated to allow runs etc redo.
-		for seq := state.FirstSeq + 1; seq < ms.state.LastSeq; seq++ {
-			if _, ok := ms.msgs[seq]; !ok {
+		state.Deleted = make([]uint64, 0, numDeleted)
+		fseq, lseq := state.FirstSeq, state.LastSeq
+		ms.dmap.Range(func(seq uint64) bool {
+			if seq < fseq || seq > lseq {
+				ms.dmap.Delete(seq)
+			} else {
 				state.Deleted = append(state.Deleted, seq)
 			}
-		}
+			return true
+		})
 	}
 	if len(state.Deleted) > 0 {
 		state.NumDeleted = len(state.Deleted)
diff --git a/server/monitor.go b/server/monitor.go
index 90f3f1de..20e38a29 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -3148,17 +3148,17 @@ func (t HealthZErrorType) MarshalJSON() ([]byte, error) {
 
 func (t *HealthZErrorType) UnmarshalJSON(data []byte) error {
 	switch string(data) {
-	case jsonString("CONNECTION"):
+	case `"CONNECTION"`:
 		*t = HealthzErrorConn
-	case jsonString("BAD_REQUEST"):
+	case `"BAD_REQUEST"`:
 		*t = HealthzErrorBadRequest
-	case jsonString("JETSTREAM"):
+	case `"JETSTREAM"`:
 		*t = HealthzErrorJetStream
-	case jsonString("ACCOUNT"):
+	case `"ACCOUNT"`:
 		*t = HealthzErrorAccount
-	case jsonString("STREAM"):
+	case `"STREAM"`:
 		*t = HealthzErrorStream
-	case jsonString("CONSUMER"):
+	case `"CONSUMER"`:
 		*t = HealthzErrorConsumer
 	default:
 		return fmt.Errorf("unknown healthz error type %q", data)
diff --git a/server/mqtt.go b/server/mqtt.go
index 730d1404..e82ddb29 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -186,17 +186,18 @@ const (
 
 	mqttInitialPubHeader        = 16 // An overkill, should need 7 bytes max
 	mqttProcessSubTooLong       = 100 * time.Millisecond
-	mqttRetainedCacheTTL        = 2 * time.Minute
+	mqttDefaultRetainedCacheTTL = 2 * time.Minute
 	mqttRetainedTransferTimeout = 10 * time.Second
 )
 
 var (
-	mqttPingResponse  = []byte{mqttPacketPingResp, 0x0}
-	mqttProtoName     = []byte("MQTT")
-	mqttOldProtoName  = []byte("MQIsdp")
-	mqttSessJailDur   = mqttSessFlappingJailDur
-	mqttFlapCleanItvl = mqttSessFlappingCleanupInterval
-	mqttJSAPITimeout  = 4 * time.Second
+	mqttPingResponse     = []byte{mqttPacketPingResp, 0x0}
+	mqttProtoName        = []byte("MQTT")
+	mqttOldProtoName     = []byte("MQIsdp")
+	mqttSessJailDur      = mqttSessFlappingJailDur
+	mqttFlapCleanItvl    = mqttSessFlappingCleanupInterval
+	mqttJSAPITimeout     = 4 * time.Second
+	mqttRetainedCacheTTL = mqttDefaultRetainedCacheTTL
 )
 
 var (
@@ -242,20 +243,25 @@ type mqttAccountSessionManager struct {
 	flapTimer  *time.Timer                    // Timer to perform some cleanup of the flappers map
 	sl         *Sublist                       // sublist allowing to find retained messages for given subscription
 	retmsgs    map[string]*mqttRetainedMsgRef // retained messages
-	rmsCache   sync.Map                       // map[string(subject)]mqttRetainedMsg
+	rmsCache   sync.Map                       // map[subject]*mqttRetainedMsg
 	jsa        mqttJSA
 	rrmLastSeq uint64        // Restore retained messages expected last sequence
 	rrmDoneCh  chan struct{} // To notify the caller that all retained messages have been loaded
 	domainTk   string        // Domain (with trailing "."), or possibly empty. This is added to session subject.
 }
 
+type mqttJSAResponse struct {
+	reply string // will be used to map to the original request in jsa.NewRequestExMulti
+	value any
+}
+
 type mqttJSA struct {
 	mu        sync.Mutex
 	id        string
 	c         *client
 	sendq     *ipQueue[*mqttJSPubMsg]
 	rplyr     string
-	replies   sync.Map
+	replies   sync.Map // [string]chan *mqttJSAResponse
 	nuid      *nuid.NUID
 	quitCh    chan struct{}
 	domain    string // Domain or possibly empty. This is added to session subject.
@@ -1232,7 +1238,7 @@ func (s *Server) mqttCreateAccountSessionManager(acc *Account, quitCh chan struc
 	// Start the go routine that will clean up cached retained messages that expired.
 	s.startGoRoutine(func() {
 		defer s.grWG.Done()
-		as.cleaupRetainedMessageCache(s, closeCh)
+		as.cleanupRetainedMessageCache(s, closeCh)
 	})
 
 	lookupStream := func(stream, txt string) (*StreamInfo, error) {
@@ -1523,51 +1529,106 @@ func (jsa *mqttJSA) prefixDomain(subject string) string {
 	return subject
 }
 
-func (jsa *mqttJSA) newRequestEx(kind, subject, cidHash string, hdr int, msg []byte, timeout time.Duration) (interface{}, error) {
-	var sb strings.Builder
-	// Either we use nuid.Next() which uses a global lock, or our own nuid object, but
-	// then it needs to be "write" protected. This approach will reduce across account
-	// contention since we won't use the global nuid's lock.
-	jsa.mu.Lock()
-	uid := jsa.nuid.Next()
-	sb.WriteString(jsa.rplyr)
-	jsa.mu.Unlock()
-
-	sb.WriteString(kind)
-	sb.WriteByte(btsep)
-	if cidHash != _EMPTY_ {
-		sb.WriteString(cidHash)
-		sb.WriteByte(btsep)
+func (jsa *mqttJSA) newRequestEx(kind, subject, cidHash string, hdr int, msg []byte, timeout time.Duration) (any, error) {
+	responses, err := jsa.newRequestExMulti(kind, subject, cidHash, []int{hdr}, [][]byte{msg}, timeout)
+	if err != nil {
+		return nil, err
+	}
+	if len(responses) != 1 {
+		return nil, fmt.Errorf("unreachable: invalid number of responses (%d)", len(responses))
 	}
-	sb.WriteString(uid)
-	reply := sb.String()
+	return responses[0].value, nil
+}
 
-	ch := make(chan interface{}, 1)
-	jsa.replies.Store(reply, ch)
+// newRequestExMulti sends multiple messages on the same subject and waits for
+// all responses. It returns the same number of responses in the same order as
+// msgs parameter. In case of a timeout it returns an error as well as all
+// responses received as a sparsely populated array, matching msgs, with nils
+// for the values that have not yet been received.
+//
+// Note that each response may represent an error and should be inspected as
+// such by the caller.
+func (jsa *mqttJSA) newRequestExMulti(kind, subject, cidHash string, hdrs []int, msgs [][]byte, timeout time.Duration) ([]*mqttJSAResponse, error) {
+	if len(hdrs) != len(msgs) {
+		return nil, fmt.Errorf("unreachable: invalid number of messages (%d) or header offsets (%d)", len(msgs), len(hdrs))
+	}
+	responseCh := make(chan *mqttJSAResponse, len(msgs))
+
+	// Generate and queue all outgoing requests, have all results reported to
+	// responseCh, and store a map of reply subjects to the original subjects'
+	// indices.
+	r2i := map[string]int{}
+	for i, msg := range msgs {
+		hdr := hdrs[i]
+		var sb strings.Builder
+		// Either we use nuid.Next() which uses a global lock, or our own nuid object, but
+		// then it needs to be "write" protected. This approach will reduce across account
+		// contention since we won't use the global nuid's lock.
+		jsa.mu.Lock()
+		uid := jsa.nuid.Next()
+		sb.WriteString(jsa.rplyr)
+		jsa.mu.Unlock()
+
+		sb.WriteString(kind)
+		sb.WriteByte(btsep)
+		if cidHash != _EMPTY_ {
+			sb.WriteString(cidHash)
+			sb.WriteByte(btsep)
+		}
+		sb.WriteString(uid)
+		reply := sb.String()
 
-	subject = jsa.prefixDomain(subject)
-	jsa.sendq.push(&mqttJSPubMsg{
-		subj:  subject,
-		reply: reply,
-		hdr:   hdr,
-		msg:   msg,
-	})
+		// Add responseCh to the reply channel map. It will be cleaned out on
+		// timeout (see below), or in processJSAPIReplies upon receiving the
+		// response.
+		jsa.replies.Store(reply, responseCh)
+
+		subject = jsa.prefixDomain(subject)
+		jsa.sendq.push(&mqttJSPubMsg{
+			subj:  subject,
+			reply: reply,
+			hdr:   hdr,
+			msg:   msg,
+		})
+		r2i[reply] = i
+	}
 
-	var i interface{}
-	// We don't want to use time.After() which causes memory growth because the timer
-	// can't be stopped and will need to expire to then be garbage collected.
+	// Wait for all responses to come back, or for the timeout to expire. We
+	// don't want to use time.After() which causes memory growth because the
+	// timer can't be stopped and will need to expire to then be garbage
+	// collected.
+	c := 0
+	responses := make([]*mqttJSAResponse, len(msgs))
+	start := time.Now()
 	t := time.NewTimer(timeout)
-	select {
-	case i = <-ch:
-		// Ensure we stop the timer so it can be quickly garbage collected.
-		t.Stop()
-	case <-jsa.quitCh:
-		return nil, ErrServerNotRunning
-	case <-t.C:
-		jsa.replies.Delete(reply)
-		return nil, fmt.Errorf("timeout for request type %q on %q (reply=%q)", kind, subject, reply)
+	defer t.Stop()
+	for {
+		select {
+		case r := <-responseCh:
+			i := r2i[r.reply]
+			responses[i] = r
+			c++
+			if c == len(msgs) {
+				return responses, nil
+			}
+
+		case <-jsa.quitCh:
+			return nil, ErrServerNotRunning
+
+		case <-t.C:
+			var reply string
+			now := time.Now()
+			for reply = range r2i { // preserve the last value for Errorf
+				jsa.replies.Delete(reply)
+			}
+
+			if len(msgs) == 1 {
+				return responses, fmt.Errorf("timeout after %v: request type %q on %q (reply=%q)", now.Sub(start), kind, subject, reply)
+			} else {
+				return responses, fmt.Errorf("timeout after %v: request type %q on %q: got %d out of %d", now.Sub(start), kind, subject, c, len(msgs))
+			}
+		}
 	}
-	return i, nil
 }
 
 func (jsa *mqttJSA) sendAck(ackSubject string) {
@@ -1667,6 +1728,30 @@ func (jsa *mqttJSA) loadLastMsgFor(streamName string, subject string) (*StoredMs
 	return lmr.Message, lmr.ToError()
 }
 
+func (jsa *mqttJSA) loadLastMsgForMulti(streamName string, subjects []string) ([]*JSApiMsgGetResponse, error) {
+	marshaled := make([][]byte, 0, len(subjects))
+	headerBytes := make([]int, 0, len(subjects))
+	for _, subject := range subjects {
+		mreq := &JSApiMsgGetRequest{LastFor: subject}
+		bb, err := json.Marshal(mreq)
+		if err != nil {
+			return nil, err
+		}
+		marshaled = append(marshaled, bb)
+		headerBytes = append(headerBytes, 0)
+	}
+
+	all, err := jsa.newRequestExMulti(mqttJSAMsgLoad, fmt.Sprintf(JSApiMsgGetT, streamName), _EMPTY_, headerBytes, marshaled, mqttJSAPITimeout)
+	// all has the same order as subjects, preserve it as we unmarshal
+	responses := make([]*JSApiMsgGetResponse, len(all))
+	for i, v := range all {
+		if v != nil {
+			responses[i] = v.value.(*JSApiMsgGetResponse)
+		}
+	}
+	return responses, err
+}
+
 func (jsa *mqttJSA) loadNextMsgFor(streamName string, subject string) (*StoredMsg, error) {
 	mreq := &JSApiMsgGetRequest{NextFor: subject}
 	req, err := json.Marshal(mreq)
@@ -1771,68 +1856,71 @@ func (as *mqttAccountSessionManager) processJSAPIReplies(_ *subscription, pc *cl
 		return
 	}
 	jsa.replies.Delete(subject)
-	ch := chi.(chan interface{})
+	ch := chi.(chan *mqttJSAResponse)
+	out := func(value any) {
+		ch <- &mqttJSAResponse{reply: subject, value: value}
+	}
 	switch token {
 	case mqttJSAStreamCreate:
 		var resp = &JSApiStreamCreateResponse{}
 		if err := json.Unmarshal(msg, resp); err != nil {
 			resp.Error = NewJSInvalidJSONError()
 		}
-		ch <- resp
+		out(resp)
 	case mqttJSAStreamUpdate:
 		var resp = &JSApiStreamUpdateResponse{}
 		if err := json.Unmarshal(msg, resp); err != nil {
 			resp.Error = NewJSInvalidJSONError()
 		}
-		ch <- resp
+		out(resp)
 	case mqttJSAStreamLookup:
 		var resp = &JSApiStreamInfoResponse{}
 		if err := json.Unmarshal(msg, &resp); err != nil {
 			resp.Error = NewJSInvalidJSONError()
 		}
-		ch <- resp
+		out(resp)
 	case mqttJSAStreamDel:
 		var resp = &JSApiStreamDeleteResponse{}
 		if err := json.Unmarshal(msg, &resp); err != nil {
 			resp.Error = NewJSInvalidJSONError()
 		}
-		ch <- resp
+		out(resp)
 	case mqttJSAConsumerCreate:
 		var resp = &JSApiConsumerCreateResponse{}
 		if err := json.Unmarshal(msg, resp); err != nil {
 			resp.Error = NewJSInvalidJSONError()
 		}
-		ch <- resp
+		out(resp)
 	case mqttJSAConsumerDel:
 		var resp = &JSApiConsumerDeleteResponse{}
 		if err := json.Unmarshal(msg, resp); err != nil {
 			resp.Error = NewJSInvalidJSONError()
 		}
-		ch <- resp
+		out(resp)
 	case mqttJSAMsgStore, mqttJSASessPersist:
 		var resp = &JSPubAckResponse{}
 		if err := json.Unmarshal(msg, resp); err != nil {
 			resp.Error = NewJSInvalidJSONError()
 		}
-		ch <- resp
+		out(resp)
 	case mqttJSAMsgLoad:
 		var resp = &JSApiMsgGetResponse{}
-		if err := json.Unmarshal(msg, resp); err != nil {
+		if err := json.Unmarshal(msg, &resp); err != nil {
 			resp.Error = NewJSInvalidJSONError()
 		}
-		ch <- resp
+		out(resp)
 	case mqttJSAStreamNames:
 		var resp = &JSApiStreamNamesResponse{}
 		if err := json.Unmarshal(msg, resp); err != nil {
 			resp.Error = NewJSInvalidJSONError()
 		}
-		ch <- resp
+		out(resp)
 	case mqttJSAMsgDelete:
 		var resp = &JSApiMsgDeleteResponse{}
 		if err := json.Unmarshal(msg, resp); err != nil {
 			resp.Error = NewJSInvalidJSONError()
 		}
-		ch <- resp
+		out(resp)
 	default:
 		pc.Warnf("Unknown reply code %q", token)
 	}
@@ -1856,8 +1944,8 @@ func (as *mqttAccountSessionManager) processRetainedMsg(_ *subscription, c *clie
 	// At this point we either recover from our own server, or process a remote retained message.
 	seq, _, _ := ackReplyInfo(reply)
 
-	// Handle this retained message
-	as.handleRetainedMsg(rm.Subject, &mqttRetainedMsgRef{sseq: seq}, rm)
+	// Handle this retained message, no need to copy the bytes.
+	as.handleRetainedMsg(rm.Subject, &mqttRetainedMsgRef{sseq: seq}, rm, false)
 
 	// If we were recovering (lastSeq > 0), then check if we are done.
 	if as.rrmLastSeq > 0 && seq >= as.rrmLastSeq {
@@ -2001,7 +2089,7 @@ func (as *mqttAccountSessionManager) createSubscription(subject string, cb msgHa
 // only used when the server shutdown.
 //
 // No lock held on entry.
-func (as *mqttAccountSessionManager) cleaupRetainedMessageCache(s *Server, closeCh chan struct{}) {
+func (as *mqttAccountSessionManager) cleanupRetainedMessageCache(s *Server, closeCh chan struct{}) {
 	tt := time.NewTicker(mqttRetainedCacheTTL)
 	defer tt.Stop()
 	for {
@@ -2013,7 +2101,7 @@ func (as *mqttAccountSessionManager) cleaupRetainedMessageCache(s *Server, close
 			i, maxScan := 0, 10*1000
 			now := time.Now()
 			as.rmsCache.Range(func(key, value interface{}) bool {
-				rm := value.(mqttRetainedMsg)
+				rm := value.(*mqttRetainedMsg)
 				if now.After(rm.expiresFromCache) {
 					as.rmsCache.Delete(key)
 				}
@@ -2123,7 +2211,7 @@ func (as *mqttAccountSessionManager) sendJSAPIrequests(s *Server, c *client, acc
 // If a message for this topic already existed, the existing record is updated
 // with the provided information.
 // Lock not held on entry.
-func (as *mqttAccountSessionManager) handleRetainedMsg(key string, rf *mqttRetainedMsgRef, rm *mqttRetainedMsg) {
+func (as *mqttAccountSessionManager) handleRetainedMsg(key string, rf *mqttRetainedMsgRef, rm *mqttRetainedMsg, copyBytesToCache bool) {
 	as.mu.Lock()
 	defer as.mu.Unlock()
 	if as.retmsgs == nil {
@@ -2151,11 +2239,7 @@ func (as *mqttAccountSessionManager) handleRetainedMsg(key string, rf *mqttRetai
 			// Update the in-memory retained message cache but only for messages
 			// that are already in the cache, i.e. have been (recently) used.
 			if rm != nil {
-				if _, ok := as.rmsCache.Load(key); ok {
-					toStore := *rm
-					toStore.expiresFromCache = time.Now().Add(mqttRetainedCacheTTL)
-					as.rmsCache.Store(key, toStore)
-				}
+				as.setCachedRetainedMsg(key, rm, true, copyBytesToCache)
 			}
 			return
 		}
@@ -2274,19 +2358,21 @@ func (as *mqttAccountSessionManager) removeSession(sess *mqttSession, lock bool)
 // Session lock held on entry. Acquires the subs lock and holds it for
 // the duration. Non-MQTT messages coming into mqttDeliverMsgCbQoS0 will be
 // waiting.
+func (sess *mqttSession) processQOS12Sub(
+	c *client, // subscribing client.
+	subject, sid []byte, isReserved bool, qos byte, jsDurName string, h msgHandler, // subscription parameters.
+) (*subscription, error) {
+	return sess.processSub(c, subject, sid, isReserved, qos, jsDurName, h, false, false, nil, false, nil)
+}
+
 func (sess *mqttSession) processSub(
-	// subscribing client.
-	c *client,
-	// subscription parameters.
-	subject, sid []byte, isReserved bool, qos byte, jsDurName string, h msgHandler,
-	// do we need to scan for shadow subscriptions? (we don't do it for QOS1+)
-	initShadow bool,
-	// len(rms) > 0 means to deliver retained messages for the subscription.
-	rms map[string]*mqttRetainedMsg,
-	// trace serialized retained messages in the log.
-	trace bool,
-	// the retained messages are kept in the account session manager.
-	as *mqttAccountSessionManager,
+	c *client, // subscribing client.
+	subject, sid []byte, isReserved bool, qos byte, jsDurName string, h msgHandler, // subscription parameters.
+	initShadow bool, // do we need to scan for shadow subscriptions? (not for QOS1+)
+	serializeRMS bool, // do we need to serialize RMS?
+	rms map[string]*mqttRetainedMsg, // preloaded rms (can be empty, or missing items if errors)
+	trace bool, // trace serialized retained messages in the log?
+	as *mqttAccountSessionManager, // needed only for rms serialization.
 ) (*subscription, error) {
 	start := time.Now()
 	defer func() {
@@ -2350,6 +2436,10 @@ func (sess *mqttSession) processSub(
 func (as *mqttAccountSessionManager) processSubs(sess *mqttSession, c *client,
 	filters []*mqttFilter, fromSubProto, trace bool) ([]*subscription, error) {
 
+	c.mu.Lock()
+	acc := c.acc
+	c.mu.Unlock()
+
 	// Helper to determine if we need to create a separate top-level
 	// subscription for a wildcard.
 	fwc := func(subject string) (bool, string, string) {
@@ -2364,29 +2454,7 @@ func (as *mqttAccountSessionManager) processSubs(sess *mqttSession, c *client,
 		return true, fwcsubject, fwcsid
 	}
 
-	// Cache and a helper to load retained messages for a given subject.
-	rms := make(map[string]*mqttRetainedMsg)
-	loadRMS := func(subject []byte) error {
-		sub := &subscription{
-			client:  c,
-			subject: subject,
-			sid:     subject,
-		}
-		c.mu.Lock()
-		acc := c.acc
-		c.mu.Unlock()
-		if err := c.addShadowSubscriptions(acc, sub, false); err != nil {
-			return err
-		}
-		// Best-effort loading the messages, logs on errors (to c.srv), loads
-		// once for subject.
-		as.loadRetainedMessagesForSubject(rms, subject, c.srv)
-		for _, ss := range sub.shadow {
-			as.loadRetainedMessagesForSubject(rms, ss.subject, c.srv)
-		}
-		return nil
-	}
-
+	rmSubjects := map[string]struct{}{}
 	// Preload retained messages for all requested subscriptions.  Also, since
 	// it's the first iteration over the filter list, do some cleanup.
 	for _, f := range filters {
@@ -2416,14 +2484,33 @@ func (as *mqttAccountSessionManager) processSubs(sess *mqttSession, c *client,
 			}
 		}
 
-		// Load retained messages.
+		// Find retained messages.
 		if fromSubProto {
-			if err := loadRMS([]byte(f.filter)); err != nil {
+			addRMSubjects := func(subject string) error {
+				sub := &subscription{
+					client:  c,
+					subject: []byte(subject),
+					sid:     []byte(subject),
+				}
+				if err := c.addShadowSubscriptions(acc, sub, false); err != nil {
+					return err
+				}
+
+				for _, sub := range append([]*subscription{sub}, sub.shadow...) {
+					as.addRetainedSubjectsForSubject(rmSubjects, bytesToString(sub.subject))
+					for _, ss := range sub.shadow {
+						as.addRetainedSubjectsForSubject(rmSubjects, bytesToString(ss.subject))
+					}
+				}
+				return nil
+			}
+
+			if err := addRMSubjects(f.filter); err != nil {
 				f.qos = mqttSubAckFailure
 				continue
 			}
 			if need, subject, _ := fwc(f.filter); need {
-				if err := loadRMS([]byte(subject)); err != nil {
+				if err := addRMSubjects(subject); err != nil {
 					f.qos = mqttSubAckFailure
 					continue
 				}
@@ -2431,6 +2518,14 @@ func (as *mqttAccountSessionManager) processSubs(sess *mqttSession, c *client,
 		}
 	}
 
+	serializeRMS := len(rmSubjects) > 0
+	var rms map[string]*mqttRetainedMsg
+	if serializeRMS {
+		// Make the best effort to load retained messages. We will identify
+		// errors in the next pass.
+		rms = as.loadRetainedMessages(rmSubjects, c)
+	}
+
 	// Small helper to add the consumer config to the session.
 	addJSConsToSess := func(sid string, cc *ConsumerConfig) {
 		if cc == nil {
@@ -2445,7 +2540,6 @@ func (as *mqttAccountSessionManager) processSubs(sess *mqttSession, c *client,
 	var err error
 	subs := make([]*subscription, 0, len(filters))
 	for _, f := range filters {
-
 		// Skip what's already been identified as a failure.
 		if f.qos == mqttSubAckFailure {
 			continue
@@ -2472,7 +2566,7 @@ func (as *mqttAccountSessionManager) processSubs(sess *mqttSession, c *client,
 			bsubject, bsid, isReserved, f.qos, // main subject
 			_EMPTY_, mqttDeliverMsgCbQoS0, // no jsDur for QOS0
 			processShadowSubs,
-			rms, trace, as) // rms is empty if not fromSubProto
+			serializeRMS, rms, trace, as)
 		sess.mu.Unlock()
 		as.mu.Unlock()
 
@@ -2506,7 +2600,7 @@ func (as *mqttAccountSessionManager) processSubs(sess *mqttSession, c *client,
 				[]byte(fwcsubject), []byte(fwcsid), isReserved, f.qos, // FWC (top-level wildcard) subject
 				_EMPTY_, mqttDeliverMsgCbQoS0, // no jsDur for QOS0
 				processShadowSubs,
-				rms, trace, as) // rms is empty if not fromSubProto
+				serializeRMS, rms, trace, as)
 			sess.mu.Unlock()
 			as.mu.Unlock()
 			if err != nil {
@@ -2532,6 +2626,7 @@ func (as *mqttAccountSessionManager) processSubs(sess *mqttSession, c *client,
 		subs = append(subs, sub)
 		addJSConsToSess(sid, jscons)
 	}
+
 	if fromSubProto {
 		err = sess.update(filters, true)
 	}
@@ -2548,20 +2643,21 @@ func (as *mqttAccountSessionManager) processSubs(sess *mqttSession, c *client,
 // Runs from the client's readLoop.
 // Account session manager lock held on entry.
 // Session lock held on entry.
-func (as *mqttAccountSessionManager) serializeRetainedMsgsForSub(rms map[string]*mqttRetainedMsg, sess *mqttSession, c *client, sub *subscription, trace bool) {
+func (as *mqttAccountSessionManager) serializeRetainedMsgsForSub(rms map[string]*mqttRetainedMsg, sess *mqttSession, c *client, sub *subscription, trace bool) error {
 	if len(as.retmsgs) == 0 || len(rms) == 0 {
-		return
+		return nil
 	}
 	result := as.sl.ReverseMatch(string(sub.subject))
 	if len(result.psubs) == 0 {
-		return
+		return nil
 	}
+	toTrace := []mqttPublish{}
 	for _, psub := range result.psubs {
 
 		rm := rms[string(psub.subject)]
 		if rm == nil {
-			// This should not happen since we pre-load messages into the cache
-			// before calling serialize.
+			// This should not happen since we pre-load messages into rms before
+			// calling serialize.
 			continue
 		}
 		var pi uint16
@@ -2591,65 +2687,90 @@ func (as *mqttAccountSessionManager) serializeRetainedMsgsForSub(rms map[string]
 		sub.mqtt.prm = append(sub.mqtt.prm, headerBytes, rm.Msg)
 		c.mu.Unlock()
 		if trace {
-			pp := mqttPublish{
+			toTrace = append(toTrace, mqttPublish{
 				topic: []byte(rm.Topic),
 				flags: flags,
 				pi:    pi,
 				sz:    len(rm.Msg),
-			}
-			c.traceOutOp("PUBLISH", []byte(mqttPubTrace(&pp)))
+			})
 		}
 	}
+	for _, pp := range toTrace {
+		c.traceOutOp("PUBLISH", []byte(mqttPubTrace(&pp)))
+	}
+	return nil
 }
 
-// Returns in the provided slice all publish retained message records that
+// Appends the stored message subjects for all retained message records that
 // match the given subscription's `subject` (which could have wildcards).
 //
 // Account session manager NOT lock held on entry.
-func (as *mqttAccountSessionManager) loadRetainedMessagesForSubject(rms map[string]*mqttRetainedMsg, topSubject []byte, log Logger) {
+func (as *mqttAccountSessionManager) addRetainedSubjectsForSubject(list map[string]struct{}, topSubject string) bool {
 	as.mu.RLock()
 	if len(as.retmsgs) == 0 {
 		as.mu.RUnlock()
-		return
+		return false
 	}
-	result := as.sl.ReverseMatch(string(topSubject))
+	result := as.sl.ReverseMatch(topSubject)
 	as.mu.RUnlock()
 
-	if len(result.psubs) == 0 {
-		return
-	}
+	added := false
 	for _, sub := range result.psubs {
 		subject := string(sub.subject)
-		if rms[subject] != nil {
-			continue // already loaded
+		if _, ok := list[subject]; ok {
+			continue
 		}
+		list[subject] = struct{}{}
+		added = true
+	}
 
-		// See if we have the retained message in the cache.
-		if rmv, _ := as.rmsCache.Load(subject); rmv != nil {
-			rm := rmv.(mqttRetainedMsg)
-			rms[subject] = &rm
-			continue
+	return added
+}
+
+type warner interface {
+	Warnf(format string, v ...any)
+}
+
+// Loads a list of retained messages given a list of stored message subjects.
+func (as *mqttAccountSessionManager) loadRetainedMessages(subjects map[string]struct{}, w warner) map[string]*mqttRetainedMsg {
+	rms := make(map[string]*mqttRetainedMsg, len(subjects))
+	ss := []string{}
+	for s := range subjects {
+		if rm := as.getCachedRetainedMsg(s); rm != nil {
+			rms[s] = rm
+		} else {
+			ss = append(ss, mqttRetainedMsgsStreamSubject+s)
 		}
+	}
 
-		// Load the retained message from the stream, and cache it for reuse in
-		// the near future.
-		loadSubject := mqttRetainedMsgsStreamSubject + subject
-		jsm, err := as.jsa.loadLastMsgFor(mqttRetainedMsgsStreamName, loadSubject)
-		if err != nil || jsm == nil {
-			log.Warnf("failed to load retained message for subject %q: %v", loadSubject, err)
+	if len(ss) == 0 {
+		return rms
+	}
+
+	results, err := as.jsa.loadLastMsgForMulti(mqttRetainedMsgsStreamName, ss)
+	// If an error occurred, warn, but then proceed with what we got.
+	if err != nil {
+		w.Warnf("error loading retained messages: %v", err)
+	}
+	for i, result := range results {
+		if result == nil {
+			continue // skip requests that timed out
+		}
+		if result.ToError() != nil {
+			w.Warnf("failed to load retained message for subject %q: %v", ss[i], err)
 			continue
 		}
 		var rm mqttRetainedMsg
-		if err := json.Unmarshal(jsm.Data, &rm); err != nil {
-			log.Warnf("failed to decode retained message for subject %q: %v", loadSubject, err)
+		if err := json.Unmarshal(result.Message.Data, &rm); err != nil {
+			w.Warnf("failed to decode retained message for subject %q: %v", ss[i], err)
 			continue
 		}
-
-		// Add the loaded retained message to the cache.
-		rm.expiresFromCache = time.Now().Add(mqttRetainedCacheTTL)
-		as.rmsCache.Store(subject, rm)
-		rms[subject] = &rm
+		// Add the loaded retained message to the cache, and to the results map.
+		key := ss[i][len(mqttRetainedMsgsStreamSubject):]
+		as.setCachedRetainedMsg(key, &rm, false, false)
+		rms[key] = &rm
 	}
+	return rms
 }
 
 // Creates the session stream (limit msgs of 1) for this client ID if it does
@@ -2833,6 +2954,32 @@ func (as *mqttAccountSessionManager) transferRetainedToPerKeySubjectStream(log *
 	return nil
 }
 
+func (as *mqttAccountSessionManager) getCachedRetainedMsg(subject string) *mqttRetainedMsg {
+	v, ok := as.rmsCache.Load(subject)
+	if !ok {
+		return nil
+	}
+	rm := v.(*mqttRetainedMsg)
+	if rm.expiresFromCache.Before(time.Now()) {
+		as.rmsCache.Delete(subject)
+		return nil
+	}
+	return rm
+}
+
+func (as *mqttAccountSessionManager) setCachedRetainedMsg(subject string, rm *mqttRetainedMsg, onlyReplace bool, copyBytesToCache bool) {
+	rm.expiresFromCache = time.Now().Add(mqttRetainedCacheTTL)
+	if onlyReplace {
+		if _, ok := as.rmsCache.Load(subject); !ok {
+			return
+		}
+	}
+	if copyBytesToCache {
+		rm.Msg = copyBytes(rm.Msg)
+	}
+	as.rmsCache.Store(subject, rm)
+}
+
 //////////////////////////////////////////////////////////////////////////////
 //
 // MQTT session related functions
@@ -4026,7 +4173,7 @@ func (c *client) mqttHandlePubRetain() {
 				sseq: smr.Sequence,
 			}
 			// Add/update the map
-			asm.handleRetainedMsg(key, rf, rm)
+			asm.handleRetainedMsg(key, rf, rm, true) // will copy the payload bytes if needs to update rmsCache
 		} else {
 			c.mu.Lock()
 			acc := c.acc
@@ -4893,9 +5040,8 @@ func (sess *mqttSession) processJSConsumer(c *client, subject, sid string,
 	// for the JS durable's deliver subject.
 	sess.mu.Lock()
 	sess.tmaxack = tmaxack
-	sub, err := sess.processSub(c, []byte(inbox), []byte(inbox),
-		isMQTTReservedSubscription(subject), qos, cc.Durable, mqttDeliverMsgCbQoS12,
-		false, nil, false, nil) // no shadow subs, no retained message delivery
+	sub, err := sess.processQOS12Sub(c, []byte(inbox), []byte(inbox),
+		isMQTTReservedSubscription(subject), qos, cc.Durable, mqttDeliverMsgCbQoS12)
 	sess.mu.Unlock()
 
 	if err != nil {
diff --git a/server/store.go b/server/store.go
index 7b433c7f..5c486b3e 100644
--- a/server/store.go
+++ b/server/store.go
@@ -1,4 +1,4 @@
-// Copyright 2019-2023 The NATS Authors
+// Copyright 2019-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
@@ -15,7 +15,6 @@ package server
 
 import (
 	"encoding/binary"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -86,6 +85,7 @@ type StreamStore interface {
 	StoreMsg(subject string, hdr, msg []byte) (uint64, int64, error)
 	StoreRawMsg(subject string, hdr, msg []byte, seq uint64, ts int64) error
 	SkipMsg() uint64
+	SkipMsgs(seq uint64, num uint64) error
 	LoadMsg(seq uint64, sm *StoreMsg) (*StoreMsg, error)
 	LoadNextMsg(filter string, wc bool, start uint64, smp *StoreMsg) (sm *StoreMsg, skip uint64, err error)
 	LoadLastMsg(subject string, sm *StoreMsg) (*StoreMsg, error)
@@ -442,14 +442,16 @@ type TemplateStore interface {
 	Delete(*streamTemplate) error
 }
 
-func jsonString(s string) string {
-	return "\"" + s + "\""
-}
-
 const (
-	limitsPolicyString    = "limits"
-	interestPolicyString  = "interest"
-	workQueuePolicyString = "workqueue"
+	limitsPolicyJSONString    = `"limits"`
+	interestPolicyJSONString  = `"interest"`
+	workQueuePolicyJSONString = `"workqueue"`
+)
+
+var (
+	limitsPolicyJSONBytes    = []byte(limitsPolicyJSONString)
+	interestPolicyJSONBytes  = []byte(interestPolicyJSONString)
+	workQueuePolicyJSONBytes = []byte(workQueuePolicyJSONString)
 )
 
 func (rp RetentionPolicy) String() string {
@@ -468,11 +470,11 @@ func (rp RetentionPolicy) String() string {
 func (rp RetentionPolicy) MarshalJSON() ([]byte, error) {
 	switch rp {
 	case LimitsPolicy:
-		return json.Marshal(limitsPolicyString)
+		return limitsPolicyJSONBytes, nil
 	case InterestPolicy:
-		return json.Marshal(interestPolicyString)
+		return interestPolicyJSONBytes, nil
 	case WorkQueuePolicy:
-		return json.Marshal(workQueuePolicyString)
+		return workQueuePolicyJSONBytes, nil
 	default:
 		return nil, fmt.Errorf("can not marshal %v", rp)
 	}
@@ -480,11 +482,11 @@ func (rp RetentionPolicy) MarshalJSON() ([]byte, error) {
 
 func (rp *RetentionPolicy) UnmarshalJSON(data []byte) error {
 	switch string(data) {
-	case jsonString(limitsPolicyString):
+	case limitsPolicyJSONString:
 		*rp = LimitsPolicy
-	case jsonString(interestPolicyString):
+	case interestPolicyJSONString:
 		*rp = InterestPolicy
-	case jsonString(workQueuePolicyString):
+	case workQueuePolicyJSONString:
 		*rp = WorkQueuePolicy
 	default:
 		return fmt.Errorf("can not unmarshal %q", data)
@@ -506,9 +508,9 @@ func (dp DiscardPolicy) String() string {
 func (dp DiscardPolicy) MarshalJSON() ([]byte, error) {
 	switch dp {
 	case DiscardOld:
-		return json.Marshal("old")
+		return []byte(`"old"`), nil
 	case DiscardNew:
-		return json.Marshal("new")
+		return []byte(`"new"`), nil
 	default:
 		return nil, fmt.Errorf("can not marshal %v", dp)
 	}
@@ -516,9 +518,9 @@ func (dp DiscardPolicy) MarshalJSON() ([]byte, error) {
 
 func (dp *DiscardPolicy) UnmarshalJSON(data []byte) error {
 	switch strings.ToLower(string(data)) {
-	case jsonString("old"):
+	case `"old"`:
 		*dp = DiscardOld
-	case jsonString("new"):
+	case `"new"`:
 		*dp = DiscardNew
 	default:
 		return fmt.Errorf("can not unmarshal %q", data)
@@ -527,9 +529,15 @@ func (dp *DiscardPolicy) UnmarshalJSON(data []byte) error {
 }
 
 const (
-	memoryStorageString = "memory"
-	fileStorageString   = "file"
-	anyStorageString    = "any"
+	memoryStorageJSONString = `"memory"`
+	fileStorageJSONString   = `"file"`
+	anyStorageJSONString    = `"any"`
+)
+
+var (
+	memoryStorageJSONBytes = []byte(memoryStorageJSONString)
+	fileStorageJSONBytes   = []byte(fileStorageJSONString)
+	anyStorageJSONBytes    = []byte(anyStorageJSONString)
 )
 
 func (st StorageType) String() string {
@@ -548,11 +556,11 @@ func (st StorageType) String() string {
 func (st StorageType) MarshalJSON() ([]byte, error) {
 	switch st {
 	case MemoryStorage:
-		return json.Marshal(memoryStorageString)
+		return memoryStorageJSONBytes, nil
 	case FileStorage:
-		return json.Marshal(fileStorageString)
+		return fileStorageJSONBytes, nil
 	case AnyStorage:
-		return json.Marshal(anyStorageString)
+		return anyStorageJSONBytes, nil
 	default:
 		return nil, fmt.Errorf("can not marshal %v", st)
 	}
@@ -560,11 +568,11 @@ func (st StorageType) MarshalJSON() ([]byte, error) {
 
 func (st *StorageType) UnmarshalJSON(data []byte) error {
 	switch string(data) {
-	case jsonString(memoryStorageString):
+	case memoryStorageJSONString:
 		*st = MemoryStorage
-	case jsonString(fileStorageString):
+	case fileStorageJSONString:
 		*st = FileStorage
-	case jsonString(anyStorageString):
+	case anyStorageJSONString:
 		*st = AnyStorage
 	default:
 		return fmt.Errorf("can not unmarshal %q", data)
@@ -573,19 +581,25 @@ func (st *StorageType) UnmarshalJSON(data []byte) error {
 }
 
 const (
-	ackNonePolicyString     = "none"
-	ackAllPolicyString      = "all"
-	ackExplicitPolicyString = "explicit"
+	ackNonePolicyJSONString     = `"none"`
+	ackAllPolicyJSONString      = `"all"`
+	ackExplicitPolicyJSONString = `"explicit"`
+)
+
+var (
+	ackNonePolicyJSONBytes     = []byte(ackNonePolicyJSONString)
+	ackAllPolicyJSONBytes      = []byte(ackAllPolicyJSONString)
+	ackExplicitPolicyJSONBytes = []byte(ackExplicitPolicyJSONString)
 )
 
 func (ap AckPolicy) MarshalJSON() ([]byte, error) {
 	switch ap {
 	case AckNone:
-		return json.Marshal(ackNonePolicyString)
+		return ackNonePolicyJSONBytes, nil
 	case AckAll:
-		return json.Marshal(ackAllPolicyString)
+		return ackAllPolicyJSONBytes, nil
 	case AckExplicit:
-		return json.Marshal(ackExplicitPolicyString)
+		return ackExplicitPolicyJSONBytes, nil
 	default:
 		return nil, fmt.Errorf("can not marshal %v", ap)
 	}
@@ -593,11 +607,11 @@ func (ap AckPolicy) MarshalJSON() ([]byte, error) {
 
 func (ap *AckPolicy) UnmarshalJSON(data []byte) error {
 	switch string(data) {
-	case jsonString(ackNonePolicyString):
+	case ackNonePolicyJSONString:
 		*ap = AckNone
-	case jsonString(ackAllPolicyString):
+	case ackAllPolicyJSONString:
 		*ap = AckAll
-	case jsonString(ackExplicitPolicyString):
+	case ackExplicitPolicyJSONString:
 		*ap = AckExplicit
 	default:
 		return fmt.Errorf("can not unmarshal %q", data)
@@ -606,16 +620,21 @@ func (ap *AckPolicy) UnmarshalJSON(data []byte) error {
 }
 
 const (
-	replayInstantPolicyString  = "instant"
-	replayOriginalPolicyString = "original"
+	replayInstantPolicyJSONString  = `"instant"`
+	replayOriginalPolicyJSONString = `"original"`
+)
+
+var (
+	replayInstantPolicyJSONBytes  = []byte(replayInstantPolicyJSONString)
+	replayOriginalPolicyJSONBytes = []byte(replayOriginalPolicyJSONString)
 )
 
 func (rp ReplayPolicy) MarshalJSON() ([]byte, error) {
 	switch rp {
 	case ReplayInstant:
-		return json.Marshal(replayInstantPolicyString)
+		return replayInstantPolicyJSONBytes, nil
 	case ReplayOriginal:
-		return json.Marshal(replayOriginalPolicyString)
+		return replayOriginalPolicyJSONBytes, nil
 	default:
 		return nil, fmt.Errorf("can not marshal %v", rp)
 	}
@@ -623,9 +642,9 @@ func (rp ReplayPolicy) MarshalJSON() ([]byte, error) {
 
 func (rp *ReplayPolicy) UnmarshalJSON(data []byte) error {
 	switch string(data) {
-	case jsonString(replayInstantPolicyString):
+	case replayInstantPolicyJSONString:
 		*rp = ReplayInstant
-	case jsonString(replayOriginalPolicyString):
+	case replayOriginalPolicyJSONString:
 		*rp = ReplayOriginal
 	default:
 		return fmt.Errorf("can not unmarshal %q", data)
@@ -634,28 +653,38 @@ func (rp *ReplayPolicy) UnmarshalJSON(data []byte) error {
 }
 
 const (
-	deliverAllPolicyString       = "all"
-	deliverLastPolicyString      = "last"
-	deliverNewPolicyString       = "new"
-	deliverByStartSequenceString = "by_start_sequence"
-	deliverByStartTimeString     = "by_start_time"
-	deliverLastPerPolicyString   = "last_per_subject"
-	deliverUndefinedString       = "undefined"
+	deliverAllPolicyJSONString       = `"all"`
+	deliverLastPolicyJSONString      = `"last"`
+	deliverNewPolicyJSONString       = `"new"`
+	deliverByStartSequenceJSONString = `"by_start_sequence"`
+	deliverByStartTimeJSONString     = `"by_start_time"`
+	deliverLastPerPolicyJSONString   = `"last_per_subject"`
+	deliverUndefinedJSONString       = `"undefined"`
+)
+
+var (
+	deliverAllPolicyJSONBytes       = []byte(deliverAllPolicyJSONString)
+	deliverLastPolicyJSONBytes      = []byte(deliverLastPolicyJSONString)
+	deliverNewPolicyJSONBytes       = []byte(deliverNewPolicyJSONString)
+	deliverByStartSequenceJSONBytes = []byte(deliverByStartSequenceJSONString)
+	deliverByStartTimeJSONBytes     = []byte(deliverByStartTimeJSONString)
+	deliverLastPerPolicyJSONBytes   = []byte(deliverLastPerPolicyJSONString)
+	deliverUndefinedJSONBytes       = []byte(deliverUndefinedJSONString)
 )
 
 func (p *DeliverPolicy) UnmarshalJSON(data []byte) error {
 	switch string(data) {
-	case jsonString(deliverAllPolicyString), jsonString(deliverUndefinedString):
+	case deliverAllPolicyJSONString, deliverUndefinedJSONString:
 		*p = DeliverAll
-	case jsonString(deliverLastPolicyString):
+	case deliverLastPolicyJSONString:
 		*p = DeliverLast
-	case jsonString(deliverLastPerPolicyString):
+	case deliverLastPerPolicyJSONString:
 		*p = DeliverLastPerSubject
-	case jsonString(deliverNewPolicyString):
+	case deliverNewPolicyJSONString:
 		*p = DeliverNew
-	case jsonString(deliverByStartSequenceString):
+	case deliverByStartSequenceJSONString:
 		*p = DeliverByStartSequence
-	case jsonString(deliverByStartTimeString):
+	case deliverByStartTimeJSONString:
 		*p = DeliverByStartTime
 	default:
 		return fmt.Errorf("can not unmarshal %q", data)
@@ -667,19 +696,19 @@ func (p *DeliverPolicy) UnmarshalJSON(data []byte) error {
 func (p DeliverPolicy) MarshalJSON() ([]byte, error) {
 	switch p {
 	case DeliverAll:
-		return json.Marshal(deliverAllPolicyString)
+		return deliverAllPolicyJSONBytes, nil
 	case DeliverLast:
-		return json.Marshal(deliverLastPolicyString)
+		return deliverLastPolicyJSONBytes, nil
 	case DeliverLastPerSubject:
-		return json.Marshal(deliverLastPerPolicyString)
+		return deliverLastPerPolicyJSONBytes, nil
 	case DeliverNew:
-		return json.Marshal(deliverNewPolicyString)
+		return deliverNewPolicyJSONBytes, nil
 	case DeliverByStartSequence:
-		return json.Marshal(deliverByStartSequenceString)
+		return deliverByStartSequenceJSONBytes, nil
 	case DeliverByStartTime:
-		return json.Marshal(deliverByStartTimeString)
+		return deliverByStartTimeJSONBytes, nil
 	default:
-		return json.Marshal(deliverUndefinedString)
+		return deliverUndefinedJSONBytes, nil
 	}
 }
 
diff --git a/server/stream.go b/server/stream.go
index bf0139b2..2e18dc62 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -1,4 +1,4 @@
-// Copyright 2019-2023 The NATS Authors
+// Copyright 2019-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
@@ -29,6 +29,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/klauspost/compress/s2"
@@ -219,7 +220,7 @@ type stream struct {
 	srv       *Server
 	client    *client
 	sysc      *client
-	sid       int
+	sid       atomic.Uint64
 	pubAck    []byte
 	outq      *jsOutQ
 	msgs      *ipQueue[*inMsg]
@@ -242,7 +243,7 @@ type stream struct {
 	mqch      chan struct{}
 	active    bool
 	ddloaded  bool
-	closed    bool
+	closed    atomic.Bool
 
 	// Mirror
 	mirror *sourceInfo
@@ -276,7 +277,7 @@ type stream struct {
 	// Clustered mode.
 	sa         *streamAssignment
 	node       RaftNode
-	catchup    bool
+	catchup    atomic.Bool
 	syncSub    *subscription
 	infoSub    *subscription
 	clMu       sync.Mutex
@@ -1955,9 +1956,9 @@ func (mset *stream) updateWithAdvisory(config *StreamConfig, sendAdvisory bool)
 // Purge will remove all messages from the stream and underlying store based on the request.
 func (mset *stream) purge(preq *JSApiStreamPurgeRequest) (purged uint64, err error) {
 	mset.mu.RLock()
-	if mset.client == nil || mset.store == nil {
+	if mset.closed.Load() {
 		mset.mu.RUnlock()
-		return 0, errors.New("invalid stream")
+		return 0, errStreamClosed
 	}
 	if mset.cfg.Sealed {
 		mset.mu.RUnlock()
@@ -2022,24 +2023,17 @@ func (mset *stream) removeMsg(seq uint64) (bool, error) {
 
 // DeleteMsg will remove a message from a stream.
 func (mset *stream) deleteMsg(seq uint64) (bool, error) {
-	mset.mu.RLock()
-	if mset.client == nil {
-		mset.mu.RUnlock()
-		return false, fmt.Errorf("invalid stream")
+	if mset.closed.Load() {
+		return false, errStreamClosed
 	}
-	mset.mu.RUnlock()
-
 	return mset.store.RemoveMsg(seq)
 }
 
 // EraseMsg will securely remove a message and rewrite the data with random data.
 func (mset *stream) eraseMsg(seq uint64) (bool, error) {
-	mset.mu.RLock()
-	if mset.client == nil {
-		mset.mu.RUnlock()
-		return false, fmt.Errorf("invalid stream")
+	if mset.closed.Load() {
+		return false, errStreamClosed
 	}
-	mset.mu.RUnlock()
 	return mset.store.EraseMsg(seq)
 }
 
@@ -2367,23 +2361,28 @@ func (mset *stream) retryMirrorConsumer() error {
 // Lock should be held.
 func (mset *stream) skipMsgs(start, end uint64) {
 	node, store := mset.node, mset.store
+	// If we are not clustered we can short circuit now with store.SkipMsgs
+	if node == nil {
+		store.SkipMsgs(start, end-start+1)
+		mset.lseq = end
+		return
+	}
+
+	// FIXME (dlc) - We should allow proposals of DeleteEange, but would need to make sure all peers support.
+	// With syncRequest was easy to add bool into request.
 	var entries []*Entry
 	for seq := start; seq <= end; seq++ {
-		if node != nil {
-			entries = append(entries, &Entry{EntryNormal, encodeStreamMsg(_EMPTY_, _EMPTY_, nil, nil, seq-1, 0)})
-			// So a single message does not get too big.
-			if len(entries) > 10_000 {
-				node.ProposeDirect(entries)
-				// We need to re-craete `entries` because there is a reference
-				// to it in the node's pae map.
-				entries = entries[:0]
-			}
-		} else {
-			mset.lseq = store.SkipMsg()
+		entries = append(entries, &Entry{EntryNormal, encodeStreamMsg(_EMPTY_, _EMPTY_, nil, nil, seq-1, 0)})
+		// So a single message does not get too big.
+		if len(entries) > 10_000 {
+			node.ProposeDirect(entries)
+			// We need to re-create `entries` because there is a reference
+			// to it in the node's pae map.
+			entries = entries[:0]
 		}
 	}
 	// Send all at once.
-	if node != nil && len(entries) > 0 {
+	if len(entries) > 0 {
 		node.ProposeDirect(entries)
 	}
 }
@@ -2435,7 +2434,7 @@ func (mset *stream) scheduleSetupMirrorConsumerRetry() {
 // Setup our mirror consumer.
 // Lock should be held.
 func (mset *stream) setupMirrorConsumer() error {
-	if mset.closed {
+	if mset.closed.Load() {
 		return errStreamClosed
 	}
 	if mset.outq == nil {
@@ -2547,7 +2546,7 @@ func (mset *stream) setupMirrorConsumer() error {
 	respCh := make(chan *JSApiConsumerCreateResponse, 1)
 	reply := infoReplySubject()
 	crSub, err := mset.subscribeInternal(reply, func(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
-		mset.unsubscribeUnlocked(sub)
+		mset.unsubscribe(sub)
 		_, msg := c.msgParts(rmsg)
 
 		var ccr JSApiConsumerCreateResponse
@@ -2556,7 +2555,10 @@ func (mset *stream) setupMirrorConsumer() error {
 			mset.setMirrorErr(ApiErrors[JSInvalidJSONErr])
 			return
 		}
-		respCh <- &ccr
+		select {
+		case respCh <- &ccr:
+		default:
+		}
 	})
 	if err != nil {
 		mirror.err = NewJSMirrorConsumerSetupFailedError(err, Unless(err))
@@ -2688,7 +2690,7 @@ func (mset *stream) setupMirrorConsumer() error {
 			mset.mu.Unlock()
 			ready.Wait()
 		case <-time.After(5 * time.Second):
-			mset.unsubscribeUnlocked(crSub)
+			mset.unsubscribe(crSub)
 			// We already waited 5 seconds, let's retry now.
 			retry = true
 		}
@@ -2819,7 +2821,7 @@ func (mset *stream) scheduleSetSourceConsumer(iname string, seq uint64, delay ti
 // Lock should be held.
 func (mset *stream) setSourceConsumer(iname string, seq uint64, startTime time.Time) {
 	// Ignore if closed.
-	if mset.closed {
+	if mset.closed.Load() {
 		return
 	}
 
@@ -2906,14 +2908,17 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64, startTime time.T
 	respCh := make(chan *JSApiConsumerCreateResponse, 1)
 	reply := infoReplySubject()
 	crSub, err := mset.subscribeInternal(reply, func(sub *subscription, c *client, _ *Account, subject, reply string, rmsg []byte) {
-		mset.unsubscribeUnlocked(sub)
+		mset.unsubscribe(sub)
 		_, msg := c.msgParts(rmsg)
 		var ccr JSApiConsumerCreateResponse
 		if err := json.Unmarshal(msg, &ccr); err != nil {
 			c.Warnf("JetStream bad source consumer create response: %q", msg)
 			return
 		}
-		respCh <- &ccr
+		select {
+		case respCh <- &ccr:
+		default:
+		}
 	})
 	if err != nil {
 		si.err = NewJSSourceConsumerSetupFailedError(err, Unless(err))
@@ -3039,7 +3044,7 @@ func (mset *stream) setSourceConsumer(iname string, seq uint64, startTime time.T
 			mset.mu.Unlock()
 			ready.Wait()
 		case <-time.After(5 * time.Second):
-			mset.unsubscribeUnlocked(crSub)
+			mset.unsubscribe(crSub)
 			// We already waited 5 seconds, let's retry now.
 			retry = true
 		}
@@ -3668,55 +3673,43 @@ func (mset *stream) unsubscribeToStream(stopping bool) error {
 	return nil
 }
 
-// Lock should be held.
+// Lock does NOT need to be held, we set the client on setup and never change it at this point.
 func (mset *stream) subscribeInternal(subject string, cb msgHandler) (*subscription, error) {
-	c := mset.client
-	if c == nil {
-		return nil, fmt.Errorf("invalid stream")
+	if mset.closed.Load() {
+		return nil, errStreamClosed
 	}
 	if cb == nil {
-		return nil, fmt.Errorf("undefined message handler")
+		return nil, errInvalidMsgHandler
 	}
-
-	mset.sid++
-
+	c := mset.client
+	sid := int(mset.sid.Add(1))
 	// Now create the subscription
-	return c.processSub([]byte(subject), nil, []byte(strconv.Itoa(mset.sid)), cb, false)
-}
-
-// Helper for unlocked stream.
-func (mset *stream) subscribeInternalUnlocked(subject string, cb msgHandler) (*subscription, error) {
-	mset.mu.Lock()
-	defer mset.mu.Unlock()
-	return mset.subscribeInternal(subject, cb)
+	return c.processSub([]byte(subject), nil, []byte(strconv.Itoa(sid)), cb, false)
 }
 
-// Lock should be held.
+// Lock does NOT need to be held, we set the client on setup and never change it at this point.
 func (mset *stream) queueSubscribeInternal(subject, group string, cb msgHandler) (*subscription, error) {
-	c := mset.client
-	if c == nil {
-		return nil, fmt.Errorf("invalid stream")
+	if mset.closed.Load() {
+		return nil, errStreamClosed
 	}
 	if cb == nil {
-		return nil, fmt.Errorf("undefined message handler")
+		return nil, errInvalidMsgHandler
 	}
-
-	mset.sid++
-
+	c := mset.client
+	sid := int(mset.sid.Add(1))
 	// Now create the subscription
-	return c.processSub([]byte(subject), []byte(group), []byte(strconv.Itoa(mset.sid)), cb, false)
+	return c.processSub([]byte(subject), []byte(group), []byte(strconv.Itoa(sid)), cb, false)
 }
 
 // This will unsubscribe us from the exact subject given.
 // We do not currently track the subs so do not have the sid.
 // This should be called only on an update.
-// Lock should be held.
+// Lock does NOT need to be held, we set the client on setup and never change it at this point.
 func (mset *stream) unsubscribeInternal(subject string) error {
-	c := mset.client
-	if c == nil {
-		return fmt.Errorf("invalid stream")
+	if mset.closed.Load() {
+		return errStreamClosed
 	}
-
+	c := mset.client
 	var sid []byte
 	c.mu.Lock()
 	for _, sub := range c.subs {
@@ -3735,18 +3728,12 @@ func (mset *stream) unsubscribeInternal(subject string) error {
 
 // Lock should be held.
 func (mset *stream) unsubscribe(sub *subscription) {
-	if sub == nil || mset.client == nil {
+	if sub == nil || mset.closed.Load() {
 		return
 	}
 	mset.client.processUnsub(sub.sid)
 }
 
-func (mset *stream) unsubscribeUnlocked(sub *subscription) {
-	mset.mu.Lock()
-	mset.unsubscribe(sub)
-	mset.mu.Unlock()
-}
-
 func (mset *stream) setupStore(fsCfg *FileStoreConfig) error {
 	mset.mu.Lock()
 	mset.created = time.Now().UTC()
@@ -4126,20 +4113,21 @@ func (mset *stream) processInboundJetStreamMsg(_ *subscription, c *client, _ *Ac
 }
 
 var (
-	errLastSeqMismatch = errors.New("last sequence mismatch")
-	errMsgIdDuplicate  = errors.New("msgid is duplicate")
-	errStreamClosed    = errors.New("stream closed")
+	errLastSeqMismatch   = errors.New("last sequence mismatch")
+	errMsgIdDuplicate    = errors.New("msgid is duplicate")
+	errStreamClosed      = errors.New("stream closed")
+	errInvalidMsgHandler = errors.New("undefined message handler")
 )
 
 // processJetStreamMsg is where we try to actually process the stream msg.
 func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte, lseq uint64, ts int64) error {
-	mset.mu.Lock()
-	c, s, store := mset.client, mset.srv, mset.store
-	if mset.closed || c == nil {
-		mset.mu.Unlock()
+	if mset.closed.Load() {
 		return errStreamClosed
 	}
 
+	mset.mu.Lock()
+	s, store := mset.srv, mset.store
+
 	// Apply the input subject transform if any
 	if mset.itr != nil {
 		ts, err := mset.itr.Match(subject)
@@ -4959,8 +4947,9 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 	jsa.mu.Unlock()
 
 	// Mark as closed, kick monitor and collect consumers first.
+	mset.closed.Store(true)
+
 	mset.mu.Lock()
-	mset.closed = true
 	// Signal to the monitor loop.
 	// Can't use qch here.
 	if mset.mqch != nil {
@@ -5064,7 +5053,6 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 	// Snapshot store.
 	store := mset.store
 	c := mset.client
-	mset.client = nil
 
 	// Clustered cleanup.
 	mset.mu.Unlock()
@@ -5509,7 +5497,7 @@ func (mset *stream) ackMsg(o *consumer, seq uint64) {
 
 	// Don't make this RLock(). We need to have only 1 running at a time to gauge interest across all consumers.
 	mset.mu.Lock()
-	if mset.closed || mset.store == nil || mset.cfg.Retention == LimitsPolicy {
+	if mset.closed.Load() || mset.cfg.Retention == LimitsPolicy {
 		mset.mu.Unlock()
 		return
 	}
@@ -5556,14 +5544,10 @@ func (mset *stream) ackMsg(o *consumer, seq uint64) {
 
 // Snapshot creates a snapshot for the stream and possibly consumers.
 func (mset *stream) snapshot(deadline time.Duration, checkMsgs, includeConsumers bool) (*SnapshotResult, error) {
-	mset.mu.RLock()
-	if mset.client == nil || mset.store == nil {
-		mset.mu.RUnlock()
-		return nil, errors.New("invalid stream")
+	if mset.closed.Load() {
+		return nil, errStreamClosed
 	}
 	store := mset.store
-	mset.mu.RUnlock()
-
 	return store.Snapshot(deadline, checkMsgs, includeConsumers)
 }
 


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 6d98f4d1..71b2a565 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -1,4 +1,4 @@
-// Copyright 2019-2023 The NATS Authors
+// Copyright 2019-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
@@ -6440,7 +6440,60 @@ func TestFileStoreExpireCacheOnLinearWalk(t *testing.T) {
 		}
 	}
 	checkNoCache()
+}
+
+func TestFileStoreSkipMsgs(t *testing.T) {
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: t.TempDir(), BlockSize: 1024},
+		StreamConfig{Name: "zzz", Subjects: []string{"*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	// Test on empty FS first.
+	// Make sure wrong starting sequence fails.
+	err = fs.SkipMsgs(10, 100)
+	require_Error(t, err, ErrSequenceMismatch)
+
+	err = fs.SkipMsgs(1, 100)
+	require_NoError(t, err)
+
+	state := fs.State()
+	require_Equal(t, state.FirstSeq, 101)
+	require_Equal(t, state.LastSeq, 100)
+	require_Equal(t, fs.numMsgBlocks(), 1)
+
+	// Now add alot.
+	err = fs.SkipMsgs(101, 100_000)
+	require_NoError(t, err)
+	state = fs.State()
+	require_Equal(t, state.FirstSeq, 100_101)
+	require_Equal(t, state.LastSeq, 100_100)
+	require_Equal(t, fs.numMsgBlocks(), 1)
+
+	// Now add in a message, and then skip to check dmap.
+	fs, err = newFileStore(
+		FileStoreConfig{StoreDir: t.TempDir(), BlockSize: 1024},
+		StreamConfig{Name: "zzz", Subjects: []string{"*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
 
+	fs.StoreMsg("foo", nil, nil)
+	err = fs.SkipMsgs(2, 10)
+	require_NoError(t, err)
+	state = fs.State()
+	require_Equal(t, state.FirstSeq, 1)
+	require_Equal(t, state.LastSeq, 11)
+	require_Equal(t, state.Msgs, 1)
+	require_Equal(t, state.NumDeleted, 10)
+	require_Equal(t, len(state.Deleted), 10)
+
+	// Check Fast State too.
+	state.Deleted = nil
+	fs.FastState(&state)
+	require_Equal(t, state.FirstSeq, 1)
+	require_Equal(t, state.LastSeq, 11)
+	require_Equal(t, state.Msgs, 1)
+	require_Equal(t, state.NumDeleted, 10)
 }
 
 ///////////////////////////////////////////////////////////////////////////
diff --git a/server/memstore_test.go b/server/memstore_test.go
index 99ce1a9b..34dae8db 100644
--- a/server/memstore_test.go
+++ b/server/memstore_test.go
@@ -1,4 +1,4 @@
-// Copyright 2019-2023 The NATS Authors
+// Copyright 2019-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
@@ -859,3 +859,54 @@ func TestMemStoreGetSeqFromTimeWithLastDeleted(t *testing.T) {
 	// Make sure we get the right value.
 	require_Equal(t, seq, 501)
 }
+
+func TestMemStoreSkipMsgs(t *testing.T) {
+	cfg := &StreamConfig{
+		Name:     "zzz",
+		Subjects: []string{"*"},
+		Storage:  MemoryStorage,
+	}
+	ms, err := newMemStore(cfg)
+	require_NoError(t, err)
+
+	// Test on empty FS first.
+	// Make sure wrong starting sequence fails.
+	err = ms.SkipMsgs(10, 100)
+	require_Error(t, err, ErrSequenceMismatch)
+
+	err = ms.SkipMsgs(1, 100)
+	require_NoError(t, err)
+
+	state := ms.State()
+	require_Equal(t, state.FirstSeq, 101)
+	require_Equal(t, state.LastSeq, 100)
+
+	// Now add alot.
+	err = ms.SkipMsgs(101, 100_000)
+	require_NoError(t, err)
+	state = ms.State()
+	require_Equal(t, state.FirstSeq, 100_101)
+	require_Equal(t, state.LastSeq, 100_100)
+
+	// Now add in a message, and then skip to check dmap.
+	ms, err = newMemStore(cfg)
+	require_NoError(t, err)
+	ms.StoreMsg("foo", nil, nil)
+
+	err = ms.SkipMsgs(2, 10)
+	require_NoError(t, err)
+	state = ms.State()
+	require_Equal(t, state.FirstSeq, 1)
+	require_Equal(t, state.LastSeq, 11)
+	require_Equal(t, state.Msgs, 1)
+	require_Equal(t, state.NumDeleted, 10)
+	require_Equal(t, len(state.Deleted), 10)
+
+	// Check Fast State too.
+	state.Deleted = nil
+	ms.FastState(&state)
+	require_Equal(t, state.FirstSeq, 1)
+	require_Equal(t, state.LastSeq, 11)
+	require_Equal(t, state.Msgs, 1)
+	require_Equal(t, state.NumDeleted, 10)
+}
diff --git a/server/mqtt_test.go b/server/mqtt_test.go
index ae6ba887..7824e690 100644
--- a/server/mqtt_test.go
+++ b/server/mqtt_test.go
@@ -1938,7 +1938,7 @@ func testMQTTSub(t testing.TB, pi uint16, c net.Conn, r *mqttReader, filters []*
 		if err != nil {
 			t.Fatal(err)
 		}
-		if qos != expected[i] {
+		if expected != nil && qos != expected[i] {
 			t.Fatalf("For topic filter %q expected qos of %v, got %v",
 				filters[i].filter, expected[i], qos)
 		}
@@ -2086,13 +2086,42 @@ func testMQTTGetPubMsgExEx(t testing.TB, _ net.Conn, r *mqttReader, b byte, pl i
 			msgLen, r.pos, len(r.buf))
 	}
 	ppayload := r.buf[r.pos : r.pos+msgLen]
-	if !bytes.Equal(payload, ppayload) {
+	if payload != nil && !bytes.Equal(payload, ppayload) {
 		t.Fatalf("Expected payload %q, got %q", payload, ppayload)
 	}
 	r.pos += msgLen
 	return pflags, pi, ptopic
 }
 
+func testMQTTReadPubPacket(t testing.TB, r *mqttReader) (flags byte, pi uint16, topic string, payload []byte) {
+	t.Helper()
+	b, pl := testMQTTReadPacket(t, r)
+	if pt := b & mqttPacketMask; pt != mqttPacketPub {
+		t.Fatalf("Expected PUBLISH packet %x, got %x", mqttPacketPub, pt)
+	}
+	flags = b & mqttPacketFlagMask
+	start := r.pos
+	topic, err := r.readString("topic name")
+	if err != nil {
+		t.Fatal(err)
+	}
+	qos := (flags & mqttPubFlagQoS) >> 1
+	if qos > 0 {
+		pi, err = r.readUint16("packet identifier")
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	msgLen := pl - (r.pos - start)
+	if r.pos+msgLen > len(r.buf) {
+		t.Fatalf("computed message length goes beyond buffer: ml=%v pos=%v lenBuf=%v",
+			msgLen, r.pos, len(r.buf))
+	}
+	payload = r.buf[r.pos : r.pos+msgLen]
+	r.pos += msgLen
+	return flags, pi, topic, payload
+}
+
 func testMQTTSendPIPacket(packetType byte, t testing.TB, c net.Conn, pi uint16) {
 	t.Helper()
 	w := newMQTTWriter(0)
@@ -3215,7 +3244,7 @@ func TestMQTTRetainedMsgNetworkUpdates(t *testing.T) {
 			for _, a := range test.order {
 				if a.add {
 					rf := &mqttRetainedMsgRef{sseq: a.seq}
-					asm.handleRetainedMsg(test.subject, rf, nil)
+					asm.handleRetainedMsg(test.subject, rf, nil, false)
 				} else {
 					asm.handleRetainedMsgDel(test.subject, a.seq)
 				}
@@ -3228,7 +3257,7 @@ func TestMQTTRetainedMsgNetworkUpdates(t *testing.T) {
 		t.Run("clear_"+subject, func(t *testing.T) {
 			// Now add a new message, which should clear the floor.
 			rf := &mqttRetainedMsgRef{sseq: 3}
-			asm.handleRetainedMsg(subject, rf, nil)
+			asm.handleRetainedMsg(subject, rf, nil, false)
 			check(t, subject, true, 3, 0)
 			// Now do a non network delete and make sure it is gone.
 			asm.handleRetainedMsgDel(subject, 0)
@@ -3249,7 +3278,7 @@ func TestMQTTRetainedMsgDel(t *testing.T) {
 	var i uint64
 	for i = 0; i < 3; i++ {
 		rf := &mqttRetainedMsgRef{sseq: i}
-		asm.handleRetainedMsg("subject", rf, nil)
+		asm.handleRetainedMsg("subject", rf, nil, false)
 	}
 	asm.handleRetainedMsgDel("subject", 2)
 	if asm.sl.count > 0 {
@@ -3277,7 +3306,8 @@ func TestMQTTRetainedMsgMigration(t *testing.T) {
 	require_NoError(t, err)
 
 	// Publish some retained messages on the old "$MQTT.rmsgs" subject.
-	for i := 0; i < 100; i++ {
+	const N = 100
+	for i := 0; i < N; i++ {
 		msg := fmt.Sprintf(
 			`{"origin":"b5IQZNtG","subject":"test%d","topic":"test%d","msg":"YmFy","flags":1}`, i, i,
 		)
@@ -3293,8 +3323,8 @@ func TestMQTTRetainedMsgMigration(t *testing.T) {
 	if si.State.NumSubjects != 1 {
 		t.Fatalf("expected 1 subject, got %d", si.State.NumSubjects)
 	}
-	if n := si.State.Subjects[`$MQTT.rmsgs`]; n != 100 {
-		t.Fatalf("expected to find 100 messages on the original subject but found %d", n)
+	if n := si.State.Subjects[`$MQTT.rmsgs`]; n != N {
+		t.Fatalf("expected to find %d messages on the original subject but found %d", N, n)
 	}
 
 	// Create an MQTT client, this will cause a migration to take place.
@@ -3304,22 +3334,22 @@ func TestMQTTRetainedMsgMigration(t *testing.T) {
 
 	testMQTTSub(t, 1, mc, rc, []*mqttFilter{{filter: "+", qos: 0}}, []byte{0})
 	topics := map[string]struct{}{}
-	for i := 0; i < 100; i++ {
+	for i := 0; i < N; i++ {
 		_, _, topic := testMQTTGetPubMsgEx(t, mc, rc, _EMPTY_, []byte("bar"))
 		topics[topic] = struct{}{}
 	}
-	if len(topics) != 100 {
+	if len(topics) != N {
 		t.Fatalf("Unexpected topics: %v", topics)
 	}
 
-	// Now look at the stream, there should be 100 messages on the new
+	// Now look at the stream, there should be N messages on the new
 	// divided subjects and none on the old undivided subject.
 	si, err = js.StreamInfo(mqttRetainedMsgsStreamName, &nats.StreamInfoRequest{
 		SubjectsFilter: `$MQTT.>`,
 	})
 	require_NoError(t, err)
-	if si.State.NumSubjects != 100 {
-		t.Fatalf("expected 100 subjects, got %d", si.State.NumSubjects)
+	if si.State.NumSubjects != N {
+		t.Fatalf("expected %d subjects, got %d", N, si.State.NumSubjects)
 	}
 	if n := si.State.Subjects[`$MQTT.rmsgs`]; n > 0 {
 		t.Fatalf("expected to find no messages on the original subject but found %d", n)
@@ -3327,7 +3357,7 @@ func TestMQTTRetainedMsgMigration(t *testing.T) {
 
 	// Check that the message counts look right. There should be one
 	// retained message per key.
-	for i := 0; i < 100; i++ {
+	for i := 0; i < N; i++ {
 		expected := fmt.Sprintf(`$MQTT.rmsgs.test%d`, i)
 		n, ok := si.State.Subjects[expected]
 		if !ok {
@@ -4953,12 +4983,12 @@ func TestMQTTPersistRetainedMsg(t *testing.T) {
 	defer c.Close()
 	testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, false)
 
-	testMQTTPublish(t, c, r, 1, false, true, "foo", 1, []byte("foo1"))
-	testMQTTPublish(t, c, r, 1, false, true, "foo", 1, []byte("foo2"))
-	testMQTTPublish(t, c, r, 1, false, true, "bar", 1, []byte("bar1"))
-	testMQTTPublish(t, c, r, 0, false, true, "baz", 1, []byte("baz1"))
+	testMQTTPublish(t, c, r, 1, false, true, "moo/foo", 1, []byte("foo1"))
+	testMQTTPublish(t, c, r, 1, false, true, "moo/foo", 1, []byte("foo2"))
+	testMQTTPublish(t, c, r, 1, false, true, "moo/bar", 1, []byte("bar1"))
+	testMQTTPublish(t, c, r, 0, false, true, "moo/baz", 1, []byte("baz1"))
 	// Remove bar
-	testMQTTPublish(t, c, r, 1, false, true, "bar", 1, nil)
+	testMQTTPublish(t, c, r, 1, false, true, "moo/bar", 1, nil)
 	testMQTTFlush(t, c, nil, r)
 	testMQTTDisconnect(t, c, nil)
 	c.Close()
@@ -4969,26 +4999,108 @@ func TestMQTTPersistRetainedMsg(t *testing.T) {
 	o.MQTT.Port = -1
 	o.StoreDir = dir
 	s = testMQTTRunServer(t, o)
-	// There is already the defer for shutdown at top of function
 
 	cisub := &mqttConnInfo{clientID: "sub", cleanSess: false}
 	c, r = testMQTTConnect(t, cisub, o.MQTT.Host, o.MQTT.Port)
 	defer c.Close()
 	testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, false)
 
-	testMQTTSub(t, 1, c, r, []*mqttFilter{{filter: "foo", qos: 1}}, []byte{1})
-	testMQTTCheckPubMsg(t, c, r, "foo", mqttPubFlagRetain|mqttPubQos1, []byte("foo2"))
+	t.Run("many subs one topic", func(t *testing.T) {
+		testMQTTSub(t, 1, c, r, []*mqttFilter{{filter: "moo/foo", qos: 1}}, []byte{1})
+		testMQTTCheckPubMsg(t, c, r, "moo/foo", mqttPubFlagRetain|mqttPubQos1, []byte("foo2"))
 
-	testMQTTSub(t, 1, c, r, []*mqttFilter{{filter: "baz", qos: 1}}, []byte{1})
-	testMQTTCheckPubMsg(t, c, r, "baz", mqttPubFlagRetain, []byte("baz1"))
+		testMQTTSub(t, 1, c, r, []*mqttFilter{{filter: "moo/baz", qos: 1}}, []byte{1})
+		testMQTTCheckPubMsg(t, c, r, "moo/baz", mqttPubFlagRetain, []byte("baz1"))
 
-	testMQTTSub(t, 1, c, r, []*mqttFilter{{filter: "bar", qos: 1}}, []byte{1})
-	testMQTTExpectNothing(t, r)
+		testMQTTSub(t, 1, c, r, []*mqttFilter{{filter: "moo/bar", qos: 1}}, []byte{1})
+		testMQTTExpectNothing(t, r)
+	})
+
+	runSingleSubscribe := func(filters []*mqttFilter) func(t *testing.T) {
+		return func(t *testing.T) {
+			testMQTTSub(t, 1, c, r, filters, nil)
+
+			for i := 0; i < 2; i++ {
+				flags, pi, topic, payload := testMQTTReadPubPacket(t, r)
+				if (flags & mqttPubFlagRetain) == 0 {
+					t.Fatalf("Expected flags to have retain set, got %v", flags)
+				}
+				if (flags & mqttPubFlagDup) != 0 {
+					t.Fatalf("Expected flags to not have Dup set, got %v", flags)
+				}
+				var expQOS byte
+				var expPayload []byte
+				switch topic {
+				case "moo/foo":
+					expQOS = mqttPubQos1
+					expPayload = []byte("foo2")
+					testMQTTSendPIPacket(mqttPacketPubAck, t, c, pi)
+				case "moo/baz":
+					expQOS = 0
+					expPayload = []byte("baz1")
+				default:
+					t.Fatalf("Unexpected topic: %v", topic)
+				}
+				if (flags & mqttPubFlagQoS) != expQOS {
+					t.Fatalf("Expected flags to have QOS %x set, got %x", expQOS, flags)
+				}
+				if string(payload) != string(expPayload) {
+					t.Fatalf("Expected payload to be %q, got %v", expPayload, string(payload))
+				}
+			}
+			testMQTTExpectNothing(t, r)
+		}
+	}
+
+	t.Run("one sub many topics", runSingleSubscribe([]*mqttFilter{
+		{filter: "moo/foo", qos: 1},
+		{filter: "moo/baz", qos: 1},
+		{filter: "moo/bar", qos: 1},
+	}))
+
+	t.Run("one sub wildcard plus", runSingleSubscribe([]*mqttFilter{
+		{filter: "moo/+", qos: 1},
+	}))
+
+	t.Run("one sub wildcard hash", runSingleSubscribe([]*mqttFilter{
+		{filter: "moo/#", qos: 1},
+	}))
 
 	testMQTTDisconnect(t, c, nil)
 	c.Close()
 }
 
+func TestMQTTRetainedMsgCleanup(t *testing.T) {
+	mqttRetainedCacheTTL = 250 * time.Millisecond
+	defer func() { mqttRetainedCacheTTL = mqttDefaultRetainedCacheTTL }()
+
+	o := testMQTTDefaultOptions()
+	s := testMQTTRunServer(t, o)
+	defer testMQTTShutdownServer(s)
+
+	ci := &mqttConnInfo{clientID: "cache", cleanSess: true}
+	c, r := testMQTTConnect(t, ci, o.MQTT.Host, o.MQTT.Port)
+	defer c.Close()
+	testMQTTCheckConnAck(t, r, mqttConnAckRCConnectionAccepted, false)
+
+	// Send a retained message.
+	testMQTTPublish(t, c, r, 1, false, true, "foo", 1, []byte("msg"))
+	testMQTTFlush(t, c, nil, r)
+
+	// Start a subscription.
+	testMQTTSub(t, 1, c, r, []*mqttFilter{{filter: "foo", qos: 1}}, []byte{1})
+	testMQTTCheckPubMsg(t, c, r, "foo", mqttPubQos1|mqttPubFlagRetain, []byte("msg"))
+
+	time.Sleep(2 * mqttRetainedCacheTTL)
+
+	// Make sure not in cache anymore
+	cli := testMQTTGetClient(t, s, "cache")
+	asm := cli.mqtt.asm
+	if v, ok := asm.rmsCache.Load("foo"); ok {
+		t.Fatalf("Should not be in cache, got %+v", v)
+	}
+}
+
 func TestMQTTConnAckFirstPacket(t *testing.T) {
 	o := testMQTTDefaultOptions()
 	o.NoLog, o.Debug, o.Trace = true, false, false
diff --git a/server/norace_test.go b/server/norace_test.go
index dc0fe3a9..a9f450db 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -1,4 +1,4 @@
-// Copyright 2018-2023 The NATS Authors
+// Copyright 2018-2024 The NATS Authors
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
@@ -9314,3 +9314,79 @@ func TestNoRaceJetStreamMirrorAndSourceConsumerFailBackoff(t *testing.T) {
 	require_True(t, si != nil)
 	require_Equal(t, si.fails, 1)
 }
+
+func TestNoRaceJetStreamClusterStreamCatchupLargeInteriorDeletes(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "R3S", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	cfg := &nats.StreamConfig{
+		Name:              "TEST",
+		Subjects:          []string{"foo.*"},
+		MaxMsgsPerSubject: 100,
+		Replicas:          1,
+	}
+
+	_, err := js.AddStream(cfg)
+	require_NoError(t, err)
+
+	msg := bytes.Repeat([]byte("Z"), 2*1024)
+	// We will create lots of interior deletes on our R1 then scale up.
+	_, err = js.Publish("foo.0", msg)
+	require_NoError(t, err)
+
+	// Create 50k messages randomly from 1-100
+	for i := 0; i < 50_000; i++ {
+		subj := fmt.Sprintf("foo.%d", rand.Intn(100)+1)
+		js.PublishAsync(subj, msg)
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+	// Now create a large gap.
+	for i := 0; i < 100_000; i++ {
+		js.PublishAsync("foo.2", msg)
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+	// Do 50k random again at end.
+	for i := 0; i < 50_000; i++ {
+		subj := fmt.Sprintf("foo.%d", rand.Intn(100)+1)
+		js.PublishAsync(subj, msg)
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	si, err := js.StreamInfo("TEST")
+	require_NoError(t, err)
+
+	cfg.Replicas = 2
+	_, err = js.UpdateStream(cfg)
+	require_NoError(t, err)
+
+	// Let catchup start.
+	c.waitOnStreamLeader(globalAccountName, "TEST")
+
+	nl := c.randomNonStreamLeader(globalAccountName, "TEST")
+	require_True(t, nl != nil)
+	mset, err := nl.GlobalAccount().lookupStream("TEST")
+	require_NoError(t, err)
+
+	checkFor(t, 10*time.Second, 500*time.Millisecond, func() error {
+		state := mset.state()
+		if state.Msgs == si.State.Msgs {
+			return nil
+		}
+		return fmt.Errorf("Msgs not equal %d vs %d", state.Msgs, si.State.Msgs)
+	})
+}