NRG: Disjoint majorities during membership changes and network partitions

This commit fixes the following bugs:

- Inconsistent Cluster Size: When a leader was partitioned from the
  cluster, immediately after proposing a EntryAddPeer. The remaining
  nodes could end up with different view of the cluster size and
  quorum. So followers could have cluster size and would not match
  the number of peers in the peer set. A subsequent leader election,
  electing one of the followers, could break the quorum system.

- Incorrect Leader Election: It was possible for a new leader to be
  elected without a proper quorum. This could happen if a partition
  occurred after a new peer was proposed but before that change was
  committed. A follower could add the uncommitted peer to its peer
  set but would not update its cluster size and quorum, leading to
  an invalid election.

Both issues are solved by making sure that when a peer is added or
removed from the membership, the cluster size and quorum are adjusted
accordingly, at the same time. Followers would first add peers when
receiving the EntryAddPeer, and then adjusting the cluster size
only after commit. This patch changes this behavior such that the
cluster size and quorum are recomputed upon receiving the EntryAddPeer
/ EntryRemovePeer proposals. This is inline with the membership
protocol proposed in Ongaro's dissertation, section 4.1.
This patch also removes the concept of a "known" peer from the Raft
layer. A node would add a peer to its peer set when first receiving
the corresponding appendEntry, and on commit it would be marked as
"known". This distinction no longer applies.

Signed-off-by: Daniele Sciascia <[email protected]>

Author: sciascid

server/monitor.go

@@ -4190,8 +4190,10 @@ func (s *Server) Raftz(opts *RaftzOptions) *RaftzStatus {
 			}
 			peer := RaftzGroupPeer{
 				Name:                s.serverNameForNode(id),
-				Known:               p.kp,
 				LastReplicatedIndex: p.li,
+				// The Raft layer no longer distinguishes between
+				// 'known' and 'unknown' peers.
+				Known: true,
 			}
 			if !p.ts.IsZero() {
 				peer.LastSeen = time.Since(p.ts).String()

server/raft.go

@@ -253,7 +253,6 @@ type catchupState struct {
 type lps struct {
 	ts time.Time // Last timestamp
 	li uint64    // Last index replicated
-	kp bool      // Known peer
 }
 
 const (
@@ -543,13 +542,13 @@ func (s *Server) initRaftNode(accName string, cfg *RaftConfig, labels pprofLabel
 	}
 
 	// Make sure to track ourselves.
-	n.peers[n.id] = &lps{time.Now(), 0, true}
+	n.peers[n.id] = &lps{time.Now(), 0}
 
 	// Track known peers
 	for _, peer := range ps.knownPeers {
 		if peer != n.id {
 			// Set these to 0 to start but mark as known peer.
-			n.peers[peer] = &lps{time.Time{}, 0, true}
+			n.peers[peer] = &lps{time.Time{}, 0}
 		}
 	}
 
@@ -2624,13 +2623,10 @@ func (n *raft) addPeer(peer string) {
 		delete(n.removed, peer)
 	}
 
-	if lp, ok := n.peers[peer]; !ok {
+	if _, ok := n.peers[peer]; !ok {
 		// We are not tracking this one automatically so we need
 		// to bump cluster size.
-		n.peers[peer] = &lps{time.Time{}, 0, true}
-	} else {
-		// Mark as added.
-		lp.kp = true
+		n.peers[peer] = &lps{time.Time{}, 0}
 	}
 
 	// Adjust cluster size and quorum if needed.
@@ -2917,13 +2913,7 @@ func (n *raft) runCatchup(ar *appendEntryResponse, indexUpdatesQ *ipQueue[uint64
 		if len(n.progress) == 0 {
 			n.progress = nil
 		}
-		// Check if this is a new peer and if so go ahead and propose adding them.
-		_, exists := n.peers[peer]
 		n.Unlock()
-		if !exists {
-			n.debug("Catchup done for %q, will add into peers", peer)
-			n.ProposeAddPeer(peer)
-		}
 		indexUpdatesQ.unregister()
 	}()
 
@@ -3188,36 +3178,16 @@ func (n *raft) applyCommit(index uint64) error {
 					data:      e.Data,
 				})
 			}
-		case EntryPeerState:
-			if n.State() != Leader {
-				if ps, err := decodePeerState(e.Data); err == nil {
-					n.processPeerState(ps)
-				}
-			}
 		case EntryAddPeer:
-			newPeer := string(e.Data)
-			n.debug("Added peer %q", newPeer)
-
-			// Store our peer in our global peer map for all peers.
-			peers.LoadOrStore(newPeer, newPeer)
-
-			n.addPeer(newPeer)
-
 			// We pass these up as well.
 			committed = append(committed, e)
 
 			// We are done with this membership change
 			n.membChanging = false
-
 		case EntryRemovePeer:
 			peer := string(e.Data)
 			n.debug("Removing peer %q", peer)
 
-			n.removePeer(peer)
-
-			// Remove from string intern map.
-			peers.Delete(peer)
-
 			// We pass these up as well.
 			committed = append(committed, e)
 
@@ -3305,25 +3275,20 @@ func (n *raft) trackResponse(ar *appendEntryResponse) bool {
 // Used to adjust cluster size and peer count based on added official peers.
 // lock should be held.
 func (n *raft) adjustClusterSizeAndQuorum() {
-	pcsz, ncsz := n.csz, 0
-	for _, peer := range n.peers {
-		if peer.kp {
-			ncsz++
-		}
-	}
-	n.csz = ncsz
+	pcsz := n.csz
+	n.csz = len(n.peers)
 	n.qn = n.csz/2 + 1
 
-	if ncsz > pcsz {
-		n.debug("Expanding our clustersize: %d -> %d", pcsz, ncsz)
+	if n.csz > pcsz {
+		n.debug("Expanding our clustersize: %d -> %d", pcsz, n.csz)
 		n.lsut = time.Now()
-	} else if ncsz < pcsz {
-		n.debug("Decreasing our clustersize: %d -> %d", pcsz, ncsz)
+	} else if n.csz < pcsz {
+		n.debug("Decreasing our clustersize: %d -> %d", pcsz, n.csz)
 		if n.State() == Leader {
 			go n.sendHeartbeat()
 		}
 	}
-	if ncsz != pcsz {
+	if n.csz != pcsz {
 		n.recreateInternalSubsLocked()
 	}
 }
@@ -3341,7 +3306,7 @@ func (n *raft) trackPeer(peer string) error {
 		}
 	}
 	if n.State() == Leader {
-		if lp, ok := n.peers[peer]; !ok || !lp.kp {
+		if _, ok := n.peers[peer]; !ok {
 			// Check if this peer had been removed previously.
 			needPeerAdd = !isRemoved
 		}
@@ -3936,15 +3901,24 @@ CONTINUE:
 					}
 				}
 			}
-		case EntryAddPeer:
-			if newPeer := string(e.Data); len(newPeer) == idLen {
-				// Track directly, but wait for commit to be official
-				if _, ok := n.peers[newPeer]; !ok {
-					n.peers[newPeer] = &lps{time.Time{}, 0, false}
+		case EntryPeerState:
+			if n.State() != Leader {
+				if ps, err := decodePeerState(e.Data); err == nil {
+					n.processPeerState(ps)
 				}
-				// Store our peer in our global peer map for all peers.
-				peers.LoadOrStore(newPeer, newPeer)
 			}
+		case EntryAddPeer:
+			peer := string(e.Data)
+			// Store our peer in our global peer map for all peers.
+			peers.LoadOrStore(peer, peer)
+			n.addPeer(peer)
+			n.debug("Added peer %q", peer)
+		case EntryRemovePeer:
+			peer := string(e.Data)
+			// Remove from string intern map.
+			peers.Delete(peer)
+			n.removePeer(peer)
+			n.debug("Removed peer %q", peer)
 		}
 	}
 
@@ -4006,10 +3980,9 @@ func (n *raft) processPeerState(ps *peerState) {
 	n.peers = make(map[string]*lps)
 	for _, peer := range ps.knownPeers {
 		if lp := old[peer]; lp != nil {
-			lp.kp = true
 			n.peers[peer] = lp
 		} else {
-			n.peers[peer] = &lps{time.Time{}, 0, true}
+			n.peers[peer] = &lps{time.Time{}, 0}
 		}
 	}
 	n.debug("Update peers from leader to %+v", n.peers)
@@ -4251,10 +4224,8 @@ func decodePeerState(buf []byte) (*peerState, error) {
 // Lock should be held.
 func (n *raft) peerNames() []string {
 	var peers []string
-	for name, peer := range n.peers {
-		if peer.kp {
-			peers = append(peers, name)
-		}
+	for name := range n.peers {
+		peers = append(peers, name)
 	}
 	return peers
 }

server/raft_helpers_test.go

@@ -57,7 +57,7 @@ func (sg smGroup) leader() stateMachine {
 	return nil
 }
 
-func (sg smGroup) followers() []stateMachine {
+func (sg smGroup) followers() smGroup {
 	var f []stateMachine
 	for _, sm := range sg {
 		if sm.node().Leader() {