gogrlx/nats-server
1
0
Fork 0
mirror of https://github.com/gogrlx/nats-server.git synced 2026-04-02 03:38:42 -07:00
Code Issues Packages Projects Releases Wiki Activity

When creating replicated mirrors where the source stream had a very large starting sequence number, the server would use excessive CPU and Memory.

Browse Source 
This is due to the mirroring functionality trying to skip messages when it detects a gap. In a replicated stream this puts excessive stress on the raft system.
This step is not needed at all if the mirror stream has no messages, we can simply jump ahead.

Signed-off-by: Derek Collison <derek@nats.io>
This commit is contained in:
Derek Collison
2023-06-15 17:20:15 -07:00
parent 367d857612
commit 087a28a13e
2 changed files with 76 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
server/stream.go
View File

@@ -1671,7 +1671,7 @@ func (mset *stream) purge(preq *JSApiStreamPurgeRequest) (purged uint64, err err
mset.mu.RUnlock()
return 0, errors.New("sealed stream")
}
store := mset.store
store, mlseq := mset.store, mset.lseq
mset.mu.RUnlock()
if preq != nil {
@@ -1683,11 +1683,17 @@ func (mset *stream) purge(preq *JSApiStreamPurgeRequest) (purged uint64, err err
return purged, err
}
// Purge consumers.
// Grab our stream state.
var state StreamState
store.FastState(&state)
fseq, lseq := state.FirstSeq, state.LastSeq
// Check if our last has moved past what our original last sequence was, if so reset.
if lseq > mlseq {
mset.setLastSeq(lseq)
}
// Purge consumers.
// Check for filtered purge.
if preq != nil && preq.Subject != _EMPTY_ {
ss := store.FilteredState(state.FirstSeq, preq.Subject)
@@ -2399,7 +2405,14 @@ func (mset *stream) setupMirrorConsumer() error {
// Check if we need to skip messages.
if state.LastSeq != ccr.ConsumerInfo.Delivered.Stream {
mset.skipMsgs(state.LastSeq+1, ccr.ConsumerInfo.Delivered.Stream)
// Check to see if delivered is past our last and we have no msgs. This will help the
// case when mirroring a stream that has a very high starting sequence number.
if state.Msgs == 0 && ccr.ConsumerInfo.Delivered.Stream > state.LastSeq {
mset.store.PurgeEx(_EMPTY_, ccr.ConsumerInfo.Delivered.Stream+1, 0)
mset.lseq = ccr.ConsumerInfo.Delivered.Stream
} else {
mset.skipMsgs(state.LastSeq+1, ccr.ConsumerInfo.Delivered.Stream)
}
}
// Capture consumer name.