Repository: nats-io/nats-server
Pull request #4929 (2024-01-09 04:58:46)

[IMPROVED] Optimize catchups for replicas and mirrors with large interior deletes.
Under scenarios where interior deletes were orders of magnitude larger than the msgs themselves, the system would need to transfer all interior deletes as skip msgs. This change adopts the ability to send DeleteRanges on streamSyncRequests, and for R1 mirrors to locally use the new store.SkipMsgs to optimize speed.

This CL also has a bug fix for memory store state with delete details, and several general improvements to the filestore impl.

Signed-off-by: Derek Collison <derek@nats.io>




Patch:
diff --git a/go.mod b/go.mod
index b295e667..5167d491 100644
--- a/go.mod
+++ b/go.mod
@@ -11,6 +11,6 @@ require (
 	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.16.0
+	golang.org/x/sys v0.15.0
 	golang.org/x/time v0.5.0
 )
diff --git a/go.sum b/go.sum
index fd94d9da..99149367 100644
--- a/go.sum
+++ b/go.sum
@@ -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.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
-golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+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/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/filestore.go b/server/filestore.go
index c7244c82..0eb959b4 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -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_cluster.go b/server/jetstream_cluster.go
index d6ff7515..520d1ed1 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()
 
@@ -1939,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])))
@@ -2197,13 +2208,9 @@ 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 || isRecovering || isRestore || time.Since(lastSnapTime) < minSnapDelta {
+		if mset == nil || isRestore || time.Since(lastSnapTime) < minSnapDelta {
 			return
 		}
 
@@ -2230,6 +2237,7 @@ 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
@@ -4555,13 +4563,9 @@ 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 recovering || (!force && time.Since(lastSnapTime) < minSnapDelta) {
+		if !force && time.Since(lastSnapTime) < minSnapDelta {
 			return
 		}
 
@@ -4612,6 +4616,7 @@ func (js *jetStream) monitorConsumer(o *consumer, ca *consumerAssignment) {
 
 	// Track if we are leader.
 	var isLeader bool
+	recovering := true
 
 	for {
 		select {
@@ -6837,6 +6842,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) {
@@ -7607,9 +7628,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.
@@ -7618,7 +7640,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
@@ -7628,14 +7650,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)
 	}
@@ -7665,6 +7695,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 {
@@ -7743,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()
@@ -7777,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()
 
@@ -7813,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 {
@@ -7888,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 {
@@ -7908,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.
@@ -7939,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.
@@ -7956,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
@@ -8006,6 +8018,7 @@ RETRY:
 					goto RETRY
 				}
 			}
+			notActive.Reset(activityInterval)
 			msgsQ.recycle(&mrecs)
 		case <-notActive.C:
 			if mrecs := msgsQ.pop(); len(mrecs) > 0 {
@@ -8034,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)
@@ -8381,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)
 
@@ -8401,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) {
@@ -8418,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()
@@ -8450,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())
@@ -8462,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:
@@ -8484,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.
@@ -8511,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)
@@ -8538,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()
@@ -8565,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/store.go b/server/store.go
index 2db46ee1..649dcecd 100644
--- a/server/store.go
+++ b/server/store.go
@@ -85,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)
diff --git a/server/stream.go b/server/stream.go
index bf0139b2..c008d4e0 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -2367,23 +2367,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)
 	}
 }


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 6d98f4d1..0235e0a1 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -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..ec6e7276 100644
--- a/server/memstore_test.go
+++ b/server/memstore_test.go
@@ -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/norace_test.go b/server/norace_test.go
index dc0fe3a9..02e73d31 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -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)
+	})
+}