Merge pull request cockroachdb#6423 from bdarnell/reject-removed-replicas

storage: Reject all messages from removed replicas

Author: bdarnell

storage/client_raft_test.go

@@ -1391,6 +1391,79 @@ func TestReplicateRogueRemovedNode(t *testing.T) {
 	finishWG.Wait()
 }
 
+func TestReplicateRemovedNodeDisruptiveElection(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	mtc := startMultiTestContext(t, 4)
+	defer mtc.Stop()
+
+	// Move the first range from the first node to the other three.
+	rangeID := roachpb.RangeID(1)
+	mtc.replicateRange(rangeID, 1, 2, 3)
+	mtc.unreplicateRange(rangeID, 0)
+	mtc.expireLeaderLeases()
+
+	// Write on the second node, to ensure that the other nodes have
+	// established leadership after the first node's removal.
+	incArgs := incrementArgs([]byte("a"), 5)
+	if _, err := client.SendWrapped(mtc.distSenders[1], nil, &incArgs); err != nil {
+		t.Fatal(err)
+	}
+
+	// Save the current term, which is the latest among the live stores.
+	findTerm := func() uint64 {
+		var term uint64
+		for i := 1; i < 4; i++ {
+			s := mtc.stores[i].RaftStatus(rangeID)
+			if s.Term > term {
+				term = s.Term
+			}
+		}
+		return term
+	}
+	term := findTerm()
+	if term == 0 {
+		t.Fatalf("expected non-zero term")
+	}
+
+	replica0 := roachpb.ReplicaDescriptor{
+		ReplicaID: roachpb.ReplicaID(mtc.stores[0].StoreID()),
+		NodeID:    roachpb.NodeID(mtc.stores[0].StoreID()),
+		StoreID:   mtc.stores[0].StoreID(),
+	}
+	replica1 := roachpb.ReplicaDescriptor{
+		ReplicaID: roachpb.ReplicaID(mtc.stores[1].StoreID()),
+		NodeID:    roachpb.NodeID(mtc.stores[1].StoreID()),
+		StoreID:   mtc.stores[1].StoreID(),
+	}
+	// Simulate an election triggered by the removed node.
+	if err := mtc.transports[0].Send(&storage.RaftMessageRequest{
+		GroupID:     rangeID,
+		ToReplica:   replica1,
+		FromReplica: replica0,
+		Message: raftpb.Message{
+			From: uint64(replica0.ReplicaID),
+			To:   uint64(replica1.ReplicaID),
+			Type: raftpb.MsgVote,
+			Term: term + 1,
+		},
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	// Wait a bit for the message to be processed.
+	// TODO(bdarnell): This will be easier to test without waiting
+	// when #5789 is done.
+	time.Sleep(10 * time.Millisecond)
+
+	// The message should have been discarded without triggering an
+	// election or changing the term.
+	newTerm := findTerm()
+	if term != newTerm {
+		t.Errorf("expected term to be constant, but changed from %v to %v", term, newTerm)
+	}
+}
+
 func TestReplicateReAddAfterDown(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 

storage/store.go

@@ -1855,6 +1855,31 @@ func (s *Store) enqueueRaftMessage(req *RaftMessageRequest) error {
 // Replica. It requires that processRaftMu is held and that s.mu is
 // not held.
 func (s *Store) handleRaftMessage(req *RaftMessageRequest) error {
+	// Drop messages that come from a node that we believe was once
+	// a member of the group but has been removed.
+	s.mu.Lock()
+	r, ok := s.mu.replicas[req.GroupID]
+	s.mu.Unlock()
+	if ok {
+		found := false
+		desc := r.Desc()
+		for _, rep := range desc.Replicas {
+			if rep.ReplicaID == req.FromReplica.ReplicaID {
+				found = true
+				break
+			}
+		}
+		// It's not a current member of the group. Is it from the future
+		// or the past?
+		if !found && req.FromReplica.ReplicaID < desc.NextReplicaID {
+			if log.V(2) {
+				log.Infof("range %s: discarding message from replica %v, older than NextReplicaID %v",
+					req.GroupID, req.FromReplica, desc.NextReplicaID)
+			}
+			return nil
+		}
+	}
+
 	switch req.Message.Type {
 	case raftpb.MsgSnap:
 		if !s.canApplySnapshot(req.GroupID, req.Message.Snapshot) {
@@ -1864,26 +1889,8 @@ func (s *Store) handleRaftMessage(req *RaftMessageRequest) error {
 			return nil
 		}
 
-	// TODO(bdarnell): handle coalesced heartbeats
 	case raftpb.MsgHeartbeat:
-		// A subset of coalesced heartbeats: drop heartbeats (but not
-		// other messages!) that come from a node that we don't believe to
-		// be a current member of the group.
-		s.mu.Lock()
-		r, ok := s.mu.replicas[req.GroupID]
-		s.mu.Unlock()
-		if ok {
-			found := false
-			for _, rep := range r.Desc().Replicas {
-				if rep.ReplicaID == req.FromReplica.ReplicaID {
-					found = true
-					break
-				}
-			}
-			if !found && req.FromReplica.ReplicaID < r.Desc().NextReplicaID {
-				return nil
-			}
-		}
+		// TODO(bdarnell): handle coalesced heartbeats.
 	}
 
 	s.mu.Lock()