Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

A simpler stream state to detect change for snapshots. #4074

Merged
merged 1 commit into from
Apr 19, 2023
Merged

A simpler stream state to detect change for snapshots. #4074

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 19 additions & 15 deletions server/jetstream_cluster.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1968,33 +1968,37 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
}
accName := acc.GetName()

// Hash of the last snapshot (fixed size in memory).
var lastSnap []byte
// Used to represent how we can detect a changed state quickly and without representing
// a complete and detailed state which could be costly in terms of memory, cpu and GC.
// This only entails how many messages, and the first and last sequence of the stream.
// This is all that is needed to detect a change, and we can get this from FilteredState()
// with and empty filter.
var lastState SimpleState
var lastSnapTime time.Time

// Highwayhash key for generating hashes.
key := make([]byte, 32)
rand.Read(key)

// Should only to be called from leader.
doSnapshot := func() {
if mset == nil || isRestore || time.Since(lastSnapTime) < minSnapDelta {
return
}

snap := mset.stateSnapshot()
ne, nb := n.Size()
hash := highwayhash.Sum(snap, key)
// Before we actually calculate the detailed state and encode it, let's check the
// simple state to detect any changes.
curState := mset.store.FilteredState(0, _EMPTY_)

// If the state hasn't changed but the log has gone way over
// the compaction size then we will want to compact anyway.
// This shouldn't happen for streams like it can for pull
// consumers on idle streams but better to be safe than sorry!
if !bytes.Equal(hash[:], lastSnap) || ne >= compactNumMin || nb >= compactSizeMin {
if err := n.InstallSnapshot(snap); err == nil {
lastSnap, lastSnapTime = hash[:], time.Now()
} else if err != errNoSnapAvailable && err != errNodeClosed {
s.Warnf("Failed to install snapshot for '%s > %s' [%s]: %v", mset.acc.Name, mset.name(), n.Group(), err)
}
ne, nb := n.Size()
if curState == lastState && ne < compactNumMin && nb < compactSizeMin {
return
}

if err := n.InstallSnapshot(mset.stateSnapshot()); err == nil {
lastState, lastSnapTime = curState, time.Now()
} else if err != errNoSnapAvailable && err != errNodeClosed {
s.Warnf("Failed to install snapshot for '%s > %s' [%s]: %v", mset.acc.Name, mset.name(), n.Group(), err)
}
}

Expand Down