NRG: Track uncommitted membership entry index in membChanging

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

Author: MauriceVanVeen

server/raft.go

@@ -171,13 +171,14 @@ type raft struct {
 	llqrt  time.Time   // Last quorum lost time
 	lsut   time.Time   // Last scale-up time
 
-	term      uint64 // The current vote term
-	pterm     uint64 // Previous term from the last snapshot
-	pindex    uint64 // Previous index from the last snapshot
-	commit    uint64 // Index of the most recent commit
-	processed uint64 // Index of the most recently processed commit
-	applied   uint64 // Index of the most recently applied commit
-	papplied  uint64 // First sequence of our log, matches when we last installed a snapshot.
+	term         uint64 // The current vote term
+	pterm        uint64 // Previous term from the last snapshot
+	pindex       uint64 // Previous index from the last snapshot
+	commit       uint64 // Index of the most recent commit
+	processed    uint64 // Index of the most recently processed commit
+	applied      uint64 // Index of the most recently applied commit
+	papplied     uint64 // First sequence of our log, matches when we last installed a snapshot.
+	membChanging uint64 // Index of uncommitted membership change entry (0 means no change in progress)
 
 	aflr uint64 // Index when to signal initial messages have been applied after becoming leader. 0 means signaling is disabled.
 
@@ -231,7 +232,6 @@ type raft struct {
 	observer     bool // The node is observing, i.e. not able to become leader
 	initializing bool // The node is new, and "empty log" checks can be temporarily relaxed.
 	scaleUp      bool // The node is part of a scale up, puts us in observer mode until the log contains data.
-	membChanging bool // There is a membership change proposal in progress
 	deleted      bool // If the node was deleted.
 }
 
@@ -954,7 +954,7 @@ func (n *raft) ProposeAddPeer(peer string) error {
 		n.RUnlock()
 		return werr
 	}
-	if n.membChanging {
+	if n.membChanging > 0 {
 		n.RUnlock()
 		return errMembershipChange
 	}
@@ -984,7 +984,7 @@ func (n *raft) ProposeRemovePeer(peer string) error {
 		return nil
 	}
 
-	if n.membChanging {
+	if n.membChanging > 0 {
 		n.RUnlock()
 		return errMembershipChange
 	}
@@ -1004,7 +1004,7 @@ func (n *raft) ProposeRemovePeer(peer string) error {
 func (n *raft) MembershipChangeInProgress() bool {
 	n.RLock()
 	defer n.RUnlock()
-	return n.membChanging
+	return n.membChanging > 0
 }
 
 // ClusterSize reports back the total cluster size.
@@ -2632,7 +2632,7 @@ func (n *raft) handleForwardedRemovePeerProposal(sub *subscription, c *client, _
 		n.RUnlock()
 		return
 	}
-	if n.membChanging {
+	if n.membChanging > 0 {
 		n.debug("Ignoring forwarded peer removal proposal, membership changing")
 		n.RUnlock()
 		return
@@ -2716,16 +2716,16 @@ func (n *raft) sendMembershipChange(e *Entry) bool {
 
 	// Only makes sense to call this with entries that change membership.
 	// Also, ignore if we're already changing membership.
-	if !e.ChangesMembership() || n.membChanging {
+	if !e.ChangesMembership() || n.membChanging > 0 {
 		return false
 	}
 
-	n.membChanging = true
 	err := n.sendAppendEntryLocked([]*Entry{e}, true)
 	if err != nil {
-		n.membChanging = false
 		return false
 	}
+	// Set to the index where we stored the membership change
+	n.membChanging = n.pindex
 
 	if e.Type == EntryAddPeer {
 		n.addPeer(string(e.Data))
@@ -3227,7 +3227,7 @@ func (n *raft) applyCommit(index uint64) error {
 			committed = append(committed, e)
 
 			// We are done with this membership change
-			n.membChanging = false
+			n.membChanging = 0
 
 		case EntryRemovePeer:
 			peer := string(e.Data)
@@ -3242,7 +3242,7 @@ func (n *raft) applyCommit(index uint64) error {
 			committed = append(committed, e)
 
 			// We are done with this membership change
-			n.membChanging = false
+			n.membChanging = 0
 
 			// If this is us and we are the leader signal the caller
 			// to attempt to stepdown.
@@ -3589,6 +3589,11 @@ func (n *raft) truncateWAL(term, index uint64) {
 	}
 	// Set after we know we have truncated properly.
 	n.pterm, n.pindex = term, index
+
+	// Check if we're truncating an uncommitted membership change.
+	if n.membChanging > 0 && n.membChanging > index {
+		n.membChanging = 0
+	}
 }
 
 // Reset our WAL. This is equivalent to truncating all data from the log.
@@ -3958,7 +3963,8 @@ CONTINUE:
 			}
 		case EntryAddPeer:
 			// When receiving or restoring, mark membership as changing.
-			n.membChanging = true
+			// Set to the index where this entry was stored (pindex is now this entry's index)
+			n.membChanging = n.pindex
 			if newPeer := string(e.Data); len(newPeer) == idLen {
 				// Track directly, but wait for commit to be official
 				if _, ok := n.peers[newPeer]; !ok {
@@ -3969,7 +3975,8 @@ CONTINUE:
 			}
 		case EntryRemovePeer:
 			// When receiving or restoring, mark membership as changing.
-			n.membChanging = true
+			// Set to the index where this entry was stored (pindex is now this entry's index)
+			n.membChanging = n.pindex
 		}
 	}
 

server/raft_test.go

@@ -4438,6 +4438,36 @@ func TestNRGUncommittedMembershipChangeOnNewLeader(t *testing.T) {
 	require_Error(t, err, errMembershipChange)
 }
 
+func TestNRGUncommittedMembershipChangeGetsTruncated(t *testing.T) {
+	n, cleanup := initSingleMemRaftNode(t)
+	defer cleanup()
+
+	nats1 := "yrzKKRBu" // "nats-1"
+
+	esm := encodeStreamMsgAllowCompress("foo", "_INBOX.foo", nil, nil, 0, 0, true)
+	entries := []*Entry{newEntry(EntryNormal, esm)}
+	aeMsg := encode(t, &appendEntry{leader: nats1, term: 1, commit: 0, pterm: 0, pindex: 0, entries: entries})
+	entries = []*Entry{newEntry(EntryAddPeer, []byte(nats1))}
+	aeAddPeer := encode(t, &appendEntry{leader: nats1, term: 1, commit: 0, pterm: 1, pindex: 1, entries: entries})
+
+	// Set up a membership change.
+	n.processAppendEntry(aeMsg, n.aesub)
+	n.processAppendEntry(aeAddPeer, n.aesub)
+	require_Equal(t, n.pindex, 2)
+	require_True(t, n.MembershipChangeInProgress())
+	require_Equal(t, n.membChanging, 2)
+
+	// If the entry containing the membership change isn't truncated, it should remain in progress.
+	n.truncateWAL(n.pterm, n.pindex)
+	require_True(t, n.MembershipChangeInProgress())
+	require_Equal(t, n.membChanging, 2)
+
+	// If the entry IS truncated, then it shouldn't be in progress anymore.
+	n.truncateWAL(n.pterm, n.pindex-1)
+	require_False(t, n.MembershipChangeInProgress())
+	require_Equal(t, n.membChanging, 0)
+}
+
 func TestNRGUncommittedMembershipChangeOnNewLeaderForwardedRemovePeerProposal(t *testing.T) {
 	n, cleanup := initSingleMemRaftNode(t)
 	defer cleanup()