Repository: nats-io/nats-server
Pull request #2947 (2022-03-23 02:14:18)

[IMPROVED] Stream recovery and Memory Utilization
Under certain circumstances a stream that was still trying to catch up at the upper layers above NRG could be elected leader staling consumers.

Improved memory usage through JetStream storage layer.

Previously we would rely more heavily on Go's garbage collector since when we loaded a block for an underlying stream we would pass references upward to avoid copies.

Now we always copy when passing back to the upper layers which allows us to not only expire our cache blocks but pool and reuse them.

The upper layers also had changes made to allow the pooling layer at that level to interoperate with the storage layer.

/cc @nats-io/core



Patch:
diff --git a/go.mod b/go.mod
index add4357a..da628ddb 100644
--- a/go.mod
+++ b/go.mod
@@ -6,7 +6,7 @@ require (
 	github.com/klauspost/compress v1.14.4
 	github.com/minio/highwayhash v1.0.2
 	github.com/nats-io/jwt/v2 v2.2.1-0.20220113022732-58e87895b296
-	github.com/nats-io/nats.go v1.13.1-0.20220308171302-2f2f6968e98d
+	github.com/nats-io/nats.go v1.13.1-0.20220318132711-e0e03e374228
 	github.com/nats-io/nkeys v0.3.0
 	github.com/nats-io/nuid v1.0.1
 	golang.org/x/crypto v0.0.0-20220112180741-5e0467b6c7ce
diff --git a/go.sum b/go.sum
index ddf89a54..ace0fd05 100644
--- a/go.sum
+++ b/go.sum
@@ -14,8 +14,8 @@ github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA
 github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
 github.com/nats-io/jwt/v2 v2.2.1-0.20220113022732-58e87895b296 h1:vU9tpM3apjYlLLeY23zRWJ9Zktr5jp+mloR942LEOpY=
 github.com/nats-io/jwt/v2 v2.2.1-0.20220113022732-58e87895b296/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k=
-github.com/nats-io/nats.go v1.13.1-0.20220308171302-2f2f6968e98d h1:zJf4l8Kp67RIZhoVeniSLZs69SHNgjLHz0aNsqPPlx8=
-github.com/nats-io/nats.go v1.13.1-0.20220308171302-2f2f6968e98d/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
+github.com/nats-io/nats.go v1.13.1-0.20220318132711-e0e03e374228 h1:czbQ9uYuV7dwLsh/0vpB+4rutgdLTYgoN5W5hf1S0eg=
+github.com/nats-io/nats.go v1.13.1-0.20220318132711-e0e03e374228/go.mod h1:BPko4oXsySz4aSWeFgOHLZs3G4Jq4ZAyE6/zMCxRT6w=
 github.com/nats-io/nkeys v0.3.0 h1:cgM5tL53EvYRU+2YLXIK0G2mJtK12Ft9oeooSZMA2G8=
 github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4=
 github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
diff --git a/server/consumer.go b/server/consumer.go
index 3c62104b..64731e3d 100644
--- a/server/consumer.go
+++ b/server/consumer.go
@@ -2149,8 +2149,9 @@ func (o *consumer) needAck(sseq uint64) bool {
 
 	// Check first if we are filtered, and if so check if this is even applicable to us.
 	if o.isFiltered() && o.mset != nil {
-		subj, _, _, _, err := o.mset.store.LoadMsg(sseq)
-		if err != nil || !o.isFilteredMatch(subj) {
+		var svp StoreMsg
+		sm, err := o.mset.store.LoadMsg(sseq, &svp)
+		if err != nil || !o.isFilteredMatch(sm.subj) {
 			o.mu.RUnlock()
 			return false
 		}
@@ -2587,9 +2588,9 @@ var (
 // Get next available message from underlying store.
 // Is partition aware and redeliver aware.
 // Lock should be held.
-func (o *consumer) getNextMsg() (subj string, hdr, msg []byte, sseq uint64, dc uint64, ts int64, err error) {
+func (o *consumer) getNextMsg() (*jsPubMsg, uint64, error) {
 	if o.mset == nil || o.mset.store == nil {
-		return _EMPTY_, nil, nil, 0, 0, 0, errBadConsumer
+		return nil, 0, errBadConsumer
 	}
 	seq, dc := o.sseq, uint64(1)
 	if o.hasSkipListPending() {
@@ -2614,8 +2615,13 @@ func (o *consumer) getNextMsg() (subj string, hdr, msg []byte, sseq uint64, dc u
 				continue
 			}
 			if seq > 0 {
-				subj, hdr, msg, ts, err = o.mset.store.LoadMsg(seq)
-				return subj, hdr, msg, seq, dc, ts, err
+				pmsg := getJSPubMsgFromPool()
+				sm, err := o.mset.store.LoadMsg(seq, &pmsg.StoreMsg)
+				if sm == nil || err != nil {
+					pmsg.returnToPool()
+					pmsg, dc = nil, 0
+				}
+				return pmsg, dc, err
 			}
 		}
 		// Fallback if all redeliveries are gone.
@@ -2626,11 +2632,12 @@ func (o *consumer) getNextMsg() (subj string, hdr, msg []byte, sseq uint64, dc u
 	if o.maxp > 0 && len(o.pending) >= o.maxp {
 		// maxp only set when ack policy != AckNone and user set MaxAckPending
 		// Stall if we have hit max pending.
-		return _EMPTY_, nil, nil, 0, 0, 0, errMaxAckPending
+		return nil, 0, errMaxAckPending
 	}
 
 	// Grab next message applicable to us.
-	subj, sseq, hdr, msg, ts, err = o.mset.store.LoadNextMsg(o.cfg.FilterSubject, o.filterWC, seq)
+	pmsg := getJSPubMsgFromPool()
+	sm, sseq, err := o.mset.store.LoadNextMsg(o.cfg.FilterSubject, o.filterWC, seq, &pmsg.StoreMsg)
 
 	if sseq >= o.sseq {
 		o.sseq = sseq + 1
@@ -2639,7 +2646,12 @@ func (o *consumer) getNextMsg() (subj string, hdr, msg []byte, sseq uint64, dc u
 		}
 	}
 
-	return subj, hdr, msg, sseq, dc, ts, err
+	if sm == nil {
+		pmsg.returnToPool()
+		return nil, 0, err
+	}
+
+	return pmsg, dc, err
 }
 
 // forceExpireFirstWaiting will force expire the first waiting.
@@ -2786,13 +2798,11 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 	// Deliver all the msgs we have now, once done or on a condition, we wait for new ones.
 	for {
 		var (
-			seq, dc     uint64
-			subj, dsubj string
-			hdr         []byte
-			msg         []byte
-			err         error
-			ts          int64
-			delay       time.Duration
+			pmsg  *jsPubMsg
+			dc    uint64
+			dsubj string
+			err   error
+			delay time.Duration
 		)
 
 		o.mu.Lock()
@@ -2812,10 +2822,11 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 			goto waitForMsgs
 		}
 
-		subj, hdr, msg, seq, dc, ts, err = o.getNextMsg()
+		// Grab our next msg.
+		pmsg, dc, err = o.getNextMsg()
 
 		// On error either wait or return.
-		if err != nil {
+		if err != nil || pmsg == nil {
 			if err == ErrStoreMsgNotFound || err == ErrStoreEOF || err == errMaxAckPending || err == errPartialCache {
 				goto waitForMsgs
 			} else {
@@ -2836,15 +2847,17 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 		} else {
 			// We will redo this one.
 			o.sseq--
+			pmsg.returnToPool()
 			goto waitForMsgs
 		}
 
 		// If we are in a replay scenario and have not caught up check if we need to delay here.
 		if o.replay && lts > 0 {
-			if delay = time.Duration(ts - lts); delay > time.Millisecond {
+			if delay = time.Duration(pmsg.ts - lts); delay > time.Millisecond {
 				o.mu.Unlock()
 				select {
 				case <-qch:
+					pmsg.returnToPool()
 					return
 				case <-time.After(delay):
 				}
@@ -2853,17 +2866,18 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 		}
 
 		// Track this regardless.
-		lts = ts
+		lts = pmsg.ts
 
 		// If we have a rate limit set make sure we check that here.
 		if o.rlimit != nil {
-			now := time.Now()
-			r := o.rlimit.ReserveN(now, len(msg)+len(hdr)+len(subj)+len(dsubj)+len(o.ackReplyT))
+			now, sm := time.Now(), &pmsg.StoreMsg
+			r := o.rlimit.ReserveN(now, len(sm.msg)+len(sm.hdr)+len(sm.subj)+len(dsubj)+len(o.ackReplyT))
 			delay := r.DelayFrom(now)
 			if delay > 0 {
 				o.mu.Unlock()
 				select {
 				case <-qch:
+					pmsg.returnToPool()
 					return
 				case <-time.After(delay):
 				}
@@ -2872,7 +2886,7 @@ func (o *consumer) loopAndGatherMsgs(qch chan struct{}) {
 		}
 
 		// Do actual delivery.
-		o.deliverMsg(dsubj, subj, hdr, msg, seq, dc, ts)
+		o.deliverMsg(dsubj, pmsg, dc)
 
 		// Reset our idle heartbeat timer if set.
 		if hb != nil {
@@ -2977,8 +2991,9 @@ func (o *consumer) adjustedPending() uint64 {
 
 // Deliver a msg to the consumer.
 // Lock should be held and o.mset validated to be non-nil.
-func (o *consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dc uint64, ts int64) {
+func (o *consumer) deliverMsg(dsubj string, pmsg *jsPubMsg, dc uint64) {
 	if o.mset == nil {
+		pmsg.returnToPool()
 		return
 	}
 	// Update pending on first attempt. This can go upside down for a short bit, that is ok.
@@ -2993,6 +3008,7 @@ func (o *consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dc uint6
 	// If headers only do not send msg payload.
 	// Add in msg size itself as header.
 	if o.cfg.HeadersOnly {
+		hdr, msg := pmsg.hdr, pmsg.msg
 		var bb bytes.Buffer
 		if len(hdr) == 0 {
 			bb.WriteString(hdrLine)
@@ -3005,12 +3021,17 @@ func (o *consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dc uint6
 		bb.WriteString(strconv.FormatInt(int64(len(msg)), 10))
 		bb.WriteString(CR_LF)
 		bb.WriteString(CR_LF)
-		hdr = bb.Bytes()
+		// Replace underlying buf which we can use directly when we send.
+		// TODO(dlc) - Probably just use directly when forming bytes.Buffer?
+		pmsg.buf = pmsg.buf[:0]
+		pmsg.buf = append(pmsg.buf, bb.Bytes()...)
+		// Replace with new header.
+		pmsg.hdr = pmsg.buf
 		// Cancel msg payload
-		msg = nil
+		pmsg.msg = nil
 	}
 
-	pmsg := newJSPubMsg(dsubj, subj, o.ackReply(seq, dseq, dc, ts, o.adjustedPending()), hdr, msg, o, seq)
+	pmsg.dsubj, pmsg.reply, pmsg.o = dsubj, o.ackReply(pmsg.seq, dseq, dc, pmsg.ts, o.adjustedPending()), o
 	psz := pmsg.size()
 
 	if o.maxpb > 0 {
@@ -3020,6 +3041,8 @@ func (o *consumer) deliverMsg(dsubj, subj string, hdr, msg []byte, seq, dc uint6
 	mset := o.mset
 	ap := o.cfg.AckPolicy
 
+	// Cant touch pmsg after this sending so capture what we need.
+	seq, ts := pmsg.seq, pmsg.ts
 	// Send message.
 	o.outq.send(pmsg)
 
diff --git a/server/filestore.go b/server/filestore.go
index 8d69b3af..3862f4cc 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -150,6 +150,7 @@ type cache struct {
 	idx  []uint32
 	lrl  uint32
 	fseq uint64
+	nra  bool
 }
 
 type msgId struct {
@@ -157,16 +158,6 @@ type msgId struct {
 	ts  int64
 }
 
-type fileStoredMsg struct {
-	subj string
-	hdr  []byte
-	msg  []byte
-	seq  uint64
-	ts   int64 // nanoseconds
-	mb   *msgBlock
-	off  int64 // offset into block file
-}
-
 const (
 	// Magic is used to identify the file store files.
 	magic = uint8(22)
@@ -230,6 +221,8 @@ const (
 	defaultOtherBlockSize = 8 * 1024 * 1024 // 8MB
 	// Default for KV based
 	defaultKVBlockSize = 8 * 1024 * 1024 // 8MB
+	// Default for minimum recycle size.
+	defaultMinRecycleBlockSize = 4 * 1024 * 1024 // 4MB
 	// max block size for now.
 	maxBlockSize = defaultStreamBlockSize
 	// Compact minimum threshold.
@@ -490,6 +483,43 @@ func (fs *fileStore) writeStreamMeta() error {
 	return nil
 }
 
+// Pools to recycle the blocks to help with memory pressure.
+var blkPoolBig sync.Pool
+var blkPoolSmall sync.Pool
+
+// Get a new msg block based on sz estimate.
+func getMsgBlockBuf(sz int) (buf []byte) {
+	var pb interface{}
+	if sz <= defaultOtherBlockSize {
+		pb, sz = blkPoolSmall.Get(), defaultOtherBlockSize
+	} else {
+		pb, sz = blkPoolBig.Get(), maxBlockSize
+	}
+	if pb != nil {
+		buf = *(pb.(*[]byte))
+	} else {
+		buf = make([]byte, sz)
+	}
+	return buf[:0]
+}
+
+// Recycle the msg block.
+func recycleMsgBlockBuf(buf []byte) {
+	if buf == nil {
+		return
+	}
+	if cap(buf) < defaultMinRecycleBlockSize {
+		return
+	}
+
+	buf = buf[:0]
+	if cap(buf) > defaultOtherBlockSize {
+		blkPoolBig.Put(&buf)
+	} else {
+		blkPoolSmall.Put(&buf)
+	}
+}
+
 const (
 	msgHdrSize     = 22
 	checksumSize   = 8
@@ -993,9 +1023,11 @@ func (fs *fileStore) expireMsgsOnRecover() {
 			break
 		}
 
+		var smv StoreMsg
+
 		// Walk messages and remove if expired.
 		for seq := mb.first.seq; seq <= mb.last.seq; seq++ {
-			sm, err := mb.cacheLookup(seq)
+			sm, err := mb.cacheLookup(seq, &smv)
 			// Process interior deleted msgs.
 			if err == errDeletedMsg {
 				// Update dmap.
@@ -1106,10 +1138,12 @@ func (fs *fileStore) GetSeqFromTime(t time.Time) uint64 {
 	lseq := mb.last.seq
 	mb.mu.RUnlock()
 
+	var smv StoreMsg
+
 	// Linear search, hence the dumb part..
 	ts := t.UnixNano()
 	for seq := fseq; seq <= lseq; seq++ {
-		sm, _, _ := mb.fetchMsg(seq)
+		sm, _, _ := mb.fetchMsg(seq, &smv)
 		if sm != nil && sm.ts >= ts {
 			return sm.seq
 		}
@@ -1118,7 +1152,7 @@ func (fs *fileStore) GetSeqFromTime(t time.Time) uint64 {
 }
 
 // Find the first matching message.
-func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64) (*fileStoredMsg, bool, error) {
+func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64, sm *StoreMsg) (*StoreMsg, bool, error) {
 	mb.mu.Lock()
 	defer mb.mu.Unlock()
 
@@ -1154,19 +1188,23 @@ func (mb *msgBlock) firstMatching(filter string, wc bool, start uint64) (*fileSt
 		}
 	}
 
+	if sm == nil {
+		sm = new(StoreMsg)
+	}
+
 	for seq := fseq; seq <= mb.last.seq; seq++ {
 		llseq := mb.llseq
-		sm, err := mb.cacheLookup(seq)
+		fsm, err := mb.cacheLookup(seq, sm)
 		if err != nil {
 			continue
 		}
-		expireOk := seq == mb.last.seq && mb.llseq == seq-1
-		if len(subs) == 1 && sm.subj == subs[0] {
-			return sm, expireOk, nil
+		expireOk := seq == mb.last.seq && mb.llseq == seq
+		if len(subs) == 1 && fsm.subj == subs[0] {
+			return fsm, expireOk, nil
 		}
 		for _, subj := range subs {
-			if sm.subj == subj {
-				return sm, expireOk, nil
+			if fsm.subj == subj {
+				return fsm, expireOk, nil
 			}
 		}
 		// If we are here we did not match, so put the llseq back.
@@ -1258,8 +1296,10 @@ func (mb *msgBlock) filteredPendingLocked(filter string, wc bool, seq uint64) (t
 		}
 		numScanIn, numScanOut := lseq-seq, seq-mb.first.seq
 
+		var smv StoreMsg
+
 		isMatch := func(seq uint64) bool {
-			if sm, _ := mb.cacheLookup(seq); sm != nil {
+			if sm, _ := mb.cacheLookup(seq, &smv); sm != nil {
 				if len(subs) == 1 && sm.subj == subs[0] {
 					return true
 				}
@@ -1361,9 +1401,10 @@ func (fs *fileStore) FilteredState(sseq uint64, subj string) SimpleState {
 		}
 	} else {
 		// Fallback to linear scan.
+		var smv StoreMsg
 		eq := compareFn(subj)
 		for seq := sseq; seq <= lseq; seq++ {
-			if sm, _ := fs.msgForSeq(seq); sm != nil && eq(sm.subj, subj) {
+			if sm, _ := fs.msgForSeq(seq, &smv); sm != nil && eq(sm.subj, subj) {
 				ss.Msgs++
 				if ss.First == 0 {
 					ss.First = seq
@@ -1449,6 +1490,7 @@ func (mb *msgBlock) setupWriteCache(buf []byte) {
 	if mb.cache != nil {
 		return
 	}
+
 	// Setup simple cache.
 	mb.cache = &cache{buf: buf}
 	// Make sure we set the proper cache offset if we have existing data.
@@ -1484,14 +1526,7 @@ func (fs *fileStore) newMsgBlockForWrite() (*msgBlock, error) {
 			lmb.closeFDsLocked()
 			if lmb.cache != nil {
 				// Reset write timestamp and see if we can expire this cache.
-				lwts, buf, llts := lmb.lwts, lmb.cache.buf, lmb.llts
-				lmb.lwts = 0
-				lmb.expireCacheLocked()
-				lmb.lwts = lwts
-				// We could check for a certain time since last load, but to be safe just reuse if no loads at all.
-				if llts == 0 && (lmb.cache == nil || lmb.cache.buf == nil) {
-					rbuf = buf[:0]
-				}
+				rbuf = lmb.tryExpireWriteCache()
 			}
 			lmb.mu.Unlock()
 		}
@@ -1908,7 +1943,8 @@ func (fs *fileStore) removeMsg(seq uint64, secure, needFSLock bool) (bool, error
 		}
 	}
 
-	sm, err := mb.cacheLookup(seq)
+	var smv StoreMsg
+	sm, err := mb.cacheLookup(seq, &smv)
 	if err != nil {
 		mb.mu.Unlock()
 		fsUnlock()
@@ -2335,7 +2371,7 @@ func (mb *msgBlock) eraseMsg(seq uint64, ri, rl int) error {
 }
 
 // Truncate this message block to the storedMsg.
-func (mb *msgBlock) truncate(sm *fileStoredMsg) (nmsgs, nbytes uint64, err error) {
+func (mb *msgBlock) truncate(sm *StoreMsg) (nmsgs, nbytes uint64, err error) {
 	// Make sure we are loaded to process messages etc.
 	if err := mb.loadMsgs(); err != nil {
 		return 0, 0, err
@@ -2432,11 +2468,12 @@ func (mb *msgBlock) selectNextFirst() {
 
 	// Need to get the timestamp.
 	// We will try the cache direct and fallback if needed.
-	sm, _ := mb.cacheLookup(seq)
+	var smv StoreMsg
+	sm, _ := mb.cacheLookup(seq, &smv)
 	if sm == nil {
 		// Slow path, need to unlock.
 		mb.mu.Unlock()
-		sm, _, _ = mb.fetchMsg(seq)
+		sm, _, _ = mb.fetchMsg(seq, &smv)
 		mb.mu.Lock()
 	}
 	if sm != nil {
@@ -2501,9 +2538,11 @@ func (mb *msgBlock) clearCache() {
 	}
 
 	if mb.cache.off == 0 {
+		recycleMsgBlockBuf(mb.cache.buf)
 		mb.cache = nil
 	} else {
 		// Clear msgs and index.
+		recycleMsgBlockBuf(mb.cache.buf)
 		mb.cache.buf = nil
 		mb.cache.idx = nil
 		mb.cache.wp = 0
@@ -2531,6 +2570,29 @@ func (mb *msgBlock) tryForceExpireCacheLocked() {
 	mb.llts = llts
 }
 
+// This is for expiration of the write cache, which will be partial with fip.
+// So we want to bypass the Pools here.
+// Lock should be held.
+func (mb *msgBlock) tryExpireWriteCache() []byte {
+	if mb.cache == nil {
+		return nil
+	}
+	lwts, buf, llts, nra := mb.lwts, mb.cache.buf, mb.llts, mb.cache.nra
+	mb.lwts, mb.cache.nra = 0, true
+	mb.expireCacheLocked()
+	mb.lwts = lwts
+	if mb.cache != nil {
+		mb.cache.nra = nra
+	}
+	// We could check for a certain time since last load, but to be safe just reuse if no loads at all.
+	if llts == 0 && (mb.cache == nil || mb.cache.buf == nil) {
+		// Clear last write time since we now are about to move on to a new lmb.
+		mb.lwts = 0
+		return buf[:0]
+	}
+	return nil
+}
+
 // Lock should be held.
 func (mb *msgBlock) expireCacheLocked() {
 	if mb.cache == nil {
@@ -2565,6 +2627,9 @@ func (mb *msgBlock) expireCacheLocked() {
 	// If we are here we will at least expire the core msg buffer.
 	// We need to capture offset in case we do a write next before a full load.
 	mb.cache.off += len(mb.cache.buf)
+	if !mb.cache.nra {
+		recycleMsgBlockBuf(mb.cache.buf)
+	}
 	mb.cache.buf = nil
 	mb.cache.wp = 0
 
@@ -2608,9 +2673,10 @@ func (fs *fileStore) cancelAgeChk() {
 func (fs *fileStore) expireMsgs() {
 	// We need to delete one by one here and can not optimize for the time being.
 	// Reason is that we need more information to adjust ack pending in consumers.
-	var sm *fileStoredMsg
+	var smv StoreMsg
+	var sm *StoreMsg
 	minAge := time.Now().UnixNano() - int64(fs.cfg.MaxAge)
-	for sm, _ = fs.msgForSeq(0); sm != nil && sm.ts <= minAge; sm, _ = fs.msgForSeq(0) {
+	for sm, _ = fs.msgForSeq(0, &smv); sm != nil && sm.ts <= minAge; sm, _ = fs.msgForSeq(0, &smv) {
 		fs.removeMsg(sm.seq, false, true)
 	}
 
@@ -3227,6 +3293,15 @@ func (mb *msgBlock) loadBlock(buf []byte) ([]byte, error) {
 		}
 	}
 
+	if buf == nil {
+		buf = getMsgBlockBuf(sz)
+		if sz > cap(buf) {
+			// We know we will make a new one so just recycle for now.
+			recycleMsgBlockBuf(buf)
+			buf = nil
+		}
+	}
+
 	if sz > cap(buf) {
 		buf = make([]byte, sz)
 	} else {
@@ -3325,7 +3400,7 @@ checkCache:
 
 // Fetch a message from this block, possibly reading in and caching the messages.
 // We assume the block was selected and is correct, so we do not do range checks.
-func (mb *msgBlock) fetchMsg(seq uint64) (*fileStoredMsg, bool, error) {
+func (mb *msgBlock) fetchMsg(seq uint64, sm *StoreMsg) (*StoreMsg, bool, error) {
 	mb.mu.Lock()
 	defer mb.mu.Unlock()
 
@@ -3334,12 +3409,12 @@ func (mb *msgBlock) fetchMsg(seq uint64) (*fileStoredMsg, bool, error) {
 			return nil, false, err
 		}
 	}
-	sm, err := mb.cacheLookup(seq)
+	fsm, err := mb.cacheLookup(seq, sm)
 	if err != nil {
 		return nil, false, err
 	}
-	expireOk := seq == mb.last.seq && mb.llseq == seq-1
-	return sm, expireOk, err
+	expireOk := seq == mb.last.seq && mb.llseq == seq
+	return fsm, expireOk, err
 }
 
 var (
@@ -3366,7 +3441,7 @@ const ebit = 1 << 63
 
 // Will do a lookup from cache.
 // Lock should be held.
-func (mb *msgBlock) cacheLookup(seq uint64) (*fileStoredMsg, error) {
+func (mb *msgBlock) cacheLookup(seq uint64, sm *StoreMsg) (*StoreMsg, error) {
 	if seq < mb.first.seq || seq > mb.last.seq {
 		return nil, ErrStoreMsgNotFound
 	}
@@ -3412,13 +3487,14 @@ func (mb *msgBlock) cacheLookup(seq uint64) (*fileStoredMsg, error) {
 	}
 
 	// Parse from the raw buffer.
-	subj, hdr, msg, mseq, ts, err := msgFromBuf(buf, hh)
-	if err != nil {
+	fsm, err := msgFromBuf(buf, sm, hh)
+	if err != nil || fsm == nil {
 		return nil, err
 	}
-	if seq != mseq {
+	if seq != fsm.seq {
+		recycleMsgBlockBuf(mb.cache.buf)
 		mb.cache.buf = nil
-		return nil, fmt.Errorf("sequence numbers for cache load did not match, %d vs %d", seq, mseq)
+		return nil, fmt.Errorf("sequence numbers for cache load did not match, %d vs %d", seq, fsm.seq)
 	}
 
 	// Clear the check bit here after we know all is good.
@@ -3426,15 +3502,16 @@ func (mb *msgBlock) cacheLookup(seq uint64) (*fileStoredMsg, error) {
 		mb.cache.idx[seq-mb.cache.fseq] = (bi | hbit)
 	}
 
-	return &fileStoredMsg{subj, hdr, msg, seq, ts, mb, int64(bi)}, nil
+	return fsm, nil
 }
 
 // Used when we are checking if discarding a message due to max msgs per subject will give us
 // enough room for a max bytes condition.
 // Lock should be already held.
 func (fs *fileStore) sizeForSeq(seq uint64) int {
+	var smv StoreMsg
 	if mb := fs.selectMsgBlock(seq); mb != nil {
-		if sm, _, _ := mb.fetchMsg(seq); sm != nil {
+		if sm, _, _ := mb.fetchMsg(seq, &smv); sm != nil {
 			return int(fileStoreMsgSize(sm.subj, sm.hdr, sm.msg))
 		}
 	}
@@ -3442,7 +3519,7 @@ func (fs *fileStore) sizeForSeq(seq uint64) int {
 }
 
 // Will return message for the given sequence number.
-func (fs *fileStore) msgForSeq(seq uint64) (*fileStoredMsg, error) {
+func (fs *fileStore) msgForSeq(seq uint64, sm *StoreMsg) (*StoreMsg, error) {
 	// TODO(dlc) - Since Store, Remove, Skip all hold the write lock on fs this will
 	// be stalled. Need another lock if want to happen in parallel.
 	fs.mu.RLock()
@@ -3466,7 +3543,7 @@ func (fs *fileStore) msgForSeq(seq uint64) (*fileStoredMsg, error) {
 		return nil, err
 	}
 
-	fsm, expireOk, err := mb.fetchMsg(seq)
+	fsm, expireOk, err := mb.fetchMsg(seq, sm)
 	if err != nil {
 		return nil, err
 	}
@@ -3481,9 +3558,9 @@ func (fs *fileStore) msgForSeq(seq uint64) (*fileStoredMsg, error) {
 }
 
 // Internal function to return msg parts from a raw buffer.
-func msgFromBuf(buf []byte, hh hash.Hash64) (string, []byte, []byte, uint64, int64, error) {
+func msgFromBuf(buf []byte, sm *StoreMsg, hh hash.Hash64) (*StoreMsg, error) {
 	if len(buf) < emptyRecordLen {
-		return _EMPTY_, nil, nil, 0, 0, errBadMsg
+		return nil, errBadMsg
 	}
 	var le = binary.LittleEndian
 
@@ -3495,7 +3572,7 @@ func msgFromBuf(buf []byte, hh hash.Hash64) (string, []byte, []byte, uint64, int
 	slen := int(le.Uint16(hdr[20:]))
 	// Simple sanity check.
 	if dlen < 0 || slen > dlen || int(rl) > len(buf) {
-		return _EMPTY_, nil, nil, 0, 0, errBadMsg
+		return nil, errBadMsg
 	}
 	data := buf[msgHdrSize : msgHdrSize+dlen]
 	// Do checksum tests here if requested.
@@ -3509,7 +3586,7 @@ func msgFromBuf(buf []byte, hh hash.Hash64) (string, []byte, []byte, uint64, int
 			hh.Write(data[slen : dlen-8])
 		}
 		if !bytes.Equal(hh.Sum(nil), data[len(data)-8:]) {
-			return _EMPTY_, nil, nil, 0, 0, errBadMsg
+			return nil, errBadMsg
 		}
 	}
 	seq := le.Uint64(hdr[4:])
@@ -3517,55 +3594,60 @@ func msgFromBuf(buf []byte, hh hash.Hash64) (string, []byte, []byte, uint64, int
 		seq = 0
 	}
 	ts := int64(le.Uint64(hdr[12:]))
-	// FIXME(dlc) - We need to not allow appends to the underlying buffer, so we will
-	// fix the capacity. This will cause a copy though in stream:internalSendLoop when
-	// we append CRLF but this was causing a race. Need to rethink more to avoid this copy.
+
+	// Create a StoreMsg if needed.
+	if sm == nil {
+		sm = new(StoreMsg)
+	} else {
+		sm.clear()
+	}
+	// To recycle the large blocks we can never pass back a reference, so need to copy for the upper
+	// layers and for us to be safe to expire, and recycle, the large msgBlocks.
 	end := dlen - 8
-	var mhdr, msg []byte
+
 	if hasHeaders {
 		hl := le.Uint32(data[slen:])
 		bi := slen + 4
 		li := bi + int(hl)
-		mhdr = data[bi:li:li]
-		msg = data[li:end:end]
+		sm.buf = append(sm.buf, data[bi:end]...)
+		li, end = li-bi, end-bi
+		sm.hdr = sm.buf[0:li:li]
+		sm.msg = sm.buf[li:end]
 	} else {
-		msg = data[slen:end:end]
+		sm.buf = append(sm.buf, data[slen:end]...)
+		sm.msg = sm.buf[0 : end-slen]
 	}
-	return string(data[:slen]), mhdr, msg, seq, ts, nil
+	sm.subj, sm.seq, sm.ts = string(data[:slen]), seq, ts
+	return sm, nil
 }
 
 // LoadMsg will lookup the message by sequence number and return it if found.
-func (fs *fileStore) LoadMsg(seq uint64) (string, []byte, []byte, int64, error) {
-	sm, err := fs.msgForSeq(seq)
-	if sm != nil {
-		return sm.subj, sm.hdr, sm.msg, sm.ts, nil
-	}
-	return _EMPTY_, nil, nil, 0, err
+func (fs *fileStore) LoadMsg(seq uint64, sm *StoreMsg) (*StoreMsg, error) {
+	return fs.msgForSeq(seq, sm)
 }
 
 // LoadLastMsg will return the last message we have that matches a given subject.
 // The subject can be a wildcard.
-func (fs *fileStore) LoadLastMsg(subject string) (subj string, seq uint64, hdr, msg []byte, ts int64, err error) {
-	var sm *fileStoredMsg
+func (fs *fileStore) LoadLastMsg(subject string, sm *StoreMsg) (*StoreMsg, error) {
 	if subject == _EMPTY_ || subject == fwcs {
-		sm, _ = fs.msgForSeq(fs.lastSeq())
+		sm, _ = fs.msgForSeq(fs.lastSeq(), sm)
 	} else if ss := fs.FilteredState(1, subject); ss.Msgs > 0 {
-		sm, _ = fs.msgForSeq(ss.Last)
+		sm, _ = fs.msgForSeq(ss.Last, sm)
+	} else {
+		sm = nil
 	}
 	if sm == nil {
-		return _EMPTY_, 0, nil, nil, 0, ErrStoreMsgNotFound
+		return nil, ErrStoreMsgNotFound
 	}
-	return sm.subj, sm.seq, sm.hdr, sm.msg, sm.ts, nil
+	return sm, nil
 }
 
-// LoadNextMsg will find the next message matching the filter subject starting at the start sequence.
-// The filter subject can be a wildcard.
-func (fs *fileStore) LoadNextMsg(filter string, wc bool, start uint64) (subj string, seq uint64, hdr, msg []byte, ts int64, err error) {
+func (fs *fileStore) LoadNextMsg(filter string, wc bool, start uint64, sm *StoreMsg) (*StoreMsg, uint64, error) {
 	fs.mu.RLock()
 	defer fs.mu.RUnlock()
 
 	if fs.closed {
-		return _EMPTY_, 0, nil, nil, 0, ErrStoreClosed
+		return nil, 0, ErrStoreClosed
 	}
 	if start < fs.state.FirstSeq {
 		start = fs.state.FirstSeq
@@ -3576,17 +3658,17 @@ func (fs *fileStore) LoadNextMsg(filter string, wc bool, start uint64) (subj str
 		if start > atomic.LoadUint64(&mb.last.seq) {
 			continue
 		}
-		if sm, expireOk, err := mb.firstMatching(filter, wc, start); err == nil {
+		if sm, expireOk, err := mb.firstMatching(filter, wc, start, sm); err == nil {
 			if expireOk && mb != fs.lmb {
 				mb.tryForceExpireCache()
 			}
-			return sm.subj, sm.seq, sm.hdr, sm.msg, sm.ts, nil
+			return sm, sm.seq, nil
 		} else if err != ErrStoreMsgNotFound {
-			return _EMPTY_, 0, nil, nil, 0, err
+			return nil, 0, err
 		}
 	}
 
-	return _EMPTY_, fs.state.LastSeq, nil, nil, 0, ErrStoreEOF
+	return nil, fs.state.LastSeq, ErrStoreEOF
 }
 
 // Type returns the type of the underlying store.
@@ -3954,6 +4036,8 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 		maxp = ss.Msgs - keep
 	}
 
+	var smv StoreMsg
+
 	fs.mu.Lock()
 	for _, mb := range fs.blks {
 		mb.mu.Lock()
@@ -3973,7 +4057,7 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 		}
 
 		for seq := f; seq <= l; seq++ {
-			if sm, _ := mb.cacheLookup(seq); sm != nil && eq(sm.subj, subject) {
+			if sm, _ := mb.cacheLookup(seq, &smv); sm != nil && eq(sm.subj, subject) {
 				rl := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
 				// Do fast in place remove.
 				// Stats
@@ -4137,9 +4221,11 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 		deleted++
 	}
 
+	var smv StoreMsg
+
 	smb.mu.Lock()
 	for mseq := smb.first.seq; mseq < seq; mseq++ {
-		sm, err := smb.cacheLookup(mseq)
+		sm, err := smb.cacheLookup(mseq, &smv)
 		if err == errDeletedMsg {
 			// Update dmap.
 			if len(smb.dmap) > 0 {
@@ -4219,7 +4305,7 @@ func (fs *fileStore) Truncate(seq uint64) error {
 		fs.mu.Unlock()
 		return ErrInvalidSequence
 	}
-	lsm, _, _ := nlmb.fetchMsg(seq)
+	lsm, _, _ := nlmb.fetchMsg(seq, nil)
 	if lsm == nil {
 		fs.mu.Unlock()
 		return ErrInvalidSequence
@@ -4395,8 +4481,9 @@ func (mb *msgBlock) removeSeqPerSubject(subj string, seq uint64) {
 	}
 
 	// TODO(dlc) - Might want to optimize this.
+	var smv StoreMsg
 	for tseq := seq + 1; tseq <= ss.Last; tseq++ {
-		if sm, _ := mb.cacheLookup(tseq); sm != nil {
+		if sm, _ := mb.cacheLookup(tseq, &smv); sm != nil {
 			if sm.subj == subj {
 				ss.First = tseq
 				return
@@ -4421,9 +4508,10 @@ func (mb *msgBlock) generatePerSubjectInfo() error {
 		mb.fss = make(map[string]*SimpleState)
 	}
 
+	var smv StoreMsg
 	fseq, lseq := mb.first.seq, mb.last.seq
 	for seq := fseq; seq <= lseq; seq++ {
-		sm, err := mb.cacheLookup(seq)
+		sm, err := mb.cacheLookup(seq, &smv)
 		if err != nil {
 			// Since we are walking by sequence we can ignore some errors that are benign to rebuilding our state.
 			if err == ErrStoreMsgNotFound || err == errDeletedMsg {
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 211cbf74..19278353 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -2535,16 +2535,13 @@ func (s *Server) jsMsgGetRequest(sub *subscription, c *client, _ *Account, subje
 		return
 	}
 
-	var subj string
-	var hdr []byte
-	var data []byte
-	var ts int64
-	seq := req.Seq
+	var svp StoreMsg
+	var sm *StoreMsg
 
 	if req.Seq > 0 {
-		subj, hdr, data, ts, err = mset.store.LoadMsg(req.Seq)
+		sm, err = mset.store.LoadMsg(req.Seq, &svp)
 	} else {
-		subj, seq, hdr, data, ts, err = mset.store.LoadLastMsg(req.LastFor)
+		sm, err = mset.store.LoadLastMsg(req.LastFor, &svp)
 	}
 	if err != nil {
 		resp.Error = NewJSNoMessageFoundError()
@@ -2552,11 +2549,11 @@ func (s *Server) jsMsgGetRequest(sub *subscription, c *client, _ *Account, subje
 		return
 	}
 	resp.Message = &StoredMsg{
-		Subject:  subj,
-		Sequence: seq,
-		Header:   hdr,
-		Data:     data,
-		Time:     time.Unix(0, ts).UTC(),
+		Subject:  sm.subj,
+		Sequence: sm.seq,
+		Header:   sm.hdr,
+		Data:     sm.msg,
+		Time:     time.Unix(0, sm.ts).UTC(),
 	}
 	s.sendAPIResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(resp))
 }
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index 6ccd1700..6015bfce 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -1442,10 +1442,9 @@ func (js *jetStream) createRaftGroup(accName string, rg *raftGroup, storage Stor
 	rg.node = n
 
 	// See if we are preferred and should start campaign immediately.
-	if n.ID() == rg.Preferred {
+	if n.ID() == rg.Preferred && n.Term() == 0 {
 		n.Campaign()
 	}
-
 	return nil
 }
 
@@ -1509,7 +1508,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment) {
 	const (
 		compactInterval = 2 * time.Minute
 		compactSizeMin  = 32 * 1024 * 1024
-		compactNumMin   = 8192
+		compactNumMin   = 65536
 	)
 
 	t := time.NewTicker(compactInterval)
@@ -1602,6 +1601,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment) {
 			} else if n.GroupLeader() != noLeader {
 				js.setStreamAssignmentRecovering(sa)
 			}
+
 			js.processStreamLeaderChange(mset, isLeader)
 		case <-t.C:
 			doSnapshot()
@@ -1807,7 +1807,8 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 
 				// We can skip if we know this is less than what we already have.
 				if lseq < last {
-					s.Debugf("Apply stream entries skipping message with sequence %d with last of %d", lseq, last)
+					s.Debugf("Apply stream entries for '%s > %s' skipping message with sequence %d with last of %d",
+						mset.account(), mset.name(), lseq, last)
 					continue
 				}
 
@@ -1831,7 +1832,8 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 					if isClusterResetErr(err) || isOutOfSpaceErr(err) {
 						return err
 					}
-					s.Debugf("Apply stream entries error processing message: %v", err)
+					s.Debugf("Apply stream entries for '%s > %s' got error processing message: %v",
+						mset.account(), mset.name(), err)
 				}
 			case deleteMsgOp:
 				md, err := decodeMsgDelete(buf[1:])
@@ -1858,8 +1860,8 @@ func (js *jetStream) applyStreamEntries(mset *stream, ce *CommittedEntry, isReco
 				}
 
 				if err != nil && !isRecovering {
-					s.Debugf("JetStream cluster failed to delete msg %d from stream %q for account %q: %v",
-						md.Seq, md.Stream, md.Client.serviceAccount(), err)
+					s.Debugf("JetStream cluster failed to delete stream msg %d from '%s > %s': %v",
+						md.Seq, md.Client.serviceAccount(), md.Stream, err)
 				}
 
 				js.mu.RLock()
@@ -5220,6 +5222,7 @@ func (mset *stream) processSnapshot(snap *streamSnapshot) error {
 	mset.clfs = snap.Failed
 	mset.store.FastState(&state)
 	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()
@@ -5240,7 +5243,9 @@ func (mset *stream) processSnapshot(snap *streamSnapshot) error {
 	}
 
 	// Pause the apply channel for our raft group while we catch up.
-	n.PauseApply()
+	if err := n.PauseApply(); err != nil {
+		return err
+	}
 	defer n.ResumeApply()
 
 	// Set our catchup state.
@@ -5308,7 +5313,8 @@ RETRY:
 	})
 	if err != nil {
 		s.Errorf("Could not subscribe to stream catchup: %v", err)
-		return err
+		err = nil
+		goto RETRY
 	}
 
 	b, _ := json.Marshal(sreq)
@@ -5368,8 +5374,11 @@ RETRY:
 		case <-qch:
 			return nil
 		case isLeader := <-lch:
-			js.processStreamLeaderChange(mset, isLeader)
-			return nil
+			if isLeader {
+				n.StepDown()
+				notActive.Reset(activityInterval)
+				goto RETRY
+			}
 		}
 	}
 }
@@ -5387,6 +5396,7 @@ func (mset *stream) processCatchupMsg(msg []byte) (uint64, error) {
 
 	mset.mu.RLock()
 	st := mset.cfg.Storage
+	ddloaded := mset.ddloaded
 	mset.mu.RUnlock()
 
 	if mset.js.limitsExceeded(st) || mset.jsa.limitsExceeded(st) {
@@ -5407,6 +5417,18 @@ func (mset *stream) processCatchupMsg(msg []byte) (uint64, error) {
 	// Update our lseq.
 	mset.setLastSeq(seq)
 
+	// Check for MsgId and if we have one here make sure to update our internal map.
+	if len(hdr) > 0 {
+		if msgId := getMsgId(hdr); msgId != _EMPTY_ {
+			if !ddloaded {
+				mset.mu.Lock()
+				mset.rebuildDedupe()
+				mset.mu.Unlock()
+			}
+			mset.storeMsgId(&ddentry{msgId, seq, ts})
+		}
+	}
+
 	return seq, nil
 }
 
@@ -5534,7 +5556,7 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 	s := mset.srv
 	defer s.grWG.Done()
 
-	const maxOutBytes = int64(1 * 1024 * 1024) // 1MB for now.
+	const maxOutBytes = int64(4 * 1024 * 1024) // 4MB for now.
 	const maxOutMsgs = int32(16384)
 	outb := int64(0)
 	outm := int32(0)
@@ -5578,8 +5600,9 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 	defer mset.clearCatchupPeer(sreq.Peer)
 
 	sendNextBatch := func() {
+		var smv StoreMsg
 		for ; seq <= last && atomic.LoadInt64(&outb) <= maxOutBytes && atomic.LoadInt32(&outm) <= maxOutMsgs; seq++ {
-			subj, hdr, msg, ts, err := mset.store.LoadMsg(seq)
+			sm, err := mset.store.LoadMsg(seq, &smv)
 			// if this is not a deleted msg, bail out.
 			if err != nil && err != ErrStoreMsgNotFound && err != errDeletedMsg {
 				// break, something changed.
@@ -5587,7 +5610,13 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 				return
 			}
 			// S2?
-			em := encodeStreamMsg(subj, _EMPTY_, hdr, msg, seq, ts)
+			var em []byte
+			if sm != nil {
+				em = encodeStreamMsg(sm.subj, _EMPTY_, sm.hdr, sm.msg, sm.seq, sm.ts)
+			} else {
+				// Skip record for deleted msg.
+				em = encodeStreamMsg(_EMPTY_, _EMPTY_, nil, nil, seq, 0)
+			}
 			// Place size in reply subject for flow control.
 			reply := fmt.Sprintf(ackReplyT, len(em))
 			atomic.AddInt64(&outb, int64(len(em)))
diff --git a/server/memstore.go b/server/memstore.go
index 9e2b16c1..e6010f7f 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -1,4 +1,4 @@
-// Copyright 2019-2021 The NATS Authors
+// Copyright 2019-2022 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
@@ -26,7 +26,7 @@ type memStore struct {
 	mu        sync.RWMutex
 	cfg       StreamConfig
 	state     StreamState
-	msgs      map[uint64]*storedMsg
+	msgs      map[uint64]*StoreMsg
 	fss       map[string]*SimpleState
 	maxp      int64
 	scb       StorageUpdateHandler
@@ -34,14 +34,6 @@ type memStore struct {
 	consumers int
 }
 
-type storedMsg struct {
-	subj string
-	hdr  []byte
-	msg  []byte
-	seq  uint64
-	ts   int64 // nanoseconds
-}
-
 func newMemStore(cfg *StreamConfig) (*memStore, error) {
 	if cfg == nil {
 		return nil, fmt.Errorf("config required")
@@ -50,7 +42,7 @@ func newMemStore(cfg *StreamConfig) (*memStore, error) {
 		return nil, fmt.Errorf("memStore requires memory storage type in config")
 	}
 	ms := &memStore{
-		msgs: make(map[uint64]*storedMsg),
+		msgs: make(map[uint64]*StoreMsg),
 		fss:  make(map[string]*SimpleState),
 		maxp: cfg.MaxMsgsPer,
 		cfg:  *cfg,
@@ -147,7 +139,15 @@ func (ms *memStore) storeRawMsg(subj string, hdr, msg []byte, seq uint64, ts int
 		hdr = copyBytes(hdr)
 	}
 
-	ms.msgs[seq] = &storedMsg{subj, hdr, msg, seq, ts}
+	// FIXME(dlc) - Could pool at this level?
+	sm := &StoreMsg{subj, nil, nil, make([]byte, 0, len(hdr)+len(msg)), seq, ts}
+	sm.buf = append(sm.buf, hdr...)
+	sm.buf = append(sm.buf, msg...)
+	if len(hdr) > 0 {
+		sm.hdr = sm.buf[:len(hdr)]
+	}
+	sm.msg = sm.buf[len(hdr):]
+	ms.msgs[seq] = sm
 	ms.state.Msgs++
 	ms.state.Bytes += memStoreMsgSize(subj, hdr, msg)
 	ms.state.LastSeq = seq
@@ -485,7 +485,7 @@ func (ms *memStore) Purge() (uint64, error) {
 	ms.state.FirstTime = time.Time{}
 	ms.state.Bytes = 0
 	ms.state.Msgs = 0
-	ms.msgs = make(map[uint64]*storedMsg)
+	ms.msgs = make(map[uint64]*StoreMsg)
 	ms.fss = make(map[string]*SimpleState)
 	ms.mu.Unlock()
 
@@ -536,7 +536,7 @@ func (ms *memStore) Compact(seq uint64) (uint64, error) {
 		ms.state.FirstSeq = seq
 		ms.state.FirstTime = time.Time{}
 		ms.state.LastSeq = seq - 1
-		ms.msgs = make(map[uint64]*storedMsg)
+		ms.msgs = make(map[uint64]*StoreMsg)
 	}
 	ms.mu.Unlock()
 
@@ -592,7 +592,7 @@ func (ms *memStore) deleteFirstMsg() bool {
 }
 
 // LoadMsg will lookup the message by sequence number and return it if found.
-func (ms *memStore) LoadMsg(seq uint64) (string, []byte, []byte, int64, error) {
+func (ms *memStore) LoadMsg(seq uint64, smp *StoreMsg) (*StoreMsg, error) {
 	ms.mu.RLock()
 	sm, ok := ms.msgs[seq]
 	last := ms.state.LastSeq
@@ -603,15 +603,20 @@ func (ms *memStore) LoadMsg(seq uint64) (string, []byte, []byte, int64, error) {
 		if seq <= last {
 			err = ErrStoreMsgNotFound
 		}
-		return _EMPTY_, nil, nil, 0, err
+		return nil, err
 	}
-	return sm.subj, sm.hdr, sm.msg, sm.ts, nil
+
+	if smp == nil {
+		smp = new(StoreMsg)
+	}
+	sm.copy(smp)
+	return smp, nil
 }
 
 // LoadLastMsg will return the last message we have that matches a given subject.
 // The subject can be a wildcard.
-func (ms *memStore) LoadLastMsg(subject string) (subj string, seq uint64, hdr, msg []byte, ts int64, err error) {
-	var sm *storedMsg
+func (ms *memStore) LoadLastMsg(subject string, smp *StoreMsg) (*StoreMsg, error) {
+	var sm *StoreMsg
 	var ok bool
 
 	ms.mu.RLock()
@@ -623,14 +628,19 @@ func (ms *memStore) LoadLastMsg(subject string) (subj string, seq uint64, hdr, m
 		sm, ok = ms.msgs[ss.Last]
 	}
 	if !ok || sm == nil {
-		return _EMPTY_, 0, nil, nil, 0, ErrStoreMsgNotFound
+		return nil, ErrStoreMsgNotFound
+	}
+
+	if smp == nil {
+		smp = new(StoreMsg)
 	}
-	return sm.subj, sm.seq, sm.hdr, sm.msg, sm.ts, nil
+	sm.copy(smp)
+	return smp, nil
 }
 
 // LoadNextMsg will find the next message matching the filter subject starting at the start sequence.
 // The filter subject can be a wildcard.
-func (ms *memStore) LoadNextMsg(filter string, wc bool, start uint64) (subj string, seq uint64, hdr, msg []byte, ts int64, err error) {
+func (ms *memStore) LoadNextMsg(filter string, wc bool, start uint64, smp *StoreMsg) (*StoreMsg, uint64, error) {
 	ms.mu.RLock()
 	defer ms.mu.RUnlock()
 
@@ -640,7 +650,7 @@ func (ms *memStore) LoadNextMsg(filter string, wc bool, start uint64) (subj stri
 
 	// If past the end no results.
 	if start > ms.state.LastSeq {
-		return _EMPTY_, ms.state.LastSeq, nil, nil, 0, ErrStoreEOF
+		return nil, ms.state.LastSeq, ErrStoreEOF
 	}
 
 	isAll := filter == _EMPTY_ || filter == fwcs
@@ -677,10 +687,14 @@ func (ms *memStore) LoadNextMsg(filter string, wc bool, start uint64) (subj stri
 
 	for nseq := fseq; nseq <= lseq; nseq++ {
 		if sm, ok := ms.msgs[nseq]; ok && (isAll || eq(sm.subj, filter)) {
-			return sm.subj, nseq, sm.hdr, sm.msg, sm.ts, nil
+			if smp == nil {
+				smp = new(StoreMsg)
+			}
+			sm.copy(smp)
+			return smp, nseq, nil
 		}
 	}
-	return _EMPTY_, ms.state.LastSeq, nil, nil, 0, ErrStoreEOF
+	return nil, ms.state.LastSeq, ErrStoreEOF
 }
 
 // RemoveMsg will remove the message from this store.
@@ -707,7 +721,7 @@ func (ms *memStore) updateFirstSeq(seq uint64) {
 		// Interior delete.
 		return
 	}
-	var nsm *storedMsg
+	var nsm *StoreMsg
 	var ok bool
 	for nseq := ms.state.FirstSeq + 1; nseq <= ms.state.LastSeq; nseq++ {
 		if nsm, ok = ms.msgs[nseq]; ok {
diff --git a/server/raft.go b/server/raft.go
index 104d21e3..96d5eeef 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -1,4 +1,4 @@
-// Copyright 2020-2021 The NATS Authors
+// Copyright 2020-2022 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
@@ -48,6 +48,7 @@ type RaftNode interface {
 	Leader() bool
 	Quorum() bool
 	Current() bool
+	Term() uint64
 	GroupLeader() string
 	HadPreviousLeader() bool
 	StepDown(preferred ...string) error
@@ -61,7 +62,7 @@ type RaftNode interface {
 	AdjustBootClusterSize(csz int) error
 	ClusterSize() int
 	ApplyQ() *ipQueue // of *CommittedEntry
-	PauseApply()
+	PauseApply() error
 	ResumeApply()
 	LeadChangeC() <-chan bool
 	QuitC() <-chan struct{}
@@ -73,7 +74,7 @@ type RaftNode interface {
 type WAL interface {
 	Type() StorageType
 	StoreMsg(subj string, hdr, msg []byte) (uint64, int64, error)
-	LoadMsg(index uint64) (subj string, hdr, msg []byte, ts int64, err error)
+	LoadMsg(index uint64, sm *StoreMsg) (*StoreMsg, error)
 	RemoveMsg(index uint64) (bool, error)
 	Compact(index uint64) (uint64, error)
 	Purge() (uint64, error)
@@ -183,8 +184,9 @@ type raft struct {
 	progress map[string]*ipQueue // of uint64
 
 	// For when we have paused our applyC.
-	paused  bool
-	hcommit uint64
+	paused    bool
+	hcommit   uint64
+	pobserver bool
 
 	// Queues and Channels
 	prop     *ipQueue // of *Entry
@@ -225,7 +227,7 @@ const (
 	minElectionTimeoutDefault = 2 * time.Second
 	maxElectionTimeoutDefault = 5 * time.Second
 	minCampaignTimeoutDefault = 100 * time.Millisecond
-	maxCampaignTimeoutDefault = 4 * minCampaignTimeoutDefault
+	maxCampaignTimeoutDefault = 8 * minCampaignTimeoutDefault
 	hbIntervalDefault         = 500 * time.Millisecond
 	lostQuorumIntervalDefault = hbIntervalDefault * 5
 )
@@ -758,20 +760,38 @@ func (n *raft) AdjustClusterSize(csz int) error {
 
 // PauseApply will allow us to pause processing of append entries onto our
 // external apply chan.
-func (n *raft) PauseApply() {
+func (n *raft) PauseApply() error {
 	n.Lock()
 	defer n.Unlock()
 
-	n.debug("Pausing apply channel")
+	if n.state == Leader {
+		return errAlreadyLeader
+	}
+	// If we are currently a candidate make sure we step down.
+	if n.state == Candidate {
+		n.stepdown.push(noLeader)
+	}
+
+	n.debug("Pausing our apply channel")
 	n.paused = true
 	n.hcommit = n.commit
+	// Also prevent us from trying to become a leader while paused and catching up.
+	n.pobserver, n.observer = n.observer, true
+	n.resetElect(48 * time.Hour)
+
+	return nil
 }
 
 func (n *raft) ResumeApply() {
 	n.Lock()
 	defer n.Unlock()
 
-	n.debug("Resuming apply channel")
+	if !n.paused {
+		return
+	}
+
+	n.debug("Resuming our apply channel")
+	n.observer, n.pobserver = n.pobserver, false
 	n.paused = false
 	// Run catchup..
 	if n.hcommit > n.commit {
@@ -783,6 +803,7 @@ func (n *raft) ResumeApply() {
 		}
 	}
 	n.hcommit = 0
+	n.resetElectionTimeout()
 }
 
 // Compact will compact our WAL.
@@ -1233,6 +1254,7 @@ func (n *raft) campaign() error {
 	}
 	n.lxfer = true
 	n.resetElect(randCampaignTimeout())
+
 	return nil
 }
 
@@ -1534,6 +1556,7 @@ func convertVoteResponse(i interface{}) *voteResponse {
 func (n *raft) runAsFollower() {
 	for {
 		elect := n.electTimer()
+
 		select {
 		case <-n.entry.ch:
 			n.processAppendEntries()
@@ -1839,6 +1862,9 @@ func (n *raft) runAsLeader() {
 	hb := time.NewTicker(hbInterval)
 	defer hb.Stop()
 
+	lq := time.NewTicker(hbInterval * 2)
+	defer lq.Stop()
+
 	for {
 		select {
 		case <-n.s.quitCh:
@@ -1876,6 +1902,7 @@ func (n *raft) runAsLeader() {
 			if n.notActive() {
 				n.sendHeartbeat()
 			}
+		case <-lq.C:
 			if n.lostQuorum() {
 				n.switchToFollower(noLeader)
 				return
@@ -1886,7 +1913,7 @@ func (n *raft) runAsLeader() {
 			if vresp == nil {
 				continue
 			}
-			if vresp.term > n.currentTerm() {
+			if vresp.term > n.Term() {
 				n.switchToFollower(noLeader)
 				return
 			}
@@ -1949,7 +1976,7 @@ func (n *raft) notActive() bool {
 }
 
 // Return our current term.
-func (n *raft) currentTerm() uint64 {
+func (n *raft) Term() uint64 {
 	n.RLock()
 	defer n.RUnlock()
 	return n.term
@@ -2134,11 +2161,12 @@ func (n *raft) catchupFollower(ar *appendEntryResponse) {
 }
 
 func (n *raft) loadEntry(index uint64) (*appendEntry, error) {
-	_, _, msg, _, err := n.wal.LoadMsg(index)
+	var smp StoreMsg
+	sm, err := n.wal.LoadMsg(index, &smp)
 	if err != nil {
 		return nil, err
 	}
-	return n.decodeAppendEntry(msg, nil, _EMPTY_)
+	return n.decodeAppendEntry(sm.msg, nil, _EMPTY_)
 }
 
 // applyCommit will update our commit index and apply the entry to the apply chan.
@@ -2744,7 +2772,7 @@ func (n *raft) processAppendEntry(ae *appendEntry, sub *subscription) {
 			case EntryLeaderTransfer:
 				if isNew {
 					maybeLeader := string(e.Data)
-					if maybeLeader == n.id {
+					if maybeLeader == n.id && !n.observer && !n.paused {
 						n.campaign()
 					}
 				}
@@ -3369,6 +3397,7 @@ func (n *raft) switchToFollower(leader string) {
 		return
 	}
 	n.debug("Switching to follower")
+
 	n.lxfer = false
 	n.updateLeader(leader)
 	n.switchState(Follower)
@@ -3380,6 +3409,11 @@ func (n *raft) switchToCandidate() {
 	if n.state == Closed {
 		return
 	}
+	// If we are catching up or are in observer mode we can not switch.
+	if n.observer || n.paused {
+		return
+	}
+
 	if n.state != Candidate {
 		n.debug("Switching to candidate")
 	} else {
diff --git a/server/store.go b/server/store.go
index f5131fe9..84dc8757 100644
--- a/server/store.go
+++ b/server/store.go
@@ -62,6 +62,16 @@ var (
 	ErrSequenceMismatch = errors.New("expected sequence does not match store")
 )
 
+// StoreMsg is the stored message format for messages that are retained by the Store layer.
+type StoreMsg struct {
+	subj string
+	hdr  []byte
+	msg  []byte
+	buf  []byte
+	seq  uint64
+	ts   int64
+}
+
 // Used to call back into the upper layers to report on changes in storage resources.
 // For the cases where its a single message we will also supply sequence number and subject.
 type StorageUpdateHandler func(msgs, bytes int64, seq uint64, subj string)
@@ -70,9 +80,9 @@ 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
-	LoadMsg(seq uint64) (subject string, hdr, msg []byte, ts int64, err error)
-	LoadLastMsg(subject string) (subj string, seq uint64, hdr, msg []byte, ts int64, err error)
-	LoadNextMsg(filter string, wc bool, start uint64) (subject string, seq uint64, hdr, msg []byte, ts int64, err 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)
 	RemoveMsg(seq uint64) (bool, error)
 	EraseMsg(seq uint64) (bool, error)
 	Purge() (uint64, error)
@@ -450,3 +460,25 @@ var errFirstSequenceMismatch = errors.New("first sequence mismatch")
 func isClusterResetErr(err error) bool {
 	return err == errLastSeqMismatch || err == ErrStoreEOF || err == errFirstSequenceMismatch
 }
+
+// Copy all fields.
+func (smo *StoreMsg) copy(sm *StoreMsg) {
+	if sm.buf != nil {
+		sm.buf = sm.buf[:0]
+	}
+	sm.buf = append(sm.buf, smo.buf...)
+	// We set cap on header in case someone wants to expand it.
+	sm.hdr, sm.msg = sm.buf[:len(smo.hdr):len(smo.hdr)], sm.buf[len(smo.hdr):]
+	sm.subj, sm.seq, sm.ts = smo.subj, smo.seq, smo.ts
+}
+
+// Clear all fields except underlying buffer but reset that if present to [:0].
+func (sm *StoreMsg) clear() {
+	if sm == nil {
+		return
+	}
+	*sm = StoreMsg{_EMPTY_, nil, nil, sm.buf, 0, 0}
+	if len(sm.buf) > 0 {
+		sm.buf = sm.buf[:0]
+	}
+}
diff --git a/server/stream.go b/server/stream.go
index 3bf0f164..e1c1f50e 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -662,6 +662,10 @@ func (mset *stream) autoTuneFileStorageBlockSize(fsCfg *FileStoreConfig) {
 // headers and msgId in them. Would need signaling from the storage layer.
 // Lock should be held.
 func (mset *stream) rebuildDedupe() {
+	if mset.ddloaded {
+		return
+	}
+
 	mset.ddloaded = true
 
 	// We have some messages. Lookup starting sequence by duplicate time window.
@@ -670,13 +674,17 @@ func (mset *stream) rebuildDedupe() {
 		return
 	}
 
+	var smv StoreMsg
 	state := mset.store.State()
 	for seq := sseq; seq <= state.LastSeq; seq++ {
-		_, hdr, _, ts, err := mset.store.LoadMsg(seq)
+		sm, err := mset.store.LoadMsg(seq, &smv)
+		if err != nil {
+			continue
+		}
 		var msgId string
-		if err == nil && len(hdr) > 0 {
-			if msgId = getMsgId(hdr); msgId != _EMPTY_ {
-				mset.storeMsgIdLocked(&ddentry{msgId, seq, ts})
+		if len(sm.hdr) > 0 {
+			if msgId = getMsgId(sm.hdr); msgId != _EMPTY_ {
+				mset.storeMsgIdLocked(&ddentry{msgId, sm.seq, sm.ts})
 			}
 		}
 		if seq == state.LastSeq {
@@ -2241,12 +2249,13 @@ func (mset *stream) setStartingSequenceForSource(sname string) {
 		return
 	}
 
+	var smv StoreMsg
 	for seq := state.LastSeq; seq >= state.FirstSeq; seq-- {
-		_, hdr, _, _, err := mset.store.LoadMsg(seq)
-		if err != nil || len(hdr) == 0 {
+		sm, err := mset.store.LoadMsg(seq, &smv)
+		if err != nil || len(sm.hdr) == 0 {
 			continue
 		}
-		ss := getHeader(JSStreamSource, hdr)
+		ss := getHeader(JSStreamSource, sm.hdr)
 		if len(ss) == 0 {
 			continue
 		}
@@ -2303,12 +2312,13 @@ func (mset *stream) startingSequenceForSources() {
 		}
 	}()
 
+	var smv StoreMsg
 	for seq := state.LastSeq; seq >= state.FirstSeq; seq-- {
-		_, hdr, _, _, err := mset.store.LoadMsg(seq)
-		if err != nil || len(hdr) == 0 {
+		sm, err := mset.store.LoadMsg(seq, &smv)
+		if err != nil || sm == nil || len(sm.hdr) == 0 {
 			continue
 		}
-		ss := getHeader(JSStreamSource, hdr)
+		ss := getHeader(JSStreamSource, sm.hdr)
 		if len(ss) == 0 {
 			continue
 		}
@@ -2880,21 +2890,26 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 		// Expected last sequence per subject.
 		if seq, exists := getExpectedLastSeqPerSubject(hdr); exists {
 			// TODO(dlc) - We could make a new store func that does this all in one.
-			_, lseq, _, _, _, err := mset.store.LoadLastMsg(subject)
+			var smv StoreMsg
+			var fseq uint64
+			sm, err := mset.store.LoadLastMsg(subject, &smv)
+			if sm != nil {
+				fseq = sm.seq
+			}
 			// If seq passed in is zero that signals we expect no msg to be present.
 			if err == ErrStoreMsgNotFound && seq == 0 {
-				lseq, err = 0, nil
+				fseq, err = 0, nil
 			}
-			if err != nil || lseq != seq {
+			if err != nil || fseq != seq {
 				mset.clfs++
 				mset.mu.Unlock()
 				if canRespond {
 					resp.PubAck = &PubAck{Stream: name}
-					resp.Error = NewJSStreamWrongLastSequenceError(lseq)
+					resp.Error = NewJSStreamWrongLastSequenceError(fseq)
 					b, _ := json.Marshal(resp)
 					outq.sendMsg(reply, b)
 				}
-				return fmt.Errorf("last sequence by subject mismatch: %d vs %d", seq, lseq)
+				return fmt.Errorf("last sequence by subject mismatch: %d vs %d", seq, fseq)
 			}
 		}
 		// Check for any rollups.
@@ -3115,37 +3130,49 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 
 // Internal message for use by jetstream subsystem.
 type jsPubMsg struct {
-	subj  string
-	dsubj string
+	dsubj string // Subject to send to, e.g. _INBOX.xxx
 	reply string
-	hdr   []byte
-	msg   []byte
-	o     *consumer
-	seq   uint64
+	StoreMsg
+	o *consumer
 }
 
 var jsPubMsgPool sync.Pool
 
-func newJSPubMsg(subj, dsubj, reply string, hdr, msg []byte, o *consumer, seq uint64) *jsPubMsg {
+func newJSPubMsg(dsubj, subj, reply string, hdr, msg []byte, o *consumer, seq uint64) *jsPubMsg {
 	var m *jsPubMsg
+	var buf []byte
 	pm := jsPubMsgPool.Get()
 	if pm != nil {
 		m = pm.(*jsPubMsg)
+		buf = m.buf[:0]
 	} else {
-		m = &jsPubMsg{}
+		m = new(jsPubMsg)
 	}
 	// When getting something from a pool it is criticical that all fields are
 	// initialized. Doing this way guarantees that if someone adds a field to
 	// the structure, the compiler will fail the build if this line is not updated.
-	(*m) = jsPubMsg{subj, dsubj, reply, hdr, msg, o, seq}
+	(*m) = jsPubMsg{dsubj, reply, StoreMsg{subj, hdr, msg, buf, seq, 0}, o}
+
 	return m
 }
 
+// Gets a jsPubMsg from the pool.
+func getJSPubMsgFromPool() *jsPubMsg {
+	pm := jsPubMsgPool.Get()
+	if pm != nil {
+		return pm.(*jsPubMsg)
+	}
+	return new(jsPubMsg)
+}
+
 func (pm *jsPubMsg) returnToPool() {
 	if pm == nil {
 		return
 	}
 	pm.subj, pm.dsubj, pm.reply, pm.hdr, pm.msg, pm.o = _EMPTY_, _EMPTY_, _EMPTY_, nil, nil, nil
+	if len(pm.buf) > 0 {
+		pm.buf = pm.buf[:0]
+	}
 	jsPubMsgPool.Put(pm)
 }
 
@@ -3153,7 +3180,7 @@ func (pm *jsPubMsg) size() int {
 	if pm == nil {
 		return 0
 	}
-	return len(pm.subj) + len(pm.reply) + len(pm.hdr) + len(pm.msg)
+	return len(pm.dsubj) + len(pm.reply) + len(pm.hdr) + len(pm.msg)
 }
 
 // Queue of *jsPubMsg for sending internal system messages.
@@ -3253,7 +3280,8 @@ func (mset *stream) internalLoop() {
 	mset.mu.RUnlock()
 
 	// Raw scratch buffer.
-	var _r [64 * 1024]byte
+	// This should be rarely used now so can be smaller.
+	var _r [1024]byte
 
 	for {
 		select {
@@ -3261,29 +3289,42 @@ func (mset *stream) internalLoop() {
 			pms := outq.pop()
 			for _, pmi := range pms {
 				pm := pmi.(*jsPubMsg)
-				c.pa.subject = []byte(pm.subj)
-				c.pa.deliver = []byte(pm.dsubj)
+				c.pa.subject = []byte(pm.dsubj)
+				c.pa.deliver = []byte(pm.subj)
 				c.pa.size = len(pm.msg) + len(pm.hdr)
 				c.pa.szb = []byte(strconv.Itoa(c.pa.size))
 				c.pa.reply = []byte(pm.reply)
 
-				msg := _r[:0]
+				// If we have an underlying buf that is the wire contents for hdr + msg, else construct on the fly.
+				var msg []byte
+				if len(pm.buf) > 0 {
+					msg = pm.buf
+				} else {
+					if len(pm.hdr) > 0 {
+						msg = pm.hdr
+						if len(pm.msg) > 0 {
+							msg = _r[:0]
+							msg = append(msg, pm.hdr...)
+							msg = append(msg, pm.msg...)
+						}
+					} else if len(pm.msg) > 0 {
+						// We own this now from a low level buffer perspective so can use directly here.
+						msg = pm.msg
+					}
+				}
+
 				if len(pm.hdr) > 0 {
 					c.pa.hdr = len(pm.hdr)
 					c.pa.hdb = []byte(strconv.Itoa(c.pa.hdr))
-					msg = append(msg, pm.hdr...)
-					msg = append(msg, pm.msg...)
 				} else {
 					c.pa.hdr = -1
 					c.pa.hdb = nil
-					if len(pm.msg) > 0 {
-						msg = append(msg, pm.msg...)
-					}
 				}
+
 				msg = append(msg, _CRLF_...)
 
 				didDeliver, _ := c.processInboundClientMsg(msg)
-				c.pa.szb = nil
+				c.pa.szb, c.pa.subject, c.pa.deliver = nil, nil, nil
 
 				// Check to see if this is a delivery for a consumer and
 				// we failed to deliver the message. If so alert the consumer.
@@ -3459,18 +3500,19 @@ func (mset *stream) stop(deleteFlag, advisory bool) error {
 }
 
 func (mset *stream) getMsg(seq uint64) (*StoredMsg, error) {
-	subj, hdr, msg, ts, err := mset.store.LoadMsg(seq)
+	var smv StoreMsg
+	sm, err := mset.store.LoadMsg(seq, &smv)
 	if err != nil {
 		return nil, err
 	}
-	sm := &StoredMsg{
-		Subject:  subj,
-		Sequence: seq,
-		Header:   hdr,
-		Data:     msg,
-		Time:     time.Unix(0, ts).UTC(),
-	}
-	return sm, nil
+	// This only used in tests directly so no need to pool etc.
+	return &StoredMsg{
+		Subject:  sm.subj,
+		Sequence: sm.seq,
+		Header:   sm.hdr,
+		Data:     sm.msg,
+		Time:     time.Unix(0, sm.ts).UTC(),
+	}, nil
 }
 
 // getConsumers will return all the current consumers for this stream.


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 1814680d..3017f848 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -1,4 +1,4 @@
-// Copyright 2019-2021 The NATS Authors
+// Copyright 2019-2022 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
@@ -67,17 +67,18 @@ func TestFileStoreBasics(t *testing.T) {
 		t.Fatalf("Expected %d bytes, got %d", expectedSize, state.Bytes)
 	}
 
-	nsubj, _, nmsg, _, err := fs.LoadMsg(2)
+	var smv StoreMsg
+	sm, err := fs.LoadMsg(2, &smv)
 	if err != nil {
 		t.Fatalf("Unexpected error looking up msg: %v", err)
 	}
-	if nsubj != subj {
-		t.Fatalf("Subjects don't match, original %q vs %q", subj, nsubj)
+	if sm.subj != subj {
+		t.Fatalf("Subjects don't match, original %q vs %q", subj, sm.subj)
 	}
-	if !bytes.Equal(nmsg, msg) {
-		t.Fatalf("Msgs don't match, original %q vs %q", msg, nmsg)
+	if !bytes.Equal(sm.msg, msg) {
+		t.Fatalf("Msgs don't match, original %q vs %q", msg, sm.msg)
 	}
-	_, _, _, _, err = fs.LoadMsg(3)
+	_, err = fs.LoadMsg(3, nil)
 	if err != nil {
 		t.Fatalf("Unexpected error looking up msg: %v", err)
 	}
@@ -120,15 +121,16 @@ func TestFileStoreMsgHeaders(t *testing.T) {
 		t.Fatalf("Wrong size for stored msg with header")
 	}
 	fs.StoreMsg(subj, hdr, msg)
-	_, shdr, smsg, _, err := fs.LoadMsg(1)
+	var smv StoreMsg
+	sm, err := fs.LoadMsg(1, &smv)
 	if err != nil {
 		t.Fatalf("Unexpected error looking up msg: %v", err)
 	}
-	if !bytes.Equal(msg, smsg) {
-		t.Fatalf("Expected same msg, got %q vs %q", smsg, msg)
+	if !bytes.Equal(msg, sm.msg) {
+		t.Fatalf("Expected same msg, got %q vs %q", sm.msg, msg)
 	}
-	if !bytes.Equal(hdr, shdr) {
-		t.Fatalf("Expected same hdr, got %q vs %q", shdr, hdr)
+	if !bytes.Equal(hdr, sm.hdr) {
+		t.Fatalf("Expected same hdr, got %q vs %q", sm.hdr, hdr)
 	}
 	if removed, _ := fs.EraseMsg(1); !removed {
 		t.Fatalf("Expected erase msg to return success")
@@ -360,11 +362,12 @@ func TestFileStoreSkipMsg(t *testing.T) {
 		t.Fatalf("Expected first to be %d and last to be %d. got first %d and last %d", numSkips+1, numSkips+5, state.FirstSeq, state.LastSeq)
 	}
 
-	subj, _, msg, _, err := fs.LoadMsg(11)
+	var smv StoreMsg
+	sm, err := fs.LoadMsg(11, &smv)
 	if err != nil {
 		t.Fatalf("Unexpected error looking up seq 11: %v", err)
 	}
-	if subj != "zzz" || string(msg) != "Hello World!" {
+	if sm.subj != "zzz" || string(sm.msg) != "Hello World!" {
 		t.Fatalf("Message did not match")
 	}
 
@@ -386,11 +389,11 @@ func TestFileStoreSkipMsg(t *testing.T) {
 	}
 	defer fs.Stop()
 
-	subj, _, msg, _, err = fs.LoadMsg(nseq)
+	sm, err = fs.LoadMsg(nseq, &smv)
 	if err != nil {
 		t.Fatalf("Unexpected error looking up seq %d: %v", nseq, err)
 	}
-	if subj != "AAA" || string(msg) != "Skip?" {
+	if sm.subj != "AAA" || string(sm.msg) != "Skip?" {
 		t.Fatalf("Message did not match")
 	}
 }
@@ -444,8 +447,9 @@ func TestFileStoreWriteExpireWrite(t *testing.T) {
 	}
 
 	// Now load them in and check.
+	var smv StoreMsg
 	for i := 1; i <= toSend*2; i++ {
-		subj, _, msg, _, err := fs.LoadMsg(uint64(i))
+		sm, err := fs.LoadMsg(uint64(i), &smv)
 		if err != nil {
 			t.Fatalf("Unexpected error looking up seq %d: %v", i, err)
 		}
@@ -453,7 +457,7 @@ func TestFileStoreWriteExpireWrite(t *testing.T) {
 		if i > toSend {
 			str = "Hello World! - 22"
 		}
-		if subj != "zzz" || string(msg) != str {
+		if sm.subj != "zzz" || string(sm.msg) != str {
 			t.Fatalf("Message did not match")
 		}
 	}
@@ -491,7 +495,7 @@ func TestFileStoreMsgLimit(t *testing.T) {
 		t.Fatalf("Expected the first sequence to be 2 now, but got %d", state.FirstSeq)
 	}
 	// Make sure we can not lookup seq 1.
-	if _, _, _, _, err := fs.LoadMsg(1); err == nil {
+	if _, err := fs.LoadMsg(1, nil); err == nil {
 		t.Fatalf("Expected error looking up seq 1 but got none")
 	}
 }
@@ -646,16 +650,17 @@ func TestFileStoreTimeStamps(t *testing.T) {
 		time.Sleep(5 * time.Millisecond)
 		fs.StoreMsg(subj, nil, msg)
 	}
+	var smv StoreMsg
 	for seq := uint64(1); seq <= 10; seq++ {
-		_, _, _, ts, err := fs.LoadMsg(seq)
+		sm, err := fs.LoadMsg(seq, &smv)
 		if err != nil {
 			t.Fatalf("Unexpected error looking up msg [%d]: %v", seq, err)
 		}
 		// These should be different
-		if ts <= last {
-			t.Fatalf("Expected different timestamps, got last %v vs %v", last, ts)
+		if sm.ts <= last {
+			t.Fatalf("Expected different timestamps, got last %v vs %v", last, sm.ts)
 		}
-		last = ts
+		last = sm.ts
 	}
 }
 
@@ -1078,11 +1083,12 @@ func TestFileStoreRemoveOutOfOrderRecovery(t *testing.T) {
 		t.Fatalf("Expected %d msgs, got %d", toStore/2, state.Msgs)
 	}
 
-	if _, _, _, _, err := fs.LoadMsg(1); err != nil {
+	var smv StoreMsg
+	if _, err := fs.LoadMsg(1, &smv); err != nil {
 		t.Fatalf("Expected to retrieve seq 1")
 	}
 	for i := 2; i <= toStore; i += 2 {
-		if _, _, _, _, err := fs.LoadMsg(uint64(i)); err == nil {
+		if _, err := fs.LoadMsg(uint64(i), &smv); err == nil {
 			t.Fatalf("Expected error looking up seq %d that should be deleted", i)
 		}
 	}
@@ -1101,11 +1107,11 @@ func TestFileStoreRemoveOutOfOrderRecovery(t *testing.T) {
 		t.Fatalf("Expected recovered states to be the same, got %+v vs %+v\n", state, state2)
 	}
 
-	if _, _, _, _, err := fs.LoadMsg(1); err != nil {
+	if _, err := fs.LoadMsg(1, &smv); err != nil {
 		t.Fatalf("Expected to retrieve seq 1")
 	}
 	for i := 2; i <= toStore; i += 2 {
-		if _, _, _, _, err := fs.LoadMsg(uint64(i)); err == nil {
+		if _, err := fs.LoadMsg(uint64(i), nil); err == nil {
 			t.Fatalf("Expected error looking up seq %d that should be deleted", i)
 		}
 	}
@@ -1233,19 +1239,18 @@ func TestFileStoreEraseMsg(t *testing.T) {
 	subj, msg := "foo", []byte("Hello World")
 	fs.StoreMsg(subj, nil, msg)
 	fs.StoreMsg(subj, nil, msg) // To keep block from being deleted.
-	_, _, smsg, _, err := fs.LoadMsg(1)
+	var smv StoreMsg
+	sm, err := fs.LoadMsg(1, &smv)
 	if err != nil {
 		t.Fatalf("Unexpected error looking up msg: %v", err)
 	}
-	if !bytes.Equal(msg, smsg) {
-		t.Fatalf("Expected same msg, got %q vs %q", smsg, msg)
+	if !bytes.Equal(msg, sm.msg) {
+		t.Fatalf("Expected same msg, got %q vs %q", sm.msg, msg)
 	}
-	// Hold for offset check later.
-	sm, _ := fs.msgForSeq(1)
 	if removed, _ := fs.EraseMsg(1); !removed {
 		t.Fatalf("Expected erase msg to return success")
 	}
-	if sm2, _ := fs.msgForSeq(1); sm2 != nil {
+	if sm2, _ := fs.msgForSeq(1, nil); sm2 != nil {
 		t.Fatalf("Expected msg to be erased")
 	}
 	fs.checkAndFlushAllBlocks()
@@ -1259,21 +1264,22 @@ func TestFileStoreEraseMsg(t *testing.T) {
 	}
 	defer fp.Close()
 
-	fp.ReadAt(buf, sm.off)
-	nsubj, _, nmsg, seq, ts, err := msgFromBuf(buf, nil)
+	fp.ReadAt(buf, 0)
+	//	nsubj, _, nmsg, seq, ts, err := msgFromBuf(buf, nil)
+	sm, err = msgFromBuf(buf, nil, nil)
 	if err != nil {
 		t.Fatalf("error reading message from block: %v", err)
 	}
-	if nsubj == subj {
+	if sm.subj == subj {
 		t.Fatalf("Expected the subjects to be different")
 	}
-	if seq != 0 && seq&ebit == 0 {
-		t.Fatalf("Expected seq to be 0, marking as deleted, got %d", seq)
+	if sm.seq != 0 && sm.seq&ebit == 0 {
+		t.Fatalf("Expected seq to be 0, marking as deleted, got %d", sm.seq)
 	}
-	if ts != 0 {
-		t.Fatalf("Expected timestamp to be 0, got %d", ts)
+	if sm.ts != 0 {
+		t.Fatalf("Expected timestamp to be 0, got %d", sm.ts)
 	}
-	if bytes.Equal(nmsg, msg) {
+	if bytes.Equal(sm.msg, msg) {
 		t.Fatalf("Expected message body to be randomized")
 	}
 }
@@ -1326,7 +1332,7 @@ func TestFileStoreEraseAndNoIndexRecovery(t *testing.T) {
 	}
 
 	for i := 2; i <= toStore; i += 2 {
-		if _, _, _, _, err := fs.LoadMsg(uint64(i)); err == nil {
+		if _, err := fs.LoadMsg(uint64(i), nil); err == nil {
 			t.Fatalf("Expected error looking up seq %d that should be erased", i)
 		}
 	}
@@ -1444,7 +1450,7 @@ func TestFileStoreWriteAndReadSameBlock(t *testing.T) {
 
 	for i := uint64(1); i <= 10; i++ {
 		fs.StoreMsg(subj, nil, msg)
-		if _, _, _, _, err := fs.LoadMsg(i); err != nil {
+		if _, err := fs.LoadMsg(i, nil); err != nil {
 			t.Fatalf("Error loading %d: %v", i, err)
 		}
 	}
@@ -1483,8 +1489,9 @@ func TestFileStoreAndRetrieveMultiBlock(t *testing.T) {
 	}
 	defer fs.Stop()
 
+	var smv StoreMsg
 	for i := uint64(1); i <= 20; i++ {
-		if _, _, _, _, err := fs.LoadMsg(i); err != nil {
+		if _, err := fs.LoadMsg(i, &smv); err != nil {
 			t.Fatalf("Error loading %d: %v", i, err)
 		}
 	}
@@ -1592,7 +1599,7 @@ func TestFileStoreReadCache(t *testing.T) {
 		return nil
 	})
 
-	fs.LoadMsg(1)
+	fs.LoadMsg(1, nil)
 	if csz := fs.cacheSize(); csz != totalBytes {
 		t.Fatalf("Expected all messages to be cached, got %d vs %d", csz, totalBytes)
 	}
@@ -1607,14 +1614,14 @@ func TestFileStoreReadCache(t *testing.T) {
 		t.Fatalf("Expected only 1 cache load, got %d", cls)
 	}
 	// Now make sure we do not reload cache if there is activity.
-	fs.LoadMsg(1)
+	fs.LoadMsg(1, nil)
 	timeout := time.Now().Add(250 * time.Millisecond)
 	for time.Now().Before(timeout) {
 		if cls := fs.cacheLoads(); cls != 2 {
 			t.Fatalf("cache loads not 2, got %d", cls)
 		}
 		time.Sleep(5 * time.Millisecond)
-		fs.LoadMsg(1) // register activity.
+		fs.LoadMsg(1, nil) // register activity.
 	}
 }
 
@@ -1638,7 +1645,7 @@ func TestFileStorePartialCacheExpiration(t *testing.T) {
 
 	// Again wait for cache to expire.
 	time.Sleep(2 * cexp)
-	if _, _, _, _, err := fs.LoadMsg(1); err != nil {
+	if _, err := fs.LoadMsg(1, nil); err != nil {
 		t.Fatalf("Error loading message 1: %v", err)
 	}
 }
@@ -1685,7 +1692,7 @@ func TestFileStorePartialIndexes(t *testing.T) {
 	}
 	// If we now load in a message in second half if we do not
 	// detect idx is a partial correctly this will panic.
-	if _, _, _, _, err := fs.LoadMsg(8); err != nil {
+	if _, err := fs.LoadMsg(8, nil); err != nil {
 		t.Fatalf("Error loading %d: %v", 1, err)
 	}
 }
@@ -2091,11 +2098,11 @@ func TestFileStoreWriteFailures(t *testing.T) {
 	if state.Msgs != lseq-1 {
 		t.Fatalf("Expected total msgs to be %d, got %d\n", lseq-1, state.Msgs)
 	}
-	if _, _, _, _, err := fs.LoadMsg(lseq); err == nil {
+	if _, err := fs.LoadMsg(lseq, nil); err == nil {
 		t.Fatalf("Expected error loading seq that failed, got none")
 	}
 	// Loading should still work.
-	if _, _, _, _, err := fs.LoadMsg(1); err != nil {
+	if _, err := fs.LoadMsg(1, nil); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
@@ -2200,9 +2207,10 @@ func TestFileStorePerf(t *testing.T) {
 		friendlyBytes(int64(toStore*storedMsgSize)),
 	)
 
+	var smv StoreMsg
 	start = time.Now()
 	for i := uint64(1); i <= toStore; i++ {
-		if _, _, _, _, err := fs.LoadMsg(i); err != nil {
+		if _, err := fs.LoadMsg(i, &smv); err != nil {
 			t.Fatalf("Error loading %d: %v", i, err)
 		}
 	}
@@ -2221,7 +2229,7 @@ func TestFileStorePerf(t *testing.T) {
 
 	start = time.Now()
 	for i := uint64(1); i <= toStore; i++ {
-		if _, _, _, _, err := fs.LoadMsg(i); err != nil {
+		if _, err := fs.LoadMsg(i, &smv); err != nil {
 			t.Fatalf("Error loading %d: %v", i, err)
 		}
 	}
@@ -2321,11 +2329,11 @@ func TestFileStoreReadBackMsgPerf(t *testing.T) {
 	// Pick something towards end of the block.
 	index := defaultStreamBlockSize/storedMsgSize - 22
 	start = time.Now()
-	fs.LoadMsg(index)
+	fs.LoadMsg(index, nil)
 	fmt.Printf("Time to load first msg [%d] = %v\n", index, time.Since(start))
 
 	start = time.Now()
-	fs.LoadMsg(index + 2)
+	fs.LoadMsg(index+2, nil)
 	fmt.Printf("Time to load second msg [%d] = %v\n", index+2, time.Since(start))
 }
 
@@ -2890,7 +2898,7 @@ func TestFileStoreStreamDeleteCacheBug(t *testing.T) {
 		t.Fatalf("Got an error on remove of %d: %v", 1, err)
 	}
 	time.Sleep(100 * time.Millisecond)
-	if _, _, _, _, err := fs.LoadMsg(2); err != nil {
+	if _, err := fs.LoadMsg(2, nil); err != nil {
 		t.Fatalf("Unexpected error looking up msg: %v", err)
 	}
 }
@@ -3398,7 +3406,7 @@ func TestFileStoreSparseCompactionWithInteriorDeletes(t *testing.T) {
 		}
 	}
 
-	_, _, _, _, err = fs.LoadMsg(900)
+	_, err = fs.LoadMsg(900, nil)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -3410,8 +3418,7 @@ func TestFileStoreSparseCompactionWithInteriorDeletes(t *testing.T) {
 	lmb.compact()
 	fs.mu.RUnlock()
 
-	_, _, _, _, err = fs.LoadMsg(900)
-	if err != nil {
+	if _, err = fs.LoadMsg(900, nil); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 }
@@ -3532,9 +3539,10 @@ func TestFileStoreFetchPerf(t *testing.T) {
 	}
 
 	// Time how long it takes us to load all messages.
+	var smv StoreMsg
 	now := time.Now()
 	for i := 0; i < n; i++ {
-		_, _, _, _, err := fs.LoadMsg(uint64(i))
+		_, err := fs.LoadMsg(uint64(i), &smv)
 		if err != nil {
 			t.Fatalf("Unexpected error looking up seq %d: %v", i, err)
 		}
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index e83efabb..40e07a49 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -2594,13 +2594,13 @@ func TestJetStreamClusterUserSnapshotAndRestore(t *testing.T) {
 	}
 
 	// Create consumer with no state.
-	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{Durable: "rip", AckPolicy: nats.AckExplicitPolicy, AckWait: time.Second})
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{Durable: "rip", AckPolicy: nats.AckExplicitPolicy})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 
 	// Create another consumer as well and give it a non-simplistic state.
-	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{Durable: "dlc", AckPolicy: nats.AckExplicitPolicy, AckWait: time.Second})
+	_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{Durable: "dlc", AckPolicy: nats.AckExplicitPolicy, AckWait: 5 * time.Second})
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -2619,8 +2619,6 @@ func TestJetStreamClusterUserSnapshotAndRestore(t *testing.T) {
 			m.AckSync()
 		}
 	}
-	nc.Flush()
-	time.Sleep(500 * time.Millisecond)
 
 	// Snapshot consumer info.
 	ci, err := jsub.ConsumerInfo()
@@ -7067,13 +7065,14 @@ func TestJetStreamClusterDomainsAndSameNameSources(t *testing.T) {
 	}
 
 	// Now make sure we have 2 msgs in our sourced stream.
-	si, err := js.StreamInfo("S")
-	if err != nil {
-		t.Fatalf("Unexpected error: %v", err)
-	}
-	if si.State.Msgs != 2 {
-		t.Fatalf("Expected 2 msgs, got %d", si.State.Msgs)
-	}
+	checkFor(t, 2*time.Second, 50*time.Millisecond, func() error {
+		si, err := js.StreamInfo("S")
+		require_NoError(t, err)
+		if si.State.Msgs != 2 {
+			return fmt.Errorf("Expected 2 msgs, got %d", si.State.Msgs)
+		}
+		return nil
+	})
 
 	// Make sure we can see our external information.
 	// This not in the Go client yet so manual for now.
@@ -10301,7 +10300,7 @@ func TestJetStreamClusterNAKBackoffs(t *testing.T) {
 	if elapsed < time.Second {
 		t.Fatalf("Took too short to redeliver, expected ~1s but got %v", elapsed)
 	}
-	if elapsed > 2*time.Second {
+	if elapsed > 3*time.Second {
 		t.Fatalf("Took too long to redeliver, expected ~1s but got %v", elapsed)
 	}
 
@@ -11177,7 +11176,7 @@ func TestJetStreamRemovedPeersAndStreamsListAndDelete(t *testing.T) {
 	// Check behavior of stream info as well. We want it to return the stream is offline and not just timeout.
 	_, err = js.StreamInfo("GONE")
 	// FIXME(dlc) - Go client not putting nats: prefix on for stream but does for consumer.
-	require_Error(t, err, NewJSStreamOfflineError())
+	require_Error(t, err, NewJSStreamOfflineError(), errors.New("nats: stream is offline"))
 
 	// Same for Consumer
 	start = time.Now()
@@ -11315,6 +11314,73 @@ func TestJetStreamDuplicateRoutesDisruptJetStreamMetaGroup(t *testing.T) {
 	c.waitOnClusterReady()
 }
 
+func TestJetStreamDuplicateMsgIdsOnCatchupAndLeaderTakeover(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "JSC", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	// Shutdown a non-leader.
+	nc.Close()
+	sr := c.randomNonStreamLeader("$G", "TEST")
+	sr.Shutdown()
+
+	nc, js = jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	m := nats.NewMsg("TEST")
+	m.Data = []byte("OK")
+
+	n := 10
+	for i := 0; i < n; i++ {
+		m.Header.Set(JSMsgId, strconv.Itoa(i))
+		_, err := js.PublishMsg(m)
+		require_NoError(t, err)
+	}
+
+	m.Header.Set(JSMsgId, "8")
+	pa, err := js.PublishMsg(m)
+	require_NoError(t, err)
+	if !pa.Duplicate {
+		t.Fatalf("Expected msg to be a duplicate")
+	}
+
+	// Now force a snapshot, want to test catchup above RAFT layer.
+	sl := c.streamLeader("$G", "TEST")
+	mset, err := sl.GlobalAccount().lookupStream("TEST")
+	require_NoError(t, err)
+	if node := mset.raftNode(); node == nil {
+		t.Fatalf("Could not get stream group name")
+	} else if err := node.InstallSnapshot(mset.stateSnapshot()); err != nil {
+		t.Fatalf("Error installing snapshot: %v", err)
+	}
+
+	// Now restart
+	sr = c.restartServer(sr)
+	c.waitOnStreamCurrent(sr, "$G", "TEST")
+
+	// Now make them the leader.
+	for sr != c.streamLeader("$G", "TEST") {
+		_, err = nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, "TEST"), nil, time.Second)
+		require_NoError(t, err)
+		c.waitOnStreamLeader("$G", "TEST")
+	}
+
+	// Make sure this gets rejected.
+	pa, err = js.PublishMsg(m)
+	require_NoError(t, err)
+	if !pa.Duplicate {
+		t.Fatalf("Expected msg to be a duplicate")
+	}
+}
+
 // Support functions
 
 // Used to setup superclusters for tests.
diff --git a/server/jetstream_test.go b/server/jetstream_test.go
index 6196b1dd..172864be 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -7339,10 +7339,9 @@ gateway {
 				if s := getStreams(js); len(s) != 1 {
 					st.Fatalf("unexpected stream count, got=%d, want=1", len(s))
 				}
-
-				err = js.DeleteStream(streamName)
-				require_NoError(st, err)
 			}
+			// Delete regardless.
+			js.DeleteStream(streamName)
 		})
 	}
 }
diff --git a/server/memstore_test.go b/server/memstore_test.go
index 5e608b48..413d39a7 100644
--- a/server/memstore_test.go
+++ b/server/memstore_test.go
@@ -1,4 +1,4 @@
-// Copyright 2019-2021 The NATS Authors
+// Copyright 2019-2022 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
@@ -45,15 +45,15 @@ func TestMemStoreBasics(t *testing.T) {
 	if state.Bytes != expectedSize {
 		t.Fatalf("Expected %d bytes, got %d", expectedSize, state.Bytes)
 	}
-	nsubj, _, nmsg, _, err := ms.LoadMsg(1)
+	sm, err := ms.LoadMsg(1, nil)
 	if err != nil {
 		t.Fatalf("Unexpected error looking up msg: %v", err)
 	}
-	if nsubj != subj {
-		t.Fatalf("Subjects don't match, original %q vs %q", subj, nsubj)
+	if sm.subj != subj {
+		t.Fatalf("Subjects don't match, original %q vs %q", subj, sm.subj)
 	}
-	if !bytes.Equal(nmsg, msg) {
-		t.Fatalf("Msgs don't match, original %q vs %q", msg, nmsg)
+	if !bytes.Equal(sm.msg, msg) {
+		t.Fatalf("Msgs don't match, original %q vs %q", msg, sm.msg)
 	}
 }
 
@@ -84,7 +84,7 @@ func TestMemStoreMsgLimit(t *testing.T) {
 		t.Fatalf("Expected the first sequence to be 2 now, but got %d", state.FirstSeq)
 	}
 	// Make sure we can not lookup seq 1.
-	if _, _, _, _, err := ms.LoadMsg(1); err == nil {
+	if _, err := ms.LoadMsg(1, nil); err == nil {
 		t.Fatalf("Expected error looking up seq 1 but got none")
 	}
 }
@@ -186,16 +186,17 @@ func TestMemStoreTimeStamps(t *testing.T) {
 		time.Sleep(5 * time.Microsecond)
 		ms.StoreMsg(subj, nil, msg)
 	}
+	var smv StoreMsg
 	for seq := uint64(1); seq <= 10; seq++ {
-		_, _, _, ts, err := ms.LoadMsg(seq)
+		sm, err := ms.LoadMsg(seq, &smv)
 		if err != nil {
 			t.Fatalf("Unexpected error looking up msg: %v", err)
 		}
 		// These should be different
-		if ts <= last {
-			t.Fatalf("Expected different timestamps, got %v", ts)
+		if sm.ts <= last {
+			t.Fatalf("Expected different timestamps, got %v", sm.ts)
 		}
-		last = ts
+		last = sm.ts
 	}
 }
 
@@ -265,12 +266,12 @@ func TestMemStoreEraseMsg(t *testing.T) {
 	}
 	subj, msg := "foo", []byte("Hello World")
 	ms.StoreMsg(subj, nil, msg)
-	_, _, smsg, _, err := ms.LoadMsg(1)
+	sm, err := ms.LoadMsg(1, nil)
 	if err != nil {
 		t.Fatalf("Unexpected error looking up msg: %v", err)
 	}
-	if !bytes.Equal(msg, smsg) {
-		t.Fatalf("Expected same msg, got %q vs %q", smsg, msg)
+	if !bytes.Equal(msg, sm.msg) {
+		t.Fatalf("Expected same msg, got %q vs %q", sm.msg, msg)
 	}
 	if removed, _ := ms.EraseMsg(1); !removed {
 		t.Fatalf("Expected erase msg to return success")
@@ -287,15 +288,15 @@ func TestMemStoreMsgHeaders(t *testing.T) {
 		t.Fatalf("Wrong size for stored msg with header")
 	}
 	ms.StoreMsg(subj, hdr, msg)
-	_, shdr, smsg, _, err := ms.LoadMsg(1)
+	sm, err := ms.LoadMsg(1, nil)
 	if err != nil {
 		t.Fatalf("Unexpected error looking up msg: %v", err)
 	}
-	if !bytes.Equal(msg, smsg) {
-		t.Fatalf("Expected same msg, got %q vs %q", smsg, msg)
+	if !bytes.Equal(msg, sm.msg) {
+		t.Fatalf("Expected same msg, got %q vs %q", sm.msg, msg)
 	}
-	if !bytes.Equal(hdr, shdr) {
-		t.Fatalf("Expected same hdr, got %q vs %q", shdr, hdr)
+	if !bytes.Equal(hdr, sm.hdr) {
+		t.Fatalf("Expected same hdr, got %q vs %q", sm.hdr, hdr)
 	}
 	if removed, _ := ms.EraseMsg(1); !removed {
 		t.Fatalf("Expected erase msg to return success")
diff --git a/server/norace_test.go b/server/norace_test.go
index 8a46931f..eff74980 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -1,4 +1,4 @@
-// Copyright 2018-2021 The NATS Authors
+// Copyright 2018-2022 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
@@ -3383,11 +3383,11 @@ func TestNoRaceJetStreamClusterCorruptWAL(t *testing.T) {
 		t.Fatalf("Unexpected error: %v", err)
 	}
 	state := fs.State()
-	_, _, msg, _, err := fs.LoadMsg(state.LastSeq)
+	sm, err := fs.LoadMsg(state.LastSeq, nil)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-	ae, err := node.decodeAppendEntry(msg, nil, _EMPTY_)
+	ae, err := node.decodeAppendEntry(sm.msg, nil, _EMPTY_)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -4356,3 +4356,171 @@ func TestNoRaceFileStoreKeyFileCleanup(t *testing.T) {
 		t.Fatalf("Expected to find only 1 key file, found %d", len(kms))
 	}
 }
+
+func TestNoRaceMsgIdPerfDuringCatchup(t *testing.T) {
+	// Uncomment to run. Needs to be on a bigger machine. Do not want as part of Travis tests atm.
+	skip(t)
+
+	c := createJetStreamClusterExplicit(t, "JSC", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.serverByName("S-1"))
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	// This will be the one we restart.
+	sl := c.streamLeader("$G", "TEST")
+	// Now move leader.
+	_, err = nc.Request(fmt.Sprintf(JSApiStreamLeaderStepDownT, "TEST"), nil, time.Second)
+	require_NoError(t, err)
+	c.waitOnStreamLeader("$G", "TEST")
+
+	// Connect to new leader.
+	nc, _ = jsClientConnect(t, c.streamLeader("$G", "TEST"))
+	defer nc.Close()
+
+	js, err = nc.JetStream(nats.PublishAsyncMaxPending(1024))
+	require_NoError(t, err)
+
+	n, ss, sr := 1_000_000, 250_000, 800_000
+	m := nats.NewMsg("TEST")
+	m.Data = []byte(strings.Repeat("Z", 2048))
+
+	// Target rate 10k msgs/sec
+	start := time.Now()
+
+	for i := 0; i < n; i++ {
+		m.Header.Set(JSMsgId, strconv.Itoa(i))
+		_, err := js.PublishMsgAsync(m)
+		require_NoError(t, err)
+		//time.Sleep(42 * time.Microsecond)
+		if i == ss {
+			fmt.Printf("SD")
+			sl.Shutdown()
+		} else if i == sr {
+			nc.Flush()
+			select {
+			case <-js.PublishAsyncComplete():
+			case <-time.After(10 * time.Second):
+			}
+			fmt.Printf("RS")
+			sl = c.restartServer(sl)
+		}
+		if i%10_000 == 0 {
+			fmt.Print("#")
+		}
+	}
+	fmt.Println()
+
+	// Wait to receive all messages.
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(20 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	tt := time.Since(start)
+	si, err := js.StreamInfo("TEST")
+	require_NoError(t, err)
+
+	fmt.Printf("Took %v to send %d msgs\n", tt, n)
+	fmt.Printf("%.0f msgs/s\n", float64(n)/tt.Seconds())
+	fmt.Printf("%.0f mb/s\n\n", float64(si.State.Bytes/(1024*1024))/tt.Seconds())
+
+	c.waitOnStreamCurrent(sl, "$G", "TEST")
+	for _, s := range c.servers {
+		mset, _ := s.GlobalAccount().lookupStream("TEST")
+		if state := mset.store.State(); state.Msgs != uint64(n) {
+			t.Fatalf("Expected server %v to have correct number of msgs %d but got %d", s, n, state.Msgs)
+		}
+	}
+}
+
+func TestNoRaceRebuildDeDupeAndMemoryPerf(t *testing.T) {
+	skip(t)
+
+	s := RunBasicJetStreamServer()
+	if config := s.JetStreamConfig(); config != nil {
+		defer removeDir(t, config.StoreDir)
+	}
+	defer s.Shutdown()
+
+	nc, js := jsClientConnect(t, s)
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{Name: "DD"})
+	require_NoError(t, err)
+
+	m := nats.NewMsg("DD")
+	m.Data = []byte(strings.Repeat("Z", 2048))
+
+	start := time.Now()
+
+	n := 1_000_000
+	for i := 0; i < n; i++ {
+		m.Header.Set(JSMsgId, strconv.Itoa(i))
+		_, err := js.PublishMsgAsync(m)
+		require_NoError(t, err)
+	}
+
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(20 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	tt := time.Since(start)
+	si, err := js.StreamInfo("DD")
+	require_NoError(t, err)
+
+	fmt.Printf("Took %v to send %d msgs\n", tt, n)
+	fmt.Printf("%.0f msgs/s\n", float64(n)/tt.Seconds())
+	fmt.Printf("%.0f mb/s\n\n", float64(si.State.Bytes/(1024*1024))/tt.Seconds())
+
+	v, _ := s.Varz(nil)
+	fmt.Printf("Memory AFTER SEND: %v\n", friendlyBytes(v.Mem))
+
+	mset, err := s.GlobalAccount().lookupStream("DD")
+	require_NoError(t, err)
+
+	mset.mu.Lock()
+	mset.ddloaded = false
+	start = time.Now()
+	mset.rebuildDedupe()
+	fmt.Printf("TOOK %v to rebuild dd\n", time.Since(start))
+	mset.mu.Unlock()
+
+	v, _ = s.Varz(nil)
+	fmt.Printf("Memory: %v\n", friendlyBytes(v.Mem))
+
+	// Now do an ephemeral consumer and whip through every message. Doing same calculations.
+	start = time.Now()
+	received, done := 0, make(chan bool)
+	sub, err := js.Subscribe("DD", func(m *nats.Msg) {
+		received++
+		if received >= n {
+			done <- true
+		}
+	}, nats.OrderedConsumer())
+	require_NoError(t, err)
+
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		if s.NumSlowConsumers() > 0 {
+			t.Fatalf("Did not receive all large messages due to slow consumer status: %d of %d", received, n)
+		}
+		t.Fatalf("Failed to receive all large messages: %d of %d\n", received, n)
+	}
+
+	fmt.Printf("TOOK %v to receive all %d msgs\n", time.Since(start), n)
+	sub.Unsubscribe()
+
+	v, _ = s.Varz(nil)
+	fmt.Printf("Memory: %v\n", friendlyBytes(v.Mem))
+}