Allow WAL entries to conflict if they are above commited index, select entry with the highest term

Signed-off-by: Marek Siarkowicz <[email protected]>

Author: serathius

tests/robustness/report/wal.go

@@ -18,6 +18,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -62,14 +63,17 @@ func PersistedRequests(lg *zap.Logger, dataDirs []string) ([]model.EtcdRequest,
 		return nil, errors.New("no data dirs")
 	}
 	entriesPersistedInWAL := make([][]raftpb.Entry, len(dataDirs))
+	var minCommitIndex uint64 = math.MaxUint64
 	for i, dir := range dataDirs {
-		_, entries, err := ReadWAL(lg, dir)
+		state, entries, err := ReadWAL(lg, dir)
 		if err != nil {
 			lg.Error("Failed to read WAL", zap.Error(err), zap.String("data-dir", dir))
+			continue
 		}
+		minCommitIndex = min(minCommitIndex, state.Commit)
 		entriesPersistedInWAL[i] = entries
 	}
-	entries, err := mergeMembersEntries(entriesPersistedInWAL)
+	entries, err := mergeMembersEntries(minCommitIndex, entriesPersistedInWAL)
 	if err != nil {
 		return nil, err
 	}
@@ -89,7 +93,7 @@ func PersistedRequests(lg *zap.Logger, dataDirs []string) ([]model.EtcdRequest,
 	return persistedRequests, nil
 }
 
-func mergeMembersEntries(memberEntries [][]raftpb.Entry) ([]raftpb.Entry, error) {
+func mergeMembersEntries(minCommitIndex uint64, memberEntries [][]raftpb.Entry) ([]raftpb.Entry, error) {
 	for _, entries := range memberEntries {
 		var lastIndex uint64
 		for _, e := range entries {
@@ -162,10 +166,20 @@ func mergeMembersEntries(memberEntries [][]raftpb.Entry) ([]raftpb.Entry, error)
 			if vote != topVotes {
 				continue
 			}
-			if entryWithMostVotes != nil && !reflect.DeepEqual(*entryWithMostVotes, memberEntries[i][memberIndices[i]]) {
-				return nil, fmt.Errorf("mismatching entries on raft index %d", raftIndex)
+			entry := memberEntries[i][memberIndices[i]]
+			if entryWithMostVotes == nil {
+				entryWithMostVotes = &entry
+				continue
+			}
+			if entryWithMostVotes.Term != entry.Term && entry.Index > minCommitIndex {
+				if entryWithMostVotes.Term < entry.Term {
+					entryWithMostVotes = &entry
+				}
+				continue
+			}
+			if !reflect.DeepEqual(*entryWithMostVotes, entry) {
+				return nil, fmt.Errorf("mismatching entries on raft index %d, mostVotes: %+v, other: %+v", raftIndex, *entryWithMostVotes, entry)
 			}
-			entryWithMostVotes = &memberEntries[i][memberIndices[i]]
 		}
 		mergedHistory = append(mergedHistory, *entryWithMostVotes)
 	}

tests/robustness/report/wal_test.go

@@ -24,10 +24,11 @@ import (
 
 func TestMergeMemberEntries(t *testing.T) {
 	tcs := []struct {
-		name          string
-		memberEntries [][]raftpb.Entry
-		expectErr     string
-		expectEntries []raftpb.Entry
+		name           string
+		minCommitIndex uint64
+		memberEntries  [][]raftpb.Entry
+		expectErr      string
+		expectEntries  []raftpb.Entry
 	}{
 		{
 			name:          "Error when empty data dir",
@@ -314,10 +315,43 @@ func TestMergeMemberEntries(t *testing.T) {
 			},
 			expectErr: "mismatching entries on raft index 1",
 		},
+		{
+			name:           "Error if entries mismatch on index before minCommitIndex",
+			minCommitIndex: 2,
+			memberEntries: [][]raftpb.Entry{
+				{
+					raftpb.Entry{Index: 1, Term: 1, Data: []byte("a")},
+					raftpb.Entry{Index: 2, Term: 1, Data: []byte("b")},
+				},
+				{
+					raftpb.Entry{Index: 1, Term: 1, Data: []byte("a")},
+					raftpb.Entry{Index: 2, Term: 2, Data: []byte("c")},
+				},
+			},
+			expectErr: "mismatching entries on raft index 2",
+		},
+		{
+			name:           "Select entry with higher term if they conflict on uncommitted index",
+			minCommitIndex: 1,
+			memberEntries: [][]raftpb.Entry{
+				{
+					raftpb.Entry{Index: 1, Term: 1, Data: []byte("a")},
+					raftpb.Entry{Index: 2, Term: 1, Data: []byte("b")},
+				},
+				{
+					raftpb.Entry{Index: 1, Term: 1, Data: []byte("a")},
+					raftpb.Entry{Index: 2, Term: 2, Data: []byte("x")},
+				},
+			},
+			expectEntries: []raftpb.Entry{
+				{Index: 1, Term: 1, Data: []byte("a")},
+				{Index: 2, Term: 2, Data: []byte("x")},
+			},
+		},
 	}
 	for _, tc := range tcs {
 		t.Run(tc.name, func(t *testing.T) {
-			entries, err := mergeMembersEntries(tc.memberEntries)
+			entries, err := mergeMembersEntries(tc.minCommitIndex, tc.memberEntries)
 			if tc.expectErr == "" {
 				require.NoError(t, err)
 			} else {