storage: harden getQuorumIndex

It wasn't looking at the progress.State, which should be harmless
but better not to trust that the Match field is correctly populated
for followers in probing status.

Note that there's a potential behavior change here: if a follower
needs a snapshot, it will have a Match field. But if we're computing
a quorum index, we implicitly assume that progress can be made from
that index since a quorum of followers "has it". A follower which
needs a snapshot is not able to help out with progress until it
has been caught up, so including it in the quorum index is not
beneficial.

Release note: None

pkg/storage/raft_log_queue.go

@@ -406,7 +406,11 @@ func computeTruncateDecision(input truncateDecisionInput) truncateDecision {
 func getQuorumIndex(raftStatus *raft.Status) uint64 {
 	match := make([]uint64, 0, len(raftStatus.Progress))
 	for _, progress := range raftStatus.Progress {
-		match = append(match, progress.Match)
+		if progress.State == raft.ProgressStateReplicate {
+			match = append(match, progress.Match)
+		} else {
+			match = append(match, 0)
+		}
 	}
 	sort.Sort(uint64Slice(match))
 	quorum := computeQuorum(len(match))

pkg/storage/raft_log_queue_test.go

@@ -92,13 +92,24 @@ func TestGetQuorumIndex(t *testing.T) {
 			Progress: make(map[uint64]raft.Progress),
 		}
 		for j, v := range c.progress {
-			status.Progress[uint64(j)] = raft.Progress{Match: v}
+			status.Progress[uint64(j)] = raft.Progress{State: raft.ProgressStateReplicate, Match: v}
 		}
 		quorumMatchedIndex := getQuorumIndex(status)
 		if c.expected != quorumMatchedIndex {
 			t.Fatalf("%d: expected %d, but got %d", i, c.expected, quorumMatchedIndex)
 		}
 	}
+
+	// Verify that only replicating followers are taken into account (i.e. others
+	// are treated as Match == 0).
+	status := &raft.Status{
+		Progress: map[uint64]raft.Progress{
+			1: {State: raft.ProgressStateReplicate, Match: 100},
+			2: {State: raft.ProgressStateSnapshot, Match: 100},
+			3: {State: raft.ProgressStateReplicate, Match: 90},
+		},
+	}
+	assert.Equal(t, uint64(90), getQuorumIndex(status))
 }
 
 func TestComputeTruncateDecision(t *testing.T) {