Repository: nats-io/nats-server
Pull request #3844 (2023-02-04 22:13:31)

[FIXED] Snapshots Issue
Due to a bug, in rare circumstances we could write an empty snapshot for applied == 0. This would cause a spinning behavior at the raft layer.

To fix this issue, given that these could be out in deployments potentially.

    1. Allow Truncate() to also properly do a reset of the store when terms were the only mismatch.
    2. During testing fixed memstore truncate and also made sure per subject info was also cleaned up.
    3. Added fix to detect a bad snapshot on initialization and remove.
    4. Do not allow snapshots for applied == 0.

Also small fixes for new staticcheck.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/accounts.go b/server/accounts.go
index 9a89c624..c7edf14f 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -1134,7 +1134,7 @@ func (a *Account) UnTrackServiceExport(service string) {
 	}
 
 	a.mu.Lock()
-	if a == nil || a.exports.services == nil {
+	if a.exports.services == nil {
 		a.mu.Unlock()
 		return
 	}
diff --git a/server/filestore.go b/server/filestore.go
index e04c5bb5..f8f52775 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -2865,6 +2865,10 @@ func (mb *msgBlock) truncate(sm *StoreMsg) (nmsgs, nbytes uint64, err error) {
 
 	// Clear our cache.
 	mb.clearCacheAndOffset()
+
+	// Redo per subject info for this block.
+	mb.resetPerSubjectInfo()
+
 	mb.mu.Unlock()
 
 	// Write our index file.
@@ -4983,8 +4987,64 @@ SKIP:
 	return purged, err
 }
 
-// Truncate will truncate a stream store up to and including seq. Sequence needs to be valid.
+// Will completely reset our store.
+func (fs *fileStore) reset() error {
+
+	fs.mu.Lock()
+	if fs.closed {
+		fs.mu.Unlock()
+		return ErrStoreClosed
+	}
+	if fs.sips > 0 {
+		fs.mu.Unlock()
+		return ErrStoreSnapshotInProgress
+	}
+
+	var purged, bytes uint64
+	cb := fs.scb
+
+	if cb != nil {
+		for _, mb := range fs.blks {
+			mb.mu.Lock()
+			purged += mb.msgs
+			bytes += mb.bytes
+			mb.dirtyCloseWithRemove(true)
+			mb.mu.Unlock()
+		}
+	}
+
+	// Reset
+	fs.state.FirstSeq = 0
+	fs.state.FirstTime = time.Time{}
+	fs.state.LastSeq = 0
+	fs.state.LastTime = time.Now().UTC()
+	// Update msgs and bytes.
+	fs.state.Msgs = 0
+	fs.state.Bytes = 0
+
+	// Reset blocks.
+	fs.blks, fs.lmb = nil, nil
+
+	// Reset subject mappings.
+	fs.psim = make(map[string]*psi)
+	fs.bim = make(map[uint32]*msgBlock)
+
+	fs.mu.Unlock()
+
+	if cb != nil {
+		cb(-int64(purged), -int64(bytes), 0, _EMPTY_)
+	}
+
+	return nil
+}
+
+// Truncate will truncate a stream store up to seq. Sequence needs to be valid.
 func (fs *fileStore) Truncate(seq uint64) error {
+	// Check for request to reset.
+	if seq == 0 {
+		return fs.reset()
+	}
+
 	fs.mu.Lock()
 
 	if fs.closed {
@@ -5017,7 +5077,6 @@ func (fs *fileStore) Truncate(seq uint64) error {
 
 	// Truncate our new last message block.
 	nmsgs, nbytes, err := nlmb.truncate(lsm)
-
 	if err != nil {
 		fs.mu.Unlock()
 		return err
@@ -5043,6 +5102,9 @@ func (fs *fileStore) Truncate(seq uint64) error {
 	fs.state.Msgs -= purged
 	fs.state.Bytes -= bytes
 
+	// Reset our subject lookup info.
+	fs.resetGlobalPerSubjectInfo()
+
 	cb := fs.scb
 	fs.mu.Unlock()
 
@@ -5249,6 +5311,22 @@ func (mb *msgBlock) removeSeqPerSubject(subj string, seq uint64, smp *StoreMsg)
 	}
 }
 
+// Lock should be held.
+func (fs *fileStore) resetGlobalPerSubjectInfo() {
+	// Clear any global subject state.
+	fs.psim = make(map[string]*psi)
+	for _, mb := range fs.blks {
+		fs.populateGlobalPerSubjectInfo(mb)
+	}
+}
+
+// Lock should be held.
+func (mb *msgBlock) resetPerSubjectInfo() error {
+	mb.fss = nil
+	mb.removePerSubjectInfoLocked()
+	return mb.generatePerSubjectInfo(true)
+}
+
 // generatePerSubjectInfo will generate the per subject info via the raw msg block.
 func (mb *msgBlock) generatePerSubjectInfo(hasLock bool) error {
 	if !hasLock {
diff --git a/server/memstore.go b/server/memstore.go
index 4cd8037c..6d84bc87 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -566,8 +566,47 @@ func (ms *memStore) Compact(seq uint64) (uint64, error) {
 	return purged, nil
 }
 
-// Truncate will truncate a stream store up to and including seq. Sequence needs to be valid.
+// Will completely reset our store.
+func (ms *memStore) reset() error {
+
+	ms.mu.Lock()
+	var purged, bytes uint64
+	cb := ms.scb
+	if cb != nil {
+		for _, sm := range ms.msgs {
+			purged++
+			bytes += memStoreMsgSize(sm.subj, sm.hdr, sm.msg)
+		}
+	}
+
+	// Reset
+	ms.state.FirstSeq = 0
+	ms.state.FirstTime = time.Time{}
+	ms.state.LastSeq = 0
+	ms.state.LastTime = time.Now().UTC()
+	// Update msgs and bytes.
+	ms.state.Msgs = 0
+	ms.state.Bytes = 0
+	// Reset msgs and fss.
+	ms.msgs = make(map[uint64]*StoreMsg)
+	ms.fss = make(map[string]*SimpleState)
+
+	ms.mu.Unlock()
+
+	if cb != nil {
+		cb(-int64(purged), -int64(bytes), 0, _EMPTY_)
+	}
+
+	return nil
+}
+
+// Truncate will truncate a stream store up to seq. Sequence needs to be valid.
 func (ms *memStore) Truncate(seq uint64) error {
+	// Check for request to reset.
+	if seq == 0 {
+		return ms.reset()
+	}
+
 	var purged, bytes uint64
 
 	ms.mu.Lock()
@@ -581,7 +620,8 @@ func (ms *memStore) Truncate(seq uint64) error {
 		if sm := ms.msgs[i]; sm != nil {
 			purged++
 			bytes += memStoreMsgSize(sm.subj, sm.hdr, sm.msg)
-			delete(ms.msgs, seq)
+			delete(ms.msgs, i)
+			ms.removeSeqPerSubject(sm.subj, i)
 		}
 	}
 	// Reset last.
@@ -590,6 +630,7 @@ func (ms *memStore) Truncate(seq uint64) error {
 	// Update msgs and bytes.
 	ms.state.Msgs -= purged
 	ms.state.Bytes -= bytes
+
 	cb := ms.scb
 	ms.mu.Unlock()
 
@@ -855,7 +896,10 @@ func (ms *memStore) FastState(state *StreamState) {
 	state.LastSeq = ms.state.LastSeq
 	state.LastTime = ms.state.LastTime
 	if state.LastSeq > state.FirstSeq {
-		state.NumDeleted = int((state.LastSeq - state.FirstSeq) - state.Msgs + 1)
+		state.NumDeleted = int((state.LastSeq - state.FirstSeq + 1) - state.Msgs)
+		if state.NumDeleted < 0 {
+			state.NumDeleted = 0
+		}
 	}
 	state.Consumers = ms.consumers
 	state.NumSubjects = len(ms.fss)
@@ -872,17 +916,18 @@ func (ms *memStore) State() StreamState {
 	state.Deleted = nil
 
 	// Calculate interior delete details.
-	if state.LastSeq > state.FirstSeq {
-		if state.NumDeleted = int((state.LastSeq - state.FirstSeq) - state.Msgs + 1); state.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 = append(state.Deleted, seq)
-				}
+	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 = append(state.Deleted, seq)
 			}
 		}
 	}
+	if len(state.Deleted) > 0 {
+		state.NumDeleted = len(state.Deleted)
+	}
 
 	return state
 }
diff --git a/server/raft.go b/server/raft.go
index 6bd7ddc1..ab5a684b 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -939,7 +939,7 @@ func (n *raft) InstallSnapshot(data []byte) error {
 	var state StreamState
 	n.wal.FastState(&state)
 
-	if n.snapfile != _EMPTY_ && state.FirstSeq >= n.applied {
+	if state.FirstSeq >= n.applied {
 		n.Unlock()
 		return nil
 	}
@@ -1075,6 +1075,7 @@ func (n *raft) setupLastSnapshot() {
 		n.pterm = snap.lastTerm
 		n.commit = snap.lastIndex
 		n.applied = snap.lastIndex
+
 		n.apply.push(&CommittedEntry{n.commit, []*Entry{{EntrySnapshot, snap.data}}})
 		if _, err := n.wal.Compact(snap.lastIndex); err != nil {
 			n.setWriteErrLocked(err)
@@ -1123,6 +1124,15 @@ func (n *raft) loadLastSnapshot() (*snapshot, error) {
 		data:      buf[20+lps : hoff],
 	}
 
+	// We had a bug in 2.9.12 that would allow snapshots on last index of 0.
+	// Detect that here and return err.
+	if snap.lastIndex == 0 {
+		n.warn("Snapshot with last index 0 is invalid, cleaning up")
+		os.Remove(n.snapfile)
+		n.snapfile = _EMPTY_
+		return nil, errSnapshotCorrupt
+	}
+
 	return snap, nil
 }
 
@@ -1241,6 +1251,7 @@ func (n *raft) GroupLeader() string {
 }
 
 // Guess the best next leader. Stepdown will check more thoroughly.
+// Lock should be held.
 func (n *raft) selectNextLeader() string {
 	nextLeader, hli := noLeader, uint64(0)
 	for peer, ps := range n.peers {
@@ -2720,7 +2731,7 @@ func (n *raft) truncateWAL(term, index uint64) {
 
 	// Check to see if we invalidated any snapshots that might have held state
 	// from the entries we are truncating.
-	if snap, _ := n.loadLastSnapshot(); snap != nil && snap.lastIndex > index {
+	if snap, _ := n.loadLastSnapshot(); snap != nil && snap.lastIndex >= index {
 		os.Remove(n.snapfile)
 		n.snapfile = _EMPTY_
 	}
diff --git a/server/stream.go b/server/stream.go
index f315e346..841fcc39 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -777,7 +777,9 @@ func (mset *stream) rebuildDedupe() {
 	}
 
 	var smv StoreMsg
-	state := mset.store.State()
+	var state StreamState
+	mset.store.FastState(&state)
+
 	for seq := sseq; seq <= state.LastSeq; seq++ {
 		sm, err := mset.store.LoadMsg(seq, &smv)
 		if err != nil {


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 925cb845..cbc2b399 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -964,7 +964,7 @@ func TestFileStoreStreamTruncate(t *testing.T) {
 
 	fs, err := newFileStoreWithCreated(
 		FileStoreConfig{StoreDir: storeDir, BlockSize: 350},
-		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
+		StreamConfig{Name: "zzz", Subjects: []string{"*"}, Storage: FileStorage},
 		time.Now(),
 		prf,
 	)
@@ -973,12 +973,19 @@ func TestFileStoreStreamTruncate(t *testing.T) {
 	}
 	defer fs.Stop()
 
+	tseq := uint64(50)
+
 	subj, toStore := "foo", uint64(100)
-	for i := uint64(1); i <= toStore; i++ {
-		if _, _, err := fs.StoreMsg(subj, nil, []byte("ok")); err != nil {
-			t.Fatalf("Error storing msg: %v", err)
-		}
+	for i := uint64(1); i < tseq; i++ {
+		_, _, err := fs.StoreMsg(subj, nil, []byte("ok"))
+		require_NoError(t, err)
+	}
+	subj = "bar"
+	for i := tseq; i <= toStore; i++ {
+		_, _, err := fs.StoreMsg(subj, nil, []byte("ok"))
+		require_NoError(t, err)
 	}
+
 	if state := fs.State(); state.Msgs != toStore {
 		t.Fatalf("Expected %d msgs, got %d", toStore, state.Msgs)
 	}
@@ -988,7 +995,6 @@ func TestFileStoreStreamTruncate(t *testing.T) {
 		t.Fatalf("Expected err of '%v', got '%v'", ErrInvalidSequence, err)
 	}
 
-	tseq := uint64(50)
 	if err := fs.Truncate(tseq); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1006,15 +1012,16 @@ func TestFileStoreStreamTruncate(t *testing.T) {
 	if err := fs.Truncate(tseq); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-	if state := fs.State(); state.Msgs != tseq-2 {
+	state := fs.State()
+	if state.Msgs != tseq-2 {
 		t.Fatalf("Expected %d msgs, got %d", tseq-2, state.Msgs)
 	}
 	expected := []uint64{10, 20}
-	if state := fs.State(); !reflect.DeepEqual(state.Deleted, expected) {
+	if !reflect.DeepEqual(state.Deleted, expected) {
 		t.Fatalf("Expected deleted to be %+v, got %+v\n", expected, state.Deleted)
 	}
 
-	before := fs.State()
+	before := state
 
 	// Make sure we can recover same state.
 	fs.Stop()
@@ -5040,3 +5047,45 @@ func TestFileStoreAllFilteredStateWithDeleted(t *testing.T) {
 	remove(8, 10, 12, 14, 16, 18)
 	checkFilteredState(7, 88, 7, 100)
 }
+
+func TestFileStoreStreamTruncateResetMultiBlock(t *testing.T) {
+	storeDir := t.TempDir()
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: storeDir, BlockSize: 128},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	subj, msg := "foo", []byte("Hello World")
+	for i := 0; i < 1000; i++ {
+		_, _, err := fs.StoreMsg(subj, nil, msg)
+		require_NoError(t, err)
+	}
+	require_True(t, fs.numMsgBlocks() == 500)
+
+	// Reset everything
+	require_NoError(t, fs.Truncate(0))
+	require_True(t, fs.numMsgBlocks() == 0)
+
+	state := fs.State()
+	require_True(t, state.Msgs == 0)
+	require_True(t, state.Bytes == 0)
+	require_True(t, state.FirstSeq == 0)
+	require_True(t, state.LastSeq == 0)
+	require_True(t, state.NumSubjects == 0)
+	require_True(t, state.NumDeleted == 0)
+
+	for i := 0; i < 1000; i++ {
+		_, _, err := fs.StoreMsg(subj, nil, msg)
+		require_NoError(t, err)
+	}
+
+	state = fs.State()
+	require_True(t, state.Msgs == 1000)
+	require_True(t, state.Bytes == 44000)
+	require_True(t, state.FirstSeq == 1)
+	require_True(t, state.LastSeq == 1000)
+	require_True(t, state.NumSubjects == 1)
+	require_True(t, state.NumDeleted == 0)
+}
diff --git a/server/memstore_test.go b/server/memstore_test.go
index c19dad85..01352c2b 100644
--- a/server/memstore_test.go
+++ b/server/memstore_test.go
@@ -353,12 +353,19 @@ func TestMemStoreStreamTruncate(t *testing.T) {
 		t.Fatalf("Unexpected error creating store: %v", err)
 	}
 
+	tseq := uint64(50)
+
 	subj, toStore := "foo", uint64(100)
-	for i := uint64(1); i <= toStore; i++ {
-		if _, _, err := ms.StoreMsg(subj, nil, []byte("ok")); err != nil {
-			t.Fatalf("Error storing msg: %v", err)
-		}
+	for i := uint64(1); i < tseq; i++ {
+		_, _, err := ms.StoreMsg(subj, nil, []byte("ok"))
+		require_NoError(t, err)
+	}
+	subj = "bar"
+	for i := tseq; i <= toStore; i++ {
+		_, _, err := ms.StoreMsg(subj, nil, []byte("ok"))
+		require_NoError(t, err)
 	}
+
 	if state := ms.State(); state.Msgs != toStore {
 		t.Fatalf("Expected %d msgs, got %d", toStore, state.Msgs)
 	}
@@ -368,7 +375,6 @@ func TestMemStoreStreamTruncate(t *testing.T) {
 		t.Fatalf("Expected err of '%v', got '%v'", ErrInvalidSequence, err)
 	}
 
-	tseq := uint64(50)
 	if err := ms.Truncate(tseq); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -386,8 +392,15 @@ func TestMemStoreStreamTruncate(t *testing.T) {
 	if err := ms.Truncate(tseq); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
+	state := ms.State()
+	if state.Msgs != tseq-2 {
+		t.Fatalf("Expected %d msgs, got %d", tseq-2, state.Msgs)
+	}
+	if state.NumSubjects != 1 {
+		t.Fatalf("Expected only 1 subject, got %d", state.NumSubjects)
+	}
 	expected := []uint64{10, 20}
-	if state := ms.State(); !reflect.DeepEqual(state.Deleted, expected) {
+	if !reflect.DeepEqual(state.Deleted, expected) {
 		t.Fatalf("Expected deleted to be %+v, got %+v\n", expected, state.Deleted)
 	}
 }
@@ -442,3 +455,43 @@ func TestMemStoreUpdateMaxMsgsPerSubject(t *testing.T) {
 		t.Fatalf("Expected to have %d stored, got %d", 10, ss.Msgs)
 	}
 }
+
+func TestMemStoreStreamTruncateReset(t *testing.T) {
+	cfg := &StreamConfig{
+		Name:     "TEST",
+		Storage:  MemoryStorage,
+		Subjects: []string{"foo"},
+	}
+	ms, err := newMemStore(cfg)
+	require_NoError(t, err)
+
+	subj, msg := "foo", []byte("Hello World")
+	for i := 0; i < 1000; i++ {
+		_, _, err := ms.StoreMsg(subj, nil, msg)
+		require_NoError(t, err)
+	}
+
+	// Reset everything
+	require_NoError(t, ms.Truncate(0))
+
+	state := ms.State()
+	require_True(t, state.Msgs == 0)
+	require_True(t, state.Bytes == 0)
+	require_True(t, state.FirstSeq == 0)
+	require_True(t, state.LastSeq == 0)
+	require_True(t, state.NumSubjects == 0)
+	require_True(t, state.NumDeleted == 0)
+
+	for i := 0; i < 1000; i++ {
+		_, _, err := ms.StoreMsg(subj, nil, msg)
+		require_NoError(t, err)
+	}
+
+	state = ms.State()
+	require_True(t, state.Msgs == 1000)
+	require_True(t, state.Bytes == 30000)
+	require_True(t, state.FirstSeq == 1)
+	require_True(t, state.LastSeq == 1000)
+	require_True(t, state.NumSubjects == 1)
+	require_True(t, state.NumDeleted == 0)
+}
diff --git a/server/websocket_test.go b/server/websocket_test.go
index c979cf83..3439f8e8 100644
--- a/server/websocket_test.go
+++ b/server/websocket_test.go
@@ -1,4 +1,4 @@
-// Copyright 2020 The NATS Authors
+// Copyright 2020-2023 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
@@ -4046,7 +4046,6 @@ func wsBenchPub(b *testing.B, numPubs int, compress bool, payload string) {
 	s := RunServer(opts)
 	defer s.Shutdown()
 
-	n := b.N
 	extra := 0
 	pubProto := []byte(fmt.Sprintf("PUB %s %d\r\n%s\r\n", testWSBenchSubject, len(payload), payload))
 	singleOpBuf := testWSCreateClientMsg(wsBinaryMessage, 1, true, compress, pubProto)
@@ -4065,7 +4064,7 @@ func wsBenchPub(b *testing.B, numPubs int, compress bool, payload string) {
 		}
 	}
 	sendBuf := testWSCreateClientMsg(wsBinaryMessage, 1, true, compress, tmp)
-	n = b.N / pb
+	n := b.N / pb
 	extra = b.N - (n * pb)
 
 	wg := sync.WaitGroup{}