[FIXED] Bug that would cause extended purge or compact to fail. (#5264)

In memstore if upto sequence was not found, e.g. deleted already, the
compact or purge command would fail.

Signed-off-by: Derek Collison <[email protected]>

server/memstore.go

@@ -842,15 +842,15 @@ func (ms *memStore) Compact(seq uint64) (uint64, error) {
 	ms.mu.Lock()
 	cb := ms.scb
 	if seq <= ms.state.LastSeq {
-		sm, ok := ms.msgs[seq]
-		if !ok {
-			ms.mu.Unlock()
-			return 0, ErrStoreMsgNotFound
-		}
 		fseq := ms.state.FirstSeq
-		ms.state.FirstSeq = seq
-		ms.state.FirstTime = time.Unix(0, sm.ts).UTC()
-
+		// Determine new first sequence.
+		for ; seq <= ms.state.LastSeq; seq++ {
+			if sm, ok := ms.msgs[seq]; ok {
+				ms.state.FirstSeq = seq
+				ms.state.FirstTime = time.Unix(0, sm.ts).UTC()
+				break
+			}
+		}
 		for seq := seq - 1; seq >= fseq; seq-- {
 			if sm := ms.msgs[seq]; sm != nil {
 				bytes += memStoreMsgSize(sm.subj, sm.hdr, sm.msg)

server/memstore_test.go

@@ -1022,3 +1022,32 @@ func TestMemStoreMultiLastSeqsMaxAllowed(t *testing.T) {
 	require_True(t, seqs == nil)
 	require_Error(t, err, ErrTooManyResults)
 }
+
+// Bug would cause PurgeEx to fail if it encountered a deleted msg at sequence to delete up to.
+func TestMemStorePurgeExWithDeletedMsgs(t *testing.T) {
+	cfg := &StreamConfig{
+		Name:     "zzz",
+		Subjects: []string{"foo"},
+		Storage:  MemoryStorage,
+	}
+	ms, err := newMemStore(cfg)
+	require_NoError(t, err)
+	defer ms.Stop()
+
+	msg := []byte("abc")
+	for i := 1; i <= 10; i++ {
+		ms.StoreMsg("foo", nil, msg)
+	}
+	ms.RemoveMsg(2)
+	ms.RemoveMsg(9) // This was the bug
+
+	n, err := ms.PurgeEx(_EMPTY_, 9, 0)
+	require_NoError(t, err)
+	require_Equal(t, n, 7)
+
+	var state StreamState
+	ms.FastState(&state)
+	require_Equal(t, state.FirstSeq, 10)
+	require_Equal(t, state.LastSeq, 10)
+	require_Equal(t, state.Msgs, 1)
+}