NRG: Mark membChanging when sending

Signed-off-by: Maurice van Veen <github@mauricevanveen.com>

Author: MauriceVanVeen

server/raft.go

@@ -943,61 +943,59 @@ func (n *raft) ForwardProposal(entry []byte) error {
 
 // ProposeAddPeer is called to add a peer to the group.
 func (n *raft) ProposeAddPeer(peer string) error {
-	n.Lock()
+	n.RLock()
 	// Check state under lock, we might not be leader anymore.
 	if n.State() != Leader {
-		n.Unlock()
+		n.RUnlock()
 		return errNotLeader
 	}
 	// Error if we had a previous write error.
 	if werr := n.werr; werr != nil {
-		n.Unlock()
+		n.RUnlock()
 		return werr
 	}
 	if n.membChanging {
-		n.Unlock()
+		n.RUnlock()
 		return errMembershipChange
 	}
 	prop := n.prop
-	n.membChanging = true
-	n.Unlock()
+	n.RUnlock()
 
 	prop.push(newProposedEntry(newEntry(EntryAddPeer, []byte(peer)), _EMPTY_))
 	return nil
 }
 
 // ProposeRemovePeer is called to remove a peer from the group.
 func (n *raft) ProposeRemovePeer(peer string) error {
-	n.Lock()
+	n.RLock()
 
 	// Error if we had a previous write error.
 	if werr := n.werr; werr != nil {
-		n.Unlock()
+		n.RUnlock()
 		return werr
 	}
 
 	if n.State() != Leader {
 		subj := n.rpsubj
-		n.Unlock()
+		n.RUnlock()
 
 		// Forward the proposal to the leader
 		n.sendRPC(subj, _EMPTY_, []byte(peer))
 		return nil
 	}
 
 	if n.membChanging {
-		n.Unlock()
+		n.RUnlock()
 		return errMembershipChange
 	}
 
 	if len(n.peers) <= 1 {
-		n.Unlock()
+		n.RUnlock()
 		return errRemoveLastNode
 	}
 
 	prop := n.prop
-	n.membChanging = true
-	n.Unlock()
+	n.RUnlock()
 
 	prop.push(newProposedEntry(newEntry(EntryRemovePeer, []byte(peer)), _EMPTY_))
 	return nil
@@ -2622,26 +2620,25 @@ func (n *raft) handleForwardedRemovePeerProposal(sub *subscription, c *client, _
 		return
 	}
 
-	n.Lock()
+	n.RLock()
 	// Check state under lock, we might not be leader anymore.
 	if n.State() != Leader || !n.leaderState.Load() {
 		n.debug("Ignoring forwarded peer removal proposal, not leader")
-		n.Unlock()
+		n.RUnlock()
 		return
 	}
 	// Error if we had a previous write error.
 	if werr := n.werr; werr != nil {
-		n.Unlock()
+		n.RUnlock()
 		return
 	}
 	if n.membChanging {
 		n.debug("Ignoring forwarded peer removal proposal, membership changing")
-		n.Unlock()
+		n.RUnlock()
 		return
 	}
 	prop := n.prop
-	n.membChanging = true
-	n.Unlock()
+	n.RUnlock()
 
 	// Need to copy since this is underlying client/route buffer.
 	peer := copyBytes(msg)
@@ -2717,11 +2714,13 @@ func (n *raft) sendMembershipChange(e *Entry) bool {
 	n.Lock()
 	defer n.Unlock()
 
-	// Only makes sense to call this with entries that change membership
-	if !e.ChangesMembership() {
+	// Only makes sense to call this with entries that change membership.
+	// Also, ignore if we're already changing membership.
+	if !e.ChangesMembership() || n.membChanging {
 		return false
 	}
 
+	n.membChanging = true
 	err := n.sendAppendEntryLocked([]*Entry{e}, true)
 	if err != nil {
 		n.membChanging = false

server/raft_test.go

@@ -4380,7 +4380,13 @@ func TestNRGProposeRemovePeerConcurrent(t *testing.T) {
 	require_NoError(t, err)
 
 	// Check that membership change is in progress.
-	require_True(t, n.MembershipChangeInProgress())
+	checkFor(t, 2*time.Second, 200*time.Millisecond, func() error {
+		if n.MembershipChangeInProgress() {
+			return nil
+		} else {
+			return errors.New("membership not in progress")
+		}
+	})
 
 	// Attempt to remove the second follower, should fail.
 	err = n.ProposeRemovePeer(locked[1].node().ID())
@@ -4627,18 +4633,19 @@ func TestNRGProposeRemovePeerAll(t *testing.T) {
 	followers := rg.followers()
 	require_Equal(t, len(followers), 2)
 
-	for _, follower := range followers {
+	peers := leader.node().Peers()
+	require_Equal(t, len(peers), 3)
+	for i, follower := range followers {
 		require_NoError(t, leader.node().ProposeRemovePeer(follower.node().ID()))
-		checkFor(t, 1*time.Second, 10*time.Millisecond, func() error {
-			if leader.node().MembershipChangeInProgress() {
-				return errors.New("membership still in progress")
-			} else {
+		checkFor(t, 2*time.Second, 10*time.Millisecond, func() error {
+			if peers = leader.node().Peers(); len(peers) == 2-i {
 				return nil
 			}
+			return errors.New("membership still in progress")
 		})
 	}
 
-	peers := leader.node().Peers()
+	peers = leader.node().Peers()
 	leaderID := leader.node().ID()
 
 	// The leader is the only one left...
@@ -4661,16 +4668,13 @@ func TestNRGLeaderResurrectsRemovedPeers(t *testing.T) {
 
 	// Remove one follower
 	require_NoError(t, leader.node().ProposeRemovePeer(followers[0].node().ID()))
-	checkFor(t, 1*time.Second, 10*time.Millisecond, func() error {
-		if leader.node().MembershipChangeInProgress() {
-			return errors.New("membership still in progress")
-		} else {
+	checkFor(t, 2*time.Second, 10*time.Millisecond, func() error {
+		if peers := leader.node().Peers(); len(peers) == 2 {
 			return nil
 		}
+		return errors.New("membership still in progress")
 	})
 
-	require_Equal(t, len(leader.node().Peers()), 2)
-
 	// Stop the leader and restart it.
 	// If bug is present: the leader resurrects the previously removed peer.
 	leader.stop()